feat(dev): wire WSS tunnel into mirrorstack dev (--tunnel)

Cargo.toml

@@ -29,7 +29,7 @@ dialoguer = { version = "0.11", default-features = false }
 console = "0.15"
 indicatif = "0.17"
 ctrlc = "3"
-tokio = { version = "1", features = ["rt", "macros", "time", "sync", "net", "io-util"] }
+tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "time", "sync", "net", "io-util"] }
 tokio-tungstenite = { version = "0.24", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] }
 futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
 

src/api.rs

@@ -76,6 +76,41 @@ pub struct CreateModuleInput<'a> {
 }
 
 /// GET /v1/auth/me — returns the authenticated user's identity.
+/// Response shape from `POST /v1/tunnel/token`. The CLI follows up with
+/// a WebSocket connect against `wss_url` carrying `?token=<token>`.
+#[derive(Deserialize, Debug)]
+pub struct TunnelToken {
+    pub token: String,
+    pub wss_url: String,
+    /// RFC3339. Server-side TTL is short (60s); we don't act on this
+    /// value (any failure triggers a fresh mint), but it's surfaced for
+    /// diagnostic logging if the connect hangs.
+    #[allow(dead_code)]
+    pub expires_at: String,
+}
+
+/// POST /v1/tunnel/token — mint a connect token for the WSS dev tunnel.
+pub fn tunnel_token(
+    http: &Client,
+    dispatch_base: &str,
+    access_token: &str,
+) -> Result<TunnelToken, ApiError> {
+    let endpoint = format!("{}/v1/tunnel/token", dispatch_base.trim_end_matches('/'));
+    let resp = http
+        .post(&endpoint)
+        .bearer_auth(access_token)
+        .header("Accept", "application/json")
+        .send()?;
+    let status = resp.status();
+    if status.is_success() {
+        return Ok(resp.json::<TunnelToken>()?);
+    }
+    if status == reqwest::StatusCode::UNAUTHORIZED || status == reqwest::StatusCode::FORBIDDEN {
+        return Err(ApiError::Unauthenticated);
+    }
+    Err(unexpected_body_error(resp))
+}
+
 pub fn me(http: &Client, api_base: &str, access_token: &str) -> Result<Identity, ApiError> {
     let endpoint = format!("{}/v1/auth/me", api_base.trim_end_matches('/'));
 

src/commands/dev/mod.rs

@@ -1,31 +1,34 @@
 //! `mirrorstack dev` — run a module locally with the supporting services
-//! it needs (Postgres for now; Redis + WSS tunnel client land in later PRs).
+//! it needs.
 //!
 //! The lifecycle is:
 //!   1. Bootstrap a `docker-compose.yml` in the cwd if missing
 //!   2. `docker compose up -d --wait` (server-side healthcheck blocks until pg is ready)
-//!   3. Spawn `go run .` with `MS_LOCAL_DB_URL` injected
-//!   4. Stream the module's stdout/stderr through a labeled prefix
-//!   5. On Ctrl-C: kill the module process (SIGKILL on unix; SIGTERM-then-SIGKILL is queued as a follow-up), then `docker compose down`
-//!
-//! No WSS tunnel yet — the module runs as a normal HTTP server on its own
-//! port. Future PR layers the tunnel registration on top.
+//!   3. (optional, --tunnel) Mint a connect token and open the WSS so
+//!      remote callers can reach this module via the platform's Leaf 1
+//!      302 path
+//!   4. Spawn `go run .` with `MS_LOCAL_DB_URL` injected
+//!   5. Stream the module's stdout/stderr through a labeled prefix
+//!   6. On Ctrl-C: kill the module (SIGKILL on unix; SIGTERM-then-SIGKILL
+//!      is queued as a follow-up — see issue #25), close the tunnel if
+//!      open, then `docker compose down`
 
 use std::io::IsTerminal;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
+use std::time::Duration;
 
-use anyhow::{Result, anyhow};
+use anyhow::{Context, Result, anyhow};
 use clap::Args;
 use console::style;
 
-use super::{ok_mark, warn_prefix};
+use super::{
+    DEFAULT_DISPATCH_BASE, ENV_DISPATCH_URL, ok_mark, resolve_base, session_expired, warn_prefix,
+};
+use crate::{api, credentials, http};
 
 mod compose;
+mod module_meta;
 mod process;
-// Used in a follow-up PR that integrates WSS into the dev lifecycle.
-// Compiled + tested now so the wire format and protocol stay correct
-// while the AWS Sender impl + IaC for the WS API GW catch up.
-#[allow(dead_code)]
 mod tunnel;
 
 #[derive(Args)]
@@ -41,13 +44,28 @@ pub struct DevArgs {
     /// when --no-compose is unset, otherwise must be supplied via env.
     #[arg(long)]
     db_url: Option<String>,
+    /// Open a WSS tunnel to the platform so deployed callers can reach this
+    /// module via Leaf 1 (302 → localhost). Requires the developer to be
+    /// signed in (via `mirrorstack login`) and the platform's dispatch
+    /// service + WS API GW to be reachable. Module identity is parsed from
+    /// `Config.ID` in main.go; --local-url defaults to http://localhost:8080.
+    #[arg(long)]
+    tunnel: bool,
+    /// URL the platform should 302 to when routing inbound calls for this
+    /// module. Only used when --tunnel is set. Default: http://localhost:8080.
+    #[arg(long)]
+    local_url: Option<String>,
 }
 
 // Matches templates/dev/docker-compose.yml.tmpl: host port 5433 (chosen so
 // the bundled compose can coexist with api-platform's own postgres on 5432).
 const DEFAULT_DB_URL: &str =
     "postgres://mirrorstack:mirrorstack@localhost:5433/mirrorstack?sslmode=disable";
 
+/// Default local URL the platform should 302 to for Leaf 1 routing. Matches
+/// the SDK's default HTTP listener port. Override with `--local-url`.
+const DEFAULT_LOCAL_URL: &str = "http://localhost:8080";
+
 pub fn run(args: DevArgs) -> Result<()> {
     let cwd = args
         .dir
@@ -62,6 +80,15 @@ pub fn run(args: DevArgs) -> Result<()> {
 
     let db_url = resolve_db_url(&args)?;
 
+    // Bring the tunnel up BEFORE compose so that a tunnel registration
+    // failure (auth expired, dispatch unreachable) doesn't waste the user's
+    // time bringing containers up first.
+    let tunnel_handle = if args.tunnel {
+        Some(open_tunnel(&cwd, args.local_url.as_deref())?)
+    } else {
+        None
+    };
+
     if !args.no_compose {
         compose::ensure_compose_file(&cwd)?;
         compose::up(&cwd)?;
@@ -70,6 +97,14 @@ pub fn run(args: DevArgs) -> Result<()> {
     eprintln!("{} module running — Ctrl-C to stop", ok_mark());
     let module_status = process::run_module(&cwd, &db_url);
 
+    if let Some((handle, runtime)) = tunnel_handle {
+        handle.shutdown();
+        // Block on the runtime briefly to let the close frame land.
+        runtime.block_on(async {
+            tokio::time::sleep(Duration::from_millis(200)).await;
+        });
+    }
+
     if !args.no_compose {
         if let Err(e) = compose::down(&cwd) {
             eprintln!(
@@ -82,6 +117,69 @@ pub fn run(args: DevArgs) -> Result<()> {
     module_status
 }
 
+/// Build a single-threaded tokio runtime, mint a connect token via
+/// dispatch, open the WSS, send `register`, and return the handle so the
+/// caller can shut it down on Ctrl-C. Runs blocking under the hood —
+/// the tunnel itself stays alive on the runtime's worker thread.
+fn open_tunnel(
+    module_dir: &Path,
+    local_url: Option<&str>,
+) -> Result<(tunnel::TunnelHandle, tokio::runtime::Runtime)> {
+    let creds = credentials::load_or_login_hint()?;
+    let dispatch_base = resolve_base(ENV_DISPATCH_URL, DEFAULT_DISPATCH_BASE);
+    let client = http::client(Duration::from_secs(15))?;
+    let module_id = module_meta::read_module_id(module_dir)?;
+    let local_url = local_url.unwrap_or(DEFAULT_LOCAL_URL);
+
+    eprintln!(
+        "{} fetching tunnel token from {}",
+        ok_mark(),
+        style(&dispatch_base).cyan().dim()
+    );
+    let token = match api::tunnel_token(&client, &dispatch_base, &creds.access_token) {
+        Ok(t) => t,
+        Err(api::ApiError::Unauthenticated) => return Err(session_expired()),
+        Err(e) => {
+            // Dispatch unreachable is a routine "platform isn't running yet"
+            // error — point the user at the right env var.
+            return Err(anyhow!(
+                "dev: tunnel-token mint failed: {e}. Check {} (or {} env var) and that the dispatch service is running.",
+                dispatch_base,
+                ENV_DISPATCH_URL
+            ));
+        }
+    };
+    // Multi-thread (one worker) so the WSS background task — pinger,
+    // server-frame reader — keeps ticking while the rest of the CLI runs
+    // its blocking module-process loop. A current-thread runtime would
+    // freeze every spawned future the moment block_on returns.
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
+        .enable_all()
+        .build()
+        .context("dev: build tokio runtime")?;
+
+    let handle = runtime.block_on(async {
+        tunnel::open(
+            &token.wss_url,
+            &token.token,
+            tunnel::RegisterPayload {
+                module_id: &module_id,
+                local_url,
+                version: env!("CARGO_PKG_VERSION"),
+            },
+        )
+        .await
+    })?;
+
+    eprintln!(
+        "{} tunnel registered (session {})",
+        ok_mark(),
+        style(&handle.session_id).cyan().dim()
+    );
+    Ok((handle, runtime))
+}
+
 fn resolve_db_url(args: &DevArgs) -> Result<String> {
     if let Some(url) = &args.db_url {
         return Ok(url.clone());
@@ -111,13 +209,19 @@ pub(super) fn line_prefix(label: &str) -> String {
 mod tests {
     use super::*;
 
+    fn args_with(no_compose: bool, db_url: Option<String>) -> DevArgs {
+        DevArgs {
+            dir: None,
+            no_compose,
+            db_url,
+            tunnel: false,
+            local_url: None,
+        }
+    }
+
     #[test]
     fn resolve_db_url_prefers_explicit_arg() {
-        let args = DevArgs {
-            dir: None,
-            no_compose: false,
-            db_url: Some("postgres://x".into()),
-        };
+        let args = args_with(false, Some("postgres://x".into()));
         assert_eq!(resolve_db_url(&args).unwrap(), "postgres://x");
     }
 
@@ -129,11 +233,7 @@ mod tests {
         if std::env::var("MS_LOCAL_DB_URL").is_ok() {
             return;
         }
-        let args = DevArgs {
-            dir: None,
-            no_compose: true,
-            db_url: None,
-        };
+        let args = args_with(true, None);
         let err = resolve_db_url(&args).unwrap_err().to_string();
         assert!(err.contains("MS_LOCAL_DB_URL"));
     }
@@ -143,11 +243,7 @@ mod tests {
         if std::env::var("MS_LOCAL_DB_URL").is_ok() {
             return;
         }
-        let args = DevArgs {
-            dir: None,
-            no_compose: false,
-            db_url: None,
-        };
+        let args = args_with(false, None);
         let url = resolve_db_url(&args).unwrap();
         assert!(url.starts_with("postgres://mirrorstack"));
         assert!(url.contains(":5433/"));