scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-11 22:53:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Fix 8 spec compliance gaps in unified-llm

Browse source 
- OpenAI adapter: include is_error flag on function_call_output items
- Anthropic adapter: extract retry_after from headers in streaming error path
- Error retryability: unknown errors now default to retryable per spec
- stream_object: add ObjectStreamResult wrapper with object() accessor
- TimeoutConfig: add From<f64> for total-only timeout shorthand
- Message::tool_result: accept serde_json::Value to preserve structured content
- GenerateParams: expose repair_tool_call field wired to execute_all_tools_with_repair
- Both generate() and stream() tool loops use repair-aware execution

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-02-20 12:15:00 -04:00
parent 88e71c64ad
commit 8a19b7be2a
6 changed files with 131 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
crates/unified-llm/src/error.rs
View file

@ -85,15 +85,23 @@ pub enum SdkError {
impl SdkError {
#[must_use]
pub const fn retryable(&self) -> bool {
matches!(
self,
Self::Provider {
kind: ProviderErrorKind::RateLimit | ProviderErrorKind::Server,
..
} | Self::RequestTimeout { .. }
| Self::Network { .. }
| Self::Stream { .. }
)
match self {
Self::Provider { kind, .. } => match kind {
ProviderErrorKind::Authentication
| ProviderErrorKind::AccessDenied
| ProviderErrorKind::NotFound
| ProviderErrorKind::InvalidRequest
| ProviderErrorKind::ContextLength
| ProviderErrorKind::QuotaExceeded
| ProviderErrorKind::ContentFilter => false,
_ => true,
},
Self::InvalidToolCall { .. }
| Self::NoObjectGenerated { .. }
| Self::Abort { .. }
| Self::Configuration { .. } => false,
_ => true,
}
}
#[must_use]
@ -279,22 +287,38 @@ mod tests {
}
#[test]
fn non_retryable_errors() {
let kinds = [
ProviderErrorKind::AccessDenied,
ProviderErrorKind::NotFound,
ProviderErrorKind::InvalidRequest,
ProviderErrorKind::ContextLength,
ProviderErrorKind::QuotaExceeded,
ProviderErrorKind::ContentFilter,
];
for kind in &kinds {
let err = SdkError::Provider {
kind: *kind,
detail: Box::new(ProviderErrorDetail::new("error", "openai")),
};
assert!(!err.retryable(), "Expected non-retryable: {err}");
}
fn non_retryable_provider_errors() {
let detail = || Box::new(ProviderErrorDetail::new("error", "openai"));
let access_denied = SdkError::Provider { kind: ProviderErrorKind::AccessDenied, detail: detail() };
assert!(!access_denied.retryable());
let not_found = SdkError::Provider { kind: ProviderErrorKind::NotFound, detail: detail() };
assert!(!not_found.retryable());
let invalid_req = SdkError::Provider { kind: ProviderErrorKind::InvalidRequest, detail: detail() };
assert!(!invalid_req.retryable());
let ctx_length = SdkError::Provider { kind: ProviderErrorKind::ContextLength, detail: detail() };
assert!(!ctx_length.retryable());
let quota = SdkError::Provider { kind: ProviderErrorKind::QuotaExceeded, detail: detail() };
assert!(!quota.retryable());
let content_filter = SdkError::Provider { kind: ProviderErrorKind::ContentFilter, detail: detail() };
assert!(!content_filter.retryable());
}
#[test]
fn non_retryable_sdk_errors() {
let invalid_tool = SdkError::InvalidToolCall { message: "bad tool".into() };
assert!(!invalid_tool.retryable());
let no_object = SdkError::NoObjectGenerated { message: "no output".into() };
assert!(!no_object.retryable());
let abort = SdkError::Abort { message: "aborted".into() };
assert!(!abort.retryable());
}
#[test]

66
crates/unified-llm/src/generate.rs
View file

@ -2,7 +2,7 @@ use crate::client::Client;
use crate::error::SdkError;
use crate::provider::StreamEventStream;
use crate::retry::retry;
use crate::tools::{execute_all_tools, Tool};
use crate::tools::{execute_all_tools_with_repair, RepairToolCallFn, Tool};
use crate::types::{
FinishReason, GenerateResult, Message, ObjectStreamEvent, Request, Response, ResponseFormat,
ResponseFormatType, RetryPolicy, StepResult, StreamEvent, TimeoutConfig, ToolCall, ToolChoice,
@ -172,7 +172,7 @@ pub async fn generate(params: GenerateParams) -> Result<GenerateResult, SdkError
if tools.iter().any(|t| t.is_active()) {
let tool_refs: Vec<&Tool> =
tools.iter().map(std::convert::AsRef::as_ref).collect();
tool_results = execute_all_tools(&tool_refs, &tool_calls, &messages, abort_signal.as_ref()).await;
tool_results = execute_all_tools_with_repair(&tool_refs, &tool_calls, &messages, abort_signal.as_ref(), params.repair_tool_call.as_ref()).await;
}
}
@ -207,7 +207,7 @@ pub async fn generate(params: GenerateParams) -> Result<GenerateResult, SdkError
for result in &last.tool_results {
messages.push(Message::tool_result(
&result.tool_call_id,
result.content.to_string(),
result.content.clone(),
result.is_error,
));
}
@ -259,6 +259,8 @@ pub struct GenerateParams {
pub abort_signal: Option<CancellationToken>,
/// Custom stop condition checked after each tool round (Section 4.3).
pub stop_when: Option<StopCondition>,
/// Callback to repair invalid tool call arguments (Section 5.8).
pub repair_tool_call: Option<RepairToolCallFn>,
}
impl GenerateParams {
@ -285,6 +287,7 @@ impl GenerateParams {
client: None,
abort_signal: None,
stop_when: None,
repair_tool_call: None,
}
}
@ -414,6 +417,12 @@ impl GenerateParams {
self.stop_when = Some(Arc::new(f));
self
}
#[must_use]
pub fn repair_tool_call(mut self, repair: RepairToolCallFn) -> Self {
self.repair_tool_call = Some(repair);
self
}
}
/// `StreamAccumulator` collects stream events into a complete Response (Section 4.4).
@ -586,6 +595,7 @@ async fn stream_with_tool_loop(params: GenerateParams) -> Result<StreamEventStre
.map(|tools| tools.iter().map(|t| t.definition.clone()).collect());
let abort_signal = params.abort_signal.clone();
let max_tool_rounds = params.max_tool_rounds;
let repair_tool_call = params.repair_tool_call.clone();
let has_active_tools = max_tool_rounds > 0
&& params
@ -703,7 +713,7 @@ async fn stream_with_tool_loop(params: GenerateParams) -> Result<StreamEventStre
let tool_refs: Vec<&Tool> =
tool_list.iter().map(std::convert::AsRef::as_ref).collect();
let tool_results = execute_all_tools(&tool_refs, &tool_calls, &messages, abort_signal.as_ref()).await;
let tool_results = execute_all_tools_with_repair(&tool_refs, &tool_calls, &messages, abort_signal.as_ref(), repair_tool_call.as_ref()).await;
if tool_results.is_empty() {
return;
@ -746,7 +756,7 @@ async fn stream_with_tool_loop(params: GenerateParams) -> Result<StreamEventStre
for result in &tool_results {
messages.push(Message::tool_result(
&result.tool_call_id,
result.content.to_string(),
result.content.clone(),
result.is_error,
));
}
@ -862,6 +872,48 @@ pub async fn generate_object(
pub type ObjectStream =
Pin<Box<dyn futures::Stream<Item = Result<ObjectStreamEvent, SdkError>> + Send>>;
/// Wraps an `ObjectStream` with an `object()` accessor for the final parsed value.
///
/// Implements `Stream<Item = Result<ObjectStreamEvent, SdkError>>` so it can be used
/// as a drop-in replacement for `ObjectStream`. Tracks the last `Complete` event's
/// object internally so callers can retrieve it after the stream ends.
pub struct ObjectStreamResult {
inner: ObjectStream,
object: Option<serde_json::Value>,
}
impl ObjectStreamResult {
fn new(inner: ObjectStream) -> Self {
Self {
inner,
object: None,
}
}
/// Returns the final parsed object after the stream has yielded a `Complete` event.
#[must_use]
pub fn object(&self) -> Option<&serde_json::Value> {
self.object.as_ref()
}
}
impl Stream for ObjectStreamResult {
type Item = Result<ObjectStreamEvent, SdkError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let inner = self.inner.as_mut();
match inner.poll_next(cx) {
Poll::Ready(Some(Ok(event))) => {
if let ObjectStreamEvent::Complete { ref object, .. } = event {
self.object = Some(object.clone());
}
Poll::Ready(Some(Ok(event)))
}
other => other,
}
}
}
/// Streaming structured output with incremental JSON parsing (Section 4.6).
///
/// Combines streaming with structured output: sets `response_format` to `json_schema`,
@ -878,7 +930,7 @@ pub type ObjectStream =
pub async fn stream_object(
params: GenerateParams,
schema: serde_json::Value,
) -> Result<ObjectStream, SdkError> {
) -> Result<ObjectStreamResult, SdkError> {
let params = GenerateParams {
response_format: Some(ResponseFormat {
kind: ResponseFormatType::JsonSchema,
@ -943,7 +995,7 @@ pub async fn stream_object(
},
);
Ok(Box::pin(mapped.flatten()))
Ok(ObjectStreamResult::new(Box::pin(mapped.flatten())))
}
#[cfg(test)]

6
crates/unified-llm/src/providers/anthropic.rs
View file

@ -3,7 +3,8 @@ use base64::{engine::general_purpose::STANDARD as BASE64_STANDARD, Engine};
use crate::error::SdkError;
use crate::provider::{ProviderAdapter, StreamEventStream};
use crate::providers::common::{
extract_system_prompt, parse_error_body, parse_rate_limit_headers, send_and_read_response,
extract_system_prompt, parse_error_body, parse_rate_limit_headers, parse_retry_after,
send_and_read_response,
};
use crate::types::{
ContentPart, FinishReason, Message, Request, Response, ResponseFormatType, Role, StreamEvent,
@ -1148,6 +1149,7 @@ impl ProviderAdapter for Adapter {
let status = http_resp.status();
if !status.is_success() {
let retry_after = parse_retry_after(http_resp.headers());
let body = http_resp.text().await.map_err(|e| SdkError::Network {
message: e.to_string(),
})?;
@ -1158,7 +1160,7 @@ impl ProviderAdapter for Adapter {
"anthropic".to_string(),
code,
raw,
None,
retry_after,
));
}

8
crates/unified-llm/src/providers/openai.rs
View file

@ -253,11 +253,15 @@ fn translate_input(messages: &[Message]) -> (Option<String>, Vec<serde_json::Val
.content
.as_str()
.map_or_else(|| tr.content.to_string(), str::to_string);
input.push(serde_json::json!({
let mut item = serde_json::json!({
"type": "function_call_output",
"call_id": tr.tool_call_id,
"output": output,
}));
});
if tr.is_error {
item["is_error"] = serde_json::json!(true);
}
input.push(item);
}
}
}

2
crates/unified-llm/src/providers/openai_compatible.rs
View file

@ -1141,7 +1141,7 @@ mod tests {
#[test]
fn translate_tool_message_has_tool_call_id() {
let msg = Message::tool_result("call_1", "72F and sunny", false);
let msg = Message::tool_result("call_1", serde_json::Value::String("72F and sunny".into()), false);
let translated = translate_messages(&[msg]);
assert_eq!(translated[0].role, "tool");
assert_eq!(translated[0].tool_call_id.as_deref(), Some("call_1"));

15
crates/unified-llm/src/types.rs
View file

@ -235,7 +235,7 @@ impl Message {
pub fn tool_result(
tool_call_id: impl Into<String>,
content: impl Into<String>,
content: serde_json::Value,
is_error: bool,
) -> Self {
let id = tool_call_id.into();
@ -243,7 +243,7 @@ impl Message {
role: Role::Tool,
content: vec![ContentPart::ToolResult(ToolResult {
tool_call_id: id.clone(),
content: serde_json::Value::String(content.into()),
content,
is_error,
image_data: None,
image_media_type: None,
@ -603,6 +603,15 @@ pub struct TimeoutConfig {
pub per_step: Option<f64>,
}
impl From<f64> for TimeoutConfig {
fn from(total: f64) -> Self {
Self {
total: Some(total),
per_step: None,
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AdapterTimeout {
pub connect: f64,
@ -797,7 +806,7 @@ mod tests {
#[test]
fn message_tool_result_constructor() {
let msg = Message::tool_result("call_123", "72F and sunny", false);
let msg = Message::tool_result("call_123", serde_json::Value::String("72F and sunny".into()), false);
assert_eq!(msg.role, Role::Tool);
assert_eq!(msg.tool_call_id, Some("call_123".to_string()));
match &msg.content[0] {