diff --git a/docs/changes/unreleased/1894-overflow-receipt-bearer.md b/docs/changes/unreleased/1894-overflow-receipt-bearer.md new file mode 100644 index 0000000000..96f0c097d8 --- /dev/null +++ b/docs/changes/unreleased/1894-overflow-receipt-bearer.md @@ -0,0 +1,12 @@ +--- +kind: fixed +title: Direct services stop before OpenRouter receipt lookups +pr: 1894 +surface: [engine] +invalidates: + - "A direct service's late receipt lookup previously followed an OpenRouter attempt base with the service's own key. Direct services now stop before any receipt lookup, regardless of the attempt base." +--- + +No shipped service routes overflow to OpenRouter. The shipped overflow doors +belong to the same vendor as the configured service. Direct-service usage +billing and OpenRouter's own late receipt recovery are unchanged. diff --git a/internal/provider/receipt.go b/internal/provider/receipt.go index 82da23e54d..4574ac6b6f 100644 --- a/internal/provider/receipt.go +++ b/internal/provider/receipt.go @@ -12,7 +12,6 @@ import ( "github.com/Agent-Field/agentfield/sdk/go/ai" "github.com/Agent-Field/codeaf/internal/guard" - "github.com/Agent-Field/codeaf/internal/modelsource" "github.com/Agent-Field/codeaf/internal/telemetry" ) @@ -171,10 +170,10 @@ func (c *Client) settle(ctx context.Context, model string, response *ai.Response return } // A DIRECT SERVICE HAS NO OPENROUTER GENERATION ROUTE TO ASK. Its missing - // usage block supplies no measured figures to record, and calling the - // router's route on the vendor would send that vendor a request and bearer - // for a fact this client already knows cannot be there. - if c.config.Direct && (base == c.config.BaseURL || telemetry.RoutingProvider(base) != modelsource.DefaultID) { + // usage block supplies no measured figures to record. An overflow route + // does not change which service owns this client's credential, so it must + // not carry that credential to a router's generation route either. + if c.config.Direct { return } sink := reconcileFrom(ctx) diff --git a/internal/provider/receipt_test.go b/internal/provider/receipt_test.go index 1928fd6c7a..5c700f21da 100644 --- a/internal/provider/receipt_test.go +++ b/internal/provider/receipt_test.go @@ -31,12 +31,19 @@ func receiptTestClient(t *testing.T, server *httptest.Server) *Client { return client } +// receiptResultCeiling bounds how long a test waits for the receipt worker to +// say what became of a call. It is a hang detector, not a pace: a passing test +// returns the moment the worker reports. Five seconds was exceeded on a shared +// box running other suites at a load average of twenty, twice, so the ceiling +// is set where only a worker that never reports can reach it. +const receiptResultCeiling = 30 * time.Second + func receiptResult(t *testing.T, results <-chan Reconciled) Reconciled { t.Helper() select { case result := <-results: return result - case <-time.After(5 * time.Second): + case <-time.After(receiptResultCeiling): t.Fatal("the receipt worker did not report what became of the call") return Reconciled{} } diff --git a/internal/provider/telemetry_usage_test.go b/internal/provider/telemetry_usage_test.go index 2ba8402d4e..dae169bf05 100644 --- a/internal/provider/telemetry_usage_test.go +++ b/internal/provider/telemetry_usage_test.go @@ -3,10 +3,13 @@ package provider import ( "context" "encoding/json" + "errors" "fmt" + "io" "net/http" "strings" "testing" + "testing/iotest" "github.com/Agent-Field/agentfield/sdk/go/ai" "github.com/Agent-Field/codeaf/internal/home" @@ -144,40 +147,129 @@ func TestOverflowUsageNamesTheActualAttemptRoute(t *testing.T) { } } -func TestLateReceiptFollowsTheOverflowRoute(t *testing.T) { +func TestV1V3DirectLateReceiptNeverUsesAnOverflowHost(t *testing.T) { + for _, base := range []string{ + "https://api.z.ai/api/coding/paas/v4", + "https://api.z.ai/api/paas/v4", + "https://openrouter.ai/api/v1", + } { + for _, reason := range []string{receiptTornReason, CutStalled.word(), receiptUnmeteredReason} { + for _, reported := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/%s/reported=%t", base, reason, reported), func(t *testing.T) { + t.Setenv(home.EnvVar, t.TempDir()) + telemetry.EnableForTest(t, true) + telemetry.VersionForTest(t, "v0.7.0") + finish := telemetry.BeginUsageSession(telemetry.ModeTask, "direct-late") + defer finish() + requests := make(chan *http.Request, receiptAttempts+1) + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests <- r.Clone(r.Context()) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"data":{"total_cost":0.02,"tokens_prompt":10,"tokens_completion":3}}`)) + }) + client, err := NewClient(Config{APIKey: "vendor-key", BaseURL: "https://api.z.ai/api/coding/paas/v4", Model: "glm-5.3", Direct: true, HTTPClient: handlerClient(handler)}) + if err != nil { + t.Fatal(err) + } + var bills []Billed + results := make(chan Reconciled, 1) + ctx := WithBilling(t.Context(), func(billed Billed) { bills = append(bills, billed) }) + ctx = WithReconcile(ctx, func(result Reconciled) { results <- result }) + pending := 0 + ctx = WithReceiptPending(ctx, func() func() { + pending++ + return func() {} + }) + response := &ai.Response{ID: "generation-direct"} + cost := 0.005 + if reported { + response.Usage = &ai.Usage{PromptTokens: 4, CompletionTokens: 1, Cost: &cost} + } + client.settle(ctx, client.config.Model, response, reason, 1, &callTrace{attemptBase: base}) + // Pending admission is synchronous. Drain an unexpected worker + // before inspecting its requests so a regression cannot pass by + // racing an assertion against the background lookup. + if pending != 0 { + t.Errorf("direct service queued %d receipts: %+v", pending, receiptResult(t, results)) + } + for len(requests) > 0 { + request := <-requests + t.Errorf("direct receipt sent %s %s with Authorization %q", request.Method, request.URL, request.Header.Get("Authorization")) + } + if len(results) != 0 { + t.Errorf("direct service produced %d late banking results", len(results)) + } + if !reported && len(bills) != 0 { + t.Errorf("missing direct usage banked %+v", bills) + } + if reported && (len(bills) != 1 || bills[0] != (Billed{Model: "glm-5.3", PromptTokens: 4, CompletionTokens: 1, Cost: cost})) { + t.Errorf("reported direct usage banked %+v", bills) + } + if rows := telemetry.SpoolContents(); len(rows) != 0 { + t.Errorf("direct settlement added %d usage events", len(rows)) + } + }) + } + } + } +} + +func TestV2RouterTornAnswerRecoversWithItsOwnBearer(t *testing.T) { t.Setenv(home.EnvVar, t.TempDir()) telemetry.EnableForTest(t, true) telemetry.VersionForTest(t, "v0.7.0") - finish := telemetry.BeginUsageSession(telemetry.ModeTask, "overflow-late") + finish := telemetry.BeginUsageSession(telemetry.ModeTask, "router-late") defer finish() results := make(chan Reconciled, 1) - handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Host != "openrouter.ai" || r.URL.Path != "/api/v1/generation" { - t.Errorf("receipt followed wrong route: %s", r.URL) + requests := make(chan *http.Request, receiptAttempts+2) + transport := roundTripFunc(func(r *http.Request) (*http.Response, error) { + requests <- r.Clone(r.Context()) + if r.URL.Path == "/api/v1/chat/completions" { + frame := `data: {"id":"generation-router","choices":[{"index":0,"delta":{"content":"begun"}}]}` + "\n\n" + return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": {"text/event-stream"}}, Body: io.NopCloser(io.MultiReader(strings.NewReader(frame), iotest.ErrReader(io.ErrUnexpectedEOF)))}, nil } - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write([]byte(`{"data":{"total_cost":0.02,"tokens_prompt":10,"tokens_completion":3}}`)) + return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": {"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"data":{"total_cost":0.02,"tokens_prompt":10,"tokens_completion":3}}`))}, nil }) - client, err := NewClient(Config{APIKey: "test", BaseURL: "https://api.anthropic.com/v1", Model: "anthropic/claude-sonnet-4", Direct: true, HTTPClient: handlerClient(handler)}) + client, err := NewClient(Config{APIKey: "router-key", BaseURL: "https://openrouter.ai/api/v1", Model: "anthropic/claude-sonnet-4", HTTPClient: &http.Client{Transport: transport}}) if err != nil { t.Fatal(err) } - ctx := WithReconcile(context.Background(), func(answer Reconciled) { results <- answer }) - client.settle(ctx, client.config.Model, &ai.Response{ID: "generation-overflow"}, receiptTornReason, 1, &callTrace{attemptBase: "https://openrouter.ai/api/v1"}) - if result := receiptResult(t, results); !result.Found { - t.Fatal("overflow receipt was not recovered") + var bills []Billed + ctx := WithStreamObserver(t.Context(), func(StreamEvent) {}) + ctx = WithCallNode(ctx, "router-node") + ctx = WithBilling(ctx, func(billed Billed) { bills = append(bills, billed) }) + ctx = WithReconcile(ctx, func(answer Reconciled) { results <- answer }) + if _, err := client.CompleteWithMessages(ctx, userMessages("hello")); !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("torn answer error = %v", err) + } + result := receiptResult(t, results) + want := Reconciled{Billed: Billed{Node: "router-node", Model: client.config.Model, PromptTokens: 10, CompletionTokens: 3, Cost: 0.02}, Ref: "generation-router", Reason: receiptTornReason, Found: true} + if result != want || len(bills) != 0 || len(results) != 0 { + t.Fatalf("banked result = %+v, want %+v; ordinary bills = %+v; extra results = %d", result, want, bills, len(results)) + } + if len(requests) != 2 { + t.Fatalf("router sent %d requests, want one completion and one receipt", len(requests)) + } + for _, expected := range []struct{ method, path, query string }{ + {http.MethodPost, "/api/v1/chat/completions", ""}, + {http.MethodGet, "/api/v1/generation", "id=generation-router"}, + } { + request := <-requests + if request.URL.Scheme != "https" || request.URL.Host != "openrouter.ai" || request.Method != expected.method || request.URL.Path != expected.path || request.URL.RawQuery != expected.query || request.Header.Get("Authorization") != "Bearer router-key" { + t.Errorf("router request = %s %s with Authorization %q", request.Method, request.URL, request.Header.Get("Authorization")) + } } rows := telemetry.SpoolContents() - if len(rows) != 1 { - t.Fatalf("expected one recovered receipt, got %d", len(rows)) + if len(rows) != 2 { + t.Fatalf("expected the torn attempt and recovered receipt, got %d events", len(rows)) } var event struct { Props map[string]any `json:"props"` } - if err := json.Unmarshal(rows[0], &event); err != nil { + if err := json.Unmarshal(rows[1], &event); err != nil { t.Fatal(err) } - if event.Props["routing_provider"] != "openrouter" || event.Props["total_tokens"] != float64(13) { + if event.Props["routing_provider"] != "openrouter" || event.Props["total_tokens"] != float64(13) || event.Props["usage_status"] != "reported" || event.Props["accounting_source"] != "provider" { t.Fatal(event.Props) } }