From 4394fc6268266c855668b1de9d9182c488fdb1a3 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 22 Oct 2026 11:40:02 -0400 Subject: [PATCH 1/3] fix(provider): keep direct-service keys off receipt lookups A direct service's late receipt recovery used to follow an attempt base that classified as OpenRouter, sending that service's own key to the router's generation route. Direct clients now stop before any receipt lookup, whatever the attempt base; ordinary usage billing still runs first, and the router road recovers its receipts exactly as before. The overflow receipt test asserted the old behaviour, so it is replaced by a direct-service test across the configured base, its metered door and an OpenRouter attempt, and a torn router answer that records each request's host and bearer. Co-Authored-By: Claude Opus 5.5 (1M context) --- internal/provider/receipt.go | 9 +- internal/provider/telemetry_usage_test.go | 124 +++++++++++++++++++--- 2 files changed, 112 insertions(+), 21 deletions(-) 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/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) } } From 4c540abca9ce54c38e15abb27d1f03a6b7a78a77 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 22 Oct 2026 11:40:21 -0400 Subject: [PATCH 2/3] docs: record the direct-service receipt guard for #1894 Co-Authored-By: Claude Opus 5.5 (1M context) --- .../unreleased/1894-overflow-receipt-bearer.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) create mode 100644 docs/changes/unreleased/1894-overflow-receipt-bearer.md 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. From 5c18595e4a5b378ce8679757a5a7f9c1826a8b35 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 22 Oct 2026 11:49:17 -0400 Subject: [PATCH 3/3] test(provider): wait for the receipt worker as long as a hang takes receiptResult gave the worker five seconds of wall clock to report. Under a shared box's load (average twenty on eight cores) that budget was exceeded twice by tests that pass in a tenth of a second alone: TestLateReceiptFollowsTheOverflowRoute and TestK10TrimmedGenerationPairsRecoveryAndBlankUsesAttempt. The ceiling is a hang detector, so it moves to thirty seconds; a passing test still returns the moment the worker reports. Co-Authored-By: Claude Opus 5.5 (1M context) --- internal/provider/receipt_test.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) 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{} }