From bbf0483b9e09673d4abb241ee4f9e38a7c5a3d12 Mon Sep 17 00:00:00 2001 From: Sheng Kun Chang Date: Wed, 20 May 2026 03:20:44 +0800 Subject: [PATCH 1/3] feat(dev/tunnel): surface platform rejections + auto-refresh access token MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related resilience fixes for `mirrorstack dev --tunnel` after running the flow end-to-end against the slug-aware platform: 1. await_register_ack ignored rpc.err — the 10s timeout was firing while the platform's rejection (module_dev_mode_off, module_not_yours) sat unread on the wire. Now correlates rpc.err frames by the register id, decodes the ErrorPayload (incl. the new optional slug field from api-platform #165), and returns a typed RegisterError variant. 2. Caller in dev/mod.rs maps RegisterError → user-actionable output: - ModuleDevModeOff: prints a warning, builds the dev-console URL (localhost:3001 for dev hosts, https://apps. for api. production, deep-linked to /dev/module//dev when the slug is carried in the payload, falling back to /dev otherwise), and opens it on Enter. - ModuleNotYours: explicit "this module isn't owned by you" message with the `mirrorstack module list` cross-check hint. - Rejected: generic "register rejected (): " so future error codes surface without a CLI change. 3. Silent access-token refresh — tunnel_token's 401 path now calls POST /v1/auth/sessions/refresh with the stored refresh token (30-day TTL), persists the rotated pair to credentials.json, and retries the mint once. Without this, the 15-minute access TTL killed long-running `mirrorstack dev` sessions even though the user's session was still valid for weeks. The refresh layer is scoped to the tunnel-token call site for now; can generalize as more long-running calls land. CLI tests stay 64-passing — these paths are integration-shaped and the existing test surface doesn't mock the WS frame plane. Co-Authored-By: Sheng Kun Chang Co-Authored-By: Claude Opus 4.7 (1M context) --- src/api.rs | 47 +++++++++++++++ src/commands/dev/mod.rs | 115 +++++++++++++++++++++++++++++++++++-- src/commands/dev/tunnel.rs | 105 +++++++++++++++++++++++++++------ 3 files changed, 242 insertions(+), 25 deletions(-) diff --git a/src/api.rs b/src/api.rs index 037754f..f11575d 100644 --- a/src/api.rs +++ b/src/api.rs @@ -214,6 +214,53 @@ pub fn create_module( }) } +/// Body of a successful `POST /v1/auth/sessions/refresh`. The platform +/// rotates the refresh token on every refresh (replay defense), so we +/// must persist the new one back to credentials. expires_at is RFC3339 +/// from the server but informational only — the CLI re-derives the +/// 15-minute access TTL locally when saving. +#[derive(Debug, Deserialize)] +pub struct RefreshResponse { + pub access_token: String, + pub refresh_token: String, + #[allow(dead_code)] + pub expires_at: String, +} + +/// POST /v1/auth/sessions/refresh — exchange a refresh token for a new +/// access + refresh pair. The CLI calls this on a 401 from any +/// access-token-bearing endpoint to recover without a fresh +/// `mirrorstack login`. A 401 from this endpoint means the refresh +/// token itself is gone (revoked, expired, or never existed) — surface +/// as Unauthenticated so callers can fall back to the login hint. +pub fn refresh_session( + http: &Client, + api_base: &str, + refresh_token: &str, +) -> Result { + #[derive(Serialize)] + struct Body<'a> { + refresh_token: &'a str, + } + let endpoint = format!( + "{}/v1/auth/sessions/refresh", + api_base.trim_end_matches('/') + ); + let resp = http + .post(&endpoint) + .header("Accept", "application/json") + .json(&Body { refresh_token }) + .send()?; + let status = resp.status(); + if status.is_success() { + return Ok(resp.json::()?); + } + if status == reqwest::StatusCode::UNAUTHORIZED { + return Err(ApiError::Unauthenticated); + } + Err(unexpected_body_error(resp)) +} + /// DELETE /v1/auth/sessions/current — revoke the supplied refresh token /// (CLI flow: token in body, not cookie). The platform treats a missing /// or already-revoked session as success, so callers can call this diff --git a/src/commands/dev/mod.rs b/src/commands/dev/mod.rs index 6de6380..ea8bf0c 100644 --- a/src/commands/dev/mod.rs +++ b/src/commands/dev/mod.rs @@ -15,14 +15,16 @@ use std::io::IsTerminal; use std::path::{Path, PathBuf}; -use std::time::Duration; +use std::time::{Duration, SystemTime}; use anyhow::{Context, Result, anyhow}; use clap::Args; use console::style; +use reqwest::blocking::Client; use super::{ - DEFAULT_DISPATCH_BASE, ENV_DISPATCH_URL, ok_mark, resolve_base, session_expired, warn_prefix, + DEFAULT_API_BASE, DEFAULT_DISPATCH_BASE, ENV_API_URL, ENV_DISPATCH_URL, ok_mark, resolve_base, + session_expired, warn_prefix, }; use crate::{api, credentials, http}; @@ -125,8 +127,9 @@ fn open_tunnel( module_dir: &Path, local_url: Option<&str>, ) -> Result<(tunnel::TunnelHandle, tokio::runtime::Runtime)> { - let creds = credentials::load_or_login_hint()?; + let mut creds = credentials::load_or_login_hint()?; let dispatch_base = resolve_base(ENV_DISPATCH_URL, DEFAULT_DISPATCH_BASE); + let api_base = resolve_base(ENV_API_URL, DEFAULT_API_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); @@ -136,7 +139,7 @@ fn open_tunnel( ok_mark(), style(&dispatch_base).cyan().dim() ); - let token = match api::tunnel_token(&client, &dispatch_base, &creds.access_token) { + let token = match mint_tunnel_token(&client, &dispatch_base, &api_base, &mut creds) { Ok(t) => t, Err(api::ApiError::Unauthenticated) => return Err(session_expired()), Err(e) => { @@ -159,7 +162,7 @@ fn open_tunnel( .build() .context("dev: build tokio runtime")?; - let handle = runtime.block_on(async { + let handle = match runtime.block_on(async { tunnel::open( &token.wss_url, &token.token, @@ -170,7 +173,21 @@ fn open_tunnel( }, ) .await - })?; + }) { + Ok(h) => h, + Err(tunnel::RegisterError::ModuleDevModeOff { slug }) => { + return Err(dev_mode_off_hint(&dispatch_base, slug.as_deref())); + } + Err(tunnel::RegisterError::ModuleNotYours) => { + return Err(anyhow!( + "dev: this module isn't owned by you — only the module owner can open a dev tunnel.\nIf you scaffolded with `mirrorstack module init`, your `Config.ID` should match the platform's record; re-check it under `mirrorstack module list`." + )); + } + Err(tunnel::RegisterError::Rejected { code, message }) => { + return Err(anyhow!("dev: register rejected ({code}): {message}")); + } + Err(tunnel::RegisterError::Transport(e)) => return Err(e), + }; eprintln!( "{} tunnel registered (session {})", @@ -180,6 +197,92 @@ fn open_tunnel( Ok((handle, runtime)) } +/// Access tokens are short (15 min per the platform's TokenConfig). +/// `mirrorstack dev` is a long-running command, so we silently refresh +/// the access token via the stored refresh token (30-day TTL) on a 401 +/// and retry the mint once. The refresh endpoint rotates the refresh +/// token too, so the rotated pair is persisted back to credentials. +/// If the refresh ALSO 401s (revoked / expired session), the original +/// Unauthenticated bubbles up and the caller surfaces session_expired. +fn mint_tunnel_token( + client: &Client, + dispatch_base: &str, + api_base: &str, + creds: &mut credentials::Credentials, +) -> Result { + match api::tunnel_token(client, dispatch_base, &creds.access_token) { + Ok(t) => Ok(t), + Err(api::ApiError::Unauthenticated) => { + let refreshed = api::refresh_session(client, api_base, &creds.refresh_token)?; + creds.access_token = refreshed.access_token; + creds.refresh_token = refreshed.refresh_token; + // expires_at is informational only on the credentials struct; + // match the platform's 15-minute access TTL. + creds.expires_at = SystemTime::now() + Duration::from_secs(15 * 60); + if let Err(e) = credentials::save(creds) { + // Save failure isn't fatal for the in-process retry (we + // hold the new access token in `creds`), but log so the + // user sees the warning if the next run re-prompts login. + eprintln!( + "{} refreshed session but failed to save credentials: {e:#}", + warn_prefix() + ); + } + api::tunnel_token(client, dispatch_base, &creds.access_token) + } + Err(e) => Err(e), + } +} + +/// Build a user-facing error for the `module_dev_mode_off` rpc.err. The +/// platform refused the tunnel because the module's `dev_mode_enabled` +/// flag is false; coach the user toward the right toggle and offer to +/// open it. When the platform carries the module slug in the error +/// payload, deep-link straight to the Dev tab; otherwise degrade to +/// the modules list (older platforms predate the slug field). +fn dev_mode_off_hint(dispatch_base: &str, slug: Option<&str>) -> anyhow::Error { + let console = dev_console_url(dispatch_base, slug); + eprintln!(); + eprintln!( + "{} dev mode is disabled for this module on the platform.", + warn_prefix() + ); + eprintln!(" Open the dev console and toggle Dev mode on:"); + eprintln!(" {}", style(&console).cyan().underlined()); + eprintln!(); + eprint!(" Press Enter to open in your browser, or Ctrl-C to cancel..."); + let _ = std::io::Write::flush(&mut std::io::stderr()); + let mut buf = String::new(); + if std::io::stdin().read_line(&mut buf).is_ok() { + let _ = open::that(&console); + } + anyhow!("dev: module dev_mode disabled — re-run after enabling it") +} + +/// Derive the dev-console web URL from the dispatch HTTP URL. +/// localhost/127.0.0.1 → http://localhost:3001; production +/// (`api.`) → `https://apps.`; anything else falls back to +/// the canonical prod console so the message stays useful. When `slug` +/// is provided, deep-link to /dev/module//dev; otherwise the +/// modules-list landing page. +fn dev_console_url(dispatch_base: &str, slug: Option<&str>) -> String { + let parsed = url::Url::parse(dispatch_base).ok(); + let host = parsed.as_ref().and_then(|u| u.host_str()); + let base = match host { + Some("localhost") | Some("127.0.0.1") | Some("::1") => { + "http://localhost:3001".to_string() + } + Some(h) if h.starts_with("api.") => { + format!("https://apps.{}", &h["api.".len()..]) + } + _ => "https://apps.mirrorstack.ai".to_string(), + }; + match slug { + Some(s) if !s.is_empty() => format!("{base}/dev/module/{s}/dev"), + _ => format!("{base}/dev"), + } +} + fn resolve_db_url(args: &DevArgs) -> Result { if let Some(url) = &args.db_url { return Ok(url.clone()); diff --git a/src/commands/dev/tunnel.rs b/src/commands/dev/tunnel.rs index dc5a189..ec19d55 100644 --- a/src/commands/dev/tunnel.rs +++ b/src/commands/dev/tunnel.rs @@ -140,6 +140,41 @@ pub(super) struct RegisterAck { pub expires_at: String, } +#[derive(Debug, Deserialize)] +struct ErrorPayload { + code: String, + message: String, + #[serde(default)] + slug: Option, +} + +/// Typed register-time rejection. Distinguishes the cases the CLI knows +/// how to coach the user through (`module_dev_mode_off`, +/// `module_not_yours`) from "every other rpc.err on the register +/// channel" so callers can match without string-sniffing. +#[derive(Debug, thiserror::Error)] +pub(super) enum RegisterError { + #[error("dev mode is disabled on this module")] + ModuleDevModeOff { + /// Module slug from the rpc.err payload — empty when the + /// platform predates the slug-in-error PR. Callers should + /// degrade to the /dev list URL when None. + slug: Option, + }, + #[error("module is not owned by you")] + ModuleNotYours, + #[error("register rejected ({code}): {message}")] + Rejected { code: String, message: String }, + #[error(transparent)] + Transport(anyhow::Error), +} + +impl From for RegisterError { + fn from(e: anyhow::Error) -> Self { + RegisterError::Transport(e) + } +} + /// Spawned-task handle. Drop or call [`shutdown`] to close the WSS. /// /// `service_token` is intentionally NOT held here — it lives on the @@ -166,8 +201,8 @@ pub(super) async fn open( wss_url: &str, ttok: &str, register: RegisterPayload<'_>, -) -> Result { - let url = with_token_param(wss_url, ttok)?; +) -> std::result::Result { + let url = with_token_param(wss_url, ttok).map_err(RegisterError::from)?; let config = WebSocketConfig { max_message_size: Some(MAX_INBOUND_FRAME_BYTES), max_frame_size: Some(MAX_INBOUND_FRAME_BYTES), @@ -176,25 +211,33 @@ pub(super) async fn open( let (ws_stream, _resp) = tokio_tungstenite::connect_async_with_config(&url, Some(config), false) .await - .with_context(|| format!("dev: connect WSS {wss_url}"))?; + .with_context(|| format!("dev: connect WSS {wss_url}")) + .map_err(RegisterError::from)?; let (mut sink, mut stream) = ws_stream.split(); let register_frame = Frame::new( FrameType::Register, - Some(serde_json::to_value(®ister).context("dev: serialize register payload")?), + Some( + serde_json::to_value(®ister) + .context("dev: serialize register payload") + .map_err(RegisterError::from)?, + ), ); let register_id = register_frame.id.clone(); - sink.send(Message::Text(serde_json::to_string(®ister_frame)?)) - .await - .context("dev: send register frame")?; + sink.send(Message::Text( + serde_json::to_string(®ister_frame).map_err(|e| RegisterError::from(anyhow!(e)))?, + )) + .await + .context("dev: send register frame") + .map_err(RegisterError::from)?; let ack = tokio::time::timeout( Duration::from_secs(10), await_register_ack(&mut sink, &mut stream, ®ister_id), ) .await - .context("dev: register_ack timeout")? - .context("dev: register_ack")?; + .context("dev: register_ack timeout") + .map_err(RegisterError::from)??; let shutdown = Arc::new(Notify::new()); tokio::spawn(run_tunnel_loop(sink, stream, shutdown.clone())); @@ -223,7 +266,7 @@ async fn await_register_ack( sink: &mut WsSink, stream: &mut WsReader, register_id: &str, -) -> Result { +) -> std::result::Result { loop { let next = stream .next() @@ -237,20 +280,44 @@ async fn await_register_ack( let _ = sink.send(Message::Pong(p)).await; continue; } - Message::Close(_) => return Err(anyhow!("server closed before register_ack")), + Message::Close(_) => return Err(anyhow!("server closed before register_ack").into()), _ => continue, }; let frame: Frame = serde_json::from_str(&text) .with_context(|| format!("dev: parse server frame: {text}"))?; - if frame.frame_type == FrameType::RegisterAck - && frame.corr_id.as_deref() == Some(register_id) - { - let payload = frame - .payload - .ok_or_else(|| anyhow!("register_ack missing payload"))?; - return serde_json::from_value(payload).context("dev: deserialize register_ack"); + // The server replies on the register's corr_id either with + // register_ack (success) or rpc.err (rejection). Anything else + // during handshake (server-initiated ping, future frame types) + // gets ignored. + if frame.corr_id.as_deref() != Some(register_id) { + continue; + } + match frame.frame_type { + FrameType::RegisterAck => { + let payload = frame + .payload + .ok_or_else(|| anyhow!("register_ack missing payload"))?; + return serde_json::from_value(payload) + .context("dev: deserialize register_ack") + .map_err(RegisterError::from); + } + FrameType::RpcErr => { + let payload = frame + .payload + .ok_or_else(|| anyhow!("rpc.err missing payload"))?; + let err: ErrorPayload = serde_json::from_value(payload) + .context("dev: deserialize rpc.err")?; + return Err(match err.code.as_str() { + "module_dev_mode_off" => RegisterError::ModuleDevModeOff { slug: err.slug }, + "module_not_yours" => RegisterError::ModuleNotYours, + _ => RegisterError::Rejected { + code: err.code, + message: err.message, + }, + }); + } + _ => continue, } - // Other frames during handshake are ignored — see fn comment. } } From 7affe2b78e4e2247c19f5b3bbc41265f295adcb2 Mon Sep 17 00:00:00 2001 From: Sheng Kun Chang Date: Wed, 20 May 2026 03:25:16 +0800 Subject: [PATCH 2/3] =?UTF-8?q?refactor(dev/tunnel):=20/simplify=20pass=20?= =?UTF-8?q?=E2=80=94=20collapse=20map=5Ferr=20chain,=20add=20RegisterResul?= =?UTF-8?q?t=20alias?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three follow-ons from the simplify review on this PR (api-platform side was reviewed and came back clean, no fixes needed): - Add From for RegisterError so the `?` operator on serde_json calls converts to RegisterError directly. tunnel.rs::open drops eight `.map_err(RegisterError::from)?` adornments — the From impl + the existing From impl do the work. - Drop the double-wrap on serde_json::to_string (was `.map_err(|e| RegisterError::from(anyhow!(e)))?`, now just `?`). - Hoist `type RegisterResult = std::result::Result` so the two handshake functions read with the same Result shape as the rest of the file. - await_register_ack's "ignore non-corr_id frames" comment was inside the match against frame_type, but the corr_id check happens above it — move the comment to the function level where the policy lives. No behavior change. cargo test stays 64/64. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/commands/dev/tunnel.rs | 53 +++++++++++++++++++------------------- 1 file changed, 26 insertions(+), 27 deletions(-) diff --git a/src/commands/dev/tunnel.rs b/src/commands/dev/tunnel.rs index ec19d55..c2b6c2b 100644 --- a/src/commands/dev/tunnel.rs +++ b/src/commands/dev/tunnel.rs @@ -175,6 +175,17 @@ impl From for RegisterError { } } +impl From for RegisterError { + fn from(e: serde_json::Error) -> Self { + RegisterError::Transport(anyhow::Error::new(e)) + } +} + +/// Local Result alias — anyhow::Result is in scope at module level, so +/// every typed handshake return path would otherwise need the +/// fully-qualified std form. Keeps signatures readable. +type RegisterResult = std::result::Result; + /// Spawned-task handle. Drop or call [`shutdown`] to close the WSS. /// /// `service_token` is intentionally NOT held here — it lives on the @@ -201,8 +212,8 @@ pub(super) async fn open( wss_url: &str, ttok: &str, register: RegisterPayload<'_>, -) -> std::result::Result { - let url = with_token_param(wss_url, ttok).map_err(RegisterError::from)?; +) -> RegisterResult { + let url = with_token_param(wss_url, ttok)?; let config = WebSocketConfig { max_message_size: Some(MAX_INBOUND_FRAME_BYTES), max_frame_size: Some(MAX_INBOUND_FRAME_BYTES), @@ -211,33 +222,24 @@ pub(super) async fn open( let (ws_stream, _resp) = tokio_tungstenite::connect_async_with_config(&url, Some(config), false) .await - .with_context(|| format!("dev: connect WSS {wss_url}")) - .map_err(RegisterError::from)?; + .with_context(|| format!("dev: connect WSS {wss_url}"))?; let (mut sink, mut stream) = ws_stream.split(); let register_frame = Frame::new( FrameType::Register, - Some( - serde_json::to_value(®ister) - .context("dev: serialize register payload") - .map_err(RegisterError::from)?, - ), + Some(serde_json::to_value(®ister).context("dev: serialize register payload")?), ); let register_id = register_frame.id.clone(); - sink.send(Message::Text( - serde_json::to_string(®ister_frame).map_err(|e| RegisterError::from(anyhow!(e)))?, - )) - .await - .context("dev: send register frame") - .map_err(RegisterError::from)?; + sink.send(Message::Text(serde_json::to_string(®ister_frame)?)) + .await + .context("dev: send register frame")?; let ack = tokio::time::timeout( Duration::from_secs(10), await_register_ack(&mut sink, &mut stream, ®ister_id), ) .await - .context("dev: register_ack timeout") - .map_err(RegisterError::from)??; + .context("dev: register_ack timeout")??; let shutdown = Arc::new(Notify::new()); tokio::spawn(run_tunnel_loop(sink, stream, shutdown.clone())); @@ -262,11 +264,14 @@ fn with_token_param(wss_url: &str, ttok: &str) -> Result { /// `register_id` arrives. Server-side ping/binary/text-without-ack are /// tolerated — the server may send heartbeat or status frames before /// the ack lands. Wrapped in `tokio::time::timeout` by the caller. +// The server replies on the register's corr_id either with +// register_ack (success) or rpc.err (rejection). Anything else during +// handshake (server-initiated ping, future frame types) gets ignored. async fn await_register_ack( sink: &mut WsSink, stream: &mut WsReader, register_id: &str, -) -> std::result::Result { +) -> RegisterResult { loop { let next = stream .next() @@ -285,10 +290,6 @@ async fn await_register_ack( }; let frame: Frame = serde_json::from_str(&text) .with_context(|| format!("dev: parse server frame: {text}"))?; - // The server replies on the register's corr_id either with - // register_ack (success) or rpc.err (rejection). Anything else - // during handshake (server-initiated ping, future frame types) - // gets ignored. if frame.corr_id.as_deref() != Some(register_id) { continue; } @@ -297,16 +298,14 @@ async fn await_register_ack( let payload = frame .payload .ok_or_else(|| anyhow!("register_ack missing payload"))?; - return serde_json::from_value(payload) - .context("dev: deserialize register_ack") - .map_err(RegisterError::from); + return Ok(serde_json::from_value(payload).context("dev: deserialize register_ack")?); } FrameType::RpcErr => { let payload = frame .payload .ok_or_else(|| anyhow!("rpc.err missing payload"))?; - let err: ErrorPayload = serde_json::from_value(payload) - .context("dev: deserialize rpc.err")?; + let err: ErrorPayload = + serde_json::from_value(payload).context("dev: deserialize rpc.err")?; return Err(match err.code.as_str() { "module_dev_mode_off" => RegisterError::ModuleDevModeOff { slug: err.slug }, "module_not_yours" => RegisterError::ModuleNotYours, From dbe4caaf3338083f3d572a12eaaee157e2997c9e Mon Sep 17 00:00:00 2001 From: Sheng Kun Chang Date: Wed, 20 May 2026 03:31:16 +0800 Subject: [PATCH 3/3] =?UTF-8?q?chore:=20cargo=20fmt=20=E2=80=94=20collapse?= =?UTF-8?q?=20multi-Some=20match=20arm=20into=20a=20single=20pattern?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI's `cargo fmt --check` rejected the localhost/loopback match in dev_console_url because the multi-arm `Some("...") | Some("...") | ...` form was wrapped onto a long single line under rustfmt's normalize. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/commands/dev/mod.rs | 4 +--- src/commands/dev/tunnel.rs | 4 +++- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/commands/dev/mod.rs b/src/commands/dev/mod.rs index ea8bf0c..df391cb 100644 --- a/src/commands/dev/mod.rs +++ b/src/commands/dev/mod.rs @@ -269,9 +269,7 @@ fn dev_console_url(dispatch_base: &str, slug: Option<&str>) -> String { let parsed = url::Url::parse(dispatch_base).ok(); let host = parsed.as_ref().and_then(|u| u.host_str()); let base = match host { - Some("localhost") | Some("127.0.0.1") | Some("::1") => { - "http://localhost:3001".to_string() - } + Some("localhost") | Some("127.0.0.1") | Some("::1") => "http://localhost:3001".to_string(), Some(h) if h.starts_with("api.") => { format!("https://apps.{}", &h["api.".len()..]) } diff --git a/src/commands/dev/tunnel.rs b/src/commands/dev/tunnel.rs index c2b6c2b..0d42280 100644 --- a/src/commands/dev/tunnel.rs +++ b/src/commands/dev/tunnel.rs @@ -298,7 +298,9 @@ async fn await_register_ack( let payload = frame .payload .ok_or_else(|| anyhow!("register_ack missing payload"))?; - return Ok(serde_json::from_value(payload).context("dev: deserialize register_ack")?); + return Ok( + serde_json::from_value(payload).context("dev: deserialize register_ack")? + ); } FrameType::RpcErr => { let payload = frame