Skip to content

fix(llm): process terminal SSE frames at EOF #106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wolfy-j merged 1 commit into from
Aug 7, 2026
Merged

fix(llm): process terminal SSE frames at EOF #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
39 changes: 28 additions & 11 deletions src/llm/src/openai/client.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -297,7 +297,7 @@ function openai_client.process_stream(stream_response, callbacks): (string?, any
})
end

local function build_result()
local function build_result(response_override)
local tool_calls_out = {}
for _, call in pairs(pending_calls) do
if call.call_id and call.name then
Expand All @@ -309,14 +309,23 @@ function openai_client.process_stream(stream_response, callbacks): (string?, any
end
end

local terminal_response = response_override or final_response
local terminal_usage = terminal_response and terminal_response.usage or final_usage
local terminal_status = terminal_response and terminal_response.status or response_status
local terminal_response_id = terminal_response and terminal_response.id or response_id
local terminal_incomplete_reason = incomplete_reason
if terminal_response and terminal_response.incomplete_details then
terminal_incomplete_reason = terminal_response.incomplete_details.reason
end

return {
content = full_content,
tool_calls = tool_calls_out,
usage = final_usage,
status = response_status,
incomplete_reason = incomplete_reason,
response_id = response_id,
response = final_response,
usage = terminal_usage,
status = terminal_status,
incomplete_reason = terminal_incomplete_reason,
response_id = terminal_response_id,
response = terminal_response,
metadata = metadata
}
end
Expand All @@ -328,11 +337,11 @@ function openai_client.process_stream(stream_response, callbacks): (string?, any
end
end

local function finish_stream(): (string, any, any)
local function finish_stream(response_override): (string, any, any)
for key, _ in pairs(pending_calls) do
emit_call(key)
end
local result: any = build_result()
local result: any = build_result(response_override)
close_stream()
on_done(result)
return full_content, nil, result
Expand All @@ -348,7 +357,15 @@ function openai_client.process_stream(stream_response, callbacks): (string?, any
return nil, err
end

if not chunk then break end
if not chunk then
if leftover == "" then break end
-- Be tolerant of a transport that closes immediately after its
-- final data line instead of sending the terminating SSE blank
-- line. Complete the frame locally so terminal success and error
-- events are still observed.
chunk = leftover .. "\n\n"
leftover = ""
end
if chunk == "" then goto continue end

if leftover ~= "" then
Expand Down Expand Up @@ -467,7 +484,7 @@ function openai_client.process_stream(stream_response, callbacks): (string?, any
-- response.completed is the terminal event. Some
-- Responses transports keep the HTTP connection alive,
-- so waiting for EOF can block an otherwise finished turn.
return finish_stream()
return finish_stream(parsed.response)
end
elseif etype == "response.incomplete" then
if parsed.response then
Expand All @@ -477,7 +494,7 @@ function openai_client.process_stream(stream_response, callbacks): (string?, any
if parsed.response.incomplete_details then
incomplete_reason = parsed.response.incomplete_details.reason
end
return finish_stream()
return finish_stream(parsed.response)
end
elseif etype == "response.failed" or etype == "error" or etype == "response.error" then
local err_payload = parsed.response and parsed.response.error or parsed.error or parsed
Expand Down
37 changes: 37 additions & 0 deletions src/llm/src/openai/client_test.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -554,6 +554,43 @@ local function define_tests()
test.eq(result.response_id, "r-terminal")
end)

it("should process a terminal event at EOF without a blank delimiter", function()
local mock_stream = build_mock_stream({
'data: {"type":"response.completed","response":{"id":"r-eof","status":"completed","usage":{"input_tokens":1,"output_tokens":1}}}'
})

local done_result: any = nil
local _, err, result = openai_client.process_stream({
stream = mock_stream,
metadata = {}
}, {
on_done = function(value) done_result = value end
})

test.is_nil(err)
test.eq(done_result.response_id, "r-eof")
test.eq(result.response_id, "r-eof")
test.not_nil(result.response)
end)

it("should process a failed event at EOF without a blank delimiter", function()
local mock_stream = build_mock_stream({
'data: {"type":"response.failed","response":{"id":"r-eof-failed","status":"failed","error":{"message":"eof failure","type":"server_error"}}}'
})

local seen_error = nil
local _, err, result = openai_client.process_stream({
stream = mock_stream,
metadata = {}
}, {
on_error = function(value) seen_error = value end
})

test.eq(err, "eof failure")
test.not_nil(seen_error)
test.eq(result.error.message, "eof failure")
end)

it("should process CRLF framing split across stream reads", function()
local mock_stream = build_mock_stream({
'data: {"type":"response.output_text.delta","delta":"Hello"}\r\n\r',
Expand Down
Loading