diff --git a/internal/proxy/no_progress_internal_test.go b/internal/proxy/no_progress_internal_test.go index 16baec45..fdde5680 100644 --- a/internal/proxy/no_progress_internal_test.go +++ b/internal/proxy/no_progress_internal_test.go @@ -2,8 +2,10 @@ package proxy import ( "context" + "encoding/json" "net/http/httptest" "strconv" + "strings" "testing" "time" @@ -49,6 +51,43 @@ func TestHandleNoProgressBreak_PreservesUserForcedPin(t *testing.T) { assert.Contains(t, rec.Body.String(), "preserving the explicit force-model pin") } +// TestHandleNoProgressBreak_OpenAIChatCompletionShape covers the FormatOpenAI +// branch that was production-dead until #825 wired ProxyOpenAIChatCompletion. +func TestHandleNoProgressBreak_OpenAIChatCompletionShape(t *testing.T) { + env, err := translate.ParseOpenAI([]byte(`{ + "model":"gpt-4o", + "messages":[{"role":"user","content":"retry"}] + }`)) + require.NoError(t, err) + rec := httptest.NewRecorder() + svc := &Service{} + + err = svc.handleNoProgressBreak( + context.Background(), rec, env, noProgressMatchThreshold, uuid.New(), + sessionKeyFromString("openai-noprogress"), + "default_high", "gpt-4o", "openai", 42, + ) + require.NoError(t, err) + assert.Equal(t, "application/json", rec.Header().Get("Content-Type")) + + var resp map[string]any + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &resp)) + assert.Equal(t, "chat.completion", resp["object"]) + assert.Equal(t, "weave-router", resp["model"]) + choices, ok := resp["choices"].([]any) + require.True(t, ok) + require.Len(t, choices, 1) + choice := choices[0].(map[string]any) + assert.Equal(t, "stop", choice["finish_reason"]) + msg := choice["message"].(map[string]any) + assert.Equal(t, "assistant", msg["role"]) + content, _ := msg["content"].(string) + assert.Contains(t, strings.ToLower(content), "no-progress loop detected") + usage, ok := resp["usage"].(map[string]any) + require.True(t, ok) + assert.EqualValues(t, 42, usage["prompt_tokens"]) +} + func TestComputeNoProgressFingerprint_StableAcrossCalls(t *testing.T) { d := router.Decision{Model: "qwen/qwen3-235b-a22b-2507", Provider: "bedrock"} a := computeNoProgressFingerprint(d, "explore RSVP files in this repository", 1, "") diff --git a/internal/proxy/no_progress_openai_test.go b/internal/proxy/no_progress_openai_test.go new file mode 100644 index 00000000..6b71772d --- /dev/null +++ b/internal/proxy/no_progress_openai_test.go @@ -0,0 +1,142 @@ +package proxy_test + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "workweave/router/internal/providers" + "workweave/router/internal/proxy" + "workweave/router/internal/router" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// openAIStuckSpawnBody is the cross-envelope no-progress positive-control shape +// for OpenAI ingress: one assistant tool_call + empty tool result, identical +// across requests so the dispatch fingerprint stays frozen (model/provider/ +// progress marker/prompt prefix). A single in-envelope tool call keeps +// detectToolCallLoop from firing; only the cross-request detector should trip. +const openAIStuckSpawnBody = `{ + "model": "gpt-4o", + "messages": [ + {"role": "user", "content": "explore RSVP files in this repository"}, + {"role": "assistant", "content": null, "tool_calls": [ + {"id": "call_1", "type": "function", "function": {"name": "Read", "arguments": "{\"path\":\"README.md\"}"}} + ]}, + {"role": "tool", "tool_call_id": "call_1", "content": ""} + ], + "tools": [{"type": "function", "function": {"name": "Read", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}}}] +}` + +func openAINoProgressSvc(fr *fakeRouter, upstream *fakeProvider) *proxy.Service { + return proxy.NewService( + fr, + map[string]providers.Client{ + providers.ProviderOpenAI: upstream, + providers.ProviderAnthropic: &fakeProvider{}, + }, + nil, false, nil, nil, + false, providers.ProviderAnthropic, "claude-haiku-4-5", + nil, + ) +} + +func openAINoProgressCtx() context.Context { + ctx := context.WithValue(context.Background(), proxy.APIKeyIDContextKey{}, "key-noprogress-openai") + return context.WithValue(ctx, proxy.InstallationIDContextKey{}, uuid.New().String()) +} + +// TestProxyOpenAIChatCompletion_NoProgressBreaksAtThreshold is the #825 +// positive control: five identical tool-bearing OpenAI dispatches in-window +// must synthesize a no-progress break (same threshold as ProxyMessages). +func TestProxyOpenAIChatCompletion_NoProgressBreaksAtThreshold(t *testing.T) { + fr := &fakeRouter{decision: router.Decision{ + Provider: providers.ProviderOpenAI, + Model: "gpt-4o", + Reason: "fresh", + }} + upstream := &fakeProvider{ + proxyResponse: func(w http.ResponseWriter) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"id":"chatcmpl_up","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}`)) + }, + } + svc := openAINoProgressSvc(fr, upstream) + ctx := openAINoProgressCtx() + + const threshold = 5 + var brokeAt int + for attempt := 1; attempt <= threshold+1; attempt++ { + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader("")) + require.NoError(t, svc.ProxyOpenAIChatCompletion(ctx, []byte(openAIStuckSpawnBody), rec, req)) + + body := rec.Body.String() + if strings.Contains(strings.ToLower(body), "no-progress loop detected") { + brokeAt = attempt + var resp map[string]any + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &resp), "break body must be JSON chat.completion") + assert.Equal(t, "chat.completion", resp["object"]) + choices, ok := resp["choices"].([]any) + require.True(t, ok) + require.NotEmpty(t, choices) + msg := choices[0].(map[string]any)["message"].(map[string]any) + assert.Equal(t, "assistant", msg["role"]) + assert.Equal(t, "stop", choices[0].(map[string]any)["finish_reason"]) + break + } + } + + require.Equal(t, threshold, brokeAt, + "ProxyOpenAIChatCompletion must break on the %d-th identical fingerprint (got brokeAt=%d, upstreamHits=%d)", + threshold, brokeAt, len(upstream.proxyBodies)) + assert.Equal(t, threshold-1, len(upstream.proxyBodies), + "upstream must run for the first %d attempts only", threshold-1) +} + +// TestProxyOpenAIChatCompletion_NoProgressNegativeControlVaryingTools confirms +// healthy OpenAI tool-bearing turns with changing tool args do not trip. +func TestProxyOpenAIChatCompletion_NoProgressNegativeControlVaryingTools(t *testing.T) { + fr := &fakeRouter{decision: router.Decision{ + Provider: providers.ProviderOpenAI, + Model: "gpt-4o", + Reason: "fresh", + }} + upstream := &fakeProvider{ + proxyResponse: func(w http.ResponseWriter) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"id":"chatcmpl_up","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}`)) + }, + } + svc := openAINoProgressSvc(fr, upstream) + ctx := openAINoProgressCtx() + + for i := 0; i < 6; i++ { + body := fmt.Sprintf(`{ + "model": "gpt-4o", + "messages": [ + {"role": "user", "content": "explore RSVP files in this repository"}, + {"role": "assistant", "content": null, "tool_calls": [ + {"id": "call_%d", "type": "function", "function": {"name": "Read", "arguments": "{\"path\":\"file-%d.md\"}"}} + ]}, + {"role": "tool", "tool_call_id": "call_%d", "content": "ok"} + ], + "tools": [{"type": "function", "function": {"name": "Read", "parameters": {"type": "object"}}}] + }`, i, i, i) + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader("")) + require.NoError(t, svc.ProxyOpenAIChatCompletion(ctx, []byte(body), rec, req)) + assert.NotContains(t, strings.ToLower(rec.Body.String()), "no-progress loop detected", + "varying tool args must not trip no-progress (attempt %d)", i+1) + } + assert.Equal(t, 6, len(upstream.proxyBodies), "all varying turns must dispatch upstream") +} diff --git a/internal/proxy/service.go b/internal/proxy/service.go index 09be662a..f095703f 100644 --- a/internal/proxy/service.go +++ b/internal/proxy/service.go @@ -4235,8 +4235,10 @@ func (s *Service) ProxyOpenAIChatCompletion(ctx context.Context, body []byte, w } baseExcludedOAI := s.excludedModelsForRequest(ctx) - // Snapshot the inbound tool-output size before any env rewrite (proactive - // compaction below, or runTurnLoop's switch handover); see toolResultBytesPtr. + // Snapshot inbound tool state before any env rewrite (proactive compaction + // below, or runTurnLoop's switch handover). Same shape as ProxyMessages: + // no-progress gating and toolResultBytesPtr must see what the client sent. + inboundToolCallCount := len(env.AssistantToolCallSignatures()) inboundLastUser := env.LastUserMessage() // Proactive context-window compaction, as in ProxyMessages. Skipped for @@ -4320,6 +4322,20 @@ func (s *Service) ProxyOpenAIChatCompletion(ctx context.Context, body []byte, w pinAgeSec := routeRes.PinAgeSec s.logPlannerOutcome(ctx, routeRes) + // Cross-envelope no-progress detector (same gate as ProxyMessages). OpenAI + // tool-bearing turns use AssistantToolCallSignatures + trailing role=tool + // HasToolResult — same envelope helpers, different wire shape. Reuses + // handleNoProgressBreak's FormatOpenAI branch. + _, agentShadowMode := AgentShadowEvalFromContext(ctx) + toolBearingTurn := inboundToolCallCount > 0 || inboundLastUser.HasToolResult + if !agentShadowMode && !routeRes.AuthoritativePerTurn && toolBearingTurn && s.noProgress != nil { + fp := computeNoProgressFingerprint(decision, promptText, feats.MessageCount, toolProgressMarker(env)) + role := roleForTier(catalog.TierFor(feats.Model)) + if looped, count := s.noProgress.recordAndDetect(routeRes.SessionKey, installationID, role, fp, time.Now()); looped { + return s.handleNoProgressBreak(ctx, w, env, count, installationID, routeRes.SessionKey, role, decision.Model, decision.Provider, feats.Tokens) + } + } + // See the ProxyMessages cache-eligibility note: subsidized requests bypass the // semantic cache (the key doesn't capture headroom-dependent model choice). cacheEligible := s.semanticCacheAllowed(ctx) && s.semanticCache != nil && !env.Stream() && decision.Metadata != nil && externalID != "" && !bypassEval && !responsesPassthrough && !billing.SubscriptionOnlyFromContext(ctx) && len(s.subsidyFactors(ctx, r.Header)) == 0