fix(provider): retain full request on continuation mismatch
- commit
- 59535ff0d74e48c04c85803ea8657a74727c07f3
- author
- brice · · 2026-08-18 09:25 UTC
- parent
- 54f793e
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +32 -1
M packages/provider/openai_responses_websocket.go +2 -1
| @@ -467,10 +467,11 @@ func (s *responsesWebSocketSession) incrementalRequest(req Request, cacheSession | |||
| 467 | 467 | // The cached response already contains the exact assistant output from the | |
| 468 | 468 | // last generation. It must be replayed verbatim in the caller transcript | |
| 469 | 469 | // before we can safely send only the strict new tool-result/user suffix. | |
| 470 | + | fullRequest := req | |
| 470 | 471 | req.Messages = append([]Message(nil), req.Messages[len(previous.Messages):]...) | |
| 471 | 472 | if len(req.Messages) > 0 && req.Messages[0].Role == RoleAssistant { | |
| 472 | 473 | if !sameResponsesOutput(req.Messages[0], s.lastOutput) { | |
| 473 | - | return req, "" | |
| 474 | + | return fullRequest, "" | |
| 474 | 475 | } | |
| 475 | 476 | req.Messages = req.Messages[1:] | |
| 476 | 477 | } | |
| @@ -467,10 +467,11 @@ func (s *responsesWebSocketSession) incrementalRequest(req Request, cacheSession | |||||
| 467 | // The cached response already contains the exact assistant output from the | 467 | // The cached response already contains the exact assistant output from the | ||
| 468 | // last generation. It must be replayed verbatim in the caller transcript | 468 | // last generation. It must be replayed verbatim in the caller transcript | ||
| 469 | // before we can safely send only the strict new tool-result/user suffix. | 469 | // before we can safely send only the strict new tool-result/user suffix. | ||
| 470 | + | fullRequest := req | |||
| 470 | req.Messages = append([]Message(nil), req.Messages[len(previous.Messages):]...) | 471 | req.Messages = append([]Message(nil), req.Messages[len(previous.Messages):]...) | ||
| 471 | if len(req.Messages) > 0 && req.Messages[0].Role == RoleAssistant { | 472 | if len(req.Messages) > 0 && req.Messages[0].Role == RoleAssistant { | ||
| 472 | if !sameResponsesOutput(req.Messages[0], s.lastOutput) { | 473 | if !sameResponsesOutput(req.Messages[0], s.lastOutput) { | ||
| 473 | - | return req, "" | 474 | + | return fullRequest, "" |
| 474 | } | 475 | } | ||
| 475 | req.Messages = req.Messages[1:] | 476 | req.Messages = req.Messages[1:] | ||
| 476 | } | 477 | } | ||
M packages/provider/openai_responses_websocket_test.go +30 -0
| @@ -261,6 +261,36 @@ func TestResponsesWebSocketWireBaselineChangeForcesFullRequest(t *testing.T) { | |||
| 261 | 261 | } | |
| 262 | 262 | } | |
| 263 | 263 | ||
| 264 | + | func TestResponsesWebSocketAssistantMismatchForcesCompleteTranscript(t *testing.T) { | |
| 265 | + | client := newResponsesWebSocketClient(&codexClient{ | |
| 266 | + | providerName: "openai", capabilities: responsesCapabilities{StablePromptCacheKey: true}, | |
| 267 | + | }).(*responsesWebSocketClient) | |
| 268 | + | first := Request{ | |
| 269 | + | Model: "gpt-5.6-sol", Context: RequestContext{CacheSessionID: "cache-root-1", ThreadID: "thread-1", TurnID: "turn-1"}, | |
| 270 | + | Messages: []Message{{Role: RoleUser, Content: []Content{TextBlock{Text: "first"}}}}, | |
| 271 | + | } | |
| 272 | + | baseline, err := client.canonicalResponsesBaseline(first, first.Context.CacheSessionID) | |
| 273 | + | if err != nil { | |
| 274 | + | t.Fatal(err) | |
| 275 | + | } | |
| 276 | + | session := &responsesWebSocketSession{} | |
| 277 | + | session.recordCompleted(first, baseline, Message{Role: RoleAssistant, Content: []Content{TextBlock{Text: "actual"}}}, "resp_1") | |
| 278 | + | second := Request{ | |
| 279 | + | Model: first.Model, Context: RequestContext{CacheSessionID: "cache-root-1", ThreadID: "thread-1", TurnID: "turn-2"}, | |
| 280 | + | Messages: append(append([]Message(nil), first.Messages...), | |
| 281 | + | Message{Role: RoleAssistant, Content: []Content{TextBlock{Text: "modified"}}}, | |
| 282 | + | Message{Role: RoleUser, Content: []Content{TextBlock{Text: "second"}}}, | |
| 283 | + | ), | |
| 284 | + | } | |
| 285 | + | streamReq, previousResponseID := session.incrementalRequest(second, second.Context.CacheSessionID, client.canonicalResponsesBaseline) | |
| 286 | + | if previousResponseID != "" { | |
| 287 | + | t.Fatalf("previous response ID = %q, want complete request", previousResponseID) | |
| 288 | + | } | |
| 289 | + | if !reflect.DeepEqual(streamReq.Messages, second.Messages) { | |
| 290 | + | t.Fatalf("stream messages = %#v, want complete transcript %#v", streamReq.Messages, second.Messages) | |
| 291 | + | } | |
| 292 | + | } | |
| 293 | + | ||
| 264 | 294 | func TestResponsesWebSocketCancellationClosesOnlyActiveConnection(t *testing.T) { | |
| 265 | 295 | closed := make(chan struct{}) | |
| 266 | 296 | var closeOnce sync.Once | |
| @@ -261,6 +261,36 @@ func TestResponsesWebSocketWireBaselineChangeForcesFullRequest(t *testing.T) { | |||||
| 261 | } | 261 | } | ||
| 262 | } | 262 | } | ||
| 263 | 263 | ||||
| 264 | + | func TestResponsesWebSocketAssistantMismatchForcesCompleteTranscript(t *testing.T) { | |||
| 265 | + | client := newResponsesWebSocketClient(&codexClient{ | |||
| 266 | + | providerName: "openai", capabilities: responsesCapabilities{StablePromptCacheKey: true}, | |||
| 267 | + | }).(*responsesWebSocketClient) | |||
| 268 | + | first := Request{ | |||
| 269 | + | Model: "gpt-5.6-sol", Context: RequestContext{CacheSessionID: "cache-root-1", ThreadID: "thread-1", TurnID: "turn-1"}, | |||
| 270 | + | Messages: []Message{{Role: RoleUser, Content: []Content{TextBlock{Text: "first"}}}}, | |||
| 271 | + | } | |||
| 272 | + | baseline, err := client.canonicalResponsesBaseline(first, first.Context.CacheSessionID) | |||
| 273 | + | if err != nil { | |||
| 274 | + | t.Fatal(err) | |||
| 275 | + | } | |||
| 276 | + | session := &responsesWebSocketSession{} | |||
| 277 | + | session.recordCompleted(first, baseline, Message{Role: RoleAssistant, Content: []Content{TextBlock{Text: "actual"}}}, "resp_1") | |||
| 278 | + | second := Request{ | |||
| 279 | + | Model: first.Model, Context: RequestContext{CacheSessionID: "cache-root-1", ThreadID: "thread-1", TurnID: "turn-2"}, | |||
| 280 | + | Messages: append(append([]Message(nil), first.Messages...), | |||
| 281 | + | Message{Role: RoleAssistant, Content: []Content{TextBlock{Text: "modified"}}}, | |||
| 282 | + | Message{Role: RoleUser, Content: []Content{TextBlock{Text: "second"}}}, | |||
| 283 | + | ), | |||
| 284 | + | } | |||
| 285 | + | streamReq, previousResponseID := session.incrementalRequest(second, second.Context.CacheSessionID, client.canonicalResponsesBaseline) | |||
| 286 | + | if previousResponseID != "" { | |||
| 287 | + | t.Fatalf("previous response ID = %q, want complete request", previousResponseID) | |||
| 288 | + | } | |||
| 289 | + | if !reflect.DeepEqual(streamReq.Messages, second.Messages) { | |||
| 290 | + | t.Fatalf("stream messages = %#v, want complete transcript %#v", streamReq.Messages, second.Messages) | |||
| 291 | + | } | |||
| 292 | + | } | |||
| 293 | + | ||||
| 264 | func TestResponsesWebSocketCancellationClosesOnlyActiveConnection(t *testing.T) { | 294 | func TestResponsesWebSocketCancellationClosesOnlyActiveConnection(t *testing.T) { | ||
| 265 | closed := make(chan struct{}) | 295 | closed := make(chan struct{}) | ||
| 266 | var closeOnce sync.Once | 296 | var closeOnce sync.Once | ||