diff --git a/crates/cli/src/launcher.rs b/crates/cli/src/launcher.rs index e36010765..065ff0ffb 100644 --- a/crates/cli/src/launcher.rs +++ b/crates/cli/src/launcher.rs @@ -492,15 +492,13 @@ impl PreparedRun { })?; let _ = std::fs::remove_file(backup); } - (_, false) => { - if cursor.path.exists() { - std::fs::remove_file(&cursor.path).map_err(|error| { - CliError::Launch(format!( - "failed to remove temporary Cursor hooks {}: {error}", - cursor.path.display() - )) - })?; - } + (_, false) if cursor.path.exists() => { + std::fs::remove_file(&cursor.path).map_err(|error| { + CliError::Launch(format!( + "failed to remove temporary Cursor hooks {}: {error}", + cursor.path.display() + )) + })?; } _ => {} } diff --git a/crates/core/src/api/event.rs b/crates/core/src/api/event.rs index e7748a178..2d722e8cd 100644 --- a/crates/core/src/api/event.rs +++ b/crates/core/src/api/event.rs @@ -172,9 +172,8 @@ pub enum ScopeCategory { /// Category-specific profile data. /// -/// Unknown wire keys are preserved in `extra`. LLM annotations are runtime-only -/// enrichment used by internal adaptive and Agent Trajectory Interchange Format -/// (ATIF) logic and are never serialized. +/// Unknown wire keys are preserved in `extra`. LLM annotations are serialized +/// under `category_profile` when a codec captures them. #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, TypedBuilder)] #[builder(field_defaults(setter(into, strip_option(ignore_invalid, fallback_suffix = "_opt"))))] pub struct CategoryProfile { @@ -215,6 +214,8 @@ impl CategoryProfile { self.model_name.is_none() && self.tool_call_id.is_none() && self.subtype.is_none() + && self.annotated_request.is_none() + && self.annotated_response.is_none() && self.extra.is_empty() } } @@ -338,6 +339,24 @@ impl Event { } } + /// Try to return this event as the canonical JSON object delivered by + /// language bindings to subscriber callbacks and ATOF exporters. + pub fn try_to_json_value(&self) -> serde_json::Result { + serde_json::to_value(self) + } + + /// Return this event as the canonical JSON object delivered by language + /// bindings to subscriber callbacks. + pub fn to_json_value(&self) -> Json { + self.try_to_json_value() + .expect("serializing an ATOF event to JSON should not fail") + } + + /// Return this event as canonical JSON. + pub fn to_json_string(&self) -> serde_json::Result { + serde_json::to_string(&self.try_to_json_value()?) + } + /// Return the lifecycle phase for scope events. pub fn scope_category(&self) -> Option { match self { diff --git a/crates/core/src/observability/atof.rs b/crates/core/src/observability/atof.rs index bb7b35d72..a51b87245 100644 --- a/crates/core/src/observability/atof.rs +++ b/crates/core/src/observability/atof.rs @@ -4,9 +4,9 @@ //! Agent Trajectory Observability Format (ATOF) JSONL exporter support for NeMo //! Flow. //! -//! The [`AtofExporter`] registers as an event subscriber and writes each raw -//! NeMo Flow Agent Trajectory Observability Format (ATOF) event as one JSON -//! object per JSONL line. +//! The [`AtofExporter`] registers as an event subscriber and writes each +//! canonical NeMo Flow Agent Trajectory Observability Format (ATOF) event as +//! one JSON object per JSONL line. use std::fs::{File, OpenOptions}; use std::io::{BufWriter, Write}; @@ -252,7 +252,10 @@ fn open_file(path: &Path, mode: AtofExporterMode) -> Result { } fn write_event(writer: &mut BufWriter, event: &Event) -> std::result::Result<(), String> { - serde_json::to_writer(&mut *writer, event).map_err(|error| error.to_string())?; + let value = event + .try_to_json_value() + .map_err(|error| error.to_string())?; + serde_json::to_writer(&mut *writer, &value).map_err(|error| error.to_string())?; writer.write_all(b"\n").map_err(|error| error.to_string())?; writer.flush().map_err(|error| error.to_string()) } diff --git a/crates/core/tests/unit/observability/atof_tests.rs b/crates/core/tests/unit/observability/atof_tests.rs index 89f93382b..7e0b10af9 100644 --- a/crates/core/tests/unit/observability/atof_tests.rs +++ b/crates/core/tests/unit/observability/atof_tests.rs @@ -4,12 +4,16 @@ //! Unit tests for the ATOF JSONL exporter. use super::*; -use crate::api::event::{BaseEvent, Event, EventCategory, MarkEvent, ScopeCategory, ScopeEvent}; +use crate::api::event::{ + BaseEvent, CategoryProfile, Event, EventCategory, MarkEvent, ScopeCategory, ScopeEvent, +}; use crate::api::runtime::NemoFlowContextState; use crate::api::runtime::global_context; use crate::api::scope::{EmitMarkEventParams, PopScopeParams, PushScopeParams, ScopeType}; -use serde_json::json; +use crate::codec::request::{AnnotatedLlmRequest, Message, MessageContent}; +use serde_json::{Map, json}; use std::fs; +use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; use uuid::Uuid; @@ -55,6 +59,50 @@ fn make_scope_start_event(name: &str) -> Event { )) } +fn make_annotated_llm_event(name: &str) -> Event { + let request = AnnotatedLlmRequest { + messages: vec![Message::User { + content: MessageContent::Text("hello".into()), + name: None, + }], + model: Some("demo-model".into()), + params: None, + tools: None, + tool_choice: None, + store: None, + previous_response_id: None, + truncation: None, + reasoning: None, + include: None, + user: None, + metadata: None, + service_tier: None, + parallel_tool_calls: None, + max_output_tokens: None, + max_tool_calls: None, + top_logprobs: None, + stream: None, + extra: Map::new(), + }; + + Event::Scope(ScopeEvent::new( + BaseEvent::builder() + .uuid(Uuid::now_v7()) + .name(name) + .data(json!({"input": true})) + .build(), + ScopeCategory::Start, + Vec::new(), + EventCategory::llm(), + Some( + CategoryProfile::builder() + .model_name("demo-model") + .annotated_request(Arc::new(request)) + .build(), + ), + )) +} + fn read_jsonl(path: &Path) -> Vec { fs::read_to_string(path) .unwrap() @@ -144,6 +192,30 @@ fn subscriber_writes_scope_and_mark_events_as_raw_jsonl() { assert_eq!(lines[1]["data"], json!({"step": 1})); } +#[test] +fn subscriber_writes_canonical_event_jsonl() { + let dir = temp_dir("atof-canonical"); + let exporter = AtofExporter::new( + AtofExporterConfig::new() + .with_output_directory(&dir) + .with_filename("events.jsonl"), + ) + .unwrap(); + let event = make_annotated_llm_event("llm-start"); + + (exporter.subscriber())(&event); + exporter.force_flush().unwrap(); + + let lines = read_jsonl(exporter.path()); + assert_eq!(lines.len(), 1); + assert_eq!(lines[0], event.try_to_json_value().unwrap()); + assert!(lines[0].get("annotated_request").is_none()); + assert_eq!( + lines[0]["category_profile"]["annotated_request"]["model"], + "demo-model" + ); +} + #[test] fn register_deregister_flush_and_shutdown_work_with_runtime_events() { let _guard = crate::observability::test_mutex().lock().unwrap(); diff --git a/crates/core/tests/unit/types_tests.rs b/crates/core/tests/unit/types_tests.rs index 8a8e10517..b3d8e1472 100644 --- a/crates/core/tests/unit/types_tests.rs +++ b/crates/core/tests/unit/types_tests.rs @@ -3,6 +3,8 @@ //! Unit tests for types in the NeMo Flow core crate. +use std::sync::Arc; + use serde_json::{Map, json}; use uuid::{Uuid, Version}; @@ -13,6 +15,48 @@ use crate::api::event::{ use crate::api::llm::{LlmAttributes, LlmHandle, LlmRequest}; use crate::api::scope::{ScopeAttributes, ScopeHandle, ScopeType}; use crate::api::tool::{ToolAttributes, ToolHandle}; +use crate::codec::request::{AnnotatedLlmRequest, Message, MessageContent}; +use crate::codec::response::AnnotatedLlmResponse; + +fn annotated_request(model: &str, text: &str) -> AnnotatedLlmRequest { + AnnotatedLlmRequest { + messages: vec![Message::User { + content: MessageContent::Text(text.into()), + name: None, + }], + model: Some(model.into()), + params: None, + tools: None, + tool_choice: None, + store: None, + previous_response_id: None, + truncation: None, + reasoning: None, + include: None, + user: None, + metadata: None, + service_tier: None, + parallel_tool_calls: None, + max_output_tokens: None, + max_tool_calls: None, + top_logprobs: None, + stream: None, + extra: Map::new(), + } +} + +fn annotated_response(id: &str, model: &str, text: &str) -> AnnotatedLlmResponse { + AnnotatedLlmResponse { + id: Some(id.into()), + model: Some(model.into()), + message: Some(MessageContent::Text(text.into())), + tool_calls: None, + finish_reason: None, + usage: None, + api_specific: None, + extra: Map::new(), + } +} #[test] fn handle_constructors_preserve_supplied_metadata() { @@ -189,6 +233,70 @@ fn event_accessors_cover_scope_tool_llm_and_mark_variants() { assert_eq!(mark_event.tool_call_id(), None); } +#[test] +fn event_json_value_uses_canonical_subscriber_shape() { + let request = annotated_request("demo-model", "hi"); + let response = annotated_response("resp-1", "demo-model", "hello"); + let event = Event::Scope(ScopeEvent::new( + BaseEvent::builder() + .name("llm") + .data(json!({"input": true})) + .metadata(json!({"trace": "abc"})) + .build(), + ScopeCategory::End, + llm_attributes_to_strings(LlmAttributes::STATEFUL), + EventCategory::llm(), + Some( + CategoryProfile::builder() + .model_name("demo-model") + .annotated_request(Arc::new(request)) + .annotated_response(Arc::new(response)) + .build(), + ), + )); + + let value = event.try_to_json_value().unwrap(); + assert_eq!(event.to_json_value(), value); + assert_eq!(value["kind"], json!("scope")); + assert_eq!(value["scope_category"], json!("end")); + assert_eq!(value["category"], json!("llm")); + assert_eq!(value["data"], json!({"input": true})); + assert_eq!(value["metadata"], json!({"trace": "abc"})); + assert!(value.get("annotated_request").is_none()); + assert!(value.get("annotated_response").is_none()); + assert_eq!( + value["category_profile"]["annotated_request"]["model"], + json!("demo-model") + ); + assert_eq!( + value["category_profile"]["annotated_response"]["id"], + json!("resp-1") + ); + + let encoded = event.to_json_string().unwrap(); + let decoded: serde_json::Value = serde_json::from_str(&encoded).unwrap(); + assert_eq!(decoded, value); +} + +#[test] +fn category_profile_wire_empty_accounts_for_annotations() { + assert!(CategoryProfile::default().is_wire_empty()); + + let request_profile = CategoryProfile::builder() + .annotated_request(Arc::new(annotated_request("demo-model", "hi"))) + .build(); + assert!(!request_profile.is_wire_empty()); + + let response_profile = CategoryProfile::builder() + .annotated_response(Arc::new(annotated_response( + "resp-1", + "demo-model", + "hello", + ))) + .build(); + assert!(!response_profile.is_wire_empty()); +} + #[test] fn atof_event_builders_construct_concrete_events() { let parent_uuid = Some(Uuid::now_v7()); diff --git a/crates/ffi/nemo_flow.h b/crates/ffi/nemo_flow.h index 72149c27c..58185bfb0 100644 --- a/crates/ffi/nemo_flow.h +++ b/crates/ffi/nemo_flow.h @@ -2295,6 +2295,15 @@ char *nemo_flow_event_name(const struct FfiEvent *ptr); */ char *nemo_flow_event_kind(const struct FfiEvent *ptr); +/** + * Return the canonical subscriber event JSON as a C string. + * Caller must free the result with `nemo_flow_string_free`. + * + * # Safety + * `ptr` must be a valid `FfiEvent` pointer or null. + */ +char *nemo_flow_event_json(const struct FfiEvent *ptr); + /** * Return the ATOF version as a C string. * diff --git a/crates/ffi/src/types/mod.rs b/crates/ffi/src/types/mod.rs index 25e8b6ca7..e99d13686 100644 --- a/crates/ffi/src/types/mod.rs +++ b/crates/ffi/src/types/mod.rs @@ -28,6 +28,7 @@ use nemo_flow::api::tool::ToolHandle; use nemo_flow::codec::traits::{LlmCodec, LlmResponseCodec}; use crate::convert::{json_to_c_string, str_to_c_string}; +use crate::error::set_last_error; #[cfg(test)] use crate::{api, convert}; @@ -600,6 +601,25 @@ pub unsafe extern "C" fn nemo_flow_event_kind(ptr: *const FfiEvent) -> *mut c_ch str_to_c_string(unsafe { &*ptr }.0.kind()) } +/// Return the canonical subscriber event JSON as a C string. +/// Caller must free the result with `nemo_flow_string_free`. +/// +/// # Safety +/// `ptr` must be a valid `FfiEvent` pointer or null. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn nemo_flow_event_json(ptr: *const FfiEvent) -> *mut c_char { + if ptr.is_null() { + return std::ptr::null_mut(); + } + match unsafe { &*ptr }.0.try_to_json_value() { + Ok(value) => json_to_c_string(&value), + Err(error) => { + set_last_error(&error.to_string()); + std::ptr::null_mut() + } + } +} + /// Return the ATOF version as a C string. /// /// # Safety diff --git a/crates/ffi/tests/integration/api_tests.rs b/crates/ffi/tests/integration/api_tests.rs index 28350e3e6..0a5afbb4f 100644 --- a/crates/ffi/tests/integration/api_tests.rs +++ b/crates/ffi/tests/integration/api_tests.rs @@ -250,6 +250,8 @@ unsafe extern "C" fn subscriber_cb(_user_data: *mut libc::c_void, event: *const "uuid": unsafe { take_string(nemo_flow_event_uuid(event)) }.unwrap_or_default(), "name": unsafe { take_string(nemo_flow_event_name(event)) }.unwrap_or_default(), "kind": unsafe { take_string(nemo_flow_ffi::types::nemo_flow_event_kind(event)) }.unwrap_or_default(), + "json": unsafe { take_string(nemo_flow_ffi::types::nemo_flow_event_json(event)) } + .map(|s| serde_json::from_str::(&s).unwrap()), "data": unsafe { take_string(nemo_flow_event_data(event)) } .map(|s| serde_json::from_str::(&s).unwrap()), "metadata": unsafe { take_string(nemo_flow_event_metadata(event)) } diff --git a/crates/ffi/tests/unit/api/core_tests.rs b/crates/ffi/tests/unit/api/core_tests.rs index 22f4f9de0..c399dd93f 100644 --- a/crates/ffi/tests/unit/api/core_tests.rs +++ b/crates/ffi/tests/unit/api/core_tests.rs @@ -540,6 +540,13 @@ fn test_ffi_error_paths_and_scope_stack() { } } +#[test] +fn test_ffi_event_json_null_pointer_returns_null() { + unsafe { + assert!(types::nemo_flow_event_json(ptr::null::()).is_null()); + } +} + #[test] fn test_ffi_tool_lifecycle_execute_and_helpers() { let _lock = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); @@ -658,6 +665,11 @@ fn test_ffi_tool_lifecycle_execute_and_helpers() { let events = lock_unpoisoned(event_log()).clone(); assert!(events.iter().any(|event| event["name"] == "ffi_tool")); + assert!(events.iter().any(|event| { + event["json"]["kind"] == json!("scope") + && event["json"]["name"] == json!("ffi_tool") + && event["json"]["category"] == json!("tool") + })); assert!( events .iter() @@ -685,6 +697,8 @@ fn test_ffi_tool_lifecycle_execute_and_helpers() { assert!(events.iter().any(|event| { event["name"] == "ffi_mark" && event["kind"] == json!("mark") + && event["json"]["kind"] == json!("mark") + && event["json"]["name"] == json!("ffi_mark") && event["data"] == json!({"mark": true}) && event["metadata"] == json!({"origin": "ffi"}) })); diff --git a/crates/ffi/tests/unit/api_tests.rs b/crates/ffi/tests/unit/api_tests.rs index 316209a96..cbc498cfd 100644 --- a/crates/ffi/tests/unit/api_tests.rs +++ b/crates/ffi/tests/unit/api_tests.rs @@ -237,6 +237,8 @@ unsafe extern "C" fn subscriber_cb(_user_data: *mut libc::c_void, event: *const "uuid": unsafe { take_string(nemo_flow_event_uuid(event)) }.unwrap_or_default(), "name": unsafe { take_string(nemo_flow_event_name(event)) }.unwrap_or_default(), "kind": unsafe { take_string(crate::types::nemo_flow_event_kind(event)) }.unwrap_or_default(), + "json": unsafe { take_string(crate::types::nemo_flow_event_json(event)) } + .map(|s| serde_json::from_str::(&s).unwrap()), "data": unsafe { take_string(nemo_flow_event_data(event)) } .map(|s| serde_json::from_str::(&s).unwrap()), "metadata": unsafe { take_string(nemo_flow_event_metadata(event)) } diff --git a/crates/node/README.md b/crates/node/README.md index ccba1cd97..de1325a8b 100644 --- a/crates/node/README.md +++ b/crates/node/README.md @@ -73,6 +73,7 @@ const { async function main() { registerSubscriber("printer", (runtimeEvent) => { console.log(`${runtimeEvent.kind} ${runtimeEvent.name}`); + console.log(JSON.stringify(runtimeEvent)); }); await withScope("demo-agent", ScopeType.Agent, async (handle) => { diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index be5d9a22a..8c1aacddb 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -2284,7 +2284,7 @@ pub fn deregister_llm_stream_execution_intercept(name: String) -> Result { /// Register a named event subscriber that receives all lifecycle events. /// -/// The `callback` receives each event as a JSON-serialized `JsEvent` object. Events are +/// The `callback` receives each event as the canonical JSON event object. Events are /// delivered asynchronously and non-blocking. Throws if a subscriber with the same `name` /// already exists. #[napi] @@ -2777,7 +2777,7 @@ pub fn scope_deregister_llm_stream_execution_intercept( /// Register a scope-local named event subscriber that receives lifecycle events /// for the specified scope. /// -/// The `callback` receives each event as a JSON-serialized `JsEvent` object. Events are +/// The `callback` receives each event as the canonical JSON event object. Events are /// delivered asynchronously and non-blocking. Throws if a subscriber with the same `name` /// already exists on the specified scope. #[napi] diff --git a/crates/node/src/callable.rs b/crates/node/src/callable.rs index a4dd53098..625b0bb92 100644 --- a/crates/node/src/callable.rs +++ b/crates/node/src/callable.rs @@ -451,7 +451,15 @@ pub fn wrap_js_event_subscriber( ) -> EventSubscriberFn { let func = Arc::new(func); Arc::new(move |event: &Event| { - let event_json = serde_json::to_value(JsEvent::from(event)).unwrap_or(Json::Null); + let event_json = match JsEvent::try_from_event(event) { + Ok(event) => event.into_json(), + Err(error) => { + record_callback_error(format!( + "nemo_flow: failed to serialize JS event subscriber payload: {error}" + )); + return; + } + }; let status = func.call(event_json, ThreadsafeFunctionCallMode::NonBlocking); if status != napi::Status::Ok { record_callback_error(format!( diff --git a/crates/node/src/types/mod.rs b/crates/node/src/types/mod.rs index d3fabf36f..3327ddc5d 100644 --- a/crates/node/src/types/mod.rs +++ b/crates/node/src/types/mod.rs @@ -304,24 +304,19 @@ impl LlmRequest { #[serde(transparent)] pub struct JsEvent(serde_json::Value); +impl JsEvent { + pub(crate) fn try_from_event(e: &Event) -> serde_json::Result { + Ok(Self(e.try_to_json_value()?)) + } + + pub(crate) fn into_json(self) -> serde_json::Value { + self.0 + } +} + impl From<&Event> for JsEvent { fn from(e: &Event) -> Self { - let mut value = serde_json::to_value(e).unwrap_or(Json::Null); - if let Json::Object(ref mut object) = value { - if let Some(request) = e.annotated_request() { - object.insert( - "annotated_request".to_string(), - serde_json::to_value(request.as_ref()).unwrap_or(Json::Null), - ); - } - if let Some(response) = e.annotated_response() { - object.insert( - "annotated_response".to_string(), - serde_json::to_value(response.as_ref()).unwrap_or(Json::Null), - ); - } - } - Self(value) + Self::try_from_event(e).expect("serializing an ATOF event to JSON should not fail") } } // --------------------------------------------------------------------------- diff --git a/crates/node/tests/scope_tests.mjs b/crates/node/tests/scope_tests.mjs index f4447b921..d7baecd72 100644 --- a/crates/node/tests/scope_tests.mjs +++ b/crates/node/tests/scope_tests.mjs @@ -259,6 +259,7 @@ describe('Subscribers', () => { assert.ok(typeof captured.uuid === 'string'); assert.ok(typeof captured.timestamp === 'string'); assert.ok(typeof captured.kind === 'string'); + assert.equal(JSON.parse(JSON.stringify(captured)).kind, captured.kind); } finally { deregisterSubscriber('node_prop_collector'); } diff --git a/crates/node/tests/typed_tests.mjs b/crates/node/tests/typed_tests.mjs index 5be99dca0..5a5c63498 100644 --- a/crates/node/tests/typed_tests.mjs +++ b/crates/node/tests/typed_tests.mjs @@ -552,9 +552,18 @@ describe('typedLlmExecute', () => { event.scope_category === 'end' && event.name === 'typed_anthropic_codec_llm', ); - assert.equal(endEvent.annotated_response.model, 'claude-3-5-sonnet'); - assert.equal(endEvent.annotated_response.message, 'Anthropic hello'); - assert.equal(endEvent.annotated_response.finish_reason, 'complete'); + assert.equal( + endEvent.category_profile.annotated_response.model, + 'claude-3-5-sonnet', + ); + assert.equal( + endEvent.category_profile.annotated_response.message, + 'Anthropic hello', + ); + assert.equal( + endEvent.category_profile.annotated_response.finish_reason, + 'complete', + ); } finally { deregisterSubscriber('typed_anthropic_codec_sub'); popScope(scope); @@ -786,9 +795,18 @@ describe('typedLlmStreamExecute', () => { event.scope_category === 'end' && event.name === 'typed_responses_stream_llm', ); - assert.equal(endEvent.annotated_response.model, 'gpt-4.1-mini'); - assert.equal(endEvent.annotated_response.message, 'hello world'); - assert.equal(endEvent.annotated_response.finish_reason, 'complete'); + assert.equal( + endEvent.category_profile.annotated_response.model, + 'gpt-4.1-mini', + ); + assert.equal( + endEvent.category_profile.annotated_response.message, + 'hello world', + ); + assert.equal( + endEvent.category_profile.annotated_response.finish_reason, + 'complete', + ); } finally { deregisterSubscriber('typed_responses_stream_sub'); deregisterLlmRequestIntercept('typed_responses_stream_req'); @@ -850,7 +868,7 @@ describe('typedLlmStreamExecute', () => { event.scope_category === 'end' && event.name === 'typed_stream_bad_response_codec_llm', ); - assert.equal(endEvent.annotated_response, undefined); + assert.equal(endEvent.category_profile?.annotated_response, undefined); } finally { deregisterSubscriber('typed_stream_bad_response_codec_sub'); popScope(scope); diff --git a/crates/python/src/py_types/events.rs b/crates/python/src/py_types/events.rs index 3cd5d9c69..95537577a 100644 --- a/crates/python/src/py_types/events.rs +++ b/crates/python/src/py_types/events.rs @@ -109,6 +109,23 @@ impl PyScopeEvent { inner: (**response).clone(), }) } + + /// Return this event as the canonical subscriber JSON dictionary. + pub(crate) fn to_dict(&self, py: Python<'_>) -> PyResult> { + let event = nemo_flow::api::event::Event::Scope(self.inner.clone()); + let value = event + .try_to_json_value() + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?; + json_to_py(py, &value) + } + + /// Return this event as canonical subscriber JSON. + pub(crate) fn to_json(&self) -> PyResult { + let event = nemo_flow::api::event::Event::Scope(self.inner.clone()); + event + .to_json_string() + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) + } } #[pyclass(name = "MarkEvent", skip_from_py_object)] @@ -180,4 +197,21 @@ impl PyMarkEvent { pub(crate) fn metadata(&self, py: Python<'_>) -> PyResult> { opt_json_to_py(py, &self.inner.base.metadata) } + + /// Return this event as the canonical subscriber JSON dictionary. + pub(crate) fn to_dict(&self, py: Python<'_>) -> PyResult> { + let event = nemo_flow::api::event::Event::Mark(self.inner.clone()); + let value = event + .try_to_json_value() + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?; + json_to_py(py, &value) + } + + /// Return this event as canonical subscriber JSON. + pub(crate) fn to_json(&self) -> PyResult { + let event = nemo_flow::api::event::Event::Mark(self.inner.clone()); + event + .to_json_string() + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) + } } diff --git a/crates/wasm/README.md b/crates/wasm/README.md index d6b53e28f..6436991e6 100644 --- a/crates/wasm/README.md +++ b/crates/wasm/README.md @@ -90,6 +90,7 @@ const { async function main() { registerSubscriber("printer", (runtimeEvent) => { console.log(`${runtimeEvent.kind} ${runtimeEvent.name}`); + console.log(JSON.stringify(runtimeEvent)); }); await withScope("demo-agent", ScopeType.Agent, async (handle) => { diff --git a/crates/wasm/src/callable.rs b/crates/wasm/src/callable.rs index 9a908c809..b8330438b 100644 --- a/crates/wasm/src/callable.rs +++ b/crates/wasm/src/callable.rs @@ -502,7 +502,15 @@ pub fn wrap_js_event_subscriber(_func: Function) -> EventSubscriberFn { pub fn wrap_js_event_subscriber(func: Function) -> EventSubscriberFn { let func = SendWrapper::new(func); std::sync::Arc::new(move |event: &Event| { - let wasm_event = WasmEvent::from(event); + let wasm_event = match WasmEvent::try_from_event(event) { + Ok(event) => event, + Err(error) => { + record_callback_error(format!( + "nemo_flow: failed to serialize JS event subscriber payload: {error}" + )); + return; + } + }; let js_event = wasm_event .serialize(&serde_wasm_bindgen::Serializer::json_compatible()) .unwrap_or(JsValue::NULL); diff --git a/crates/wasm/src/types/mod.rs b/crates/wasm/src/types/mod.rs index cb0c00019..b196ae03b 100644 --- a/crates/wasm/src/types/mod.rs +++ b/crates/wasm/src/types/mod.rs @@ -419,26 +419,15 @@ impl LlmRequest { #[serde(transparent)] pub struct WasmEvent(Json); +impl WasmEvent { + pub(crate) fn try_from_event(e: &Event) -> serde_json::Result { + Ok(Self(e.try_to_json_value()?)) + } +} + impl From<&Event> for WasmEvent { fn from(e: &Event) -> Self { - let mut value = serde_json::to_value(e).unwrap_or(Json::Null); - - if let Json::Object(ref mut object) = value { - if let Some(request) = e.annotated_request() { - object.insert( - "annotated_request".to_string(), - serde_json::to_value(request.as_ref()).unwrap_or(Json::Null), - ); - } - if let Some(response) = e.annotated_response() { - object.insert( - "annotated_response".to_string(), - serde_json::to_value(response.as_ref()).unwrap_or(Json::Null), - ); - } - } - - Self(value) + Self::try_from_event(e).expect("serializing an ATOF event to JSON should not fail") } } diff --git a/crates/wasm/tests/integration/scope_tests.rs b/crates/wasm/tests/integration/scope_tests.rs index 1937f53c1..a35883f82 100644 --- a/crates/wasm/tests/integration/scope_tests.rs +++ b/crates/wasm/tests/integration/scope_tests.rs @@ -503,6 +503,15 @@ fn test_subscriber_event_properties() { let kind = js_sys::Reflect::get(&event, &"kind".into()).unwrap(); assert!(kind.is_string(), "Event should have kind string"); + let encoded = js_sys::JSON::stringify(&event).unwrap(); + let decoded = js_sys::JSON::parse(&encoded.as_string().unwrap()).unwrap(); + let decoded_kind = js_sys::Reflect::get(&decoded, &"kind".into()).unwrap(); + assert_eq!( + decoded_kind.as_string(), + kind.as_string(), + "Event should be directly JSON serializable" + ); + deregister_subscriber("wasm_prop_collector").unwrap(); js_sys::eval("delete globalThis.__wasm_evt_props").unwrap(); } diff --git a/docs/about/concepts/events.md b/docs/about/concepts/events.md index c50cf5c52..fc6ee2959 100644 --- a/docs/about/concepts/events.md +++ b/docs/about/concepts/events.md @@ -77,14 +77,17 @@ ATOF uses one `data` field. For scope events, `data` is the semantic input on Category-specific fields live under `category_profile`. NeMo Flow uses `model_name` for LLM events, `tool_call_id` for tool events, and `subtype` for -custom-category events. Unknown profile fields are preserved so newer -producers can interoperate with older consumers. +custom-category events. LLM codec annotations, when present, are serialized +under `category_profile.annotated_request` on LLM start events and +`category_profile.annotated_response` on LLM end events. Unknown profile fields +are preserved so newer producers can interoperate with older consumers. ### Annotated Request and Response Data -LLM codecs can enrich LLM events with runtime-only annotated request and -response data. These annotations are available to in-process subscribers and -exporters, but they are not serialized into the ATOF wire event. +LLM codecs can enrich LLM events with annotated request and response data. These +annotations are part of the canonical event JSON under `category_profile` when +they are present, so ATOF JSONL export and in-process subscriber JSON expose the +same payload shape. ## How Events Are Produced diff --git a/docs/about/concepts/subscribers.md b/docs/about/concepts/subscribers.md index 71780d81e..69a6b8107 100644 --- a/docs/about/concepts/subscribers.md +++ b/docs/about/concepts/subscribers.md @@ -61,6 +61,52 @@ exporter handoff. Some subscribers stay inside the process and power custom logging, analytics, or debugging logic. +#### Host Integration Event JSON + +For host integrations that need a serialized event payload, use the event +object's canonical JSON helpers instead of reconstructing payloads from native +attributes. Python subscribers can call `event.to_dict()` or `event.to_json()` +from the callback while still using the normal subscriber registration API. + +This pattern is useful when an agent runtime, framework adapter, or plugin host +already has its own lifecycle hooks but wants NeMo Flow to be the shared +telemetry representation. The host integration maps those hooks into NeMo Flow +scopes, LLM calls, tool calls, or marks. NeMo Flow emits the canonical ATOF event +stream, and each subscriber chooses whether to consume the native event object, +the canonical JSON helper, or an exporter-specific translation. + +```{mermaid} +flowchart + Host[Host Integration] + + subgraph NeMoFlow[NeMo Flow] + direction TB + Binding[Binding API] + Core[Rust Core Runtime] + Events[Canonical ATOF Event Stream] + Observer[In-Process Subscriber] + Json[Canonical Event JSON] + Exporters[Exporter Subscribers] + Backends[JSONL / ATIF / OTLP] + + Binding -->|emits scopes, tools, LLMs, marks| Core + Core --> Events + Events --> Observer + Observer -->|to_dict / to_json / JSON| Json + Events --> Exporters + Exporters --> Backends + end + + Host -->|maps lifecycle hooks| Binding + Json -. host consumes canonical telemetry .-> Host +``` + +The important boundary is that subscribers do not define the event schema. They +receive the runtime event and may serialize it through the binding helper when +they need a stable JSON payload. Exporter subscribers, such as the ATOF JSONL +exporter, consume the same event stream and serialize the same canonical event +shape for their target backend. + ### Forwarding and Export Some subscribers translate the event stream into external formats or transport @@ -115,6 +161,8 @@ plugin component. Use these practices when applying the concept in application or integration code. - Use a plain subscriber when you want in-process custom behavior. +- Use `event.to_dict()` or `event.to_json()` when a host runtime or exporter + needs the canonical event JSON shape in-process. - Use a scope-local subscriber when the observation should disappear with the owning scope. - Use a plugin-installed subscriber when the behavior should be reusable and diff --git a/docs/integrate-frameworks/provider-response-codecs.md b/docs/integrate-frameworks/provider-response-codecs.md index 81d865ae8..8322e1cf6 100644 --- a/docs/integrate-frameworks/provider-response-codecs.md +++ b/docs/integrate-frameworks/provider-response-codecs.md @@ -211,7 +211,7 @@ nemo_flow.subscribers.register("response-debugger", on_event) import { registerSubscriber } from 'nemo-flow-node'; registerSubscriber('response-debugger', (event) => { - const annotated = event.annotated_response; + const annotated = event.category_profile?.annotated_response; if (!annotated) { return; } diff --git a/go/nemo_flow/README.md b/go/nemo_flow/README.md index 3805247e1..7bce110bf 100644 --- a/go/nemo_flow/README.md +++ b/go/nemo_flow/README.md @@ -95,6 +95,7 @@ import ( func main() { if err := nemo.RegisterSubscriber("printer", func(event nemo.Event) { fmt.Printf("%s %s\n", event.Kind(), event.Name()) + fmt.Println(string(event.JSON())) }); err != nil { log.Fatal(err) } diff --git a/go/nemo_flow/scope_test.go b/go/nemo_flow/scope_test.go index ae1e4c248..d3fa046cc 100644 --- a/go/nemo_flow/scope_test.go +++ b/go/nemo_flow/scope_test.go @@ -8,6 +8,7 @@ import ( "fmt" "sync" "testing" + "time" ) const pushScopeFailed = "PushScope failed: %v" @@ -120,6 +121,82 @@ func assertJSONFieldNumber(t *testing.T, raw json.RawMessage, field string, want } } +func TestEventJSONHelpers(t *testing.T) { + var captured Event + capturedCh := make(chan struct{}, 1) + var mu sync.Mutex + subscriberName := "go_event_json_sub" + + _ = DeregisterSubscriber(subscriberName) + if err := RegisterSubscriber(subscriberName, func(event Event) { + if event.Name() == "go_json_mark" { + mu.Lock() + captured = event + mu.Unlock() + select { + case capturedCh <- struct{}{}: + default: + } + } + }); err != nil { + t.Fatalf("RegisterSubscriber failed: %v", err) + } + defer DeregisterSubscriber(subscriberName) + + if err := EmitEvent("go_json_mark", WithEventData(json.RawMessage(`{"ok":true}`))); err != nil { + t.Fatalf("EmitEvent failed: %v", err) + } + + select { + case <-capturedCh: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for subscriber to capture event") + } + + mu.Lock() + event := captured + mu.Unlock() + if event == nil { + t.Fatal("expected subscriber to capture event") + } + + var payload map[string]interface{} + if err := json.Unmarshal(event.JSON(), &payload); err != nil { + t.Fatalf("event.JSON returned invalid JSON: %v", err) + } + if payload["kind"] != "mark" || payload["name"] != "go_json_mark" { + t.Fatalf("unexpected event JSON payload: %#v", payload) + } + + marshaled, err := json.Marshal(event) + if err != nil { + t.Fatalf("json.Marshal(event) failed: %v", err) + } + var marshaledPayload map[string]interface{} + if err := json.Unmarshal(marshaled, &marshaledPayload); err != nil { + t.Fatalf("json.Marshal(event) returned invalid JSON: %v", err) + } + if marshaledPayload["kind"] != payload["kind"] || marshaledPayload["name"] != payload["name"] { + t.Fatalf("MarshalJSON payload mismatch: raw=%#v marshaled=%#v", payload, marshaledPayload) + } +} + +func TestEventBaseJSONHandlesNilPointer(t *testing.T) { + var base eventBase + + if raw := base.JSON(); raw != nil { + t.Fatalf("expected nil JSON for nil event pointer, got %s", raw) + } + + marshaled, err := json.Marshal(base) + if err != nil { + t.Fatalf("json.Marshal(eventBase) failed: %v", err) + } + if string(marshaled) != "null" { + t.Fatalf("expected nil event base to marshal as null, got %s", marshaled) + } +} + func runConcurrentScopePushPopWorker(errCh chan<- error) { stack, err := NewScopeStack() if err != nil { diff --git a/go/nemo_flow/types.go b/go/nemo_flow/types.go index 9a5c0a6d8..fd7cd6a42 100644 --- a/go/nemo_flow/types.go +++ b/go/nemo_flow/types.go @@ -56,6 +56,7 @@ extern void nemo_flow_llm_request_free(FfiLLMRequest* ptr); extern char* nemo_flow_event_uuid(const FfiEvent* ptr); extern char* nemo_flow_event_name(const FfiEvent* ptr); extern char* nemo_flow_event_kind(const FfiEvent* ptr); +extern char* nemo_flow_event_json(const FfiEvent* ptr); extern char* nemo_flow_event_atof_version(const FfiEvent* ptr); extern char* nemo_flow_event_scope_category(const FfiEvent* ptr); extern char* nemo_flow_event_category(const FfiEvent* ptr); @@ -380,6 +381,8 @@ type Event interface { ToolCallID() string AnnotatedRequest() json.RawMessage AnnotatedResponse() json.RawMessage + JSON() json.RawMessage + MarshalJSON() ([]byte, error) } type eventBase struct { @@ -409,6 +412,7 @@ type eventSnapshot struct { toolCallID string annotatedRequest json.RawMessage annotatedResponse json.RawMessage + eventJSON json.RawMessage } func (e eventBase) UUID() string { @@ -540,6 +544,19 @@ func (e eventBase) AnnotatedResponse() json.RawMessage { } return goJSONOpt(C.nemo_flow_event_annotated_response(e.ptr)) } +func (e eventBase) JSON() json.RawMessage { + if e.snapshot != nil { + return cloneJSON(e.snapshot.eventJSON) + } + return goJSONOpt(C.nemo_flow_event_json(e.ptr)) +} +func (e eventBase) MarshalJSON() ([]byte, error) { + raw := e.JSON() + if raw == nil { + return []byte("null"), nil + } + return cloneJSON(raw), nil +} // ScopeEvent is the typed wrapper for an ATOF scope lifecycle event. type ScopeEvent struct{ eventBase } @@ -571,6 +588,7 @@ func newEvent(ptr *C.FfiEvent) Event { toolCallID: goStringOpt((*C.char)(C.nemo_flow_event_tool_call_id(unsafe.Pointer(ptr)))), annotatedRequest: goJSONOpt(C.nemo_flow_event_annotated_request(ptr)), annotatedResponse: goJSONOpt(C.nemo_flow_event_annotated_response(ptr)), + eventJSON: goJSONOpt(C.nemo_flow_event_json(ptr)), }, } switch base.Kind() { diff --git a/pyproject.toml b/pyproject.toml index 95b140ba1..c55c0ae94 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -120,8 +120,9 @@ omit = ["python/nemo_flow/integrations/*"] [tool.ty.analysis] # nemo_flow._native is a compiled Rust extension (built by maturin) that only # exists after `uv sync` / `pip install -e .`. Suppress unresolved-import for it. -# LangChain and LangGraph are optional integration dependencies which aren't installed by default. -allowed-unresolved-imports = ["langchain.**", "langchain_*.**", "langgraph.**", "nemo_flow._native", "pytest"] +# LangChain, LangGraph, and Deep Agents are optional integration dependencies +# which aren't installed by default. +allowed-unresolved-imports = ["deepagents.**", "langchain.**", "langchain_*.**", "langgraph.**", "nemo_flow._native", "pytest"] [tool.ruff] line-length = 120 diff --git a/python/nemo_flow/README.md b/python/nemo_flow/README.md index 0ddd7e618..d76299165 100644 --- a/python/nemo_flow/README.md +++ b/python/nemo_flow/README.md @@ -145,6 +145,28 @@ with nemo_flow.scope.scope("demo-agent", nemo_flow.ScopeType.Agent) as handle: nemo_flow.subscribers.deregister("printer") ``` +For host integrations that need a serialized event shape, consume the +canonical JSON payload from the subscriber event object: + +```python +import json +import nemo_flow + + +def on_event(event) -> None: + payload = event.to_dict() + print(payload["kind"], payload["name"]) + assert json.loads(event.to_json()) == payload + + +nemo_flow.subscribers.register("host-exporter", on_event) +try: + with nemo_flow.scope.scope("demo-agent", nemo_flow.ScopeType.Agent): + nemo_flow.scope.event("initialized", data={"binding": "python"}) +finally: + nemo_flow.subscribers.deregister("host-exporter") +``` + ## Package Surface The public package modules are: diff --git a/python/nemo_flow/_native.pyi b/python/nemo_flow/_native.pyi index abb9e06eb..5b8917efe 100644 --- a/python/nemo_flow/_native.pyi +++ b/python/nemo_flow/_native.pyi @@ -603,6 +603,12 @@ class ScopeEvent: def annotated_response(self) -> Optional[AnnotatedLLMResponse]: """Return the normalized LLM response annotation, if present.""" ... + def to_dict(self) -> _JsonObject: + """Return this event as the canonical subscriber JSON dictionary.""" + ... + def to_json(self) -> str: + """Return this event as canonical subscriber JSON.""" + ... class MarkEvent: """ATOF point-in-time mark event emitted to subscribers. @@ -659,6 +665,12 @@ class MarkEvent: def data_schema(self) -> Optional[_JsonObject]: """Return a schema descriptor for ``data``, if one is present.""" ... + def to_dict(self) -> _JsonObject: + """Return this event as the canonical subscriber JSON dictionary.""" + ... + def to_json(self) -> str: + """Return this event as canonical subscriber JSON.""" + ... class AtifExporter: """ATIF trajectory exporter that collects events and exports trajectories. diff --git a/python/nemo_flow/py.typed b/python/nemo_flow/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/python/tests/test_event_json.py b/python/tests/test_event_json.py new file mode 100644 index 000000000..9136226b6 --- /dev/null +++ b/python/tests/test_event_json.py @@ -0,0 +1,53 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for canonical subscriber event JSON helpers.""" + +from __future__ import annotations + +import json +from uuid import uuid4 + +from nemo_flow import MarkEvent, ScopeEvent, ScopeType, scope, subscribers + + +def _subscriber_name(prefix: str) -> str: + return f"{prefix}-{uuid4()}" + + +def test_subscriber_events_expose_canonical_json_helpers(): + events = [] + name = _subscriber_name("py-event-json") + subscribers.register(name, events.append) + try: + with scope.scope( + "json-scope", + ScopeType.Agent, + input={"input": True}, + metadata={"trace": "abc"}, + ): + scope.event("json-mark", data={"mark": True}, metadata={"source": "test"}) + finally: + subscribers.deregister(name) + + scope_event = next( + event + for event in events + if isinstance(event, ScopeEvent) and event.name == "json-scope" and event.scope_category == "start" + ) + scope_payload = scope_event.to_dict() + assert scope_payload["kind"] == "scope" + assert scope_payload["scope_category"] == "start" + assert scope_payload["category"] == "agent" + assert scope_payload["name"] == "json-scope" + assert scope_payload["data"] == {"input": True} + assert scope_payload["metadata"] == {"trace": "abc"} + assert json.loads(scope_event.to_json()) == scope_payload + + mark_event = next(event for event in events if isinstance(event, MarkEvent) and event.name == "json-mark") + mark_payload = mark_event.to_dict() + assert mark_payload["kind"] == "mark" + assert mark_payload["name"] == "json-mark" + assert mark_payload["data"] == {"mark": True} + assert mark_payload["metadata"] == {"source": "test"} + assert json.loads(mark_event.to_json()) == mark_payload