Skip to content
Merged
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
4 changes: 0 additions & 4 deletions .rust-file-sizes.json
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,6 @@
"crates/agentic-server/src/main.rs": {
"limit": 644,
"reason": "Server configuration resolves the web_fetch settings with the same file and environment precedence as the other tool settings."
},
"crates/agentic-server-core/src/storage/models/item.rs": {
"limit": 510,
"reason": "The item CRUD now owns response-history row identity, as requested in review; consolidating the insert here adds the row/public-ID bookkeeping. Splitting the item storage module is a separate change."
}
},
"generated": {}
Expand Down
3 changes: 3 additions & 0 deletions ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -596,6 +596,9 @@ access happen — those live in `tool/`, `executor/`, and `storage/` respectivel
`InOutItem::into_input_items` uses the existing output-to-input conversion and
client-tool lowering for model history. Storage row markers and legacy decoding
remain in `storage/`; there is no separate conversation-item enum.
Item retrieval and listing return `InOutItem` directly. `into_public` assigns the
storage identity and expands shorthand messages on the returned item;
serialization uses the item types themselves, without an item-response wrapper.
`ResponsesInput::input_items` uses `InOutItem::as_input` to borrow submitted input
items and project outputs. The existing compaction-window selection accepts these
borrowed items, so estimation avoids copying message and image payloads and model
Expand Down
2 changes: 2 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,8 @@ flowchart LR
| --- | --- | --- |
| `POST /v1/responses` | OpenAI-compatible Responses API with state, tools, and streaming | ✅ |
| `GET /v1/responses/{response_id}` | Retrieve a locally stored response | ✅ |
| `DELETE /v1/responses/{response_id}` | Delete a locally stored response | ✅ |
| `GET /v1/responses/{response_id}/input_items` | List the input items of a stored response | ✅ |
| `GET /v1/responses` | WebSocket transport for the Responses API | ✅ |
| `POST /v1/responses/compact` | Compact direct input or a stored response chain | ✅ |
| `/v1/conversations` · `/v1/conversations/{id}/items` | Conversations API: create, retrieve, update, and delete conversations and their items | ✅ |
Expand Down
6 changes: 3 additions & 3 deletions crates/agentic-llm-d/tests/split_execution_integration.rs
Original file line number Diff line number Diff line change
Expand Up @@ -609,7 +609,7 @@ async fn a_turn_that_cannot_be_stored_is_refused() {
let ctx = exec_ctx().await;

let unknown = hydrate(request("hi", Some("resp_missing")), &ctx).await;
assert_eq!(status_of(&unknown.expect_err("unknown id")), 404);
assert_eq!(status_of(&unknown.expect_err("unknown id")), 400);

// A caller-supplied body gets none of the in-process parser's defaults,
// including the catch-all used for compatibility on trusted internal paths.
Expand Down Expand Up @@ -746,7 +746,7 @@ async fn a_turn_that_is_not_written_still_returns() {
[agentic_core::types::io::OutputItem::FunctionCall(call)] if call.call_id.is_empty()
));
let orphan = hydrate(request("and then?", Some(&id)), &ctx).await;
assert_eq!(status_of(&orphan.expect_err("never stored")), 404);
assert_eq!(status_of(&orphan.expect_err("never stored")), 400);

// A failed response remains returned-but-unstored even when its partial
// item representations disagree; no unusable continuation can be created.
Expand All @@ -762,7 +762,7 @@ async fn a_turn_that_is_not_written_still_returns() {
.expect("failed partial output is not a persistence error");
assert_eq!(payload.status, "error");
let orphan = hydrate(request("and then?", Some(&failed_id)), &ctx).await;
assert_eq!(status_of(&orphan.expect_err("failed SSE was never stored")), 404);
assert_eq!(status_of(&orphan.expect_err("failed SSE was never stored")), 400);
}

/// Until an identical retry can be proven identical, a reused id is refused.
Expand Down
8 changes: 8 additions & 0 deletions crates/agentic-server-core/src/executor/compaction.rs
Original file line number Diff line number Diff line change
Expand Up @@ -455,13 +455,17 @@ mod tests {

fn function_image_output(encoded_bytes: usize) -> InputItem {
InputItem::FunctionCallOutput(FunctionToolResultMessage {
id: None,
status: None,
call_id: "call_view_image".to_owned(),
output: ToolCallOutput::Content(vec![ToolOutputContent::InputImage(inline_image(encoded_bytes))]),
})
}

fn custom_image_output(encoded_bytes: usize) -> InputItem {
InputItem::CustomToolCallOutput(CustomToolCallOutputMessage {
status: None,
id: None,
call_id: "call_view_image".to_owned(),
name: Some("view_image".to_owned()),
output: ToolCallOutput::Content(vec![ToolOutputContent::InputImage(inline_image(encoded_bytes))]),
Expand Down Expand Up @@ -585,6 +589,8 @@ mod tests {
}),
user_message("first"),
InputItem::FunctionCallOutput(FunctionToolResultMessage {
id: None,
status: None,
call_id: "call_1".to_owned(),
output: "tool output".into(),
}),
Expand Down Expand Up @@ -700,6 +706,8 @@ mod tests {
let input = ResponsesInput::from(vec![
user_message("hello context"),
InputItem::FunctionCallOutput(FunctionToolResultMessage {
id: None,
status: None,
call_id: "call_1".to_owned(),
output: "substantial tool output".into(),
}),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ use crate::executor::{
use crate::types::{
agent::{AgentCompletion, AgentIdentity, AgentTurnKey},
agent_commands::CollaborationResult,
event::MessageStatus,
io::output::OutputTextContent,
io::{
AgentMessage, AgentMessageContent, FunctionToolResultMessage, InputItem, MultiAgentAction,
Expand Down Expand Up @@ -43,6 +44,8 @@ impl MultiAgentRun {
.stored
.history
.push(InputItem::FunctionCallOutput(FunctionToolResultMessage {
id: Some(uuid7_str("fc_")),
status: Some(MessageStatus::Completed),
call_id: call_id.clone(),
output: ToolCallOutput::Text(text.clone()),
}));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -483,6 +483,8 @@ async fn concurrent_children_pause_and_resume_from_durable_tree_json_and_sse() {
.iter()
.map(|call| {
InputItem::FunctionCallOutput(FunctionToolResultMessage {
id: None,
status: None,
call_id: call.call_id.clone(),
output: ToolCallOutput::Text("proposal details".into()),
})
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,8 @@ fn batch(response: &str, ids: &[&str]) -> ClientToolOutputBatch {
.iter()
.map(|id| {
ClientToolOutput::Function(FunctionToolResultMessage {
id: None,
status: None,
call_id: (*id).into(),
output: ToolCallOutput::Text(format!("output for {id}")),
})
Expand Down
39 changes: 26 additions & 13 deletions crates/agentic-server-core/src/executor/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -117,10 +117,13 @@ pub enum ExecutorError {
/// The request's `max_tool_calls` is not an integer of at least one.
#[error(transparent)]
InvalidMaxToolCalls(#[from] MaxToolCallsError),
/// A recognized request value that cannot be used in this combination, reported with
/// `code: unsupported_value` and the offending request parameter.
/// A rejected request parameter value, with its public code and parameter name.
#[error("{message}")]
UnsupportedValue { param: &'static str, message: &'static str },
UnsupportedValue {
param: String,
code: Option<&'static str>,
message: String,
},
/// A recognized request member this gateway cannot honour (`background: true`, `prompt`).
#[error(transparent)]
UnsupportedParameter(#[from] UnsupportedParameterError),
Expand Down Expand Up @@ -240,6 +243,9 @@ impl ExecutorError {
| Self::JsonError(_)
| Self::PayloadTooLarge(_)
| Self::Storage(StorageError::ItemCursorNotFound { .. }) => "invalid_request_error",
Self::Storage(StorageError::NotFound { resource_type, .. }) if resource_type == "Response" => {
"invalid_request_error"
}
Self::Storage(e) if e.is_not_found() => "not_found",
Self::Storage(e) if e.is_validation() => "invalid_request_error",
Self::Conflict(_) => "conflict_error",
Expand Down Expand Up @@ -270,7 +276,7 @@ impl ExecutorError {
Self::ResourceLimitExceeded { .. } => "response_resource_limit_exceeded",
Self::InvalidPromptCacheOptions(PromptCacheOptionsError::Unsupported) => "unsupported_parameter",
Self::InvalidMaxToolCalls(error) => error.code(),
Self::UnsupportedValue { .. } => "unsupported_value",
Self::UnsupportedValue { code, .. } => code.unwrap_or("unsupported_value"),
Self::UnsupportedParameter(error) => error.code(),
Self::Tool(ToolError::UnknownParameter { .. }) => "unknown_parameter",
other => other.error_type(),
Expand All @@ -282,15 +288,15 @@ impl ExecutorError {
pub fn error_param(&self) -> Option<&str> {
match self.client_visible_error() {
Self::Storage(StorageError::InvalidItemId { param, .. })
| Self::Tool(ToolError::UnknownParameter { param }) => Some(param),
| Self::Tool(ToolError::UnknownParameter { param })
| Self::UnsupportedValue { param, .. } => Some(param),
Self::Storage(StorageError::ItemAlreadyInConversation) => Some("items"),
Self::Storage(StorageError::ItemCursorNotFound { .. }) => Some("after"),
Self::ConversationLocked { .. } => Some("conversation"),
Self::PreviousResponseNotFound { .. } => Some("previous_response_id"),
Self::Tool(ToolError::MissingOutput { .. } | ToolError::UnknownCallOutput { .. }) => Some("input"),
Self::InvalidPromptCacheOptions(_) => Some("prompt_cache_options"),
Self::InvalidMaxToolCalls(_) => Some(MAX_TOOL_CALLS_PARAM),
Self::UnsupportedValue { param, .. } => Some(param),
Self::UnsupportedParameter(error) => Some(error.param()),
_ => None,
}
Expand All @@ -317,11 +323,15 @@ impl ExecutorError {

/// Builds the OpenAI-compatible error object shared by HTTP and SSE.
pub(crate) fn response_error(&self) -> serde_json::Value {
let code = if matches!(
self.client_visible_error(),
Self::Tool(ToolError::MissingOutput { .. } | ToolError::UnknownCallOutput { .. })
| Self::Storage(StorageError::ItemCursorNotFound { .. })
) {
let missing_response = matches!(self.client_visible_error(),
Self::Storage(StorageError::NotFound { resource_type, .. }) if resource_type == "Response");
let code = if missing_response
|| matches!(
self.client_visible_error(),
Self::Tool(ToolError::MissingOutput { .. } | ToolError::UnknownCallOutput { .. })
| Self::Storage(StorageError::ItemCursorNotFound { .. })
| Self::UnsupportedValue { code: None, .. }
) {
serde_json::Value::Null
} else {
serde_json::json!(self.error_code())
Expand All @@ -332,6 +342,8 @@ impl ExecutorError {
error.insert("code".to_owned(), code);
if let Some(param) = self.error_param() {
error.insert("param".to_owned(), serde_json::json!(param));
} else if missing_response {
error.insert("param".to_owned(), serde_json::Value::Null);
}
serde_json::Value::Object(error)
}
Expand Down Expand Up @@ -565,8 +577,9 @@ mod tests {
#[test]
fn unsupported_value_matches_openai_error_envelope() {
let error = ExecutorError::UnsupportedValue {
param: "multi_agent",
message: "Async parallel tool calls are not yet supported in multi-agent mode.",
param: "multi_agent".into(),
code: Some("unsupported_value"),
message: "Async parallel tool calls are not yet supported in multi-agent mode.".into(),
};
assert_eq!(
envelope(error),
Expand Down
42 changes: 12 additions & 30 deletions crates/agentic-server-core/src/executor/modes/conversation.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,10 @@
use crate::storage::{
ConversationData, ConversationSnapshot, ConversationStore, ConversationVersion, ResponseMetadata, StorageError,
};
use crate::types::conversations::{ConversationMetadata, ItemResponse, ListItemsResponse};
use crate::types::conversations::ConversationMetadata;
use crate::types::io::InOutItem;
use crate::types::io::OutputItem;
use crate::types::io::{ItemPage, ListItemsResponse};

use super::CommittedResponse;
use crate::executor::error::{ExecutorError, ExecutorResult};
Expand Down Expand Up @@ -194,7 +195,7 @@ impl ConversationHandler {
tenant_id: &str,
conversation_id: &str,
items: Vec<InOutItem>,
) -> ExecutorResult<Vec<ItemResponse>> {
) -> ExecutorResult<Vec<InOutItem>> {
if items.is_empty() || items.len() > 20 {
return Err(ExecutorError::InvalidRequest(
"items must contain between 1 and 20 items".into(),
Expand All @@ -211,7 +212,7 @@ impl ConversationHandler {
let mut item_responses = Vec::with_capacity(created.len());
for db_item in created {
let conversation_item = InOutItem::try_from(&db_item)?;
item_responses.push(ItemResponse::new(db_item.public_id().to_owned(), conversation_item));
item_responses.push(conversation_item.into_public(db_item.public_id().to_owned()));
}

Ok(item_responses)
Expand All @@ -225,41 +226,22 @@ impl ConversationHandler {
&self,
tenant_id: &str,
conversation_id: &str,
limit: i64,
after: Option<&str>,
order: &str,
page: ItemPage<'_>,
) -> ExecutorResult<ListItemsResponse> {
// Validate order parameter
if order != "asc" && order != "desc" {
return Err(ExecutorError::InvalidRequest("order must be 'asc' or 'desc'".into()));
}

let limit = limit.clamp(1, 100);
let order_enum = if order == "desc" {
crate::types::conversations::ItemOrder::Desc
} else {
crate::types::conversations::ItemOrder::Asc
};

// Fetch limit + 1 to determine if there are more items
// Fetch one extra item to determine if there are more items
let mut items = self
.store
.list_items(tenant_id, conversation_id, limit + 1, after, order_enum)
.list_items(tenant_id, conversation_id, page.limit.probe(), page.after, page.order)
.await
.map_err(ExecutorError::Storage)?;

// Safe: limit is clamped to [1, 100], far below usize range
#[allow(clippy::cast_possible_wrap)]
let has_more = (items.len() as i64) > limit;
if has_more {
items.pop(); // Remove the extra item
}
let has_more = items.len() > page.limit.get();
items.truncate(page.limit.get());

// Perform fallible conversion - propagate errors instead of silently dropping items
let mut item_responses = Vec::with_capacity(items.len());
for db_item in items {
let conversation_item = InOutItem::try_from(&db_item)?;
item_responses.push(ItemResponse::new(db_item.public_id().to_owned(), conversation_item));
item_responses.push(conversation_item.into_public(db_item.public_id().to_owned()));
}

Ok(ListItemsResponse::new(item_responses, has_more))
Expand All @@ -274,15 +256,15 @@ impl ConversationHandler {
tenant_id: &str,
conversation_id: &str,
item_id: &str,
) -> ExecutorResult<ItemResponse> {
) -> ExecutorResult<InOutItem> {
let db_item = self
.store
.retrieve_item(tenant_id, conversation_id, item_id)
.await
.map_err(ExecutorError::Storage)?;

let conversation_item = InOutItem::try_from(&db_item)?;
Ok(ItemResponse::new(db_item.public_id().to_owned(), conversation_item))
Ok(conversation_item.into_public(db_item.public_id().to_owned()))
}

/// Deletes an item by ID, returning the updated conversation.
Expand Down
Loading
Loading