openai_responses_continuation_websocket_test.go RawHistoryFolder
1package provider
2
3import (
4 "context"
5 "encoding/json"
6 "net/http/httptest"
7 "strings"
8 "sync"
9 "testing"
10
11 "golang.org/x/net/websocket"
12)
13
14// newContinuationWebSocketServer serves two sequential Responses on one
15// WebSocket connection: interim text with explicit end_turn:false, then
16// final text with no end_turn field (missing still terminates).
17func newContinuationWebSocketServer(t *testing.T, mu *sync.Mutex, requests *[]map[string]any) *httptest.Server {
18 t.Helper()
19 texts := []string{"part one", "final"}
20 endTurns := []any{false, nil}
21 return httptest.NewServer(websocket.Handler(func(conn *websocket.Conn) {
22 defer conn.Close()
23 for i := 0; i < 2; i++ {
24 var payload map[string]any
25 if err := websocket.JSON.Receive(conn, &payload); err != nil {
26 t.Errorf("receive websocket request: %v", err)
27 return
28 }
29 mu.Lock()
30 *requests = append(*requests, payload)
31 mu.Unlock()
32 response := map[string]any{
33 "id": "resp_" + string(rune('1'+i)),
34 "usage": map[string]any{"input_tokens": 1, "output_tokens": 1},
35 }
36 if endTurns[i] != nil {
37 response["end_turn"] = endTurns[i]
38 }
39 for _, event := range []map[string]any{
40 {"type": "response.created", "response": map[string]any{"id": response["id"]}},
41 {"type": "response.output_item.added", "output_index": 0, "item": map[string]any{"type": "message"}},
42 {"type": "response.output_text.delta", "output_index": 0, "delta": texts[i]},
43 {"type": "response.completed", "response": response},
44 } {
45 _ = websocket.JSON.Send(conn, event)
46 }
47 }
48 }))
49}
50
51// streamContinuationResponse collects one Responses stream into its stop
52// reason and assembled text.
53func streamContinuationResponse(t *testing.T, client Client, req Request) (StopReason, string) {
54 t.Helper()
55 stream, err := client.Stream(context.Background(), req)
56 if err != nil {
57 t.Fatal(err)
58 }
59 var (
60 done EventDone
61 text strings.Builder
62 )
63 for event := range stream {
64 switch e := event.(type) {
65 case EventTextDelta:
66 text.WriteString(e.Delta)
67 case EventDone:
68 done = e
69 }
70 }
71 if done.Err != nil {
72 t.Fatalf("stream error: %v", done.Err)
73 }
74 return done.Stop, text.String()
75}
76
77// TestResponsesContinuationWebSocket pins explicit continuation over the
78// Responses WebSocket transport: the first response carries
79// end_turn:false, the second request reuses the connection with the
80// expected previous-response/incremental state, and the second response
81// terminates normally.
82func TestResponsesContinuationWebSocket(t *testing.T) {
83 var (
84 mu sync.Mutex
85 requests []map[string]any
86 )
87 server := newContinuationWebSocketServer(t, &mu, &requests)
88 defer server.Close()
89
90 inner := &codexClient{
91 token: "test-token",
92 baseURL: server.URL + "/v1/responses",
93 providerName: "openai",
94 capabilities: responsesCapabilities{StablePromptCacheKey: true},
95 }
96 client := newResponsesWebSocketClient(inner)
97
98 first := Request{
99 Model: "gpt-5.6-sol",
100 Context: RequestContext{CacheSessionID: "cache-root-1", ThreadID: "thread-1", TurnID: "turn-1"},
101 Messages: []Message{{Role: RoleUser, Content: []Content{TextBlock{Text: "first"}}}},
102 }
103 if stop, text := streamContinuationResponse(t, client, first); stop != StopContinue || text != "part one" {
104 t.Fatalf("first response = (%q, %q), want (continue, part one)", stop, text)
105 }
106
107 second := Request{
108 Model: "gpt-5.6-sol",
109 Context: RequestContext{CacheSessionID: "cache-root-1", ThreadID: "thread-1", TurnID: "turn-2"},
110 Messages: append(append([]Message(nil), first.Messages...),
111 Message{Role: RoleAssistant, Content: []Content{TextBlock{Text: "part one"}}},
112 Message{Role: RoleUser, Content: []Content{TextBlock{Text: "second"}}},
113 ),
114 }
115 if stop, text := streamContinuationResponse(t, client, second); stop != StopEnd || text != "final" {
116 t.Fatalf("second response = (%q, %q), want (end, final)", stop, text)
117 }
118
119 mu.Lock()
120 defer mu.Unlock()
121 if len(requests) != 2 {
122 t.Fatalf("websocket requests = %d, want 2 on one connection", len(requests))
123 }
124 if got := requests[1]["previous_response_id"]; got != "resp_1" {
125 t.Fatalf("previous_response_id = %#v, want resp_1", got)
126 }
127 input, ok := requests[1]["input"].([]any)
128 if !ok || len(input) != 1 {
129 t.Fatalf("incremental input = %#v", requests[1]["input"])
130 }
131 encoded, _ := json.Marshal(input[0])
132 if !strings.Contains(string(encoded), "second") || strings.Contains(string(encoded), "part one") {
133 t.Fatalf("incremental input = %s", encoded)
134 }
135}
136
137// TestResponsesReasoningWebSocketTerminalBackfill pins terminal-only
138// encrypted reasoning over the shared WebSocket path: the streamed item
139// carries only an ID, the terminal response output supplies the payload,
140// and the final assistant message keeps it for full-history replay.
141func TestResponsesReasoningWebSocketTerminalBackfill(t *testing.T) {
142 server := httptest.NewServer(websocket.Handler(func(conn *websocket.Conn) {
143 defer conn.Close()
144 var payload map[string]any
145 if err := websocket.JSON.Receive(conn, &payload); err != nil {
146 return
147 }
148 response := map[string]any{
149 "id": "resp-ws-1",
150 "usage": map[string]any{"input_tokens": 1, "output_tokens": 1},
151 "output": []any{
152 map[string]any{"type": "reasoning", "id": "rs-ws-a", "encrypted_content": "blob-ws", "summary": []any{}},
153 map[string]any{"type": "message", "id": "msg-ws-1", "status": "completed", "role": "assistant", "content": []any{map[string]any{"type": "output_text", "text": "hello", "annotations": []any{}}}},
154 },
155 }
156 for _, event := range []map[string]any{
157 {"type": "response.created", "response": map[string]any{"id": "resp-ws-1"}},
158 {"type": "response.output_item.added", "output_index": 0, "item": map[string]any{"type": "reasoning", "id": "rs-ws-a"}},
159 {"type": "response.output_item.added", "output_index": 1, "item": map[string]any{"type": "message"}},
160 {"type": "response.output_text.delta", "output_index": 1, "delta": "hello"},
161 {"type": "response.completed", "response": response},
162 } {
163 _ = websocket.JSON.Send(conn, event)
164 }
165 }))
166 defer server.Close()
167
168 inner := &codexClient{
169 token: "test-token",
170 baseURL: server.URL + "/v1/responses",
171 providerName: "openai",
172 capabilities: responsesCapabilities{StablePromptCacheKey: true},
173 }
174 client := newResponsesWebSocketClient(inner)
175
176 stream, err := client.Stream(context.Background(), Request{
177 Model: "gpt-5.6-sol",
178 Context: RequestContext{CacheSessionID: "cache-ws", ThreadID: "thread-ws-reasoning", TurnID: "turn-1"},
179 Messages: []Message{{Role: RoleUser, Content: []Content{TextBlock{Text: "hi"}}}},
180 })
181 if err != nil {
182 t.Fatal(err)
183 }
184 var done EventDone
185 for event := range stream {
186 if e, ok := event.(EventDone); ok {
187 done = e
188 }
189 }
190 if done.Err != nil {
191 t.Fatalf("stream error: %v", done.Err)
192 }
193 var sawReasoning bool
194 for _, c := range done.Message.Content {
195 if rb, ok := c.(ReasoningBlock); ok && rb.ID == "rs-ws-a" && rb.Encrypted == "blob-ws" {
196 sawReasoning = true
197 }
198 }
199 if !sawReasoning {
200 t.Fatalf("websocket reasoning not backfilled: %#v", done.Message.Content)
201 }
202
203 wire, err := inner.buildRequest(Request{
204 Model: "gpt-5.6-sol",
205 Messages: []Message{
206 {Role: RoleUser, Content: []Content{TextBlock{Text: "hi"}}},
207 done.Message,
208 },
209 })
210 if err != nil {
211 t.Fatal(err)
212 }
213 var replayed bool
214 for _, item := range wire.Input {
215 if reasoning, ok := item.(codexReasoningItem); ok && reasoning.ID == "rs-ws-a" && reasoning.EncryptedContent == "blob-ws" {
216 replayed = true
217 }
218 }
219 if !replayed {
220 t.Fatalf("terminal websocket payload did not reach full-history request: %#v", wire.Input)
221 }
222}