Skip to main content

scv_client/
lib.rs

1//! What every local client of the SCV daemon needs, without depending on the
2//! server: the instance [`Layout`] (every path under `SCV_HOME`, and the
3//! instance's service unit name), the delegation-depth variable a delegated SCV
4//! inherits, the chat log ([`history`]), framed reading and writing
5//! ([`Connection`], [`read_frame`]),
6//! private instance files ([`fs::replace_private`]), [`Secret`] values
7//! that never print, byte-bounded text
8//! ([`text::utf8_prefix`]), and [`control`] for daemon management requests,
9//! which fail with a typed [`ControlError`].
10
11#![forbid(unsafe_code)]
12
13mod connection;
14pub mod fs;
15pub mod history;
16mod layout;
17mod secret;
18pub mod text;
19pub use connection::{Connection, read_frame, write_message};
20pub use layout::{Layout, Stray};
21pub use secret::Secret;
22
23use anyhow::Result;
24use scv_protocol::{
25    ClientMessage, DaemonCommand, DaemonStatus, ErrorCode, Frame, FrameDecoder, Overflow,
26    PROTOCOL_VERSION, ServerEvent,
27};
28use std::{fmt, path::Path, time::Duration};
29use tokio::{io::BufReader, net::UnixStream};
30
31/// Largest management reply, counting its line ending. Workbench actions
32/// reply with a snapshot the daemon may send up to its 8 MiB frame limit;
33/// a smaller cap here reports failure after the action already took effect.
34const MAX_CONTROL_FRAME_BYTES: usize = 8 * 1024 * 1024;
35
36/// Environment variable carrying a delegated process's depth; SCV sets it on
37/// every agent it starts.
38pub const DELEGATION_DEPTH_VARIABLE: &str = "SCV_DELEGATION_DEPTH";
39
40/// The delegation depth to declare in `session.start`: this process's own,
41/// when an SCV started it, so a delegated client cannot reset the count by
42/// connecting to a daemon.
43pub fn inherited_delegation_depth() -> Option<u32> {
44    parse_delegation_depth(std::env::var(DELEGATION_DEPTH_VARIABLE).ok().as_deref())
45}
46
47fn parse_delegation_depth(value: Option<&str>) -> Option<u32> {
48    value
49        .and_then(|value| value.trim().parse().ok())
50        .filter(|depth| *depth > 0)
51}
52
53/// Why a [`control`] request failed.
54#[derive(Debug)]
55#[non_exhaustive]
56pub enum ControlError {
57    /// No daemon accepted the connection: it is not running, or the socket
58    /// is stale. Nothing was sent.
59    Unavailable(std::io::Error),
60    /// The daemon refused the request. A daemon that predates a command
61    /// refuses it with [`ErrorCode::InvalidJson`], as a frame it cannot parse.
62    Server {
63        /// Why, as the daemon's stable code.
64        code: ErrorCode,
65        /// What went wrong, for people.
66        message: String,
67    },
68    /// No answer within the helper's time limit. The daemon may still carry
69    /// out a mutation; query status before retrying.
70    TimedOut,
71    /// The exchange broke off, or the daemon answered something this client
72    /// does not understand. A mutation's outcome is unknown.
73    Protocol(String),
74}
75
76impl fmt::Display for ControlError {
77    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
78        match self {
79            Self::Unavailable(_) => formatter
80                .write_str("SCV daemon unavailable; start it with `scv start` or `scv run`"),
81            Self::Server { message, .. } => formatter.write_str(message),
82            Self::TimedOut => formatter
83                .write_str("SCV management request timed out; query status before retrying"),
84            Self::Protocol(message) => formatter.write_str(message),
85        }
86    }
87}
88
89impl std::error::Error for ControlError {
90    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
91        match self {
92            Self::Unavailable(error) => Some(error),
93            _ => None,
94        }
95    }
96}
97
98/// How long a management exchange may take. A command that can stop a
99/// running account waits for it to stop, and an email account finishes a
100/// mail action under way first, for up to 65 seconds; a workbench action
101/// that may run `git` on worktrees (each command up to 30 seconds), or
102/// wait for an agent to accept steering (up to 30 seconds), gets 90.
103pub fn control_timeout(command: &DaemonCommand) -> Duration {
104    match command {
105        DaemonCommand::Reload
106        | DaemonCommand::ChannelSet { .. }
107        | DaemonCommand::ChannelLogout { .. }
108        | DaemonCommand::WorkbenchStart { .. }
109        | DaemonCommand::WorkbenchResume { .. }
110        | DaemonCommand::WorkbenchSend { .. }
111        | DaemonCommand::Worktrees
112        | DaemonCommand::WorktreeRemove { .. } => Duration::from_secs(90),
113        _ => Duration::from_secs(20),
114    }
115}
116
117/// A bounded management exchange. Never retries mutations on ambiguous failure.
118pub async fn control(path: &Path, command: DaemonCommand) -> Result<DaemonStatus, ControlError> {
119    let broken = |error: std::io::Error| ControlError::Protocol(format!("{error}"));
120    tokio::time::timeout(control_timeout(&command), async {
121        let stream = UnixStream::connect(path)
122            .await
123            .map_err(ControlError::Unavailable)?;
124        let (reader, writer) = stream.into_split();
125        let mut connection = Connection::new(
126            BufReader::new(reader),
127            writer,
128            FrameDecoder::new(MAX_CONTROL_FRAME_BYTES, Overflow::Stop),
129        );
130        for message in [
131            ClientMessage::initialize("init", "scv-control"),
132            ClientMessage::DaemonControl {
133                request_id: "control".into(),
134                command,
135            },
136        ] {
137            connection.send(&message).await.map_err(broken)?;
138            let bytes = match connection.read().await.map_err(broken)? {
139                Frame::Line(bytes) => bytes,
140                Frame::TooLarge => {
141                    return Err(ControlError::Protocol(
142                        "SCV status exceeds frame limit".into(),
143                    ));
144                }
145                Frame::End | Frame::Truncated(_) => {
146                    return Err(ControlError::Protocol(
147                        "SCV daemon closed the management connection".into(),
148                    ));
149                }
150            };
151            let event = serde_json::from_slice::<ServerEvent>(&bytes)
152                .map_err(|error| ControlError::Protocol(format!("{error}")))?;
153            match event {
154                ServerEvent::Initialized {
155                    protocol_version: PROTOCOL_VERSION,
156                    ..
157                } if matches!(message, ClientMessage::Initialize { .. }) => {}
158                ServerEvent::DaemonStatus { status, .. } => return Ok(status),
159                ServerEvent::Error { code, message, .. } => {
160                    return Err(ControlError::Server { code, message });
161                }
162                _ => {
163                    return Err(ControlError::Protocol(
164                        "unexpected SCV management response; upgrade/restart the daemon".into(),
165                    ));
166                }
167            }
168        }
169        Err(ControlError::Protocol("SCV daemon omitted status".into()))
170    })
171    .await
172    .unwrap_or(Err(ControlError::TimedOut))
173}
174
175#[cfg(test)]
176mod tests;