soothe-client 0.3.2

WebSocket client for talking to a running soothe-daemon
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

//! Session bootstrap helpers.

use std::time::Duration;

use serde_json::{json, Map, Value};

use crate::client::Client;
use crate::errors::{Error, Result};
use crate::Config;

/// Options for `bootstrap_loop_session`.
#[derive(Debug, Clone, Default)]
pub struct BootstrapOptions {
    /// Resume an existing loop id (skip loop_new).
    pub resume_loop_id: Option<String>,
    /// Workspace path (`client_workspace`).
    pub workspace: Option<String>,
    /// User id.
    pub user_id: Option<String>,
    /// Client workspace id.
    pub client_workspace_id: Option<String>,
    /// Stream delivery mode.
    pub stream_delivery: String,
    /// Ephemeral loop flag.
    pub is_ephemeral: bool,
}

impl BootstrapOptions {
    /// Default adaptive delivery.
    pub fn new() -> Self {
        Self {
            stream_delivery: "adaptive".into(),
            ..Default::default()
        }
    }
}

/// Connect with bounded retries (cold-start races).
pub async fn connect_with_retries(
    client: &Client,
    max_retries: u32,
    retry_delay: Duration,
) -> Result<()> {
    let max = if max_retries == 0 { 40 } else { max_retries };
    let delay = if retry_delay.is_zero() {
        Duration::from_millis(250)
    } else {
        retry_delay
    };
    let mut last = None;
    for attempt in 1..=max {
        match client.connect().await {
            Ok(()) => return Ok(()),
            Err(e) => {
                last = Some(e);
                if attempt < max {
                    tokio::time::sleep(delay).await;
                }
            }
        }
    }
    Err(last.unwrap_or_else(|| Error::msg("connect failed")))
}

/// Run loop_new|reattach → subscribe. Assumes handshake already done in `connect`.
pub async fn bootstrap_loop_session(
    client: &Client,
    opts: BootstrapOptions,
    cfg: Option<&Config>,
) -> Result<Map<String, Value>> {
    let default_cfg = Config::default();
    let cfg = cfg.unwrap_or(&default_cfg);
    let delivery = if matches!(
        opts.stream_delivery.as_str(),
        "batch" | "adaptive" | "streaming"
    ) {
        opts.stream_delivery.as_str()
    } else {
        "adaptive"
    };

    let loop_id = if let Some(resume) = opts
        .resume_loop_id
        .as_ref()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
    {
        resume
    } else {
        let mut params = Map::new();
        if let Some(ws) = &opts.workspace {
            if !ws.is_empty() {
                params.insert("client_workspace".into(), json!(ws));
            }
        }
        if let Some(uid) = &opts.user_id {
            if !uid.is_empty() {
                params.insert("user_id".into(), json!(uid));
            }
        }
        if let Some(wsid) = &opts.client_workspace_id {
            if !wsid.is_empty() {
                params.insert("client_workspace_id".into(), json!(wsid));
            }
        }
        if opts.is_ephemeral {
            params.insert("is_ephemeral".into(), json!(true));
        }
        let resp = client
            .request("loop_new", params, cfg.loop_status_timeout)
            .await
            .map_err(|e| Error::msg(format!("loop_new: {e}")))?;
        let lid = resp
            .get("loop_id")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .trim()
            .to_string();
        if lid.is_empty() {
            return Err(Error::msg("loop_new response missing loop_id"));
        }
        lid
    };

    client
        .loop_subscribe(&loop_id, delivery)
        .await
        .map_err(|e| Error::msg(format!("loop_subscribe: {e}")))?;

    let mut out = Map::new();
    out.insert("type".into(), json!("session_ready"));
    out.insert("loop_id".into(), json!(loop_id));
    out.insert("success".into(), json!(true));
    out.insert("autopilot_mode".into(), json!("solo"));
    Ok(out)
}