boxxkite_client/
audit.rs

1//! Audit log: `GET /v1/sandboxes/{id}/log` (paginated) and
2//! `GET /v1/sandboxes/{id}/watch` (a live Server-Sent Events feed). Mirrors
3//! `sdk-python`'s `get_log`/`watch`.
4
5use std::pin::Pin;
6
7use futures_util::{Stream, StreamExt};
8use reqwest::Method;
9use reqwest_eventsource::{Event, EventSource};
10use serde::Deserialize;
11
12use crate::client::Client;
13use crate::error::BoxxkiteError;
14
15/// One row of `GET /v1/sandboxes/{id}/log`'s audit trail -- one exec/file
16/// operation, whether issued by the agent (`source: "agent"`) or by a human
17/// during a `takeover` session (`source: "human_takeover"`).
18#[derive(Debug, Clone, Deserialize)]
19pub struct AuditLogEntry {
20    pub id: String,
21    pub session_id: String,
22    /// `"agent"` or `"human_takeover"`.
23    pub source: String,
24    pub operation: String,
25    pub detail: serde_json::Value,
26    pub exit_code: Option<i64>,
27    pub output_truncated: Option<String>,
28    pub started_at: String,
29    pub duration_ms: i64,
30    /// Hash-chain digest for this row (`docs/TAMPER-EVIDENT-AUDIT-DESIGN.md`),
31    /// or `None` for legacy rows written before hash-chaining was added.
32    pub row_hash: Option<String>,
33    pub prev_hash: Option<String>,
34}
35
36/// `GET /v1/sandboxes/{id}/log`'s response.
37#[derive(Debug, Clone, Deserialize)]
38pub struct AuditLogResponse {
39    pub entries: Vec<AuditLogEntry>,
40    pub limit: i64,
41    pub offset: i64,
42    pub total: i64,
43}
44
45/// Optional `get_log` pagination parameters.
46#[derive(Debug, Clone, Default)]
47pub struct GetLogOptions {
48    limit: Option<u32>,
49    offset: Option<u32>,
50}
51
52impl GetLogOptions {
53    pub fn new() -> Self {
54        Self::default()
55    }
56
57    /// Maximum number of entries to return (1-500, server default 50).
58    pub fn limit(mut self, limit: u32) -> Self {
59        self.limit = Some(limit);
60        self
61    }
62
63    /// Number of entries to skip, oldest-first.
64    pub fn offset(mut self, offset: u32) -> Self {
65        self.offset = Some(offset);
66        self
67    }
68}
69
70impl Client {
71    /// `GET /v1/sandboxes/{id}/log` -- paginated exec/file-op audit history.
72    pub async fn get_log(
73        &self,
74        session_id: &str,
75        options: GetLogOptions,
76    ) -> Result<AuditLogResponse, BoxxkiteError> {
77        let mut query = Vec::new();
78        if let Some(limit) = options.limit {
79            query.push(("limit", limit.to_string()));
80        }
81        if let Some(offset) = options.offset {
82            query.push(("offset", offset.to_string()));
83        }
84        let builder = self
85            .request(Method::GET, &format!("/v1/sandboxes/{session_id}/log"))
86            .query(&query);
87        self.send(builder).await
88    }
89
90    /// `GET /v1/sandboxes/{id}/watch` -- streams new audit-log entries as
91    /// they're written, one [`AuditLogEntry`] per Server-Sent Event. This is
92    /// a live feed of exec/file operations as the control-plane logs them,
93    /// not a live terminal (that's [`Client::takeover`]).
94    ///
95    /// The stream ends when the sandbox session is destroyed or the caller
96    /// drops it. Bring `futures_util::StreamExt` (or any other `Stream`
97    /// combinator crate) into scope to consume it:
98    ///
99    /// ```no_run
100    /// # use futures_util::StreamExt;
101    /// # async fn example(client: boxxkite_client::Client, session_id: &str) {
102    /// let mut entries = client.watch(session_id);
103    /// while let Some(entry) = entries.next().await {
104    ///     match entry {
105    ///         Ok(entry) => println!("{entry:?}"),
106    ///         Err(err) => {
107    ///             eprintln!("watch error: {err}");
108    ///             break;
109    ///         }
110    ///     }
111    /// }
112    /// # }
113    /// ```
114    pub fn watch(
115        &self,
116        session_id: &str,
117    ) -> Pin<Box<dyn Stream<Item = Result<AuditLogEntry, BoxxkiteError>> + Send + 'static>> {
118        let request_builder =
119            self.request(Method::GET, &format!("/v1/sandboxes/{session_id}/watch"));
120
121        Box::pin(async_stream::stream! {
122            let mut event_source = match EventSource::new(request_builder) {
123                Ok(event_source) => event_source,
124                Err(err) => {
125                    yield Err(BoxxkiteError::Config(format!("failed to build watch request: {err}")));
126                    return;
127                }
128            };
129
130            while let Some(event) = event_source.next().await {
131                match event {
132                    Ok(Event::Open) => continue,
133                    Ok(Event::Message(message)) => {
134                        yield serde_json::from_str::<AuditLogEntry>(&message.data).map_err(BoxxkiteError::from);
135                    }
136                    Err(reqwest_eventsource::Error::StreamEnded) => break,
137                    Err(reqwest_eventsource::Error::InvalidStatusCode(status, response)) => {
138                        let bytes = response.bytes().await.unwrap_or_default();
139                        yield Err(crate::error::api_error_from_bytes(status.as_u16(), &bytes));
140                        break;
141                    }
142                    Err(err) => {
143                        yield Err(BoxxkiteError::from(err));
144                        break;
145                    }
146                }
147            }
148            event_source.close();
149        })
150    }
151}
boxxkite_client/audit.rs

boxxkite_client/
audit.rs
