degenbot_cli_core/
operator.rs

1//! The operator command-channel client: JSON-lines over a Unix domain socket
2//! (ADR-051 D6;).
3//!
4//! `degenbot fleet posture [show|set]` and `degenbot path [add|discover]` are
5//! clients of a LIVE bot's `OperatorServer`. The versioned wire protocol is
6//! documented in `src/degenbot/operator/operator_channel.py`; that header +
7//! the server's framing are read-only contract and this module must not drift
8//! from them.
9//!
10//! # Wire shape (one JSON object per line, newline-terminated)
11//!
12//! Request lines nest the payload at the top level (the exact shape Python's
13//! `send_command` writes and the server's `_decode_request` reads):
14//!
15//! ```text
16//! {"op": "add_path", "payload": {"steps": [...], "directions": [true]}}
17//! {"op": "discover", "payload": {"bound": 5}}
18//! {"op": "set_fleet_posture", "payload": {"cordon_enter_events": 2}}
19//! {"op": "get_fleet_posture", "payload": {}}
20//! ```
21//!
22//! Response lines:
23//!
24//! ```text
25//! {"ok": true, "detail": "..."}
26//! {"ok": true, "detail": "", "effective": {...}}
27//! {"ok": false, "error": "..."}
28//! ```
29//!
30//! # Division of validation
31//!
32//! This client adds **wire hygiene only** - an unknown `cordon_*` key, an empty
33//! posture patch, an unknown hop-family string - refused BEFORE the socket is
34//! touched by [`validate_posture_patch`] / [`parse_hop_token`]. The six cordon
35//! values are DOMAIN validation and stay the authority of the server
36//! (`PosturePolicyPatch::validate` in the workers core); this crate deliberately
37//! does not import or re-implement it, so a client built earlier never second-
38//! guesses a host built later.
39//!
40//! # Socket resolution
41//!
42//! `--socket` > `DEGENBOT_OPERATOR_SOCKET` (the shell-environment layer, read
43//! through the `degenbot-config` `EnvVars` seam) > `~/.config/degenbot/operator.sock`
44//! (the example driver's plug-in default). No typed config-file key exists for
45//! the operator socket, so the env layer is the only config surface; a future
46//! typed key would slot in above the env layer at exactly one site.
47
48#[cfg(unix)]
49use std::future::Future;
50use std::path::{Path, PathBuf};
51#[cfg(unix)]
52use std::time::Duration;
53
54use degenbot_config::EnvVars;
55use serde_json::{json, Map, Value};
56#[cfg(unix)]
57use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
58#[cfg(unix)]
59use tokio::net::UnixStream;
60
61use crate::error::CliError;
62
63/// The environment variable naming the live bot's operator socket (the shell
64/// layer of the cascade). Empty is treated exactly like unset.
65pub const SOCKET_ENV: &str = "DEGENBOT_OPERATOR_SOCKET";
66
67/// The plug-in default socket path (the settlement-bot example's
68/// `--operator-socket` default family). The leading `~` expands against `HOME`.
69pub const SOCKET_DEFAULT: &str = "~/.config/degenbot/operator.sock";
70
71/// The `HOME` layer used to expand a leading `~` in the default.
72const HOME_ENV: &str = "HOME";
73
74/// Per-request connect/read timeout (mirrors the server's `request_timeout`
75/// default of 60s).
76#[cfg(unix)]
77const REQUEST_TIMEOUT: Duration = Duration::from_secs(60);
78
79/// The refusal `send_request` returns on a host with no Unix domain socket.
80/// The channel is a UDS protocol end to end (the Python `OperatorServer`
81/// cannot bind one on Windows either), so off Unix the command arms stay
82/// compiled and fail here, at the one transport site.
83#[cfg(not(unix))]
84const UDS_UNSUPPORTED: &str = "the operator command channel requires a Unix domain socket, \
85     which this platform does not provide";
86
87/// The six fleet-posture threshold key names `set_fleet_posture` accepts (the
88/// typed `DEGENBOT_FLEET_CORDON_*` keys). Anything else is refused at the wire
89/// before it reaches the host.
90pub const FLEET_POSTURE_THRESHOLD_KEYS: [&str; 6] = [
91    "cordon_enter_events",
92    "cordon_enter_window_ms",
93    "cordon_duty_percent",
94    "cordon_duty_window_ms",
95    "cordon_exit_clean_ms",
96    "cordon_sim_intake_floor",
97];
98
99/// The documented sentinel for `--cordon-sim-intake-floor`: the literal
100/// `null` (case-insensitive) restores half the slot cap; any other value is a
101/// threshold. The host's `PosturePolicyPatch` carries this as the
102/// `Some(None)` override.
103pub const SIM_INTAKE_FLOOR_RESTORE: &str = "null";
104
105/// A pool family on the wire (`V2` / `V3` / `V4`).
106#[derive(Debug, Clone, Copy, PartialEq, Eq)]
107pub enum PathFamily {
108    /// `V2`.
109    V2,
110    /// `V3`.
111    V3,
112    /// `V4`.
113    V4,
114}
115
116impl PathFamily {
117    /// The wire spelling (uppercase).
118    #[must_use]
119    pub const fn as_str(self) -> &'static str {
120        match self {
121            Self::V2 => "V2",
122            Self::V3 => "V3",
123            Self::V4 => "V4",
124        }
125    }
126
127    /// Parse a case-insensitive family string.
128    #[must_use]
129    pub fn parse(raw: &str) -> Option<Self> {
130        match raw.to_ascii_uppercase().as_str() {
131            "V2" => Some(Self::V2),
132            "V3" => Some(Self::V3),
133            "V4" => Some(Self::V4),
134            _ => None,
135        }
136    }
137}
138
139/// One hop in an `add_path` `steps` array.
140#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct PathStep {
142    /// The hop's pool family.
143    pub family: PathFamily,
144    /// The pool address (`0x` + 40 hex).
145    pub address: String,
146    /// The V4 pool id (`0x` + 64 hex); `None` for V2/V3 and for a V4 hop that
147    /// omitted it.
148    pub hash: Option<String>,
149}
150
151/// A `--direction` choice: one bit applied to every hop.
152#[derive(Debug, Clone, Copy, PartialEq, Eq)]
153pub enum PathDirection {
154    /// Zero-for-one: `true` for each hop.
155    Zfo,
156    /// One-for-zero: `false` for each hop.
157    Ozf,
158}
159
160impl PathDirection {
161    /// The argv spelling.
162    #[must_use]
163    pub const fn as_str(self) -> &'static str {
164        match self {
165            Self::Zfo => "zfo",
166            Self::Ozf => "ozf",
167        }
168    }
169
170    /// Parse the argv spelling (case-sensitive, mirroring click's Choice).
171    #[must_use]
172    pub fn parse(raw: &str) -> Option<Self> {
173        match raw {
174            "zfo" => Some(Self::Zfo),
175            "ozf" => Some(Self::Ozf),
176            _ => None,
177        }
178    }
179
180    /// The per-hop wire bit.
181    #[must_use]
182    pub const fn is_zfo(self) -> bool {
183        matches!(self, Self::Zfo)
184    }
185}
186
187/// Parse one `FAMILY:ADDRESS[:HASH]` hop token into a wire [`PathStep`]
188/// (mirrors `_parse_hop`). Family is case-insensitive; the V4 hash is carried
189/// only for V4 and only when non-empty.
190///
191/// # Errors
192///
193/// [`CliError::OperatorHygiene`] when the family is not V2/V3/V4 or the
194/// address is absent/empty - refused before any socket work.
195pub fn parse_hop_token(hop: &str) -> Result<PathStep, CliError> {
196    let parts: Vec<&str> = hop.split(':').collect();
197    let raw_family = parts.first().copied().unwrap_or_default();
198    let family = PathFamily::parse(raw_family).ok_or_else(|| {
199        CliError::OperatorHygiene(format!("--hop family must be V2|V3|V4, got {raw_family:?}"))
200    })?;
201    let address = parts
202        .get(1)
203        .copied()
204        .filter(|address| !address.is_empty())
205        .ok_or_else(|| CliError::OperatorHygiene(format!("--hop {hop:?} is missing an address")))?;
206    let hash = if family == PathFamily::V4 {
207        parts
208            .get(2)
209            .copied()
210            .filter(|hash| !hash.is_empty())
211            .map(ToString::to_string)
212    } else {
213        None
214    };
215    Ok(PathStep {
216        family,
217        address: address.to_string(),
218        hash,
219    })
220}
221
222/// A `set_fleet_posture` threshold value.
223///
224/// The wire carries integers for the five count/window keys and a float for
225/// `cordon_duty_percent`; `Null` is the `cordon_sim_intake_floor` restore
226/// sentinel.
227#[derive(Debug, Clone, Copy, PartialEq)]
228pub enum PosturePatchValue {
229    /// An integer threshold.
230    Int(u64),
231    /// A float threshold (`cordon_duty_percent`).
232    Float(f64),
233    /// Restore half the slot cap (`cordon_sim_intake_floor` only).
234    Null,
235}
236
237impl PosturePatchValue {
238    /// The JSON value the wire carries. A non-finite float cannot cross JSON
239    /// and degrades to `null` (the host then rejects it - domain validation
240    /// stays server-side).
241    #[must_use]
242    pub fn json_value(self) -> Value {
243        match self {
244            Self::Int(value) => Value::from(value),
245            Self::Float(value) => {
246                serde_json::Number::from_f64(value).map_or(Value::Null, Value::Number)
247            }
248            Self::Null => Value::Null,
249        }
250    }
251}
252
253/// One `cordon_*` entry in a partial posture patch.
254#[derive(Debug, Clone, PartialEq)]
255pub struct PosturePatchEntry {
256    /// One of [`FLEET_POSTURE_THRESHOLD_KEYS`].
257    pub key: String,
258    /// The threshold value.
259    pub value: PosturePatchValue,
260}
261
262impl PosturePatchEntry {
263    /// Build an entry from a key + typed value.
264    #[must_use]
265    pub fn new(key: impl Into<String>, value: PosturePatchValue) -> Self {
266        Self {
267            key: key.into(),
268            value,
269        }
270    }
271
272    /// Build an integer entry.
273    #[must_use]
274    pub fn int(key: impl Into<String>, value: u64) -> Self {
275        Self::new(key, PosturePatchValue::Int(value))
276    }
277
278    /// Build a float entry.
279    #[must_use]
280    pub fn float(key: impl Into<String>, value: f64) -> Self {
281        Self::new(key, PosturePatchValue::Float(value))
282    }
283}
284
285/// Parse the `cordon_sim_intake_floor` flag text: the literal `null`
286/// (case-insensitive) is the documented restore sentinel
287/// ([`SIM_INTAKE_FLOOR_RESTORE`]), anything else must be an unsigned integer.
288///
289/// # Errors
290///
291/// [`CliError::OperatorHygiene`] when the text is neither the sentinel nor an
292/// unsigned integer.
293pub fn parse_sim_intake_floor(raw: &str) -> Result<PosturePatchValue, CliError> {
294    if raw.eq_ignore_ascii_case(SIM_INTAKE_FLOOR_RESTORE) {
295        return Ok(PosturePatchValue::Null);
296    }
297    raw.trim()
298        .parse::<u64>()
299        .map(PosturePatchValue::Int)
300        .map_err(|_| {
301            CliError::OperatorHygiene(format!(
302                "cordon_sim_intake_floor must be an integer or {SIM_INTAKE_FLOOR_RESTORE:?} to                  restore half the slot cap, got {raw:?}"
303            ))
304        })
305}
306
307/// Client-side wire hygiene for a posture patch: no unknown `cordon_*` key and
308/// no empty patch. Mirrors the host's `handle_fleet_posture_op` guard messages;
309/// the host's typed `PosturePolicyPatch::validate` remains the value authority.
310///
311/// # Errors
312///
313/// [`CliError::OperatorHygiene`] with the refused key list or the empty-patch
314/// message.
315pub fn validate_posture_patch(patch: &[PosturePatchEntry]) -> Result<(), CliError> {
316    let mut unknown: Vec<&str> = patch
317        .iter()
318        .map(|entry| entry.key.as_str())
319        .filter(|key| !FLEET_POSTURE_THRESHOLD_KEYS.contains(key))
320        .collect();
321    unknown.sort_unstable();
322    unknown.dedup();
323    if !unknown.is_empty() {
324        return Err(CliError::OperatorHygiene(format!(
325            "unknown fleet-posture threshold key(s): {}",
326            unknown.join(", ")
327        )));
328    }
329    if patch.is_empty() {
330        return Err(CliError::OperatorHygiene(
331            "set_fleet_posture needs at least one threshold key (empty patch)".to_string(),
332        ));
333    }
334    Ok(())
335}
336
337/// The four ops the live host serves, encoded as one request line.
338#[derive(Debug, Clone, PartialEq)]
339pub enum WireRequest {
340    /// `add_path`: enqueue one specific path.
341    AddPath {
342        /// The hop steps, in path order.
343        steps: Vec<PathStep>,
344        /// Per-hop direction bits, or `None` to let the host auto-resolve.
345        directions: Option<Vec<bool>>,
346    },
347    /// `discover`: one bounded on-demand discovery sweep.
348    Discover {
349        /// Maximum paths to process, or `None` for the host's default.
350        bound: Option<u64>,
351    },
352    /// `set_fleet_posture`: a partial patch over the six `cordon_*` keys.
353    SetFleetPosture {
354        /// The patch entries (must already pass [`validate_posture_patch`]).
355        patch: Vec<PosturePatchEntry>,
356    },
357    /// `get_fleet_posture`: read the live thresholds + posture.
358    GetFleetPosture,
359}
360
361impl WireRequest {
362    /// The wire op name.
363    #[must_use]
364    pub const fn op_name(&self) -> &'static str {
365        match self {
366            Self::AddPath { .. } => "add_path",
367            Self::Discover { .. } => "discover",
368            Self::SetFleetPosture { .. } => "set_fleet_posture",
369            Self::GetFleetPosture => "get_fleet_posture",
370        }
371    }
372
373    /// The request payload object.
374    #[must_use]
375    pub fn payload(&self) -> Value {
376        match self {
377            Self::AddPath { steps, directions } => {
378                let steps: Vec<Value> = steps.iter().map(step_json).collect();
379                let directions = directions.as_ref().map_or(Value::Null, |bits| json!(bits));
380                json!({ "steps": steps, "directions": directions })
381            }
382            Self::Discover { bound } => json!({ "bound": bound }),
383            Self::SetFleetPosture { patch } => {
384                let mut payload = Map::new();
385                for entry in patch {
386                    payload.insert(entry.key.clone(), entry.value.json_value());
387                }
388                Value::Object(payload)
389            }
390            Self::GetFleetPosture => json!({}),
391        }
392    }
393
394    /// Encode the one request line the server reads (JSON object + newline).
395    #[must_use]
396    pub fn encode_line(&self) -> String {
397        let mut envelope = Map::new();
398        envelope.insert("op".to_string(), Value::from(self.op_name()));
399        envelope.insert("payload".to_string(), self.payload());
400        let mut line =
401            serde_json::to_string(&Value::Object(envelope)).unwrap_or_else(|_| "{}".to_string());
402        line.push('\n');
403        line
404    }
405}
406
407/// One wire `steps` entry (only the V4 hash key is carried, and only when set).
408fn step_json(step: &PathStep) -> Value {
409    let mut object = Map::new();
410    object.insert("family".to_string(), Value::from(step.family.as_str()));
411    object.insert("address".to_string(), Value::from(step.address.clone()));
412    if let Some(hash) = &step.hash {
413        object.insert("hash".to_string(), Value::from(hash.clone()));
414    }
415    Value::Object(object)
416}
417
418/// A decoded response frame.
419#[derive(Debug, Clone, PartialEq)]
420pub enum WireResponse {
421    /// `{"ok": true, ...}`.
422    Ok {
423        /// The `detail` string (`""` when absent).
424        detail: String,
425        /// The `effective` policy object, when the op echoes one.
426        effective: Option<Value>,
427    },
428    /// `{"ok": false, "error": "..."}` - including the host's unknown-op reply.
429    Err {
430        /// The `error` string.
431        error: String,
432    },
433}
434
435impl WireResponse {
436    /// Whether the host accepted the command.
437    #[must_use]
438    pub const fn is_ok(&self) -> bool {
439        matches!(self, Self::Ok { .. })
440    }
441
442    /// The success detail (`""` for an error frame).
443    #[must_use]
444    pub fn detail(&self) -> &str {
445        match self {
446            Self::Ok { detail, .. } => detail,
447            Self::Err { .. } => "",
448        }
449    }
450
451    /// The echoed effective policy, when present.
452    #[must_use]
453    pub fn effective(&self) -> Option<&Value> {
454        match self {
455            Self::Ok { effective, .. } => effective.as_ref(),
456            Self::Err { .. } => None,
457        }
458    }
459
460    /// The host's error string, when the frame was a refusal.
461    #[must_use]
462    pub fn error(&self) -> Option<&str> {
463        match self {
464            Self::Err { error } => Some(error),
465            Self::Ok { .. } => None,
466        }
467    }
468}
469
470/// Decode one response line. A malformed/non-object/missing-`ok` frame is a
471/// protocol error; a well-formed `{"ok": false, ...}` frame (including the
472/// host's unknown-op reply) is [`WireResponse::Err`], never a panic.
473///
474/// # Errors
475///
476/// [`CliError::OperatorProtocol`] for an empty line, invalid UTF-8/JSON, a
477/// non-object frame, or an `ok` field that is missing/non-boolean.
478pub fn decode_response(line: &str) -> Result<WireResponse, CliError> {
479    let trimmed = line.trim_end_matches(['\n', '\r']);
480    if trimmed.is_empty() {
481        return Err(CliError::OperatorProtocol(
482            "operator host sent an empty response line".to_string(),
483        ));
484    }
485    let value: Value = serde_json::from_str(trimmed).map_err(|err| {
486        CliError::OperatorProtocol(format!("invalid operator response JSON: {err}"))
487    })?;
488    let Some(object) = value.as_object() else {
489        return Err(CliError::OperatorProtocol(
490            "operator response is not a JSON object".to_string(),
491        ));
492    };
493    match object.get("ok") {
494        Some(Value::Bool(true)) => Ok(WireResponse::Ok {
495            detail: object
496                .get("detail")
497                .and_then(Value::as_str)
498                .unwrap_or_default()
499                .to_string(),
500            effective: object.get("effective").cloned(),
501        }),
502        Some(Value::Bool(false)) => Ok(WireResponse::Err {
503            error: object
504                .get("error")
505                .and_then(Value::as_str)
506                .unwrap_or("operator host reported failure without an error message")
507                .to_string(),
508        }),
509        Some(_) => Err(CliError::OperatorProtocol(
510            "operator response 'ok' is not a boolean".to_string(),
511        )),
512        None => Err(CliError::OperatorProtocol(
513            "operator response is missing 'ok'".to_string(),
514        )),
515    }
516}
517
518/// Render an `effective` object as one compact JSON object with its top-level
519/// keys sorted - the Python CLI's `json.dumps(effective, sort_keys=True)`.
520/// `None` renders `{}` (the Python `response.get("effective", {})` default).
521///
522/// The effective policy is a flat object by protocol, so sorting the top level
523/// is equivalent to Python's recursive `sort_keys`.
524#[must_use]
525pub fn render_json_sorted(value: Option<&Value>) -> String {
526    match value {
527        Some(Value::Object(map)) => {
528            let mut entries: Vec<(&String, &Value)> = map.iter().collect();
529            entries.sort_by(|a, b| a.0.cmp(b.0));
530            let inner = entries
531                .iter()
532                .map(|(key, value)| {
533                    let key = serde_json::to_string(key).unwrap_or_default();
534                    let value = serde_json::to_string(value).unwrap_or_default();
535                    format!("{key}: {value}")
536                })
537                .collect::<Vec<_>>()
538                .join(", ");
539            format!("{{{inner}}}")
540        }
541        Some(other) => serde_json::to_string(other).unwrap_or_default(),
542        None => "{}".to_string(),
543    }
544}
545
546/// Resolve the operator socket: `--socket` > `DEGENBOT_OPERATOR_SOCKET` >
547/// `~/.config/degenbot/operator.sock`. The winning value has a leading `~`
548/// expanded against `HOME` through the env seam.
549#[must_use]
550pub fn resolve_socket(env: &dyn EnvVars, cli_socket: Option<&str>) -> PathBuf {
551    let env_value = env.get(SOCKET_ENV);
552    let raw = match (non_empty(cli_socket), non_empty(env_value.as_deref())) {
553        (Some(cli), _) => cli,
554        (None, Some(value)) => value,
555        (None, None) => SOCKET_DEFAULT,
556    };
557    expand_tilde(env, raw)
558}
559
560/// Treat an empty string exactly like an absent layer.
561fn non_empty(value: Option<&str>) -> Option<&str> {
562    value.filter(|value| !value.is_empty())
563}
564
565/// Expand a leading `~` (or `~`/) against `HOME` from the env seam (mirrors
566/// the `degenbot-config` resolver's expansion for the database-path default).
567fn expand_tilde(env: &dyn EnvVars, raw: &str) -> PathBuf {
568    let home = env.get(HOME_ENV).filter(|home| !home.is_empty());
569    if raw == "~" {
570        if let Some(home) = home {
571            return PathBuf::from(home);
572        }
573    } else if let Some(rest) = raw.strip_prefix("~/") {
574        if let Some(home) = home {
575            return Path::new(&home).join(rest);
576        }
577    }
578    PathBuf::from(raw)
579}
580
581/// Send one request to the operator host and decode its single response line.
582///
583/// The arms build a [`WireRequest`] and call this; the transport is a
584/// short-lived `tokio` runtime over a Unix-domain socket (the arms are sync).
585///
586/// # Errors
587///
588/// [`CliError::OperatorProtocol`] when the socket is unreachable, the exchange
589/// times out, or the response is malformed; [`CliError::RuntimeNested`] when
590/// called from inside an existing `tokio` runtime.
591pub fn send_request(socket: &Path, request: &WireRequest) -> Result<WireResponse, CliError> {
592    let line = request.encode_line();
593    exchange_blocking(socket, line)
594}
595
596/// Drive one encoded request line to its response line over the operator
597/// socket.
598#[cfg(unix)]
599fn exchange_blocking(socket: &Path, line: String) -> Result<WireResponse, CliError> {
600    block_on_operator(exchange(socket, line))?
601}
602
603/// Off Unix there is no Unix domain socket to reach: the same
604/// [`CliError::OperatorProtocol`] the unreachable-socket arm raises reports it.
605#[cfg(not(unix))]
606fn exchange_blocking(socket: &Path, line: String) -> Result<WireResponse, CliError> {
607    let _ = (socket, line);
608    Err(CliError::OperatorProtocol(UDS_UNSUPPORTED.to_string()))
609}
610
611/// Drive the exchange future on a self-built runtime, reusing the console's one
612/// runtime helper. Only `RuntimeNested` is reachable from
613/// [`crate::block::block_on`]; any other mapping is a build failure and is
614/// reported as the operator-protocol error it is.
615#[cfg(unix)]
616fn block_on_operator<F: Future>(future: F) -> Result<F::Output, CliError> {
617    crate::block::block_on(future).map_err(|err| match err {
618        CliError::RuntimeNested => err,
619        other => CliError::OperatorProtocol(other.message()),
620    })
621}
622
623/// One request line out, one response line in (the server closes after the
624/// reply). Unix-only: the wire rides a Unix domain socket.
625#[cfg(unix)]
626async fn exchange(socket: &Path, line: String) -> Result<WireResponse, CliError> {
627    let connect = tokio::time::timeout(REQUEST_TIMEOUT, UnixStream::connect(socket))
628        .await
629        .map_err(|_| {
630            CliError::OperatorProtocol(format!(
631                "timed out connecting to operator socket {}",
632                socket.display()
633            ))
634        })?;
635    let mut stream = connect.map_err(|err| {
636        CliError::OperatorProtocol(format!(
637            "cannot reach operator socket {}: {err}",
638            socket.display()
639        ))
640    })?;
641    stream.write_all(line.as_bytes()).await.map_err(|err| {
642        CliError::OperatorProtocol(format!(
643            "failed writing to operator socket {}: {err}",
644            socket.display()
645        ))
646    })?;
647    stream.flush().await.map_err(|err| {
648        CliError::OperatorProtocol(format!(
649            "failed flushing operator socket {}: {err}",
650            socket.display()
651        ))
652    })?;
653    let mut reader = BufReader::new(stream);
654    let mut buffer = Vec::new();
655    let read = tokio::time::timeout(REQUEST_TIMEOUT, reader.read_until(b'\n', &mut buffer))
656        .await
657        .map_err(|_| {
658            CliError::OperatorProtocol(format!(
659                "timed out waiting for a response from operator socket {}",
660                socket.display()
661            ))
662        })?;
663    let read = read.map_err(|err| {
664        CliError::OperatorProtocol(format!(
665            "failed reading from operator socket {}: {err}",
666            socket.display()
667        ))
668    })?;
669    if read == 0 {
670        return Err(CliError::OperatorProtocol(format!(
671            "no response from operator server at {}",
672            socket.display()
673        )));
674    }
675    let text = String::from_utf8(buffer).map_err(|err| {
676        CliError::OperatorProtocol(format!("operator response is not valid UTF-8: {err}"))
677    })?;
678    decode_response(&text)
679}
degenbot_cli_core/operator.rs

degenbot_cli_core/
operator.rs
