// OpenAI streaming wire-format errors and partial tool calls must never become executable output.
import albedo/openai_api/sse
import albedo/openai_api/stream
import albedo/openai_api/types
import gleam/dynamic/decode
import gleam/json
import gleam/list
import gleam/option.{type Option, None, Some}
pub fn responses_incomplete_never_exposes_executable_calls_test() -> Result(
#(stream.State, List(types.Event), Option(types.Turn)),
types.Error,
) {
let data =
"{\"type\":\"response.incomplete\",\"response\":{\"output\":[{\"type\":\"function_call\",\"call_id\":\"danger\",\"name\":\"delete\",\"arguments\":\"{}\"}],\"incomplete_details\":{\"reason\":\"max_output_tokens\"}}}"
let assert Ok(#(
_,
[],
Some(types.Turn(None, [_], [], None, types.LengthLimit, None)),
)) = send(stream.new(types.Responses), "response.incomplete", data)
}
pub fn responses_provider_failures_and_malformed_fields_are_typed_test() -> Nil {
let failed =
"{\"type\":\"response.failed\",\"response\":{\"error\":{\"message\":\"capacity\"}}}"
assert send(stream.new(types.Responses), "response.failed", failed)
== Error(types.ProviderError("capacity"))
let malformed = "{\"type\":\"response.created\",\"response\":{\"id\":42}}"
let assert Error(types.InvalidEvent(_)) =
send(stream.new(types.Responses), "response.created", malformed)
let unsupported =
"{\"type\":\"response.custom_tool_call_input.delta\",\"delta\":\"rm\"}"
assert send(
stream.new(types.Responses),
"response.custom_tool_call_input.delta",
unsupported,
)
== Error(types.Unsupported(
"unsupported Responses semantic event: response.custom_tool_call_input.delta",
))
}
pub fn responses_argument_deltas_name_the_tool_their_item_opened_with_test() -> Result(
#(stream.State, List(types.Event), Option(types.Turn)),
types.Error,
) {
let added =
"{\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"call_id\":\"c\",\"name\":\"web_search\",\"arguments\":\"\"}}"
let assert Ok(#(state, [], None)) =
send(stream.new(types.Responses), "response.output_item.added", added)
let delta =
"{\"type\":\"response.function_call_arguments.delta\",\"output_index\":1,\"delta\":\"{\\\"q\"}"
let assert Ok(#(_, [types.ArgumentsDelta(1, "web_search", "{\"q")], None)) =
send(state, "response.function_call_arguments.delta", delta)
}
pub fn chat_accumulates_interleaved_tools_usage_and_native_replay_test() -> Nil {
let state = stream.new(types.ChatCompletions)
let first =
"{\"id\":\"chat_1\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"hel\",\"reasoning_content\":\"think \"},\"finish_reason\":null}]}"
let assert Ok(#(
state,
[
types.Started("chat_1"),
types.ThinkingDelta("think "),
types.TextDelta(0, 0, "hel"),
],
None,
)) = send(state, "", first)
let one =
"{\"id\":\"chat_1\",\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":1,\"id\":\"call_b\",\"type\":\"function\",\"function\":{\"name\":\"beta\",\"arguments\":\"{\\\"b\\\":\"}},{\"index\":0,\"id\":\"call_a\",\"type\":\"function\",\"function\":{\"name\":\"alpha\",\"arguments\":\"{\\\"a\\\":\"}}]},\"finish_reason\":null}]}"
let assert Ok(#(
state,
[
types.ArgumentsDelta(1, "beta", "{\"b\":"),
types.ArgumentsDelta(0, "alpha", "{\"a\":"),
],
None,
)) = send(state, "", one)
let two =
"{\"id\":\"chat_1\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"lo\",\"refusal\":\"no\",\"reasoning_content\":\"carefully\",\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"1}\"}},{\"index\":1,\"function\":{\"arguments\":\"2}\"}}]},\"finish_reason\":\"tool_calls\"}]}"
let assert Ok(#(state, events, None)) = send(state, "", two)
assert events
== [
types.ThinkingDelta("carefully"),
types.TextDelta(0, 0, "lo"),
types.ArgumentsDelta(0, "alpha", "1}"),
types.ArgumentsDelta(1, "beta", "2}"),
]
let usage =
"{\"id\":\"chat_1\",\"choices\":[],\"usage\":{\"prompt_tokens\":5,\"completion_tokens\":8,\"total_tokens\":13,\"prompt_tokens_details\":{\"cached_tokens\":0}}}"
let assert Ok(#(state, [], None)) = send(state, "", usage)
let assert Ok(#(
_,
[],
Some(types.Turn(
Some("chat_1"),
[message],
[
types.ToolCall("call_a", "alpha", "{\"a\":1}"),
types.ToolCall("call_b", "beta", "{\"b\":2}"),
],
Some(types.Usage(5, 8, Some(0), None, None, None, None)),
types.ToolCalls,
None,
)),
)) = send(state, "", "[DONE]")
let decoder = {
use content <- decode.field("content", decode.string)
use refusal <- decode.field("refusal", decode.string)
use reasoning <- decode.field("reasoning_content", decode.string)
use tool_ids <- decode.field(
"tool_calls",
decode.list({
use id <- decode.field("id", decode.string)
decode.success(id)
}),
)
decode.success(#(content, refusal, reasoning, tool_ids))
}
assert types.inspect_item(message, decoder)
== Ok(#("hello", "no", "think carefully", ["call_a", "call_b"]))
}
pub fn cache_usage_details_tolerate_null_and_reject_malformed_values_test() -> Result(
#(stream.State, List(types.Event), Option(types.Turn)),
types.Error,
) {
let responses_null =
"{\"type\":\"response.completed\",\"response\":{\"output\":[],\"usage\":{\"input_tokens\":3,\"output_tokens\":2,\"input_tokens_details\":null}}}"
let assert Ok(#(
_,
[],
Some(types.Turn(
_,
[],
[],
Some(types.Usage(3, 2, None, None, None, None, None)),
_,
None,
)),
)) = send(stream.new(types.Responses), "", responses_null)
let chat_null =
"{\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":2,\"prompt_tokens_details\":null}}"
let assert Ok(#(state, _, None)) =
send(stream.new(types.ChatCompletions), "", chat_null)
let assert Ok(#(
_,
[],
Some(types.Turn(
_,
_,
[],
Some(types.Usage(3, 2, None, None, None, None, None)),
_,
None,
)),
)) = send(state, "", "[DONE]")
let responses_malformed =
"{\"type\":\"response.completed\",\"response\":{\"output\":[],\"usage\":{\"input_tokens\":3,\"output_tokens\":2,\"input_tokens_details\":{\"cached_tokens\":\"3\"}}}}"
let assert Error(types.InvalidEvent(_)) =
send(stream.new(types.Responses), "", responses_malformed)
let chat_malformed =
"{\"choices\":[],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":2,\"prompt_tokens_details\":{\"cached_tokens\":\"3\"}}}"
let assert Error(types.InvalidEvent(_)) =
send(stream.new(types.ChatCompletions), "", chat_malformed)
}
pub fn responses_separates_reasoning_summary_parts_test() -> Result(
#(stream.State, List(types.Event), Option(types.Turn)),
types.Error,
) {
let part = fn(index) {
"{\"type\":\"response.reasoning_summary_part.added\",\"output_index\":0,\"summary_index\":"
<> index
<> ",\"part\":{\"type\":\"summary_text\",\"text\":\"\"}}"
}
let state = stream.new(types.Responses)
let assert Ok(#(state, [], None)) = send(state, "", part("0"))
let assert Ok(#(_state, [types.ThinkingDelta("\n\n")], None)) =
send(state, "", part("1"))
}
pub fn chat_rejects_multiple_choices_and_nonempty_unknown_semantics_test() -> Result(
#(stream.State, List(types.Event), Option(types.Turn)),
types.Error,
) {
let multiple =
"{\"choices\":[{\"index\":0,\"delta\":{}},{\"index\":1,\"delta\":{}}]}"
assert send(stream.new(types.ChatCompletions), "", multiple)
== Error(types.Unsupported("multiple chat completion choices"))
let audio =
"{\"choices\":[{\"index\":0,\"delta\":{\"audio\":{\"id\":\"a\"}}}]}"
let assert Error(types.InvalidEvent(_)) =
send(stream.new(types.ChatCompletions), "", audio)
}
pub fn chat_ignores_only_empty_unknown_fields_test() -> Nil {
list.each(["null", "\"\"", "[]", "{}"], fn(value) {
let data =
"{\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hello\",\"audio\":"
<> value
<> "}}]}"
let assert Ok(#(_, [types.TextDelta(0, 0, "hello")], None)) =
send(stream.new(types.ChatCompletions), "", data)
})
list.each(["\"x\"", "[0]", "{\"id\":null}", "0", "false", "true"], fn(value) {
let data =
"{\"choices\":[{\"index\":0,\"delta\":{\"audio\":" <> value <> "}}]}"
let assert Error(types.InvalidEvent(_)) =
send(stream.new(types.ChatCompletions), "", data)
})
}
pub fn chat_rejects_non_object_deltas_test() -> Nil {
list.each(["null", "[]", "42", "\"x\"", "false"], fn(value) {
let data = "{\"choices\":[{\"index\":0,\"delta\":" <> value <> "}]}"
let assert Error(types.InvalidEvent(_)) =
send(stream.new(types.ChatCompletions), "", data)
})
}
pub fn chat_provider_errors_take_precedence_only_with_string_messages_test() -> Nil {
list.each(["", "capacity"], fn(message) {
let data =
json.object([
#("error", json.object([#("message", json.string(message))])),
#(
"choices",
json.array(["hello"], fn(content) {
json.object([
#("index", json.int(0)),
#("delta", json.object([#("content", json.string(content))])),
])
}),
),
])
|> json.to_string
assert send(stream.new(types.ChatCompletions), "", data)
== Error(types.ProviderError(message))
})
list.each(
[
"null", "42", "\"x\"", "{}", "{\"message\":null}", "{\"message\":42}",
"{\"message\":false}", "{\"message\":[]}",
],
fn(value) {
let data =
"{\"error\":"
<> value
<> ",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hello\"}}]}"
let assert Ok(#(_, [types.TextDelta(0, 0, "hello")], None)) =
send(stream.new(types.ChatCompletions), "", data)
},
)
let assert Error(types.InvalidEvent(_)) =
send(stream.new(types.ChatCompletions), "", "{\"error\":{\"message\":42}}")
Nil
}
pub fn chat_requires_finish_and_suppresses_partial_tools_on_limits_test() -> Nil {
assert send(stream.new(types.ChatCompletions), "", "[DONE]")
== Error(types.InvalidEvent("[DONE] before chat finish reason"))
let partial =
"{\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_partial\",\"type\":\"function\",\"function\":{\"name\":\"unsafe\",\"arguments\":\"{\"}}]},\"finish_reason\":\"length\"}]}"
let assert Ok(#(state, _, None)) =
send(stream.new(types.ChatCompletions), "", partial)
let assert Ok(#(
_,
[],
Some(types.Turn(None, [message], [], None, types.LengthLimit, None)),
)) = send(state, "", "[DONE]")
let decoder = {
use calls <- decode.optional_field(
"tool_calls",
None,
decode.optional(decode.list(decode.dynamic)),
)
decode.success(calls)
}
assert types.inspect_item(message, decoder) == Ok(None)
}
pub fn malformed_json_and_content_filter_finish_are_typed_test() -> Result(
#(stream.State, List(types.Event), Option(types.Turn)),
types.Error,
) {
let assert Error(types.InvalidEvent(_)) =
send(stream.new(types.ChatCompletions), "", "{")
let assert Error(types.InvalidEvent(_)) =
send(stream.new(types.Responses), "", "[]")
let filtered =
"{\"choices\":[{\"index\":0,\"delta\":{\"content\":\"safe prefix\"},\"finish_reason\":\"content_filter\"}]}"
let assert Ok(#(state, _, None)) =
send(stream.new(types.ChatCompletions), "", filtered)
let assert Ok(#(
_,
[],
Some(types.Turn(_, _, [], _, types.ContentFiltered, None)),
)) = send(state, "", "[DONE]")
}
pub fn chat_assembles_name_fragments_and_encodes_tool_only_content_as_null_test() -> Result(
#(stream.State, List(types.Event), Option(types.Turn)),
types.Error,
) {
let first =
"{\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call1\",\"function\":{\"name\":\"read_\",\"arguments\":\"{\"}}]}}]}"
let second =
"{\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"name\":\"file\",\"arguments\":\"}\"}}]},\"finish_reason\":\"tool_calls\"}]}"
let assert Ok(#(state, _, None)) =
send(stream.new(types.ChatCompletions), "", first)
let assert Ok(#(state, _, None)) = send(state, "", second)
let assert Ok(#(state, _, Some(turn))) = send(state, "", "[DONE]")
assert turn.tool_calls == [types.ToolCall("call1", "read_file", "{}")]
let assert [item] = turn.output
let encoded = types.replay_json(item) |> json.to_string
assert json.parse(
encoded,
decode.at(["content"], decode.optional(decode.string)),
)
== Ok(None)
let assert Error(types.InvalidEvent(_)) = send(state, "", "[DONE]")
}
pub fn chat_rejects_data_after_finish_reason_test() -> Result(
#(stream.State, List(types.Event), Option(types.Turn)),
types.Error,
) {
let first =
"{\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}"
let late = "{\"choices\":[{\"index\":0,\"delta\":{\"content\":\"late\"}}]}"
let assert Ok(#(state, _, None)) =
send(stream.new(types.ChatCompletions), "", first)
let assert Error(types.InvalidEvent(_)) = send(state, "", late)
}
pub fn responses_rejects_incomplete_call_in_completed_response_test() -> Result(
#(stream.State, List(types.Event), Option(types.Turn)),
types.Error,
) {
let event =
"{\"type\":\"response.completed\",\"response\":{\"output\":[{\"type\":\"function_call\",\"call_id\":\"c\",\"name\":\"tool\",\"arguments\":\"{}\",\"status\":\"in_progress\"}]}}"
let assert Error(types.InvalidEvent(_)) =
send(stream.new(types.Responses), "", event)
}
pub fn compatible_reasoning_fields_are_preserved_without_renaming_test() -> Nil {
let first =
"{\"choices\":[{\"index\":0,\"delta\":{\"reasoning\":\"r1\",\"reasoning_content\":\"c1\"}}]}"
let second =
"{\"choices\":[{\"index\":0,\"delta\":{\"reasoning\":\"r2\",\"reasoning_content\":\"c2\",\"content\":\"answer\"},\"finish_reason\":\"stop\"}]}"
let assert Ok(#(state, _, None)) =
send(stream.new(types.ChatCompletions), "", first)
let assert Ok(#(state, _, None)) = send(state, "", second)
let assert Ok(#(_, _, Some(turn))) = send(state, "", "[DONE]")
let assert [item] = turn.output
assert types.inspect_item(item, decode.at(["reasoning"], decode.string))
== Ok("r1r2")
assert types.inspect_item(
item,
decode.at(["reasoning_content"], decode.string),
)
== Ok("c1c2")
}
pub fn reasoning_details_merge_fragments_by_index_and_keep_metadata_test() -> Nil {
let first =
"{\"choices\":[{\"index\":0,\"delta\":{\"reasoning\":\"a\",\"reasoning_details\":[{\"type\":\"reasoning.text\",\"index\":0,\"format\":\"unknown\",\"text\":\"a\"}]}}]}"
let second =
"{\"choices\":[{\"index\":0,\"delta\":{\"reasoning\":\"b\",\"reasoning_details\":[{\"type\":\"reasoning.text\",\"index\":0,\"format\":\"unknown\",\"text\":\"b\"}],\"content\":\"answer\"},\"finish_reason\":\"stop\"}]}"
let assert Ok(#(state, _, None)) =
send(stream.new(types.ChatCompletions), "", first)
let assert Ok(#(state, _, None)) = send(state, "", second)
let assert Ok(#(_, _, Some(turn))) = send(state, "", "[DONE]")
let assert [item] = turn.output
assert types.inspect_item(
item,
decode.at(
["reasoning_details"],
decode.list(decode.at(["text"], decode.string)),
),
)
== Ok(["ab"])
assert types.inspect_item(
item,
decode.at(
["reasoning_details"],
decode.list(decode.at(["format"], decode.string)),
),
)
== Ok(["unknown"])
}
pub fn responses_uses_streamed_done_items_when_terminal_output_is_empty_test() -> Nil {
let state = stream.new(types.Responses)
let done =
"{\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"python\",\"arguments\":\"{\\\"code\\\":\\\"20 + 22\\\",\\\"timeout_ms\\\":1000}\",\"status\":\"completed\"}}"
let assert Ok(#(state, [], None)) = send(state, "", done)
let completed =
"{\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"output\":[],\"usage\":{\"input_tokens\":10,\"output_tokens\":5}}}"
let assert Ok(#(
_,
[types.Started("resp_1")],
Some(types.Turn(
Some("resp_1"),
[_],
[types.ToolCall("call_1", "python", arguments)],
_,
types.ToolCalls,
None,
)),
)) = send(state, "", completed)
assert arguments == "{\"code\":\"20 + 22\",\"timeout_ms\":1000}"
}
fn send(
state: stream.State,
name: String,
data: String,
) -> Result(#(stream.State, List(types.Event), Option(types.Turn)), types.Error) {
stream.feed(state, sse.Event(name, data))
}