fix(openai-ws): track pending passthrough turns

This commit is contained in:
aofee
2026-09-02 14:59:13 +08:00
parent 5097b31457
commit a7f1ed8fa4
2 changed files with 66 additions and 6 deletions
@@ -538,7 +538,7 @@ func runUpstreamToClient(
// 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) != "" {
if graceful && state.hasUnfinishedTurn() {
graceful = false
err = errors.New("upstream websocket closed before terminal event: " + err.Error())
}
@@ -585,7 +585,14 @@ func runUpstreamToClient(
}
observedEvent = observeUpstreamMessage(state, payload, startAt, nowFn, onUsageParseFailure)
case coderws.MessageBinary:
// binary frame 直接透传,不进入 JSON 观测路径(避免无效解析开销)。
// Binary frames remain opaque for usage/result observation, but a JSON
// terminal still settles relay lifecycle. Otherwise the pending-turn
// disconnect guard would turn an already-delivered terminal into a false
// missing-terminal failure when the upstream closes normally.
if isTerminalEvent(strings.TrimSpace(gjson.GetBytes(payload, "type").String())) {
state.consumePendingTurnStartedAt()
openAIWSRelayDiscardActiveTurnTiming(state)
}
}
emitTurnComplete(onTurnComplete, state, observedEvent)
if dropDownstreamWrites != nil && dropDownstreamWrites.Load() {
@@ -995,6 +1002,13 @@ func (s *relayState) consumePendingTurnStartedAt() time.Time {
return *startedAt
}
func (s *relayState) hasUnfinishedTurn() bool {
if s == nil {
return false
}
return s.pendingTurnStart.Load() != nil || s.activeTurn != nil
}
func openAIWSRelayDeleteTurnTiming(state *relayState, responseID string) (relayTurnTiming, bool) {
if state == nil || state.turnTimingByID == nil {
return relayTurnTiming{}, false
@@ -309,11 +309,46 @@ func TestRelay_UpstreamDisconnect(t *testing.T) {
defer cancel()
result, relayExit := Relay(ctx, clientConn, upstreamConn, firstPayload, RelayOptions{})
// 上游 EOF 属于 disconnect,标记为 graceful
require.Nil(t, relayExit, "上游 EOF 应被视为 graceful disconnect")
require.NotNil(t, relayExit)
require.Equal(t, "read_upstream", relayExit.Stage)
require.False(t, relayExit.Graceful)
require.ErrorContains(t, relayExit.Err, "upstream websocket closed before terminal event")
require.Equal(t, "gpt-4o", result.RequestModel)
}
func TestRelay_UpstreamNormalCloseBeforeResponseIDIsFailure(t *testing.T) {
t.Parallel()
clientConn := newPassthroughTestFrameConn(nil, false)
upstreamConn := &eofReplacementFrameConn{
FrameConn: newPassthroughTestFrameConn([]passthroughTestFrame{
{
msgType: coderws.MessageText,
payload: []byte(`{"type":"response.created","status":"in_progress"}`),
},
{
msgType: coderws.MessageText,
payload: []byte(`{"type":"response.in_progress","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.RequestID)
require.Empty(t, result.TerminalEventType)
require.Len(t, clientConn.Writes(), 2)
}
func TestRelay_UpstreamNormalCloseBeforeTerminalIsFailure(t *testing.T) {
t.Parallel()
@@ -921,7 +956,10 @@ func TestRelay_BinaryFramePassthrough(t *testing.T) {
defer cancel()
result, relayExit := Relay(ctx, clientConn, upstreamConn, firstPayload, RelayOptions{})
require.Nil(t, relayExit)
require.NotNil(t, relayExit)
require.Equal(t, "read_upstream", relayExit.Stage)
require.False(t, relayExit.Graceful)
require.ErrorContains(t, relayExit.Err, "upstream websocket closed before terminal event")
// binary frame 不解析 usage
require.Equal(t, 0, result.Usage.InputTokens)
@@ -986,7 +1024,12 @@ func TestRelay_PreservesFirstMessageType(t *testing.T) {
t.Parallel()
clientConn := newPassthroughTestFrameConn(nil, false)
upstreamConn := newPassthroughTestFrameConn(nil, true)
upstreamConn := newPassthroughTestFrameConn([]passthroughTestFrame{
{
msgType: coderws.MessageBinary,
payload: []byte(`{"type":"response.completed","response":{"id":"resp_binary_first_type"}}`),
},
}, true)
firstPayload := []byte(`{"type":"response.create","model":"gpt-4o","input":[]}`)
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
@@ -1001,6 +1044,9 @@ func TestRelay_PreservesFirstMessageType(t *testing.T) {
require.Len(t, upstreamWrites, 1)
require.Equal(t, coderws.MessageBinary, upstreamWrites[0].msgType)
require.Equal(t, firstPayload, upstreamWrites[0].payload)
clientWrites := clientConn.Writes()
require.Len(t, clientWrites, 1)
require.Equal(t, coderws.MessageBinary, clientWrites[0].msgType)
}
func TestRelay_UsageParseFailureDoesNotBlockRelay(t *testing.T) {