Merge pull request #55 from manaflow-ai/fix-fable-observability

Fable-chain observability: truthful give-up log, attributed Bedrock gateway traffic (v0.1.33)

Author: lawrencecchen

internal/proxy/bedrock.go

@@ -82,7 +82,7 @@ func (s Server) bedrockHandler() http.Handler {
 		resp, sourceName, err := s.signAndForwardBedrockWithHeaders(r.Context(), r.Method, upstreamPath, r.URL.RawQuery, headers, body)
 		if err != nil {
 			if s.Logger != nil {
-				s.Logger.Error("bedrock upstream request failed", "path", upstreamPath, "error", err)
+				s.Logger.Error("bedrock upstream request failed", "path", upstreamPath, "remote_addr", clientRemoteIP(r), "user_agent", r.UserAgent(), "error", err)
 			}
 			http.Error(w, "bedrock upstream request failed", http.StatusBadGateway)
 			return
@@ -102,7 +102,7 @@ func (s Server) bedrockHandler() http.Handler {
 		model := bedrockModelFromPath(upstreamPath)
 		if resp.StatusCode == http.StatusTooManyRequests {
 			if s.Logger != nil {
-				s.Logger.Warn("bedrock throttled", "model", model, "path", upstreamPath, "bedrock_source", sourceName)
+				s.Logger.Warn("bedrock throttled", "model", model, "path", upstreamPath, "remote_addr", clientRemoteIP(r), "user_agent", r.UserAgent(), "bedrock_source", sourceName)
 			}
 			cfg.onThrottle(sourceName, model)
 		}

internal/proxy/bedrock_test.go

@@ -1,8 +1,10 @@
 package proxy
 
 import (
+	"bytes"
 	"context"
 	"io"
+	"log/slog"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -157,6 +159,38 @@ func TestBedrockHandlerRetriesNextSourceOnThrottle(t *testing.T) {
 	}
 }
 
+func TestBedrockHandlerLogsClientAttributionOnThrottle(t *testing.T) {
+	var logBuf bytes.Buffer
+	rt := bedrockRoundTripFunc(func(req *http.Request) (*http.Response, error) {
+		return &http.Response{
+			StatusCode: http.StatusTooManyRequests,
+			Body:       io.NopCloser(strings.NewReader(`{"error":"throttled"}`)),
+			Header:     http.Header{"Content-Type": []string{"application/json"}},
+		}, nil
+	})
+	s := Server{
+		Logger: slog.New(slog.NewTextHandler(&logBuf, nil)),
+		Bedrock: &BedrockConfig{
+			Region:    "us-east-1",
+			Sources:   []BedrockCredentialSource{{Name: "aw0", Credentials: namedBedrockCreds("AKIAONE")}},
+			Transport: rt,
+		},
+	}
+
+	req := httptest.NewRequest(http.MethodPost, "/bedrock/model/us.anthropic.claude-fable-5/invoke", strings.NewReader(`{"anthropic_version":"bedrock-2023-05-31"}`))
+	req.Header.Set("User-Agent", "sr claude-aws")
+	rec := httptest.NewRecorder()
+	s.bedrockHandler().ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusTooManyRequests {
+		t.Fatalf("status = %d, want 429 (body %q)", rec.Code, rec.Body.String())
+	}
+	logs := logBuf.String()
+	if !strings.Contains(logs, "bedrock throttled") || !strings.Contains(logs, "remote_addr=") || !strings.Contains(logs, "user_agent=\"sr claude-aws\"") {
+		t.Fatalf("expected throttle log with remote_addr and user_agent; logs=\n%s", logs)
+	}
+}
+
 func TestBedrockHandlerDisabledWithoutConfig(t *testing.T) {
 	s := Server{}
 	rec := httptest.NewRecorder()

internal/proxy/claude_fable_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"errors"
 	"io"
+	"log/slog"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -364,6 +365,125 @@ func TestUsageLimitRetryTransportClaudeFableFallsBackAfterPoolExhausted(t *testi
 	}
 }
 
+func TestUsageLimitRetryTransportClaudeFableFallbackSuppressesExhaustedLog(t *testing.T) {
+	server, store := claudeFailoverServer(t)
+	if _, err := store.Put("claude", "session-fable", "cooked@example.com", ""); err != nil {
+		t.Fatal(err)
+	}
+	stub := &stubRoundTripper{responses: func(req *http.Request) *http.Response {
+		h := http.Header{}
+		h.Set("Anthropic-Ratelimit-Unified-Status", "rejected")
+		return &http.Response{
+			StatusCode: http.StatusTooManyRequests,
+			Body:       io.NopCloser(strings.NewReader(`{"type":"error","error":{"type":"rate_limit_error","message":"Error"}}`)),
+			Header:     h,
+		}
+	}}
+	bedrockRT := bedrockRoundTripFunc(func(req *http.Request) (*http.Response, error) {
+		return &http.Response{
+			StatusCode: http.StatusOK,
+			Body:       io.NopCloser(strings.NewReader(`{"type":"message","usage":{"output_tokens":3}}`)),
+			Header:     http.Header{"Content-Type": []string{"application/json"}},
+		}, nil
+	})
+	server.Bedrock = &BedrockConfig{
+		Region:    "us-east-1",
+		Sources:   []BedrockCredentialSource{{Name: "aw0", Credentials: staticBedrockCreds()}},
+		Transport: bedrockRT,
+	}
+	var logBuf bytes.Buffer
+	logger := slog.New(slog.NewTextHandler(&logBuf, nil))
+	fableBody := []byte(`{"model":"claude-fable-5","max_tokens":8,"messages":[]}`)
+	req, err := http.NewRequest(http.MethodPost, "https://api.anthropic.com/v1/messages", bytes.NewReader(fableBody))
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.GetBody = func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(fableBody)), nil }
+	req.Header.Set("Authorization", "Bearer tok-cooked")
+	transport := usageLimitRetryTransport{
+		base:        stub,
+		server:      &server,
+		logger:      logger,
+		provider:    accounts.ProviderClaude,
+		agent:       "claude",
+		session:     "session-fable",
+		account:     "cooked@example.com",
+		method:      http.MethodPost,
+		path:        "/v1/messages",
+		maxAttempts: 4,
+		fableFallback: func() (*http.Response, bool) {
+			return server.claudeFableFallbackResponse(req, fableBody)
+		},
+	}
+	response, err := transport.RoundTrip(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d, want 200 from bedrock after pool exhausted", response.StatusCode)
+	}
+	logs := logBuf.String()
+	if !strings.Contains(logs, "serving claude fable via fallback chain") {
+		t.Fatalf("expected fable fallback save log; logs=\n%s", logs)
+	}
+	if strings.Contains(logs, "returned to client after failover exhausted") {
+		t.Fatalf("did not expect client-facing failover-exhaustion signal on fallback save; logs=\n%s", logs)
+	}
+}
+
+func TestUsageLimitRetryTransportClaudeFableFallbackFailureLogsExhausted(t *testing.T) {
+	server, store := claudeFailoverServer(t)
+	if _, err := store.Put("claude", "session-fable", "cooked@example.com", ""); err != nil {
+		t.Fatal(err)
+	}
+	stub := &stubRoundTripper{responses: func(req *http.Request) *http.Response {
+		h := http.Header{}
+		h.Set("Anthropic-Ratelimit-Unified-Status", "rejected")
+		return &http.Response{
+			StatusCode: http.StatusTooManyRequests,
+			Body:       io.NopCloser(strings.NewReader(`{"type":"error","error":{"type":"rate_limit_error","message":"Error"}}`)),
+			Header:     h,
+		}
+	}}
+	var logBuf bytes.Buffer
+	logger := slog.New(slog.NewTextHandler(&logBuf, nil))
+	fableBody := []byte(`{"model":"claude-fable-5","max_tokens":8,"messages":[]}`)
+	req, err := http.NewRequest(http.MethodPost, "https://api.anthropic.com/v1/messages", bytes.NewReader(fableBody))
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.GetBody = func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(fableBody)), nil }
+	req.Header.Set("Authorization", "Bearer tok-cooked")
+	transport := usageLimitRetryTransport{
+		base:        stub,
+		server:      &server,
+		logger:      logger,
+		provider:    accounts.ProviderClaude,
+		agent:       "claude",
+		session:     "session-fable",
+		account:     "cooked@example.com",
+		method:      http.MethodPost,
+		path:        "/v1/messages",
+		maxAttempts: 4,
+		fableFallback: func() (*http.Response, bool) {
+			return nil, false
+		},
+	}
+	response, err := transport.RoundTrip(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusTooManyRequests {
+		t.Fatalf("status = %d, want pool 429", response.StatusCode)
+	}
+	logs := logBuf.String()
+	if !strings.Contains(logs, "claude rate-limit returned to client after failover exhausted") {
+		t.Fatalf("expected client-facing failover-exhaustion signal; logs=\n%s", logs)
+	}
+}
+
 func TestOauthRetryAccountSkipsAPIKeyAccountsWhenOAuthOnly(t *testing.T) {
 	server, _ := claudeFailoverServer(t)
 	server.Accounts = append(server.Accounts, accounts.Account{

internal/proxy/proxy.go

@@ -3404,21 +3404,21 @@ func (t usageLimitRetryTransport) RoundTrip(req *http.Request) (*http.Response,
 			if t.server == nil {
 				reason = "no_server"
 			}
-			t.logClaudeFailoverExhausted(response, accountID, reason, attempt, maxAttempts, len(tried))
 			if fallback, ok := t.fableFallbackResponse(response, accountID, reason); ok {
 				return fallback, nil
 			}
+			t.logClaudeFailoverExhausted(response, accountID, reason, attempt, maxAttempts, len(tried))
 			return response, nil
 		}
 		nextAccount, pickErr := t.server.oauthRetryAccount(req.Context(), t.provider, t.agent, t.session, t.userEmail, t.poolModel, tried, t.fableFallback != nil)
 		if pickErr != nil {
 			if t.logger != nil {
 				t.logger.Warn("usage-limit retry has no alternate account", "agent", t.agent, "session", t.session, "account", accountID, "method", t.method, "path", t.path, "upstream", t.upstream, "error", pickErr)
 			}
-			t.logClaudeFailoverExhausted(response, accountID, "no_alternate_account", attempt, maxAttempts, len(tried))
 			if fallback, ok := t.fableFallbackResponse(response, accountID, "no_alternate_account"); ok {
 				return fallback, nil
 			}
+			t.logClaudeFailoverExhausted(response, accountID, "no_alternate_account", attempt, maxAttempts, len(tried))
 			return response, nil
 		}
 		body, bodyErr := req.GetBody()

package.json

@@ -1,6 +1,6 @@
 {
   "name": "subrouter",
-  "version": "0.1.32",
+  "version": "0.1.33",
   "description": "Routes AI coding-agent traffic across subscription accounts and API keys.",
   "license": "MIT",
   "homepage": "https://github.com/manaflow-ai/subrouter#readme",

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "subrouter"
-version = "0.1.32"
+version = "0.1.33"
 description = "Routes AI coding-agent traffic across subscription accounts and API keys."
 readme = "README.md"
 requires-python = ">=3.9"