fix(openai-ws): reject close before terminal event

This commit is contained in:
aofee
2026-09-01 20:44:11 +08:00
parent 0d27f45ead
commit bfe0a5a873
2 changed files with 56 additions and 2 deletions
@@ -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
@@ -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()