Add JWT expiry warning and account-stream heartbeat/staleness detection (fixes #225)

crates/standx-cli/src/commands/maker/mod.rs

@@ -34,7 +34,7 @@ use model::is_order_rejection;
 use model::{is_maker_order, position_for_symbol, MakerExit, PendingPlace};
 #[cfg(test)]
 use notify::webhook_body;
-use notify::{MakerNotifier, PositionChange, RiskNotice};
+use notify::{token_expiry_level, MakerNotifier, PositionChange, RiskNotice, TokenExpiryLevel};
 use pipeline::{CycleRequest, CycleState};
 use recovery::{
     cancel_maker_orders_with_retry, order_response_reconnect_available, reconcile_ledger_snapshot,
@@ -71,6 +71,14 @@ pub fn panic_webhook_body(format: AlertWebhookFormat, text: &str) -> serde_json:
 /// Env var gating live order placement. The live path ships code-complete but
 /// locked until it has been supervised-tested against production.
 const LIVE_MAKER_ENV: &str = "STANDX_ENABLE_LIVE_MAKER";
+
+/// Warn when the JWT has under 2h of life left; escalate under 15m. Token
+/// lifetime caps run duration (there is no renewal endpoint), so an operator
+/// needs lead time to re-authenticate before the bot halts.
+const TOKEN_EXPIRY_WARN_SECS: i64 = 2 * 60 * 60;
+const TOKEN_EXPIRY_CRITICAL_SECS: i64 = 15 * 60;
+/// Throttle disk/env credential reloads for the expiry check.
+const TOKEN_EXPIRY_CHECK_INTERVAL: Duration = Duration::from_secs(60);
 pub async fn handle_maker(
     command: MakerCommands,
     output_format: OutputFormat,

crates/standx-cli/src/commands/maker/notify.rs

@@ -245,6 +245,35 @@ impl MakerNotifier {
     }
 }
 
+/// Severity band for the maker's JWT remaining-lifetime monitor.
+///
+/// Token lifetime is a hard cap on run duration: once the JWT expires every
+/// reconnect and REST call is rejected and the bot halts. There is no renewal
+/// endpoint in the codebase, so the best we can do is warn early enough for an
+/// operator to re-authenticate.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub(super) enum TokenExpiryLevel {
+    Ok,
+    Warning,
+    Critical,
+}
+
+/// Classify remaining token lifetime into a severity band. Thresholds are in
+/// seconds; `critical_below` should be <= `warn_below`.
+pub(super) fn token_expiry_level(
+    remaining_secs: i64,
+    warn_below: i64,
+    critical_below: i64,
+) -> TokenExpiryLevel {
+    if remaining_secs <= critical_below {
+        TokenExpiryLevel::Critical
+    } else if remaining_secs <= warn_below {
+        TokenExpiryLevel::Warning
+    } else {
+        TokenExpiryLevel::Ok
+    }
+}
+
 pub(super) struct RiskNotice<'a> {
     pub(super) kind: &'a str,
     pub(super) severity: &'a str,
@@ -295,3 +324,40 @@ mod tests {
         );
     }
 }
+
+#[cfg(test)]
+mod token_expiry_tests {
+    use super::*;
+
+    #[test]
+    fn classifies_remaining_lifetime_into_bands() {
+        let warn = 2 * 60 * 60; // 2h
+        let critical = 15 * 60; // 15m
+        assert_eq!(
+            token_expiry_level(6 * 60 * 60, warn, critical),
+            TokenExpiryLevel::Ok
+        );
+        assert_eq!(
+            token_expiry_level(warn, warn, critical),
+            TokenExpiryLevel::Warning
+        );
+        assert_eq!(
+            token_expiry_level(30 * 60, warn, critical),
+            TokenExpiryLevel::Warning
+        );
+        assert_eq!(
+            token_expiry_level(critical, warn, critical),
+            TokenExpiryLevel::Critical
+        );
+        assert_eq!(
+            token_expiry_level(0, warn, critical),
+            TokenExpiryLevel::Critical
+        );
+    }
+
+    #[test]
+    fn severity_bands_are_ordered_for_escalation_checks() {
+        assert!(TokenExpiryLevel::Critical > TokenExpiryLevel::Warning);
+        assert!(TokenExpiryLevel::Warning > TokenExpiryLevel::Ok);
+    }
+}

crates/standx-cli/src/commands/maker/runtime.rs

@@ -903,11 +903,60 @@ pub(super) async fn run_maker(
     let mut order_response_reconnect_attempts_used = 0_u32;
     let mut account_stream_reconnect_attempts_used = 0_u32;
     let mut account_position_mismatch: Option<f64> = None;
+    // JWT expiry monitor: highest severity already alerted, plus a throttle so
+    // credentials are only reloaded from disk/env periodically.
+    let mut token_expiry_alerted = TokenExpiryLevel::Ok;
+    let mut last_token_expiry_check: Option<std::time::Instant> = None;
     let mut runtime_state = MakerState::starting();
     runtime_state.reduce(MakerEvent::StartupReady);
 
     let exit = 'main: loop {
         if args.live {
+            // JWT expiry monitor. There is no renewal endpoint, so we can only
+            // warn: escalate through Warning → Critical and alert once per band.
+            let due = last_token_expiry_check
+                .map(|last| last.elapsed() >= TOKEN_EXPIRY_CHECK_INTERVAL)
+                .unwrap_or(true);
+            if due {
+                last_token_expiry_check = Some(std::time::Instant::now());
+                if let Ok(creds) = Credentials::load() {
+                    let remaining = creds.remaining_seconds();
+                    let level = token_expiry_level(
+                        remaining,
+                        TOKEN_EXPIRY_WARN_SECS,
+                        TOKEN_EXPIRY_CRITICAL_SECS,
+                    );
+                    if level > token_expiry_alerted {
+                        token_expiry_alerted = level;
+                        let (severity, event) = match level {
+                            TokenExpiryLevel::Critical => ("critical", "token_expiry_critical"),
+                            _ => ("warning", "token_expiry_warning"),
+                        };
+                        let minutes = remaining / 60;
+                        let message = format!(
+                            "auth token expires in ~{minutes}m ({}); no renewal endpoint — run 'standx auth login' before it lapses or the bot will halt",
+                            creds.expires_at_string()
+                        );
+                        notifier
+                            .risk(
+                                RiskNotice {
+                                    kind: "token_expiry",
+                                    severity,
+                                    event,
+                                    message: &message,
+                                    symbol: &symbol,
+                                    cycle,
+                                    position_before: None,
+                                    position_after: None,
+                                    expected: None,
+                                    observed: None,
+                                },
+                                false,
+                            )
+                            .await;
+                    }
+                }
+            }
             if let Some(health) = account_stream_health
                 .as_ref()
                 .filter(|health| !health.is_healthy())

crates/standx-sdk/src/account_stream.rs

@@ -15,6 +15,13 @@ use tokio_tungstenite::{connect_async, tungstenite::Message};
 
 const DEFAULT_ACCOUNT_STREAM_URL: &str = "wss://perps.standx.com/ws-stream/v1";
 const ACCOUNT_STREAM_ROTATE_AFTER: Duration = Duration::from_secs(23 * 60 * 60 + 50 * 60);
+/// How often we send a client-side ping to keep the connection observably
+/// alive and to elicit a pong (which resets the idle deadline).
+const ACCOUNT_STREAM_PING_INTERVAL: Duration = Duration::from_secs(30);
+/// If no inbound frame (data, ping, or pong) arrives within this window the
+/// connection is treated as stale — this catches half-open TCP sessions where
+/// the socket never errors but the peer has silently gone away.
+const ACCOUNT_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(90);
 
 fn string_or_number<'de, D>(deserializer: D) -> std::result::Result<String, D::Error>
 where
@@ -160,6 +167,8 @@ pub struct AccountStream {
     url: String,
     token: String,
     epoch: u64,
+    ping_interval: Duration,
+    idle_timeout: Duration,
 }
 
 impl AccountStream {
@@ -175,6 +184,8 @@ impl AccountStream {
             url: DEFAULT_ACCOUNT_STREAM_URL.to_string(),
             token: credentials.token,
             epoch,
+            ping_interval: ACCOUNT_STREAM_PING_INTERVAL,
+            idle_timeout: ACCOUNT_STREAM_IDLE_TIMEOUT,
         })
     }
 
@@ -184,9 +195,18 @@ impl AccountStream {
             url: url.into(),
             token: token.into(),
             epoch,
+            ping_interval: ACCOUNT_STREAM_PING_INTERVAL,
+            idle_timeout: ACCOUNT_STREAM_IDLE_TIMEOUT,
         }
     }
 
+    #[cfg(test)]
+    fn with_heartbeat(mut self, ping_interval: Duration, idle_timeout: Duration) -> Self {
+        self.ping_interval = ping_interval;
+        self.idle_timeout = idle_timeout;
+        self
+    }
+
     pub async fn connect(
         &self,
         channels: &[AccountChannel],
@@ -264,10 +284,22 @@ impl AccountStream {
         let health = AccountStreamHealth::new(self.epoch);
         let task_health = health.clone();
         let epoch = self.epoch;
+        let ping_interval = self.ping_interval;
+        let idle_timeout = self.idle_timeout;
         let _ = tx.send(AccountEvent::Connected { epoch }).await;
         let handle = tokio::spawn(async move {
             let rotation = tokio::time::sleep(ACCOUNT_STREAM_ROTATE_AFTER);
             tokio::pin!(rotation);
+            // First ping fires after one interval (not immediately), so the
+            // handshake and any buffered startup frames are handled first.
+            let mut ping = tokio::time::interval_at(
+                tokio::time::Instant::now() + ping_interval,
+                ping_interval,
+            );
+            ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+            // Read-side idle deadline; reset on every inbound frame.
+            let idle = tokio::time::sleep(idle_timeout);
+            tokio::pin!(idle);
             loop {
                 let message = tokio::select! {
                     _ = &mut rotation => {
@@ -276,8 +308,29 @@ impl AccountStream {
                         let _ = tx.send(AccountEvent::Disconnected { reason }).await;
                         return;
                     }
+                    _ = ping.tick() => {
+                        if let Err(error) = write.send(Message::Ping(Vec::new().into())).await {
+                            let reason = format!("account-stream ping failed: {error}");
+                            task_health.mark_unhealthy(reason.clone());
+                            let _ = tx.send(AccountEvent::Error { reason }).await;
+                            return;
+                        }
+                        continue;
+                    }
+                    _ = &mut idle => {
+                        let reason = format!(
+                            "account stream idle for {}s (no ping/pong/data; connection likely half-open)",
+                            idle_timeout.as_secs()
+                        );
+                        task_health.mark_unhealthy(reason.clone());
+                        let _ = tx.send(AccountEvent::Disconnected { reason }).await;
+                        return;
+                    }
                     message = read.next() => message,
                 };
+                // Any inbound frame proves the peer is alive; extend the deadline.
+                idle.as_mut()
+                    .reset(tokio::time::Instant::now() + idle_timeout);
                 let Some(message) = message else {
                     let reason = "account stream ended without a close frame".to_string();
                     task_health.mark_unhealthy(reason.clone());
@@ -479,4 +532,49 @@ mod tests {
         .unwrap();
         handle.await.unwrap();
     }
+
+    #[tokio::test]
+    async fn idle_connection_is_marked_unhealthy() {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("ws://{}", listener.local_addr().unwrap());
+        // Server authenticates then stays connected but silent forever,
+        // simulating a half-open connection that never errors or closes.
+        let server = tokio::spawn(async move {
+            let (socket, _) = listener.accept().await.unwrap();
+            let mut websocket = accept_async(socket).await.unwrap();
+            let _auth = websocket.next().await.unwrap().unwrap();
+            websocket
+                .send(Message::Text(
+                    r#"{"seq":1,"channel":"auth","data":{"code":200,"msg":"success"}}"#.into(),
+                ))
+                .await
+                .unwrap();
+            // Absorb the client's pings without replying, then hold the socket
+            // open so the client can only detect death via the idle timeout.
+            while let Some(Ok(_)) = websocket.next().await {}
+        });
+
+        // Idle timeout well below the ping interval so the idle deadline, not a
+        // ping write failure, is what trips health.
+        let stream = AccountStream::with_url_and_token(url, "jwt", 7)
+            .with_heartbeat(Duration::from_secs(10), Duration::from_millis(200));
+        let (mut events, health, handle) = stream.connect(&[AccountChannel::Order]).await.unwrap();
+        assert_eq!(
+            events.recv().await,
+            Some(AccountEvent::Connected { epoch: 7 })
+        );
+
+        let disconnect = tokio::time::timeout(Duration::from_secs(2), events.recv())
+            .await
+            .expect("idle timeout should surface a disconnect");
+        match disconnect {
+            Some(AccountEvent::Disconnected { reason }) => {
+                assert!(reason.contains("idle"), "unexpected reason: {reason}");
+            }
+            other => panic!("expected idle Disconnected, got {other:?}"),
+        }
+        assert!(!health.is_healthy());
+        handle.await.unwrap();
+        server.abort();
+    }
 }