diff --git a/backend/internal/service/openai_ws_v2/passthrough_relay.go b/backend/internal/service/openai_ws_v2/passthrough_relay.go index e28223e091..fdb539a073 100644 --- a/backend/internal/service/openai_ws_v2/passthrough_relay.go +++ b/backend/internal/service/openai_ws_v2/passthrough_relay.go @@ -533,17 +533,26 @@ func runUpstreamToClient( msgType, payload, err := upstreamConn.ReadFrame(ctx) if err != nil { emitTurnComplete(onTurnComplete, state, finalizePendingBareError(state, nowFn())) + graceful := isDisconnectError(err) + // A clean WebSocket close only describes the transport handshake. Once + // the upstream has started a Responses turn, success still requires a + // terminal protocol event. Treat an early 1000/EOF as a relay failure so + // the adapter does not report relay_completed with an active turn. + if graceful && openAIWSRelayActiveTurnID(state) != "" { + graceful = false + err = errors.New("upstream websocket closed before terminal event: " + err.Error()) + } emitRelayTrace(onTrace, RelayTraceEvent{ Stage: "read_upstream_failed", Direction: "upstream_to_client", Error: err.Error(), - Graceful: isDisconnectError(err), + Graceful: graceful, WroteDownstream: wroteDownstream, }) exitCh <- relayExitSignal{ stage: "read_upstream", err: err, - graceful: isDisconnectError(err), + graceful: graceful, wroteDownstream: wroteDownstream, } return diff --git a/backend/internal/service/openai_ws_v2/passthrough_relay_test.go b/backend/internal/service/openai_ws_v2/passthrough_relay_test.go index 28b896f248..a5505b4981 100644 --- a/backend/internal/service/openai_ws_v2/passthrough_relay_test.go +++ b/backend/internal/service/openai_ws_v2/passthrough_relay_test.go @@ -41,6 +41,11 @@ type closeSpyFrameConn struct { closeCalls atomic.Int32 } +type eofReplacementFrameConn struct { + FrameConn + err error +} + type cancelJoinProbeFrameConn struct { readStarted chan struct{} readCanceled chan struct{} @@ -65,6 +70,14 @@ func newPassthroughTestFrameConn(frames []passthroughTestFrame, autoClose bool) return c } +func (c *eofReplacementFrameConn) ReadFrame(ctx context.Context) (coderws.MessageType, []byte, error) { + msgType, payload, err := c.FrameConn.ReadFrame(ctx) + if errors.Is(err, io.EOF) { + return msgType, payload, c.err + } + return msgType, payload, err +} + func (c *passthroughTestFrameConn) ReadFrame(ctx context.Context) (coderws.MessageType, []byte, error) { if ctx == nil { ctx = context.Background() @@ -301,6 +314,38 @@ func TestRelay_UpstreamDisconnect(t *testing.T) { require.Equal(t, "gpt-4o", result.RequestModel) } +func TestRelay_UpstreamNormalCloseBeforeTerminalIsFailure(t *testing.T) { + t.Parallel() + + clientConn := newPassthroughTestFrameConn(nil, false) + upstreamConn := &eofReplacementFrameConn{ + FrameConn: newPassthroughTestFrameConn([]passthroughTestFrame{ + { + msgType: coderws.MessageText, + payload: []byte(`{"type":"response.created","response":{"id":"resp_incomplete","status":"in_progress"}}`), + }, + { + msgType: coderws.MessageText, + payload: []byte(`{"type":"response.in_progress","response":{"id":"resp_incomplete","status":"in_progress"}}`), + }, + }, true), + err: coderws.CloseError{Code: coderws.StatusNormalClosure}, + } + + firstPayload := []byte(`{"type":"response.create","model":"gpt-5.6-sol","input":[]}`) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + result, relayExit := Relay(ctx, clientConn, upstreamConn, firstPayload, RelayOptions{}) + require.NotNil(t, relayExit) + require.Equal(t, "read_upstream", relayExit.Stage) + require.False(t, relayExit.Graceful) + require.True(t, relayExit.WroteDownstream) + require.ErrorContains(t, relayExit.Err, "upstream websocket closed before terminal event") + require.Empty(t, result.TerminalEventType) + require.Len(t, clientConn.Writes(), 2) +} + func TestRelay_ClientDisconnect(t *testing.T) { t.Parallel()