diff --git a/Cargo.lock b/Cargo.lock index ffc583b..88e7e3a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -134,6 +134,16 @@ dependencies = [ "syn", ] +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys 0.52.0", +] + [[package]] name = "fallible-iterator" version = "0.3.0" @@ -952,6 +962,16 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "slab" version = "0.4.12" @@ -1092,6 +1112,7 @@ dependencies = [ "libc", "mio", "pin-project-lite", + "signal-hook-registry", "socket2", "tokio-macros", "windows-sys 0.61.2", diff --git a/crates/executor/Cargo.toml b/crates/executor/Cargo.toml index 0b571e2..ed4ce9f 100644 --- a/crates/executor/Cargo.toml +++ b/crates/executor/Cargo.toml @@ -13,5 +13,5 @@ rusqlite = { version = "0.32", features = ["bundled"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" sha2 = "0.10" -tokio = { version = "1.40", features = ["macros", "rt-multi-thread", "time", "sync"] } +tokio = { version = "1.40", features = ["macros", "rt-multi-thread", "time", "sync", "signal"] } tokio-tungstenite = { version = "0.24", features = ["rustls-tls-webpki-roots"] } diff --git a/crates/executor/src/bitget.rs b/crates/executor/src/bitget.rs index e679283..f0da366 100644 --- a/crates/executor/src/bitget.rs +++ b/crates/executor/src/bitget.rs @@ -10,7 +10,9 @@ use std::collections::HashMap; use std::time::{SystemTime, UNIX_EPOCH}; use crate::config::ExecutorConfig; -use crate::types::{MarketUpdate, OrderRecord, PrivateWsUpdate}; +use crate::types::{ + AccountSnapshotUpdate, MarketUpdate, OrderRecord, PositionRecord, PrivateWsUpdate, +}; type HmacSha256 = Hmac; @@ -420,11 +422,48 @@ pub fn parse_public_ws_message(text: &str) -> Result> { })) } -pub fn parse_private_ws_message(text: &str) -> Result { +pub fn parse_private_ws_message(text: &str, cfg: &ExecutorConfig) -> Result { if text == "pong" { return Ok(PrivateWsUpdate::default()); } let value: Value = serde_json::from_str(text)?; + let mut update = PrivateWsUpdate::default(); + + // Connection-control events arrive WITHOUT an arg.channel: the login ack + // ({"event":"login","code":0,...}) confirms auth; any {"event":"error",...} + // or a non-zero login code is an auth/subscribe failure. The WS loop gates + // private-state readiness on the ack and emits websocket_auth_failed on the + // error, so the parser must surface both. Data messages (orders/positions/ + // account) have no top-level "event" and fall through to the channel dispatch. + // ponytail: Bitget serializes `code` as a NUMBER on the login ack + // ({"event":"login","code":0}) but as a STRING on REST/error bodies; accept + // both so a numeric ack isn't mis-read as a missing/failed code. + if let Some(event) = value.get("event").and_then(Value::as_str) { + if event == "login" { + if ws_code(&value) == "0" { + update.login_ack = true; + } else { + update.auth_error = Some(format!( + "login code {}: {}", + ws_code(&value), + value.get("msg").and_then(Value::as_str).unwrap_or("") + )); + } + return Ok(update); + } + if event == "error" { + update.auth_error = Some(format!( + "code {}: {}", + ws_code(&value), + value.get("msg").and_then(Value::as_str).unwrap_or("") + )); + return Ok(update); + } + // Other control events (e.g. subscribe ack) carry no private data; return + // the empty update so the read loop skips it. + return Ok(update); + } + let channel = value .pointer("/arg/channel") .and_then(Value::as_str) @@ -434,7 +473,6 @@ pub fn parse_private_ws_message(text: &str) -> Result { .and_then(Value::as_array) .cloned() .unwrap_or_default(); - let mut update = PrivateWsUpdate::default(); if channel == "orders" { for row in data { @@ -445,7 +483,7 @@ pub fn parse_private_ws_message(text: &str) -> Result { exchange_order_id: Some(order_id), client_oid, intent_id: None, - symbol: "ETH/USDT:USDT".to_string(), + symbol: resolve_symbol(cfg, &str_field(&row, "instId")), side: str_field(&row, "side"), action: str_field(&row, "tradeSide"), order_type: str_field(&row, "orderType"), @@ -469,11 +507,74 @@ pub fn parse_private_ws_message(text: &str) -> Result { last_error: None, }); } + } else if channel == "positions" { + for row in data { + let size = row.get("total").and_then(num_or_str_f64).unwrap_or(0.0); + if size.abs() < 1e-12 { + continue; // no open position + } + let entry_price = row + .get("openPriceAvg") + .and_then(num_or_str_f64) + .unwrap_or(0.0); + update.positions.push(PositionRecord { + symbol: resolve_symbol(cfg, &str_field(&row, "instId")), + side: str_field(&row, "holdSide"), + notional: size.abs() * entry_price, + entry_price, + unrealized_pnl: row + .get("unrealizedPL") + .and_then(num_or_str_f64) + .unwrap_or(0.0), + ownership: "system".to_string(), + opened_at: None, + adopted_at: None, + source_intent_id: None, + raw_json: row.to_string(), + }); + } + } else if channel == "account" { + // ponytail: a single account message can carry multiple marginCoin rows; + // the configured marginCoin is the one the risk gate cares about. Take the + // first matching row, fall back to the first row if none matches. equity is + // accountEquity on REST and either equity/accountEquity on WS — accept both. + let row = data + .iter() + .find(|r| str_field(r, "marginCoin") == cfg.margin_coin) + .or_else(|| data.first()); + if let Some(row) = row { + let equity = row + .get("accountEquity") + .and_then(num_or_str_f64) + .or_else(|| row.get("equity").and_then(num_or_str_f64)); + let available = row.get("available").and_then(num_or_str_f64); + if let (Some(equity), Some(available)) = (equity, available) { + update.account = Some(AccountSnapshotUpdate { + equity, + available_margin: available, + unrealized_pnl: row + .get("unrealizedPL") + .and_then(num_or_str_f64) + .unwrap_or(0.0), + }); + } + } } Ok(update) } +/// Map a private-WS message instId (e.g. "ETHUSDT") to the configured display +/// symbol (e.g. "ETH/USDT:USDT"). A foreign instId falls through to the raw +/// string — never silently relabel a non-configured symbol as ETH. +fn resolve_symbol(cfg: &ExecutorConfig, inst_id: &str) -> String { + if inst_id == cfg.bitget_symbol { + cfg.symbol.clone() + } else { + inst_id.to_string() + } +} + fn str_field(value: &Value, key: &str) -> String { value .get(key) @@ -482,6 +583,18 @@ fn str_field(value: &Value, key: &str) -> String { .to_string() } +/// Read a connection-control `code` field as a string regardless of whether +/// Bitget serialized it as a number (login ack: `"code":0`) or a string +/// (error/REST bodies: `"code":"30001"`). Empty string when absent. +fn ws_code(value: &Value) -> String { + match value.get("code") { + Some(Value::String(s)) => s.clone(), + Some(Value::Number(n)) => n.to_string(), + Some(other) => other.to_string(), + None => String::new(), + } +} + fn parse_f64(value: Option<&Value>, name: &str) -> Result { value .and_then(Value::as_str) @@ -490,18 +603,54 @@ fn parse_f64(value: Option<&Value>, name: &str) -> Result { .with_context(|| format!("parse {name}")) } -pub async fn verify_public_ws_connects(cfg: &ExecutorConfig) -> Result<()> { - cfg.validate_demo_only()?; - let (mut socket, _) = tokio_tungstenite::connect_async(&cfg.public_ws_url).await?; - use futures_util::{SinkExt, StreamExt}; - let msg = serde_json::json!({ +/// Private WS login payload: signs `GET /user/verify` with the per-connection +/// timestamp. Pure so the one-shot verify and the long-running WS loop build the +/// identical login (and the test pins the signature path). +pub fn private_login_message(cfg: &ExecutorConfig, timestamp: &str) -> serde_json::Value { + serde_json::json!({ + "op": "login", + "args": [{ + "apiKey": cfg.secrets.api_key, + "passphrase": cfg.secrets.passphrase, + "timestamp": timestamp, + "sign": websocket_sign(&cfg.secrets.api_secret, timestamp) + }] + }) +} + +/// Private WS subscribe payload for orders/positions/account (orders+positions +/// key on instId, account keys on coin; "default" carries every symbol for the +/// product type). Pure so the one-shot verify and the long-running WS loop build +/// the identical subscription. Sent after login. +pub fn private_subscribe_message(cfg: &ExecutorConfig) -> serde_json::Value { + serde_json::json!({ + "op": "subscribe", + "args": [ + {"instType": cfg.product_type, "channel": "orders", "instId": "default"}, + {"instType": cfg.product_type, "channel": "positions", "instId": "default"}, + {"instType": cfg.product_type, "channel": "account", "coin": "default"} + ] + }) +} + +/// Public books5 subscribe payload for the configured symbol. Pure so the one-shot +/// verify and the long-running WS loop build the identical subscription. +pub fn public_books5_subscribe_message(cfg: &ExecutorConfig) -> serde_json::Value { + serde_json::json!({ "op": "subscribe", "args": [{ "instType": cfg.product_type, "channel": "books5", "instId": cfg.bitget_symbol }] - }); + }) +} + +pub async fn verify_public_ws_connects(cfg: &ExecutorConfig) -> Result<()> { + cfg.validate_demo_only()?; + let (mut socket, _) = tokio_tungstenite::connect_async(&cfg.public_ws_url).await?; + use futures_util::{SinkExt, StreamExt}; + let msg = public_books5_subscribe_message(cfg); socket .send(tokio_tungstenite::tungstenite::Message::Text( msg.to_string(), @@ -522,15 +671,7 @@ pub async fn verify_private_ws_connects(cfg: &ExecutorConfig) -> Result<()> { let (mut socket, _) = tokio_tungstenite::connect_async(&cfg.private_ws_url).await?; use futures_util::{SinkExt, StreamExt}; let timestamp = now_seconds(); - let login = serde_json::json!({ - "op": "login", - "args": [{ - "apiKey": cfg.secrets.api_key, - "passphrase": cfg.secrets.passphrase, - "timestamp": timestamp, - "sign": websocket_sign(&cfg.secrets.api_secret, ×tamp) - }] - }); + let login = private_login_message(cfg, ×tamp); socket .send(tokio_tungstenite::tungstenite::Message::Text( login.to_string(), @@ -540,8 +681,17 @@ pub async fn verify_private_ws_connects(cfg: &ExecutorConfig) -> Result<()> { .await? .ok_or_else(|| anyhow::anyhow!("private websocket closed"))??; let text = msg.into_text()?; - if text.contains("\"event\":\"error\"") || !text.contains("\"login\"") { - bail!("private websocket login failed: {text}"); + // ponytail: reuse the daemon's parser instead of a string-contains probe so + // the one-shot verify and the long-running loop judge login the same way — + // including Bitget's NUMERIC login code ({"event":"login","code":0}), which + // a "contains \"login\"" check accepts but a future stricter check could + // diverge on. Surfaces the real code+msg on failure instead of raw text. + let update = parse_private_ws_message(&text, cfg)?; + if let Some(detail) = update.auth_error { + bail!("private websocket login failed: {detail}"); + } + if !update.login_ack { + bail!("private websocket login not acked: {text}"); } Ok(()) } @@ -613,11 +763,146 @@ mod tests { }] }"#; - let update = parse_private_ws_message(raw).unwrap(); + let cfg = ExecutorConfig::demo_for_tests(); + let update = parse_private_ws_message(raw, &cfg).unwrap(); assert_eq!(update.orders.len(), 1); assert_eq!(update.orders[0].client_oid, "client-1"); assert_eq!(update.orders[0].status, "live"); + // de-hardcoded symbol: instId ETHUSDT resolves to the configured display symbol. + assert_eq!(update.orders[0].symbol, "ETH/USDT:USDT"); + } + + #[test] + fn private_subscribe_message_covers_orders_positions_account() { + let cfg = ExecutorConfig::demo_for_tests(); + let msg = private_subscribe_message(&cfg); + let text = msg.to_string(); + + assert!(text.contains("\"op\":\"subscribe\"")); + assert!(text.contains("\"channel\":\"orders\"")); + assert!(text.contains("\"channel\":\"positions\"")); + assert!(text.contains("\"channel\":\"account\"")); + assert!(text.contains("\"instType\":\"USDT-FUTURES\"")); + } + + #[test] + fn parses_private_position_message() { + let raw = r#"{ + "action":"snapshot", + "arg":{"instType":"USDT-FUTURES","instId":"default","channel":"positions"}, + "data":[{ + "instId":"ETHUSDT", + "holdSide":"long", + "total":"0.05", + "openPriceAvg":"3000", + "unrealizedPL":"1.5" + }] + }"#; + + let cfg = ExecutorConfig::demo_for_tests(); + let update = parse_private_ws_message(raw, &cfg).unwrap(); + + assert_eq!(update.positions.len(), 1); + let pos = &update.positions[0]; + assert_eq!(pos.symbol, "ETH/USDT:USDT"); + assert_eq!(pos.side, "long"); + assert!((pos.entry_price - 3000.0).abs() < 1e-9); + assert!((pos.notional - 150.0).abs() < 1e-9); + assert!((pos.unrealized_pnl - 1.5).abs() < 1e-9); + } + + #[test] + fn parses_private_account_message() { + let raw = r#"{ + "action":"snapshot", + "arg":{"instType":"USDT-FUTURES","instId":"default","channel":"account"}, + "data":[{ + "marginCoin":"USDT", + "available":"500", + "accountEquity":"1000", + "unrealizedPL":"-2" + }] + }"#; + + let cfg = ExecutorConfig::demo_for_tests(); + let update = parse_private_ws_message(raw, &cfg).unwrap(); + + let acct = update.account.expect("account snapshot parsed"); + assert!((acct.equity - 1000.0).abs() < 1e-9); + assert!((acct.available_margin - 500.0).abs() < 1e-9); + assert!((acct.unrealized_pnl - (-2.0)).abs() < 1e-9); + } + + #[test] + fn parses_private_login_ack() { + // Bitget's real login ack serializes code as a NUMBER: {"event":"login","code":0}. + // The WS loop waits for this before treating private state as ready, so the + // parser must set login_ack for the numeric form (and the string form too). + let cfg = ExecutorConfig::demo_for_tests(); + let update = parse_private_ws_message(r#"{"event":"login","code":0}"#, &cfg).unwrap(); + assert!(update.login_ack); + assert!(update.auth_error.is_none()); + // defensive: a string "0" must also ack (some clients/gateways vary) + let update_s = parse_private_ws_message(r#"{"event":"login","code":"0"}"#, &cfg).unwrap(); + assert!(update_s.login_ack); + } + + #[test] + fn parses_private_login_failure_code_as_auth_error() { + // A login response with a non-zero code (numeric, like the ack). Must + // surface as auth_error so the loop emits websocket_auth_failed and stays + // not-ready; login_ack must NOT be set. + let cfg = ExecutorConfig::demo_for_tests(); + let update = parse_private_ws_message( + r#"{"event":"login","code":30001,"msg":"sign invalid"}"#, + &cfg, + ) + .unwrap(); + assert!(!update.login_ack); + let err = update.auth_error.expect("auth error parsed"); + assert!(err.contains("30001")); + assert!(err.contains("sign invalid")); + } + + #[test] + fn parses_private_error_event_as_auth_error() { + // A generic {"event":"error",...} (string code, as on REST/error bodies). + let cfg = ExecutorConfig::demo_for_tests(); + let update = parse_private_ws_message( + r#"{"event":"error","code":"30005","msg":"login timeout"}"#, + &cfg, + ) + .unwrap(); + assert!(!update.login_ack); + let err = update.auth_error.expect("auth error parsed"); + assert!(err.contains("30005")); + assert!(err.contains("login timeout")); + } + + #[test] + fn public_books5_subscribe_payload_targets_demo_symbol() { + let cfg = ExecutorConfig::demo_for_tests(); + let msg = public_books5_subscribe_message(&cfg); + let text = msg.to_string(); + + assert!(text.contains("\"op\":\"subscribe\"")); + assert!(text.contains("\"channel\":\"books5\"")); + assert!(text.contains("\"instId\":\"ETHUSDT\"")); + assert!(text.contains("\"instType\":\"USDT-FUTURES\"")); + } + + #[test] + fn private_login_payload_uses_websocket_signature() { + let cfg = ExecutorConfig::demo_for_tests(); + let msg = private_login_message(&cfg, "1538054050"); + let text = msg.to_string(); + + assert!(text.contains("\"op\":\"login\"")); + assert!(text.contains("\"apiKey\":\"key\"")); + assert!(text.contains("\"passphrase\":\"pass\"")); + assert!(text.contains("\"timestamp\":\"1538054050\"")); + assert!(text.contains(&websocket_sign("secret", "1538054050"))); } #[test] diff --git a/crates/executor/src/config.rs b/crates/executor/src/config.rs index 351ec73..9a5dae9 100644 --- a/crates/executor/src/config.rs +++ b/crates/executor/src/config.rs @@ -84,7 +84,7 @@ impl ExecutorConfig { pub fn validate_demo_only(&self) -> Result<()> { if self.mode != TradingMode::Demo { - bail!("third milestone executor only supports Bitget demo mode"); + bail!("prodigy executor only supports Bitget demo mode"); } if self.secrets.api_key.trim().is_empty() || self.secrets.api_secret.trim().is_empty() @@ -149,7 +149,7 @@ mod tests { } #[test] - fn live_mode_is_rejected_for_third_milestone() { + fn live_mode_is_rejected() { let cfg = ExecutorConfig { mode: TradingMode::Live, ..ExecutorConfig::demo_for_tests() diff --git a/crates/executor/src/daemon.rs b/crates/executor/src/daemon.rs new file mode 100644 index 0000000..2959d0a --- /dev/null +++ b/crates/executor/src/daemon.rs @@ -0,0 +1,1229 @@ +use anyhow::Result; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use crate::config::ExecutorConfig; + +#[derive(Debug, Clone, Default)] +pub struct DaemonOptions { + pub max_runtime: Option, +} + +/// Cross-task "please run a REST reconcile on the next tick" signal. The WS loops +/// set it after a successful (re)connect (spec: reconcile after WS reconnect); the +/// daemon main loop takes it on each tick. Arc-shared so the WS tasks and the main +/// loop can reach it without channel plumbing. Pure accessors so the set/take +/// semantics are unit-testable without a network round-trip. +#[derive(Debug, Clone, Default)] +pub struct ReconcileSignal { + pending: Arc, +} + +impl ReconcileSignal { + pub fn new() -> Self { + Self::default() + } + /// WS loop calls this after a successful (re)connect. + pub fn request(&self) { + self.pending.store(true, Ordering::SeqCst); + } + /// Main loop calls this each tick; returns true iff a reconcile was requested, + /// and clears the flag (so one reconnect triggers exactly one reconcile). + pub fn take(&self) -> bool { + self.pending.swap(false, Ordering::SeqCst) + } +} + +/// Cross-task "private account state is ready" flag. The private WS loop sets it +/// true after a successful (re)connect (login + subscribe sent without error) and +/// false on disconnect; the daemon's intent loop reads it when risk-checking a new +/// OPEN (spec: refuse new opening exposure when private state is not ready). Close / +/// reduce / cancel bypass it (they're risk-reducing and use REST). Arc-shared so the +/// WS task and the main loop can reach it without channel plumbing. Pure accessors so +/// the set/get semantics are unit-testable without a network round-trip. +/// +/// ponytail: "ready" means the private WS is connected, authenticated, and subscribed +/// (login + subscribe sends succeeded) — NOT "we have received the first data push". An +/// idle demo account may never push an orders/positions message, so gating on first +/// data would deadlock all opens on an idle account. Connection-level readiness is the +/// correct, non-deadlocking signal. +#[derive(Debug, Clone, Default)] +pub struct PrivateStateReady { + ready: Arc, +} + +impl PrivateStateReady { + pub fn new() -> Self { + Self::default() + } + /// Private WS loop: set true after a successful (re)connect, false on disconnect. + pub fn set(&self, value: bool) { + self.ready.store(value, Ordering::SeqCst); + } + /// Intent loop: is private state ready (private WS connected + subscribed)? + pub fn is_ready(&self) -> bool { + self.ready.load(Ordering::SeqCst) + } +} + +pub async fn run_daemon(cfg: ExecutorConfig, options: DaemonOptions) -> Result<()> { + cfg.validate_demo_only()?; + let conn = rusqlite::Connection::open(&cfg.db_path)?; + // ponytail: WAL persists in the DB file header (idempotent; matches src/prodigy/db.py). + // M4 has Python writing intents, the daemon R/W, the private WS writing, and Telegram + // reading — all concurrent. WAL lets those readers/writers proceed in parallel instead of + // serializing on a single rollback journal; busy_timeout makes them wait out SQLITE_BUSY. + if let Err(err) = conn.pragma_update(None, "journal_mode", "wal") { + // WAL is the assumed journal mode for concurrent Python/daemon/WS/Telegram + // access. A failure isn't fatal (busy_timeout still serializes writers), but + // surface it as a warning event so the operator knows the DB isn't WAL — M4 + // claims WAL-compatible behavior and shouldn't swallow this silently. + crate::db::write_event( + &conn, + "warning", + "daemon", + &format!("WAL journal_mode setup failed: {err}"), + "{}", + )?; + } + conn.busy_timeout(Duration::from_secs(5))?; + let rest = crate::bitget::BitgetRestClient::new(cfg.clone())?; + + if cfg.test_reset_demo_state { + crate::db::write_event( + &conn, + "warning", + "daemon", + "test reset requested in daemon mode", + "{}", + )?; + } + + rest.set_leverage(cfg.leverage).await.map_err(|e| { + anyhow::anyhow!( + "set-leverage failed (configured {}x): {e} — refusing to trade at unknown leverage", + cfg.leverage + ) + })?; + // Startup reconcile BEFORE processing intents: repair any local/exchange + // divergence left over from a prior run so the first tick starts from + // exchange-truth. (Daemon mode does NOT call reset_demo_symbol_state here — + // reset is the one-shot's job; daemon only logs the warning above.) + crate::reconcile::reconcile_once( + &conn, + &rest, + "daemon-startup", + !cfg.test_reset_demo_state, + cfg.telegram_bot_token.as_deref(), + cfg.telegram_chat_id.as_deref(), + ) + .await?; + crate::db::write_event(&conn, "info", "daemon", "daemon started", "{}")?; + + let market_cache = Arc::new(tokio::sync::Mutex::new( + crate::executor::MarketCache::default(), + )); + let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); + + // Cross-task signal: a WS (re)connect sets it, the main loop consumes it on + // each tick to run a REST reconcile immediately (spec: reconcile after WS + // reconnect) instead of waiting for the periodic interval. + let reconcile_signal = ReconcileSignal::new(); + // Cross-task signal: the private WS loop sets this true after a successful + // (re)connect (login + subscribe sent) and false on disconnect; the intent + // loop reads it to gate NEW opening exposure (spec: refuse new opening + // exposure when private state is not ready). Only the PRIVATE WS owns it — + // public WS and telegram do not. + let private_ready = PrivateStateReady::new(); + + let mut public_task = tokio::spawn(run_public_ws_loop( + cfg.clone(), + market_cache.clone(), + shutdown_rx.clone(), + reconcile_signal.clone(), + )); + let mut private_task = tokio::spawn(run_private_ws_loop( + cfg.clone(), + shutdown_rx.clone(), + reconcile_signal.clone(), + private_ready.clone(), + )); + let mut telegram_task = tokio::spawn(run_telegram_query_loop(cfg.clone(), shutdown_rx.clone())); + + // ponytail: monotonic Instant for the bounded-runtime check — immune to + // wall-clock skew that SystemTime would inject mid-loop. + let started = tokio::time::Instant::now(); + let mut poll = tokio::time::interval(Duration::from_millis(250)); + let mut last_reconcile_ms = crate::bitget::now_ms().parse::().unwrap_or(0); + + loop { + tokio::select! { + _ = shutdown_signal() => { + log_shutdown_requested(&conn)?; + break; + } + _ = poll.tick() => { + if options.max_runtime.is_some_and(|max| started.elapsed() >= max) { + crate::db::write_event( + &conn, + "info", + "daemon", + "bounded daemon runtime elapsed", + "{}", + )?; + break; + } + let now_ms = crate::bitget::now_ms().parse::().unwrap_or(0); + // Reconnect-triggered reconcile takes priority over the periodic + // interval: a WS (re)connect set the signal, so run the SAME + // error-isolated reconcile immediately and reset the interval + // clock (so the next periodic pass is a full interval away). + if reconcile_signal.take() { + if let Err(err) = crate::reconcile::reconcile_once( + &conn, + &rest, + "daemon-ws-reconnect", + !cfg.test_reset_demo_state, + cfg.telegram_bot_token.as_deref(), + cfg.telegram_chat_id.as_deref(), + ) + .await + { + crate::db::write_event( + &conn, + "warning", + "reconcile", + &format!("reconcile failed: {err}"), + "{}", + )?; + } + last_reconcile_ms = now_ms; + } + if should_run_reconcile(now_ms, last_reconcile_ms, cfg.reconcile_interval_secs) { + // Periodic reconcile errors are LOGGED, not propagated: a single + // flaky REST pass must not bring the daemon down (the next tick + // retries). Same isolation as the intent-loop below. + if let Err(err) = crate::reconcile::reconcile_once( + &conn, + &rest, + "daemon-periodic", + !cfg.test_reset_demo_state, + cfg.telegram_bot_token.as_deref(), + cfg.telegram_chat_id.as_deref(), + ) + .await + { + crate::db::write_event( + &conn, + "warning", + "reconcile", + &format!("reconcile failed: {err}"), + "{}", + )?; + } + last_reconcile_ms = now_ms; + } + + let mut local_cache = { + let cache = market_cache.lock().await; + cache.clone() + }; + // Error isolation: a stale-market or REST failure here (common in + // the first few hundred ms before the public WS delivers, or on a + // transient network blip) is logged as an event and the loop + // continues — the daemon must not crash on a loop-iteration error. + // The next tick retries once the WS cache is fresh. + if let Err(err) = crate::executor::process_pending_intents_once( + &conn, + &cfg, + &rest, + &mut local_cache, + private_ready.is_ready(), + ) + .await + { + crate::db::write_event( + &conn, + "error", + "intent_loop", + &format!("intent loop failed: {err}"), + "{}", + )?; + } + } + } + } + + // Shutdown ordering: signal WS loops via the watch channel, then give them + // a short grace window to observe it and return cooperatively (flush/close). + // abort() is the hard fallback so the process still exits within the + // bounded test runtime if a task is stuck mid-await on a socket read. + let _ = shutdown_tx.send(true); + let _ = tokio::time::timeout( + Duration::from_millis(200), + futures_util::future::join3(&mut public_task, &mut private_task, &mut telegram_task), + ) + .await; + public_task.abort(); + private_task.abort(); + telegram_task.abort(); + crate::db::write_event(&conn, "info", "daemon", "daemon stopped", "{}")?; + Ok(()) +} + +/// Pure gate for the periodic reconcile cadence: true once `interval_secs` +/// have elapsed since the last reconcile. Saturating subtraction keeps it +/// safe against clock-skew-driven `now < last` orderings. +pub fn should_run_reconcile(now_ms: i64, last_reconcile_ms: i64, interval_secs: u64) -> bool { + now_ms.saturating_sub(last_reconcile_ms) >= (interval_secs as i64) * 1000 +} + +/// Shared shutdown-requested event write for both the ctrl_c (SIGINT) and +/// SIGTERM arms of `run_daemon`'s main select. Same body either way so a +/// production signal (SIGTERM from `kill`/systemd/container stop) gets the +/// identical graceful-shutdown audit trail as an interactive Ctrl+C. +fn log_shutdown_requested(conn: &rusqlite::Connection) -> Result<()> { + crate::db::write_event(conn, "info", "daemon", "shutdown requested", "{}") +} + +/// Wait for SIGTERM. Production daemons receive SIGTERM from `kill`, systemd +/// and container stop; without this handler the default disposition kills the +/// process hard — no "shutdown requested" event, no task abort, no +/// "daemon stopped". Unix-only (Windows has no SIGTERM equivalent here). +#[cfg(unix)] +async fn sigterm() { + use tokio::signal::unix::{signal, SignalKind}; + let mut s = signal(SignalKind::terminate()).expect("install SIGTERM handler"); + s.recv().await; +} + +/// Neutral shutdown-signal future for the main select: resolves on either +/// ctrl_c (SIGINT) or, on Unix, SIGTERM. Wrapping both in one future lets +/// `tokio::select!` take a single branch (the macro rejects `#[cfg]` on its +/// own arms). Same shutdown path either way — SIGTERM is the signal +/// production daemons actually receive. +#[cfg(unix)] +async fn shutdown_signal() { + tokio::select! { + _ = tokio::signal::ctrl_c() => {} + _ = sigterm() => {} + } +} + +#[cfg(not(unix))] +async fn shutdown_signal() { + let _ = tokio::signal::ctrl_c().await; +} + +/// Pure glue: stamp a parsed public-WS books5 update into the shared market cache +/// with the local-received time. Wraps `MarketCache::update_at` so the WS loop and +/// its test share one call site, and the freshness window stays LOCAL-received. +pub fn apply_public_market_update( + cache: &mut crate::executor::MarketCache, + update: crate::types::MarketUpdate, + local_received_at_ms: i64, +) { + cache.update_at(update, local_received_at_ms); +} + +/// Pure glue: write a parsed private-WS update (orders/fills/positions) to SQLite +/// via the existing db upsert/insert helpers. Re-applying the same update is safe +/// (upserts by PK, fills insert-or-ignore). Wraps the three writes so the WS loop +/// and its test share one call site. Apply errors are surfaced (the loop logs them +/// and never crashes the daemon). +pub fn apply_private_ws_update( + conn: &rusqlite::Connection, + update: crate::types::PrivateWsUpdate, +) -> Result<()> { + for order in update.orders { + // ponytail: the private WS is a fast cache, not the source of truth for + // order identity. Only refresh orders we already placed locally — never + // insert a new row (that would steal identity from the REST execution + // path: intent_id stays NULL and system_net_base ignores a real system + // position) and never adopt a manual/imported order before REST reconcile + // detects it (local_oids would then contain it and reconcile would skip + // imported/manual detection). REST reconcile remains the authority for + // discovering new orders; the executor owns intent_id. + if crate::db::order_exists(conn, &order.client_oid)? { + crate::db::refresh_order_from_ws(conn, &order)?; + } + } + for fill in update.fills { + crate::db::insert_fill(conn, &fill)?; + } + for position in update.positions { + // ponytail: WS positions refresh market fields only — REST reconcile owns + // ownership classification (refresh_position_from_ws preserves + // ownership/adopted_at/source_intent_id on conflict; upsert_position would + // clobber them). Spec: if WS and REST disagree, REST wins. + crate::db::refresh_position_from_ws(conn, &position)?; + } + // The private-WS `account` event is parsed (PrivateWsUpdate.account) but NOT + // persisted: equity_snapshots is REST reconcile's authoritative table + // (telegram /pnl and /risk read it). Writing WS-derived equity there would + // let the display disagree with the last REST reconcile. REST wins. + Ok(()) +} + +/// Long-running public-WS loop: connect, subscribe books5 for the configured +/// symbol, parse every incoming books5 message and refresh the shared market +/// cache with a LOCAL timestamp. Disconnects (or shutdown) reset the loop. +/// Spawned by `run_daemon`; demo-only invariant enforced at entry. +pub async fn run_public_ws_loop( + cfg: ExecutorConfig, + market_cache: Arc>, + shutdown: tokio::sync::watch::Receiver, + reconcile_signal: ReconcileSignal, +) -> Result<()> { + use futures_util::{SinkExt, StreamExt}; + use tokio_tungstenite::tungstenite::Message; + + cfg.validate_demo_only()?; + let mut shutdown = shutdown; + loop { + if *shutdown.borrow() { + return Ok(()); + } + match tokio_tungstenite::connect_async(&cfg.public_ws_url).await { + Ok((mut socket, _)) => { + // ponytail: a failed subscribe send is recoverable — log and + // break to the outer reconnect loop (1s backoff) instead of + // killing the WS task with `?`. A dead WS loop is undetectable + // (the main loop never monitors the JoinHandle), so we keep it + // alive via reconnect until shutdown. + if let Err(err) = socket + .send(Message::Text( + crate::bitget::public_books5_subscribe_message(&cfg).to_string(), + )) + .await + { + eprintln!("public ws subscribe send failed: {err}; reconnecting"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + // (Re)connect succeeded and we're subscribed: request a REST + // reconcile on the next main-loop tick (spec: reconcile after + // WS reconnect) so any books5 / cache gap is repaired promptly. + reconcile_signal.request(); + 'inner: loop { + tokio::select! { + _ = shutdown.changed() => { + if *shutdown.borrow() { + return Ok(()); + } + } + msg = socket.next() => { + let Some(msg) = msg else { break 'inner; }; + let Ok(msg) = msg else { break 'inner; }; + let Ok(text) = msg.into_text() else { continue; }; + match crate::bitget::parse_public_ws_message(&text) { + Ok(Some(update)) => { + let now_ms = crate::bitget::now_ms().parse::().unwrap_or(0); + let mut cache = market_cache.lock().await; + apply_public_market_update(&mut cache, update, now_ms); + } + Ok(None) => {} + Err(err) => { + eprintln!("public ws parse error: {err}"); + } + } + } + } + } + eprintln!("public ws socket closed; reconnecting"); + } + Err(err) => { + eprintln!("public ws disconnected: {err}"); + } + } + // ponytail: fixed 1s reconnect backoff on EVERY reconnect path + // (connect failure AND mid-stream socket close/error); exponential + // backoff if disconnects become frequent (a flapping link would hammer + // the endpoint and risk a temporary IP block). Shutdown exits return + // above before reaching here, so they aren't delayed. + tokio::time::sleep(Duration::from_secs(1)).await; + } +} + +/// Optional read-only Telegram polling loop (M4). Runs ONLY when both +/// `telegram_bot_token` and `telegram_chat_id` are configured — otherwise it +/// returns immediately, since Telegram is not an execution dependency. It +/// long-polls `getUpdates`, filters to the operator's chat_id only (other +/// chats get no reply), and answers recognized `/status /positions /orders +/// /pnl /risk` commands via `telegram_query::query_response`. `/stop /resume +/// /close_all` are refused by the query layer (M4 forbids remote trading +/// control). +/// +/// Error isolation: EVERY network/parse/SQLite error here is logged and the +/// loop continues — a flaky getUpdates or a transient DB lock must NEVER crash +/// the daemon. Uses the same hoisted-shutdown `select!` pattern as the WS +/// loops so the 1s throttle never blocks a shutdown. Open its own SQLite +/// connection per update batch (rusqlite Connection is not Sync). +pub async fn run_telegram_query_loop( + cfg: ExecutorConfig, + shutdown: tokio::sync::watch::Receiver, +) -> Result<()> { + let (Some(token), Some(chat_id)) = + (cfg.telegram_bot_token.clone(), cfg.telegram_chat_id.clone()) + else { + return Ok(()); + }; + let client = reqwest::Client::new(); + let mut offset: i64 = 0; + let mut shutdown = shutdown; + loop { + if *shutdown.borrow() { + return Ok(()); + } + let get_url = format!("https://api.telegram.org/bot{token}/getUpdates"); + // ponytail: a failed long-poll (network blip, 5xx) is logged and we + // back off via the shutdown-aware sleep below — never propagated, the + // daemon must not die on a Telegram outage. + let response = client + .get(&get_url) + .query(&[ + ("timeout", "10".to_string()), + ("offset", offset.to_string()), + ]) + .send() + .await; + if let Ok(resp) = response { + if let Ok(value) = resp.json::().await { + if let Some(updates) = value.get("result").and_then(serde_json::Value::as_array) { + for update in updates { + if let Some(id) = + update.get("update_id").and_then(serde_json::Value::as_i64) + { + offset = id + 1; + } + let message = update.get("message").unwrap_or(&serde_json::Value::Null); + let chat = message + .get("chat") + .and_then(|c| c.get("id")) + .and_then(serde_json::Value::as_i64) + .map(|v| v.to_string()); + // Security gate: only the operator's configured chat + // gets any reply; messages from any other chat are + // ignored (offset still advances so they aren't redelivered). + if chat.as_deref() != Some(chat_id.as_str()) { + continue; + } + let Some(text) = message.get("text").and_then(serde_json::Value::as_str) + else { + continue; + }; + match rusqlite::Connection::open(&cfg.db_path) { + Ok(conn) => { + if let Err(err) = conn + .busy_timeout(std::time::Duration::from_secs(5)) + .map_err(anyhow::Error::from) + { + eprintln!("telegram sqlite busy_timeout error: {err}"); + continue; + } + let reply = match crate::telegram_query::query_response(&conn, text) + { + Ok(reply) => reply, + Err(err) => { + eprintln!("telegram query error: {err}"); + continue; + } + }; + if let Some(reply) = reply { + let send_url = + format!("https://api.telegram.org/bot{token}/sendMessage"); + // ponytail: best-effort send — a failed + // sendMessage is dropped on the floor; the + // operator can re-issue the command. + let _ = client + .post(send_url) + .form(&[ + ("chat_id", chat_id.as_str()), + ("text", reply.as_str()), + ]) + .send() + .await; + } + } + Err(err) => eprintln!("telegram sqlite open error: {err}"), + } + } + } + } + } + // ponytail: hoisted shutdown-aware throttle — same pattern as the WS + // loops, so a shutdown observed mid-throttle returns promptly instead + // of sleeping the full 1s (the Task 4 backoff-bug fix applied here too). + tokio::select! { + _ = shutdown.changed() => { + if *shutdown.borrow() { + return Ok(()); + } + } + _ = tokio::time::sleep(Duration::from_secs(1)) => {} + } + } +} + +/// Record a `websocket_auth_failed` event in SQLite and fire the demo Telegram +/// notification for it (`notify::should_send_telegram` gates demo delivery). +/// Best-effort throughout: a sqlite/telegram failure only logs to stderr — the +/// private-WS loop must not die on a logging path (a dead WS loop is +/// undetectable). Shared by the PRE-ready login-failure path and the POST-ready +/// mid-session auth-error path so the operator sees the failure either way. +/// ponytail: extracted so both auth-failure paths emit the identical event + +/// notification instead of drifting apart. +async fn emit_websocket_auth_failed(cfg: &ExecutorConfig, detail: &str) { + match rusqlite::Connection::open(&cfg.db_path) { + Ok(conn) => { + let _ = conn.busy_timeout(Duration::from_secs(5)); + if let Err(err) = crate::db::write_event( + &conn, + "warning", + "private_ws", + &format!("websocket auth failed: {detail}"), + "{}", + ) { + eprintln!("websocket_auth_failed event write error: {err}"); + } + } + Err(err) => eprintln!("websocket_auth_failed event sqlite open error: {err}"), + } + crate::notify::send_telegram( + cfg.telegram_bot_token.as_deref(), + cfg.telegram_chat_id.as_deref(), + "websocket_auth_failed", + &format!("private websocket login failed: {detail}"), + ) + .await + .ok(); +} + +/// Long-running private-WS loop: connect, send the per-connection login (signed +/// `GET /user/verify`), then parse every incoming message and apply orders/fills/ +/// positions to SQLite via `apply_private_ws_update`. A parse error or a SQLite +/// apply error is logged and the loop continues — the daemon must not crash on a +/// bad update. Disconnects (or shutdown) reset the loop. Spawned by `run_daemon` +/// (Task 7 wires it); demo-only invariant enforced at entry. +pub async fn run_private_ws_loop( + cfg: ExecutorConfig, + shutdown: tokio::sync::watch::Receiver, + reconcile_signal: ReconcileSignal, + private_ready: PrivateStateReady, +) -> Result<()> { + use futures_util::{SinkExt, StreamExt}; + use tokio_tungstenite::tungstenite::Message; + + cfg.validate_demo_only()?; + let mut shutdown = shutdown; + loop { + if *shutdown.borrow() { + return Ok(()); + } + // ponytail: flip not-ready at the top of every outer iteration — a + // disconnect that broke the inner read-loop lands here on its next + // iteration, so readiness reflects the disconnect BEFORE we attempt to + // reconnect. set(true) below only fires after the subscribe send + // succeeds, so a connect/login/subscribe failure stays not-ready. + private_ready.set(false); + match tokio_tungstenite::connect_async(&cfg.private_ws_url).await { + Ok((mut socket, _)) => { + let timestamp = crate::bitget::now_seconds(); + // ponytail: failed login/subscribe sends are recoverable — log + // and break to the outer reconnect loop (1s backoff) instead of + // killing the WS task with `?`. A dead WS loop is undetectable, + // so we keep it alive via reconnect until shutdown. + if let Err(err) = socket + .send(Message::Text( + crate::bitget::private_login_message(&cfg, ×tamp).to_string(), + )) + .await + { + eprintln!("private ws login send failed: {err}; reconnecting"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + // Wait for the login ack BEFORE subscribing or marking ready. A + // failed login (bad signature / expired timestamp / banned key) + // arrives as {"event":"error",...} or a non-zero login code; the + // parser surfaces it as auth_error. Only {"event":"login","code":"0"} + // makes private state ready. Spec: emit websocket_auth_failed on + // auth failure and keep new opens gated out until a successful + // reconnect re-acks. Bounded by a 10s ack deadline so a dead + // socket can't hang the loop. + let ack_deadline = std::time::Instant::now() + Duration::from_secs(10); + let mut acked = false; + let mut auth_failure: Option = None; + while std::time::Instant::now() < ack_deadline { + let msg = tokio::select! { + _ = shutdown.changed() => { + if *shutdown.borrow() { + return Ok(()); + } + continue; + } + msg = socket.next() => match msg { + Some(Ok(m)) => m, + Some(Err(_)) | None => break, + }, + }; + let Ok(text) = msg.into_text() else { + continue; + }; + match crate::bitget::parse_private_ws_message(&text, &cfg) { + Ok(u) if u.auth_error.is_some() => { + auth_failure = u.auth_error; + break; + } + Ok(u) if u.login_ack => { + acked = true; + break; + } + Ok(_) => continue, + Err(err) => { + eprintln!("private ws parse error pre-ack: {err}"); + continue; + } + } + } + if let Some(detail) = auth_failure { + // Auth failed: surface it (event + demo Telegram) and stay + // not-ready. A corrected key/timestamp re-acks on the next + // connect. ponytail: sleep BEFORE continue so a bad key / + // consistently-rejected login can't tight-loop reconnect and + // flood the events table / Telegram — same 1s backoff every + // other reconnect path takes below (the bare `continue` here + // previously skipped the outer-loop sleep at the bottom). + emit_websocket_auth_failed(&cfg, &detail).await; + eprintln!("private ws login rejected: {detail}; reconnecting"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + if !acked { + eprintln!("private ws login ack timed out; reconnecting"); + // ponytail: same 1s backoff — an ack that never comes (dead + // socket, network blackhole) must not tight-loop the 10s ack + // wait + immediate reconnect. + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + // Subscribe to orders/positions/account now that login is acked. + if let Err(err) = socket + .send(Message::Text( + crate::bitget::private_subscribe_message(&cfg).to_string(), + )) + .await + { + eprintln!("private ws subscribe send failed: {err}; reconnecting"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + // (Re)connect succeeded, login was ACKED, and we're subscribed: + // request a REST reconcile on the next main-loop tick (spec: + // reconcile after WS reconnect) so any orders/fills/positions gap + // is repaired, and mark private state READY so new opens are no + // longer gated out (spec: refuse new opening exposure when + // private state is not ready). Readiness is connection-level: we + // wait for the login ack but NOT for a first data push — an idle + // demo account may never push one, and gating on it would + // deadlock all opens on an idle account. + private_ready.set(true); + reconcile_signal.request(); + loop { + tokio::select! { + _ = shutdown.changed() => { + if *shutdown.borrow() { + return Ok(()); + } + } + msg = socket.next() => { + let Some(msg) = msg else { break; }; + let Ok(msg) = msg else { break; }; + let Ok(text) = msg.into_text() else { continue; }; + let update = match crate::bitget::parse_private_ws_message(&text, &cfg) { + Ok(update) => update, + Err(err) => { + eprintln!("private ws parse error: {err}"); + continue; + } + }; + // A post-ready {"event":"error",...} (subscribe args + // rejected, key revoked mid-session, etc.) sets + // auth_error but carries no orders/fills/positions/ + // account. Without this check the empty-skip below + // would drop it and private state would stay READY + // while the private channel is actually broken — new + // opens would proceed on a dead feed. Surface it, + // gate opens back out, and break to the outer + // reconnect loop (1s backoff + fresh login). + if let Some(detail) = &update.auth_error { + private_ready.set(false); + emit_websocket_auth_failed(&cfg, detail).await; + eprintln!( + "private ws session error after ready: {detail}; reconnecting" + ); + break; + } + if update.orders.is_empty() + && update.fills.is_empty() + && update.positions.is_empty() + && update.account.is_none() + { + continue; + } + match rusqlite::Connection::open(&cfg.db_path) { + Ok(conn) => { + // ponytail: busy_timeout failure is benign + // (sqlite3_busy_timeout essentially never errors); + // skip this batch instead of killing the WS task. + if let Err(err) = conn.busy_timeout(std::time::Duration::from_secs(5)) { + eprintln!("private ws sqlite busy_timeout error: {err}"); + continue; + } + if let Err(err) = apply_private_ws_update(&conn, update) { + eprintln!("private ws sqlite apply error: {err}"); + } + } + Err(err) => eprintln!("private ws sqlite open error: {err}"), + } + } + } + } + // The socket died (close/err broke the read loop) — flip readiness + // OFF immediately so a 250ms intent-loop tick in the ~1s reconnect + // window can't slip a new OPEN through the gate on stale readiness. + // (set(false) at the top of the next outer iteration would be ~1s late.) + private_ready.set(false); + eprintln!("private ws socket closed; reconnecting"); + } + Err(err) => { + // Connect itself failed: readiness is already false (set at the top + // of this outer iteration), but set again for clarity/symmetry. + private_ready.set(false); + eprintln!("private ws disconnected: {err}"); + } + } + // ponytail: fixed 1s reconnect backoff on EVERY reconnect path — hoisted + // to the outer loop so BOTH connect failure and mid-stream close back off + // (the Task 4 bug only slept the connect-err arm, letting a flapping + // mid-stream disconnect hammer the endpoint). Add exponential backoff if + // disconnects become frequent. Shutdown exits return above before here. + tokio::time::sleep(Duration::from_secs(1)).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::{ExecutorConfig, TradingMode}; + + fn test_conn() -> rusqlite::Connection { + let conn = rusqlite::Connection::open_in_memory().unwrap(); + conn.execute_batch(include_str!("../../../schema/001_initial.sql")) + .unwrap(); + conn.execute_batch(include_str!("../../../schema/002_execution.sql")) + .unwrap(); + conn + } + + #[test] + fn daemon_options_default_runs_forever() { + let options = DaemonOptions::default(); + + assert!(options.max_runtime.is_none()); + } + + #[test] + fn should_run_reconcile_when_interval_elapsed() { + assert!(should_run_reconcile(10_000, 0, 10)); + assert!(!should_run_reconcile(9_999, 0, 10)); + } + + #[test] + fn reconcile_signal_request_then_take() { + let sig = ReconcileSignal::new(); + assert!(!sig.take(), "fresh signal should not request reconcile"); + sig.request(); + assert!(sig.take(), "after request, take is true"); + assert!( + !sig.take(), + "take clears the flag (one reconnect → one reconcile)" + ); + } + + #[test] + fn reconcile_signal_shared_between_clones() { + let sig = ReconcileSignal::new(); + let clone = sig.clone(); + clone.request(); // WS task sets via its clone + assert!( + sig.take(), + "main loop sees the request through its own handle" + ); + } + + #[test] + fn private_state_ready_default_false_then_set() { + let r = PrivateStateReady::new(); + assert!(!r.is_ready(), "fresh signal should be not-ready"); + r.set(true); + assert!(r.is_ready()); + r.set(false); + assert!(!r.is_ready()); + } + + #[test] + fn private_state_ready_shared_between_clones() { + let r = PrivateStateReady::new(); + let clone = r.clone(); + clone.set(true); // private WS task sets via its clone + assert!( + r.is_ready(), + "main loop sees the readiness through its own handle" + ); + } + + #[test] + fn daemon_allows_bounded_runtime_for_tests() { + let options = DaemonOptions { + max_runtime: Some(std::time::Duration::from_millis(5)), + }; + + assert_eq!( + options.max_runtime.unwrap(), + std::time::Duration::from_millis(5) + ); + } + + #[tokio::test] + async fn emit_websocket_auth_failed_writes_event_to_db() { + // Regression for Fix 1/2: the shared helper both auth-failure paths use + // must persist a websocket_auth_failed event so the operator sees a + // broken private channel. It opens its own connection from cfg.db_path, + // so point it at a temp file initialized with the schema. + let dir = std::env::temp_dir(); + let db_path = dir.join(format!( + "prodigy-test-emit-auth-{}.sqlite", + std::process::id() + )); + { + let conn = rusqlite::Connection::open(&db_path).unwrap(); + conn.execute_batch(include_str!("../../../schema/001_initial.sql")) + .unwrap(); + conn.execute_batch(include_str!("../../../schema/002_execution.sql")) + .unwrap(); + } + let cfg = ExecutorConfig { + db_path: db_path.clone(), + // telegram None → send_telegram is a no-op (helper stays best-effort). + telegram_bot_token: None, + telegram_chat_id: None, + ..ExecutorConfig::demo_for_tests() + }; + + emit_websocket_auth_failed(&cfg, "login code 30001: sign invalid").await; + + let conn = rusqlite::Connection::open(&db_path).unwrap(); + let row: (String, String, String) = conn + .query_row( + "select severity, component, message from events \ + where message like '%websocket auth failed%' order by created_at desc limit 1", + [], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + ) + .unwrap(); + assert_eq!(row.0, "warning"); + assert_eq!(row.1, "private_ws"); + assert!(row.2.contains("sign invalid")); + + let _ = std::fs::remove_file(&db_path); + } + + #[tokio::test] + async fn daemon_rejects_non_demo_mode_before_opening_db() { + let cfg = ExecutorConfig { + mode: TradingMode::Live, + ..ExecutorConfig::demo_for_tests() + }; + + let err = run_daemon( + cfg, + DaemonOptions { + max_runtime: Some(std::time::Duration::from_millis(1)), + }, + ) + .await + .unwrap_err(); + + assert!(err.to_string().contains("demo")); + } + + #[test] + fn public_ws_update_refreshes_market_cache() { + let mut cache = crate::executor::MarketCache::default(); + + apply_public_market_update( + &mut cache, + crate::types::MarketUpdate { + symbol: "ETHUSDT".to_string(), + best_bid: 100.0, + best_ask: 101.0, + exchange_ts_ms: 10, + }, + 1_000, + ); + + assert!(cache.latest_fresh(1_500, 3).is_some()); + } + + #[test] + fn private_ws_update_upserts_orders_and_positions() { + let conn = test_conn(); + // Fix-A: the private WS only refreshes orders we already placed locally — + // it no longer INSERTS orders. So pre-insert the order (as the executor + // would) with an intent_id, then assert the WS push refreshes it (status + // flips) and keeps the count at 1. The position assertion (count 1) still + // holds: refresh_position_from_ws inserts on first write. + // FK: orders.intent_id references trade_intents — seed the parent intent. + conn.execute( + "insert into trade_intents (intent_id, created_at, symbol, side, action, + target_notional, max_order_notional, status, source) + values ('intent-1','2026-07-01T00:00:00Z','ETHUSDT','long','open',100,100,'executed','t')", + [], + ) + .unwrap(); + crate::db::upsert_order( + &conn, + &crate::types::OrderRecord { + order_id: "local-order-1".to_string(), + exchange_order_id: None, + client_oid: "client-1".to_string(), + intent_id: Some("intent-1".to_string()), + symbol: "ETHUSDT".to_string(), + side: "buy".to_string(), + action: "open".to_string(), + order_type: "market".to_string(), + status: "submitted".to_string(), + price: Some(100.0), + size: 0.1, + filled_size: 0.0, + attempt: 1, + raw_json: "{}".to_string(), + last_error: None, + }, + ) + .unwrap(); + + let update = crate::types::PrivateWsUpdate { + login_ack: false, + auth_error: None, + orders: vec![crate::types::OrderRecord { + exchange_order_id: Some("ex-1".to_string()), + status: "filled".to_string(), + filled_size: 0.1, + intent_id: None, + ..crate::types::OrderRecord { + order_id: "local-order-1".to_string(), + exchange_order_id: None, + client_oid: "client-1".to_string(), + intent_id: None, + symbol: "ETHUSDT".to_string(), + side: "buy".to_string(), + action: "open".to_string(), + order_type: "market".to_string(), + status: "filled".to_string(), + price: Some(100.0), + size: 0.1, + filled_size: 0.1, + attempt: 1, + raw_json: "{}".to_string(), + last_error: None, + } + }], + positions: vec![crate::types::PositionRecord { + symbol: "ETH/USDT:USDT".to_string(), + side: "long".to_string(), + notional: 10.0, + entry_price: 100.0, + unrealized_pnl: 1.0, + ownership: "system".to_string(), + opened_at: Some("now".to_string()), + adopted_at: None, + source_intent_id: None, + raw_json: "{}".to_string(), + }], + fills: vec![], + account: None, + }; + + apply_private_ws_update(&conn, update).unwrap(); + + let order_count: i64 = conn + .query_row("select count(*) from orders", [], |r| r.get(0)) + .unwrap(); + let position_count: i64 = conn + .query_row("select count(*) from positions", [], |r| r.get(0)) + .unwrap(); + assert_eq!(order_count, 1); + assert_eq!(position_count, 1); + // The known order was REFRESHED: status flipped submitted -> filled, and + // the executor's intent_id was preserved (not clobbered to NULL by WS). + let (status, intent_id): (String, Option) = conn + .query_row( + "select status, intent_id from orders where client_oid = 'client-1'", + [], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .unwrap(); + assert_eq!(status, "filled"); + assert_eq!(intent_id.as_deref(), Some("intent-1")); + } + + #[test] + fn apply_private_ws_update_does_not_persist_account_snapshot() { + // Finding 2 regression: a private-WS `account` event must NOT be written + // into the REST-authoritative equity_snapshots table (reconcile owns it; + // telegram /pnl and /risk read it). WS is only a fast cache. The account + // event is still PARSED (PrivateWsUpdate.account is populated here), but + // not persisted. Spec: REST wins. + let conn = rusqlite::Connection::open_in_memory().unwrap(); + conn.execute_batch(include_str!("../../../schema/001_initial.sql")) + .unwrap(); + conn.execute_batch(include_str!("../../../schema/002_execution.sql")) + .unwrap(); + + let update = crate::types::PrivateWsUpdate { + account: Some(crate::types::AccountSnapshotUpdate { + equity: 999.0, + available_margin: 500.0, + unrealized_pnl: -2.0, + }), + ..Default::default() + }; + + apply_private_ws_update(&conn, update).unwrap(); + + let count: i64 = conn + .query_row("select count(*) from equity_snapshots", [], |r| r.get(0)) + .unwrap(); + assert_eq!(count, 0, "WS account events must not be persisted"); + } + + #[test] + fn apply_private_ws_update_skips_unknown_orders() { + // Fix-A regression: the private WS is a fast cache, NOT the source of truth + // for order identity. apply_private_ws_update must only refresh orders we + // already placed locally — it must neither steal identity from the REST + // execution path (inserting a system order with intent_id NULL so + // system_net_base drops it) nor adopt a manual/imported order before REST + // reconcile detects it (local_oids would then contain it and reconcile + // would skip imported/manual detection). + let conn = test_conn(); + // FK: orders.intent_id references trade_intents — seed the parent intent. + conn.execute( + "insert into trade_intents (intent_id, created_at, symbol, side, action, + target_notional, max_order_notional, status, source) + values ('intent-a','2026-07-01T00:00:00Z','ETH/USDT:USDT','long','open',100,100,'executed','t')", + [], + ) + .unwrap(); + // Pre-existing LOCAL system order (order-A) — known to us, with intent_id. + crate::db::upsert_order( + &conn, + &crate::types::OrderRecord { + order_id: "order-a".to_string(), + exchange_order_id: None, + client_oid: "client-a".to_string(), + intent_id: Some("intent-a".to_string()), + symbol: "ETH/USDT:USDT".to_string(), + side: "buy".to_string(), + action: "open".to_string(), + order_type: "limit".to_string(), + status: "submitted".to_string(), + price: Some(3000.0), + size: 0.05, + filled_size: 0.0, + attempt: 1, + raw_json: "{}".to_string(), + last_error: None, + }, + ) + .unwrap(); + + // WS update: order-A refresh (intent_id None, now filled) + order-B that has + // NO local row (a manual/unknown order the WS would otherwise adopt). + let update = crate::types::PrivateWsUpdate { + orders: vec![ + crate::types::OrderRecord { + exchange_order_id: Some("ex-a".to_string()), + status: "filled".to_string(), + filled_size: 0.05, + intent_id: None, + ..crate::types::OrderRecord { + order_id: "order-a".to_string(), + exchange_order_id: None, + client_oid: "client-a".to_string(), + intent_id: None, + symbol: "ETH/USDT:USDT".to_string(), + side: "buy".to_string(), + action: "open".to_string(), + order_type: "limit".to_string(), + status: "filled".to_string(), + price: Some(3000.0), + size: 0.05, + filled_size: 0.05, + attempt: 1, + raw_json: "{}".to_string(), + last_error: None, + } + }, + crate::types::OrderRecord { + order_id: "order-b".to_string(), + exchange_order_id: Some("ex-b".to_string()), + client_oid: "client-b".to_string(), + intent_id: None, + symbol: "ETH/USDT:USDT".to_string(), + side: "buy".to_string(), + action: "open".to_string(), + order_type: "market".to_string(), + status: "filled".to_string(), + price: Some(3000.0), + size: 0.02, + filled_size: 0.02, + attempt: 1, + raw_json: "{}".to_string(), + last_error: None, + }, + ], + ..Default::default() + }; + + apply_private_ws_update(&conn, update).unwrap(); + + // order-A was refreshed: status flipped to filled, intent_id preserved. + let a: (String, f64, Option) = conn + .query_row( + "select status, filled_size, intent_id from orders where client_oid = 'client-a'", + [], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + ) + .unwrap(); + assert_eq!(a.0, "filled"); + assert!((a.1 - 0.05).abs() < 1e-9); + assert_eq!( + a.2.as_deref(), + Some("intent-a"), + "known order's intent_id must be preserved" + ); + // order-B (unknown/manual) was NOT adopted: no row inserted. + let b: i64 = conn + .query_row( + "select count(*) from orders where client_oid = 'client-b'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(b, 0, "WS must not adopt an unknown/manual order"); + } +} diff --git a/crates/executor/src/db.rs b/crates/executor/src/db.rs index 7ce0041..0d0b3e2 100644 --- a/crates/executor/src/db.rs +++ b/crates/executor/src/db.rs @@ -97,6 +97,50 @@ pub fn upsert_order(conn: &Connection, order: &OrderRecord) -> Result<()> { Ok(()) } +/// True iff we already have a local order row for this client_oid. The private-WS +/// loop uses this to refresh ONLY orders we placed locally — it never inserts a +/// row the executor didn't write (REST reconcile owns order discovery). +pub fn order_exists(conn: &Connection, client_oid: &str) -> Result { + let n: i64 = conn.query_row( + "select count(*) from orders where client_oid = ?", + params![client_oid], + |r| r.get(0), + )?; + Ok(n > 0) +} + +/// Refresh an ALREADY-LOCAL order's live fields from a private-WS update WITHOUT +/// touching identity columns (order_id, client_oid, intent_id stay as the +/// executor wrote them). REST reconcile owns order discovery; the executor owns +/// intent_id. The WS only refreshes status/filled_size/price/raw_json (the fast +/// cache). Bare UPDATE by positional params (excluded.* is UPSERT-only) and it +/// matches zero rows when no local row exists — callers gate on order_exists. +pub fn refresh_order_from_ws(conn: &Connection, order: &OrderRecord) -> Result<()> { + conn.execute( + "update orders set + exchange_order_id = coalesce(?1, exchange_order_id), + status = ?2, + price = ?3, + size = ?4, + filled_size = ?5, + updated_at = datetime('now'), + attempt = ?6, + raw_json = ?7 + where client_oid = ?8", + params![ + order.exchange_order_id, + order.status, + order.price, + order.size, + order.filled_size, + order.attempt, + order.raw_json, + order.client_oid, + ], + )?; + Ok(()) +} + /// Remove the local positions row for a symbol. Called by reconcile when the /// exchange no longer lists a position for it (manual full-close): exchange state /// wins, so the local row must not keep reporting a position Bitget no longer @@ -142,6 +186,42 @@ pub fn upsert_position(conn: &Connection, position: &PositionRecord) -> Result<( Ok(()) } +/// Refresh a position's live market fields from a private-WS update WITHOUT +/// overwriting reconcile's authoritative ownership classification. REST +/// reconcile owns ownership/adopted_at/source_intent_id; the WS feed only +/// refreshes side/notional/entry_price/unrealized_pnl/raw_json (the fast cache). +/// If no local row exists yet, insert with WS-supplied ownership ("system") +/// pending the next reconcile reclassification. (Spec: if WS and REST disagree, +/// REST wins — so we never let a WS push downgrade an existing classification.) +pub fn refresh_position_from_ws(conn: &Connection, position: &PositionRecord) -> Result<()> { + conn.execute( + "insert into positions ( + symbol, side, notional, entry_price, unrealized_pnl, updated_at, + ownership, opened_at, adopted_at, source_intent_id, raw_json + ) values (?, ?, ?, ?, ?, datetime('now'), ?, ?, ?, ?, ?) + on conflict(symbol) do update set + side = excluded.side, + notional = excluded.notional, + entry_price = excluded.entry_price, + unrealized_pnl = excluded.unrealized_pnl, + updated_at = datetime('now'), + raw_json = excluded.raw_json", + params![ + position.symbol, + position.side, + position.notional, + position.entry_price, + position.unrealized_pnl, + position.ownership, + position.opened_at, + position.adopted_at, + position.source_intent_id, + position.raw_json, + ], + )?; + Ok(()) +} + /// client_oids of orders we already have locally (used to detect exchange orders /// we're missing → repair). Reconciliation inserts the missing ones. pub fn local_order_client_oids(conn: &Connection) -> Result> { @@ -1188,6 +1268,81 @@ mod tests { assert!((net - 0.05).abs() < 1e-9 && side == "long"); } + #[test] + fn refresh_position_from_ws_preserves_reconcile_ownership() { + // Finding 1 regression: a private-WS positions push must NOT clobber the + // ownership classification (imported/system), adopted_at, or source_intent_id + // that REST reconcile authoritatively wrote. WS only refreshes market-movement + // fields (side/notional/entry_price/unrealized_pnl/raw_json); reconcile's + // ownership stays put until the next reconcile reclassifies. Spec: REST wins. + use crate::types::PositionRecord; + let conn = memory_db(); + // Reconcile's authoritative write: position classified imported + adopted. + let reconciled = PositionRecord { + symbol: "ETH/USDT:USDT".to_string(), + side: "long".to_string(), + notional: 1000.0, + entry_price: 3000.0, + unrealized_pnl: 12.0, + ownership: "imported".to_string(), + opened_at: Some("2026-07-01T00:00:00Z".to_string()), + adopted_at: Some("2026-07-01T00:00:00Z".to_string()), + source_intent_id: Some("intent-9".to_string()), + raw_json: "{\"src\":\"reconcile\"}".to_string(), + }; + upsert_position(&conn, &reconciled).unwrap(); + + // Private-WS push for the same symbol: WS parser hardcodes ownership "system" + // and different market fields. + let ws = PositionRecord { + notional: 1100.0, + unrealized_pnl: 50.0, + ownership: "system".to_string(), + adopted_at: None, + source_intent_id: None, + raw_json: "{\"src\":\"ws\"}".to_string(), + ..reconciled.clone() + }; + refresh_position_from_ws(&conn, &ws).unwrap(); + + let row = conn + .query_row( + "select notional, unrealized_pnl, ownership, adopted_at, source_intent_id, raw_json + from positions where symbol='ETH/USDT:USDT'", + [], + |r| { + Ok(( + r.get::<_, f64>(0)?, + r.get::<_, f64>(1)?, + r.get::<_, String>(2)?, + r.get::<_, Option>(3)?, + r.get::<_, Option>(4)?, + r.get::<_, String>(5)?, + )) + }, + ) + .unwrap(); + // Market fields refreshed from WS. + assert!((row.0 - 1100.0).abs() < 1e-9, "notional should refresh"); + assert!((row.1 - 50.0).abs() < 1e-9, "unrealized_pnl should refresh"); + // Ownership authority preserved. + assert_eq!( + row.2, "imported", + "ownership must stay reconcile-authoritative" + ); + assert_eq!( + row.3.as_deref(), + Some("2026-07-01T00:00:00Z"), + "adopted_at must be preserved" + ); + assert_eq!( + row.4.as_deref(), + Some("intent-9"), + "source_intent_id must be preserved" + ); + assert_eq!(row.5, "{\"src\":\"ws\"}", "raw_json should refresh"); + } + #[test] fn set_order_filled_from_detail_partial_keeps_submitted() { // A partial cumulative fill (< ordered size) updates filled_size but does @@ -1221,4 +1376,97 @@ mod tests { assert_eq!(status, "submitted"); assert!((filled - 0.02).abs() < 1e-9); } + + #[test] + fn refresh_order_from_ws_preserves_intent_id_and_does_not_insert() { + // Fix-A regression: a private-WS order refresh must update the live fields + // (status/filled_size/exchange_order_id/price/...) but MUST preserve the + // identity columns the executor wrote (intent_id). The WS order parser sets + // intent_id None; if a refresh clobbered intent_id, system_net_base_for_symbol + // (filters intent_id is not null) would drop a real system position and + // reconcile would mis-classify it. Also it must never INSERT a row. + let conn = memory_db(); + conn.execute( + "insert into trade_intents (intent_id, created_at, symbol, side, action, + target_notional, max_order_notional, status, source) + values ('intent-7','2026-07-01T00:00:00Z','ETH/USDT:USDT','long','open',100,100,'executed','t')", + [], + ) + .unwrap(); + // Executor's authoritative write: system order, submitted, no fill yet. + let system = OrderRecord { + order_id: "order-1".to_string(), + exchange_order_id: None, + client_oid: "client-1".to_string(), + intent_id: Some("intent-7".to_string()), + symbol: "ETH/USDT:USDT".to_string(), + side: "buy".to_string(), + action: "open".to_string(), + order_type: "limit".to_string(), + status: "submitted".to_string(), + price: Some(3000.0), + size: 0.05, + filled_size: 0.0, + attempt: 1, + raw_json: "{\"src\":\"rest\"}".to_string(), + last_error: None, + }; + upsert_order(&conn, &system).unwrap(); + + // Private-WS push for the SAME client_oid with intent_id None (WS parser), + // status filled, fill size 0.05, exchange order id now known. + let ws = OrderRecord { + exchange_order_id: Some("ex-1".to_string()), + status: "filled".to_string(), + filled_size: 0.05, + intent_id: None, + raw_json: "{\"src\":\"ws\"}".to_string(), + ..system.clone() + }; + refresh_order_from_ws(&conn, &ws).unwrap(); + + let row = conn + .query_row( + "select status, filled_size, exchange_order_id, intent_id, raw_json + from orders where client_oid = 'client-1'", + [], + |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, f64>(1)?, + r.get::<_, Option>(2)?, + r.get::<_, Option>(3)?, + r.get::<_, String>(4)?, + )) + }, + ) + .unwrap(); + // Live fields refreshed from WS. + assert_eq!(row.0, "filled"); + assert!((row.1 - 0.05).abs() < 1e-9); + assert_eq!(row.2.as_deref(), Some("ex-1")); + assert_eq!(row.4, "{\"src\":\"ws\"}"); + // Identity preserved: the executor's intent_id is untouched. + assert_eq!( + row.3.as_deref(), + Some("intent-7"), + "intent_id must be preserved on WS refresh" + ); + + // CRITICAL: a refresh on a client_oid with NO local row must not INSERT. + let fresh = OrderRecord { + client_oid: "client-unknown".to_string(), + order_id: "order-2".to_string(), + ..ws.clone() + }; + refresh_order_from_ws(&conn, &fresh).unwrap(); + let unknown: i64 = conn + .query_row( + "select count(*) from orders where client_oid = 'client-unknown'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(unknown, 0, "refresh must not insert an unknown order"); + } } diff --git a/crates/executor/src/executor.rs b/crates/executor/src/executor.rs index 3ad56a9..0fa8f0b 100644 --- a/crates/executor/src/executor.rs +++ b/crates/executor/src/executor.rs @@ -16,16 +16,24 @@ use crate::types::{MarketUpdate, OrderRecord, TradeIntent}; #[derive(Debug, Clone, Default)] pub struct MarketCache { latest: Option, + local_received_at_ms: Option, } impl MarketCache { pub fn update(&mut self, update: MarketUpdate) { + let now_ms = crate::bitget::now_ms().parse::().unwrap_or(0); + self.update_at(update, now_ms); + } + + pub fn update_at(&mut self, update: MarketUpdate, local_received_at_ms: i64) { self.latest = Some(update); + self.local_received_at_ms = Some(local_received_at_ms); } pub fn latest_fresh(&self, now_ms: i64, stale_after_secs: u64) -> Option { let update = self.latest.clone()?; - let age_ms = now_ms.saturating_sub(update.exchange_ts_ms); + let received_at = self.local_received_at_ms?; + let age_ms = now_ms.saturating_sub(received_at); if age_ms <= (stale_after_secs as i64) * 1000 { Some(update) } else { @@ -34,13 +42,253 @@ impl MarketCache { } } -/// Turn a (possibly stale/missing) market snapshot into a Result. Used at the -/// maker-retry and taker-fallback decision points: a maker order times out after -/// up to 15s while the cache's freshness window is 3s, so re-placing on the -/// pre-timeout price would post a stale limit. None (stale or unavailable) must -/// FAIL the intent rather than place on a price we can't trust. -fn require_fresh_market(market: Option) -> Result { - market.ok_or_else(|| anyhow::anyhow!("market data is stale; cannot place on stale price")) +/// What the intent loop should do for one intent, given WS cache freshness. +/// Pure so the open-vs-close gating is unit-testable without a network round-trip. +/// +/// - Opening actions (open/add/reverse) need FRESH WS market to price a limit/market +/// order on the current level. A stale/missing WS cache DEFERS them (keep pending) +/// — never fail — so a transient WS gap can't permanently reject an intent (spec: +/// stale market blocks only new opening exposure). REST ticker is NOT used for +/// opens here: the whole point of the WS cache is fresh best bid/ask, and falling +/// back to a (possibly also stale/lagged) REST ticker for an OPEN would defeat the +/// freshness gate. The open defers until WS is fresh again. +/// - Risk-reducing actions (close/reduce/cancel) proceed via a fresh REST ticker +/// regardless of WS staleness (spec line 95: close/cancel/de-risk stay allowed). +#[derive(Debug, Clone, PartialEq)] +pub enum MarketGate { + /// Use this fresh WS snapshot (opening action, cache fresh). + UseWs(MarketUpdate), + /// Defer (leave pending) — opening action but WS cache stale/missing. Retry next tick. + DeferOpen, + /// Risk-reducing action; proceed by fetching a fresh REST ticker (WS cache irrelevant). + UseRestTicker, +} + +pub(crate) fn market_gate(action: &str, cache: Option) -> MarketGate { + if is_opening_action(action) { + match cache { + Some(m) => MarketGate::UseWs(m), + None => MarketGate::DeferOpen, + } + } else { + MarketGate::UseRestTicker + } +} + +/// Minimum gap (ms) between two "deferred open" events for the same intent. +// ponytail: the daemon ticks every 250ms; without a per-intent cap a WS outage +// floods the events table (one row per tick per pending open). 10s balances +// observability (a stuck-deferred intent still logs every ~10s) against write +// volume. Generalize to config if multiple symbols/intents make this too coarse. +const DEFER_EVENT_WINDOW_MS: i64 = 10_000; + +/// True if a deferred-open event was emitted recently enough to suppress another +/// emit. Pure over (last_emit_ms, now_ms) so the rate-limit is unit-testable +/// without the REST/WS harness; the loop persists last_emit_ms in executor_state. +fn defer_event_recently_emitted(last_emit_ms: Option, now_ms: i64) -> bool { + match last_emit_ms { + Some(last) => now_ms - last < DEFER_EVENT_WINDOW_MS, + None => false, + } +} + +/// Best-effort durable failure for an intent that hit an infrastructure error +/// AFTER it was accepted. Logs the error event, then flips the intent to +/// `failed` so a stuck-`accepted` row can't wedge the loop (accept_intent only +/// matches `pending`, so the loop would otherwise never retry it). reconcile +/// owns the real order/position truth. Both writes are best-effort: a sqlite +/// failure here is logged but must not propagate (the caller is mid-batch and a +/// logging error must not strand the rest of the intents — same isolation as +/// the per-intent snapshot/equity writes above). +fn fail_intent_after_infra_error(conn: &Connection, intent_id: &str, err: &anyhow::Error) { + let _ = db::write_event( + conn, + "error", + "intent_loop", + &format!("intent {intent_id} infrastructure error: {err}"), + "{}", + ); + if let Err(fail_err) = db::fail_intent(conn, intent_id, &format!("infrastructure error: {err}")) + { + let _ = db::write_event( + conn, + "error", + "intent_loop", + &format!("could not mark intent {intent_id} failed: {fail_err}"), + "{}", + ); + } +} + +/// Process every pending intent in DB order using the same code path the +/// one-shot `run_once_or_loop` uses. Extracted so the daemon (Task 7) can drive +/// the loop on its own cadence without duplicating the per-intent wiring +/// (account snapshot, equity row, freshness gate, process_one_intent, event). +/// Behavior is identical to the prior inline loop; it returns the count only so +/// a caller may log it. +pub async fn process_pending_intents_once( + conn: &rusqlite::Connection, + cfg: &ExecutorConfig, + rest: &BitgetRestClient, + market_cache: &mut MarketCache, + private_state_ready: bool, +) -> Result { + let intents = db::pending_intents(conn)?; + let mut processed = 0usize; + for intent in intents { + // Gate FIRST: a stale public WS defers an OPEN before any REST call. The + // daemon ticks every 250ms; if the WS is down, fetching a REST account + // snapshot for a pending open each tick would hammer REST for nothing. + // Close/reduce (UseRestTicker) and fresh-WS opens (UseWs) proceed below. + let now_ms = crate::bitget::now_ms().parse::().unwrap_or(0); + let cache = market_cache.latest_fresh(now_ms, cfg.stale_market_data_secs); + // DeferOpen is exactly (opening action + no fresh WS cache); check it without + // consuming `cache` so the same Option feeds the market_gate resolution below. + if cache.is_none() && is_opening_action(&intent.action) { + // Opening action + stale/missing WS: defer (leave pending). Do NOT + // fail — a transient WS gap must not permanently reject an open. + // Stays pending; the next tick retries once WS is fresh. + // + // Rate-limit the defer event: the daemon ticks every 250ms, so a WS + // outage would otherwise flood the events table with one row per tick + // per pending open. Emit at most once per DEFER_EVENT_WINDOW per + // intent (last-emit timestamp persisted in executor_state so the cap + // survives across ticks). The intent defers either way. + if !defer_event_recently_emitted( + db::get_executor_state(conn, &format!("deferred_open:{}", intent.intent_id)) + .unwrap_or(None) + .and_then(|s| s.parse::().ok()), + now_ms, + ) { + let _ = db::set_executor_state( + conn, + &format!("deferred_open:{}", intent.intent_id), + &now_ms.to_string(), + ); + let _ = db::write_event( + conn, + "info", + "intent_loop", + &format!("deferred open {}: market data stale", intent.intent_id), + "{}", + ); + } + continue; + } + + // Fetch a FRESH account snapshot per intent. A run can carry multiple + // opening intents; if the first fills, its gross notional/equity have + // moved by the time the second is risk-checked. Sharing one snapshot + // across the loop would let a later intent pass the cap on stale equity + // or double-count the headroom the first fill already consumed. + // ponytail: per-intent error isolation — a transient account-snapshot + // failure must not strand the rest of the batch (head-of-line blocking). + // Log + continue; the next tick retries. + let account = match fetch_account_snapshot(rest, private_state_ready).await { + Ok(a) => a, + Err(err) => { + let _ = db::write_event( + conn, + "warning", + "intent_loop", + &format!("account snapshot failed for {}: {err}", intent.intent_id), + "{}", + ); + continue; + } + }; + // ponytail: isolate the per-intent audit-row write too — a sqlite + // failure here (disk full / DB locked) must not strand the rest of the + // batch. Skip this intent and retry next tick; the snapshot was fetched + // but couldn't be persisted, so don't process the intent either. + if let Err(err) = db::insert_equity_snapshot( + conn, + account.equity, + account.available_margin, + account.unrealized_pnl_24h, + 0.0, + ) { + let _ = db::write_event( + conn, + "warning", + "intent_loop", + &format!( + "equity snapshot write failed for {}: {err}", + intent.intent_id + ), + "{}", + ); + continue; + } + + // Resolve the market snapshot: a fresh-WS open reuses the cached level; + // a close/reduce (UseRestTicker) fetches a FRESH REST ticker regardless + // of WS staleness (spec: close/cancel/de-risk stay allowed when safe) and + // refreshes the cache so later opens in the same batch see it. + let market = match market_gate(&intent.action, cache) { + MarketGate::UseWs(m) => m, + MarketGate::UseRestTicker => match fetch_market_snapshot(cfg, rest).await { + Ok(m) => { + market_cache.update(m.clone()); + m + } + Err(err) => { + let _ = db::write_event( + conn, + "warning", + "intent_loop", + &format!("market refresh failed for {}: {err}", intent.intent_id), + "{}", + ); + continue; + } + }, + // DeferOpen was already short-circuited above; this arm satisfies + // exhaustiveness and is never reached here. + MarketGate::DeferOpen => continue, + }; + + // ponytail: process_one_intent fails intents DURABLY via db::fail_intent + // and returns Ok(()) for known failure modes (risk reject, place-order + // reject, market-refresh failure, cancel-confirm failure, max-commands). + // A residual Err is genuine infrastructure error (REST/sqlite) that fired + // AFTER accept_intent flipped the intent pending→accepted. accept_intent + // only matches `pending`, so a stuck-accepted intent would be invisible to + // every future tick (pending_intents selects pending only) — i.e. wedged + // forever. Fail it durably here so the loop can't strand it; reconcile + // owns the real order/position truth, so "failed" means "we lost track of + // this attempt", not "no order was placed". This single guard covers every + // bare `?` inside process_one_intent (sizing, retry/taker account snapshot, + // order-row writes) — fixing the wedge once, where all Err paths converge, + // instead of at each `?`. + if let Err(err) = process_one_intent( + conn, + cfg, + rest, + intent.clone(), + market, + account, + market_cache, + ) + .await + { + fail_intent_after_infra_error(conn, &intent.intent_id, &err); + } + // Only count intents that actually reached process_one_intent (UseWs / + // UseRestTicker arms). A deferred open stays pending and is NOT counted, + // so the caller can tell the batch had headroom left to retry. + // ponytail: the processed-event write is best-effort — a logging failure + // here must not strand the batch; the println and count are non-failing. + if let Err(err) = db::write_event(conn, "info", "executor", "processed intent", "{}") { + eprintln!( + "processed-intent event write failed for {}: {err}", + intent.intent_id + ); + } + println!("processed {}", intent.intent_id); + processed += 1; + } + Ok(processed) } pub async fn run_once_or_loop(cfg: ExecutorConfig) -> Result<()> { @@ -77,49 +325,31 @@ pub async fn run_once_or_loop(cfg: ExecutorConfig) -> Result<()> { ) .await?; - let intents = db::pending_intents(&conn)?; // ponytail: seed the market cache once with a real ticker (not the WS stream // yet); latest_fresh per intent rejects openings when the cached price is // older than cfg.stale_market_data_secs. let mut market_cache = MarketCache::default(); market_cache.update(fetch_market_snapshot(&cfg, &rest).await?); - for intent in intents { - // Fetch a FRESH account snapshot per intent. A run can carry multiple - // opening intents; if the first fills, its gross notional/equity have - // moved by the time the second is risk-checked. Sharing one snapshot - // across the loop would let a later intent pass the cap on stale equity - // or double-count the headroom the first fill already consumed. - let account = fetch_account_snapshot(&rest).await?; - db::insert_equity_snapshot( - &conn, - account.equity, - account.available_margin, - account.unrealized_pnl_24h, - 0.0, - )?; - let now_ms = crate::bitget::now_ms().parse::().unwrap_or(0); - let market = - require_fresh_market(market_cache.latest_fresh(now_ms, cfg.stale_market_data_secs))?; - process_one_intent( - &conn, - &cfg, - &rest, - intent.clone(), - market, - account, - &mut market_cache, - ) - .await?; - db::write_event(&conn, "info", "executor", "processed intent", "{}")?; - println!("processed {}", intent.intent_id); - } + // ponytail: pass true — the one-shot already verified the private WS connects + // via verify_private_ws_connects above, so private state IS ready at one-shot + // time. The daemon instead threads its live PrivateStateReady signal here. + process_pending_intents_once(&conn, &cfg, &rest, &mut market_cache, true).await?; Ok(()) } /// Fetch real account equity/margin/unrealized PnL + sum positions into gross /// notional. Parses the Bitget v2 mix account + all-position response into the /// AccountRiskSnapshot the risk gate consumes — no hardcoded fiction. -async fn fetch_account_snapshot(rest: &BitgetRestClient) -> Result { +/// `private_state_ready` is the daemon's live PrivateStateReady signal value at +/// the call site: the per-intent ENTRY snapshot (process_pending_intents_once) +/// passes the live value so an open is gated on it; the mid-execution re-checks +/// inside process_one_intent (maker-retry, taker) pass `true` — those run AFTER +/// the open was already gated at entry, and aborting a half-placed open on a WS +/// flap would be worse than completing it (spec: refuse new OPENING exposure). +async fn fetch_account_snapshot( + rest: &BitgetRestClient, + private_state_ready: bool, +) -> Result { let acct = rest.get_account_snapshot().await?; let data = acct .get("data") @@ -187,7 +417,7 @@ async fn fetch_account_snapshot(rest: &BitgetRestClient) -> Result f64 { if step <= 0.0 { return value; } - (value / step).floor() * step + // ponytail: floating-point can store a clean lot slightly below its true + // value (one lot of 0.01 is 0.009999999999999995 in f64). A bare + // (value/step).floor() maps 0.9999...→0, dropping a whole lot — and at one + // lot sizes the order to 0, which Bitget rejects ("less than minimum order + // quantity"). Add an epsilon in LOT-COUNT space: 1e-9 is far smaller than + // the lot step (0.01) so a genuine sub-lot remainder (0.005, 0.0199) is + // never bumped up to the next lot, but far larger than FP division error, so + // noise at a lot boundary is absorbed. Money path: rounding stays DOWN, so + // the placed size never exceeds the approved notional. + let lots = value / step; + (lots + 1e-9).floor() * step } -fn format_size(value: f64) -> String { +pub fn format_size(value: f64) -> String { // ponytail: round to the lot step before formatting so Bitget accepts the size. let rounded = round_down_to_step(value, MIN_ORDER_BASE); format!("{rounded:.4}") @@ -894,7 +1134,11 @@ pub async fn process_one_intent( if attempt > 1 { // Fetch a FRESH account snapshot so the re-check uses current // equity/margin, not the stale snapshot from run start. - let fresh_account = fetch_account_snapshot(rest).await?; + // ponytail: pass true — this re-check runs MID-EXECUTION of an + // already-gated open (entry snapshot gated it); aborting here on + // a WS flap would strand a half-placed open. Spec scopes the + // private-state gate to NEW opening exposure. + let fresh_account = fetch_account_snapshot(rest, true).await?; if let Err(reason) = check_intent( &intent, &fresh_account, @@ -1118,7 +1362,9 @@ pub async fn process_one_intent( } }; // Re-check risk with a FRESH account snapshot before taker fallback. - let fresh_account = fetch_account_snapshot(rest).await?; + // ponytail: pass true — mid-execution of an already-gated open; see + // the maker-retry re-check above for the same rationale. + let fresh_account = fetch_account_snapshot(rest, true).await?; if let Err(reason) = check_intent( &intent, &fresh_account, @@ -1270,6 +1516,75 @@ mod tests { use super::*; use crate::types::{MarketUpdate, TradeIntent}; + fn memory_db() -> Connection { + let conn = Connection::open_in_memory().unwrap(); + conn.execute_batch(include_str!("../../../schema/001_initial.sql")) + .unwrap(); + conn.execute_batch(include_str!("../../../schema/002_execution.sql")) + .unwrap(); + conn + } + + fn insert_pending_intent(conn: &Connection, intent_id: &str) { + conn.execute( + "insert into trade_intents ( + intent_id, created_at, symbol, side, action, target_notional, + max_order_notional, status, source + ) values (?1, '2026-07-01T00:00:00Z', 'ETH/USDT:USDT', + 'long', 'open', 100, 100, 'pending', 'test')", + rusqlite::params![intent_id], + ) + .unwrap(); + } + + fn intent_status(conn: &Connection, intent_id: &str) -> String { + conn.query_row( + "select status from trade_intents where intent_id = ?1", + rusqlite::params![intent_id], + |row| row.get(0), + ) + .unwrap() + } + + #[test] + fn infra_error_marks_accepted_intent_failed_not_stuck() { + // Regression: process_one_intent accepts the intent (pending→accepted) + // BEFORE driving the state machine, and accept_intent only matches + // `pending`. So a later infrastructure Err would leave the row accepted + // and invisible to every future tick — wedged forever. fail_intent_after_ + // infra_error must flip it to `failed` so the loop can't strand it. + let conn = memory_db(); + insert_pending_intent(&conn, "i1"); + assert!(db::accept_intent(&conn, "i1").unwrap()); + assert_eq!(intent_status(&conn, "i1"), "accepted"); + + fail_intent_after_infra_error(&conn, "i1", &anyhow::anyhow!("rest timeout")); + + assert_eq!( + intent_status(&conn, "i1"), + "failed", + "infra error after accept must not leave the intent stuck accepted" + ); + let err: String = conn + .query_row( + "select error from trade_intents where intent_id = 'i1'", + [], + |row| row.get(0), + ) + .unwrap(); + assert!(err.contains("rest timeout")); + assert!(err.contains("infrastructure error")); + } + + #[test] + fn defer_event_rate_limit_suppresses_recent_emits() { + // Never emitted → emit. Within the 10s window → suppress. Past it → emit. + assert!(!defer_event_recently_emitted(None, 10_000)); + assert!(defer_event_recently_emitted(Some(9_500), 10_000)); // 500ms ago + assert!(!defer_event_recently_emitted(Some(0), 10_000)); // 10s ago → emit + assert!(defer_event_recently_emitted(Some(10_000), 10_000)); // exactly now + } + #[test] fn maker_open_long_uses_best_bid() { let intent = TradeIntent { @@ -1354,6 +1669,26 @@ mod tests { assert_eq!(format_size(0.0199), "0.01"); } + #[test] + fn round_down_to_step_absorbs_fp_noise_at_lot_boundary() { + // 0.01 stored as 0.009999999999999995 is ONE lot, not zero. A bare + // (value/step).floor() maps 0.9999...→0, under-sizing the order to 0 + // (Bitget then rejects "less than minimum order quantity"). The epsilon + // absorbs the noise so a value at-or-above an integer lot keeps that lot. + assert_eq!(round_down_to_step(0.009999999999999995, 0.01), 0.01); + // a genuine sub-lot remainder must NOT be bumped up to the next lot + assert_eq!(round_down_to_step(0.0199, 0.01), 0.01); // 1.99 lots → 1 + assert_eq!(round_down_to_step(0.005, 0.01), 0.0); // half a lot → 0 + } + + #[test] + fn format_size_handles_fp_noise_close_size() { + // The observed fill can come back as 0.009999999999999995 (one lot, + // FP-noisy); formatting it for a reduce-only close must yield "0.01", not + // the raw repr the exchange rejects as below the minimum order quantity. + assert_eq!(format_size(0.009999999999999995), "0.01"); + } + #[test] fn is_open_blocked_only_for_active_override() { // money-path open gate: only the literal "active" state blocks opening. @@ -1427,20 +1762,30 @@ mod tests { } #[test] - fn stale_market_cache_returns_none() { + fn market_cache_uses_local_received_time_for_freshness() { let mut cache = MarketCache::default(); - cache.update(MarketUpdate { - symbol: "ETHUSDT".to_string(), - best_bid: 3000.0, - best_ask: 3000.5, - exchange_ts_ms: 1, - }); + cache.update_at( + MarketUpdate { + symbol: "ETHUSDT".to_string(), + best_bid: 100.0, + best_ask: 101.0, + exchange_ts_ms: 10, + }, + 1_000, + ); + + // Freshness is judged by LOCAL-received time, not exchange time: the WS + // loop stamps when it receives the message locally, so the staleness + // window (stale_after_secs=3 -> 3000ms) measures now - local_received. + assert!(cache.latest_fresh(3_999, 3).is_some()); + assert!(cache.latest_fresh(4_001, 3).is_none()); + } + + #[test] + fn market_cache_rejects_missing_snapshot() { + let cache = MarketCache::default(); - // units are ms: stale_after_secs(3) -> 3000ms threshold. - // age 3999ms (now 4000 - ts 1) exceeds it -> stale -> None. - assert!(cache.latest_fresh(4_000, 3).is_none()); - // age 999ms (now 1000 - ts 1) is under it -> fresh -> Some. - assert!(cache.latest_fresh(1_000, 3).is_some()); + assert!(cache.latest_fresh(1_000, 3).is_none()); } #[test] @@ -1458,20 +1803,37 @@ mod tests { } #[test] - fn require_fresh_market_rejects_missing_snapshot() { - // A maker order can wait up to 15s before timing out; the freshness - // window is 3s. Re-placing the retry/taker on the pre-timeout price - // would post a stale limit, so a None (stale/unavailable) snapshot must - // be an error the intent fails on — not a silent place-on-old-price. - assert!(require_fresh_market(None).is_err()); - let fresh = MarketUpdate { + fn market_gate_defers_open_when_cache_stale() { + // Opening action with a stale/missing WS cache must DEFER (stay pending), + // never fail — a transient WS gap must not permanently reject an open. + assert_eq!(market_gate("open", None), MarketGate::DeferOpen); + assert_eq!(market_gate("add", None), MarketGate::DeferOpen); + } + + #[test] + fn market_gate_uses_ws_for_open_when_fresh() { + let m = MarketUpdate { + symbol: "ETHUSDT".to_string(), + best_bid: 3000.0, + best_ask: 3000.5, + exchange_ts_ms: 1, + }; + // Fresh WS cache for an opening action -> use it directly. + assert_eq!(market_gate("open", Some(m.clone())), MarketGate::UseWs(m)); + } + + #[test] + fn market_gate_uses_rest_ticker_for_close_regardless_of_cache() { + let m = MarketUpdate { symbol: "ETHUSDT".to_string(), best_bid: 3000.0, best_ask: 3000.5, exchange_ts_ms: 1, }; - let got = require_fresh_market(Some(fresh.clone())).unwrap(); - assert_eq!(got.best_bid, fresh.best_bid); + // Risk-reducing actions ignore WS staleness; they proceed via a fresh + // REST ticker whether the WS cache is missing or present. + assert_eq!(market_gate("close", None), MarketGate::UseRestTicker); + assert_eq!(market_gate("close", Some(m)), MarketGate::UseRestTicker); } #[test] diff --git a/crates/executor/src/lib.rs b/crates/executor/src/lib.rs index 10ede59..6329f52 100644 --- a/crates/executor/src/lib.rs +++ b/crates/executor/src/lib.rs @@ -1,5 +1,6 @@ pub mod bitget; pub mod config; +pub mod daemon; pub mod db; pub mod executor; pub mod manual_override; @@ -7,4 +8,5 @@ pub mod notify; pub mod reconcile; pub mod risk; pub mod state; +pub mod telegram_query; pub mod types; diff --git a/crates/executor/src/main.rs b/crates/executor/src/main.rs index e681e7f..1ceae88 100644 --- a/crates/executor/src/main.rs +++ b/crates/executor/src/main.rs @@ -2,56 +2,167 @@ use anyhow::{bail, Result}; use prodigy_executor::config::{load_env_file, DemoSecrets, ExecutorConfig}; use prodigy_executor::executor; use std::env; -use std::path::Path; +use std::path::{Path, PathBuf}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RunMode { + Once, + Daemon, +} + +#[derive(Debug)] +struct ParsedExecutorArgs { + cfg: ExecutorConfig, + run_mode: RunMode, + max_runtime_ms: Option, +} #[tokio::main] async fn main() -> Result<()> { - let cfg = parse_args_and_config()?; - cfg.validate_demo_only()?; - executor::run_once_or_loop(cfg).await + let parsed = parse_args_and_config()?; + parsed.cfg.validate_demo_only()?; + match parsed.run_mode { + RunMode::Once => executor::run_once_or_loop(parsed.cfg).await, + RunMode::Daemon => { + prodigy_executor::daemon::run_daemon( + parsed.cfg, + prodigy_executor::daemon::DaemonOptions { + max_runtime: parsed.max_runtime_ms.map(std::time::Duration::from_millis), + }, + ) + .await + } + } +} + +fn parse_args_and_config() -> Result { + parse_args_from(env::args()) +} + +// Production entry point for arg parsing. Loads the `.env.local` file, overlays +// real process env vars (real env wins over the file, matching prior behavior), +// then delegates to the pure `parse_args_from_env`. Kept as a thin wrapper so +// the actual CLI parsing can be unit-tested with an injected env map and no +// filesystem/secret dependency. +fn parse_args_from(args: I) -> Result +where + I: IntoIterator, + S: Into, +{ + let mut env_file = load_env_file(&find_env_local())?; + for key in [ + "BITGET_DEMO_API_KEY", + "BITGET_DEMO_API_SECRET", + "BITGET_DEMO_SECRET_KEY", + "BITGET_DEMO_API_PASSPHRASE", + "BITGET_DEMO_PASSPHRASE", + "TELEGRAM_BOT_TOKEN", + "TELEGRAM_CHAT_ID", + ] { + if let Ok(value) = env::var(key) { + env_file.insert(key.to_string(), value); + } + } + parse_args_from_env(args, &env_file) } -fn parse_args_and_config() -> Result { +// Pure CLI parsing + secret resolution against an injected env map. Reads NO +// filesystem and NO real process env — everything comes from `env_file`. This +// makes the parse-mode unit tests hermetic. +fn parse_args_from_env( + args: I, + env_file: &std::collections::HashMap, +) -> Result +where + I: IntoIterator, + S: Into, +{ let mut cfg = ExecutorConfig::demo_for_tests(); - let env_file = load_env_file(Path::new(".env.local"))?; - let mut args = env::args().skip(1); + let mut run_mode = RunMode::Once; + let mut max_runtime_ms: Option = None; + let mut explicit_once = false; + let mut explicit_daemon = false; + + let mut args = args.into_iter().map(|s| s.into()).skip(1); while let Some(arg) = args.next() { match arg.as_str() { "--db" => { cfg.db_path = args .next() - .unwrap_or_else(|| "var/prodigy.sqlite".into()) + .unwrap_or_else(|| "var/prodigy.sqlite".to_string()) .into() } - "--once" => {} + "--once" => { + explicit_once = true; + run_mode = RunMode::Once; + } + "--daemon" => { + explicit_daemon = true; + run_mode = RunMode::Daemon; + } + "--max-runtime-ms" => { + let value = args + .next() + .ok_or_else(|| anyhow::anyhow!("--max-runtime-ms requires a value"))?; + max_runtime_ms = Some(value.parse().map_err(|_| { + anyhow::anyhow!("--max-runtime-ms must be a non-negative integer") + })?); + } "--test-reset-demo-state" => cfg.test_reset_demo_state = true, "--mode" => { - let value = args.next().unwrap_or_else(|| "demo".into()); + let value = args.next().unwrap_or_else(|| "demo".to_string()); if value != "demo" { - bail!("third milestone executor only supports --mode demo"); + bail!("prodigy executor only supports --mode demo"); } } other => bail!("unknown argument: {other}"), } } + if explicit_once && explicit_daemon { + bail!("cannot use --once and --daemon together"); + } + cfg.secrets = DemoSecrets { - api_key: read_secret(&["BITGET_DEMO_API_KEY"], &env_file)?, + api_key: read_secret(&["BITGET_DEMO_API_KEY"], env_file)?, // ponytail: .env.local ships two naming conventions; accept either so the // demo creds load regardless of which key the operator set. api_secret: read_secret( &["BITGET_DEMO_API_SECRET", "BITGET_DEMO_SECRET_KEY"], - &env_file, + env_file, )?, passphrase: read_secret( &["BITGET_DEMO_API_PASSPHRASE", "BITGET_DEMO_PASSPHRASE"], - &env_file, + env_file, )?, }; - cfg.telegram_bot_token = read_optional(&["TELEGRAM_BOT_TOKEN"], &env_file); - cfg.telegram_chat_id = read_optional(&["TELEGRAM_CHAT_ID"], &env_file); - Ok(cfg) + cfg.telegram_bot_token = read_optional(&["TELEGRAM_BOT_TOKEN"], env_file); + cfg.telegram_chat_id = read_optional(&["TELEGRAM_CHAT_ID"], env_file); + Ok(ParsedExecutorArgs { + cfg, + run_mode, + max_runtime_ms, + }) +} + +// ponytail: cargo runs tests with CWD = crate dir, but .env.local lives at the +// workspace root. Walk up to find it, mirroring the integration-test helper in +// tests/bitget_demo.rs. Also lets `main()` run from any subdir. +fn find_env_local() -> PathBuf { + let mut dir = match env::current_dir() { + Ok(d) => d, + Err(_) => return Path::new(".env.local").to_path_buf(), + }; + loop { + let candidate = dir.join(".env.local"); + if candidate.exists() { + return candidate; + } + if !dir.pop() { + return Path::new(".env.local").to_path_buf(); + } + } } fn read_secret( @@ -66,6 +177,83 @@ fn read_optional( env_file: &std::collections::HashMap, ) -> Option { keys.iter() - .filter_map(|k| env::var(k).ok().or_else(|| env_file.get(*k).cloned())) + .filter_map(|k| env_file.get(*k).cloned()) .find(|v| !v.is_empty()) } + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + // Injected fake demo creds so the parse tests never touch `.env.local` or + // real machine secrets. These are obviously-fake test values. + fn fake_env() -> HashMap { + let mut m = HashMap::new(); + m.insert("BITGET_DEMO_API_KEY".into(), "test-key".into()); + m.insert("BITGET_DEMO_API_SECRET".into(), "test-secret".into()); + m.insert("BITGET_DEMO_API_PASSPHRASE".into(), "test-pass".into()); + m + } + + #[test] + fn parses_once_mode_by_default() { + let parsed = parse_args_from_env(["prodigy-executor"], &fake_env()).unwrap(); + + assert_eq!(parsed.run_mode, RunMode::Once); + assert_eq!( + parsed.cfg.db_path, + std::path::PathBuf::from("var/prodigy.sqlite") + ); + } + + #[test] + fn parses_daemon_mode_and_db_path() { + let parsed = parse_args_from_env( + [ + "prodigy-executor", + "--daemon", + "--db", + "/tmp/prodigy-test.sqlite", + ], + &fake_env(), + ) + .unwrap(); + + assert_eq!(parsed.run_mode, RunMode::Daemon); + assert_eq!( + parsed.cfg.db_path, + std::path::PathBuf::from("/tmp/prodigy-test.sqlite") + ); + } + + #[test] + fn rejects_once_and_daemon_together() { + let err = parse_args_from_env(["prodigy-executor", "--once", "--daemon"], &HashMap::new()) + .unwrap_err(); + + assert!(err + .to_string() + .contains("cannot use --once and --daemon together")); + } + + #[test] + fn parses_bounded_daemon_runtime_for_tests() { + let parsed = parse_args_from_env( + ["prodigy-executor", "--daemon", "--max-runtime-ms", "1500"], + &fake_env(), + ) + .unwrap(); + + assert_eq!(parsed.run_mode, RunMode::Daemon); + assert_eq!(parsed.max_runtime_ms, Some(1500)); + } + + #[test] + fn rejects_live_mode_before_execution() { + let err = parse_args_from_env(["prodigy-executor", "--mode", "live"], &HashMap::new()) + .unwrap_err(); + + assert!(err.to_string().contains("only supports --mode demo")); + } +} diff --git a/crates/executor/src/notify.rs b/crates/executor/src/notify.rs index 7593d79..d51ce43 100644 --- a/crates/executor/src/notify.rs +++ b/crates/executor/src/notify.rs @@ -1,4 +1,4 @@ -use anyhow::Result; +use anyhow::{Context, Result}; use reqwest::Client; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -50,12 +50,21 @@ pub async fn send_telegram( return Ok(()); }; let url = format!("https://api.telegram.org/bot{token}/sendMessage"); - Client::new() + // ponytail: M4 spec — "Telegram delivery failure must not block execution, + // order management, or reconcile" (design line 141). reqwest's default has no + // request timeout, so a hung/slow Telegram POST would block every direct + // awaiter (reconcile pass, private-WS auth-failure helper). BOUND it to 3s; + // a timeout returns Err, which every caller already swallows (.ok() / let _ =), + // so Telegram being down degrades to "no notification" instead of "stuck loop". + // Single point of fix — no per-caller timeouts. + let request = Client::new() .post(url) .form(&[("chat_id", chat), ("text", text)]) - .send() - .await? - .error_for_status()?; + .send(); + let response = tokio::time::timeout(std::time::Duration::from_secs(3), request) + .await + .context("telegram send timed out")??; + response.error_for_status()?; Ok(()) } @@ -117,4 +126,43 @@ mod tests { "manual_override_cleared" )); } + + // Regression for the M4 spec: "Telegram delivery failure must not block + // execution, order management, or reconcile". send_telegram must never hang + // the caller. The real-network timeout path can't be exercised without an + // injectable URL (the host is hard-coded), so this pins the two non-network + // invariants: (a) a suppressed kind short-circuits to Ok before any I/O, and + // (b) missing token/chat short-circuits to Ok before any I/O. Both guarantee + // the most common callers (every demo-kind in notify, and any M4 daemon run + // without configured Telegram creds) return promptly. The 3s send() bound is + // a one-line constant verified by inspection + the live daemon smoke. + #[tokio::test] + async fn send_telegram_short_circuits_without_hanging() { + // Suppressed kind: no I/O attempted. + let suppressed = tokio::time::timeout( + std::time::Duration::from_secs(3), + send_telegram(Some("t"), Some("c"), "info", "x"), + ) + .await + .expect("suppressed kind must not hang"); + assert!(suppressed.is_ok()); + + // Missing token: no I/O attempted even for an active kind. + let no_token = tokio::time::timeout( + std::time::Duration::from_secs(3), + send_telegram(None, None, "critical", "x"), + ) + .await + .expect("missing-creds must not hang"); + assert!(no_token.is_ok()); + + // Active kind, partial creds (no chat): still short-circuits. + let partial = tokio::time::timeout( + std::time::Duration::from_secs(3), + send_telegram(Some("t"), None, "critical", "x"), + ) + .await + .expect("partial-creds must not hang"); + assert!(partial.is_ok()); + } } diff --git a/crates/executor/src/risk.rs b/crates/executor/src/risk.rs index 3e7b62d..467021f 100644 --- a/crates/executor/src/risk.rs +++ b/crates/executor/src/risk.rs @@ -37,32 +37,47 @@ pub fn check_intent( account: &AccountRiskSnapshot, params: &RiskParams, ) -> Result { - if !account.private_state_is_ready { + // Opening actions create new exposure and are subject to the full risk gate. + // Close/reduce/cancel are risk-REDUCING and bypass the open-only limits + // (M4 risk priority: de-risk outranks new-opening limits). They still must + // not run on a busted (equity<=0) account. + // ponytail: inlined here (not imported from executor.rs) because risk.rs + // cannot depend on executor.rs — executor depends on risk. + let is_open = matches!(intent.action.as_str(), "open" | "add" | "reverse"); + + if is_open && !account.private_state_is_ready { return Err("private account state is not ready".to_string()); } - if !account.market_is_fresh && intent.action == "open" { + if is_open && !account.market_is_fresh { return Err("market data is stale".to_string()); } if account.equity <= 0.0 { return Err("equity is not positive".to_string()); } - if account.available_margin < account.equity * params.min_available_margin_fraction { + if is_open && account.available_margin < account.equity * params.min_available_margin_fraction { return Err("available margin is too low".to_string()); } let suspended = account.unrealized_pnl_24h <= -account.equity * params.trading_suspension_unrealized_loss_x_equity; - if suspended && intent.action == "open" { + if suspended && is_open { return Err("trading suspended by 24h unrealized loss".to_string()); } - let total_cap = account.equity * params.total_notional_cap_x_equity; - let remaining = (total_cap - account.gross_notional).max(0.0); - let approved = intent - .target_notional - .min(intent.max_order_notional) - .min(remaining); - if approved <= 0.0 && intent.action == "open" { + // Opens are clipped by the total-notional cap (new exposure). Close/reduce/cancel + // are NOT new exposure — they use the intent's notional directly, uncapped, so a + // cap-exhausted account can still de-risk. + let approved = if is_open { + let total_cap = account.equity * params.total_notional_cap_x_equity; + let remaining = (total_cap - account.gross_notional).max(0.0); + intent + .target_notional + .min(intent.max_order_notional) + .min(remaining) + } else { + intent.target_notional.min(intent.max_order_notional) + }; + if approved <= 0.0 && is_open { return Err("notional cap reached".to_string()); } @@ -167,4 +182,120 @@ mod tests { assert!(err.contains("notional cap reached")); } + + #[test] + fn close_is_allowed_when_total_notional_cap_exhausted() { + // Cap exhausted (remaining 0): a close must still size its full notional. + let account = AccountRiskSnapshot { + equity: 1_000.0, + available_margin: 1_000.0, + unrealized_pnl_24h: 0.0, + gross_notional: 5_000.0, // == cap → remaining 0 + market_is_fresh: true, + private_state_is_ready: true, + }; + + let decision = check_intent( + &intent("close", 200.0, 200.0), + &account, + &RiskParams::default(), + ) + .unwrap(); + + assert_eq!(decision.approved_notional, 200.0); + + // Open on the same cap-exhausted account IS still blocked. + let err = check_intent( + &intent("open", 200.0, 200.0), + &account, + &RiskParams::default(), + ) + .unwrap_err(); + assert!(err.contains("notional cap reached")); + } + + #[test] + fn close_is_allowed_when_available_margin_low() { + // available_margin below the 5% floor would block an open; close must + // still go through (closing REDUCES margin usage — that's the point). + let account = AccountRiskSnapshot { + equity: 1_000.0, + available_margin: 10.0, // < 5% of 1000 = 50 + unrealized_pnl_24h: 0.0, + gross_notional: 0.0, + market_is_fresh: true, + private_state_is_ready: true, + }; + + let decision = check_intent( + &intent("close", 100.0, 100.0), + &account, + &RiskParams::default(), + ) + .unwrap(); + + assert_eq!(decision.approved_notional, 100.0); + + let err = check_intent( + &intent("open", 100.0, 100.0), + &account, + &RiskParams::default(), + ) + .unwrap_err(); + assert!(err.contains("available margin is too low")); + } + + #[test] + fn close_ignores_stale_market_and_unready_private_state() { + // Private WS down + stale market would block open; close bypasses both + // (close must be possible during a WS outage — REST still works). + let account = AccountRiskSnapshot { + equity: 1_000.0, + available_margin: 1_000.0, + unrealized_pnl_24h: 0.0, + gross_notional: 0.0, + market_is_fresh: false, + private_state_is_ready: false, + }; + + let decision = check_intent( + &intent("close", 100.0, 100.0), + &account, + &RiskParams::default(), + ) + .unwrap(); + + assert_eq!(decision.approved_notional, 100.0); + + let err = check_intent( + &intent("open", 100.0, 100.0), + &account, + &RiskParams::default(), + ) + .unwrap_err(); + assert!(err.contains("private account state is not ready")); + } + + #[test] + fn close_still_blocked_when_equity_not_positive() { + // Bust guard applies to ALL actions — a busted (equity<=0) account is + // already liquidated; close must not run there either. + let account = AccountRiskSnapshot { + equity: 0.0, + available_margin: 0.0, + unrealized_pnl_24h: 0.0, + gross_notional: 0.0, + market_is_fresh: true, + private_state_is_ready: true, + }; + + let err = check_intent( + &intent("close", 100.0, 100.0), + &account, + &RiskParams::default(), + ) + .unwrap_err(); + + assert!(err.contains("equity is not positive")); + } } diff --git a/crates/executor/src/telegram_query.rs b/crates/executor/src/telegram_query.rs new file mode 100644 index 0000000..5e8f484 --- /dev/null +++ b/crates/executor/src/telegram_query.rs @@ -0,0 +1,241 @@ +//! Read-only Telegram query formatting (M4). +//! +//! Maps the operator's `/status /positions /orders /pnl /risk` commands to +//! short SQLite-backed replies. These are STRICTLY read-only: no rows are +//! written, no side effects on the trading system. `/stop /resume /close_all` +//! are explicit non-grata in M4 — remote trading control is forbidden, so +//! they get a fixed "not supported in M4" refusal rather than acting. +//! +//! `query_response` returns `Ok(None)` for anything that isn't a recognized +//! command, so the polling loop simply doesn't reply to noise. + +use anyhow::Result; +use rusqlite::Connection; + +/// Map a single command line to its read-only reply, or `None` if it isn't a +/// recognized command (no reply). Remote trading controls are refused. +pub fn query_response(conn: &Connection, text: &str) -> Result> { + let command = text.split_whitespace().next().unwrap_or(""); + match command { + "/status" => Ok(Some(status_response(conn)?)), + "/positions" => Ok(Some(positions_response(conn)?)), + "/orders" => Ok(Some(orders_response(conn)?)), + "/pnl" => Ok(Some(pnl_response(conn)?)), + "/risk" => Ok(Some(risk_response(conn)?)), + "/stop" | "/resume" | "/close_all" => Ok(Some( + "remote trading controls are not supported in M4".to_string(), + )), + _ => Ok(None), + } +} + +fn status_response(conn: &Connection) -> Result { + let events: i64 = conn.query_row("select count(*) from events", [], |r| r.get(0))?; + let pending: i64 = conn.query_row( + "select count(*) from trade_intents where status = 'pending'", + [], + |r| r.get(0), + )?; + Ok(format!( + "status: daemon\npending_intents: {pending}\nevents: {events}" + )) +} + +fn positions_response(conn: &Connection) -> Result { + let mut stmt = conn.prepare( + "select symbol, side, notional, entry_price, unrealized_pnl, ownership + from positions order by symbol", + )?; + let rows = stmt.query_map([], |row| { + Ok(format!( + "{} {} notional={} entry={} upnl={} ownership={}", + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, f64>(2)?, + row.get::<_, f64>(3)?, + row.get::<_, f64>(4)?, + row.get::<_, String>(5)?, + )) + })?; + let lines = rows.collect::, _>>()?; + Ok(if lines.is_empty() { + "positions: none".to_string() + } else { + lines.join("\n") + }) +} + +fn orders_response(conn: &Connection) -> Result { + let mut stmt = conn.prepare( + "select client_oid, symbol, side, action, status, size, filled_size + from orders order by updated_at desc limit 10", + )?; + let rows = stmt.query_map([], |row| { + Ok(format!( + "{} {} {} {} status={} size={} filled={}", + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, f64>(5)?, + row.get::<_, f64>(6)?, + )) + })?; + let lines = rows.collect::, _>>()?; + Ok(if lines.is_empty() { + "orders: none".to_string() + } else { + lines.join("\n") + }) +} + +fn pnl_response(conn: &Connection) -> Result { + let unrealized: f64 = conn.query_row( + "select coalesce(sum(unrealized_pnl), 0) from positions", + [], + |r| r.get(0), + )?; + let equity: Option = conn + .query_row( + "select equity from equity_snapshots order by created_at desc limit 1", + [], + |r| r.get(0), + ) + .ok(); + // ponytail: M4 /pnl is unrealized-only — it does NOT report realized PnL. Realized + // PnL per closed trade is a future-milestone concern (it needs the per-trade fills + // ledger aggregated by close intent); M4 surfaces only live unrealized PnL + the + // last REST equity snapshot. Saying so explicitly so an operator isn't misled into + // treating this as total account PnL. + Ok(format!( + "pnl (unrealized-only, M4):\nunrealized={unrealized}\nequity={}\nrealized=n/a (not tracked in M4)", + equity.unwrap_or(0.0) + )) +} + +fn risk_response(conn: &Connection) -> Result { + let manual_overrides: i64 = conn.query_row( + "select count(*) from executor_state + where key like 'manual_override:%' and value = 'active'", + [], + |r| r.get(0), + )?; + let available_margin: Option = conn + .query_row( + "select available_margin from equity_snapshots order by created_at desc limit 1", + [], + |r| r.get(0), + ) + .ok(); + Ok(format!( + "risk:\nmanual_overrides={manual_overrides}\navailable_margin={}", + available_margin.unwrap_or(0.0) + )) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_conn() -> rusqlite::Connection { + let conn = rusqlite::Connection::open_in_memory().unwrap(); + conn.execute_batch(include_str!("../../../schema/001_initial.sql")) + .unwrap(); + conn.execute_batch(include_str!("../../../schema/002_execution.sql")) + .unwrap(); + conn + } + + #[test] + fn status_query_reads_sqlite_without_side_effects() { + let conn = test_conn(); + crate::db::write_event(&conn, "info", "daemon", "daemon started", "{}").unwrap(); + + let response = query_response(&conn, "/status").unwrap().unwrap(); + + assert!(response.contains("status")); + assert!(response.contains("daemon")); + } + + #[test] + fn positions_query_lists_current_positions() { + let conn = test_conn(); + conn.execute( + "insert into positions ( + symbol, side, notional, entry_price, unrealized_pnl, updated_at, + ownership, raw_json + ) values ('ETH/USDT:USDT', 'long', 100.0, 2000.0, 3.5, 'now', 'system', '{}')", + [], + ) + .unwrap(); + + let response = query_response(&conn, "/positions").unwrap().unwrap(); + + assert!(response.contains("ETH/USDT:USDT")); + assert!(response.contains("long")); + assert!(response.contains("3.5")); + } + + #[test] + fn orders_query_lists_recent_orders() { + let conn = test_conn(); + conn.execute( + "insert into orders ( + order_id, client_oid, intent_id, symbol, side, action, order_type, + status, price, size, filled_size, created_at, updated_at + ) values ( + 'order-1', 'client-xyz', null, 'ETH/USDT:USDT', 'buy', 'open', + 'limit', 'filled', 2000.0, 0.5, 0.5, 'now', 'now' + )", + [], + ) + .unwrap(); + + let response = query_response(&conn, "/orders").unwrap().unwrap(); + + assert!(response.contains("client-xyz")); + assert!(response.contains("ETH/USDT:USDT")); + assert!(response.contains("status=filled")); + } + + #[test] + fn remote_control_commands_are_not_supported_in_m4() { + for command in ["/stop", "/resume", "/close_all"] { + let conn = test_conn(); + let response = query_response(&conn, command).unwrap().unwrap(); + assert!(response.contains("not supported in M4")); + } + } + + #[test] + fn unrecognized_command_returns_none_so_no_reply_is_sent() { + let conn = test_conn(); + // ponytail: None = "don't reply" — the polling loop only sends on Some, + // so noise from other chats or stray text never spams the operator. + assert!(query_response(&conn, "hello world").unwrap().is_none()); + } + + #[test] + fn pnl_query_reports_unrealized_and_equity() { + let conn = test_conn(); + crate::db::insert_equity_snapshot(&conn, 1000.0, 500.0, -2.0, 0.0).unwrap(); + + let response = query_response(&conn, "/pnl").unwrap().unwrap(); + + assert!(response.contains("unrealized=0")); + assert!(response.contains("equity=1000")); + } + + #[test] + fn risk_query_counts_active_manual_overrides() { + let conn = test_conn(); + crate::db::insert_equity_snapshot(&conn, 1000.0, 500.0, 0.0, 0.0).unwrap(); + crate::db::set_executor_state(&conn, "manual_override:ETH/USDT:USDT", "active").unwrap(); + + let response = query_response(&conn, "/risk").unwrap().unwrap(); + + assert!(response.contains("manual_overrides=1")); + assert!(response.contains("available_margin=500")); + } +} diff --git a/crates/executor/src/types.rs b/crates/executor/src/types.rs index b7f4707..a3b93ee 100644 --- a/crates/executor/src/types.rs +++ b/crates/executor/src/types.rs @@ -64,9 +64,26 @@ pub struct MarketUpdate { pub exchange_ts_ms: i64, } +/// Equity snapshot parsed from a private-WS `account` event. A fast cache update; +/// the REST reconcile path remains the source of truth. +#[derive(Debug, Clone, PartialEq)] +pub struct AccountSnapshotUpdate { + pub equity: f64, + pub available_margin: f64, + pub unrealized_pnl: f64, +} + #[derive(Debug, Clone, Default, PartialEq)] pub struct PrivateWsUpdate { pub orders: Vec, pub fills: Vec, pub positions: Vec, + pub account: Option, + /// True when a private-WS `{"event":"login","code":"0",...}` ack was parsed. + /// The WS loop waits for this before treating private state as ready. + pub login_ack: bool, + /// Set when a private-WS `{"event":"error",...}` or a non-zero login code is + /// parsed (auth/subscribe failure). Carries the exchange message so the loop + /// can emit a `websocket_auth_failed` event and stay not-ready until reconnect. + pub auth_error: Option, } diff --git a/crates/executor/tests/bitget_demo.rs b/crates/executor/tests/bitget_demo.rs index 708827a..e820728 100644 --- a/crates/executor/tests/bitget_demo.rs +++ b/crates/executor/tests/bitget_demo.rs @@ -272,7 +272,11 @@ async fn bitget_demo_can_open_and_reduce_only_close_market_order() { product_type: cfg.product_type.clone(), margin_mode: cfg.margin_mode.clone(), margin_coin: cfg.margin_coin.clone(), - size: format!("{opened_size}"), + // ponytail: round the observed fill to the lot step before formatting + // — the raw repr (0.009999999999999995) is one lot, but Bitget rejects + // it as "less than minimum order quantity". Reuse the production sizer + // so this close path exercises the same lot-rounding as the executor. + size: prodigy_executor::executor::format_size(opened_size), price: None, side: "sell".to_string(), order_type: "market".to_string(), diff --git a/tests/test_executor_integration.py b/tests/test_executor_integration.py index 337992c..959cdb8 100644 --- a/tests/test_executor_integration.py +++ b/tests/test_executor_integration.py @@ -125,3 +125,106 @@ def test_rust_demo_executor_processes_pending_intent(tmp_path): f"total ({orders_filled_total}) — would indicate a double-count" ) assert event_count >= 1 + + +def test_rust_demo_daemon_processes_pending_intent_once(tmp_path): + _demo_depth_diagnostic() + db_path = tmp_path / "prodigy.sqlite" + + with connect(db_path) as conn: + init_db(conn) + write_trade_intent( + conn, + TradeIntent( + intent_id="daemon-intent-1", + created_at="2026-07-03T00:00:00Z", + symbol="ETH/USDT:USDT", + side="long", + action="open", + target_notional=100.0, + max_order_notional=100.0, + source="test", + reason="daemon integration", + model_version="smoke-test", + ), + ) + + result = subprocess.run( + [ + "cargo", + "run", + "-q", + "-p", + "prodigy-executor", + "--", + "--db", + str(db_path), + "--daemon", + "--max-runtime-ms", + "30000", + "--test-reset-demo-state", + ], + check=True, + text=True, + capture_output=True, + # Startup reconcile + set-leverage against the real demo API take ~7s + # before the loop starts, and a single intent-processing tick can run + # ~30-40s: the maker open path waits open_maker_timeout_seconds (15s) + # per attempt, refreshes, then falls back to a taker that the phantom + # demo book often rejects, finally failing "left to reconcile". The + # bounded-runtime check fires between ticks, so the first tick runs to + # the intent's terminal state regardless of max-runtime-ms; the bound + # only governs when the *next* tick exits. 90s leaves room for compile + # + startup + one full processing tick. + timeout=90, + ) + + with connect(db_path) as conn: + intent = conn.execute( + "select status, error from trade_intents where intent_id = 'daemon-intent-1'" + ).fetchone() + order_count = conn.execute("select count(*) from orders").fetchone()[0] + event_count = conn.execute("select count(*) from events").fetchone()[0] + # No zero-fill order may be marked filled — the M4 anti-false-fill + # invariant, mirrored from the --once test. + false_fills = conn.execute( + "select count(*) from orders where status = 'filled' and filled_size <= 0" + ).fetchone()[0] + + # Honest terminal state: executed when the demo book is tradable, failed + # with a diagnostic when it is phantom-liquid (see _demo_depth_diagnostic). + # 'pending'/'accepted' (stuck) is never accepted after bounded daemon runtime. + assert intent["status"] in ("executed", "failed"), ( + f"expected a terminal state (executed|failed), got {intent['status']}" + ) + if intent["status"] == "failed": + assert intent["error"], "a failed intent must record a diagnostic error" + assert order_count >= 1, "expected at least one demo order to be attempted" + assert event_count >= 1, "daemon must record startup + reconcile + intent events" + assert false_fills == 0, "an order must not be marked filled with no fill" + assert "daemon" in result.stdout or result.stderr == "" + + +def test_rust_daemon_rejects_live_mode(tmp_path): + db_path = tmp_path / "prodigy.sqlite" + result = subprocess.run( + [ + "cargo", + "run", + "-q", + "-p", + "prodigy-executor", + "--", + "--db", + str(db_path), + "--daemon", + "--mode", + "live", + ], + check=False, + text=True, + capture_output=True, + ) + + assert result.returncode != 0 + assert "only supports --mode demo" in result.stderr