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::{self, SubscriberDelivery};
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    pop_scope_inner(params, false).map(|_| ())
300}
301
302/// Pop the current scope and return a receipt for its scope-end subscriber delivery.
303///
304/// The receipt covers sanitizer and subscriber processing for the scope-end event.
305/// It does not wait for unrelated events queued after that event.
306#[doc(hidden)]
307pub fn pop_scope_with_subscriber_delivery(
308    params: PopScopeParams<'_>,
309) -> Result<SubscriberDelivery> {
310    pop_scope_inner(params, true)?.ok_or_else(|| {
311        FlowError::Internal("tracked scope pop did not create a subscriber delivery receipt".into())
312    })
313}
314
315fn pop_scope_inner(
316    params: PopScopeParams<'_>,
317    track_delivery: bool,
318) -> Result<Option<SubscriberDelivery>> {
319    ensure_runtime_owner()?;
320    let scope_stack = current_scope_stack();
321    let (scope, event, subscribers, emission_scope_stack) = {
322        let scope_guard = scope_stack
323            .read()
324            .map_err(|error| scope_stack_lock_error(error, "pop"))?;
325        let top = scope_guard.top();
326        if top.uuid != *params.handle_uuid {
327            if scope_guard.find(params.handle_uuid).is_some() {
328                return Err(FlowError::InvalidArgument(
329                    "scope handle is not at the top of the stack".into(),
330                ));
331            }
332            return Err(FlowError::NotFound("scope handle not found".into()));
333        }
334        let scope_subscribers = scope_guard.collect_scope_local_subscribers();
335        let subscribers = snapshot_event_subscribers(scope_subscribers)?;
336        let scope = top.clone();
337        let context = global_context();
338        let state = context
339            .read()
340            .map_err(|error| FlowError::Internal(error.to_string()))?;
341        let event = state.build_scope_end_event(
342            EndScopeHandleParams::builder()
343                .handle(&scope)
344                .data_opt(params.output)
345                .timestamp_opt(params.timestamp)
346                .metadata_opt(params.metadata)
347                .build(),
348        );
349        (scope, event, subscribers, scope_stack.clone())
350    };
351    // Capture the scope-local chain before removing its owner. The event is
352    // published later, but scope cleanup must not change the middleware that
353    // was visible when the end event was emitted.
354    let sanitizers = snapshot_event_sanitizers(&event, &emission_scope_stack).unwrap_or_default();
355    let publication_scope_stack = snapshot_scope_stack(&emission_scope_stack)?;
356    let removed = task_scope_remove(params.handle_uuid)?;
357    debug_assert_eq!(removed.uuid, scope.uuid);
358    if track_delivery {
359        subscriber_dispatcher::dispatch_sanitized_event_with_delivery(
360            event,
361            sanitizers,
362            &subscribers,
363            publication_scope_stack,
364        )
365        .map(Some)
366    } else {
367        let _ = subscriber_dispatcher::dispatch_sanitized_event(
368            event,
369            sanitizers,
370            &subscribers,
371            publication_scope_stack,
372        );
373        Ok(None)
374    }
375}
376
377/// Emit a standalone mark event under the current or provided scope.
378///
379/// This creates a point-in-time lifecycle event without pushing or popping a
380/// new scope.
381///
382/// # Parameters
383/// - `name`: Event name to emit.
384/// - `parent`: Optional explicit parent scope. When `None`, the current top of
385///   stack is used.
386/// - `data`: Optional JSON payload recorded on the emitted event.
387/// - `metadata`: Optional JSON metadata recorded on the emitted event.
388/// - `timestamp`: Optional timestamp recorded on the emitted mark event. When
389///   `None`, the current UTC time is used.
390///
391/// # Returns
392/// A [`Result`] that is `Ok(())` after the event has been queued for
393/// sanitization and publication.
394///
395/// # Errors
396/// Returns an error when the runtime owner check fails or when internal state
397/// cannot be read safely.
398///
399/// # Notes
400/// The mark event is queued with subscriber and sanitizer snapshots captured
401/// from the active scope stack.
402pub fn event(params: EmitMarkEventParams<'_>) -> Result<()> {
403    ensure_runtime_owner()?;
404    let parent_uuid = resolve_parent_uuid(params.parent);
405    let scope_stack = current_scope_stack();
406    let (event, subscribers, emission_scope_stack) = {
407        let subscribers = if params.name == COMPACTION_EVENT_NAME {
408            let mut scope_guard = scope_stack
409                .write()
410                .map_err(|error| scope_stack_lock_error(error, "mark"))?;
411            let subscribers =
412                snapshot_event_subscribers(scope_guard.collect_scope_local_subscribers())?;
413            scope_guard.mark_agent_fresh(parent_uuid);
414            subscribers
415        } else {
416            let scope_guard = scope_stack
417                .read()
418                .map_err(|error| scope_stack_lock_error(error, "mark"))?;
419            snapshot_event_subscribers(scope_guard.collect_scope_local_subscribers())?
420        };
421        let context = global_context();
422        let state = context
423            .read()
424            .map_err(|error| FlowError::Internal(error.to_string()))?;
425        let event = state.create_event(MarkEvent::new(
426            BaseEvent::builder()
427                .name(params.name)
428                .parent_uuid_opt(parent_uuid)
429                .timestamp(params.timestamp.unwrap_or_else(Utc::now))
430                .data_opt(params.data)
431                .data_schema_opt(params.data_schema)
432                .metadata_opt(params.metadata)
433                .build(),
434            params.category,
435            params.category_profile,
436        ));
437        (event, subscribers, scope_stack.clone())
438    };
439    let sanitizers = snapshot_event_sanitizers(&event, &emission_scope_stack).unwrap_or_default();
440    let _ = subscriber_dispatcher::dispatch_sanitized_event(
441        event,
442        sanitizers,
443        &subscribers,
444        emission_scope_stack,
445    );
446    Ok(())
447}
nemo_relay/api/scope.rs

nemo_relay/api/
scope.rs
