zeph_core/
json_event_layer.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! `JsonEventLayer`: a [`crate::runtime_layer::RuntimeLayer`] that emits tool events via [`JsonEventSink`].
5//!
6//! Install this layer on the agent when `--json` is active. It is the *canonical*
7//! emitter for `tool_call` and `tool_result` events — `JsonCliChannel` intentionally
8//! no-ops its corresponding channel methods to avoid double-emission.
9//!
10//! All tool arguments and outputs pass through [`crate::redact::scrub_content`] before
11//! emission so secrets (API keys, bearer tokens, passwords) are not written to the JSONL
12//! stream.
13
14use std::future::Future;
15use std::pin::Pin;
16use std::sync::Arc;
17
18use zeph_tools::ToolError;
19use zeph_tools::executor::{ToolCall, ToolOutput};
20
21use crate::json_event_sink::{JsonEvent, JsonEventSink};
22use crate::runtime_layer::{BeforeToolResult, LayerContext, RuntimeLayer};
23
24/// `RuntimeLayer` that forwards tool events to a [`JsonEventSink`].
25pub struct JsonEventLayer {
26    sink: Arc<JsonEventSink>,
27}
28
29impl JsonEventLayer {
30    /// Create a new layer sharing `sink` with `JsonCliChannel`.
31    #[must_use]
32    pub fn new(sink: Arc<JsonEventSink>) -> Self {
33        Self { sink }
34    }
35}
36
37impl RuntimeLayer for JsonEventLayer {
38    fn before_tool<'a>(
39        &'a self,
40        _ctx: &'a LayerContext<'_>,
41        call: &'a ToolCall,
42    ) -> Pin<Box<dyn Future<Output = BeforeToolResult> + Send + 'a>> {
43        // Serialize args, scrub secrets, then re-parse so the sink receives a clean Value.
44        let raw = serde_json::Value::Object(call.params.clone());
45        let raw_str = raw.to_string();
46        let scrubbed_str = crate::redact::scrub_content(&raw_str);
47        let args_value: serde_json::Value =
48            serde_json::from_str(&scrubbed_str).unwrap_or(serde_json::Value::Null);
49        self.sink.emit(&JsonEvent::ToolCall {
50            tool: call.tool_id.as_ref(),
51            args: &args_value,
52            id: call.tool_call_id.as_str(),
53        });
54        Box::pin(std::future::ready(None))
55    }
56
57    fn after_tool<'a>(
58        &'a self,
59        _ctx: &'a LayerContext<'_>,
60        call: &'a ToolCall,
61        result: &'a Result<Option<ToolOutput>, ToolError>,
62    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
63        let err_str;
64        let scrubbed_err;
65        let scrubbed_out;
66        let (output, is_error) = match result {
67            Ok(Some(out)) => {
68                scrubbed_out = crate::redact::scrub_content(&out.summary);
69                (scrubbed_out.as_ref(), false)
70            }
71            Ok(None) => ("", false),
72            Err(e) => {
73                err_str = e.to_string();
74                scrubbed_err = crate::redact::scrub_content(&err_str);
75                (scrubbed_err.as_ref(), true)
76            }
77        };
78        self.sink.emit(&JsonEvent::ToolResult {
79            tool: call.tool_id.as_ref(),
80            id: call.tool_call_id.as_str(),
81            output,
82            is_error,
83        });
84        Box::pin(std::future::ready(()))
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use std::io::Write;
91    use std::sync::{Arc, Mutex};
92
93    use crate::runtime_layer::LayerContext;
94
95    use super::*;
96
97    /// `Write` impl backed by a shared buffer so the test can read back what `JsonEventSink`
98    /// wrote after it takes ownership of the writer.
99    #[derive(Clone)]
100    struct SharedBuffer(Arc<Mutex<Vec<u8>>>);
101
102    impl Write for SharedBuffer {
103        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
104            self.0.lock().unwrap().write(buf)
105        }
106        fn flush(&mut self) -> std::io::Result<()> {
107            Ok(())
108        }
109    }
110
111    fn make_call(tool_id: &str, tool_call_id: &str) -> ToolCall {
112        ToolCall {
113            tool_id: zeph_common::ToolName::new(tool_id),
114            tool_call_id: tool_call_id.to_owned(),
115            ..Default::default()
116        }
117    }
118
119    fn emitted_ids(buf: &[u8], event_name: &str) -> Vec<String> {
120        String::from_utf8_lossy(buf)
121            .lines()
122            .filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
123            .filter(|v| v["event"] == event_name)
124            .map(|v| v["id"].as_str().unwrap_or_default().to_owned())
125            .collect()
126    }
127
128    /// Regression for #5680: `before_tool`/`after_tool` must emit `tool_call_id` (unique per
129    /// invocation) as the JSON event `id`, not `tool_id` (the tool's name, shared across every
130    /// call to the same tool in a turn). Two calls to the same tool with distinct
131    /// `tool_call_id`s must produce distinct `id` values in the emitted events.
132    #[tokio::test]
133    async fn before_and_after_tool_emit_distinct_ids_for_same_tool() {
134        let buf = Arc::new(Mutex::new(Vec::new()));
135        let sink = Arc::new(JsonEventSink::with_writer(SharedBuffer(buf.clone())));
136        let layer = JsonEventLayer::new(sink);
137        let ctx = LayerContext {
138            conversation_id: None,
139            turn_number: 0,
140        };
141
142        let call_a = make_call("shell", "call-a");
143        let call_b = make_call("shell", "call-b");
144        let ok_result: Result<Option<ToolOutput>, ToolError> = Ok(None);
145
146        layer.before_tool(&ctx, &call_a).await;
147        layer.before_tool(&ctx, &call_b).await;
148        layer.after_tool(&ctx, &call_a, &ok_result).await;
149        layer.after_tool(&ctx, &call_b, &ok_result).await;
150
151        let snapshot = buf.lock().unwrap().clone();
152        let call_ids = emitted_ids(&snapshot, "tool_call");
153        let result_ids = emitted_ids(&snapshot, "tool_result");
154
155        assert_eq!(call_ids, vec!["call-a", "call-b"]);
156        assert_eq!(result_ids, vec!["call-a", "call-b"]);
157    }
158}
zeph_core/json_event_layer.rs

zeph_core/
json_event_layer.rs
