Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions proxy/handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -1310,6 +1310,18 @@ func (h *Handler) handleClaudeStream(ctx context.Context, w http.ResponseWriter,
if ctx.Err() != nil {
return
}
if isKiroInputTooLongError(err) {
h.recordFailureWithDetails("claude", model, account.ID, err)
if !messageStarted {
h.sendClaudeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
return
}
h.sendSSE(w, flusher, "error", map[string]interface{}{
"type": "error",
"error": map[string]string{"type": "invalid_request_error", "message": err.Error()},
})
return
}
lastErr = err
excluded[account.ID] = true
if !isStreamIntegrityError(err) {
Expand Down Expand Up @@ -1607,6 +1619,11 @@ func (h *Handler) handleClaudeNonStream(ctx context.Context, w http.ResponseWrit
if ctx.Err() != nil {
return
}
if isKiroInputTooLongError(err) {
h.recordFailureWithDetails("claude", model, account.ID, err)
h.sendClaudeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
return
}
lastErr = err
excluded[account.ID] = true
if !isStreamIntegrityError(err) {
Expand Down Expand Up @@ -2091,6 +2108,19 @@ func (h *Handler) handleOpenAIStream(ctx context.Context, w http.ResponseWriter,
if ctx.Err() != nil {
return
}
if isKiroInputTooLongError(err) {
h.recordFailureWithDetails("openai", model, account.ID, err)
if !responseStarted {
h.sendOpenAIError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
return
}
data, _ := json.Marshal(map[string]interface{}{
"error": map[string]string{"message": err.Error(), "type": "invalid_request_error"},
})
fmt.Fprintf(w, "data: %s\n\n", data)
flusher.Flush()
return
}
lastErr = err
excluded[account.ID] = true
// Integrity failures are upstream hiccups, not account faults.
Expand Down Expand Up @@ -2234,6 +2264,11 @@ func (h *Handler) handleOpenAINonStream(ctx context.Context, w http.ResponseWrit
if ctx.Err() != nil {
return
}
if isKiroInputTooLongError(err) {
h.recordFailureWithDetails("openai", model, account.ID, err)
h.sendOpenAIError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
return
}
lastErr = err
excluded[account.ID] = true
// Integrity failures are upstream hiccups, not account faults.
Expand Down
299 changes: 299 additions & 0 deletions proxy/input_too_long_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,299 @@
package proxy

import (
"context"
"encoding/json"
"fmt"
"kiro-go/config"
accountpool "kiro-go/pool"
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"sync/atomic"
"testing"
"time"
)

const replayElisionMarker = "[Earlier turns of this conversation were omitted to fit the model's input limit.]"

func replayFixtureText(tag string, approxTokens int) string {
unit := tag + " lorem ipsum dolor sit amet consectetur adipiscing elit sed do eiusmod. "
return strings.Repeat(unit, approxTokens/8+1)
}

func storeOversizedReplayChain(t *testing.T, turns, tokensPerSide int) *ResponsesObject {
t.Helper()
if err := config.Init(filepath.Join(t.TempDir(), "config.json")); err != nil {
t.Fatalf("config.Init: %v", err)
}

var newest *ResponsesObject
previousID := ""
for i := 0; i < turns; i++ {
input, err := json.Marshal(fmt.Sprintf("turn %d user: %s", i, replayFixtureText(fmt.Sprintf("t%d", i), tokensPerSide)))
if err != nil {
t.Fatalf("marshal turn %d: %v", i, err)
}
response := &ResponsesObject{
ID: fmt.Sprintf("resp_oversized_%d", i),
Object: "response",
Status: "completed",
Model: "gpt-5.6-terra",
StoredInput: input,
PreviousResponseID: previousID,
Output: []ResponseOutputItem{{
Type: "message",
Role: "assistant",
Content: []ResponseContentPart{{
Type: "output_text",
Text: fmt.Sprintf("turn %d assistant: %s", i, replayFixtureText("answer", tokensPerSide)),
}},
}},
}
if err := saveResponse(response); err != nil {
t.Fatalf("save turn %d: %v", i, err)
}
previousID = response.ID
newest = response
}
return newest
}

func replayTokenEstimateForTest(messages []OpenAIMessage) int {
total := 0
for _, message := range messages {
total += estimateOpenAIContentTokens(message.Content)
total += estimateApproxTokens(message.ToolCallID)
for _, call := range message.ToolCalls {
total += estimateApproxTokens(call.Function.Name)
total += estimateApproxTokens(call.Function.Arguments)
}
}
return total
}

// A compacting Codex client keeps previous_response_id while replacing its
// current input with a short summary. Replaying the full pre-compaction chain
// defeats compaction and eventually triggers CONTENT_LENGTH_EXCEEDS_THRESHOLD.
func TestResponsesReplayHistoryLeavesRoomForCompactedInput(t *testing.T) {
newest := storeOversizedReplayChain(t, 32, 5_000)
replayed := expandPreviousResponseHistory(newest, "gpt-5.6-terra")
budget := getContextWindowSize("gpt-5.6-terra") / 2
got := replayTokenEstimateForTest(replayed)
if got > budget {
t.Fatalf("replayed history uses %d tokens, over %d-token budget", got, budget)
}

var joined strings.Builder
for _, message := range replayed {
if text, ok := message.Content.(string); ok {
joined.WriteString(text)
joined.WriteByte('\n')
}
}
transcript := joined.String()
if !strings.Contains(transcript, "turn 31 user") {
t.Fatal("newest turn was dropped")
}
if strings.Contains(transcript, "turn 0 user") {
t.Fatal("oldest turn must be dropped when replay exceeds its budget")
}
if !strings.Contains(transcript, replayElisionMarker) {
t.Fatal("history elision must be disclosed to the model")
}
}

// A deterministic request-size rejection cannot succeed on another endpoint.
// Retrying it only multiplies latency and upstream load.
func TestCallKiroAPIDoesNotFallbackOnContentLengthRejection(t *testing.T) {
if err := config.Init(filepath.Join(t.TempDir(), "config.json")); err != nil {
t.Fatalf("config.Init: %v", err)
}
var hits atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
hits.Add(1)
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte(`{"message":"Input is too long.","reason":"CONTENT_LENGTH_EXCEEDS_THRESHOLD"}`))
}))
defer server.Close()

oldResolver := resolveKiroEndpoints
resolveKiroEndpoints = func(*config.Account) []kiroEndpoint {
return []kiroEndpoint{
{URL: server.URL, Origin: "AI_EDITOR", Name: "Kiro IDE"},
{URL: server.URL, Origin: "AI_EDITOR", Name: "CodeWhisperer"},
{URL: server.URL, Origin: "AI_EDITOR", Name: "AmazonQ"},
}
}
t.Cleanup(func() { resolveKiroEndpoints = oldResolver })

oldClient := kiroHttpStore.Load()
kiroHttpStore.Store(&http.Client{Timeout: time.Second, Transport: &http.Transport{}})
t.Cleanup(func() { kiroHttpStore.Store(oldClient) })

account := &config.Account{
ID: "content-length-account",
AccessToken: "token",
ProfileArn: "arn:aws:codewhisperer:profile/test",
}
err := CallKiroAPIContext(context.Background(), account, integrityTestPayload(), &KiroStreamCallback{})
if err == nil || !strings.Contains(err.Error(), "CONTENT_LENGTH_EXCEEDS_THRESHOLD") {
t.Fatalf("expected content-length error, got %v", err)
}
if hits.Load() != 1 {
t.Fatalf("content-length rejection must not fallback, hits=%d", hits.Load())
}
}

// The outer account loop must also stop after the first deterministic 400.
func TestResponsesDoesNotRotateAccountsOnContentLengthRejection(t *testing.T) {
var hits atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
hits.Add(1)
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte(`{"message":"Input is too long.","reason":"CONTENT_LENGTH_EXCEEDS_THRESHOLD"}`))
}))
defer server.Close()

if err := config.Init(filepath.Join(t.TempDir(), "config.json")); err != nil {
t.Fatalf("config.Init: %v", err)
}
for i := 0; i < maxAccountRetryAttempts; i++ {
if err := config.AddAccount(config.Account{
ID: fmt.Sprintf("oversize-account-%d", i),
Enabled: true,
AccessToken: "token",
ProfileArn: "arn:aws:codewhisperer:profile/test",
}); err != nil {
t.Fatalf("add account %d: %v", i, err)
}
}
if err := config.UpdatePreferredEndpoint("kiro"); err != nil {
t.Fatalf("set preferred endpoint: %v", err)
}
if err := config.UpdateEndpointFallback(false); err != nil {
t.Fatalf("disable fallback: %v", err)
}
t.Cleanup(swapKiroEndpointsForTest(t, server))

pool := accountpool.GetPool()
pool.Reload()
handler := &Handler{pool: pool, promptCache: newPromptCacheTracker(defaultPromptCacheTTL)}
req := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{
"model":"gpt-5.6-terra",
"input":"compacted summary",
"store":false
}`))
rec := httptest.NewRecorder()
handler.handleOpenAIResponses(rec, req)

if hits.Load() != 1 {
t.Fatalf("content-length rejection must not rotate accounts, hits=%d", hits.Load())
}
if rec.Code != http.StatusBadRequest {
t.Fatalf("status=%d, want 400; body=%s", rec.Code, rec.Body.String())
}

streamReq := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{
"model":"gpt-5.6-terra",
"input":"compacted summary",
"stream":true,
"store":false
}`))
streamRec := httptest.NewRecorder()
handler.handleOpenAIResponses(streamRec, streamReq)

if hits.Load() != 2 {
t.Fatalf("streaming rejection must use one additional attempt, total hits=%d", hits.Load())
}
streamBody := streamRec.Body.String()
if !strings.Contains(streamBody, "response.failed") || !strings.Contains(streamBody, `"type":"invalid_request_error"`) {
t.Fatalf("stream must report a non-retryable invalid request, body=%s", streamBody)
}
if strings.Count(streamBody, "event: response.in_progress") != 1 {
t.Fatalf("stream must not rotate accounts, body=%s", streamBody)
}
}

// Codex custom tool outputs can exceed a megabyte. They become structured
// current-message toolResults, which the byte-level payload limiter must trim;
// shortening only currentMessage.content leaves the oversized copy untouched.
func TestOpenAIToKiroTruncatesOversizedCurrentCustomToolResult(t *testing.T) {
call := ToolCall{ID: "call_large", Type: "function"}
call.Function.Name = "exec"
call.Function.Arguments = `{"input":"read a large command output"}`
payload := OpenAIToKiro(&OpenAIRequest{
Model: "gpt-5.6-terra",
Messages: []OpenAIMessage{
{Role: "assistant", Content: "", ToolCalls: []ToolCall{call}},
{Role: "tool", ToolCallID: "call_large", Content: strings.Repeat("large tool output line\n", 80_000)},
},
}, false)

encoded, err := json.Marshal(payload)
if err != nil {
t.Fatalf("marshal payload: %v", err)
}
if len(encoded) > maxPayloadBytes {
t.Fatalf("payload remains oversized after truncation: %d > %d", len(encoded), maxPayloadBytes)
}
ctx := payload.ConversationState.CurrentMessage.UserInputMessage.UserInputMessageContext
if ctx == nil || len(ctx.ToolResults) != 1 {
t.Fatalf("expected one structured tool result, got %#v", ctx)
}
result := ctx.ToolResults[0]
if result.ToolUseID != "call_large" {
t.Fatalf("tool result pairing changed: %q", result.ToolUseID)
}
if len(result.Content) != 1 || !strings.Contains(result.Content[0].Text, "Tool output truncated") {
t.Fatalf("tool output truncation was not disclosed: %#v", result.Content)
}
}

// Once a tool result is followed by another user message it moves into Kiro
// history and sanitizeKiroHistory flattens it into ordinary text. The minimum
// recent-history count must not make one multi-megabyte entry untrimmable.
func TestOpenAIToKiroTruncatesOversizedToolResultMovedIntoHistory(t *testing.T) {
call := ToolCall{ID: "call_historical_large", Type: "function"}
call.Function.Name = "exec"
call.Function.Arguments = `{"input":"read a large command output"}`
payload := OpenAIToKiro(&OpenAIRequest{
Model: "gpt-5.6-sol",
Messages: []OpenAIMessage{
{Role: "assistant", Content: "", ToolCalls: []ToolCall{call}},
{Role: "tool", ToolCallID: "call_historical_large", Content: strings.Repeat("historical tool output line\n", 120_000)},
{Role: "user", Content: "continue"},
},
}, false)

encoded, err := json.Marshal(payload)
if err != nil {
t.Fatalf("marshal payload: %v", err)
}
if len(encoded) > maxPayloadBytes {
t.Fatalf("payload remains oversized after history truncation: %d > %d", len(encoded), maxPayloadBytes)
}
historyTokenLimit := int(float64(getContextWindowSize("gpt-5.6-sol")) * payloadHistoryWindowShare)
if got := estimateKiroHistoryTokens(payload.ConversationState.History); got > historyTokenLimit {
t.Fatalf("history remains over token budget: %d > %d", got, historyTokenLimit)
}
if len(payload.ConversationState.History) == 0 {
t.Fatal("history structure must remain present")
}
var historyText strings.Builder
for _, entry := range payload.ConversationState.History {
if entry.UserInputMessage != nil {
historyText.WriteString(entry.UserInputMessage.Content)
}
if entry.AssistantResponseMessage != nil {
historyText.WriteString(entry.AssistantResponseMessage.Content)
}
}
if !strings.Contains(historyText.String(), "Earlier message content truncated") {
t.Fatal("history content truncation must be disclosed to the model")
}
if payload.ConversationState.CurrentMessage.UserInputMessage.Content != "continue" {
t.Fatalf("current user instruction was changed: %q", payload.ConversationState.CurrentMessage.UserInputMessage.Content)
}
}
10 changes: 8 additions & 2 deletions proxy/kiro.go
Original file line number Diff line number Diff line change
Expand Up @@ -479,8 +479,14 @@ endpointLoop:
if resp.StatusCode != 200 {
errBody, _ := io.ReadAll(resp.Body)
resp.Body.Close()
lastErr = fmt.Errorf("HTTP %d from %s: %s", resp.StatusCode, ep.Name, string(errBody))
// Authentication errors and payment errors are not retried across endpoints.
lastErr = newKiroHTTPError(resp.StatusCode, ep.Name, errBody)
// Authentication, payment, and deterministic request-shape errors
// cannot succeed on another endpoint. In particular, retrying an
// oversized payload only multiplies latency and upstream load.
if isKiroInputTooLongError(lastErr) {
logger.Warnf("[KiroAPI] Endpoint %s rejected oversized input; not retrying: %v", ep.Name, lastErr)
return lastErr
}
if resp.StatusCode == 401 || resp.StatusCode == 403 || resp.StatusCode == 402 {
return lastErr
}
Expand Down
Loading