fix(provider): gate Responses cache continuation
- commit
- 54f793e779445fbf1298698eb4994f90591f0fd8
- author
- brice · · 2026-08-18 09:21 UTC
- parent
- c21eb9c
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
6 files · +105 -8
M docs/providers.md +4 -2
| @@ -159,8 +159,10 @@ route uses the root identity as `prompt_cache_key`; declared Codex routes also | |||
| 159 | 159 | receive their documented session and thread routing headers. A generic | |
| 160 | 160 | OpenAI-compatible endpoint never receives these extensions merely because it | |
| 161 | 161 | accepts an OpenAI-shaped request. JSON event mode also reports sanitized cache | |
| 162 | - | diagnostics (`eligible`, `mode`, `transport`, and `continuation`). These | |
| 163 | - | records never include prompts, durable IDs, request bodies, or credentials. | |
| 162 | + | diagnostics (`eligible`, `mode`, `transport`, and `continuation`). `eligible` | |
| 163 | + | is a conservative local prompt-length estimate; OpenAI performs the authoritative | |
| 164 | + | tokenization and cache decision. These records never include prompts, durable | |
| 165 | + | IDs, request bodies, or credentials. | |
| 164 | 166 | ||
| 165 | 167 | When a provider reports cache detail, zut displays cache-read hit rate as | |
| 166 | 168 | `cache reads / cache-reporting prompt input`. A shown `0%` is a confirmed | |
| @@ -159,8 +159,10 @@ route uses the root identity as `prompt_cache_key`; declared Codex routes also | |||||
| 159 | receive their documented session and thread routing headers. A generic | 159 | receive their documented session and thread routing headers. A generic | ||
| 160 | OpenAI-compatible endpoint never receives these extensions merely because it | 160 | OpenAI-compatible endpoint never receives these extensions merely because it | ||
| 161 | accepts an OpenAI-shaped request. JSON event mode also reports sanitized cache | 161 | accepts an OpenAI-shaped request. JSON event mode also reports sanitized cache | ||
| 162 | - | diagnostics (`eligible`, `mode`, `transport`, and `continuation`). These | 162 | + | diagnostics (`eligible`, `mode`, `transport`, and `continuation`). `eligible` |
| 163 | - | records never include prompts, durable IDs, request bodies, or credentials. | 163 | + | is a conservative local prompt-length estimate; OpenAI performs the authoritative |
| 164 | + | tokenization and cache decision. These records never include prompts, durable | |||
| 165 | + | IDs, request bodies, or credentials. | |||
| 164 | 166 | ||||
| 165 | When a provider reports cache detail, zut displays cache-read hit rate as | 167 | When a provider reports cache detail, zut displays cache-read hit rate as | ||
| 166 | `cache reads / cache-reporting prompt input`. A shown `0%` is a confirmed | 168 | `cache reads / cache-reporting prompt input`. A shown `0%` is a confirmed | ||
M docs/rpc.md +1 -1
| @@ -193,7 +193,7 @@ Stream notifications during a `prompt` or `compact`. None carry an `id`. | |||
| 193 | 193 | |---|---|---| | |
| 194 | 194 | | `turn_start` | `step` | Beginning of one model call (max-steps loop iteration) | | |
| 195 | 195 | | `request_started` | `provider`, `model`, `scope`, `attempt`, `max_attempts` | A provider (`scope=provider`) or agent (`scope=agent`) request attempt started | | |
| 196 | - | | `cache_diagnostics` | `eligible`, `mode`, `transport`, `continuation` | Sanitized cache-routing state; it never includes prompts, IDs, request bodies, or credentials | | |
| 196 | + | | `cache_diagnostics` | `eligible`, `mode`, `transport`, `continuation` | Sanitized cache-routing state. `eligible` is a conservative local prompt-length estimate; it never includes prompts, IDs, request bodies, or credentials | | |
| 197 | 197 | | `retry_scheduled` | `scope`, `attempt`, `max_attempts`, `delay_ms` | The upcoming attempt is delayed; `delay_ms` is a non-negative integer | | |
| 198 | 198 | | `user_message` | `content`, `time` | The submitted prompt as it was added to the transcript | | |
| 199 | 199 | | `assistant_start` | (none) | Provider connection opened; waiting for assistant streaming | | |
| @@ -193,7 +193,7 @@ Stream notifications during a `prompt` or `compact`. None carry an `id`. | |||||
| 193 | |---|---|---| | 193 | |---|---|---| | ||
| 194 | | `turn_start` | `step` | Beginning of one model call (max-steps loop iteration) | | 194 | | `turn_start` | `step` | Beginning of one model call (max-steps loop iteration) | | ||
| 195 | | `request_started` | `provider`, `model`, `scope`, `attempt`, `max_attempts` | A provider (`scope=provider`) or agent (`scope=agent`) request attempt started | | 195 | | `request_started` | `provider`, `model`, `scope`, `attempt`, `max_attempts` | A provider (`scope=provider`) or agent (`scope=agent`) request attempt started | | ||
| 196 | - | | `cache_diagnostics` | `eligible`, `mode`, `transport`, `continuation` | Sanitized cache-routing state; it never includes prompts, IDs, request bodies, or credentials | | 196 | + | | `cache_diagnostics` | `eligible`, `mode`, `transport`, `continuation` | Sanitized cache-routing state. `eligible` is a conservative local prompt-length estimate; it never includes prompts, IDs, request bodies, or credentials | |
| 197 | | `retry_scheduled` | `scope`, `attempt`, `max_attempts`, `delay_ms` | The upcoming attempt is delayed; `delay_ms` is a non-negative integer | | 197 | | `retry_scheduled` | `scope`, `attempt`, `max_attempts`, `delay_ms` | The upcoming attempt is delayed; `delay_ms` is a non-negative integer | | ||
| 198 | | `user_message` | `content`, `time` | The submitted prompt as it was added to the transcript | | 198 | | `user_message` | `content`, `time` | The submitted prompt as it was added to the transcript | | ||
| 199 | | `assistant_start` | (none) | Provider connection opened; waiting for assistant streaming | | 199 | | `assistant_start` | (none) | Provider connection opened; waiting for assistant streaming | | ||
M packages/provider/openai_codex.go +14 -1
| @@ -399,16 +399,29 @@ func splitCallID(id string) (string, string) { | |||
| 399 | 399 | return id, "" | |
| 400 | 400 | } | |
| 401 | 401 | ||
| 402 | + | const minOpenAIPromptCacheTokens = 1024 | |
| 403 | + | ||
| 402 | 404 | func responsesCacheDiagnostics(wire *codexRequest, transport, continuation string) CacheDiagnostics { | |
| 403 | 405 | diagnostics := CacheDiagnostics{Mode: "none", Transport: transport, Continuation: continuation} | |
| 404 | 406 | if wire == nil || wire.PromptCacheOptions == nil { | |
| 405 | 407 | return diagnostics | |
| 406 | 408 | } | |
| 407 | - | diagnostics.Eligible = true | |
| 408 | 409 | diagnostics.Mode = wire.PromptCacheOptions.Mode | |
| 410 | + | diagnostics.Eligible = estimatedOpenAIPromptTokens(wire) >= minOpenAIPromptCacheTokens | |
| 409 | 411 | return diagnostics | |
| 410 | 412 | } | |
| 411 | 413 | ||
| 414 | + | // estimatedOpenAIPromptTokens supplies a deliberately conservative local | |
| 415 | + | // eligibility signal. OpenAI makes the final tokenization and cache decision; | |
| 416 | + | // this estimate never leaves the process and is not persisted in diagnostics. | |
| 417 | + | func estimatedOpenAIPromptTokens(wire *codexRequest) int { | |
| 418 | + | encoded, err := json.Marshal(wire) | |
| 419 | + | if err != nil { | |
| 420 | + | return 0 | |
| 421 | + | } | |
| 422 | + | return len(encoded) / 4 | |
| 423 | + | } | |
| 424 | + | ||
| 412 | 425 | func usesCodexCLIRouting(model string) bool { | |
| 413 | 426 | switch model { | |
| 414 | 427 | case "gpt-5.6-luna", "gpt-5.6-luna-pro", "gpt-5.6-terra", "gpt-5.6-terra-pro": | |
| @@ -399,16 +399,29 @@ func splitCallID(id string) (string, string) { | |||||
| 399 | return id, "" | 399 | return id, "" | ||
| 400 | } | 400 | } | ||
| 401 | 401 | ||||
| 402 | + | const minOpenAIPromptCacheTokens = 1024 | |||
| 403 | + | ||||
| 402 | func responsesCacheDiagnostics(wire *codexRequest, transport, continuation string) CacheDiagnostics { | 404 | func responsesCacheDiagnostics(wire *codexRequest, transport, continuation string) CacheDiagnostics { | ||
| 403 | diagnostics := CacheDiagnostics{Mode: "none", Transport: transport, Continuation: continuation} | 405 | diagnostics := CacheDiagnostics{Mode: "none", Transport: transport, Continuation: continuation} | ||
| 404 | if wire == nil || wire.PromptCacheOptions == nil { | 406 | if wire == nil || wire.PromptCacheOptions == nil { | ||
| 405 | return diagnostics | 407 | return diagnostics | ||
| 406 | } | 408 | } | ||
| 407 | - | diagnostics.Eligible = true | |||
| 408 | diagnostics.Mode = wire.PromptCacheOptions.Mode | 409 | diagnostics.Mode = wire.PromptCacheOptions.Mode | ||
| 410 | + | diagnostics.Eligible = estimatedOpenAIPromptTokens(wire) >= minOpenAIPromptCacheTokens | |||
| 409 | return diagnostics | 411 | return diagnostics | ||
| 410 | } | 412 | } | ||
| 411 | 413 | ||||
| 414 | + | // estimatedOpenAIPromptTokens supplies a deliberately conservative local | |||
| 415 | + | // eligibility signal. OpenAI makes the final tokenization and cache decision; | |||
| 416 | + | // this estimate never leaves the process and is not persisted in diagnostics. | |||
| 417 | + | func estimatedOpenAIPromptTokens(wire *codexRequest) int { | |||
| 418 | + | encoded, err := json.Marshal(wire) | |||
| 419 | + | if err != nil { | |||
| 420 | + | return 0 | |||
| 421 | + | } | |||
| 422 | + | return len(encoded) / 4 | |||
| 423 | + | } | |||
| 424 | + | ||||
| 412 | func usesCodexCLIRouting(model string) bool { | 425 | func usesCodexCLIRouting(model string) bool { | ||
| 413 | switch model { | 426 | switch model { | ||
| 414 | case "gpt-5.6-luna", "gpt-5.6-luna-pro", "gpt-5.6-terra", "gpt-5.6-terra-pro": | 427 | case "gpt-5.6-luna", "gpt-5.6-luna-pro", "gpt-5.6-terra", "gpt-5.6-terra-pro": | ||
M packages/provider/openai_codex_test.go +17 -0
| @@ -13,6 +13,23 @@ import ( | |||
| 13 | 13 | "golang.org/x/net/websocket" | |
| 14 | 14 | ) | |
| 15 | 15 | ||
| 16 | + | func TestResponsesCacheDiagnosticsRequireEstimatedThreshold(t *testing.T) { | |
| 17 | + | newWire := func(text string) *codexRequest { | |
| 18 | + | return &codexRequest{ | |
| 19 | + | PromptCacheOptions: &codexPromptCacheOptions{Mode: "implicit"}, | |
| 20 | + | Input: []any{codexInputMessage{Role: "user", Content: []any{ | |
| 21 | + | codexInputText{Type: "input_text", Text: text}, | |
| 22 | + | }}}, | |
| 23 | + | } | |
| 24 | + | } | |
| 25 | + | if diagnostics := responsesCacheDiagnostics(newWire(strings.Repeat("a", 64)), "http_sse", "full"); diagnostics.Eligible { | |
| 26 | + | t.Fatalf("short request diagnostics = %#v, want ineligible", diagnostics) | |
| 27 | + | } | |
| 28 | + | if diagnostics := responsesCacheDiagnostics(newWire(strings.Repeat("a", 5_000)), "http_sse", "full"); !diagnostics.Eligible { | |
| 29 | + | t.Fatalf("long request diagnostics = %#v, want eligible", diagnostics) | |
| 30 | + | } | |
| 31 | + | } | |
| 32 | + | ||
| 16 | 33 | // An image-only tool result must not serialize to an empty | |
| 17 | 34 | // function_call_output (the Responses API may reject it) and a | |
| 18 | 35 | // following user-message image must serialize as input_image so the | |
| @@ -13,6 +13,23 @@ import ( | |||||
| 13 | "golang.org/x/net/websocket" | 13 | "golang.org/x/net/websocket" | ||
| 14 | ) | 14 | ) | ||
| 15 | 15 | ||||
| 16 | + | func TestResponsesCacheDiagnosticsRequireEstimatedThreshold(t *testing.T) { | |||
| 17 | + | newWire := func(text string) *codexRequest { | |||
| 18 | + | return &codexRequest{ | |||
| 19 | + | PromptCacheOptions: &codexPromptCacheOptions{Mode: "implicit"}, | |||
| 20 | + | Input: []any{codexInputMessage{Role: "user", Content: []any{ | |||
| 21 | + | codexInputText{Type: "input_text", Text: text}, | |||
| 22 | + | }}}, | |||
| 23 | + | } | |||
| 24 | + | } | |||
| 25 | + | if diagnostics := responsesCacheDiagnostics(newWire(strings.Repeat("a", 64)), "http_sse", "full"); diagnostics.Eligible { | |||
| 26 | + | t.Fatalf("short request diagnostics = %#v, want ineligible", diagnostics) | |||
| 27 | + | } | |||
| 28 | + | if diagnostics := responsesCacheDiagnostics(newWire(strings.Repeat("a", 5_000)), "http_sse", "full"); !diagnostics.Eligible { | |||
| 29 | + | t.Fatalf("long request diagnostics = %#v, want eligible", diagnostics) | |||
| 30 | + | } | |||
| 31 | + | } | |||
| 32 | + | ||||
| 16 | // An image-only tool result must not serialize to an empty | 33 | // An image-only tool result must not serialize to an empty | ||
| 17 | // function_call_output (the Responses API may reject it) and a | 34 | // function_call_output (the Responses API may reject it) and a | ||
| 18 | // following user-message image must serialize as input_image so the | 35 | // following user-message image must serialize as input_image so the | ||
M packages/provider/openai_responses_websocket.go +31 -4
| @@ -2,7 +2,9 @@ package provider | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "crypto/sha256" | |
| 5 | 6 | "crypto/tls" | |
| 7 | + | "encoding/hex" | |
| 6 | 8 | "encoding/json" | |
| 7 | 9 | "fmt" | |
| 8 | 10 | "net/http" | |
| @@ -40,6 +42,7 @@ type responsesWebSocketSession struct { | |||
| 40 | 42 | // lastRequest is the complete request accepted by the server. Continuation | |
| 41 | 43 | // must compare every response-context setting, not merely its transcript. | |
| 42 | 44 | lastRequest Request | |
| 45 | + | lastBaseline string | |
| 43 | 46 | lastOutput Message | |
| 44 | 47 | lastResponseID string | |
| 45 | 48 | } | |
| @@ -179,7 +182,11 @@ func (c *responsesWebSocketClient) Stream(ctx context.Context, req Request) (<-c | |||
| 179 | 182 | } | |
| 180 | 183 | }() | |
| 181 | 184 | ||
| 182 | - | streamReq, previousResponseID := session.incrementalRequest(req) | |
| 185 | + | fullBaseline, err := c.canonicalResponsesBaseline(req, cacheSessionID) | |
| 186 | + | if err != nil { | |
| 187 | + | return nil, err | |
| 188 | + | } | |
| 189 | + | streamReq, previousResponseID := session.incrementalRequest(req, cacheSessionID, c.canonicalResponsesBaseline) | |
| 183 | 190 | payload, err := c.websocketPayload(streamReq, cacheSessionID, previousResponseID) | |
| 184 | 191 | if err == nil { | |
| 185 | 192 | wire, wireErr := c.http.buildRequest(streamReq) | |
| @@ -259,7 +266,7 @@ func (c *responsesWebSocketClient) Stream(ctx context.Context, req Request) (<-c | |||
| 259 | 266 | terminal = true | |
| 260 | 267 | stateMu.Unlock() | |
| 261 | 268 | completed = true | |
| 262 | - | session.recordCompleted(req, output, responseID) | |
| 269 | + | session.recordCompleted(req, fullBaseline, output, responseID) | |
| 263 | 270 | }) | |
| 264 | 271 | close(finished) | |
| 265 | 272 | if !completed { | |
| @@ -269,6 +276,18 @@ func (c *responsesWebSocketClient) Stream(ctx context.Context, req Request) (<-c | |||
| 269 | 276 | return out, nil | |
| 270 | 277 | } | |
| 271 | 278 | ||
| 279 | + | // canonicalResponsesBaseline hashes the complete, encoded WebSocket request | |
| 280 | + | // without a server response ID. The digest lets continuation compare every | |
| 281 | + | // wire-visible setting without retaining prompts, IDs, or credentials. | |
| 282 | + | func (c *responsesWebSocketClient) canonicalResponsesBaseline(req Request, cacheSessionID string) (string, error) { | |
| 283 | + | payload, err := c.websocketPayload(req, cacheSessionID, "") | |
| 284 | + | if err != nil { | |
| 285 | + | return "", err | |
| 286 | + | } | |
| 287 | + | digest := sha256.Sum256(payload) | |
| 288 | + | return hex.EncodeToString(digest[:]), nil | |
| 289 | + | } | |
| 290 | + | ||
| 272 | 291 | func (c *responsesWebSocketClient) websocketPayload(req Request, cacheSessionID, previousResponseID string) ([]byte, error) { | |
| 273 | 292 | wire, err := c.http.buildRequest(req) | |
| 274 | 293 | if err != nil { | |
| @@ -432,13 +451,19 @@ func (c *responsesWebSocketClient) forwardResponsesHTTP(ctx context.Context, req | |||
| 432 | 451 | } | |
| 433 | 452 | } | |
| 434 | 453 | ||
| 435 | - | func (s *responsesWebSocketSession) incrementalRequest(req Request) (Request, string) { | |
| 454 | + | func (s *responsesWebSocketSession) incrementalRequest(req Request, cacheSessionID string, canonicalBaseline func(Request, string) (string, error)) (Request, string) { | |
| 436 | 455 | s.mu.Lock() | |
| 437 | 456 | defer s.mu.Unlock() | |
| 438 | 457 | previous := s.lastRequest | |
| 439 | 458 | if s.lastResponseID == "" || !sameResponsesContinuationConfig(previous, req) || len(req.Messages) < len(previous.Messages) || !reflect.DeepEqual(req.Messages[:len(previous.Messages)], previous.Messages) { | |
| 440 | 459 | return req, "" | |
| 441 | 460 | } | |
| 461 | + | prefix := req | |
| 462 | + | prefix.Messages = append([]Message(nil), req.Messages[:len(previous.Messages)]...) | |
| 463 | + | baseline, err := canonicalBaseline(prefix, cacheSessionID) | |
| 464 | + | if err != nil || baseline != s.lastBaseline { | |
| 465 | + | return req, "" | |
| 466 | + | } | |
| 442 | 467 | // The cached response already contains the exact assistant output from the | |
| 443 | 468 | // last generation. It must be replayed verbatim in the caller transcript | |
| 444 | 469 | // before we can safely send only the strict new tool-result/user suffix. | |
| @@ -499,12 +524,13 @@ func (s *responsesWebSocketSession) connection() (*websocket.Conn, <-chan sseEve | |||
| 499 | 524 | return s.conn, s.raw | |
| 500 | 525 | } | |
| 501 | 526 | ||
| 502 | - | func (s *responsesWebSocketSession) recordCompleted(request Request, output Message, responseID string) { | |
| 527 | + | func (s *responsesWebSocketSession) recordCompleted(request Request, baseline string, output Message, responseID string) { | |
| 503 | 528 | s.mu.Lock() | |
| 504 | 529 | defer s.mu.Unlock() | |
| 505 | 530 | request.Messages = append([]Message(nil), request.Messages...) | |
| 506 | 531 | request.Lifecycle = nil | |
| 507 | 532 | s.lastRequest = request | |
| 533 | + | s.lastBaseline = baseline | |
| 508 | 534 | s.lastOutput = output | |
| 509 | 535 | s.lastResponseID = responseID | |
| 510 | 536 | } | |
| @@ -535,6 +561,7 @@ func (s *responsesWebSocketSession) invalidateLocked() { | |||
| 535 | 561 | s.raw = nil | |
| 536 | 562 | s.readerDone = nil | |
| 537 | 563 | s.lastRequest = Request{} | |
| 564 | + | s.lastBaseline = "" | |
| 538 | 565 | s.lastOutput = Message{} | |
| 539 | 566 | s.lastResponseID = "" | |
| 540 | 567 | } | |
| @@ -2,7 +2,9 @@ package provider | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "crypto/sha256" | |||
| 5 | "crypto/tls" | 6 | "crypto/tls" | ||
| 7 | + | "encoding/hex" | |||
| 6 | "encoding/json" | 8 | "encoding/json" | ||
| 7 | "fmt" | 9 | "fmt" | ||
| 8 | "net/http" | 10 | "net/http" | ||
| @@ -40,6 +42,7 @@ type responsesWebSocketSession struct { | |||||
| 40 | // lastRequest is the complete request accepted by the server. Continuation | 42 | // lastRequest is the complete request accepted by the server. Continuation | ||
| 41 | // must compare every response-context setting, not merely its transcript. | 43 | // must compare every response-context setting, not merely its transcript. | ||
| 42 | lastRequest Request | 44 | lastRequest Request | ||
| 45 | + | lastBaseline string | |||
| 43 | lastOutput Message | 46 | lastOutput Message | ||
| 44 | lastResponseID string | 47 | lastResponseID string | ||
| 45 | } | 48 | } | ||
| @@ -179,7 +182,11 @@ func (c *responsesWebSocketClient) Stream(ctx context.Context, req Request) (<-c | |||||
| 179 | } | 182 | } | ||
| 180 | }() | 183 | }() | ||
| 181 | 184 | ||||
| 182 | - | streamReq, previousResponseID := session.incrementalRequest(req) | 185 | + | fullBaseline, err := c.canonicalResponsesBaseline(req, cacheSessionID) |
| 186 | + | if err != nil { | |||
| 187 | + | return nil, err | |||
| 188 | + | } | |||
| 189 | + | streamReq, previousResponseID := session.incrementalRequest(req, cacheSessionID, c.canonicalResponsesBaseline) | |||
| 183 | payload, err := c.websocketPayload(streamReq, cacheSessionID, previousResponseID) | 190 | payload, err := c.websocketPayload(streamReq, cacheSessionID, previousResponseID) | ||
| 184 | if err == nil { | 191 | if err == nil { | ||
| 185 | wire, wireErr := c.http.buildRequest(streamReq) | 192 | wire, wireErr := c.http.buildRequest(streamReq) | ||
| @@ -259,7 +266,7 @@ func (c *responsesWebSocketClient) Stream(ctx context.Context, req Request) (<-c | |||||
| 259 | terminal = true | 266 | terminal = true | ||
| 260 | stateMu.Unlock() | 267 | stateMu.Unlock() | ||
| 261 | completed = true | 268 | completed = true | ||
| 262 | - | session.recordCompleted(req, output, responseID) | 269 | + | session.recordCompleted(req, fullBaseline, output, responseID) |
| 263 | }) | 270 | }) | ||
| 264 | close(finished) | 271 | close(finished) | ||
| 265 | if !completed { | 272 | if !completed { | ||
| @@ -269,6 +276,18 @@ func (c *responsesWebSocketClient) Stream(ctx context.Context, req Request) (<-c | |||||
| 269 | return out, nil | 276 | return out, nil | ||
| 270 | } | 277 | } | ||
| 271 | 278 | ||||
| 279 | + | // canonicalResponsesBaseline hashes the complete, encoded WebSocket request | |||
| 280 | + | // without a server response ID. The digest lets continuation compare every | |||
| 281 | + | // wire-visible setting without retaining prompts, IDs, or credentials. | |||
| 282 | + | func (c *responsesWebSocketClient) canonicalResponsesBaseline(req Request, cacheSessionID string) (string, error) { | |||
| 283 | + | payload, err := c.websocketPayload(req, cacheSessionID, "") | |||
| 284 | + | if err != nil { | |||
| 285 | + | return "", err | |||
| 286 | + | } | |||
| 287 | + | digest := sha256.Sum256(payload) | |||
| 288 | + | return hex.EncodeToString(digest[:]), nil | |||
| 289 | + | } | |||
| 290 | + | ||||
| 272 | func (c *responsesWebSocketClient) websocketPayload(req Request, cacheSessionID, previousResponseID string) ([]byte, error) { | 291 | func (c *responsesWebSocketClient) websocketPayload(req Request, cacheSessionID, previousResponseID string) ([]byte, error) { | ||
| 273 | wire, err := c.http.buildRequest(req) | 292 | wire, err := c.http.buildRequest(req) | ||
| 274 | if err != nil { | 293 | if err != nil { | ||
| @@ -432,13 +451,19 @@ func (c *responsesWebSocketClient) forwardResponsesHTTP(ctx context.Context, req | |||||
| 432 | } | 451 | } | ||
| 433 | } | 452 | } | ||
| 434 | 453 | ||||
| 435 | - | func (s *responsesWebSocketSession) incrementalRequest(req Request) (Request, string) { | 454 | + | func (s *responsesWebSocketSession) incrementalRequest(req Request, cacheSessionID string, canonicalBaseline func(Request, string) (string, error)) (Request, string) { |
| 436 | s.mu.Lock() | 455 | s.mu.Lock() | ||
| 437 | defer s.mu.Unlock() | 456 | defer s.mu.Unlock() | ||
| 438 | previous := s.lastRequest | 457 | previous := s.lastRequest | ||
| 439 | if s.lastResponseID == "" || !sameResponsesContinuationConfig(previous, req) || len(req.Messages) < len(previous.Messages) || !reflect.DeepEqual(req.Messages[:len(previous.Messages)], previous.Messages) { | 458 | if s.lastResponseID == "" || !sameResponsesContinuationConfig(previous, req) || len(req.Messages) < len(previous.Messages) || !reflect.DeepEqual(req.Messages[:len(previous.Messages)], previous.Messages) { | ||
| 440 | return req, "" | 459 | return req, "" | ||
| 441 | } | 460 | } | ||
| 461 | + | prefix := req | |||
| 462 | + | prefix.Messages = append([]Message(nil), req.Messages[:len(previous.Messages)]...) | |||
| 463 | + | baseline, err := canonicalBaseline(prefix, cacheSessionID) | |||
| 464 | + | if err != nil || baseline != s.lastBaseline { | |||
| 465 | + | return req, "" | |||
| 466 | + | } | |||
| 442 | // The cached response already contains the exact assistant output from the | 467 | // The cached response already contains the exact assistant output from the | ||
| 443 | // last generation. It must be replayed verbatim in the caller transcript | 468 | // last generation. It must be replayed verbatim in the caller transcript | ||
| 444 | // 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. | ||
| @@ -499,12 +524,13 @@ func (s *responsesWebSocketSession) connection() (*websocket.Conn, <-chan sseEve | |||||
| 499 | return s.conn, s.raw | 524 | return s.conn, s.raw | ||
| 500 | } | 525 | } | ||
| 501 | 526 | ||||
| 502 | - | func (s *responsesWebSocketSession) recordCompleted(request Request, output Message, responseID string) { | 527 | + | func (s *responsesWebSocketSession) recordCompleted(request Request, baseline string, output Message, responseID string) { |
| 503 | s.mu.Lock() | 528 | s.mu.Lock() | ||
| 504 | defer s.mu.Unlock() | 529 | defer s.mu.Unlock() | ||
| 505 | request.Messages = append([]Message(nil), request.Messages...) | 530 | request.Messages = append([]Message(nil), request.Messages...) | ||
| 506 | request.Lifecycle = nil | 531 | request.Lifecycle = nil | ||
| 507 | s.lastRequest = request | 532 | s.lastRequest = request | ||
| 533 | + | s.lastBaseline = baseline | |||
| 508 | s.lastOutput = output | 534 | s.lastOutput = output | ||
| 509 | s.lastResponseID = responseID | 535 | s.lastResponseID = responseID | ||
| 510 | } | 536 | } | ||
| @@ -535,6 +561,7 @@ func (s *responsesWebSocketSession) invalidateLocked() { | |||||
| 535 | s.raw = nil | 561 | s.raw = nil | ||
| 536 | s.readerDone = nil | 562 | s.readerDone = nil | ||
| 537 | s.lastRequest = Request{} | 563 | s.lastRequest = Request{} | ||
| 564 | + | s.lastBaseline = "" | |||
| 538 | s.lastOutput = Message{} | 565 | s.lastOutput = Message{} | ||
| 539 | s.lastResponseID = "" | 566 | s.lastResponseID = "" | ||
| 540 | } | 567 | } | ||
M packages/provider/openai_responses_websocket_test.go +38 -0
| @@ -5,6 +5,7 @@ import ( | |||
| 5 | 5 | "encoding/json" | |
| 6 | 6 | "net/http" | |
| 7 | 7 | "net/http/httptest" | |
| 8 | + | "reflect" | |
| 8 | 9 | "strings" | |
| 9 | 10 | "sync" | |
| 10 | 11 | "testing" | |
| @@ -223,6 +224,43 @@ func TestResponsesWebSocketReusesSessionAndSendsIncrementalInput(t *testing.T) { | |||
| 223 | 224 | } | |
| 224 | 225 | } | |
| 225 | 226 | ||
| 227 | + | func TestResponsesWebSocketWireBaselineChangeForcesFullRequest(t *testing.T) { | |
| 228 | + | mappedModel := "gpt-5.6-sol-a" | |
| 229 | + | client := newResponsesWebSocketClient(&codexClient{ | |
| 230 | + | providerName: "openai", | |
| 231 | + | capabilities: responsesCapabilities{StablePromptCacheKey: true}, | |
| 232 | + | modelName: func(string) string { return mappedModel }, | |
| 233 | + | }).(*responsesWebSocketClient) | |
| 234 | + | first := Request{ | |
| 235 | + | Model: "gpt-5.6-sol", | |
| 236 | + | Context: RequestContext{CacheSessionID: "cache-root-1", ThreadID: "thread-1", TurnID: "turn-1"}, | |
| 237 | + | Messages: []Message{{Role: RoleUser, Content: []Content{TextBlock{Text: "first"}}}}, | |
| 238 | + | } | |
| 239 | + | baseline, err := client.canonicalResponsesBaseline(first, first.Context.CacheSessionID) | |
| 240 | + | if err != nil { | |
| 241 | + | t.Fatal(err) | |
| 242 | + | } | |
| 243 | + | session := &responsesWebSocketSession{} | |
| 244 | + | session.recordCompleted(first, baseline, Message{Role: RoleAssistant, Content: []Content{TextBlock{Text: "ok"}}}, "resp_1") | |
| 245 | + | ||
| 246 | + | mappedModel = "gpt-5.6-sol-b" | |
| 247 | + | second := Request{ | |
| 248 | + | Model: first.Model, | |
| 249 | + | Context: RequestContext{CacheSessionID: "cache-root-1", ThreadID: "thread-1", TurnID: "turn-2"}, | |
| 250 | + | Messages: append(append([]Message(nil), first.Messages...), | |
| 251 | + | Message{Role: RoleAssistant, Content: []Content{TextBlock{Text: "ok"}}}, | |
| 252 | + | Message{Role: RoleUser, Content: []Content{TextBlock{Text: "second"}}}, | |
| 253 | + | ), | |
| 254 | + | } | |
| 255 | + | streamReq, previousResponseID := session.incrementalRequest(second, second.Context.CacheSessionID, client.canonicalResponsesBaseline) | |
| 256 | + | if previousResponseID != "" { | |
| 257 | + | t.Fatalf("previous response ID = %q, want full request after wire baseline change", previousResponseID) | |
| 258 | + | } | |
| 259 | + | if !reflect.DeepEqual(streamReq.Messages, second.Messages) { | |
| 260 | + | t.Fatalf("stream messages = %#v, want complete transcript %#v", streamReq.Messages, second.Messages) | |
| 261 | + | } | |
| 262 | + | } | |
| 263 | + | ||
| 226 | 264 | func TestResponsesWebSocketCancellationClosesOnlyActiveConnection(t *testing.T) { | |
| 227 | 265 | closed := make(chan struct{}) | |
| 228 | 266 | var closeOnce sync.Once | |
| @@ -5,6 +5,7 @@ import ( | |||||
| 5 | "encoding/json" | 5 | "encoding/json" | ||
| 6 | "net/http" | 6 | "net/http" | ||
| 7 | "net/http/httptest" | 7 | "net/http/httptest" | ||
| 8 | + | "reflect" | |||
| 8 | "strings" | 9 | "strings" | ||
| 9 | "sync" | 10 | "sync" | ||
| 10 | "testing" | 11 | "testing" | ||
| @@ -223,6 +224,43 @@ func TestResponsesWebSocketReusesSessionAndSendsIncrementalInput(t *testing.T) { | |||||
| 223 | } | 224 | } | ||
| 224 | } | 225 | } | ||
| 225 | 226 | ||||
| 227 | + | func TestResponsesWebSocketWireBaselineChangeForcesFullRequest(t *testing.T) { | |||
| 228 | + | mappedModel := "gpt-5.6-sol-a" | |||
| 229 | + | client := newResponsesWebSocketClient(&codexClient{ | |||
| 230 | + | providerName: "openai", | |||
| 231 | + | capabilities: responsesCapabilities{StablePromptCacheKey: true}, | |||
| 232 | + | modelName: func(string) string { return mappedModel }, | |||
| 233 | + | }).(*responsesWebSocketClient) | |||
| 234 | + | first := Request{ | |||
| 235 | + | Model: "gpt-5.6-sol", | |||
| 236 | + | Context: RequestContext{CacheSessionID: "cache-root-1", ThreadID: "thread-1", TurnID: "turn-1"}, | |||
| 237 | + | Messages: []Message{{Role: RoleUser, Content: []Content{TextBlock{Text: "first"}}}}, | |||
| 238 | + | } | |||
| 239 | + | baseline, err := client.canonicalResponsesBaseline(first, first.Context.CacheSessionID) | |||
| 240 | + | if err != nil { | |||
| 241 | + | t.Fatal(err) | |||
| 242 | + | } | |||
| 243 | + | session := &responsesWebSocketSession{} | |||
| 244 | + | session.recordCompleted(first, baseline, Message{Role: RoleAssistant, Content: []Content{TextBlock{Text: "ok"}}}, "resp_1") | |||
| 245 | + | ||||
| 246 | + | mappedModel = "gpt-5.6-sol-b" | |||
| 247 | + | second := Request{ | |||
| 248 | + | Model: first.Model, | |||
| 249 | + | Context: RequestContext{CacheSessionID: "cache-root-1", ThreadID: "thread-1", TurnID: "turn-2"}, | |||
| 250 | + | Messages: append(append([]Message(nil), first.Messages...), | |||
| 251 | + | Message{Role: RoleAssistant, Content: []Content{TextBlock{Text: "ok"}}}, | |||
| 252 | + | Message{Role: RoleUser, Content: []Content{TextBlock{Text: "second"}}}, | |||
| 253 | + | ), | |||
| 254 | + | } | |||
| 255 | + | streamReq, previousResponseID := session.incrementalRequest(second, second.Context.CacheSessionID, client.canonicalResponsesBaseline) | |||
| 256 | + | if previousResponseID != "" { | |||
| 257 | + | t.Fatalf("previous response ID = %q, want full request after wire baseline change", previousResponseID) | |||
| 258 | + | } | |||
| 259 | + | if !reflect.DeepEqual(streamReq.Messages, second.Messages) { | |||
| 260 | + | t.Fatalf("stream messages = %#v, want complete transcript %#v", streamReq.Messages, second.Messages) | |||
| 261 | + | } | |||
| 262 | + | } | |||
| 263 | + | ||||
| 226 | func TestResponsesWebSocketCancellationClosesOnlyActiveConnection(t *testing.T) { | 264 | func TestResponsesWebSocketCancellationClosesOnlyActiveConnection(t *testing.T) { | ||
| 227 | closed := make(chan struct{}) | 265 | closed := make(chan struct{}) | ||
| 228 | var closeOnce sync.Once | 266 | var closeOnce sync.Once | ||