feat(ocsf): carry structured events on sandbox log stream

crates/openshell-cli/src/run.rs

@@ -9020,6 +9020,7 @@ mod tests {
                 .iter()
                 .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
                 .collect(),
+            ocsf_json: Vec::new(),
         }
     }
 

crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs

@@ -523,6 +523,7 @@ impl OpenShell for TestOpenShell {
                                 message: message.to_string(),
                                 source: "gateway".to_string(),
                                 fields: HashMap::new(),
+                                ocsf_json: Vec::new(),
                             })),
                         }))
                         .await;

crates/openshell-server/src/tracing_bus.rs

@@ -113,6 +113,10 @@ impl TracingLogBus {
         if log.sandbox_id.is_empty() {
             return;
         }
+        let mut log = log;
+        if let Some(event) = decode_ocsf_event(&log) {
+            log.message = event.format_shorthand();
+        }
         let evt = SandboxStreamEvent {
             payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log(
                 log.clone(),
@@ -163,6 +167,26 @@ impl TracingLogBus {
     }
 }
 
+/// Decode the structured OCSF event a sandbox line carries, if any.
+///
+/// A malformed payload is logged and ignored rather than dropping the line.
+fn decode_ocsf_event(log: &SandboxLogLine) -> Option<openshell_ocsf::OcsfEvent> {
+    if log.ocsf_json.is_empty() {
+        return None;
+    }
+    match serde_json::from_slice(&log.ocsf_json) {
+        Ok(event) => Some(event),
+        Err(error) => {
+            tracing::warn!(
+                sandbox_id = %log.sandbox_id,
+                %error,
+                "discarding undecodable OCSF payload from sandbox log line"
+            );
+            None
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 struct SandboxLogLayer {
     bus: TracingLogBus,
@@ -208,6 +232,7 @@ where
             message: msg,
             source: "gateway".to_string(),
             fields: HashMap::new(),
+            ocsf_json: Vec::new(),
         };
         let evt = SandboxStreamEvent {
             payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Log(
@@ -274,6 +299,7 @@ mod tests {
             message: message.to_string(),
             source: "gateway".to_string(),
             fields: HashMap::new(),
+            ocsf_json: Vec::new(),
         }
     }
 
@@ -339,6 +365,56 @@ mod tests {
         bus.remove("nonexistent");
     }
 
+    #[test]
+    fn external_lines_render_shorthand_from_the_structured_event() {
+        use openshell_ocsf::{
+            ActionId, ActivityId, DispositionId, Endpoint, NetworkActivityBuilder, SeverityId,
+            StatusId,
+        };
+
+        let event = NetworkActivityBuilder::new(&ocsf_ctx("sb-wire"))
+            .activity(ActivityId::Open)
+            .action(ActionId::Denied)
+            .disposition(DispositionId::Blocked)
+            .severity(SeverityId::Medium)
+            .status(StatusId::Failure)
+            .dst_endpoint(Endpoint::from_domain("blocked.example.com", 443))
+            .message("CONNECT denied blocked.example.com:443")
+            .build();
+        let expected = event.format_shorthand();
+
+        let bus = TracingLogBus::new();
+        let mut line = make_log_event("sb-wire", "");
+        line.ocsf_json = event.to_json_line().unwrap().into_bytes();
+        bus.publish_external(line);
+
+        let tail = bus.tail("sb-wire", 10);
+        assert_eq!(tail.len(), 1);
+        assert_eq!(log_message(&tail[0]).message, expected);
+    }
+
+    #[test]
+    fn external_lines_without_ocsf_json_keep_their_message() {
+        let bus = TracingLogBus::new();
+        bus.publish_external(make_log_event("sb-plain", "already rendered"));
+
+        let tail = bus.tail("sb-plain", 10);
+        assert_eq!(log_message(&tail[0]).message, "already rendered");
+    }
+
+    #[test]
+    fn undecodable_ocsf_json_does_not_drop_the_line() {
+        let bus = TracingLogBus::new();
+        let mut line = make_log_event("sb-bad", "fallback text");
+        line.ocsf_json = b"{not valid json".to_vec();
+        bus.publish_external(line);
+
+        // A malformed payload must not silently swallow the record.
+        let tail = bus.tail("sb-bad", 10);
+        assert_eq!(tail.len(), 1);
+        assert_eq!(log_message(&tail[0]).message, "fallback text");
+    }
+
     #[test]
     fn publish_after_remove_does_not_resurrect_the_bus_entry() {
         let bus = TracingLogBus::new();

crates/openshell-server/tests/ocsf_wire_equivalence.rs

@@ -0,0 +1,135 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Gateway-rendered log text must match what the sandbox would have rendered.
+//!
+//! A decode gap corrupts `openshell logs` output rather than erroring.
+
+use std::net::{IpAddr, Ipv4Addr};
+
+use openshell_ocsf::{
+    ActionId, ActivityId, AppLifecycleBuilder, ConfigStateChangeBuilder, DetectionFindingBuilder,
+    DispositionId, Endpoint, EventOrigin, FindingInfo, HttpActivityBuilder, HttpMethod,
+    HttpRequest, HttpResponse, NetworkActivityBuilder, OcsfEvent, Process, ProcessActivityBuilder,
+    SandboxContext, SeverityId, SshActivityBuilder, StateId, StatusId, Url,
+};
+
+fn ctx() -> SandboxContext {
+    SandboxContext {
+        sandbox_id: "sb-1".to_string(),
+        sandbox_name: "agent-01".to_string(),
+        container_image: "ghcr.io/nvidia/openshell/sandbox:0.42.1".to_string(),
+        hostname: "openshell-sb-1".to_string(),
+        product_version: "0.42.1".to_string(),
+        proxy_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
+        proxy_port: 8888,
+        origin: EventOrigin::Sandbox,
+    }
+}
+
+/// Serialize as the supervisor does, then decode as the gateway does.
+fn across_the_wire(event: &OcsfEvent) -> OcsfEvent {
+    let bytes = event.to_json_line().expect("serialize").into_bytes();
+    serde_json::from_slice(&bytes).expect("gateway should decode the payload")
+}
+
+fn assert_renders_identically(label: &str, event: &OcsfEvent) {
+    assert_eq!(
+        across_the_wire(event).format_shorthand(),
+        event.format_shorthand(),
+        "{label}: gateway-rendered text differs from sandbox-rendered text"
+    );
+}
+
+#[test]
+fn network_activity_renders_identically_across_the_wire() {
+    let event = NetworkActivityBuilder::new(&ctx())
+        .activity(ActivityId::Open)
+        .action(ActionId::Denied)
+        .disposition(DispositionId::Blocked)
+        .severity(SeverityId::Medium)
+        .status(StatusId::Failure)
+        .dst_endpoint(Endpoint::from_domain("api.example.com", 443))
+        .src_endpoint_addr(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 5)), 51234)
+        .actor_process(Process::new("/usr/bin/curl", 4711).with_cmd_line("curl -sS https://x"))
+        .firewall_rule("default-deny-egress", "opa")
+        .message("CONNECT denied api.example.com:443")
+        .build();
+    assert_renders_identically("network_activity", &event);
+}
+
+#[test]
+fn http_activity_renders_identically_across_the_wire() {
+    let event = HttpActivityBuilder::new(&ctx())
+        .activity(ActivityId::Open)
+        .action(ActionId::Allowed)
+        .disposition(DispositionId::Allowed)
+        .severity(SeverityId::Informational)
+        .status(StatusId::Success)
+        .http_request(HttpRequest {
+            http_method: HttpMethod::Get,
+            url: Some(Url::new("https", "api.example.com", "/v1/items", 443)),
+        })
+        .http_response(HttpResponse { code: 200 })
+        .message("GET /v1/items 200")
+        .build();
+    assert_renders_identically("http_activity", &event);
+}
+
+#[test]
+fn ssh_activity_renders_identically_across_the_wire() {
+    let event = SshActivityBuilder::new(&ctx())
+        .activity(ActivityId::Open)
+        .severity(SeverityId::Informational)
+        .status(StatusId::Success)
+        .dst_endpoint(Endpoint::from_domain("sandbox.local", 22))
+        .message("ssh session accepted")
+        .build();
+    assert_renders_identically("ssh_activity", &event);
+}
+
+#[test]
+fn process_activity_renders_identically_across_the_wire() {
+    let event = ProcessActivityBuilder::new(&ctx())
+        .activity(ActivityId::Open)
+        .severity(SeverityId::Informational)
+        .status(StatusId::Success)
+        .process(Process::new("/usr/bin/python3", 4713).with_cmd_line("python3 -m pytest"))
+        .message("process started")
+        .build();
+    assert_renders_identically("process_activity", &event);
+}
+
+#[test]
+fn detection_finding_renders_identically_across_the_wire() {
+    let event = DetectionFindingBuilder::new(&ctx())
+        .finding_info(FindingInfo::new("finding-1", "Sandbox bypass attempt"))
+        .severity(SeverityId::High)
+        .is_alert(true)
+        .evidence("dst_host", "169.254.169.254")
+        .message("bypass attempt detected")
+        .build();
+    assert_renders_identically("detection_finding", &event);
+}
+
+#[test]
+fn config_state_change_renders_identically_across_the_wire() {
+    let event = ConfigStateChangeBuilder::new(&ctx())
+        .state(StateId::Other, "policy-loaded")
+        .severity(SeverityId::Informational)
+        .status(StatusId::Success)
+        .message("policy reloaded")
+        .unmapped("policy_version", 7)
+        .build();
+    assert_renders_identically("config_state_change", &event);
+}
+
+#[test]
+fn application_lifecycle_renders_identically_across_the_wire() {
+    let event = AppLifecycleBuilder::new(&ctx())
+        .activity(ActivityId::Open)
+        .severity(SeverityId::Informational)
+        .message("supervisor started")
+        .build();
+    assert_renders_identically("application_lifecycle", &event);
+}

crates/openshell-supervisor-process/src/log_push.rs

@@ -51,22 +51,25 @@ impl<S: Subscriber> Layer<S> for LogPushLayer {
             return;
         }
 
-        // OCSF events carry their payload in a thread-local; extract the
-        // shorthand representation for the push message. Non-OCSF events
-        // use the original visitor-based extraction.
-        let (msg, fields) = if meta.target() == openshell_ocsf::OCSF_TARGET {
-            if let Some(ocsf_event) = openshell_ocsf::clone_current_event() {
-                (
-                    ocsf_event.format_shorthand(),
-                    std::collections::HashMap::new(),
-                )
-            } else {
+        // OCSF events carry their payload in a thread-local. Send the event
+        // itself; the gateway renders the display text from it.
+        let (msg, fields, ocsf_json) = if meta.target() == openshell_ocsf::OCSF_TARGET {
+            let Some(ocsf_event) = openshell_ocsf::clone_current_event() else {
                 return;
-            }
+            };
+            let Ok(json) = ocsf_event.to_json_line() else {
+                return;
+            };
+            (
+                String::new(),
+                std::collections::HashMap::new(),
+                json.into_bytes(),
+            )
         } else {
             let mut visitor = LogVisitor::default();
             event.record(&mut visitor);
-            visitor.into_parts(meta.name())
+            let (msg, fields) = visitor.into_parts(meta.name());
+            (msg, fields, Vec::new())
         };
 
         let ts = openshell_core::time::now_ms();
@@ -85,6 +88,7 @@ impl<S: Subscriber> Layer<S> for LogPushLayer {
             message: msg,
             source: "sandbox".to_string(),
             fields,
+            ocsf_json,
         };
 
         // Best-effort: drop if the channel is full (don't block tracing).
@@ -102,7 +106,6 @@ pub fn spawn_log_push_task(
     sandbox_id: String,
 ) -> (mpsc::Sender<SandboxLogLine>, tokio::task::JoinHandle<()>) {
     let (tx, rx) = mpsc::channel::<SandboxLogLine>(1024);
-
     let handle = tokio::spawn(run_push_loop(endpoint, sandbox_id, rx));
 
     (tx, handle)
@@ -354,7 +357,7 @@ mod tests {
     }
 
     #[test]
-    fn ocsf_events_push_shorthand_with_ocsf_level_and_no_fields() {
+    fn ocsf_events_push_the_structured_event_not_shorthand() {
         let event = NetworkActivityBuilder::new(&ocsf_ctx())
             .activity(ActivityId::Open)
             .action(ActionId::Denied)
@@ -364,17 +367,58 @@ mod tests {
             .dst_endpoint(Endpoint::from_domain("blocked.example.com", 443))
             .message("CONNECT denied blocked.example.com:443".to_string())
             .build();
+        let expected_json = event.to_json().expect("serialize");
         let expected_shorthand = event.format_shorthand();
 
         let lines = capture(16, || ocsf_emit!(event));
+        assert_eq!(lines.len(), 1);
+        let line = &lines[0];
+
+        assert!(
+            !line.ocsf_json.is_empty(),
+            "structured event should be sent"
+        );
+        let decoded: serde_json::Value =
+            serde_json::from_slice(&line.ocsf_json).expect("payload should be valid JSON");
+        assert_eq!(decoded, expected_json);
+
+        // The receiver renders the display text, so it is not sent.
+        assert!(line.message.is_empty());
+
+        // What the receiver will render must match what the sandbox would have.
+        let decoded_event: openshell_ocsf::OcsfEvent =
+            serde_json::from_slice(&line.ocsf_json).expect("payload should decode");
+        assert_eq!(decoded_event.format_shorthand(), expected_shorthand);
+    }
+
+    #[test]
+    fn non_ocsf_lines_carry_no_ocsf_payload() {
+        let lines = capture(16, || {
+            tracing::info!(target: "test_target", "plain line");
+        });
+        assert!(lines[0].ocsf_json.is_empty());
+        assert_eq!(lines[0].message, "plain line");
+    }
+
+    #[test]
+    fn ocsf_events_push_with_ocsf_level_and_no_fields() {
+        let event = NetworkActivityBuilder::new(&ocsf_ctx())
+            .activity(ActivityId::Open)
+            .action(ActionId::Denied)
+            .disposition(DispositionId::Blocked)
+            .severity(SeverityId::Medium)
+            .status(StatusId::Failure)
+            .dst_endpoint(Endpoint::from_domain("blocked.example.com", 443))
+            .message("CONNECT denied blocked.example.com:443".to_string())
+            .build();
+        let lines = capture(16, || ocsf_emit!(event));
 
         assert_eq!(lines.len(), 1);
         let line = &lines[0];
         assert_eq!(line.level, "OCSF");
         assert_eq!(line.target, openshell_ocsf::OCSF_TARGET);
         assert_eq!(line.source, "sandbox");
         assert_eq!(line.sandbox_id, "sb-test");
-        assert_eq!(line.message, expected_shorthand);
         assert!(line.fields.is_empty());
         assert!(line.timestamp_ms > 0);
     }
@@ -454,6 +498,7 @@ mod tests {
             message: message.to_string(),
             source: "sandbox".to_string(),
             fields: std::collections::HashMap::new(),
+            ocsf_json: Vec::new(),
         }
     }