Merge pull request #6470 from aofee/codex/fix-ws-passthrough-missing-terminal

fix(openai-ws): reject passthrough close before terminal event
This commit is contained in:
Wesley Liddick
2026-09-02 09:42:13 +08:00
committed by GitHub
2 changed files with 56 additions and 2 deletions
@@ -533,17 +533,26 @@ func runUpstreamToClient(
msgType, payload, err := upstreamConn.ReadFrame(ctx) msgType, payload, err := upstreamConn.ReadFrame(ctx)
if err != nil { if err != nil {
emitTurnComplete(onTurnComplete, state, finalizePendingBareError(state, nowFn())) 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{ emitRelayTrace(onTrace, RelayTraceEvent{
Stage: "read_upstream_failed", Stage: "read_upstream_failed",
Direction: "upstream_to_client", Direction: "upstream_to_client",
Error: err.Error(), Error: err.Error(),
Graceful: isDisconnectError(err), Graceful: graceful,
WroteDownstream: wroteDownstream, WroteDownstream: wroteDownstream,
}) })
exitCh <- relayExitSignal{ exitCh <- relayExitSignal{
stage: "read_upstream", stage: "read_upstream",
err: err, err: err,
graceful: isDisconnectError(err), graceful: graceful,
wroteDownstream: wroteDownstream, wroteDownstream: wroteDownstream,
} }
return return
@@ -41,6 +41,11 @@ type closeSpyFrameConn struct {
closeCalls atomic.Int32 closeCalls atomic.Int32
} }
type eofReplacementFrameConn struct {
FrameConn
err error
}
type cancelJoinProbeFrameConn struct { type cancelJoinProbeFrameConn struct {
readStarted chan struct{} readStarted chan struct{}
readCanceled chan struct{} readCanceled chan struct{}
@@ -65,6 +70,14 @@ func newPassthroughTestFrameConn(frames []passthroughTestFrame, autoClose bool)
return c 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) { func (c *passthroughTestFrameConn) ReadFrame(ctx context.Context) (coderws.MessageType, []byte, error) {
if ctx == nil { if ctx == nil {
ctx = context.Background() ctx = context.Background()
@@ -301,6 +314,38 @@ func TestRelay_UpstreamDisconnect(t *testing.T) {
require.Equal(t, "gpt-4o", result.RequestModel) 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) { func TestRelay_ClientDisconnect(t *testing.T) {
t.Parallel() t.Parallel()