Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 18 additions & 3 deletions ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -493,6 +493,11 @@ classification and reshapes those raw calls accordingly:
synthetic `function_call` events named `tool_search` are projected into that same
public lifecycle after validation.

Shell functions are restored to `shell_call` for client execution by default.
When an application explicitly registers a shell executor, the translator instead
suppresses those internal function frames using the registry's resolved ownership;
the existing gateway event plan emits the shell call's added/done lifecycle.

It also buffers function-call events that arrive before the call's name is known
(bounded at 256 KiB) and replays them once the name resolves.

Expand Down Expand Up @@ -740,13 +745,23 @@ the behavioral layer — routing, handler traits, normalization, and execution.
reused across requests, specifically for gateway tools that need **lazy, per-request
connection setup**: MCP servers (connects and caches `McpClient`s keyed by server
URL, falling back to connecting a fresh request-declared server) and the shared
`WebSearchHandler`. As of today it only has slots for `ToolType::Mcp` and
`ToolType::WebSearch`; `GatewayExecutorRegistration` has typed variants for those
supported slots. Client-owned
`WebSearchHandler`. It also has an optional, application-provided `ShellExecutor`
slot. `GatewayExecutorRegistration::Shell` is an explicit execution grant; an
unregistered shell declaration remains client-executed. `ShellExecutor` accepts
a typed call with bounded action limits and cancellation and returns typed command
outputs. The adapter binds into the existing gateway scheduler, not a second tool loop.
Client-owned
tools (`function`, `custom`, `namespace`) never touch this file; their registry
entries are inserted with `ToolOwnership::Client` and no `GatewayExecutors`
involvement.

Shell item history is preserved publicly in storage. At the inference boundary,
`ShellHandler::model_input` lowers shell calls and outputs into matching function
history, just as declarations and explicit shell selectors are normalized. For an
opt-in gateway executor, storage additionally retains the canonical internal function
call/output pair; rehydration omits that pair's public shell-call projection to avoid
replaying the invocation twice. Client-executed shell history is not omitted.

**To add a new tool type:**
1. Implement `ToolHandler`, including its typed `ToolParams`, for it. If it's client-executed,
stop there — see `function.rs`/`custom.rs` for the pattern.
Expand Down
2 changes: 2 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions crates/agentic-server-core/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ serde_json.workspace = true
sse-stream.workspace = true
thiserror.workspace = true
tokio = { workspace = true, features = ["time"] }
tokio-util = { workspace = true, features = ["rt"] }
tracing.workspace = true
url.workspace = true

Expand Down
35 changes: 34 additions & 1 deletion crates/agentic-server-core/src/events/normalize.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
use serde_json::Value;

use super::types::{EventFrame, EventPayload, SSEEventType, SSEItemType, WireEvent};
use super::types::{EventFrame, EventPayload, SSEEventType, SSEItemType, ShellCommandUpdate, WireEvent};
use crate::types::io::OutputItem;
use crate::utils::common::{deserialize_from_str_opt, deserialize_from_value_opt};

/// Normalize a raw SSE data line into a typed [`EventFrame`].
Expand Down Expand Up @@ -64,6 +65,9 @@ fn extract_payload(event_type: SSEEventType, json: &Value) -> EventPayload {
SSEEventType::FunctionCallArgumentsDone => extract_fn_call_args_done(json),
SSEEventType::CustomToolCallInputDelta => extract_custom_tool_call_input_delta(json),
SSEEventType::CustomToolCallInputDone => extract_custom_tool_call_input_done(json),
SSEEventType::ShellCallCommandAdded => extract_shell_call_command_added(json),
SSEEventType::ShellCallCommandDelta => extract_shell_call_command_delta(json),
SSEEventType::ShellCallCommandDone => extract_shell_call_command_done(json),

SSEEventType::ReasoningTextDelta => extract_reasoning_text_delta(json),
SSEEventType::ReasoningTextDone => extract_reasoning_text_done(json),
Expand Down Expand Up @@ -133,6 +137,14 @@ fn extract_output_item_added(json: &Value) -> EventPayload {
name: json_str_opt(item, "name"),
namespace: json_str_opt(item, "namespace"),
call_id: json_str_opt(item, "call_id"),
shell_call: if item["type"] == "shell_call" {
match deserialize_from_value_opt::<OutputItem>(item.clone()) {
Some(OutputItem::ShellCall(call)) => Some(Box::new(call)),
_ => None,
}
} else {
None
},
}
}

Expand Down Expand Up @@ -169,6 +181,27 @@ fn extract_text_done(json: &Value) -> EventPayload {
}
}

fn extract_shell_call_command_added(json: &Value) -> EventPayload {
extract_shell_command(json, ShellCommandUpdate::Added(json_str(json, "command")))
}

fn extract_shell_call_command_delta(json: &Value) -> EventPayload {
extract_shell_command(json, ShellCommandUpdate::Delta(json_str(json, "delta")))
}

fn extract_shell_call_command_done(json: &Value) -> EventPayload {
extract_shell_command(json, ShellCommandUpdate::Done(json_str(json, "command")))
}

fn extract_shell_command(json: &Value, update: ShellCommandUpdate) -> EventPayload {
EventPayload::ShellCallCommand {
item_id: json_str(json, "item_id"),
output_index: json_u32(json, "output_index"),
command_index: json_u32(json, "command_index"),
update,
}
}

fn extract_fn_call_args_delta(json: &Value) -> EventPayload {
EventPayload::FunctionCallArgsDelta {
delta: json_str(json, "delta"),
Expand Down
34 changes: 33 additions & 1 deletion crates/agentic-server-core/src/events/types.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};

use crate::types::io::{OutputItem, ResponseUsage};
use crate::types::io::{OutputItem, ResponseUsage, ShellCall};

/// The type of an output item received during streaming.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
Expand All @@ -13,6 +13,7 @@ pub enum SSEItemType {
WebSearchCall,
McpCall,
McpListTools,
ShellCall,
Compaction,
Message,
}
Expand All @@ -28,6 +29,7 @@ impl SSEItemType {
Self::WebSearchCall => "web_search_call",
Self::McpCall => "mcp_call",
Self::McpListTools => "mcp_list_tools",
Self::ShellCall => "shell_call",
Self::Compaction => "compaction",
Self::Message => "message",
}
Expand All @@ -52,6 +54,7 @@ impl std::str::FromStr for SSEItemType {
"web_search_call" => Ok(Self::WebSearchCall),
"mcp_call" => Ok(Self::McpCall),
"mcp_list_tools" => Ok(Self::McpListTools),
"shell_call" => Ok(Self::ShellCall),
"compaction" => Ok(Self::Compaction),
"message" => Ok(Self::Message),
_ => Err(()),
Expand All @@ -71,6 +74,7 @@ impl TryFrom<&OutputItem> for SSEItemType {
OutputItem::WebSearchCall(_) => Ok(Self::WebSearchCall),
OutputItem::McpCall(_) => Ok(Self::McpCall),
OutputItem::McpListTools(_) => Ok(Self::McpListTools),
OutputItem::ShellCall(_) => Ok(Self::ShellCall),
OutputItem::Reasoning(_) => Ok(Self::Reasoning),
OutputItem::Compaction(_) => Ok(Self::Compaction),
OutputItem::Unknown => Err(()),
Expand Down Expand Up @@ -125,6 +129,9 @@ pub enum SSEEventType {
FunctionCallArgumentsDone,
CustomToolCallInputDelta,
CustomToolCallInputDone,
ShellCallCommandAdded,
ShellCallCommandDelta,
ShellCallCommandDone,

// Reasoning
ReasoningTextDelta,
Expand Down Expand Up @@ -171,6 +178,9 @@ impl From<&str> for SSEEventType {
"response.function_call_arguments.done" => Self::FunctionCallArgumentsDone,
"response.custom_tool_call_input.delta" => Self::CustomToolCallInputDelta,
"response.custom_tool_call_input.done" => Self::CustomToolCallInputDone,
"response.shell_call_command.added" => Self::ShellCallCommandAdded,
"response.shell_call_command.delta" => Self::ShellCallCommandDelta,
"response.shell_call_command.done" => Self::ShellCallCommandDone,
"response.reasoning_text.delta" => Self::ReasoningTextDelta,
"response.reasoning_text.done" => Self::ReasoningTextDone,
"response.reasoning_part.added" => Self::ReasoningPartAdded,
Expand Down Expand Up @@ -215,6 +225,9 @@ impl TryFrom<SSEEventType> for &'static str {
SSEEventType::FunctionCallArgumentsDone => Ok("response.function_call_arguments.done"),
SSEEventType::CustomToolCallInputDelta => Ok("response.custom_tool_call_input.delta"),
SSEEventType::CustomToolCallInputDone => Ok("response.custom_tool_call_input.done"),
SSEEventType::ShellCallCommandAdded => Ok("response.shell_call_command.added"),
SSEEventType::ShellCallCommandDelta => Ok("response.shell_call_command.delta"),
SSEEventType::ShellCallCommandDone => Ok("response.shell_call_command.done"),
SSEEventType::ReasoningTextDelta => Ok("response.reasoning_text.delta"),
SSEEventType::ReasoningTextDone => Ok("response.reasoning_text.done"),
SSEEventType::ReasoningPartAdded => Ok("response.reasoning_part.added"),
Expand Down Expand Up @@ -276,6 +289,14 @@ impl WireEvent {
}
}

/// One command's incremental lifecycle within a shell output item.
#[derive(Debug, Clone)]
pub enum ShellCommandUpdate {
Added(String),
Delta(String),
Done(String),
}

/// Typed payload extracted from an SSE event's JSON data.
#[derive(Debug, Clone)]
#[non_exhaustive]
Expand All @@ -296,6 +317,8 @@ pub enum EventPayload {
name: Option<String>,
namespace: Option<String>,
call_id: Option<String>,
/// Preserve the typed initial shell item before command events arrive.
shell_call: Option<Box<ShellCall>>,
},

/// `response.output_item.done`
Expand Down Expand Up @@ -339,6 +362,12 @@ pub enum EventPayload {
},

/// `response.custom_tool_call_input.delta`
ShellCallCommand {
item_id: String,
output_index: u32,
command_index: u32,
update: ShellCommandUpdate,
},
CustomToolCallInputDelta {
delta: String,
item_id: String,
Expand Down Expand Up @@ -468,6 +497,9 @@ mod tests {
SSEEventType::FunctionCallArgumentsDone,
SSEEventType::CustomToolCallInputDelta,
SSEEventType::CustomToolCallInputDone,
SSEEventType::ShellCallCommandAdded,
SSEEventType::ShellCallCommandDelta,
SSEEventType::ShellCallCommandDone,
SSEEventType::ReasoningTextDelta,
SSEEventType::ReasoningTextDone,
SSEEventType::ReasoningPartAdded,
Expand Down
70 changes: 69 additions & 1 deletion crates/agentic-server-core/src/events/validate.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ pub(crate) fn validate_frame(frame: &EventFrame) -> Result<ValidatedFrame<'_>, E
SSEEventType::Other => Ok(ValidatedFrame { item: None }),
event_type => {
let output_index = required_output_index(frame, event_name)?;
let item_id = required_str(&frame.wire.rest, "item_id", event_name)?;
let item_id = validate_event_item_id(frame, event_name)?;
validate_event_fields(&frame.wire.rest, event_type, event_name)?;
let item_type = expected_item_type(event_type).ok_or_else(|| {
invalid(format!(
Expand Down Expand Up @@ -80,6 +80,9 @@ fn expected_item_type(event_type: SSEEventType) -> Option<SSEItemType> {
SSEEventType::CustomToolCallInputDelta | SSEEventType::CustomToolCallInputDone => {
Some(SSEItemType::CustomToolCall)
}
SSEEventType::ShellCallCommandAdded
| SSEEventType::ShellCallCommandDelta
| SSEEventType::ShellCallCommandDone => Some(SSEItemType::ShellCall),
SSEEventType::ReasoningTextDelta
| SSEEventType::ReasoningTextDone
| SSEEventType::ReasoningPartAdded
Expand Down Expand Up @@ -192,12 +195,30 @@ fn validate_output_item<'a>(
})
}

fn validate_event_item_id<'a>(frame: &'a EventFrame, event_name: &str) -> Result<&'a str, EventError> {
if matches!(
frame.event_type,
SSEEventType::ShellCallCommandAdded | SSEEventType::ShellCallCommandDelta | SSEEventType::ShellCallCommandDone
) && !frame.wire.rest.contains_key("item_id")
{
// Only native shell command events may omit the ID and resolve by output_index.
return Ok("");
}
let item_id = required_str(&frame.wire.rest, "item_id", event_name)?;
Ok(item_id)
}

fn validate_event_fields(
event: &Map<String, Value>,
event_type: SSEEventType,
event_name: &str,
) -> Result<(), EventError> {
match event_type {
SSEEventType::ShellCallCommandAdded
| SSEEventType::ShellCallCommandDelta
| SSEEventType::ShellCallCommandDone => {
required_u32(event, "command_index", event_name)?;
}
SSEEventType::OutputTextDelta
| SSEEventType::OutputTextDone
| SSEEventType::ContentPartAdded
Expand All @@ -218,6 +239,7 @@ fn validate_event_fields(
SSEEventType::OutputTextDelta
| SSEEventType::FunctionCallArgumentsDelta
| SSEEventType::CustomToolCallInputDelta
| SSEEventType::ShellCallCommandDelta
| SSEEventType::ReasoningTextDelta
| SSEEventType::ReasoningSummaryTextDelta
| SSEEventType::McpCallArgumentsDelta => Some("delta"),
Expand All @@ -226,6 +248,7 @@ fn validate_event_fields(
}
SSEEventType::FunctionCallArgumentsDone | SSEEventType::McpCallArgumentsDone => Some("arguments"),
SSEEventType::CustomToolCallInputDone => Some("input"),
SSEEventType::ShellCallCommandAdded | SSEEventType::ShellCallCommandDone => Some("command"),
SSEEventType::ContentPartAdded
| SSEEventType::ContentPartDone
| SSEEventType::ReasoningPartAdded
Expand Down Expand Up @@ -308,3 +331,48 @@ fn missing_field(owner: &str, field: &str) -> EventError {
fn invalid(message: impl Into<String>) -> EventError {
EventError(message.into())
}

#[cfg(test)]
mod tests {
use serde_json::json;

use super::validate_frame;
use crate::events::normalize_sse_line;

#[test]
fn only_native_shell_command_events_allow_omitted_item_id() {
for (event_type, allows_omitted_id) in [
("response.shell_call_command.added", true),
("response.shell_call_command.delta", true),
("response.shell_call_command.done", true),
("response.function_call_arguments.delta", false),
("response.function_call_arguments.done", false),
("response.custom_tool_call_input.delta", false),
("response.custom_tool_call_input.done", false),
("response.mcp_call_arguments.delta", false),
] {
let event = json!({
"type": event_type,
"output_index": 0,
"command_index": 0,
"command": "pwd",
"delta": "",
"arguments": "{}",
"input": "pwd"
});
let frame = normalize_sse_line(&format!("data: {event}")).unwrap();
assert_eq!(validate_frame(&frame).is_ok(), allows_omitted_id, "{event_type}");

for item_id in [json!(""), json!(null), json!(42), json!("sh_1")] {
let mut event = event.clone();
event["item_id"] = item_id.clone();
let frame = normalize_sse_line(&format!("data: {event}")).unwrap();
assert_eq!(
validate_frame(&frame).is_ok(),
item_id == json!("sh_1"),
"{event_type} with item_id={item_id}"
);
}
}
}
}
Loading