From e0a0b5a4b7e600dc5e676f2e4228c94713e2192d Mon Sep 17 00:00:00 2001 From: Luis Pater Date: Mon, 13 Jul 2026 03:22:47 +0800 Subject: [PATCH] feat(codex): add summary detection and processing for local compaction in websocket input Closes: #4227 --- .../openai/openai_responses_websocket.go | 61 +++++- .../openai/openai_responses_websocket_test.go | 182 ++++++++++++++++++ 2 files changed, 242 insertions(+), 1 deletion(-) diff --git a/sdk/api/handlers/openai/openai_responses_websocket.go b/sdk/api/handlers/openai/openai_responses_websocket.go index d749a5e0..d9cd1190 100644 --- a/sdk/api/handlers/openai/openai_responses_websocket.go +++ b/sdk/api/handlers/openai/openai_responses_websocket.go @@ -37,6 +37,8 @@ const ( wsDoneMarker = "[DONE]" wsTurnStateHeader = "x-codex-turn-state" wsTimelineBodyKey = "WEBSOCKET_TIMELINE_OVERRIDE" + + codexLocalCompactionSummaryPrefix = "Another language model started to solve this problem and produced a summary of its thinking process. You also have access to the state of the tools that were used by that language model. Use this to build on the work that has already been done and avoid duplicating work. Here is the summary produced by the other language model, use the information in this summary to assist with your own analysis:" ) var responsesWebsocketUpgrader = websocket.Upgrader{ @@ -684,12 +686,16 @@ func shouldReplaceWebsocketTranscript(rawJSON []byte, nextInput gjson.Result) bo if requestType != wsRequestTypeCreate && requestType != wsRequestTypeAppend { return false } - if strings.TrimSpace(gjson.GetBytes(rawJSON, "previous_response_id").String()) != "" { + previousResponseID := gjson.GetBytes(rawJSON, "previous_response_id") + if strings.TrimSpace(previousResponseID.String()) != "" { return false } if !nextInput.Exists() || !nextInput.IsArray() { return false } + if requestType == wsRequestTypeCreate && !previousResponseID.Exists() && inputHasCodexLocalCompactionSummary(nextInput) { + return true + } for _, item := range nextInput.Array() { switch strings.TrimSpace(item.Get("type").String()) { @@ -705,6 +711,59 @@ func shouldReplaceWebsocketTranscript(rawJSON []byte, nextInput gjson.Result) bo return false } +func inputHasCodexLocalCompactionSummary(input gjson.Result) bool { + if !input.IsArray() { + return false + } + + hasSummary := false + for index, item := range input.Array() { + itemType := strings.TrimSpace(item.Get("type").String()) + if itemType == "additional_tools" { + tools := item.Get("tools") + if index != 0 || strings.TrimSpace(item.Get("role").String()) != "developer" || !tools.IsArray() { + return false + } + for _, tool := range tools.Array() { + if !tool.IsObject() || strings.TrimSpace(tool.Get("type").String()) == "" { + return false + } + } + continue + } + if itemType != "" && itemType != "message" { + return false + } + + role := strings.TrimSpace(item.Get("role").String()) + if role != "user" && role != "developer" { + return false + } + if role == "user" && strings.HasPrefix(codexLocalCompactionMessageText(item), codexLocalCompactionSummaryPrefix+"\n") { + hasSummary = true + } + } + return hasSummary +} + +func codexLocalCompactionMessageText(message gjson.Result) string { + content := message.Get("content") + if content.Type == gjson.String { + return content.String() + } + if !content.IsArray() { + return "" + } + + var text strings.Builder + for _, part := range content.Array() { + if strings.TrimSpace(part.Get("type").String()) == "input_text" { + text.WriteString(part.Get("text").String()) + } + } + return text.String() +} + func inputSatisfiesPendingToolCalls(input gjson.Result, pendingCallIDs []string) bool { if len(pendingCallIDs) == 0 { return true diff --git a/sdk/api/handlers/openai/openai_responses_websocket_test.go b/sdk/api/handlers/openai/openai_responses_websocket_test.go index f4f16b7a..4cd522e4 100644 --- a/sdk/api/handlers/openai/openai_responses_websocket_test.go +++ b/sdk/api/handlers/openai/openai_responses_websocket_test.go @@ -660,6 +660,188 @@ func TestNormalizeResponsesWebsocketRequestSkipsPreviousResponseIDWhenPendingOut } } +func TestNormalizeResponsesWebsocketRequestReplacesCodexLocalCompactionTranscript(t *testing.T) { + lastRequest := []byte(`{"model":"gpt-5.6-sol","stream":true,"instructions":"be helpful","input":[ + {"type":"message","role":"user","id":"old-user","content":[{"type":"input_text","text":"old prompt"}]}, + {"type":"function_call_output","id":"old-tool-output","call_id":"old-call","output":"old result"} + ]}`) + lastResponseOutput := []byte(`[ + {"type":"function_call","id":"old-tool-call","call_id":"old-call","name":"lookup","arguments":"{}"}, + {"type":"message","role":"assistant","id":"old-assistant","content":[{"type":"output_text","text":"old answer"}]} + ]`) + raw := []byte(fmt.Sprintf(`{"type":"response.create","input":[ + {"type":"additional_tools","role":"developer","tools":[]}, + {"role":"developer","id":"initial-context","content":"workspace context"}, + {"type":"message","role":"user","id":"compacted-user","content":[{"type":"input_text","text":"retained context"}]}, + {"role":"user","id":"local-summary","content":%q}, + {"type":"message","role":"developer","id":"turn-context","content":[{"type":"input_text","text":"current workspace context"}]}, + {"role":"user","id":"incoming-user","content":"continue the task"} + ],"parallel_tool_calls":true,"client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"}}`, codexLocalCompactionSummaryPrefix+"\nThe compacted summary.")) + + normalized, next, errMsg := normalizeResponsesWebsocketRequestWithMode(raw, lastRequest, lastResponseOutput, false, false) + if errMsg != nil { + t.Fatalf("unexpected error: %v", errMsg.Error) + } + if gjson.GetBytes(normalized, "previous_response_id").Exists() { + t.Fatalf("replacement request must not include previous_response_id: %s", normalized) + } + if got, want := gjson.GetBytes(normalized, "input").Raw, gjson.GetBytes(raw, "input").Raw; got != want { + t.Fatalf("replacement input did not preserve the complete new transcript:\n got: %s\nwant: %s", got, want) + } + input := gjson.GetBytes(normalized, "input").Array() + wantIDs := []string{"", "initial-context", "compacted-user", "local-summary", "turn-context", "incoming-user"} + if len(input) != len(wantIDs) { + t.Fatalf("replacement input len = %d, want %d: %s", len(input), len(wantIDs), normalized) + } + for index, wantID := range wantIDs { + if got := input[index].Get("id").String(); got != wantID { + t.Fatalf("replacement input[%d].id = %q, want %q: %s", index, got, wantID, normalized) + } + } + if got := input[0].Get("type").String(); got != "additional_tools" { + t.Fatalf("input[0].type = %q, want additional_tools: %s", got, normalized) + } + if got := input[0].Get("role").String(); got != "developer" { + t.Fatalf("input[0].role = %q, want developer: %s", got, normalized) + } + if tools := input[0].Get("tools"); !tools.IsArray() || len(tools.Array()) != 0 { + t.Fatalf("input[0] empty tools array was not preserved: %s", normalized) + } + for _, staleID := range []string{"old-user", "old-tool-output", "old-tool-call", "old-assistant"} { + if bytes.Contains(normalized, []byte(staleID)) { + t.Fatalf("replacement input contains stale item %q: %s", staleID, normalized) + } + } + if got := gjson.GetBytes(normalized, "model").String(); got != "gpt-5.6-sol" { + t.Fatalf("model = %q, want gpt-5.6-sol", got) + } + if got := gjson.GetBytes(normalized, "instructions").String(); got != "be helpful" { + t.Fatalf("instructions = %q, want be helpful", got) + } + if !gjson.GetBytes(normalized, "stream").Bool() { + t.Fatalf("stream must be enabled: %s", normalized) + } + if !gjson.GetBytes(normalized, "parallel_tool_calls").Bool() { + t.Fatalf("parallel_tool_calls was not preserved: %s", normalized) + } + if got := gjson.GetBytes(normalized, "client_metadata.ws_request_header_x_openai_internal_codex_responses_lite").String(); got != "true" { + t.Fatalf("Responses Lite client metadata = %q, want true: %s", got, normalized) + } + if !bytes.Equal(next, normalized) { + t.Fatalf("next request snapshot should match normalized request") + } +} + +func TestShouldReplaceWebsocketTranscriptCodexLocalCompactionSemantics(t *testing.T) { + compactedInput := gjson.Parse(fmt.Sprintf(`[ + {"type":"message","role":"developer","content":[{"type":"input_text","text":"initial context"}]}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"retained context"}]}, + {"type":"message","role":"user","content":[{"type":"input_text","text":%q}]} + ]`, codexLocalCompactionSummaryPrefix+"\nSummary body.")) + if !shouldReplaceWebsocketTranscript([]byte(`{"type":"response.create"}`), compactedInput) { + t.Fatal("Codex local compaction input must replace the websocket transcript") + } + for _, request := range []string{ + `{"type":"response.create","previous_response_id":"resp-1"}`, + `{"type":"response.create","previous_response_id":""}`, + `{"type":"response.create","previous_response_id":null}`, + } { + if shouldReplaceWebsocketTranscript([]byte(request), compactedInput) { + t.Fatalf("request carrying previous_response_id must not use the local compaction rule: %s", request) + } + } + if shouldReplaceWebsocketTranscript([]byte(`{"type":"response.append"}`), compactedInput) { + t.Fatal("response.append must not be treated as a full local compaction reset") + } + + ordinaryInput := gjson.Parse(`[ + {"type":"message","role":"developer","content":"Please summarize future messages."}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"Please create a compacted summary of this text."}]} + ]`) + if shouldReplaceWebsocketTranscript([]byte(`{"type":"response.create"}`), ordinaryInput) { + t.Fatal("ordinary user/developer input must not replace the transcript") + } +} + +func TestCodexLocalCompactionSummaryContentShapes(t *testing.T) { + tests := []struct { + name string + content string + want bool + }{ + {name: "string content", content: fmt.Sprintf(`%q`, codexLocalCompactionSummaryPrefix+"\nSummary body."), want: true}, + {name: "multiple input text parts", content: fmt.Sprintf(`[{"type":"input_text","text":%q},{"type":"input_text","text":"\nSummary body."}]`, codexLocalCompactionSummaryPrefix), want: true}, + {name: "non-text part before summary", content: fmt.Sprintf(`[{"type":"input_image","image_url":"data:image/png;base64,AA=="},{"type":"input_text","text":%q}]`, codexLocalCompactionSummaryPrefix+"\nSummary body."), want: true}, + {name: "bare prefix", content: fmt.Sprintf(`%q`, codexLocalCompactionSummaryPrefix), want: false}, + {name: "prefix followed by space", content: fmt.Sprintf(`%q`, codexLocalCompactionSummaryPrefix+" Summary body."), want: false}, + {name: "summary after ordinary text", content: fmt.Sprintf(`[{"type":"input_text","text":"ordinary text"},{"type":"input_text","text":%q}]`, codexLocalCompactionSummaryPrefix+"\nSummary body."), want: false}, + {name: "developer summary", content: fmt.Sprintf(`%q`, codexLocalCompactionSummaryPrefix+"\nSummary body."), want: false}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + role := "user" + if test.name == "developer summary" { + role = "developer" + } + input := gjson.Parse(fmt.Sprintf(`[{"type":"message","role":%q,"content":%s}]`, role, test.content)) + if got := inputHasCodexLocalCompactionSummary(input); got != test.want { + t.Fatalf("inputHasCodexLocalCompactionSummary() = %t, want %t", got, test.want) + } + }) + } +} + +func TestCodexLocalCompactionSummaryAdditionalToolsConstraints(t *testing.T) { + summary := fmt.Sprintf(`{"role":"user","content":%q}`, codexLocalCompactionSummaryPrefix+"\nSummary body.") + tests := []struct { + name string + input string + want bool + }{ + {name: "Responses Lite tools first", input: fmt.Sprintf(`[{"type":"additional_tools","role":"developer","tools":[{"type":"custom","name":"exec"}]},%s]`, summary), want: true}, + {name: "tools after message", input: fmt.Sprintf(`[%s,{"type":"additional_tools","role":"developer","tools":[{"type":"custom","name":"exec"}]}]`, summary)}, + {name: "tools with user role", input: fmt.Sprintf(`[{"type":"additional_tools","role":"user","tools":[{"type":"custom","name":"exec"}]},%s]`, summary)}, + {name: "tools missing array", input: fmt.Sprintf(`[{"type":"additional_tools","role":"developer"},%s]`, summary)}, + {name: "tools not array", input: fmt.Sprintf(`[{"type":"additional_tools","role":"developer","tools":{}},%s]`, summary)}, + {name: "tools empty", input: fmt.Sprintf(`[{"type":"additional_tools","role":"developer","tools":[]},%s]`, summary), want: true}, + {name: "malformed tool", input: fmt.Sprintf(`[{"type":"additional_tools","role":"developer","tools":[null]},%s]`, summary)}, + {name: "arbitrary input item", input: fmt.Sprintf(`[{"type":"unknown","role":"developer"},%s]`, summary)}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := inputHasCodexLocalCompactionSummary(gjson.Parse(test.input)); got != test.want { + t.Fatalf("inputHasCodexLocalCompactionSummary() = %t, want %t", got, test.want) + } + }) + } +} + +func TestCodexLocalCompactionSummaryRejectsOrdinaryHistoryItems(t *testing.T) { + tests := []struct { + name string + historyItem string + wantReplace bool + }{ + {name: "reasoning", historyItem: `{"type":"reasoning","id":"reasoning-1"}`}, + {name: "assistant", historyItem: `{"type":"message","role":"assistant","id":"assistant-1"}`, wantReplace: true}, + {name: "function call", historyItem: `{"type":"function_call","call_id":"call-1"}`, wantReplace: true}, + {name: "function call output", historyItem: `{"type":"function_call_output","call_id":"call-1"}`}, + {name: "custom tool call", historyItem: `{"type":"custom_tool_call","call_id":"call-1"}`, wantReplace: true}, + {name: "custom tool call output", historyItem: `{"type":"custom_tool_call_output","call_id":"call-1"}`}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + input := gjson.Parse(fmt.Sprintf(`[%s,{"type":"message","role":"user","content":[{"type":"input_text","text":%q}]}]`, test.historyItem, codexLocalCompactionSummaryPrefix+"\nSummary body.")) + if inputHasCodexLocalCompactionSummary(input) { + t.Fatal("ordinary transcript history must not match the local user-summary shape") + } + if got := shouldReplaceWebsocketTranscript([]byte(`{"type":"response.create"}`), input); got != test.wantReplace { + t.Fatalf("shouldReplaceWebsocketTranscript() = %t, want %t", got, test.wantReplace) + } + }) + } +} + func TestNormalizeResponsesWebsocketRequestWithPreviousResponseIDMergedWhenIncrementalDisabled(t *testing.T) { lastRequest := []byte(`{"model":"test-model","stream":true,"input":[{"type":"message","id":"msg-1"}]}`) lastResponseOutput := []byte(`[