fix(server): do not append [DONE] after an upstream in-band stream error - #334
fix(server): do not append [DONE] after an upstream in-band stream error#334enwaiax wants to merge 1 commit into
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Enterprise Run ID: 📒 Files selected for processing (3)
WalkthroughThe translation encoder now reports whether an in-band error ended the stream. The server passes this outcome to SSE framing. OpenAI Chat framing suppresses ChangesStream outcome propagation
Estimated code review effort: 3 (Moderate) | ~20 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
An upstream SSE error event is a well-formed JSON event, so it flows through frame_event() as ordinary data and leaves the framing loop's `failed` flag unset. The OpenAI Chat sentinel is then appended to a stream that did not complete, and an SDK client that stops at [DONE] reports the truncated, failed turn as a successful completion. The translation layer already detects this: encode_stream() checks state.errored and returns early, deliberately skipping codec.finish(). That outcome was simply never communicated to the serving layer, which cannot otherwise distinguish a clean EOF from an early stop. Expose it via StreamOutcome and consult it before emitting the sentinel. encode_stream() keeps its signature and delegates to the new encode_stream_with_outcome(), so existing callers are unaffected. Only the Chat leg was affected: [DONE] is Chat-specific, and the terminal events for Anthropic Messages and OpenAI Responses come from codec.finish(), which the early return already skips. Signed-off-by: enwaiax <32839114+enwaiax@users.noreply.github.com>
b64e2e9 to
ce68902
Compare
nachiketb-nvidia
left a comment
nachiketb-nvidia
left a comment
There was a problem hiding this comment.
Thanks for the catch, seems a rebase off main is in order.
LGTM otherwise!
nachiketb-nvidia
left a comment
nachiketb-nvidia
left a comment
There was a problem hiding this comment.
Retracting my approval for now while this still needs follow-up before merge.
| /// success sentinel (OpenAI Chat's `[DONE]`) to a failed stream. This flag | ||
| /// carries that outcome across the layer boundary; `false` until the encoder | ||
| /// observes an error. | ||
| pub type StreamOutcome = std::sync::Arc<std::sync::atomic::AtomicBool>; |
There was a problem hiding this comment.
think we can solve this more locally without adding StreamOutcome.
The Arc<AtomicBool> works, but it feels like the wrong abstraction here. The server is the layer appending [DONE], so the server should also own the decision to suppress it. In this case the signal is already present in the stream item: an OpenAI Chat error frame like {"error": ...} is terminal and
should not be followed by [DONE].
Could we keep this entirely inside sse.rs with a small private enum instead?
enum FrameOutcome {
Continue(Event),
TerminalError(Event),
}
fn frame_event(value: Value, target_format: WireFormat) -> Result<FrameOutcome, BoxError> {
let is_terminal_error =
target_format == WireFormat::OpenAiChat && value.get("error").is_some();
let event = Event::default().data(serde_json::to_string(&value)?);
Ok(if is_terminal_error {
FrameOutcome::TerminalError(event)
} else {
FrameOutcome::Continue(event)
})
}
Then frame_stream can keep local state:
let mut failed = false;
while let Some(item) = stream.next().await {
match item {
Ok(value) => match frame_event(value, target_format) {
Ok(FrameOutcome::Continue(event)) => yield Ok(event),
Ok(FrameOutcome::TerminalError(event)) => {
failed = true;
yield Ok(event);
break;
}
Err(error) => {
failed = true;
yield Ok(error_event(error));
break;
}
},
Err(error) => {
failed = true;
yield Ok(error_event(error));
break;
}
}
}
if !failed && target_format == WireFormat::OpenAiChat {
yield Ok(Event::default().data("[DONE]"));
}That keeps the fix targeted:
- no translation API change
- no public StreamOutcome
- no Arc / memory ordering discussion for a single stream loop
- the [DONE] decision stays next to the code that emits [DONE]
Longer term, if we want stream termination to be a first-class contract, I’d rather see a typed translated stream item with an explicit terminal status. But for this PR, keeping it private and local to SSE framing seems cleaner.
2 participants
Problem
When an upstream emits an SSE error event mid-stream,
/v1/chat/completionsforwards that error frame and then still appends
data: [DONE]— the OpenAI Chatsuccess sentinel. An SDK client that stops at
[DONE]treats the truncated,failed turn as a normally completed answer.
Captured against the release binary (loopback upstream, one content frame followed
by one error frame):
/v1/messagesand/v1/responsesare unaffected — neither uses a[DONE]sentinel, and both already terminate on the error event.
Root cause
crates/switchyard-server/src/sse.rstracks a localfailedflag, but only aframing/JSON error or a transport-level stream error sets it. An upstream in-band
error event is a well-formed JSON event, so it flows through
frame_event()asordinary data and leaves
failed == false:The translation layer already knows the stream failed:
encode_stream()checksstate.erroredand returns early, deliberately skippingcodec.finish(). Thatoutcome was simply never communicated to the serving layer, which cannot otherwise
tell a clean EOF from an early stop.
This also matches the contract already stated in
crates/switchyard-translation/src/helpers.rs: "An in-band error is terminal forevery target format: the encoder emits the pre-error content and the error, then
drops any later chunk."
Fix
Expose the outcome across the layer boundary and consult it before emitting the
sentinel.
[DONE]stays in the serving layer — framing is the serving layer's jobper the
RawEventStreamdoc contract, and[DONE]is not a JSON event object, soit does not belong in a codec.
StreamOutcome(a sharedAtomicBool) set whereencode_streamalreadyreturns early on
state.erroredencode_stream_with_outcome();encode_stream()keeps its signature anddelegates to it, so existing callers (e.g.
libsy-llm-client) are untouchedframe_stream()skips the sentinel when the outcome reports an in-band errorTests
Two new unit tests in
sse.rs:upstream_in_band_error_suppresses_the_done_marker— the reported defectclean_stream_still_emits_the_done_marker— guards the OpenAI Chat contract forthe success path
The pre-existing
stream_error_terminates_without_done_markerstill passes.Verification
End-to-end A/B against the same loopback upstream, same test, only the binary
swapped:
Note
The Python
switchyard serveentry point has the same defect(
switchyard/lib/endpoints/sse_helpers.pyyields[DONE]unless an exception israised, and an upstream error event is not an exception). It has no equivalent
erroredsignal to read, so it needs its own fix rather than a port of this one —not included here to keep this PR to a single component.
Summary by CodeRabbit