agentcat_sdk/
custom.rs

1//! [`CustomEvent`]: an event the server publishes by itself, apart from tool calls, and the two
2//! forms that publish it: [`publish_string`] for the session-string form
3//! on an [`AgentCat`] handle and [`publish_server`] for the server form on a
4//! [`CallContext`](crate::CallContext) or a [`Tracked`](crate::Tracked). Both only queue the event.
5use std::time::{Duration, SystemTime};
6
7use rmcp::model::Implementation;
8use serde_json::{Map, Value};
9use tokio::runtime::Handle;
10
11use crate::client::AgentCat;
12use crate::error::Error;
13use crate::event::ClientIdentity;
14use crate::event::assemble::{EventJob, SDK_LANGUAGE};
15use crate::hooks::PendingHooks;
16use crate::pipeline::EventObject;
17use crate::pipeline::tags::validate_tags;
18use crate::wire;
19
20#[cfg(test)]
21mod tests;
22
23/// The data of a custom event. Every field is optional; each setter consumes the event and
24/// returns it changed.
25///
26/// ```
27/// use std::time::Duration;
28/// use agentcat_sdk::CustomEvent;
29///
30/// let event = CustomEvent::new()
31///     .session_id("ses_1")
32///     .message("cache warmed")
33///     .duration(Duration::from_millis(40));
34/// # let _ = event;
35/// ```
36#[derive(Clone, Debug, Default)]
37pub struct CustomEvent {
38    pub(crate) session_id: Option<String>,
39    pub(crate) resource_name: Option<String>,
40    pub(crate) parameters: Option<Value>,
41    pub(crate) response: Option<Value>,
42    pub(crate) message: Option<String>,
43    pub(crate) duration: Option<Duration>,
44    pub(crate) is_error: Option<bool>,
45    pub(crate) error: Option<Value>,
46    pub(crate) tags: Option<Map<String, Value>>,
47    pub(crate) properties: Option<Map<String, Value>>,
48}
49
50impl CustomEvent {
51    /// An event with no data.
52    #[must_use]
53    pub fn new() -> Self {
54        Self::default()
55    }
56
57    /// The session the event belongs to.
58    #[must_use]
59    pub fn session_id(mut self, id: impl Into<String>) -> Self {
60        self.session_id = Some(id.into());
61        self
62    }
63
64    /// The name of the resource the event is about.
65    #[must_use]
66    pub fn resource_name(mut self, name: impl Into<String>) -> Self {
67        self.resource_name = Some(name.into());
68        self
69    }
70
71    /// The parameters of what happened.
72    #[must_use]
73    pub fn parameters(mut self, parameters: Value) -> Self {
74        self.parameters = Some(parameters);
75        self
76    }
77
78    /// The response of what happened.
79    #[must_use]
80    pub fn response(mut self, response: Value) -> Self {
81        self.response = Some(response);
82        self
83    }
84
85    /// A free text about the event.
86    #[must_use]
87    pub fn message(mut self, message: impl Into<String>) -> Self {
88        self.message = Some(message.into());
89        self
90    }
91
92    /// How long it took.
93    #[must_use]
94    pub fn duration(mut self, duration: Duration) -> Self {
95        self.duration = Some(duration);
96        self
97    }
98
99    /// Whether the event is an error.
100    #[must_use]
101    pub fn is_error(mut self, is_error: bool) -> Self {
102        self.is_error = Some(is_error);
103        self
104    }
105
106    /// The error, when there is one.
107    #[must_use]
108    pub fn error(mut self, error: Value) -> Self {
109        self.error = Some(error);
110        self
111    }
112
113    /// Tags on the event.
114    #[must_use]
115    pub fn tags(mut self, tags: Map<String, Value>) -> Self {
116        self.tags = Some(tags);
117        self
118    }
119
120    /// Properties on the event.
121    #[must_use]
122    pub fn properties(mut self, properties: Map<String, Value>) -> Self {
123        self.properties = Some(properties);
124        self
125    }
126}
127
128/// The event type of every custom event.
129const CUSTOM_EVENT_TYPE: &str = "agentcat:custom";
130
131/// The warning that a server-form event without a session logs.
132const NO_SESSION_WARNING: &str = "publish_custom_event: no session_id provided; event will be published without a session. Pass CustomEvent::session_id to attribute it to a task.";
133
134/// Sets `key` of `event` when there is a value; `None` leaves no key.
135fn put(event: &mut EventObject, key: &str, value: Option<Value>) {
136    if let Some(value) = value {
137        event.insert(key.to_owned(), value);
138    }
139}
140
141/// A duration as an event holds it: whole milliseconds, at most `u64::MAX`.
142fn millis(duration: Duration) -> Value {
143    Value::from(u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
144}
145
146/// The data fields that both forms write after the session, in the event's key order up to
147/// `userIntent`.
148struct Payload {
149    resource_name: Option<String>,
150    parameters: Option<Value>,
151    response: Option<Value>,
152    message: Option<String>,
153}
154
155/// Writes `payload` into `event`, in this key order: `resourceName`, `parameters`, `response`,
156/// `userIntent`.
157fn put_payload(event: &mut EventObject, payload: Payload) {
158    let Payload {
159        resource_name,
160        parameters,
161        response,
162        message,
163    } = payload;
164    put(event, "resourceName", resource_name.map(Value::String));
165    put(event, "parameters", parameters);
166    put(event, "response", response);
167    put(event, "userIntent", message.map(Value::String));
168}
169
170/// A [`CustomEvent`] that passed the checks of both forms ([`accept`]), taken apart once.
171struct Accepted {
172    /// The event's own session, when it is not empty.
173    session_id: Option<String>,
174    payload: Payload,
175    duration: Option<Duration>,
176    is_error: Option<bool>,
177    error: Option<Value>,
178    tags: Option<Map<String, Value>>,
179    properties: Option<Map<String, Value>>,
180}
181
182/// The checks both forms make before they build anything: `Ok(None)` on a handle that
183/// `AGENTCAT_ENABLED` turned off, else `event` taken apart.
184///
185/// # Errors
186///
187/// [`Error::ProjectIdRequired`] for an empty `project_id` on a handle that is on.
188fn accept(
189    agentcat: &AgentCat,
190    project_id: &str,
191    event: CustomEvent,
192) -> Result<Option<Accepted>, Error> {
193    if !agentcat.inner().enabled {
194        return Ok(None);
195    }
196    if project_id.is_empty() {
197        return Err(Error::ProjectIdRequired);
198    }
199    let CustomEvent {
200        session_id,
201        resource_name,
202        parameters,
203        response,
204        message,
205        duration,
206        is_error,
207        error,
208        tags,
209        properties,
210    } = event;
211    Ok(Some(Accepted {
212        session_id: session_id.filter(|own| !own.is_empty()),
213        payload: Payload {
214            resource_name,
215            parameters,
216            response,
217            message,
218        },
219        duration,
220        is_error,
221        error,
222        tags,
223        properties,
224    }))
225}
226
227/// The customer's `tags` and `properties`, last in the event: `tags` is the key with the valid
228/// tags, or `null` when none is valid, whenever the customer set tags (an empty map too);
229/// `properties` only with at least one key. The log lines of dropped tags go to `log`.
230fn put_metadata(
231    event: &mut EventObject,
232    tags: Option<Map<String, Value>>,
233    properties: Option<Map<String, Value>>,
234    log: &mut Vec<String>,
235) {
236    if let Some(tags) = tags {
237        let valid = validate_tags(&tags, log);
238        event.insert("tags".to_owned(), valid.map_or(Value::Null, Value::Object));
239    }
240    if let Some(properties) = properties.filter(|properties| !properties.is_empty()) {
241        event.insert("properties".to_owned(), Value::Object(properties));
242    }
243}
244
245/// Hands the log lines, then (when there is one) the event followed by the `Published custom
246/// event` line, to the handle's sink.
247fn hand_over(agentcat: &AgentCat, log: Vec<String>, job: Option<EventJob>, session_id: &str) {
248    let sink = &agentcat.inner().sink;
249    for line in log {
250        sink.log(line);
251    }
252    let whose = if session_id.is_empty() {
253        "without a session".to_owned()
254    } else {
255        format!("for session {session_id}")
256    };
257    let published = format!("Published custom event {whose} with type '{CUSTOM_EVENT_TYPE}'");
258    match job {
259        Some(job) => sink.event_then_log(job, published),
260        None => sink.log(published),
261    }
262}
263
264/// The session-string form: the event goes to the handle's queue with the project
265/// `project_id` as given, the session of the event's own `session_id` or else `session_id`
266/// verbatim, no redaction hook and no server or client fields. `now` is the event's timestamp.
267/// A handle that `AGENTCAT_ENABLED` turned off returns `Ok(())` at once and builds nothing.
268///
269/// # Errors
270///
271/// [`Error::ProjectIdRequired`] for an empty `project_id` on a handle that is on.
272pub(crate) fn publish_string(
273    agentcat: &AgentCat,
274    session_id: &str,
275    project_id: &str,
276    event: CustomEvent,
277    now: SystemTime,
278) -> Result<(), Error> {
279    let Some(event) = accept(agentcat, project_id, event)? else {
280        return Ok(());
281    };
282    let session = event.session_id.unwrap_or_else(|| session_id.to_owned());
283
284    let mut log = Vec::new();
285    let mut object = EventObject::new();
286    object.insert("sessionId".to_owned(), Value::String(session.clone()));
287    object.insert("projectId".to_owned(), Value::String(project_id.to_owned()));
288    object.insert("eventType".to_owned(), Value::from(CUSTOM_EVENT_TYPE));
289    object.insert("timestamp".to_owned(), Value::String(wire::iso_millis(now)));
290    put_payload(&mut object, event.payload);
291    put(&mut object, "duration", event.duration.map(millis));
292    put(&mut object, "isError", event.is_error.map(Value::Bool));
293    put(&mut object, "error", event.error);
294    put_metadata(&mut object, event.tags, event.properties, &mut log);
295
296    let job = EventJob {
297        event: object,
298        hooks: PendingHooks::default(),
299        hook_mode: false,
300        runtime: None,
301        string_redaction: false,
302        event_redaction: false,
303        backtrace: None,
304    };
305    hand_over(agentcat, log, Some(job), &session);
306    Ok(())
307}
308
309/// The server form: the event with the handle's project (`project_id` only has to be non-empty),
310/// the event's own session or none, the given client and server, the event hook of the options
311/// (never the string hook) on the runtime of the caller, and nothing queued while tracing is off.
312/// `created` is the event's timestamp and `published` the moment the event is published: a
313/// missing or zero `duration` becomes the milliseconds between them. A handle that
314/// `AGENTCAT_ENABLED` turned off returns `Ok(())` at once and builds nothing.
315///
316/// # Errors
317///
318/// [`Error::ProjectIdRequired`] for an empty `project_id` on a handle that is on.
319pub(crate) fn publish_server(
320    agentcat: &AgentCat,
321    project_id: &str,
322    event: CustomEvent,
323    client: Option<&ClientIdentity>,
324    server: Option<&Implementation>,
325    created: SystemTime,
326    published: SystemTime,
327) -> Result<(), Error> {
328    let Some(event) = accept(agentcat, project_id, event)? else {
329        return Ok(());
330    };
331    let mut log = Vec::new();
332    let session = event.session_id.unwrap_or_else(|| {
333        log.push(NO_SESSION_WARNING.to_owned());
334        String::new()
335    });
336    let mut custom = EventObject::new();
337    put_metadata(&mut custom, event.tags, event.properties, &mut log);
338
339    let shared = agentcat.inner();
340    if !shared.options.tracing() {
341        hand_over(agentcat, log, None, &session);
342        return Ok(());
343    }
344
345    let duration = event
346        .duration
347        .filter(|given| !given.is_zero())
348        .unwrap_or_else(|| published.duration_since(created).unwrap_or_default());
349    let text = |value: &str| Value::String(value.to_owned());
350    let mut object = EventObject::new();
351    object.insert("id".to_owned(), text(""));
352    object.insert("sessionId".to_owned(), Value::String(session.clone()));
353    object.insert("projectId".to_owned(), text(&shared.project_id));
354    object.insert("eventType".to_owned(), text(CUSTOM_EVENT_TYPE));
355    object.insert(
356        "timestamp".to_owned(),
357        Value::String(wire::iso_millis(created)),
358    );
359    object.insert("duration".to_owned(), millis(duration));
360    object.insert("sdkLanguage".to_owned(), text(SDK_LANGUAGE));
361    object.insert(
362        "agentcatVersion".to_owned(),
363        text(env!("CARGO_PKG_VERSION")),
364    );
365    put(
366        &mut object,
367        "serverName",
368        server.map(|info| text(&info.name)),
369    );
370    put(
371        &mut object,
372        "serverVersion",
373        server.map(|info| text(&info.version)),
374    );
375    put(
376        &mut object,
377        "clientName",
378        client.and_then(|client| client.name.as_deref()).map(text),
379    );
380    put(
381        &mut object,
382        "clientVersion",
383        client
384            .and_then(|client| client.version.as_deref())
385            .map(text),
386    );
387    object.insert("identifyActorData".to_owned(), Value::Object(Map::new()));
388    put_payload(&mut object, event.payload);
389    put(&mut object, "isError", event.is_error.map(Value::Bool));
390    put(&mut object, "error", event.error);
391    object.extend(custom);
392
393    let job = EventJob {
394        event: object,
395        hooks: PendingHooks::default(),
396        hook_mode: false,
397        runtime: Handle::try_current().ok(),
398        string_redaction: false,
399        event_redaction: true,
400        backtrace: None,
401    };
402    hand_over(agentcat, log, Some(job), &session);
403    Ok(())
404}
agentcat_sdk/custom.rs

agentcat_sdk/
custom.rs
