fix(TASK-044): streaming response.incomplete parity — keep partial output, record usage
Review findings: the streaming arm now mirrors the non-streaming twin (openai_extract_responses): partial text/tool calls are returned rather than discarded by a hard error; only an empty truncated response bails. Billed usage on truncated turns is recorded before either outcome. Also: CHANGELOG notes the streaming-incomplete surfacing and corrects the include/strict phrasing (sent on every responses-wire body).
This commit is contained in:
3 files changed
+69
-1
No files matched your search
@@ -944,6 +944,18 @@ fn openai_responses_handle_event(data: &Value, handler: &mut SseHandler) -> Resu
|
||||
None => bail!("Response failed: {data}"),
|
||||
},
|
||||
Some("response.incomplete") => {
|
||||
// Truncated turns are the most expensive ones; record their
|
||||
// billed usage before deciding how to surface the truncation.
|
||||
if let Some(usage) = openai_parse_responses_usage(&data["response"]["usage"]) {
|
||||
debug!("token-usage: {usage:?}");
|
||||
handler.usage(usage);
|
||||
}
|
||||
// Parity with the non-streaming path (`openai_extract_responses`):
|
||||
// partial text or tool calls are returned rather than discarded
|
||||
// by a hard error; only an empty truncated response bails.
|
||||
if handler.has_received_visible_output() {
|
||||
return Ok(true);
|
||||
}
|
||||
match data["response"]["incomplete_details"]["reason"].as_str() {
|
||||
Some(reason) => bail!("The response was cut off: {reason}"),
|
||||
None => bail!("The response was cut off: {data}"),
|
||||
@@ -1406,6 +1418,53 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn responses_stream_incomplete_without_reason_still_bails() {
|
||||
let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
|
||||
let abort_signal = crate::utils::create_abort_signal();
|
||||
let mut handler = SseHandler::new(sender, abort_signal);
|
||||
let event = json!({
|
||||
"type": "response.incomplete",
|
||||
"response": {}
|
||||
});
|
||||
|
||||
let err = openai_responses_handle_event(&event, &mut handler)
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
|
||||
assert!(err.contains("The response was cut off"), "unexpected error: {err}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn responses_stream_incomplete_with_partial_output_ends_stream_and_records_usage() {
|
||||
let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
|
||||
let abort_signal = crate::utils::create_abort_signal();
|
||||
let mut handler = SseHandler::new(sender, abort_signal);
|
||||
handler.set_silent(true);
|
||||
handler.text("partial answer").unwrap();
|
||||
let event = json!({
|
||||
"type": "response.incomplete",
|
||||
"response": {
|
||||
"incomplete_details": { "reason": "max_output_tokens" },
|
||||
"usage": {
|
||||
"input_tokens": 50,
|
||||
"output_tokens": 10,
|
||||
"input_tokens_details": { "cached_tokens": 25 }
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let done = openai_responses_handle_event(&event, &mut handler).unwrap();
|
||||
|
||||
assert!(done, "partial output should end the stream, not bail");
|
||||
let (text, _, _, usage) = handler.take();
|
||||
assert_eq!(text, "partial answer");
|
||||
let usage = usage.expect("usage should be recorded for truncated turns");
|
||||
assert_eq!(usage.input_tokens, Some(50));
|
||||
assert_eq!(usage.output_tokens, Some(10));
|
||||
assert_eq!(usage.cache_read_input_tokens, Some(25));
|
||||
}
|
||||
|
||||
fn openai_config(name: &str, auth: Option<&str>, oauth: Option<OAuthConfig>) -> OpenAIConfig {
|
||||
OpenAIConfig {
|
||||
name: Some(name.into()),
|
||||
|
||||
@@ -215,6 +215,14 @@ impl SseHandler {
|
||||
!self.buffer.is_empty() || !self.tool_calls.is_empty() || !self.thinking.is_empty()
|
||||
}
|
||||
|
||||
/// Whether visible output (text or tool calls) has been accumulated.
|
||||
/// Unlike [`Self::has_received_content`], thinking blocks do not count:
|
||||
/// a stream truncated before any visible output is still an empty
|
||||
/// response to the caller.
|
||||
pub fn has_received_visible_output(&self) -> bool {
|
||||
!self.buffer.is_empty() || !self.tool_calls.is_empty()
|
||||
}
|
||||
|
||||
pub fn abort(&self) -> AbortSignal {
|
||||
self.abort_signal.clone()
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user