running_process/client/
telemetry.rs

1//! Client helpers for daemon-owned session tee telemetry.
2
3use std::path::{Path, PathBuf};
4
5use crate::client::{ClientError, DaemonClient};
6use crate::proto::daemon::{
7    DaemonRequest, GetSessionTeeStatusRequest, RegisterSessionTeeRequest, RequestType, StatusCode,
8    TeeBackpressure as ProtoTeeBackpressure, TeeFileMode as ProtoTeeFileMode,
9    TeeSessionKind as ProtoTeeSessionKind, TeeSinkKind, TeeStreamKind as ProtoTeeStreamKind,
10    UnregisterSessionTeeRequest,
11};
12
13#[derive(Clone, Copy, Debug, Eq, PartialEq)]
14/// Session transport that owns the stream being tee'd.
15pub enum SessionTeeKind {
16    /// Daemon-owned pseudo-terminal session.
17    Pty,
18    /// Daemon-owned pipe-backed session.
19    Pipe,
20}
21
22#[derive(Clone, Copy, Debug, Eq, PartialEq)]
23/// Session byte stream that can be mirrored to a tee sink.
24pub enum SessionTeeStream {
25    /// Combined PTY output bytes.
26    PtyOutput,
27    /// Pipe session standard output bytes.
28    Stdout,
29    /// Pipe session standard error bytes.
30    Stderr,
31    /// Bytes written successfully to a pipe session's standard input.
32    Stdin,
33}
34
35#[derive(Clone, Copy, Debug, Eq, PartialEq)]
36/// File open mode used when registering a file tee.
37pub enum SessionTeeFileMode {
38    /// Append tee output to the file if it already exists.
39    Append,
40    /// Truncate the file before writing tee output.
41    Truncate,
42}
43
44#[derive(Clone, Copy, Debug, Eq, PartialEq)]
45/// Backpressure policy for bounded tee queues.
46pub enum SessionTeeBackpressure {
47    /// Keep the session reader non-blocking and account for dropped bytes.
48    DropOldest,
49    /// Block the session reader until the tee sink accepts more bytes.
50    Block,
51}
52
53#[derive(Clone, Debug, Eq, PartialEq)]
54/// Request used to register a daemon-managed file tee for a session stream.
55pub struct SessionTeeFileRequest {
56    /// Session identifier returned by the spawn API.
57    pub session_id: String,
58    /// Kind of session that owns `session_id`.
59    pub session_kind: SessionTeeKind,
60    /// Stream to mirror into the file.
61    pub stream: SessionTeeStream,
62    /// Destination file path on the daemon host.
63    pub path: PathBuf,
64    /// File open mode for the destination path.
65    pub mode: SessionTeeFileMode,
66    /// 0 means use the daemon default.
67    pub queue_capacity: u32,
68    /// Whether the daemon writes marker lines when tee bytes are missed.
69    pub write_missed_markers: bool,
70    /// Queue behavior when the file sink cannot keep up.
71    pub backpressure: SessionTeeBackpressure,
72}
73
74impl SessionTeeFileRequest {
75    /// Create a file tee request with append mode and daemon-default queue size.
76    pub fn new<P>(
77        session_id: impl Into<String>,
78        session_kind: SessionTeeKind,
79        stream: SessionTeeStream,
80        path: P,
81    ) -> Self
82    where
83        P: AsRef<Path>,
84    {
85        Self {
86            session_id: session_id.into(),
87            session_kind,
88            stream,
89            path: path.as_ref().to_path_buf(),
90            mode: SessionTeeFileMode::Append,
91            queue_capacity: 0,
92            write_missed_markers: true,
93            backpressure: SessionTeeBackpressure::DropOldest,
94        }
95    }
96
97    /// Open the destination file in truncate mode instead of append mode.
98    pub fn truncate(mut self) -> Self {
99        self.mode = SessionTeeFileMode::Truncate;
100        self
101    }
102
103    /// Set the bounded queue capacity; `0` keeps the daemon default.
104    pub fn queue_capacity(mut self, capacity: u32) -> Self {
105        self.queue_capacity = capacity;
106        self
107    }
108
109    /// Disable marker lines for bytes missed by the file tee.
110    pub fn suppress_missed_markers(mut self) -> Self {
111        self.write_missed_markers = false;
112        self
113    }
114
115    /// Set the queue backpressure policy for this tee.
116    pub fn backpressure(mut self, backpressure: SessionTeeBackpressure) -> Self {
117        self.backpressure = backpressure;
118        self
119    }
120}
121
122#[derive(Clone, Copy, Debug, Eq, PartialEq)]
123/// Current daemon status for a registered session tee.
124pub struct SessionTeeStatus {
125    /// Stream associated with the registered tee handle.
126    pub stream: SessionTeeStream,
127    /// Number of stream bytes missed by this tee.
128    pub missed_bytes: u64,
129    /// Whether the tee sink has disconnected from the session stream.
130    pub disconnected: bool,
131}
132
133impl DaemonClient {
134    /// Register a daemon-owned file tee and return its opaque tee handle.
135    pub fn register_session_file_tee(
136        &mut self,
137        request: &SessionTeeFileRequest,
138    ) -> Result<u64, ClientError> {
139        let daemon_request = DaemonRequest {
140            id: self.next_request_id(),
141            r#type: RequestType::RegisterSessionTee.into(),
142            protocol_version: 1,
143            register_session_tee: Some(RegisterSessionTeeRequest {
144                session_id: request.session_id.clone(),
145                session_kind: proto_session_kind(request.session_kind) as i32,
146                stream: proto_stream_kind(request.stream) as i32,
147                sink_kind: TeeSinkKind::File as i32,
148                file_path: crate::platform::fs::encode_path_bytes(&request.path),
149                file_mode: proto_file_mode(request.mode) as i32,
150                queue_capacity: request.queue_capacity,
151                suppress_missed_markers: !request.write_missed_markers,
152                backpressure: proto_backpressure(request.backpressure) as i32,
153            }),
154            ..Default::default()
155        };
156        let response = self.send_request(daemon_request)?;
157        ensure_ok(&response)?;
158        let payload = response
159            .register_session_tee
160            .ok_or_else(|| ClientError::Server {
161                code: StatusCode::Internal,
162                message: "register_session_tee response missing payload".into(),
163            })?;
164        Ok(payload.tee_handle)
165    }
166
167    /// Unregister a previously registered session tee handle.
168    pub fn unregister_session_tee(
169        &mut self,
170        session_kind: SessionTeeKind,
171        session_id: &str,
172        tee_handle: u64,
173    ) -> Result<(), ClientError> {
174        let daemon_request = DaemonRequest {
175            id: self.next_request_id(),
176            r#type: RequestType::UnregisterSessionTee.into(),
177            protocol_version: 1,
178            unregister_session_tee: Some(UnregisterSessionTeeRequest {
179                session_id: session_id.to_string(),
180                session_kind: proto_session_kind(session_kind) as i32,
181                tee_handle,
182            }),
183            ..Default::default()
184        };
185        let response = self.send_request(daemon_request)?;
186        ensure_ok(&response)
187    }
188
189    /// Fetch the current status for a registered session tee handle.
190    pub fn get_session_tee_status(
191        &mut self,
192        session_kind: SessionTeeKind,
193        session_id: &str,
194        tee_handle: u64,
195    ) -> Result<SessionTeeStatus, ClientError> {
196        let daemon_request = DaemonRequest {
197            id: self.next_request_id(),
198            r#type: RequestType::GetSessionTeeStatus.into(),
199            protocol_version: 1,
200            get_session_tee_status: Some(GetSessionTeeStatusRequest {
201                session_id: session_id.to_string(),
202                session_kind: proto_session_kind(session_kind) as i32,
203                tee_handle,
204            }),
205            ..Default::default()
206        };
207        let response = self.send_request(daemon_request)?;
208        ensure_ok(&response)?;
209        let payload = response
210            .get_session_tee_status
211            .ok_or_else(|| ClientError::Server {
212                code: StatusCode::Internal,
213                message: "get_session_tee_status response missing payload".into(),
214            })?;
215        let stream =
216            ProtoTeeStreamKind::try_from(payload.stream).map_err(|_| ClientError::Server {
217                code: StatusCode::Internal,
218                message: "get_session_tee_status response has invalid stream".into(),
219            })?;
220        Ok(SessionTeeStatus {
221            stream: client_stream_kind(stream)?,
222            missed_bytes: payload.missed_bytes,
223            disconnected: payload.disconnected,
224        })
225    }
226}
227
228fn ensure_ok(response: &crate::proto::daemon::DaemonResponse) -> Result<(), ClientError> {
229    if response.code == StatusCode::Ok as i32 {
230        return Ok(());
231    }
232    let code = StatusCode::try_from(response.code).unwrap_or(StatusCode::UnknownRequest);
233    Err(ClientError::Server {
234        code,
235        message: response.message.clone(),
236    })
237}
238
239fn proto_session_kind(kind: SessionTeeKind) -> ProtoTeeSessionKind {
240    match kind {
241        SessionTeeKind::Pty => ProtoTeeSessionKind::Pty,
242        SessionTeeKind::Pipe => ProtoTeeSessionKind::Pipe,
243    }
244}
245
246fn proto_stream_kind(stream: SessionTeeStream) -> ProtoTeeStreamKind {
247    match stream {
248        SessionTeeStream::PtyOutput => ProtoTeeStreamKind::PtyOutput,
249        SessionTeeStream::Stdout => ProtoTeeStreamKind::Stdout,
250        SessionTeeStream::Stderr => ProtoTeeStreamKind::Stderr,
251        SessionTeeStream::Stdin => ProtoTeeStreamKind::Stdin,
252    }
253}
254
255fn client_stream_kind(stream: ProtoTeeStreamKind) -> Result<SessionTeeStream, ClientError> {
256    match stream {
257        ProtoTeeStreamKind::PtyOutput => Ok(SessionTeeStream::PtyOutput),
258        ProtoTeeStreamKind::Stdout => Ok(SessionTeeStream::Stdout),
259        ProtoTeeStreamKind::Stderr => Ok(SessionTeeStream::Stderr),
260        ProtoTeeStreamKind::Stdin => Ok(SessionTeeStream::Stdin),
261        ProtoTeeStreamKind::Unspecified => Err(ClientError::Server {
262            code: StatusCode::Internal,
263            message: "get_session_tee_status response has unspecified stream".into(),
264        }),
265    }
266}
267
268fn proto_file_mode(mode: SessionTeeFileMode) -> ProtoTeeFileMode {
269    match mode {
270        SessionTeeFileMode::Append => ProtoTeeFileMode::Append,
271        SessionTeeFileMode::Truncate => ProtoTeeFileMode::Truncate,
272    }
273}
274
275fn proto_backpressure(backpressure: SessionTeeBackpressure) -> ProtoTeeBackpressure {
276    match backpressure {
277        SessionTeeBackpressure::DropOldest => ProtoTeeBackpressure::DropOldest,
278        SessionTeeBackpressure::Block => ProtoTeeBackpressure::Block,
279    }
280}
281
282#[cfg(test)]
283#[path = "../tests/client_telemetry_coverage.rs"]
284mod coverage_tests;
running_process/client/telemetry.rs

running_process/client/
telemetry.rs
