nemo_relay/api/
scope.rs

1// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4use crate::api::event::{BaseEvent, CategoryProfile, DataSchema, EventCategory, MarkEvent};
5use crate::api::runtime::global_context;
6use crate::api::runtime::scope_stack::snapshot_scope_stack;
7use crate::api::runtime::subscriber_dispatcher;
8use crate::api::runtime::{
9    current_scope_stack, task_scope_push, task_scope_remove, task_scope_top,
10};
11use crate::api::shared::{
12    ensure_runtime_owner, resolve_parent_uuid, snapshot_event_sanitizers,
13    snapshot_event_subscribers,
14};
15use crate::error::{FlowError, Result};
16use crate::json::Json;
17use chrono::{DateTime, Utc};
18use serde::{Deserialize, Serialize};
19use typed_builder::TypedBuilder;
20use uuid::Uuid;
21
22pub use nemo_relay_types::api::scope::{HandleAttributes, ScopeAttributes, ScopeType};
23
24/// Canonical mark-event name used to indicate agent context compaction.
25pub const COMPACTION_EVENT_NAME: &str = "compaction";
26
27/// Runtime-owned handle identifying an active or completed scope.
28#[derive(Debug, Clone, Serialize, Deserialize, TypedBuilder)]
29#[builder(field_defaults(setter(strip_option(ignore_invalid, fallback_suffix = "_opt"))))]
30pub struct ScopeHandle {
31    /// Unique scope identifier.
32    #[builder(default = Uuid::now_v7())]
33    pub uuid: Uuid,
34    /// Timestamp captured when the scope handle was created.
35    #[builder(default = Utc::now())]
36    pub started_at: DateTime<Utc>,
37    /// Semantic category of the scope.
38    pub scope_type: ScopeType,
39    /// Human-readable scope name.
40    #[builder(setter(into))]
41    pub name: String,
42    /// Optional application payload stored on the handle.
43    #[builder(default)]
44    pub data: Option<Json>,
45    /// Optional metadata attached to the scope.
46    #[builder(default)]
47    pub metadata: Option<Json>,
48    /// Scope behavior flags.
49    #[builder(default = ScopeAttributes::empty())]
50    pub attributes: ScopeAttributes,
51    /// UUID of the parent scope, if any.
52    #[builder(default)]
53    pub parent_uuid: Option<Uuid>,
54}
55
56fn scope_stack_lock_error(error: impl std::fmt::Display, operation: &'static str) -> FlowError {
57    log::error!(
58        target: "nemo_relay.runtime",
59        event = "scope_stack_unavailable",
60        operation = operation;
61        "Scope operation failed because the scope stack lock is poisoned: {error}"
62    );
63    FlowError::Internal(error.to_string())
64}
65
66/// Builder parameters for [`push_scope`].
67#[derive(TypedBuilder)]
68#[builder(field_defaults(setter(strip_option(ignore_invalid, fallback_suffix = "_opt"))))]
69pub struct PushScopeParams<'a> {
70    /// Human-readable scope name recorded on emitted lifecycle events.
71    pub name: &'a str,
72    /// Semantic category for the new scope.
73    pub scope_type: ScopeType,
74    /// Optional explicit parent scope.
75    #[builder(default)]
76    pub parent: Option<&'a ScopeHandle>,
77    /// Scope attribute bitflags applied to the new scope.
78    #[builder(default = ScopeAttributes::empty())]
79    pub attributes: ScopeAttributes,
80    /// Optional application payload stored on the scope handle.
81    #[builder(default)]
82    pub data: Option<Json>,
83    /// Optional JSON metadata recorded on the emitted start event.
84    #[builder(default)]
85    pub metadata: Option<Json>,
86    /// Optional JSON payload exported as the scope start event data.
87    #[builder(default)]
88    pub input: Option<Json>,
89    /// Optional timestamp recorded on the emitted start event.
90    #[builder(default)]
91    pub timestamp: Option<DateTime<Utc>>,
92}
93
94/// Builder parameters for [`NemoRelayContextState::create_scope_handle`].
95#[derive(Debug, Clone, TypedBuilder)]
96#[builder(field_defaults(setter(strip_option(ignore_invalid, fallback_suffix = "_opt"))))]
97pub struct CreateScopeHandleParams<'a> {
98    /// Human-readable scope name.
99    pub name: &'a str,
100    /// Optional parent scope UUID.
101    #[builder(default)]
102    pub parent_uuid: Option<Uuid>,
103    /// Semantic category of the scope.
104    pub scope_type: ScopeType,
105    /// Scope attribute bitflags.
106    #[builder(default = ScopeAttributes::empty())]
107    pub attributes: ScopeAttributes,
108    /// Optional application payload stored on the handle.
109    #[builder(default)]
110    pub data: Option<Json>,
111    /// Optional metadata stored on the handle.
112    #[builder(default)]
113    pub metadata: Option<Json>,
114    /// Optional timestamp captured as the handle start time and reused by the
115    /// emitted start event. When omitted, the current UTC time is used.
116    #[builder(default)]
117    pub timestamp: Option<DateTime<Utc>>,
118}
119
120/// Builder parameters for [`NemoRelayContextState::build_scope_end_event`].
121#[derive(Debug, Clone, TypedBuilder)]
122#[builder(field_defaults(setter(strip_option(ignore_invalid, fallback_suffix = "_opt"))))]
123pub struct EndScopeHandleParams<'a> {
124    /// Scope handle to serialize into the emitted end event.
125    pub handle: &'a ScopeHandle,
126    /// Optional JSON payload exported as the semantic scope output.
127    #[builder(default)]
128    pub data: Option<Json>,
129    /// Optional metadata to be appended to the metadata set when the scope was created.
130    #[builder(default)]
131    pub metadata: Option<Json>,
132    /// Optional timestamp recorded on the emitted end event. When omitted, the
133    /// runtime records the current UTC time, or one microsecond after the
134    /// handle start time if the current time is not later.
135    #[builder(default)]
136    pub timestamp: Option<DateTime<Utc>>,
137}
138
139/// Builder parameters for [`pop_scope`].
140#[derive(TypedBuilder)]
141#[builder(field_defaults(setter(strip_option(ignore_invalid, fallback_suffix = "_opt"))))]
142pub struct PopScopeParams<'a> {
143    /// UUID of the scope that should be popped.
144    pub handle_uuid: &'a Uuid,
145    /// Optional JSON payload exported as the semantic scope output.
146    #[builder(default)]
147    pub output: Option<Json>,
148    /// Optional JSON payload metadata to be appended to the metadata set when the scope was created.
149    #[builder(default)]
150    pub metadata: Option<Json>,
151    /// Optional timestamp recorded on the emitted end event. When omitted, the
152    /// runtime records the current UTC time, or one microsecond after the
153    /// handle start time if the current time is not later.
154    #[builder(default)]
155    pub timestamp: Option<DateTime<Utc>>,
156}
157
158/// Builder parameters for [`event`].
159#[derive(TypedBuilder)]
160#[builder(field_defaults(setter(strip_option(ignore_invalid, fallback_suffix = "_opt"))))]
161pub struct EmitMarkEventParams<'a> {
162    /// Event name to emit.
163    pub name: &'a str,
164    /// Optional explicit parent scope.
165    #[builder(default)]
166    pub parent: Option<&'a ScopeHandle>,
167    /// Optional JSON payload recorded as the mark data.
168    #[builder(default)]
169    pub data: Option<Json>,
170    /// Optional schema identifier for the mark data.
171    #[builder(default)]
172    pub data_schema: Option<DataSchema>,
173    /// Optional JSON metadata recorded on the emitted event.
174    #[builder(default)]
175    pub metadata: Option<Json>,
176    /// Optional semantic category for the mark.
177    #[builder(default)]
178    pub category: Option<EventCategory>,
179    /// Optional category-specific mark profile.
180    #[builder(default)]
181    pub category_profile: Option<CategoryProfile>,
182    /// Optional timestamp recorded on the emitted mark event. When omitted, the
183    /// current UTC time is used.
184    #[builder(default)]
185    pub timestamp: Option<DateTime<Utc>>,
186}
187
188/// Return the current scope at the top of the active stack.
189///
190/// This reads the task-local or thread-local scope stack without mutating it
191/// and returns a clone of the current top-most [`ScopeHandle`].
192///
193/// # Returns
194/// A [`Result`] containing the current [`ScopeHandle`] when the runtime owner
195/// check succeeds.
196///
197/// # Errors
198/// Returns an error when the current binding has not initialized the shared
199/// runtime ownership correctly.
200pub fn get_handle() -> Result<ScopeHandle> {
201    ensure_runtime_owner()?;
202    Ok(task_scope_top())
203}
204
205/// Push a new scope onto the active scope stack.
206///
207/// This creates a new [`ScopeHandle`], emits a scope-start event to global and
208/// scope-local subscribers, and makes the new scope the current top of stack.
209///
210/// # Parameters
211/// - `name`: Human-readable scope name recorded on emitted lifecycle events.
212/// - `scope_type`: Semantic category for the new scope.
213/// - `parent`: Optional explicit parent scope. When `None`, the current top of
214///   stack is used as the parent.
215/// - `attributes`: Bitflags that modify scope behavior and observability.
216/// - `data`: Optional application payload stored on the returned handle.
217/// - `metadata`: Optional JSON metadata recorded on the emitted start event.
218/// - `input`: Optional JSON payload exported as the Agent Trajectory
219///   Observability Format (ATOF) data payload.
220/// - `timestamp`: Optional timestamp recorded as the handle start time and on
221///   the emitted start event. When `None`, the current UTC time is used.
222///
223/// # Returns
224/// A [`Result`] containing the newly created [`ScopeHandle`].
225///
226/// # Errors
227/// Returns an error when the runtime owner check fails or when internal state
228/// cannot be read safely.
229///
230/// # Notes
231/// The start event is queued with subscriber and sanitizer snapshots captured
232/// while the new scope is active.
233pub fn push_scope(params: PushScopeParams<'_>) -> Result<ScopeHandle> {
234    ensure_runtime_owner()?;
235    let parent_uuid = resolve_parent_uuid(params.parent);
236    let (handle, event, subscribers, emission_scope_stack) = {
237        let scope_stack = current_scope_stack();
238        let scope_guard = scope_stack
239            .read()
240            .map_err(|error| scope_stack_lock_error(error, "push"))?;
241        let scope_subscribers = scope_guard.collect_scope_local_subscribers();
242        let subscribers = snapshot_event_subscribers(scope_subscribers)?;
243        let context = global_context();
244        let state = context
245            .read()
246            .map_err(|error| FlowError::Internal(error.to_string()))?;
247        let handle_params = CreateScopeHandleParams::builder()
248            .name(params.name)
249            .parent_uuid_opt(parent_uuid)
250            .scope_type(params.scope_type)
251            .attributes(params.attributes)
252            .data_opt(params.data)
253            .metadata_opt(params.metadata)
254            .timestamp_opt(params.timestamp)
255            .build();
256        let handle = state.create_scope_handle(handle_params);
257        let event = state.build_scope_start_event(&handle, params.input);
258        (handle, event, subscribers, scope_stack.clone())
259    };
260    task_scope_push(handle.clone());
261    let sanitizers = snapshot_event_sanitizers(&event, &emission_scope_stack).unwrap_or_default();
262    let _ = subscriber_dispatcher::dispatch_sanitized_event(
263        event,
264        sanitizers,
265        &subscribers,
266        emission_scope_stack,
267    );
268    Ok(handle)
269}
270
271/// Pop the current scope from the active scope stack.
272///
273/// This emits a scope-end event for the target scope and removes any
274/// scope-local registrations owned by that scope.
275///
276/// # Parameters
277/// - `handle_uuid`: UUID of the scope that should be popped.
278/// - `output`: Optional JSON payload exported as the semantic scope output.
279/// - `timestamp`: Optional timestamp recorded on the emitted end event. When
280///   `None`, the runtime uses the current UTC time, or one microsecond after
281///   the handle start time if the current time is not later.
282///
283/// # Returns
284/// A [`Result`] that is `Ok(())` when the scope was popped successfully.
285///
286/// # Errors
287/// Returns [`FlowError::InvalidArgument`] when the target scope exists but is
288/// not the current top of stack, and [`FlowError::NotFound`] when the UUID is
289/// unknown to the active stack.
290///
291/// # Notes
292/// The implicit root scope cannot be removed.
293///
294/// Scope-end emission snapshots the visible scope-local sanitizers before
295/// removing the scope. Publication is then queued after removal using that
296/// snapshot, so cleanup does not change the middleware applied to the emitted
297/// event.
298pub fn pop_scope(params: PopScopeParams<'_>) -> Result<()> {
299    ensure_runtime_owner()?;
300    let scope_stack = current_scope_stack();
301    let (scope, event, subscribers, emission_scope_stack) = {
302        let scope_guard = scope_stack
303            .read()
304            .map_err(|error| scope_stack_lock_error(error, "pop"))?;
305        let top = scope_guard.top();
306        if top.uuid != *params.handle_uuid {
307            if scope_guard.find(params.handle_uuid).is_some() {
308                return Err(FlowError::InvalidArgument(
309                    "scope handle is not at the top of the stack".into(),
310                ));
311            }
312            return Err(FlowError::NotFound("scope handle not found".into()));
313        }
314        let scope_subscribers = scope_guard.collect_scope_local_subscribers();
315        let subscribers = snapshot_event_subscribers(scope_subscribers)?;
316        let scope = top.clone();
317        let context = global_context();
318        let state = context
319            .read()
320            .map_err(|error| FlowError::Internal(error.to_string()))?;
321        let event = state.build_scope_end_event(
322            EndScopeHandleParams::builder()
323                .handle(&scope)
324                .data_opt(params.output)
325                .timestamp_opt(params.timestamp)
326                .metadata_opt(params.metadata)
327                .build(),
328        );
329        (scope, event, subscribers, scope_stack.clone())
330    };
331    // Capture the scope-local chain before removing its owner. The event is
332    // published later, but scope cleanup must not change the middleware that
333    // was visible when the end event was emitted.
334    let sanitizers = snapshot_event_sanitizers(&event, &emission_scope_stack).unwrap_or_default();
335    let publication_scope_stack = snapshot_scope_stack(&emission_scope_stack)?;
336    let removed = task_scope_remove(params.handle_uuid)?;
337    debug_assert_eq!(removed.uuid, scope.uuid);
338    let _ = subscriber_dispatcher::dispatch_sanitized_event(
339        event,
340        sanitizers,
341        &subscribers,
342        publication_scope_stack,
343    );
344    Ok(())
345}
346
347/// Emit a standalone mark event under the current or provided scope.
348///
349/// This creates a point-in-time lifecycle event without pushing or popping a
350/// new scope.
351///
352/// # Parameters
353/// - `name`: Event name to emit.
354/// - `parent`: Optional explicit parent scope. When `None`, the current top of
355///   stack is used.
356/// - `data`: Optional JSON payload recorded on the emitted event.
357/// - `metadata`: Optional JSON metadata recorded on the emitted event.
358/// - `timestamp`: Optional timestamp recorded on the emitted mark event. When
359///   `None`, the current UTC time is used.
360///
361/// # Returns
362/// A [`Result`] that is `Ok(())` after the event has been queued for
363/// sanitization and publication.
364///
365/// # Errors
366/// Returns an error when the runtime owner check fails or when internal state
367/// cannot be read safely.
368///
369/// # Notes
370/// The mark event is queued with subscriber and sanitizer snapshots captured
371/// from the active scope stack.
372pub fn event(params: EmitMarkEventParams<'_>) -> Result<()> {
373    ensure_runtime_owner()?;
374    let parent_uuid = resolve_parent_uuid(params.parent);
375    let scope_stack = current_scope_stack();
376    let (event, subscribers, emission_scope_stack) = {
377        let subscribers = if params.name == COMPACTION_EVENT_NAME {
378            let mut scope_guard = scope_stack
379                .write()
380                .map_err(|error| scope_stack_lock_error(error, "mark"))?;
381            let subscribers =
382                snapshot_event_subscribers(scope_guard.collect_scope_local_subscribers())?;
383            scope_guard.mark_agent_fresh(parent_uuid);
384            subscribers
385        } else {
386            let scope_guard = scope_stack
387                .read()
388                .map_err(|error| scope_stack_lock_error(error, "mark"))?;
389            snapshot_event_subscribers(scope_guard.collect_scope_local_subscribers())?
390        };
391        let context = global_context();
392        let state = context
393            .read()
394            .map_err(|error| FlowError::Internal(error.to_string()))?;
395        let event = state.create_event(MarkEvent::new(
396            BaseEvent::builder()
397                .name(params.name)
398                .parent_uuid_opt(parent_uuid)
399                .timestamp(params.timestamp.unwrap_or_else(Utc::now))
400                .data_opt(params.data)
401                .data_schema_opt(params.data_schema)
402                .metadata_opt(params.metadata)
403                .build(),
404            params.category,
405            params.category_profile,
406        ));
407        (event, subscribers, scope_stack.clone())
408    };
409    let sanitizers = snapshot_event_sanitizers(&event, &emission_scope_stack).unwrap_or_default();
410    let _ = subscriber_dispatcher::dispatch_sanitized_event(
411        event,
412        sanitizers,
413        &subscribers,
414        emission_scope_stack,
415    );
416    Ok(())
417}
nemo_relay/api/scope.rs

nemo_relay/api/
scope.rs
