soothe_client/
session.rs

1//! Session bootstrap helpers.
2
3use std::time::Duration;
4
5use serde_json::{json, Map, Value};
6
7use crate::client::Client;
8use crate::errors::{Error, Result};
9use crate::Config;
10
11/// Options for `bootstrap_loop_session`.
12#[derive(Debug, Clone, Default)]
13pub struct BootstrapOptions {
14    /// Resume an existing loop id (skip loop_new).
15    pub resume_loop_id: Option<String>,
16    /// Workspace path (`client_workspace`).
17    pub workspace: Option<String>,
18    /// User id.
19    pub user_id: Option<String>,
20    /// Client workspace id.
21    pub client_workspace_id: Option<String>,
22    /// Stream delivery mode.
23    pub stream_delivery: String,
24    /// Ephemeral loop flag.
25    pub is_ephemeral: bool,
26}
27
28impl BootstrapOptions {
29    /// Default adaptive delivery.
30    pub fn new() -> Self {
31        Self {
32            stream_delivery: "adaptive".into(),
33            ..Default::default()
34        }
35    }
36}
37
38/// Connect with bounded retries (cold-start races).
39pub async fn connect_with_retries(
40    client: &Client,
41    max_retries: u32,
42    retry_delay: Duration,
43) -> Result<()> {
44    let max = if max_retries == 0 { 40 } else { max_retries };
45    let delay = if retry_delay.is_zero() {
46        Duration::from_millis(250)
47    } else {
48        retry_delay
49    };
50    let mut last = None;
51    for attempt in 1..=max {
52        match client.connect().await {
53            Ok(()) => return Ok(()),
54            Err(e) => {
55                last = Some(e);
56                if attempt < max {
57                    tokio::time::sleep(delay).await;
58                }
59            }
60        }
61    }
62    Err(last.unwrap_or_else(|| Error::msg("connect failed")))
63}
64
65/// Run loop_new|reattach → subscribe. Assumes handshake already done in `connect`.
66pub async fn bootstrap_loop_session(
67    client: &Client,
68    opts: BootstrapOptions,
69    cfg: Option<&Config>,
70) -> Result<Map<String, Value>> {
71    let default_cfg = Config::default();
72    let cfg = cfg.unwrap_or(&default_cfg);
73    let delivery = if matches!(
74        opts.stream_delivery.as_str(),
75        "batch" | "adaptive" | "streaming"
76    ) {
77        opts.stream_delivery.as_str()
78    } else {
79        "adaptive"
80    };
81
82    let loop_id = if let Some(resume) = opts
83        .resume_loop_id
84        .as_ref()
85        .map(|s| s.trim().to_string())
86        .filter(|s| !s.is_empty())
87    {
88        resume
89    } else {
90        let mut params = Map::new();
91        if let Some(ws) = &opts.workspace {
92            if !ws.is_empty() {
93                params.insert("client_workspace".into(), json!(ws));
94            }
95        }
96        if let Some(uid) = &opts.user_id {
97            if !uid.is_empty() {
98                params.insert("user_id".into(), json!(uid));
99            }
100        }
101        if let Some(wsid) = &opts.client_workspace_id {
102            if !wsid.is_empty() {
103                params.insert("client_workspace_id".into(), json!(wsid));
104            }
105        }
106        if opts.is_ephemeral {
107            params.insert("is_ephemeral".into(), json!(true));
108        }
109        let resp = client
110            .request("loop_new", params, cfg.loop_status_timeout)
111            .await
112            .map_err(|e| Error::msg(format!("loop_new: {e}")))?;
113        let lid = resp
114            .get("loop_id")
115            .and_then(|v| v.as_str())
116            .unwrap_or("")
117            .trim()
118            .to_string();
119        if lid.is_empty() {
120            return Err(Error::msg("loop_new response missing loop_id"));
121        }
122        lid
123    };
124
125    client
126        .loop_subscribe(&loop_id, delivery)
127        .await
128        .map_err(|e| Error::msg(format!("loop_subscribe: {e}")))?;
129
130    let mut out = Map::new();
131    out.insert("type".into(), json!("session_ready"));
132    out.insert("loop_id".into(), json!(loop_id));
133    out.insert("success".into(), json!(true));
134    out.insert("autopilot_mode".into(), json!("solo"));
135    Ok(out)
136}
soothe_client/session.rs

soothe_client/
session.rs
