From afe670f612e9de4931648feef596b859e6ebe017 Mon Sep 17 00:00:00 2001 From: Enrico Piovesan Date: Sat, 8 Aug 2026 00:14:54 -0600 Subject: [PATCH] feat(runtime): thread real emitted events through LocalExecutor, retire output-JSON convention LocalExecutor::execute now returns LocalExecutionOutput { value, emitted_events } instead of a bare Value, so both BoundLocalExecutor (PlacementRouter path) and ArtifactRouter (workflow-internal node execution) carry real WASM-sourced events instead of discarding them. Native-emitted events are validated against the capability's `emits` list and Subscribable service_type before use. workflows.rs now publishes to EventBroker and reads emitted events from the structured field, removing the old JSON "emitted_events" key convention. Migrates all ~21 LocalExecutor implementors across traverse-runtime, traverse-mcp, and traverse-cli to the new return type. Governing spec: 101-local-executor-event-emission (ADR-0037) Closes #996 Co-Authored-By: Claude Sonnet 5 --- crates/traverse-cli/src/http_api.rs | 20 +- crates/traverse-cli/src/main.rs | 16 +- crates/traverse-mcp/src/lib.rs | 11 +- crates/traverse-mcp/src/stdio_server.rs | 11 +- .../examples/load_workspace_app_state.rs | 9 +- .../traverse-runtime/src/artifact_router.rs | 105 +++++- crates/traverse-runtime/src/lib.rs | 336 ++++++++++++++--- crates/traverse-runtime/src/workflows.rs | 341 ++++++++++++------ .../tests/doc_approval_pipeline.rs | 38 +- .../tests/placement_router_live_wiring.rs | 62 +++- crates/traverse-runtime/tests/runtime.rs | 22 +- 11 files changed, 725 insertions(+), 246 deletions(-) diff --git a/crates/traverse-cli/src/http_api.rs b/crates/traverse-cli/src/http_api.rs index f57c2a30..c4e14141 100644 --- a/crates/traverse-cli/src/http_api.rs +++ b/crates/traverse-cli/src/http_api.rs @@ -7289,7 +7289,9 @@ mod tests { RegistryProvenance, RegistryScope, SourceKind, SourceReference, WorkflowEdge, WorkflowNode, WorkflowNodeInput, WorkflowNodeOutput, }; - use traverse_runtime::{LocalExecutionFailure, LocalExecutionFailureCode}; + use traverse_runtime::{ + LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutionOutput, + }; #[test] fn serve_error_display_keeps_bind_and_accept_context() { @@ -7323,11 +7325,17 @@ mod tests { &self, _capability: &ResolvedCapability, _input: &Value, - ) -> Result { - self.result.clone().map_err(|msg| LocalExecutionFailure { - code: LocalExecutionFailureCode::ExecutionFailed, - message: msg, - }) + ) -> Result { + self.result + .clone() + .map(|value| LocalExecutionOutput { + value, + emitted_events: Vec::new(), + }) + .map_err(|msg| LocalExecutionFailure { + code: LocalExecutionFailureCode::ExecutionFailed, + message: msg, + }) } } diff --git a/crates/traverse-cli/src/main.rs b/crates/traverse-cli/src/main.rs index 129ee835..4fcc6a29 100644 --- a/crates/traverse-cli/src/main.rs +++ b/crates/traverse-cli/src/main.rs @@ -42,9 +42,9 @@ use traverse_registry::{ }; use traverse_runtime::executor::{SUPPORTED_HOST_ABI_VERSION, verify_wasm_host_abi_bytes}; use traverse_runtime::{ - ArtifactRouter, LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutor, Runtime, - RuntimeExecutionOutcome, RuntimeRequest, RuntimeResultStatus, RuntimeTrace, - parse_runtime_request, + ArtifactRouter, LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutionOutput, + LocalExecutor, Runtime, RuntimeExecutionOutcome, RuntimeRequest, RuntimeResultStatus, + RuntimeTrace, parse_runtime_request, }; #[derive(Debug)] @@ -4838,8 +4838,8 @@ impl LocalExecutor for ExpeditionExampleExecutor { &self, capability: &traverse_registry::ResolvedCapability, input: &Value, - ) -> Result { - match capability.contract.id.as_str() { + ) -> Result { + let value = match capability.contract.id.as_str() { "expedition.planning.capture-expedition-objective" => { execute_capture_expedition_objective(input) } @@ -4860,7 +4860,11 @@ impl LocalExecutor for ExpeditionExampleExecutor { other => Err(executor_failure(&format!( "unsupported expedition example capability: {other}" ))), - } + }?; + Ok(LocalExecutionOutput { + value, + emitted_events: Vec::new(), + }) } } diff --git a/crates/traverse-mcp/src/lib.rs b/crates/traverse-mcp/src/lib.rs index 44b18ff2..bcd6d697 100644 --- a/crates/traverse-mcp/src/lib.rs +++ b/crates/traverse-mcp/src/lib.rs @@ -566,8 +566,8 @@ mod tests { WorkflowRegistration, }; use traverse_runtime::{ - LocalExecutionFailure, RuntimeContext, RuntimeIntent, RuntimeLookup, RuntimeLookupScope, - RuntimeResultStatus, + LocalExecutionFailure, LocalExecutionOutput, RuntimeContext, RuntimeIntent, RuntimeLookup, + RuntimeLookupScope, RuntimeResultStatus, }; #[test] @@ -1318,8 +1318,11 @@ mod tests { &self, _capability: &ResolvedCapability, _input: &Value, - ) -> Result { - Ok(json!({"draft_id": "draft-001"})) + ) -> Result { + Ok(LocalExecutionOutput { + value: json!({"draft_id": "draft-001"}), + emitted_events: Vec::new(), + }) } } diff --git a/crates/traverse-mcp/src/stdio_server.rs b/crates/traverse-mcp/src/stdio_server.rs index 80c4ca3b..7fce6f2b 100644 --- a/crates/traverse-mcp/src/stdio_server.rs +++ b/crates/traverse-mcp/src/stdio_server.rs @@ -1073,8 +1073,9 @@ impl LocalExecutor for ExpeditionExampleExecutor { &self, capability: &traverse_registry::ResolvedCapability, input: &Value, - ) -> Result { - match capability.contract.id.as_str() { + ) -> Result + { + let value = match capability.contract.id.as_str() { "expedition.planning.capture-expedition-objective" => { execute_capture_expedition_objective(input) } @@ -1091,7 +1092,11 @@ impl LocalExecutor for ExpeditionExampleExecutor { other => Err(executor_failure(&format!( "unsupported expedition capability for stdio execution: {other}" ))), - } + }?; + Ok(traverse_runtime::LocalExecutionOutput { + value, + emitted_events: Vec::new(), + }) } } diff --git a/crates/traverse-runtime/examples/load_workspace_app_state.rs b/crates/traverse-runtime/examples/load_workspace_app_state.rs index fa177512..f752a8a9 100644 --- a/crates/traverse-runtime/examples/load_workspace_app_state.rs +++ b/crates/traverse-runtime/examples/load_workspace_app_state.rs @@ -4,7 +4,7 @@ use std::path::Path; use traverse_registry::{ DiscoveryQuery, LookupScope, ResolvedCapability, WorkspaceAppStateErrorCode, }; -use traverse_runtime::{LocalExecutionFailure, LocalExecutor, Runtime}; +use traverse_runtime::{LocalExecutionFailure, LocalExecutionOutput, LocalExecutor, Runtime}; #[derive(Debug)] struct ConformanceExecutor; @@ -14,8 +14,11 @@ impl LocalExecutor for ConformanceExecutor { &self, _capability: &ResolvedCapability, _input: &Value, - ) -> Result { - Ok(json!({"status": "not_executed"})) + ) -> Result { + Ok(LocalExecutionOutput { + value: json!({"status": "not_executed"}), + emitted_events: Vec::new(), + }) } } diff --git a/crates/traverse-runtime/src/artifact_router.rs b/crates/traverse-runtime/src/artifact_router.rs index f343f719..6013bbfa 100644 --- a/crates/traverse-runtime/src/artifact_router.rs +++ b/crates/traverse-runtime/src/artifact_router.rs @@ -1,13 +1,16 @@ use crate::executor::ExecutorError; #[cfg(feature = "wasmtime-executor")] use crate::executor::{ArtifactType, CapabilityExecutor, ExecutorCapability, WasmExecutor}; -use crate::{LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutor}; +use crate::{ + LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutionOutput, LocalExecutor, +}; use serde_json::Value; use std::collections::BTreeMap; use std::sync::Arc; use traverse_registry::ResolvedCapability; -type NativeHandler = dyn Fn(&Value) -> Result + Send + Sync; +type NativeHandler = + dyn Fn(&Value) -> Result + Send + Sync; /// Production local-execution boundary for registered artifacts. /// @@ -48,7 +51,10 @@ impl ArtifactRouter { /// Registers one host-provided native handler for an exact capability id. pub fn register_native_handler(&mut self, capability_id: impl Into, handler: F) where - F: Fn(&Value) -> Result + Send + Sync + 'static, + F: Fn(&Value) -> Result + + Send + + Sync + + 'static, { self.native_handlers .insert(capability_id.into(), Arc::new(handler)); @@ -60,7 +66,7 @@ impl LocalExecutor for ArtifactRouter { &self, capability: &ResolvedCapability, input: &Value, - ) -> Result { + ) -> Result { if let Some(binary) = &capability.artifact.binary { #[cfg(feature = "wasmtime-executor")] { @@ -80,16 +86,20 @@ impl LocalExecutor for ArtifactRouter { service_type: capability.contract.service_type.clone(), }; // Events emitted via `traverse_host::emit_event` during this - // call are intentionally discarded here: `ArtifactRouter` - // bridges directly into `LocalExecutor`, bypassing - // `PlacementRouter` Step 5's `EventBroker` publish (a - // pre-existing gap for workflow-internal node execution, - // unrelated to this ABI — see spec 098's Capability - // Boundary). + // call are returned as real `LocalExecutionOutput.emitted_events` + // (spec 101-local-executor-event-emission FR-003) — already + // ABI-validated by `WasmExecutor`. `ArtifactRouter` itself + // does not publish them (FR-004): it is used both directly + // by `workflows.rs` and, via `BoundLocalExecutor`, by + // `PlacementRouter` Step 5, so publishing here would + // double-publish on the live `Runtime::execute()` path. return self .wasm .execute(&executor_capability, input) - .map(|output| output.value) + .map(|output| LocalExecutionOutput { + value: output.value, + emitted_events: output.emitted_events, + }) .map_err(|error| map_executor_error(&error)); } #[cfg(not(feature = "wasmtime-executor"))] @@ -212,11 +222,17 @@ mod tests { assert_eq!(failure.code, LocalExecutionFailureCode::ConstraintViolated); router.register_native_handler("hello.world.say-hello", |_| { - Ok(serde_json::json!({"ok": true})) + Ok(LocalExecutionOutput { + value: serde_json::json!({"ok": true}), + emitted_events: Vec::new(), + }) }); assert_eq!( router.execute(&capability, &serde_json::json!({})), - Ok(serde_json::json!({"ok": true})) + Ok(LocalExecutionOutput { + value: serde_json::json!({"ok": true}), + emitted_events: Vec::new(), + }) ); } @@ -236,6 +252,69 @@ mod tests { assert_eq!(failure.message, "registered artifact execution failed"); } + #[cfg(feature = "wasmtime-executor")] + #[test] + fn wasm_execution_success_returns_real_value_and_emitted_events() { + use sha2::{Digest, Sha256}; + use std::fmt::Write as _; + + // Spec 101-local-executor-event-emission FR-003: on a successful + // WASM execution, `ArtifactRouter` must return the executor's real + // `value`/`emitted_events`, not discard or reshape them. + let wat_src = r#" + (module + (import "wasi_snapshot_preview1" "fd_write" + (func $fd_write (param i32 i32 i32 i32) (result i32))) + (memory (export "memory") 1) + (data (i32.const 8) "{}") + (func $_start (export "_start") + (i32.store (i32.const 0) (i32.const 8)) + (i32.store (i32.const 4) (i32.const 2)) + (drop (call $fd_write (i32.const 1) (i32.const 0) (i32.const 1) (i32.const 4))) + ) + ) + "#; + let wasm_bytes = wat::parse_str(wat_src).expect("WAT source should parse"); + + let mut hasher = Sha256::new(); + hasher.update(&wasm_bytes); + let checksum = hasher + .finalize() + .iter() + .fold(String::new(), |mut acc, byte| { + let _ = write!(acc, "{byte:02x}"); + acc + }); + + let tmp = format!( + "/tmp/traverse-artifact-router-test-{}.wasm", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_nanos()) + ); + std::fs::write(&tmp, &wasm_bytes).expect("temp wasm module should write"); + + let mut capability = resolved_capability(Some(BinaryReference { + format: BinaryFormat::Wasm, + location: tmp.clone(), + signature: None, + })); + capability.artifact.digests.binary_digest = Some(format!("sha256:{checksum}")); + + let result = ArtifactRouter::new() + .expect("router should initialize") + .execute(&capability, &serde_json::json!({})); + std::fs::remove_file(&tmp).ok(); + + assert_eq!( + result, + Ok(LocalExecutionOutput { + value: serde_json::json!({}), + emitted_events: Vec::new(), + }) + ); + } + #[test] fn executor_errors_map_to_stable_local_failure_codes() { let errors = [ diff --git a/crates/traverse-runtime/src/lib.rs b/crates/traverse-runtime/src/lib.rs index 3ac9ca3e..4931e7d1 100644 --- a/crates/traverse-runtime/src/lib.rs +++ b/crates/traverse-runtime/src/lib.rs @@ -36,7 +36,8 @@ use std::path::Path; use std::sync::{Arc, Mutex}; use trace::TraceStore; use traverse_contracts::{ - EventReference, ExecutionTarget, HostApiAccess, Lifecycle, NetworkAccess, ViolationRecord, + EventReference, ExecutionTarget, HostApiAccess, Lifecycle, NetworkAccess, ServiceType, + ViolationRecord, }; use traverse_registry::{ CapabilityRegistration, CapabilityRegistry, DiscoveryQuery, ImplementationKind, LookupScope, @@ -318,7 +319,16 @@ pub trait LocalExecutor: Send + Sync { &self, capability: &ResolvedCapability, input: &Value, - ) -> Result; + ) -> Result; +} + +/// A [`LocalExecutor`]'s output: the capability's JSON value plus any events +/// it emitted (spec 101-local-executor-event-emission FR-001). Mirrors +/// [`crate::executor::ExecutorOutput`] for the WASM `CapabilityExecutor` path. +#[derive(Debug, Clone, PartialEq)] +pub struct LocalExecutionOutput { + pub value: Value, + pub emitted_events: Vec, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -1636,13 +1646,19 @@ where ); } - // `BoundLocalExecutor` always returns an empty - // `emitted_events` (the WASM-only `traverse_host::emit_event` - // ABI has no equivalent for a host-provided `LocalExecutor` - // closure), so `response.emitted_events` is structurally - // always empty here regardless of what the caller's executor - // does internally. - let emitted_events: Vec = Vec::new(); + // `BoundLocalExecutor` now threads a `LocalExecutor`'s real + // `emitted_events` into `ExecutorOutput` (spec + // 101-local-executor-event-emission FR-002), already + // validated and published by `PlacementRouter` Step 5, so + // `response.emitted_events` carries the real events here. + let emitted_events: Vec = response + .emitted_events + .iter() + .map(|event| EventReference { + event_id: event.event_type.clone(), + version: event.version.clone(), + }) + .collect(); successful_execution_outcome( context, selected, @@ -1686,16 +1702,56 @@ where _capability: &ExecutorCapability, input: &Value, ) -> Result { - self.executor + let output = self + .executor .execute(&self.selected, input) - .map(|value| ExecutorOutput { - value, - emitted_events: Vec::new(), - }) - .map_err(|failure| ExecutorError::ExecutionFailed(failure.message)) + .map_err(|failure| ExecutorError::ExecutionFailed(failure.message))?; + validate_natively_emitted_events(&self.selected.contract, &output.emitted_events) + .map_err(ExecutorError::ExecutionFailed)?; + Ok(ExecutorOutput { + value: output.value, + emitted_events: output.emitted_events, + }) } } +/// Validates events a native [`LocalExecutor`] implementor populated +/// directly (not via the WASM `traverse_host::emit_event` ABI, which +/// validates synchronously at call time) against the executing capability +/// contract's `emits` list and `service_type`, mirroring spec +/// `098-capability-event-host-abi`'s WASM-boundary checks (spec +/// `101-local-executor-event-emission` FR-007/FR-008). Runs after the +/// native closure has already returned — there is no host-function call +/// boundary to reject mid-call the way WASM has — but always before any +/// publish to [`EventBroker`]. +fn validate_natively_emitted_events( + contract: &traverse_contracts::CapabilityContract, + emitted_events: &[TraverseEvent], +) -> Result<(), String> { + if emitted_events.is_empty() { + return Ok(()); + } + if contract.service_type != ServiceType::Subscribable { + return Err(format!( + "capability '{}' emitted events but its service_type is not Subscribable", + contract.id + )); + } + for event in emitted_events { + let declared = contract + .emits + .iter() + .any(|decl| decl.event_id == event.event_type && decl.version == event.version); + if !declared { + return Err(format!( + "capability '{}' emitted an undeclared event {}@{}", + contract.id, event.event_type, event.version + )); + } + } + Ok(()) +} + /// Best-effort broker used only when the default in-process broker cannot be constructed. #[derive(Debug, Default)] struct DiscardEventBroker; @@ -3580,16 +3636,19 @@ mod tests { assert!(discard.cancel("missing").is_err()); } - #[test] - fn bound_local_executor_never_publishes_events_through_placement_router() { - // `BoundLocalExecutor` bridges a host-provided `LocalExecutor` (a - // native Rust closure) into `CapabilityExecutor`. Spec - // 098-capability-event-host-abi's `traverse_host::emit_event` is a - // WASM-only mechanism (FR-001) — a native closure has no ABI to call - // and the output-JSON `emitted_events` convention it used to rely on - // is removed (FR-004), so `BoundLocalExecutor::execute` always - // returns an empty `emitted_events` list. This test documents that - // intentional gap rather than asserting an event reaches the broker. + /// Shared harness for the `bound_local_executor_*` publish tests: builds a + /// broker with one registered/subscribed event type, a `Subscribable` + /// resolved capability declaring that event in its `emits` list, and runs + /// the given `LocalExecutor` through `PlacementRouter` exactly as the live + /// `Runtime::execute()` path does. Returns the router response and the + /// broker (so callers can poll it) plus the subscription id. + fn run_native_executor_through_placement_router( + executor: E, + ) -> ( + Result, + Arc, + String, + ) { use super::events::{ EventBroker, EventCatalog, EventCatalogEntry, InProcessBroker, LifecycleStatus, }; @@ -3597,20 +3656,181 @@ mod tests { use super::placement::PlacementConstraintEvaluator; use super::router::{CapabilityExecutorRegistry, PlacementRouter, RouterRequest}; use super::trace::TraceStore; - use super::{LocalExecutionFailure, LocalExecutor}; - use serde_json::Value; + let event_type = "dev.traverse.native.live-emitted"; + let catalog = Arc::new(EventCatalog::new()); + catalog + .register(EventCatalogEntry { + event_type: event_type.to_string(), + owner: "native.live".to_string(), + version: "1.0.0".to_string(), + lifecycle_status: LifecycleStatus::Active, + consumer_count: 0, + }) + .expect("catalog entry should register"); + let broker = Arc::new(InProcessBroker::new(catalog).expect("broker should construct")); + let subscription = broker + .subscribe(event_type, "0") + .expect("subscribe should succeed"); + let trace_store = Arc::new(Mutex::new(TraceStore::new())); + + let mut selected = resolved_capability(None, Lifecycle::Active); + selected.contract.service_type = ServiceType::Subscribable; + selected.contract.event_trigger = Some("dev.traverse.native.trigger".to_string()); + selected.contract.emits = vec![traverse_contracts::EventReference { + event_id: event_type.to_string(), + version: "1.0.0".to_string(), + }]; + selected.contract.permitted_targets = vec![ExecutionTarget::Local, ExecutionTarget::Cloud]; + + let mut registry = CapabilityExecutorRegistry::new(); + registry.insert( + ArtifactType::Native, + Box::new(super::BoundLocalExecutor { + executor: Arc::new(executor), + selected: selected.clone(), + }), + ); + let router = PlacementRouter::new( + PlacementConstraintEvaluator, + registry, + Arc::clone(&trace_store), + broker.clone(), + ); + + let response = router.execute(RouterRequest { + capability_id: selected.record.id.clone(), + artifact_type: ArtifactType::Native, + contract: selected.contract.clone(), + target_hint: Some(ExecutionTarget::Local), + runtime_snapshot: super::idle_runtime_snapshot(), + input: json!({}), + executor_capability: super::executor_capability_for(&selected, ArtifactType::Native), + trace_id_override: Some("trace_native_live".to_string()), + }); + + (response, broker, subscription.subscription_id) + } + + fn native_traverse_event(event_type: &str) -> super::events::TraverseEvent { + super::events::TraverseEvent { + id: "native-event-1".to_string(), + source: "traverse-runtime/test.native".to_string(), + event_type: event_type.to_string(), + datacontenttype: "application/json".to_string(), + time: "2026-01-01T00:00:00Z".to_string(), + data: json!({}), + owner: "test.native".to_string(), + version: "1.0.0".to_string(), + lifecycle_status: super::events::LifecycleStatus::Active, + deduplication_id: Some("native-event-1".to_string()), + ordering_scope: Some("test.native".to_string()), + correlation_id: None, + causation_id: None, + subject_id: None, + actor_id: None, + } + } + + #[test] + fn bound_local_executor_publishes_declared_native_events_through_placement_router() { + use super::events::EventBroker; + use serde_json::Value; + // Spec 101-local-executor-event-emission FR-002: `BoundLocalExecutor` + // now threads a native `LocalExecutor`'s real `emitted_events` into + // `ExecutorOutput`, so `PlacementRouter` Step 5 publishes them for + // `Subscribable` capabilities exactly as it already does for the WASM + // `CapabilityExecutor` path. This replaces the old test documenting + // that gap as expected behavior. struct NativeEmitExecutor; - impl LocalExecutor for NativeEmitExecutor { + impl super::LocalExecutor for NativeEmitExecutor { + fn execute( + &self, + _capability: &ResolvedCapability, + _input: &Value, + ) -> Result { + Ok(super::LocalExecutionOutput { + value: json!({ "draft_id": "native-1" }), + emitted_events: vec![native_traverse_event("dev.traverse.native.live-emitted")], + }) + } + } + + let (response, broker, subscription_id) = + run_native_executor_through_placement_router(NativeEmitExecutor); + let response = response.expect("native placement router execution should succeed"); + + assert_eq!(response.trace_id, "trace_native_live"); + assert_eq!(response.emitted_events.len(), 1); + let poll = broker + .poll(&subscription_id, 10) + .expect("poll should succeed"); + assert_eq!(poll.events.len(), 1); + assert_eq!( + poll.events[0].event.event_type, + "dev.traverse.native.live-emitted" + ); + } + + #[test] + fn bound_local_executor_rejects_undeclared_native_event() { + use super::events::EventBroker; + use serde_json::Value; + // Spec 101-local-executor-event-emission FR-007/FR-008: an event a + // native `LocalExecutor` populates that is not in the capability + // contract's `emits` list must fail the whole execution, not be + // silently dropped or published. + struct UndeclaredEmitExecutor; + impl super::LocalExecutor for UndeclaredEmitExecutor { + fn execute( + &self, + _capability: &ResolvedCapability, + _input: &Value, + ) -> Result { + Ok(super::LocalExecutionOutput { + value: json!({ "draft_id": "native-1" }), + emitted_events: vec![native_traverse_event("dev.traverse.native.undeclared")], + }) + } + } + + let (response, broker, subscription_id) = + run_native_executor_through_placement_router(UndeclaredEmitExecutor); + assert!(response.is_err()); + let poll = broker + .poll(&subscription_id, 10) + .expect("poll should succeed"); + assert!(poll.events.is_empty()); + } + + #[test] + fn bound_local_executor_rejects_native_event_from_non_subscribable_capability() { + use serde_json::Value; + // Spec 101-local-executor-event-emission FR-007/FR-008: a + // non-`Subscribable` capability that populates `emitted_events` must + // fail the whole execution, mirroring the WASM ABI's FR-003 gate. + struct NonSubscribableEmitExecutor; + impl super::LocalExecutor for NonSubscribableEmitExecutor { fn execute( &self, _capability: &ResolvedCapability, _input: &Value, - ) -> Result { - Ok(json!({ "draft_id": "native-1" })) + ) -> Result { + Ok(super::LocalExecutionOutput { + value: json!({ "draft_id": "native-1" }), + emitted_events: vec![native_traverse_event("dev.traverse.native.live-emitted")], + }) } } + use super::events::{ + EventBroker, EventCatalog, EventCatalogEntry, InProcessBroker, LifecycleStatus, + }; + use super::executor::ArtifactType; + use super::placement::PlacementConstraintEvaluator; + use super::router::{CapabilityExecutorRegistry, PlacementRouter, RouterRequest}; + use super::trace::TraceStore; + let event_type = "dev.traverse.native.live-emitted"; let catalog = Arc::new(EventCatalog::new()); catalog @@ -3629,8 +3849,7 @@ mod tests { let trace_store = Arc::new(Mutex::new(TraceStore::new())); let mut selected = resolved_capability(None, Lifecycle::Active); - selected.contract.service_type = ServiceType::Subscribable; - selected.contract.event_trigger = Some("dev.traverse.native.trigger".to_string()); + selected.contract.service_type = ServiceType::Stateless; selected.contract.emits = vec![traverse_contracts::EventReference { event_id: event_type.to_string(), version: "1.0.0".to_string(), @@ -3641,7 +3860,7 @@ mod tests { registry.insert( ArtifactType::Native, Box::new(super::BoundLocalExecutor { - executor: Arc::new(NativeEmitExecutor), + executor: Arc::new(NonSubscribableEmitExecutor), selected: selected.clone(), }), ); @@ -3652,34 +3871,22 @@ mod tests { broker.clone(), ); - let response = router - .execute(RouterRequest { - capability_id: selected.record.id.clone(), - artifact_type: ArtifactType::Native, - contract: selected.contract.clone(), - target_hint: Some(ExecutionTarget::Local), - runtime_snapshot: super::idle_runtime_snapshot(), - input: json!({}), - executor_capability: super::executor_capability_for( - &selected, - ArtifactType::Native, - ), - trace_id_override: Some("trace_native_live".to_string()), - }) - .expect("native placement router execution should succeed"); + let response = router.execute(RouterRequest { + capability_id: selected.record.id.clone(), + artifact_type: ArtifactType::Native, + contract: selected.contract.clone(), + target_hint: Some(ExecutionTarget::Local), + runtime_snapshot: super::idle_runtime_snapshot(), + input: json!({}), + executor_capability: super::executor_capability_for(&selected, ArtifactType::Native), + trace_id_override: Some("trace_native_live_non_subscribable".to_string()), + }); - assert_eq!(response.trace_id, "trace_native_live"); + assert!(response.is_err()); let poll = broker .poll(&subscription.subscription_id, 10) .expect("poll should succeed"); assert!(poll.events.is_empty()); - assert!( - trace_store - .lock() - .expect("trace lock") - .get("trace_native_live") - .is_some() - ); } #[test] @@ -4964,7 +5171,13 @@ mod tests { let result = executor.execute(&capability, &json!({})); - assert_eq!(result, Ok(json!({"draft_id": "draft"}))); + assert_eq!( + result, + Ok(super::LocalExecutionOutput { + value: json!({"draft_id": "draft"}), + emitted_events: Vec::new(), + }) + ); } #[test] @@ -5520,8 +5733,11 @@ mod tests { &self, _capability: &ResolvedCapability, _input: &serde_json::Value, - ) -> Result { - Ok(json!({"draft_id": "draft"})) + ) -> Result { + Ok(super::LocalExecutionOutput { + value: json!({"draft_id": "draft"}), + emitted_events: Vec::new(), + }) } } @@ -5532,7 +5748,7 @@ mod tests { &self, _capability: &ResolvedCapability, _input: &serde_json::Value, - ) -> Result { + ) -> Result { Err(super::LocalExecutionFailure { code: super::LocalExecutionFailureCode::ExecutionFailed, message: "forced failure".to_string(), diff --git a/crates/traverse-runtime/src/workflows.rs b/crates/traverse-runtime/src/workflows.rs index cef737e1..67d0d327 100644 --- a/crates/traverse-runtime/src/workflows.rs +++ b/crates/traverse-runtime/src/workflows.rs @@ -8,7 +8,7 @@ use crate::{ use serde::{Deserialize, Serialize}; use serde_json::{Map, Value, json}; use std::collections::BTreeSet; -use traverse_contracts::EventReference; +use traverse_contracts::{EventReference, ServiceType}; use traverse_registry::{ LookupScope, RegistryScope, ResolvedCapability, ResolvedWorkflow, WorkflowEdge, WorkflowEdgePredicate, WorkflowEdgeTrigger, WorkflowNode, @@ -601,7 +601,7 @@ where }; if let Err(error) = validate_payload_against_contract( - &output, + &output.value, &capability.contract.outputs.schema, RuntimeErrorCode::OutputValidationFailed, "workflow node output does not satisfy the capability output contract", @@ -622,8 +622,64 @@ where )); } - update_state(&mut state, node, &output); - let node_emitted = emitted_events(&output); + // Spec 101-local-executor-event-emission FR-007/FR-008: validate + // natively-populated events before using them for anything, + // mirroring spec 098's WASM-boundary checks. + if let Err(validation_message) = crate::validate_natively_emitted_events( + &capability.contract, + &output.emitted_events, + ) { + let mut failed = visited; + if let Some(last) = failed.last_mut() { + last.status = WorkflowTraversalStepStatus::Failed; + } + return Err(workflow_failure( + request, + WorkflowTraversalFailureReason::StepExecutionFailed, + runtime_error( + RuntimeErrorCode::ContractViolation, + &validation_message, + json!({"node_id": node.node_id}), + ), + failed, + traversed, + emitted, + event_evidence, + warnings, + )); + } + + update_state(&mut state, node, &output.value); + + // Spec 101-local-executor-event-emission FR-005: publish to + // EventBroker for Subscribable capabilities, structurally + // analogous to PlacementRouter Step 5 (same gate, same + // best-effort semantics — a publish error is recorded but does + // not fail the workflow step). + if capability.contract.service_type == ServiceType::Subscribable { + for event in &output.emitted_events { + let _ = self.event_broker.publish(event.clone()); + } + } + + // Spec 101-local-executor-event-emission FR-006: Pass-1 + // event-driven edge matching reads from the structured + // `LocalExecutionOutput.emitted_events` field, not a + // JSON-parsed "emitted_events" convention key. + let node_emitted: Vec = output + .emitted_events + .iter() + .enumerate() + .map(|(index, event)| EmittedEventRecord { + record_id: format!("event_record_{index}"), + event: EventReference { + event_id: event.event_type.clone(), + version: event.version.clone(), + }, + payload: Some(event.data.clone()), + source: None, + }) + .collect(); emitted.extend(node_emitted.iter().map(|record| record.event.clone())); if let Some(last) = visited.last_mut() { last.status = WorkflowTraversalStepStatus::Completed; @@ -979,33 +1035,6 @@ fn final_workflow_output(state: &Map, output_projection: &[String Value::Object(projected) } -fn emitted_events(output: &Value) -> Vec { - let Value::Object(object) = output else { - return Vec::new(); - }; - let Some(Value::Array(events)) = object.get("emitted_events") else { - return Vec::new(); - }; - events - .iter() - .enumerate() - .filter_map(|(index, event)| { - let Value::Object(event) = event else { - return None; - }; - Some(EmittedEventRecord { - record_id: format!("event_record_{index}"), - event: EventReference { - event_id: event.get("event_id")?.as_str()?.to_string(), - version: event.get("version")?.as_str()?.to_string(), - }, - payload: event.get("payload").cloned(), - source: None, - }) - }) - .collect() -} - fn waiting_edge_contexts( workflow_execution_id: &str, edges: &[WorkflowEdge], @@ -1358,8 +1387,8 @@ mod tests { use crate::security::RuntimeSecurityConfig; use crate::{ CandidateCollectionRecord, LocalExecutionFailure, LocalExecutionFailureCode, - RuntimeContext, RuntimeIntent, RuntimeLookup, RuntimeLookupScope, RuntimeRequest, - RuntimeResultStatus, SelectionRecord, + LocalExecutionOutput, RuntimeContext, RuntimeIntent, RuntimeLookup, RuntimeLookupScope, + RuntimeRequest, RuntimeResultStatus, SelectionRecord, }; use serde_json::json; use std::sync::Arc; @@ -1411,7 +1440,7 @@ mod tests { } #[test] - fn workflow_helpers_cover_state_and_event_extraction_paths() { + fn workflow_helpers_cover_state_and_edge_paths() { let scalar = workflow_state(&json!("value")); assert_eq!(scalar.get("input"), Some(&json!("value"))); @@ -1433,30 +1462,6 @@ mod tests { assert_eq!(state.get("draft_id"), Some(&json!("draft-1"))); update_state(&mut state, &node, &json!("not-an-object")); - assert!(emitted_events(&json!("nope")).is_empty()); - assert_eq!( - emitted_events(&json!({ - "emitted_events": [ - "bad", - { - "event_id": "content.comments.draft-created", - "version": "1.0.0", - "payload": {"severity": "normal"} - }, - {"event_id": "bad"} - ] - })), - vec![EmittedEventRecord { - record_id: "event_record_1".to_string(), - event: EventReference { - event_id: "content.comments.draft-created".to_string(), - version: "1.0.0".to_string(), - }, - payload: Some(json!({"severity": "normal"})), - source: None, - }] - ); - let edge = WorkflowEdge { edge_id: "edge".to_string(), from: "a".to_string(), @@ -1604,6 +1609,54 @@ mod tests { ); } + /// Spec 101-local-executor-event-emission FR-005 / acceptance scenario 2: + /// a workflow node's declared, `Subscribable`-gated event both satisfies + /// same-execution waiting edges (already covered by + /// `executes_workflow_deterministically_and_supports_workflow_backed_capabilities`) + /// AND is published to `EventBroker` for external consumers, mirroring + /// `PlacementRouter` Step 5's publish semantics. + #[test] + fn workflow_node_emitted_events_are_published_to_event_broker() { + let broker = event_catalog_broker_fixture("content.comments.draft-created", "1.0.0"); + let subscription = broker + .subscribe("content.comments.draft-created", "0") + .unwrap_or_else(|error| unreachable!("{error:?}")); + + let workflow_registry = workflow_registry_fixture(); + let runtime = Runtime::new(capability_registry_fixture(), WorkflowExecutor) + .with_workflow_registry(workflow_registry) + .with_security_config(RuntimeSecurityConfig::development()) + .with_event_broker(broker.clone()); + + let workflow = runtime.execute_workflow(valid_workflow_request()); + assert_eq!(workflow.result.status, WorkflowTraversalStatus::Completed); + + let poll = broker + .poll(&subscription.subscription_id, 10) + .unwrap_or_else(|error| unreachable!("{error:?}")); + assert_eq!(poll.events.len(), 1); + assert_eq!( + poll.events[0].event.event_type, + "content.comments.draft-created" + ); + } + + /// Spec 101-local-executor-event-emission FR-005 / acceptance scenario 7: + /// an `EventBroker` publish failure at workflow-node publish time is + /// recorded (best-effort, matching `PlacementRouter` Step 5) but does not + /// fail the workflow step — a broker outage does not fail traversal. + #[test] + fn workflow_node_publish_failure_does_not_fail_workflow_step() { + let workflow_registry = workflow_registry_fixture(); + let runtime = Runtime::new(capability_registry_fixture(), WorkflowExecutor) + .with_workflow_registry(workflow_registry) + .with_security_config(RuntimeSecurityConfig::development()) + .with_event_broker(Arc::new(AlwaysFailingBroker)); + + let workflow = runtime.execute_workflow(valid_workflow_request()); + assert_eq!(workflow.result.status, WorkflowTraversalStatus::Completed); + } + #[test] fn workflow_failures_cover_not_found_missing_events_and_step_failures() { let workflow_registry = workflow_registry_fixture(); @@ -2435,6 +2488,22 @@ mod tests { Some(WorkflowTraversalFailureReason::StepExecutionFailed) ); + // Spec 101-local-executor-event-emission FR-007/FR-008: a workflow + // node that emits an event its contract does not declare must fail + // the step with a contract-violation error, mirroring + // `PlacementRouter`'s native-boundary validation. + let undeclared_event_runtime = Runtime::new( + capability_registry_fixture(), + UndeclaredEventWorkflowExecutor, + ) + .with_workflow_registry(workflow_registry_fixture()) + .with_security_config(RuntimeSecurityConfig::development()); + let undeclared_event = undeclared_event_runtime.execute_workflow(valid_workflow_request()); + assert_eq!( + undeclared_event.evidence.result.failure_reason, + Some(WorkflowTraversalFailureReason::StepExecutionFailed) + ); + let direct_success = runtime.traverse_workflow( &valid_workflow_request(), &resolved_workflow(workflow_definition_fixture( @@ -3089,8 +3158,8 @@ mod tests { &self, capability: &ResolvedCapability, _input: &Value, - ) -> Result { - let output = match capability.record.id.as_str() { + ) -> Result { + let value = match capability.record.id.as_str() { "content.comments.pipeline-validate" => json!({"valid": true, "issues": []}), "content.comments.pipeline-process" => json!({ "title": "Hello world", @@ -3101,7 +3170,10 @@ mod tests { }), _ => json!({"summary": "Hello world (fleeting)", "wordCount": 3}), }; - Ok(output) + Ok(LocalExecutionOutput { + value, + emitted_events: Vec::new(), + }) } } @@ -3112,9 +3184,12 @@ mod tests { &self, capability: &ResolvedCapability, _input: &Value, - ) -> Result { + ) -> Result { match capability.record.id.as_str() { - "content.comments.pipeline-validate" => Ok(json!({"valid": true, "issues": []})), + "content.comments.pipeline-validate" => Ok(LocalExecutionOutput { + value: json!({"valid": true, "issues": []}), + emitted_events: Vec::new(), + }), other => Err(LocalExecutionFailure { code: LocalExecutionFailureCode::ExecutionFailed, message: format!("step failed: {other}"), @@ -3390,6 +3465,12 @@ mod tests { inputs: Value, outputs: Value, ) -> CapabilityContract { + // Subscribable requires a non-empty event_trigger (registry + // ContractValidationFailed otherwise); only capabilities that + // actually declare `emits` need to be Subscribable at all (spec + // 101-local-executor-event-emission FR-007/FR-008 rejects a + // non-Subscribable capability's emitted events). + let has_emits = !emits.is_empty(); CapabilityContract { kind: "capability_contract".to_string(), schema_version: "1.0.0".to_string(), @@ -3451,14 +3532,22 @@ mod tests { evidence_type: EvidenceType::ContractValidation, status: EvidenceStatus::Passed, }], - service_type: ServiceType::Stateless, + service_type: if has_emits { + ServiceType::Subscribable + } else { + ServiceType::Stateless + }, permitted_targets: vec![ ExecutionTarget::Local, ExecutionTarget::Cloud, ExecutionTarget::Edge, ExecutionTarget::Device, ], - event_trigger: None, + event_trigger: if has_emits { + Some(format!("{id}.triggered")) + } else { + None + }, connector_requirements: Vec::new(), state_schema: None, } @@ -3513,26 +3602,28 @@ mod tests { &self, capability: &ResolvedCapability, _input: &Value, - ) -> Result { - let output = match capability.record.id.as_str() { - "content.comments.create-comment-draft" => json!({ - "draft_id": "draft-1", - "emitted_events": [ - {"event_id": "content.comments.draft-created", "version": "1.0.0"} - ] - }), - "content.comments.validate-comment" => json!({ - "draft_id": "draft-1", - "emitted_events": [ - {"event_id": "content.comments.validated", "version": "1.0.0"} - ] - }), - "content.comments.persist-comment" => json!({ - "comment_id": "comment-1" - }), - _ => json!({}), + ) -> Result { + let (value, emitted_events) = match capability.record.id.as_str() { + "content.comments.create-comment-draft" => ( + json!({"draft_id": "draft-1"}), + vec![sample_traverse_event( + "content.comments.draft-created", + "1.0.0", + )], + ), + "content.comments.validate-comment" => ( + json!({"draft_id": "draft-1"}), + vec![sample_traverse_event("content.comments.validated", "1.0.0")], + ), + "content.comments.persist-comment" => { + (json!({"comment_id": "comment-1"}), Vec::new()) + } + _ => (json!({}), Vec::new()), }; - Ok(output) + Ok(LocalExecutionOutput { + value, + emitted_events, + }) } } @@ -3543,7 +3634,7 @@ mod tests { &self, _capability: &ResolvedCapability, _input: &Value, - ) -> Result { + ) -> Result { Err(LocalExecutionFailure { code: LocalExecutionFailureCode::ExecutionFailed, message: "boom".to_string(), @@ -3560,23 +3651,25 @@ mod tests { &self, capability: &ResolvedCapability, _input: &Value, - ) -> Result { - let output = match capability.record.id.as_str() { - "content.comments.create-comment-draft" => json!({ - "draft_id": "draft-1", - "emitted_events": [ - {"event_id": "content.comments.draft-created", "version": "1.0.0"} - ] - }), - "content.comments.validate-comment" => json!({ - "draft_id": "draft-1" - }), - "content.comments.persist-comment" => json!({ - "comment_id": "comment-1" - }), - _ => json!({}), + ) -> Result { + let (value, emitted_events) = match capability.record.id.as_str() { + "content.comments.create-comment-draft" => ( + json!({"draft_id": "draft-1"}), + vec![sample_traverse_event( + "content.comments.draft-created", + "1.0.0", + )], + ), + "content.comments.validate-comment" => (json!({"draft_id": "draft-1"}), Vec::new()), + "content.comments.persist-comment" => { + (json!({"comment_id": "comment-1"}), Vec::new()) + } + _ => (json!({}), Vec::new()), }; - Ok(output) + Ok(LocalExecutionOutput { + value, + emitted_events, + }) } } @@ -3585,16 +3678,44 @@ mod tests { &self, capability: &ResolvedCapability, _input: &Value, - ) -> Result { - let output = match capability.record.id.as_str() { - "content.comments.create-comment-draft" => json!({ - "emitted_events": [ - {"event_id": "content.comments.draft-created", "version": "1.0.0"} - ] - }), - _ => json!({}), + ) -> Result { + let (value, emitted_events) = match capability.record.id.as_str() { + "content.comments.create-comment-draft" => ( + json!({}), + vec![sample_traverse_event( + "content.comments.draft-created", + "1.0.0", + )], + ), + _ => (json!({}), Vec::new()), }; - Ok(output) + Ok(LocalExecutionOutput { + value, + emitted_events, + }) + } + } + + struct UndeclaredEventWorkflowExecutor; + + impl LocalExecutor for UndeclaredEventWorkflowExecutor { + fn execute( + &self, + _capability: &ResolvedCapability, + _input: &Value, + ) -> Result { + // The workflow under test in `workflow_failures_cover_not_found_missing_events_and_step_failures` + // fails FR-007/FR-008 validation at its start node, so this + // executor is only ever invoked for + // "content.comments.create-comment-draft" and never reaches a + // second node. + Ok(LocalExecutionOutput { + value: json!({"draft_id": "draft-1"}), + emitted_events: vec![sample_traverse_event( + "content.comments.undeclared-event", + "1.0.0", + )], + }) } } diff --git a/crates/traverse-runtime/tests/doc_approval_pipeline.rs b/crates/traverse-runtime/tests/doc_approval_pipeline.rs index 7ea2abdf..b1b55007 100644 --- a/crates/traverse-runtime/tests/doc_approval_pipeline.rs +++ b/crates/traverse-runtime/tests/doc_approval_pipeline.rs @@ -12,7 +12,7 @@ use traverse_registry::{ }; use traverse_runtime::security::RuntimeSecurityConfig; use traverse_runtime::{ - LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutor, Runtime, + LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutionOutput, LocalExecutor, Runtime, WorkflowExecutionRequest, WorkflowLookupScope, WorkflowTraversalStatus, }; @@ -26,28 +26,34 @@ impl LocalExecutor for DocApprovalExecutor { &self, capability: &ResolvedCapability, input: &Value, - ) -> Result { + ) -> Result { match capability.contract.id.as_str() { "doc-approval.analyze" => { let document = input["document"].as_str().unwrap_or_default(); - Ok(json!({ - "docType": if document.contains("Invoice") { "invoice" } else { "contract" }, - "parties": ["Acme Corp", "Globex LLC"], - "amounts": ["USD 12000.00"], - "confidence": "high", - "recommendation": "review", - })) + Ok(LocalExecutionOutput { + value: json!({ + "docType": if document.contains("Invoice") { "invoice" } else { "contract" }, + "parties": ["Acme Corp", "Globex LLC"], + "amounts": ["USD 12000.00"], + "confidence": "high", + "recommendation": "review", + }), + emitted_events: Vec::new(), + }) } "doc-approval.recommend" => { let confidence = input["confidence"].as_str().unwrap_or_default(); let doc_type = input["docType"].as_str().unwrap_or_default(); - Ok(json!({ - "recommendation": if confidence == "high" { "approve" } else { "escalate" }, - "rationale": format!( - "deterministic {doc_type} analysis with {confidence} confidence" - ), - "confidence": confidence, - })) + Ok(LocalExecutionOutput { + value: json!({ + "recommendation": if confidence == "high" { "approve" } else { "escalate" }, + "rationale": format!( + "deterministic {doc_type} analysis with {confidence} confidence" + ), + "confidence": confidence, + }), + emitted_events: Vec::new(), + }) } other => Err(LocalExecutionFailure { code: LocalExecutionFailureCode::ConstraintViolated, diff --git a/crates/traverse-runtime/tests/placement_router_live_wiring.rs b/crates/traverse-runtime/tests/placement_router_live_wiring.rs index e6db2e39..11ad44b2 100644 --- a/crates/traverse-runtime/tests/placement_router_live_wiring.rs +++ b/crates/traverse-runtime/tests/placement_router_live_wiring.rs @@ -24,9 +24,9 @@ use traverse_runtime::events::{ use traverse_runtime::security::RuntimeSecurityConfig; use traverse_runtime::trace::{TraceOutcome, TraceStore}; use traverse_runtime::{ - ExecutionFailureReason, LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutor, - Runtime, RuntimeContext, RuntimeErrorCode, RuntimeLookup, RuntimeLookupScope, RuntimeRequest, - RuntimeResultStatus, RuntimeState, + ExecutionFailureReason, LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutionOutput, + LocalExecutor, Runtime, RuntimeContext, RuntimeErrorCode, RuntimeLookup, RuntimeLookupScope, + RuntimeRequest, RuntimeResultStatus, RuntimeState, }; #[derive(Debug, Default)] @@ -53,10 +53,38 @@ struct EmittingNativeExecutor; impl LocalExecutor for EmittingNativeExecutor { fn execute( &self, - _capability: &traverse_registry::ResolvedCapability, + capability: &traverse_registry::ResolvedCapability, _input: &Value, - ) -> Result { - Ok(json!({ "draft_id": "draft-live-001" })) + ) -> Result { + // Spec 101-local-executor-event-emission FR-001/FR-002: emit every + // event this capability's contract declares, so tests exercising the + // live `Runtime::execute()` path can assert it reaches `EventBroker`. + let emitted_events = capability + .contract + .emits + .iter() + .map(|declared| TraverseEvent { + id: format!("{}-event", capability.contract.id), + source: format!("traverse-runtime/{}", capability.contract.id), + event_type: declared.event_id.clone(), + datacontenttype: "application/json".to_string(), + time: "2026-08-06T00:00:00Z".to_string(), + data: json!({}), + owner: capability.contract.id.clone(), + version: declared.version.clone(), + lifecycle_status: LifecycleStatus::Active, + deduplication_id: Some(format!("{}-event", capability.contract.id)), + ordering_scope: Some(capability.contract.id.clone()), + correlation_id: None, + causation_id: None, + subject_id: None, + actor_id: None, + }) + .collect(); + Ok(LocalExecutionOutput { + value: json!({ "draft_id": "draft-live-001" }), + emitted_events, + }) } } @@ -67,7 +95,7 @@ impl LocalExecutor for FailingLiveExecutor { &self, _capability: &traverse_registry::ResolvedCapability, _input: &Value, - ) -> Result { + ) -> Result { Err(LocalExecutionFailure { code: LocalExecutionFailureCode::ExecutionFailed, message: "live executor failed".to_string(), @@ -284,16 +312,14 @@ fn exact_request(capability_id: &str) -> RuntimeRequest { } #[test] -fn live_native_execution_completes_and_writes_trace_without_publishing_events() { +fn live_native_execution_publishes_declared_events_and_writes_trace() { // `Runtime::execute`'s live path bridges the host-provided `LocalExecutor` // through `BoundLocalExecutor` into `PlacementRouter`. Spec - // 098-capability-event-host-abi's `traverse_host::emit_event` is a - // WASM-only mechanism (FR-001) that a native `LocalExecutor` closure has - // no way to call, and the output-JSON `emitted_events` convention it - // used to rely on is removed (FR-004) — so a native capability can no - // longer emit events at all through this path. This test documents that - // intentional gap: execution and trace recording still succeed, but no - // event reaches `EventBroker` or the trace's `emitted_events`. + // 101-local-executor-event-emission FR-002 threads a native + // `LocalExecutor`'s real `emitted_events` into `ExecutorOutput`, so + // `PlacementRouter` Step 5 publishes them for `Subscribable` capabilities + // exactly as it already does for the WASM `CapabilityExecutor` path. + // This replaces the old test documenting that gap as expected behavior. let event_type = "dev.traverse.live.draft-created"; let broker = broker_with_event(event_type); let subscription = broker @@ -313,12 +339,14 @@ fn live_native_execution_completes_and_writes_trace_without_publishing_events() let outcome = runtime.execute(exact_request("live.wiring.native-subject")); assert_eq!(outcome.result.status, RuntimeResultStatus::Completed); - assert!(outcome.trace.emitted_events.is_empty()); + assert_eq!(outcome.trace.emitted_events.len(), 1); + assert_eq!(outcome.trace.emitted_events[0].event_id, event_type); let poll = broker .poll(&subscription.subscription_id, 10) .expect("poll should succeed"); - assert!(poll.events.is_empty()); + assert_eq!(poll.events.len(), 1); + assert_eq!(poll.events[0].event.event_type, event_type); let store = trace_store.lock().expect("trace store lock"); let entries = store.list_public(Some("live.wiring.native-subject")); diff --git a/crates/traverse-runtime/tests/runtime.rs b/crates/traverse-runtime/tests/runtime.rs index daa96048..06c8cf07 100644 --- a/crates/traverse-runtime/tests/runtime.rs +++ b/crates/traverse-runtime/tests/runtime.rs @@ -15,9 +15,9 @@ use traverse_runtime::security::RuntimeSecurityConfig; use traverse_runtime::{ BrowserRuntimeSubscriptionLifecycleStatus, BrowserRuntimeSubscriptionMessage, BrowserRuntimeSubscriptionRequest, CandidateReason, ExecutionFailureReason, ExecutionStatus, - LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutor, PlacementTarget, Runtime, - RuntimeContext, RuntimeErrorCode, RuntimeLookup, RuntimeLookupScope, RuntimeRequest, - RuntimeResultStatus, RuntimeState, SelectionFailureReason, SelectionStatus, + LocalExecutionFailure, LocalExecutionFailureCode, LocalExecutionOutput, LocalExecutor, + PlacementTarget, Runtime, RuntimeContext, RuntimeErrorCode, RuntimeLookup, RuntimeLookupScope, + RuntimeRequest, RuntimeResultStatus, RuntimeState, SelectionFailureReason, SelectionStatus, browser_subscription_messages, parse_runtime_request, }; @@ -1142,8 +1142,11 @@ impl LocalExecutor for EchoExecutor { &self, _capability: &traverse_registry::ResolvedCapability, _input: &Value, - ) -> Result { - Ok(json!({"draft_id": "draft-001"})) + ) -> Result { + Ok(LocalExecutionOutput { + value: json!({"draft_id": "draft-001"}), + emitted_events: Vec::new(), + }) } } @@ -1154,7 +1157,7 @@ impl LocalExecutor for FailingExecutor { &self, _capability: &traverse_registry::ResolvedCapability, _input: &Value, - ) -> Result { + ) -> Result { Err(LocalExecutionFailure { code: LocalExecutionFailureCode::ExecutionFailed, message: "executor failed".to_string(), @@ -1169,8 +1172,11 @@ impl LocalExecutor for WrongOutputExecutor { &self, _capability: &traverse_registry::ResolvedCapability, _input: &Value, - ) -> Result { - Ok(json!({"missing": "draft_id"})) + ) -> Result { + Ok(LocalExecutionOutput { + value: json!({"missing": "draft_id"}), + emitted_events: Vec::new(), + }) } }