diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0215dec..54397b2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -9,20 +9,20 @@ jobs: lint: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 - - uses: actions/setup-go@v5 + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5 with: go-version: "1.26" - name: golangci-lint - uses: golangci/golangci-lint-action@v7 + uses: golangci/golangci-lint-action@9fae48acfc02a90574d7c304a1758ef9895495fa # v7 with: version: v2.11.4 test: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 - - uses: actions/setup-go@v5 + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5 with: go-version: "1.26" - name: Run tests @@ -36,15 +36,15 @@ jobs: exit 1 fi - name: Upload coverage - uses: codecov/codecov-action@v4 + uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4 with: file: coverage.out vet: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 - - uses: actions/setup-go@v5 + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5 with: go-version: "1.26" - run: go vet ./... @@ -52,8 +52,8 @@ jobs: dashboard: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 - - uses: actions/setup-node@v4 + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4 with: node-version: "22" - name: Build dashboard diff --git a/.github/workflows/docker-build.yml b/.github/workflows/docker-build.yml index 81467fd..d00ff79 100644 --- a/.github/workflows/docker-build.yml +++ b/.github/workflows/docker-build.yml @@ -4,16 +4,38 @@ on: branches: [main, staging] jobs: + ci-gate: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 + - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5 + with: + go-version: "1.26" + - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4 + with: + node-version: "22" + - name: Lint + uses: golangci/golangci-lint-action@9fae48acfc02a90574d7c304a1758ef9895495fa # v7 + with: + version: v2.11.4 + - name: Test + run: go test -race ./... + - name: Vet + run: go vet ./... + - name: Dashboard build + run: cd dashboard/svelte && npm ci && npm run build && npm run check + docker: + needs: [ci-gate] runs-on: ubuntu-latest permissions: contents: read packages: write steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 - name: Log in to GHCR - uses: docker/login-action@v3 + uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3 with: registry: ghcr.io username: ${{ github.actor }} @@ -29,7 +51,7 @@ jobs: fi - name: Build and push to GHCR - uses: docker/build-push-action@v5 + uses: docker/build-push-action@ca052bb54ab0790a636c9b5f226502c73d547a25 # v5 with: context: . push: true diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 2578837..addf36d 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -10,13 +10,13 @@ jobs: contents: write packages: write steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 with: fetch-depth: 0 - - uses: actions/setup-go@v5 + - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5 with: go-version: "1.26" - - uses: actions/setup-node@v4 + - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4 with: node-version: "22" @@ -24,7 +24,7 @@ jobs: run: cd dashboard/svelte && npm ci && npm run build - name: Run GoReleaser - uses: goreleaser/goreleaser-action@v6 + uses: goreleaser/goreleaser-action@e435ccd777264be153ace6237001ef4d979d3a7a # v6 with: version: latest args: release --clean @@ -32,14 +32,14 @@ jobs: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - name: Log in to GHCR - uses: docker/login-action@v3 + uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3 with: registry: ghcr.io username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} - name: Log in to Docker Hub - uses: docker/login-action@v3 + uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3 with: username: ${{ secrets.DOCKERHUB_USERNAME }} password: ${{ secrets.DOCKERHUB_TOKEN }} @@ -53,7 +53,7 @@ jobs: echo "minor=$MINOR" >> $GITHUB_OUTPUT - name: Build and push Docker image - uses: docker/build-push-action@v5 + uses: docker/build-push-action@ca052bb54ab0790a636c9b5f226502c73d547a25 # v5 with: context: . push: true diff --git a/.gitignore b/.gitignore index fc4871f..1781395 100644 --- a/.gitignore +++ b/.gitignore @@ -31,16 +31,24 @@ Thumbs.db # Environment .env .env.local +*.key + +# Coverage (explicit) +coverage.out + +# Node +node_modules/ # Dashboard build output dashboard/svelte/build/ -dashboard/svelte/node_modules/ dashboard/svelte/.svelte-kit/ # Temporary files tmp/ .tools/ +# Runtime data +.oberwatch/ oberwatch *.db *.db-shm diff --git a/BRANCHING.md b/BRANCHING.md index 84d9766..8fddc91 100644 --- a/BRANCHING.md +++ b/BRANCHING.md @@ -1,13 +1,13 @@ # Branching Strategy -Oberwatch uses GitHub Flow with a protected `staging` branch in front of `main`. +Oberwatch uses a protected `staging` branch in front of `main`. ## Branch Overview -- `main` is the production-ready branch. Every merge to `main` should be stable and eligible for release. +- `main` is the beta / pre-production branch. Every merge to `main` should be stable enough for preview testing and eligible for a tagged release. - `staging` is the integration branch. Feature work lands here first for combined testing. - `feature/*` branches are short-lived branches created from `staging` for normal work. -- `hotfix/*` branches are short-lived branches created from `main` for urgent production fixes. +- `hotfix/*` branches are short-lived branches created from `main` for urgent fixes that must be promoted quickly toward the next beta or tagged release. ## Feature Workflow @@ -32,10 +32,10 @@ Every pull request to `staging` must pass CI before merge. When `staging` is stable, open a pull request from `staging` to `main`. - Merge to `main` only after CI passes. -- Use squash merge to keep history clean. -- Treat every merge to `main` as production-ready. +- Use a merge commit so `main` preserves the exact tested integration state from `staging`. +- Treat every merge to `main` as beta / pre-production ready, not as the final stable release. -Merges to `main` publish the `latest` container image. Merges to `staging` publish the `staging` container image. +Merges to `main` publish the `beta` container image. Merges to `staging` publish the `staging` container image. ## Tagged Releases @@ -43,7 +43,7 @@ Tagged releases are created from `main` only. - Create a tag like `v0.1.0` on `main`. - Pushing the tag runs the release workflow. -- The release workflow publishes binaries plus container images to GHCR and Docker Hub. +- The release workflow publishes binaries plus stable container images to GHCR and Docker Hub. - Container tags include the full version, the minor version alias, and `latest`. ## Hotfix Workflow @@ -62,6 +62,8 @@ Open the hotfix pull request against `main`. After it merges, cherry-pick the sa - Create `feature/*` branches from `staging`. - Open normal pull requests into `staging`. +- Use squash merge for normal feature PRs into `staging`. - Promote `staging` into `main` when the integration branch is stable. +- Use a merge commit for `staging` -> `main` promotion PRs. - Create release tags from `main` only. - Create `hotfix/*` branches from `main`, merge them to `main`, then backport to `staging`. diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ca73c1..687185b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,10 +7,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.1.0] - 2026-03-30 + ### Added -- Initial project scaffolding -- Go module setup (`github.com/OberWatch/oberwatch`) -- Package structure: proxy, budget, trace, eval, config, dashboard, alert, provider, pricing -- Makefile with build, test, lint targets -- CI/CD pipeline configuration -- Community files (CONTRIBUTING, CODE_OF_CONDUCT, SECURITY) +- Reverse proxy for OpenAI, Anthropic, Google, and Ollama APIs +- Streaming (SSE) passthrough with accurate token counting +- Agent auto-registration on first proxied request via X-Oberwatch-Agent header +- Per-agent budget enforcement with configurable actions: reject, downgrade, alert, kill +- Model auto-downgrade chains when budget threshold is reached +- Global budget cap with configurable period (hourly, daily, weekly, monthly) +- Runaway detection with configurable request rate window and auto-kill +- Emergency stop with resume — pauses all agent proxy traffic without affecting dashboard or API +- Agent management via dashboard: rename, edit budgets, kill, enable, reset +- Budget persistence in SQLite with periodic in-memory to disk flush +- Webhook and Slack alert integrations with deduplication +- Dashboard: Overview, Agents, Costs, and Settings pages +- Real-time dashboard updates via Server-Sent Events +- First-run onboarding wizard with admin account creation +- Session-based authentication with 24h expiry and password change +- Built-in model pricing tables for OpenAI, Anthropic, and Google models +- One-line install script for Linux and macOS with systemd service setup +- Docker support with named volume persistence +- Docker Compose template with commented enterprise service placeholders +- GHCR image publishing: beta (from main), staging (from staging branch) +- Docker Hub image publishing on tagged releases +- CI/CD: lint, test, 80% coverage gate, dashboard build, automated releases via GoReleaser +- Docker volume mount detection with startup warning when data is not persisted diff --git a/Dockerfile b/Dockerfile index f3f88c5..2fb1179 100644 --- a/Dockerfile +++ b/Dockerfile @@ -20,11 +20,14 @@ RUN CGO_ENABLED=1 go build -ldflags="-s -w -X main.channel=${CHANNEL} -X main.co # Runtime stage FROM alpine:3.20 -RUN apk add --no-cache ca-certificates sqlite-libs +RUN apk add --no-cache ca-certificates sqlite-libs && \ + addgroup -S oberwatch && adduser -S oberwatch -G oberwatch && \ + mkdir -p /data && chown oberwatch:oberwatch /data COPY --from=builder /app/oberwatch /usr/local/bin/oberwatch EXPOSE 8080 VOLUME ["/data"] +USER oberwatch ENTRYPOINT ["oberwatch"] CMD ["serve"] diff --git a/README.md b/README.md index 0803891..aeb45e5 100644 --- a/README.md +++ b/README.md @@ -78,9 +78,12 @@ Then open `http://localhost:8080` to complete setup. ### Docker ```bash -docker run -d -p 8080:8080 -v oberwatch-data:/data ghcr.io/oberwatch/oberwatch:latest +docker run -d -p 8080:8080 -v oberwatch-data:/data ghcr.io/oberwatch/oberwatch:beta ``` +For preview testing, use the beta channel instead of the staging channel: + + ### Docker Compose (for Enterprise Edition) ```bash diff --git a/docs/production.md b/docs/production.md index b3d5a64..9411316 100644 --- a/docs/production.md +++ b/docs/production.md @@ -1,3 +1,124 @@ -# Production Deployment +# Production Deployment Guide -TODO: document reverse proxy, TLS termination, persistent storage, backups, and hardened deployment patterns for Oberwatch. +## Reverse Proxy with TLS + +> ⚠️ Never expose Oberwatch without TLS in production. +> API keys pass through the proxy in plain text request headers. +> Always terminate TLS in front of Oberwatch. + +### Caddy (recommended — automatic TLS) + +Install Caddy, then create `/etc/caddy/Caddyfile`: + +```caddy +oberwatch.yourdomain.com { + reverse_proxy localhost:8080 +} +``` + +Caddy automatically provisions and renews Let's Encrypt certificates. +Start with: `sudo systemctl enable --now caddy` + +### Nginx + +```nginx +server { + listen 443 ssl http2; + server_name oberwatch.yourdomain.com; + + ssl_certificate /etc/letsencrypt/live/oberwatch.yourdomain.com/fullchain.pem; + ssl_certificate_key /etc/letsencrypt/live/oberwatch.yourdomain.com/privkey.pem; + + location / { + proxy_pass http://127.0.0.1:8080; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + + # Required for SSE (Server-Sent Events) + proxy_buffering off; + proxy_cache off; + proxy_read_timeout 86400s; + proxy_http_version 1.1; + proxy_set_header Connection ''; + } +} + +server { + listen 80; + server_name oberwatch.yourdomain.com; + return 301 https://$host$request_uri; +} +``` + +## Firewall + +Only expose port 443 publicly. Keep port 8080 on localhost only. + +```bash +sudo ufw allow 443/tcp +sudo ufw allow 22/tcp +sudo ufw deny 8080/tcp +sudo ufw enable +``` + +## Persistent Storage + +### Binary install +Data lives in `~/.oberwatch/data/oberwatch.db` by default. + +### Docker +Always use a named volume. Without `-v`, data is lost on container removal: + +```bash +docker run -d -p 8080:8080 -v oberwatch-data:/data ghcr.io/oberwatch/oberwatch:latest +``` + +## Backups + +SQLite supports hot backups — copy the file while Oberwatch is running: + +```bash +# Binary install +cp ~/.oberwatch/data/oberwatch.db ~/backups/oberwatch-$(date +%Y%m%d).db + +# Docker +docker cp oberwatch:/data/oberwatch.db ~/backups/oberwatch-$(date +%Y%m%d).db +``` + +Automate with a daily cron job. + +## Upgrading + +### Binary +Re-run the install script. It upgrades the binary without touching config or data: + +```bash +curl -fsSL https://raw.githubusercontent.com/OberWatch/oberwatch/main/scripts/install.sh | sh +``` + +### Docker + +```bash +docker pull ghcr.io/oberwatch/oberwatch:latest +docker stop oberwatch && docker rm oberwatch +docker run -d -p 8080:8080 -v oberwatch-data:/data --name oberwatch ghcr.io/oberwatch/oberwatch:latest +``` + +## Resource Requirements + +Oberwatch is lightweight: +- RAM: ~30-50MB baseline +- CPU: minimal — it proxies requests, it doesn't run inference +- Disk: ~500MB-1GB for a busy deployment (100K requests/day, 7-day retention) + +## Health Monitoring + +Poll the health endpoint from your uptime monitor: + +```bash +curl http://localhost:8080/_oberwatch/api/v1/health +``` + +Returns HTTP 200 with status `"ok"` when healthy. Includes `emergency_stop` state and provider reachability. diff --git a/go.mod b/go.mod index d0d97b0..fd8f5d8 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/BurntSushi/toml v1.6.0 github.com/mattn/go-sqlite3 v1.14.22 github.com/spf13/cobra v1.10.2 - golang.org/x/crypto v0.44.0 + golang.org/x/crypto v0.45.0 ) require ( diff --git a/go.sum b/go.sum index c3bf99a..34985a4 100644 --- a/go.sum +++ b/go.sum @@ -11,6 +11,6 @@ github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiT github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY= github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= -golang.org/x/crypto v0.44.0 h1:A97SsFvM3AIwEEmTBiaxPPTYpDC47w720rdiiUvgoAU= -golang.org/x/crypto v0.44.0/go.mod h1:013i+Nw79BMiQiMsOPcVCB5ZIJbYkerPrGnOa00tvmc= +golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q= +golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/internal/budget/manager_test.go b/internal/budget/manager_test.go index becb45e..7c7636f 100644 --- a/internal/budget/manager_test.go +++ b/internal/budget/manager_test.go @@ -851,6 +851,383 @@ func TestRecordSpend_NegativeIgnored(t *testing.T) { } } +func TestRenameAgent(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + oldName string + newName string + wantError bool + }{ + {name: "same name is no-op", oldName: "agent-a", newName: "agent-a", wantError: false}, + {name: "rename to new name succeeds", oldName: "agent-a", newName: "agent-renamed", wantError: false}, + {name: "rename to existing name fails", oldName: "agent-a", newName: "agent-b", wantError: true}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + cfg := baseGateConfig() + store, err := storage.NewSQLiteStore(filepath.Join(t.TempDir(), "rename.db"), 0, nil) + if err != nil { + t.Fatalf("NewSQLiteStore() error = %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + + clock := newMockClock(time.Date(2026, time.March, 26, 12, 0, 0, 0, time.UTC)) + manager, err := NewPersistentManagerWithClockAndDispatcher(cfg, nil, store, clock, nil) + if err != nil { + t.Fatalf("NewPersistentManagerWithClockAndDispatcher() error = %v", err) + } + t.Cleanup(func() { _ = manager.Close() }) + + manager.RecordSpend("agent-a", 2.5) + manager.RecordSpend("agent-b", 1.0) + if flushErr := manager.Flush(context.Background()); flushErr != nil { + t.Fatalf("Flush() error = %v", flushErr) + } + + if tt.oldName == tt.newName { + if renameErr := manager.RenameAgent(context.Background(), tt.oldName, tt.newName); renameErr != nil { + t.Fatalf("RenameAgent(same) error = %v", renameErr) + } + return + } + + renameErr := manager.RenameAgent(context.Background(), tt.oldName, tt.newName) + if tt.wantError { + if renameErr == nil { + t.Fatal("RenameAgent() error = nil, want non-nil") + } + return + } + if renameErr != nil { + t.Fatalf("RenameAgent() error = %v", renameErr) + } + + view := manager.GetBudget(tt.newName) + if view.SpentUSD != 2.5 { + t.Fatalf("renamed agent spent = %v, want 2.5", view.SpentUSD) + } + oldView := manager.GetBudget(tt.oldName) + if oldView.SpentUSD != 0 { + t.Fatalf("old agent spent = %v, want 0", oldView.SpentUSD) + } + }) + } +} + +func TestCloneState(t *testing.T) { + t.Parallel() + + //nolint:govet // keep test table readable. + tests := []struct { + name string + state *agentState + }{ + {name: "nil state", state: nil}, + { + name: "populated state", + state: &agentState{ + spentUSD: 5.0, + lastAlertedPct: 50, + periodStartedAt: time.Date(2026, time.March, 26, 12, 0, 0, 0, time.UTC), + periodResetsAt: time.Date(2026, time.March, 27, 12, 0, 0, 0, time.UTC), + lastSeenAt: time.Date(2026, time.March, 26, 13, 0, 0, 0, time.UTC), + requestTimes: []time.Time{time.Date(2026, time.March, 26, 12, 30, 0, 0, time.UTC)}, + triggeredAlerts: map[float64]bool{50: true}, + killed: true, + disableReason: disableReasonManualKill, + dirty: true, + }, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + cloned := cloneState(tt.state) + if cloned == nil { + t.Fatal("cloneState() returned nil") + } + if tt.state == nil { + if len(cloned.triggeredAlerts) != 0 { + t.Fatalf("cloneState(nil).triggeredAlerts len = %d, want 0", len(cloned.triggeredAlerts)) + } + return + } + if cloned.spentUSD != tt.state.spentUSD { + t.Fatalf("cloned.spentUSD = %v, want %v", cloned.spentUSD, tt.state.spentUSD) + } + if cloned.killed != tt.state.killed { + t.Fatalf("cloned.killed = %v, want %v", cloned.killed, tt.state.killed) + } + if cloned.disableReason != tt.state.disableReason { + t.Fatalf("cloned.disableReason = %q, want %q", cloned.disableReason, tt.state.disableReason) + } + // Verify deep copy: mutate original, cloned should not change. + tt.state.triggeredAlerts[80] = true + if _, found := cloned.triggeredAlerts[80]; found { + t.Fatal("cloned triggeredAlerts shares reference with original") + } + }) + } +} + +func TestPersistedDisableReason(t *testing.T) { + t.Parallel() + + //nolint:govet // keep test table readable. + tests := []struct { + name string + status string + limitUSD float64 + spentUSD float64 + actionOnExceed config.BudgetAction + want string + }{ + {name: "budget_exceeded status", status: "budget_exceeded", want: disableReasonBudgetExceeded}, + {name: "manual_kill status", status: "manual_kill", want: disableReasonManualKill}, + {name: "runaway_detected status", status: "runaway_detected", want: disableReasonRunaway}, + {name: "killed with budget exceeded", status: "killed", limitUSD: 10, spentUSD: 10, actionOnExceed: config.BudgetActionKill, want: disableReasonBudgetExceeded}, + {name: "killed manual", status: "killed", limitUSD: 10, spentUSD: 5, actionOnExceed: config.BudgetActionKill, want: disableReasonManualKill}, + {name: "active status", status: "active", want: disableReasonNone}, + {name: "empty status", status: "", want: disableReasonNone}, + {name: "unknown status", status: "something_else", want: disableReasonNone}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got := persistedDisableReason(tt.status, tt.limitUSD, tt.spentUSD, tt.actionOnExceed) + if got != tt.want { + t.Fatalf("persistedDisableReason() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestPersistedStatusForState(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + state *agentState + want string + }{ + {name: "nil state", state: nil, want: "active"}, + {name: "not killed", state: &agentState{killed: false}, want: "active"}, + {name: "killed with reason", state: &agentState{killed: true, disableReason: disableReasonRunaway}, want: disableReasonRunaway}, + {name: "killed no reason", state: &agentState{killed: true, disableReason: ""}, want: "killed"}, + {name: "killed whitespace reason", state: &agentState{killed: true, disableReason: " "}, want: "killed"}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got := persistedStatusForState(tt.state) + if got != tt.want { + t.Fatalf("persistedStatusForState() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestFirstNonZeroTime(t *testing.T) { + t.Parallel() + + fallback := time.Date(2026, time.March, 26, 12, 0, 0, 0, time.UTC) + value := time.Date(2026, time.March, 25, 6, 0, 0, 0, time.UTC) + + //nolint:govet // keep test table readable. + tests := []struct { + name string + value time.Time + fallback time.Time + want time.Time + }{ + {name: "zero value uses fallback", value: time.Time{}, fallback: fallback, want: fallback.UTC()}, + {name: "non-zero value used", value: value, fallback: fallback, want: value.UTC()}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got := firstNonZeroTime(tt.value, tt.fallback) + if !got.Equal(tt.want) { + t.Fatalf("firstNonZeroTime() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestTriggeredAlerts(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + thresholds []float64 + limitUSD float64 + spentUSD float64 + wantFired int + }{ + {name: "no thresholds", thresholds: nil, limitUSD: 10, spentUSD: 5, wantFired: 0}, + {name: "none crossed", thresholds: []float64{50, 80}, limitUSD: 10, spentUSD: 4, wantFired: 0}, + {name: "one crossed", thresholds: []float64{50, 80}, limitUSD: 10, spentUSD: 6, wantFired: 1}, + {name: "all crossed", thresholds: []float64{50, 80, 100}, limitUSD: 10, spentUSD: 10, wantFired: 3}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got := triggeredAlerts(tt.thresholds, tt.limitUSD, tt.spentUSD) + if len(got) != tt.wantFired { + t.Fatalf("triggeredAlerts() fired = %d, want %d", len(got), tt.wantFired) + } + }) + } +} + +func TestToBudgetView_NegativeRemaining(t *testing.T) { + t.Parallel() + + policy := agentPolicy{limitUSD: 5, period: config.BudgetPeriodDaily, actionOnExceed: config.BudgetActionReject} + state := &agentState{spentUSD: 8, triggeredAlerts: make(map[float64]bool)} + + view := toBudgetView("over-agent", policy, state) + if view.RemainingUSD != 0 { + t.Fatalf("toBudgetView().RemainingUSD = %v, want 0", view.RemainingUSD) + } + if view.Status != "active" { + t.Fatalf("toBudgetView().Status = %q, want %q", view.Status, "active") + } +} + +func TestPersistentManagerWithClockAndDispatcher(t *testing.T) { + t.Parallel() + + cfg := baseGateConfig() + store, err := storage.NewSQLiteStore(filepath.Join(t.TempDir(), "dispatch.db"), 0, nil) + if err != nil { + t.Fatalf("NewSQLiteStore() error = %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + + clock := newMockClock(time.Date(2026, time.March, 26, 12, 0, 0, 0, time.UTC)) + dispatcher := &capturingDispatcher{} + manager, err := NewPersistentManagerWithClockAndDispatcher(cfg, nil, store, clock, dispatcher) + if err != nil { + t.Fatalf("NewPersistentManagerWithClockAndDispatcher() error = %v", err) + } + t.Cleanup(func() { _ = manager.Close() }) + + manager.RecordSpend("agent-a", 5.1) // 51% — crosses 50% threshold + events := dispatcher.snapshot() + foundThreshold := false + for _, e := range events { + if e.Type == alert.TypeBudgetThreshold { + foundThreshold = true + } + } + if !foundThreshold { + t.Fatal("expected budget_threshold alert, got none") + } + + if flushErr := manager.Flush(context.Background()); flushErr != nil { + t.Fatalf("Flush() error = %v", flushErr) + } + record, found, getErr := store.GetAgent(context.Background(), "agent-a") + if getErr != nil { + t.Fatalf("GetAgent() error = %v", getErr) + } + if !found { + t.Fatal("GetAgent() found = false, want true") + } + if record.BudgetSpentUSD != 5.1 { + t.Fatalf("persisted BudgetSpentUSD = %v, want 5.1", record.BudgetSpentUSD) + } +} + +func TestLoadPersistedAgents_RestoresState(t *testing.T) { + t.Parallel() + + cfg := baseGateConfig() + dbPath := filepath.Join(t.TempDir(), "load.db") + store, err := storage.NewSQLiteStore(dbPath, 0, nil) + if err != nil { + t.Fatalf("NewSQLiteStore() error = %v", err) + } + + clock := newMockClock(time.Date(2026, time.March, 26, 12, 0, 0, 0, time.UTC)) + + // Create first manager, add state, flush, close. + m1, err := NewPersistentManagerWithClockAndDispatcher(cfg, nil, store, clock, nil) + if err != nil { + t.Fatalf("NewPersistentManagerWithClockAndDispatcher(m1) error = %v", err) + } + m1.RecordSpend("persisted-agent", 7.5) + m1.KillAgent("killed-agent") + if flushErr := m1.Flush(context.Background()); flushErr != nil { + t.Fatalf("m1.Flush() error = %v", flushErr) + } + _ = m1.Close() + _ = store.Close() + + // Reopen store and create new manager — should load persisted state. + store2, err := storage.NewSQLiteStore(dbPath, 0, nil) + if err != nil { + t.Fatalf("NewSQLiteStore(reopen) error = %v", err) + } + t.Cleanup(func() { _ = store2.Close() }) + + m2, err := NewPersistentManagerWithClockAndDispatcher(cfg, nil, store2, clock, nil) + if err != nil { + t.Fatalf("NewPersistentManagerWithClockAndDispatcher(m2) error = %v", err) + } + t.Cleanup(func() { _ = m2.Close() }) + + view := m2.GetBudget("persisted-agent") + if view.SpentUSD != 7.5 { + t.Fatalf("loaded SpentUSD = %v, want 7.5", view.SpentUSD) + } + + killedView := m2.GetBudget("killed-agent") + if killedView.Status != "killed" { + t.Fatalf("loaded killed agent status = %q, want %q", killedView.Status, "killed") + } +} + +func TestFlushAgentIfNeeded_EmptyAgent(t *testing.T) { + t.Parallel() + + cfg := baseGateConfig() + store, err := storage.NewSQLiteStore(filepath.Join(t.TempDir(), "flush-empty.db"), 0, nil) + if err != nil { + t.Fatalf("NewSQLiteStore() error = %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + + clock := newMockClock(time.Date(2026, time.March, 26, 12, 0, 0, 0, time.UTC)) + manager, err := NewPersistentManagerWithClockAndDispatcher(cfg, nil, store, clock, nil) + if err != nil { + t.Fatalf("NewPersistentManagerWithClockAndDispatcher() error = %v", err) + } + t.Cleanup(func() { _ = manager.Close() }) + + // Should not panic or error for empty/whitespace agent. + manager.flushAgentIfNeeded("") + manager.flushAgentIfNeeded(" ") +} + func TestBudgetViewsAndMutations(t *testing.T) { t.Parallel() diff --git a/internal/config/config.go b/internal/config/config.go index 88bd6b5..cb23e1d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -61,6 +61,7 @@ type UpstreamConfig struct { DefaultProvider ProviderConfigName `toml:"default_provider"` OpenAI ProviderEndpoint `toml:"openai"` Anthropic ProviderEndpoint `toml:"anthropic"` + Google ProviderEndpoint `toml:"google"` Ollama ProviderEndpoint `toml:"ollama"` Custom ProviderEndpoint `toml:"custom"` Timeout string `toml:"timeout"` @@ -244,6 +245,7 @@ func DefaultConfig() Config { DefaultProvider: ProviderOpenAI, OpenAI: ProviderEndpoint{BaseURL: "https://api.openai.com"}, Anthropic: ProviderEndpoint{BaseURL: "https://api.anthropic.com"}, + Google: ProviderEndpoint{BaseURL: "https://generativelanguage.googleapis.com/v1beta/openai"}, Ollama: ProviderEndpoint{BaseURL: "http://localhost:11434"}, Custom: ProviderEndpoint{}, Timeout: "120s", diff --git a/internal/config/init.go b/internal/config/init.go index e011b54..26239ab 100644 --- a/internal/config/init.go +++ b/internal/config/init.go @@ -55,7 +55,7 @@ tls_key = "" # ----------------------------------------------------------------------------- [upstream] # Default upstream provider when auto-detection is ambiguous. -# Options: "openai", "anthropic", "ollama", "custom" +# Options: "openai", "anthropic", "google", "ollama", "custom" # Default: "openai" default_provider = "openai" @@ -73,6 +73,11 @@ base_url = "https://api.openai.com" # Default: "https://api.anthropic.com" base_url = "https://api.anthropic.com" +[upstream.google] +# Base URL for Google Gemini OpenAI-compatible API. +# Default: "https://generativelanguage.googleapis.com/v1beta/openai" +base_url = "https://generativelanguage.googleapis.com/v1beta/openai" + [upstream.ollama] # Base URL for Ollama. # Default: "http://localhost:11434" diff --git a/internal/proxy/provider_routing.go b/internal/proxy/provider_routing.go new file mode 100644 index 0000000..ec3d14f --- /dev/null +++ b/internal/proxy/provider_routing.go @@ -0,0 +1,475 @@ +package proxy + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "strings" + "time" + + "github.com/OberWatch/oberwatch/internal/config" + "github.com/OberWatch/oberwatch/internal/pricing" + "github.com/OberWatch/oberwatch/internal/storage" +) + +const providerOverrideHeader = "X-Oberwatch-Provider" + +type resolvedRouteContextKey struct{} + +type resolvedRoute struct { + provider config.ProviderConfigName + anthropicCompat bool +} + +func resolvedRouteFromContext(ctx context.Context) (resolvedRoute, bool) { + route, ok := ctx.Value(resolvedRouteContextKey{}).(resolvedRoute) + return route, ok +} + +func providerRoutingMiddleware(defaultProvider config.ProviderConfigName, targets map[config.ProviderConfigName]*url.URL, logger *slog.Logger) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !isKnownProxyPath(r.URL.Path) { + next.ServeHTTP(w, r) + return + } + + var requestBody []byte + if r.Body != nil { + body, err := io.ReadAll(r.Body) + if err != nil { + writeConfigError(w, fmt.Sprintf("read request body: %v", err)) + return + } + requestBody = body + r.Body = io.NopCloser(bytes.NewReader(requestBody)) + r.ContentLength = int64(len(requestBody)) + r.GetBody = func() (io.ReadCloser, error) { + return io.NopCloser(bytes.NewReader(requestBody)), nil + } + } + + route, err := resolveProvider(r, requestBody, defaultProvider) + if err != nil { + writeProviderResolutionError(w, err.Error(), http.StatusBadRequest) + return + } + if _, ok := targets[route.provider]; !ok { + writeProviderResolutionError(w, fmt.Sprintf("provider %q is not configured", route.provider), http.StatusBadRequest) + return + } + if logger != nil { + logger.Debug("resolved upstream provider", "path", r.URL.Path, "provider", route.provider, "anthropic_compat", route.anthropicCompat) + } + + *r = *r.WithContext(context.WithValue(r.Context(), resolvedRouteContextKey{}, route)) + next.ServeHTTP(w, r) + }) + } +} + +func resolveProvider(r *http.Request, requestBody []byte, defaultProvider config.ProviderConfigName) (resolvedRoute, error) { + normalized := strings.TrimSuffix(strings.TrimSpace(r.URL.Path), "/") + if normalized == "" { + normalized = "/" + } + + switch { + case normalized == "/v1/messages": + return resolvedRoute{provider: config.ProviderAnthropic}, nil + case strings.HasPrefix(normalized, "/api/"): + return resolvedRoute{provider: config.ProviderOllama}, nil + case normalized == "/v1/models": + provider, ok := parseProviderOverride(r) + if !ok { + return resolvedRoute{}, fmt.Errorf("%s header is required for /v1/models", providerOverrideHeader) + } + return resolvedRoute{provider: provider}, nil + case isOpenAICompatiblePath(normalized): + if provider, ok := parseProviderOverride(r); ok { + route := resolvedRoute{provider: provider} + if provider == config.ProviderAnthropic { + if normalized != "/v1/chat/completions" { + return resolvedRoute{}, fmt.Errorf("anthropic compatibility only supports /v1/chat/completions") + } + route.anthropicCompat = true + } + return route, nil + } + + model, _ := extractModelAndStream(requestBody) + if model == "" { + return resolvedRoute{}, fmt.Errorf("could not resolve provider: request model is required") + } + + provider, ok := detectProviderForModel(model) + if !ok { + return resolvedRoute{}, fmt.Errorf("could not resolve provider for model %q", model) + } + if provider == config.ProviderAnthropic { + if normalized != "/v1/chat/completions" { + return resolvedRoute{}, fmt.Errorf("anthropic compatibility only supports /v1/chat/completions") + } + return resolvedRoute{provider: provider, anthropicCompat: true}, nil + } + return resolvedRoute{provider: provider}, nil + default: + if defaultProvider == "" { + return resolvedRoute{}, fmt.Errorf("could not resolve provider for path %q", r.URL.Path) + } + return resolvedRoute{provider: defaultProvider}, nil + } +} + +func parseProviderOverride(r *http.Request) (config.ProviderConfigName, bool) { + raw := strings.TrimSpace(strings.ToLower(r.Header.Get(providerOverrideHeader))) + switch config.ProviderConfigName(raw) { + case config.ProviderOpenAI, config.ProviderAnthropic, config.ProviderGoogle, config.ProviderOllama, config.ProviderCustom: + return config.ProviderConfigName(raw), true + default: + return "", false + } +} + +func detectProviderForModel(model string) (config.ProviderConfigName, bool) { + normalized := strings.ToLower(strings.TrimSpace(model)) + switch { + case strings.HasPrefix(normalized, "claude-"): + return config.ProviderAnthropic, true + case strings.HasPrefix(normalized, "gemini-"): + return config.ProviderGoogle, true + case strings.HasPrefix(normalized, "gpt-") || strings.HasPrefix(normalized, "o1") || strings.HasPrefix(normalized, "o3") || strings.HasPrefix(normalized, "o4") || strings.HasPrefix(normalized, "text-embedding-"): + return config.ProviderOpenAI, true + case strings.HasPrefix(normalized, "llama"), strings.HasPrefix(normalized, "mistral"), strings.HasPrefix(normalized, "mixtral"), strings.HasPrefix(normalized, "qwen"), strings.HasPrefix(normalized, "phi"), strings.HasPrefix(normalized, "gemma"), strings.HasPrefix(normalized, "deepseek"), strings.HasPrefix(normalized, "codellama"), strings.HasPrefix(normalized, "dolphin"), strings.HasPrefix(normalized, "orca"), strings.HasPrefix(normalized, "wizard"), strings.HasPrefix(normalized, "vicuna"), strings.HasPrefix(normalized, "solar"), strings.HasPrefix(normalized, "granite"), strings.HasPrefix(normalized, "smollm"): + return config.ProviderOllama, true + default: + return "", false + } +} + +func isOpenAICompatiblePath(path string) bool { + switch strings.TrimSuffix(strings.TrimSpace(path), "/") { + case "/v1/chat/completions", "/v1/completions", "/v1/embeddings": + return true + default: + return false + } +} + +func writeProviderResolutionError(w http.ResponseWriter, message string, statusCode int) { + payload, err := json.Marshal(map[string]any{ + "error": map[string]any{ + "code": "provider_resolution_failed", + "message": message, + }, + }) + if err != nil { + http.Error(w, message, statusCode) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(statusCode) + _, _ = w.Write(payload) +} + +func serveAnthropicOpenAICompat(w http.ResponseWriter, r *http.Request, route resolvedRoute, targets map[config.ProviderConfigName]*url.URL, hooks Hooks) { + target := targets[route.provider] + if target == nil { + writeProviderResolutionError(w, fmt.Sprintf("provider %q is not configured", route.provider), http.StatusBadRequest) + return + } + + requestBody, err := io.ReadAll(r.Body) + if err != nil { + writeConfigError(w, fmt.Sprintf("read request body: %v", err)) + return + } + + anthropicBody, model, streaming, err := openAIChatToAnthropic(requestBody) + if err != nil { + writeProviderResolutionError(w, err.Error(), http.StatusBadRequest) + return + } + if streaming { + writeProviderResolutionError(w, "anthropic OpenAI SDK compatibility currently supports non-streaming chat completions only", http.StatusBadRequest) + return + } + + upstreamURL := *target + upstreamURL.Path = "/v1/messages" + upstreamURL.RawQuery = r.URL.RawQuery + + upstreamReq, err := http.NewRequestWithContext(r.Context(), http.MethodPost, upstreamURL.String(), bytes.NewReader(anthropicBody)) + if err != nil { + writeConfigError(w, fmt.Sprintf("build anthropic compatibility request: %v", err)) + return + } + copyCompatibilityHeaders(upstreamReq.Header, r.Header) + upstreamReq.Header.Set("Content-Type", "application/json") + if upstreamReq.Header.Get("anthropic-version") == "" { + upstreamReq.Header.Set("anthropic-version", "2023-06-01") + } + if apiKey := extractBearerToken(r); apiKey != "" { + upstreamReq.Header.Set("x-api-key", apiKey) + } else if apiKey := strings.TrimSpace(r.Header.Get("x-api-key")); apiKey != "" { + upstreamReq.Header.Set("x-api-key", apiKey) + } + + resp, err := http.DefaultClient.Do(upstreamReq) + if err != nil { + http.Error(w, fmt.Sprintf("upstream proxy error: %v", err), http.StatusBadGateway) + return + } + defer func() { _ = resp.Body.Close() }() + + body, err := io.ReadAll(resp.Body) + if err != nil { + http.Error(w, fmt.Sprintf("read upstream response: %v", err), http.StatusBadGateway) + return + } + + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + copyCompatibilityResponseHeaders(w.Header(), resp.Header) + w.WriteHeader(resp.StatusCode) + _, _ = w.Write(body) + return + } + + translated, err := anthropicToOpenAIChatCompletion(body, model) + if err != nil { + writeConfigError(w, fmt.Sprintf("translate anthropic response: %v", err)) + return + } + + if hooks.Budget != nil && hooks.Pricing != nil { + if meta, ok := r.Context().Value(budgetContextKey{}).(budgetRequestMeta); ok { + usage := pricing.ExtractUsageFromResponse(string(config.ProviderAnthropic), body, hooks.Logger) + cost := hooks.Pricing.CalculateCost(meta.model, usage.InputTokens, usage.OutputTokens) + hooks.Budget.RecordSpend(meta.agent, cost) + if hooks.CostSink != nil { + hooks.CostSink.Enqueue(storage.CostRecord{ + Agent: meta.agent, + Model: meta.model, + Provider: meta.provider, + TraceID: meta.traceID, + TaskID: meta.taskID, + OriginalModel: meta.originalModel, + InputTokens: usage.InputTokens, + OutputTokens: usage.OutputTokens, + CostUSD: cost, + Downgraded: meta.downgraded, + CreatedAt: time.Now().UTC(), + }) + } + } + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(translated) +} + +func openAIChatToAnthropic(raw []byte) ([]byte, string, bool, error) { + //nolint:govet // local decode structs mirror the external payload shape. + var request struct { + Model string `json:"model"` + //nolint:govet // local decode structs mirror the external payload shape. + Messages []struct { + Role string `json:"role"` + Content any `json:"content"` + } `json:"messages"` + MaxTokens int `json:"max_tokens"` + Temperature *float64 `json:"temperature,omitempty"` + Stream bool `json:"stream"` + } + if err := json.Unmarshal(raw, &request); err != nil { + return nil, "", false, fmt.Errorf("decode OpenAI chat completion request: %w", err) + } + model := strings.TrimSpace(request.Model) + if model == "" { + return nil, "", false, fmt.Errorf("request model is required") + } + + var systemParts []string + messages := make([]map[string]any, 0, len(request.Messages)) + for _, message := range request.Messages { + content := extractMessageContentText(message.Content) + if content == "" { + continue + } + if strings.EqualFold(message.Role, "system") { + systemParts = append(systemParts, content) + continue + } + messages = append(messages, map[string]any{ + "role": message.Role, + "content": content, + }) + } + if len(messages) == 0 { + return nil, "", false, fmt.Errorf("at least one non-system message is required") + } + + maxTokens := request.MaxTokens + if maxTokens <= 0 { + maxTokens = 1024 + } + + payload := map[string]any{ + "model": model, + "messages": messages, + "max_tokens": maxTokens, + } + if len(systemParts) > 0 { + payload["system"] = strings.Join(systemParts, "\n\n") + } + if request.Temperature != nil { + payload["temperature"] = *request.Temperature + } + encoded, err := json.Marshal(payload) + if err != nil { + return nil, "", false, fmt.Errorf("encode anthropic compatibility request: %w", err) + } + return encoded, model, request.Stream, nil +} + +func anthropicToOpenAIChatCompletion(raw []byte, model string) ([]byte, error) { + var response struct { + ID string `json:"id"` + Model string `json:"model"` + StopReason string `json:"stop_reason"` + Content []struct { + Type string `json:"type"` + Text string `json:"text"` + } `json:"content"` + Usage struct { + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + } `json:"usage"` + } + if err := json.Unmarshal(raw, &response); err != nil { + return nil, fmt.Errorf("decode anthropic response: %w", err) + } + if strings.TrimSpace(response.Model) != "" { + model = response.Model + } + + var contentBuilder strings.Builder + for _, block := range response.Content { + if block.Type != "text" { + continue + } + contentBuilder.WriteString(block.Text) + } + + payload := map[string]any{ + "id": response.ID, + "object": "chat.completion", + "created": time.Now().Unix(), + "model": model, + "choices": []map[string]any{ + { + "index": 0, + "message": map[string]any{ + "role": "assistant", + "content": contentBuilder.String(), + }, + "finish_reason": mapAnthropicStopReason(response.StopReason), + }, + }, + "usage": map[string]any{ + "prompt_tokens": response.Usage.InputTokens, + "completion_tokens": response.Usage.OutputTokens, + "total_tokens": response.Usage.InputTokens + response.Usage.OutputTokens, + }, + } + encoded, err := json.Marshal(payload) + if err != nil { + return nil, fmt.Errorf("encode OpenAI-compatible response: %w", err) + } + return encoded, nil +} + +func extractMessageContentText(content any) string { + switch value := content.(type) { + case string: + return strings.TrimSpace(value) + case []any: + parts := make([]string, 0, len(value)) + for _, item := range value { + block, ok := item.(map[string]any) + if !ok { + continue + } + if strings.TrimSpace(fmt.Sprintf("%v", block["type"])) != "text" { + continue + } + text := strings.TrimSpace(fmt.Sprintf("%v", block["text"])) + if text != "" { + parts = append(parts, text) + } + } + return strings.Join(parts, "\n") + default: + return "" + } +} + +func extractBearerToken(r *http.Request) string { + authorization := strings.TrimSpace(r.Header.Get("Authorization")) + if authorization == "" { + return "" + } + parts := strings.Fields(authorization) + if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") { + return "" + } + return strings.TrimSpace(parts[1]) +} + +func copyCompatibilityHeaders(dst http.Header, src http.Header) { + for key, values := range src { + switch { + case strings.EqualFold(key, "Authorization"): + continue + case strings.EqualFold(key, "X-API-Key"): + continue + case strings.HasPrefix(strings.ToLower(key), oberwatchHeaderPrefixLower): + continue + case strings.EqualFold(key, "Content-Length"): + continue + case strings.EqualFold(key, "Host"): + continue + } + for _, value := range values { + dst.Add(key, value) + } + } +} + +func copyCompatibilityResponseHeaders(dst http.Header, src http.Header) { + for key, values := range src { + if strings.EqualFold(key, "Content-Length") { + continue + } + for _, value := range values { + dst.Add(key, value) + } + } +} + +func mapAnthropicStopReason(reason string) string { + switch strings.TrimSpace(reason) { + case "max_tokens": + return "length" + case "tool_use": + return "tool_calls" + default: + return "stop" + } +} diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index 23dba5e..35547a1 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -74,7 +74,10 @@ func New(cfg config.Config, hooks Hooks) (*Server, error) { reverseProxy := &httputil.ReverseProxy{ FlushInterval: -1, Rewrite: func(req *httputil.ProxyRequest) { - targetProvider := detectProvider(req.In.URL.Path, defaultProvider) + targetProvider := defaultProvider + if route, ok := resolvedRouteFromContext(req.In.Context()); ok { + targetProvider = route.provider + } target := targets[targetProvider] req.SetURL(target) @@ -121,11 +124,20 @@ func New(cfg config.Config, hooks Hooks) (*Server, error) { hooks.Dashboard.ServeHTTP(w, r) return } + if route, ok := resolvedRouteFromContext(r.Context()); ok && route.anthropicCompat { + serveAnthropicOpenAICompat(w, r, route, targets, hooks) + return + } reverseProxy.ServeHTTP(w, r) }) - chained := chain(proxyHandler, gateMiddleware(hooks), traceMiddleware(hooks.Trace)) + chained := chain( + proxyHandler, + providerRoutingMiddleware(defaultProvider, targets, hooks.Logger), + gateMiddleware(hooks), + traceMiddleware(hooks.Trace), + ) return &Server{handler: chained}, nil } @@ -146,6 +158,7 @@ func buildTargets(cfg config.UpstreamConfig) (map[config.ProviderConfigName]*url endpoints := []endpoint{ {name: config.ProviderOpenAI, baseURL: cfg.OpenAI.BaseURL}, {name: config.ProviderAnthropic, baseURL: cfg.Anthropic.BaseURL}, + {name: config.ProviderGoogle, baseURL: cfg.Google.BaseURL}, {name: config.ProviderOllama, baseURL: cfg.Ollama.BaseURL}, {name: config.ProviderCustom, baseURL: cfg.Custom.BaseURL}, } @@ -177,22 +190,6 @@ func buildTargets(cfg config.UpstreamConfig) (map[config.ProviderConfigName]*url return targets, nil } -func detectProvider(path string, defaultProvider config.ProviderConfigName) config.ProviderConfigName { - normalized := strings.TrimSuffix(path, "/") - if normalized == "" { - normalized = path - } - - switch normalized { - case "/v1/chat/completions", "/v1/completions": - return config.ProviderOpenAI - case "/v1/messages": - return config.ProviderAnthropic - default: - return defaultProvider - } -} - func shouldServeDashboard(method string, requestPath string) bool { if method != http.MethodGet && method != http.MethodHead { return false @@ -202,6 +199,9 @@ func shouldServeDashboard(method string, requestPath string) bool { func isKnownProxyPath(requestPath string) bool { normalized := strings.TrimSuffix(strings.TrimSpace(requestPath), "/") + if strings.HasPrefix(normalized, "/api/") { + return true + } switch normalized { case "/v1/chat/completions", "/v1/completions", "/v1/embeddings", "/v1/models", "/v1/messages": return true @@ -289,13 +289,16 @@ func gateMiddleware(hooks Hooks) func(http.Handler) http.Handler { meta := budgetRequestMeta{ agent: agent, model: model, - provider: string(detectProvider(r.URL.Path, config.ProviderOpenAI)), + provider: string(config.ProviderOpenAI), traceID: strings.TrimSpace(r.Header.Get("X-Oberwatch-Trace-ID")), taskID: strings.TrimSpace(r.Header.Get("X-Oberwatch-Task")), originalModel: originalModel, streaming: streaming, downgraded: downgraded, } + if route, ok := resolvedRouteFromContext(r.Context()); ok { + meta.provider = string(route.provider) + } *r = *r.WithContext(context.WithValue(r.Context(), budgetContextKey{}, meta)) } next.ServeHTTP(w, r) diff --git a/internal/proxy/proxy_budget_test.go b/internal/proxy/proxy_budget_test.go index b7a6f24..abd6d09 100644 --- a/internal/proxy/proxy_budget_test.go +++ b/internal/proxy/proxy_budget_test.go @@ -110,7 +110,7 @@ func TestServer_BudgetDowngradeRewritesModelAndRecordsSpend(t *testing.T) { cfg.Gate.DefaultBudget.LimitUSD = 10 cfg.Gate.DefaultBudget.ActionOnExceed = config.BudgetActionDowngrade cfg.Gate.DowngradeThresholdPct = 50 - cfg.Gate.DefaultDowngradeChain = []string{"claude-opus-4-6", "claude-sonnet-4-6", "claude-haiku-4-5"} + cfg.Gate.DefaultDowngradeChain = []string{"gpt-4.1", "gpt-4.1-mini", "gpt-4o-mini"} manager := budget.NewManager(cfg.Gate, nil) manager.RecordSpend("email-agent", 5.5) @@ -124,7 +124,7 @@ func TestServer_BudgetDowngradeRewritesModelAndRecordsSpend(t *testing.T) { server := newTestServer(t, proxyServer) t.Cleanup(server.Close) - req, err := http.NewRequest(http.MethodPost, server.URL+"/v1/chat/completions", strings.NewReader(`{"model":"claude-opus-4-6","stream":false,"messages":[{"role":"user","content":"hello"}]}`)) + req, err := http.NewRequest(http.MethodPost, server.URL+"/v1/chat/completions", strings.NewReader(`{"model":"gpt-4.1","stream":false,"messages":[{"role":"user","content":"hello"}]}`)) if err != nil { t.Fatalf("NewRequest() error = %v", err) } @@ -143,7 +143,7 @@ func TestServer_BudgetDowngradeRewritesModelAndRecordsSpend(t *testing.T) { _, _ = io.ReadAll(resp.Body) gotBody := <-capturedBody - if !strings.Contains(gotBody, `"model":"claude-sonnet-4-6"`) { + if !strings.Contains(gotBody, `"model":"gpt-4.1-mini"`) { t.Fatalf("forwarded body = %s, want downgraded model", gotBody) } diff --git a/internal/proxy/proxy_test.go b/internal/proxy/proxy_test.go index 14d6eea..a2cabc6 100644 --- a/internal/proxy/proxy_test.go +++ b/internal/proxy/proxy_test.go @@ -26,39 +26,33 @@ type capturedRequest struct { func TestServer_ProviderDetectionRouting(t *testing.T) { t.Parallel() + //nolint:govet // keep test table fields grouped for readability. tests := []struct { name string path string + requestBody string defaultProvider config.ProviderConfigName wantProvider string + requestHeaders map[string]string }{ - { - name: "chat completions route to openai", - path: "/v1/chat/completions", - defaultProvider: config.ProviderAnthropic, - wantProvider: "openai", - }, - { - name: "completions route to openai", - path: "/v1/completions", - defaultProvider: config.ProviderAnthropic, - wantProvider: "openai", - }, { name: "messages route to anthropic", path: "/v1/messages", + requestBody: `{"model":"claude-haiku-4-5","max_tokens":64,"messages":[{"role":"user","content":"hi"}]}`, defaultProvider: config.ProviderOpenAI, wantProvider: "anthropic", }, { - name: "unknown path uses default provider", + name: "models uses explicit provider header", path: "/v1/models", defaultProvider: config.ProviderAnthropic, - wantProvider: "anthropic", + wantProvider: "openai", + requestHeaders: map[string]string{providerOverrideHeader: string(config.ProviderOpenAI)}, }, { - name: "routes with trailing slash still map correctly", + name: "routes with trailing slash by model family", path: "/v1/chat/completions/", + requestBody: `{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hello"}]}`, defaultProvider: config.ProviderAnthropic, wantProvider: "openai", }, @@ -82,7 +76,21 @@ func TestServer_ProviderDetectionRouting(t *testing.T) { })) t.Cleanup(anthropicServer.Close) + googleServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + received <- "google" + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(googleServer.Close) + + ollamaServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + received <- "ollama" + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(ollamaServer.Close) + cfg := testConfig(openAIServer.URL, anthropicServer.URL) + cfg.Upstream.Google.BaseURL = googleServer.URL + cfg.Upstream.Ollama.BaseURL = ollamaServer.URL cfg.Upstream.DefaultProvider = tt.defaultProvider proxyServer, err := New(cfg, Hooks{}) if err != nil { @@ -92,10 +100,13 @@ func TestServer_ProviderDetectionRouting(t *testing.T) { server := newTestServer(t, proxyServer) t.Cleanup(server.Close) - req, err := http.NewRequest(http.MethodPost, server.URL+tt.path, strings.NewReader(`{"hello":"world"}`)) + req, err := http.NewRequest(http.MethodPost, server.URL+tt.path, strings.NewReader(tt.requestBody)) if err != nil { t.Fatalf("NewRequest() error = %v", err) } + for key, value := range tt.requestHeaders { + req.Header.Set(key, value) + } resp, err := server.Client().Do(req) if err != nil { @@ -121,6 +132,245 @@ func TestServer_ProviderDetectionRouting(t *testing.T) { } } +func TestServer_OpenAICompatibleRoutesByModel(t *testing.T) { + t.Parallel() + + //nolint:govet // keep test table fields grouped for readability. + tests := []struct { + name string + model string + wantProvider string + }{ + {name: "gpt routes to openai", model: "gpt-4o-mini", wantProvider: "openai"}, + {name: "gemini routes to google", model: "gemini-2.5-flash", wantProvider: "google"}, + {name: "llama routes to ollama", model: "llama3.2", wantProvider: "ollama"}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + received := make(chan string, 1) + openAIServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + received <- "openai" + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(openAIServer.Close) + + anthropicServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + received <- "anthropic" + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(anthropicServer.Close) + + googleServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + received <- "google" + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(googleServer.Close) + + ollamaServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + received <- "ollama" + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(ollamaServer.Close) + + cfg := testConfig(openAIServer.URL, anthropicServer.URL) + cfg.Upstream.Google.BaseURL = googleServer.URL + cfg.Upstream.Ollama.BaseURL = ollamaServer.URL + proxyServer, err := New(cfg, Hooks{}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + server := newTestServer(t, proxyServer) + t.Cleanup(server.Close) + + reqBody := fmt.Sprintf(`{"model":%q,"messages":[{"role":"user","content":"hello"}]}`, tt.model) + req, err := http.NewRequest(http.MethodPost, server.URL+"/v1/chat/completions", strings.NewReader(reqBody)) + if err != nil { + t.Fatalf("NewRequest() error = %v", err) + } + + resp, err := server.Client().Do(req) + if err != nil { + t.Fatalf("Do() error = %v", err) + } + t.Cleanup(func() { _ = resp.Body.Close() }) + + if resp.StatusCode != http.StatusOK { + t.Fatalf("status code = %d, want %d", resp.StatusCode, http.StatusOK) + } + + select { + case got := <-received: + if got != tt.wantProvider { + t.Fatalf("routed provider = %q, want %q", got, tt.wantProvider) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for upstream request") + } + }) + } +} + +func TestServer_AnthropicOpenAICompatTransformsRequestAndResponse(t *testing.T) { + t.Parallel() + + requestBodies := make(chan string, 1) + anthropicServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + t.Fatalf("ReadAll() error = %v", err) + } + requestBodies <- string(body) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"msg_123","model":"claude-haiku-4-5","content":[{"type":"text","text":"anthropic ok"}],"usage":{"input_tokens":18,"output_tokens":10},"stop_reason":"end_turn"}`)) + })) + t.Cleanup(anthropicServer.Close) + + openAIServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Fatal("openai upstream should not be called for claude compatibility request") + })) + t.Cleanup(openAIServer.Close) + + cfg := testConfig(openAIServer.URL, anthropicServer.URL) + proxyServer, err := New(cfg, Hooks{}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + server := newTestServer(t, proxyServer) + t.Cleanup(server.Close) + + req, err := http.NewRequest(http.MethodPost, server.URL+"/v1/chat/completions", strings.NewReader(`{"model":"claude-haiku-4-5","messages":[{"role":"user","content":"hello"}]}`)) + if err != nil { + t.Fatalf("NewRequest() error = %v", err) + } + req.Header.Set("Authorization", "Bearer anthropic-key") + + resp, err := server.Client().Do(req) + if err != nil { + t.Fatalf("Do() error = %v", err) + } + t.Cleanup(func() { _ = resp.Body.Close() }) + + if resp.StatusCode != http.StatusOK { + t.Fatalf("status code = %d, want %d", resp.StatusCode, http.StatusOK) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("ReadAll() error = %v", err) + } + if !strings.Contains(string(body), `"object":"chat.completion"`) { + t.Fatalf("response body = %q, want OpenAI-compatible shape", string(body)) + } + if !strings.Contains(string(body), `"content":"anthropic ok"`) { + t.Fatalf("response body = %q, want translated content", string(body)) + } + + gotRequest := <-requestBodies + if !strings.Contains(gotRequest, `"model":"claude-haiku-4-5"`) || !strings.Contains(gotRequest, `"max_tokens":1024`) { + t.Fatalf("anthropic request = %s, want translated anthropic body", gotRequest) + } +} + +func TestServer_UnknownOpenAICompatibleModelReturnsProviderResolutionError(t *testing.T) { + t.Parallel() + + openAIServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Fatal("upstream should not be called when provider resolution fails") + })) + t.Cleanup(openAIServer.Close) + + anthropicServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Fatal("upstream should not be called when provider resolution fails") + })) + t.Cleanup(anthropicServer.Close) + + cfg := testConfig(openAIServer.URL, anthropicServer.URL) + proxyServer, err := New(cfg, Hooks{}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + server := newTestServer(t, proxyServer) + t.Cleanup(server.Close) + + req, err := http.NewRequest(http.MethodPost, server.URL+"/v1/chat/completions", strings.NewReader(`{"model":"mystery-model","messages":[{"role":"user","content":"hello"}]}`)) + if err != nil { + t.Fatalf("NewRequest() error = %v", err) + } + + resp, err := server.Client().Do(req) + if err != nil { + t.Fatalf("Do() error = %v", err) + } + t.Cleanup(func() { _ = resp.Body.Close() }) + + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status code = %d, want %d", resp.StatusCode, http.StatusBadRequest) + } + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("ReadAll() error = %v", err) + } + if !strings.Contains(string(body), "provider_resolution_failed") { + t.Fatalf("body = %q, want provider resolution error", string(body)) + } +} + +func TestServer_OllamaNativeAPIRoutesToOllama(t *testing.T) { + t.Parallel() + + received := make(chan string, 1) + ollamaServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + received <- r.URL.Path + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(ollamaServer.Close) + + openAIServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Fatal("openai upstream should not be called for native ollama requests") + })) + t.Cleanup(openAIServer.Close) + + anthropicServer := newTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Fatal("anthropic upstream should not be called for native ollama requests") + })) + t.Cleanup(anthropicServer.Close) + + cfg := testConfig(openAIServer.URL, anthropicServer.URL) + cfg.Upstream.Ollama.BaseURL = ollamaServer.URL + proxyServer, err := New(cfg, Hooks{}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + server := newTestServer(t, proxyServer) + t.Cleanup(server.Close) + + req, err := http.NewRequest(http.MethodPost, server.URL+"/api/chat", strings.NewReader(`{"model":"llama3.2","messages":[{"role":"user","content":"hello"}],"stream":false}`)) + if err != nil { + t.Fatalf("NewRequest() error = %v", err) + } + + resp, err := server.Client().Do(req) + if err != nil { + t.Fatalf("Do() error = %v", err) + } + t.Cleanup(func() { _ = resp.Body.Close() }) + + if resp.StatusCode != http.StatusOK { + t.Fatalf("status code = %d, want %d", resp.StatusCode, http.StatusOK) + } + if got := <-received; got != "/api/chat" { + t.Fatalf("routed path = %q, want %q", got, "/api/chat") + } +} + func TestServer_NonStreamingPassthroughAndHeaderStripping(t *testing.T) { t.Parallel() @@ -584,7 +834,7 @@ func TestServer_ProxyErrorReturnsBadGateway(t *testing.T) { server := newTestServer(t, proxyServer) t.Cleanup(server.Close) - req, err := http.NewRequest(http.MethodPost, server.URL+tt.path, bytes.NewBufferString(`{"stream":false}`)) + req, err := http.NewRequest(http.MethodPost, server.URL+tt.path, bytes.NewBufferString(`{"model":"gpt-4o-mini","stream":false}`)) if err != nil { t.Fatalf("NewRequest() error = %v", err) } diff --git a/internal/proxy/proxy_unit_test.go b/internal/proxy/proxy_unit_test.go index d776c98..06110ab 100644 --- a/internal/proxy/proxy_unit_test.go +++ b/internal/proxy/proxy_unit_test.go @@ -1,12 +1,14 @@ package proxy import ( + "context" "encoding/json" "errors" "io" "log/slog" "net/http" "net/http/httptest" + "net/url" "reflect" "strings" "sync" @@ -18,44 +20,44 @@ import ( "github.com/OberWatch/oberwatch/internal/storage" ) -func TestDetectProvider_TableDriven(t *testing.T) { +func TestDetectProviderForModel_TableDriven(t *testing.T) { t.Parallel() + //nolint:govet // keep test table fields grouped for readability. tests := []struct { - name string - path string - defaultProvider config.ProviderConfigName - want config.ProviderConfigName + name string + model string + want config.ProviderConfigName + ok bool }{ { - name: "chat completions routes to openai", - path: "/v1/chat/completions", - defaultProvider: config.ProviderAnthropic, - want: config.ProviderOpenAI, + name: "gpt routes to openai", + model: "gpt-4o-mini", + want: config.ProviderOpenAI, + ok: true, }, { - name: "completions routes to openai", - path: "/v1/completions", - defaultProvider: config.ProviderAnthropic, - want: config.ProviderOpenAI, + name: "gemini routes to google", + model: "gemini-2.5-flash", + want: config.ProviderGoogle, + ok: true, }, { - name: "messages routes to anthropic", - path: "/v1/messages", - defaultProvider: config.ProviderOpenAI, - want: config.ProviderAnthropic, + name: "claude routes to anthropic", + model: "claude-haiku-4-5", + want: config.ProviderAnthropic, + ok: true, }, { - name: "unknown path uses default", - path: "/v1/models", - defaultProvider: config.ProviderCustom, - want: config.ProviderCustom, + name: "llama routes to ollama", + model: "llama3.1:8b", + want: config.ProviderOllama, + ok: true, }, { - name: "trailing slash normalized", - path: "/v1/chat/completions/", - defaultProvider: config.ProviderAnthropic, - want: config.ProviderOpenAI, + name: "unknown model is rejected", + model: "mystery-provider-model", + ok: false, }, } @@ -64,12 +66,830 @@ func TestDetectProvider_TableDriven(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Parallel() - got := detectProvider(tt.path, tt.defaultProvider) + got, ok := detectProviderForModel(tt.model) + if ok != tt.ok { + t.Fatalf("detectProviderForModel(%q) ok = %v, want %v", tt.model, ok, tt.ok) + } if got != tt.want { - t.Fatalf("detectProvider(%q, %q) = %q, want %q", tt.path, tt.defaultProvider, got, tt.want) + t.Fatalf("detectProviderForModel(%q) = %q, want %q", tt.model, got, tt.want) + } + }) + } +} + +func TestResolveProvider_TableDriven(t *testing.T) { + t.Parallel() + + //nolint:govet // keep test table fields grouped for readability. + tests := []struct { + name string + path string + requestBody string + header string + want resolvedRoute + wantErrSub string + }{ + { + name: "messages uses anthropic", + path: "/v1/messages", + requestBody: `{"model":"claude-haiku-4-5"}`, + want: resolvedRoute{provider: config.ProviderAnthropic}, + }, + { + name: "openai-compatible gemini uses google", + path: "/v1/chat/completions", + requestBody: `{"model":"gemini-2.5-flash"}`, + want: resolvedRoute{provider: config.ProviderGoogle}, + }, + { + name: "claude chat uses anthropic compatibility", + path: "/v1/chat/completions", + requestBody: `{"model":"claude-haiku-4-5"}`, + want: resolvedRoute{provider: config.ProviderAnthropic, anthropicCompat: true}, + }, + { + name: "models requires provider header", + path: "/v1/models", + wantErrSub: providerOverrideHeader, + }, + { + name: "models accepts explicit provider header", + path: "/v1/models", + header: string(config.ProviderOpenAI), + want: resolvedRoute{provider: config.ProviderOpenAI}, + }, + { + name: "unknown model fails", + path: "/v1/chat/completions", + requestBody: `{"model":"mystery-provider-model"}`, + wantErrSub: "could not resolve provider", + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + req := httptest.NewRequest(http.MethodPost, tt.path, strings.NewReader(tt.requestBody)) + if tt.header != "" { + req.Header.Set(providerOverrideHeader, tt.header) + } + + got, err := resolveProvider(req, []byte(tt.requestBody), config.ProviderOpenAI) + if tt.wantErrSub != "" { + if err == nil || !strings.Contains(err.Error(), tt.wantErrSub) { + t.Fatalf("resolveProvider() error = %v, want substring %q", err, tt.wantErrSub) + } + return + } + if err != nil { + t.Fatalf("resolveProvider() error = %v", err) + } + if !reflect.DeepEqual(got, tt.want) { + t.Fatalf("resolveProvider() = %#v, want %#v", got, tt.want) + } + }) + } +} + +func TestOpenAIChatToAnthropic_TableDriven(t *testing.T) { + t.Parallel() + + //nolint:govet // keep test table fields grouped for readability. + tests := []struct { + name string + requestBody string + wantContains []string + wantModel string + wantStream bool + wantErrSub string + }{ + { + name: "translates simple chat request", + requestBody: `{"model":"claude-haiku-4-5","messages":[{"role":"system","content":"be concise"},{"role":"user","content":"hello"}]}`, + wantContains: []string{ + `"model":"claude-haiku-4-5"`, + `"system":"be concise"`, + `"role":"user"`, + `"content":"hello"`, + `"max_tokens":1024`, + }, + wantModel: "claude-haiku-4-5", + wantStream: false, + }, + { + name: "missing model fails", + requestBody: `{"messages":[{"role":"user","content":"hello"}]}`, + wantErrSub: "request model is required", + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + gotBody, gotModel, gotStream, err := openAIChatToAnthropic([]byte(tt.requestBody)) + if tt.wantErrSub != "" { + if err == nil || !strings.Contains(err.Error(), tt.wantErrSub) { + t.Fatalf("openAIChatToAnthropic() error = %v, want substring %q", err, tt.wantErrSub) + } + return + } + if err != nil { + t.Fatalf("openAIChatToAnthropic() error = %v", err) + } + if gotModel != tt.wantModel { + t.Fatalf("model = %q, want %q", gotModel, tt.wantModel) + } + if gotStream != tt.wantStream { + t.Fatalf("stream = %v, want %v", gotStream, tt.wantStream) + } + for _, want := range tt.wantContains { + if !strings.Contains(string(gotBody), want) { + t.Fatalf("body = %q, want substring %q", string(gotBody), want) + } + } + }) + } +} + +func TestAnthropicToOpenAIChatCompletion_TableDriven(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + responseBody string + wantContains []string + }{ + { + name: "translates anthropic response", + responseBody: `{"id":"msg_123","model":"claude-haiku-4-5","content":[{"type":"text","text":"hello back"}],"usage":{"input_tokens":18,"output_tokens":10},"stop_reason":"end_turn"}`, + wantContains: []string{`"object":"chat.completion"`, `"model":"claude-haiku-4-5"`, `"content":"hello back"`, `"prompt_tokens":18`, `"completion_tokens":10`}, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got, err := anthropicToOpenAIChatCompletion([]byte(tt.responseBody), "claude-haiku-4-5") + if err != nil { + t.Fatalf("anthropicToOpenAIChatCompletion() error = %v", err) + } + for _, want := range tt.wantContains { + if !strings.Contains(string(got), want) { + t.Fatalf("body = %q, want substring %q", string(got), want) + } + } + }) + } +} + +func TestProviderRoutingHelpers_TableDriven(t *testing.T) { + t.Parallel() + + //nolint:govet // keep test table fields grouped for readability. + tests := []struct { + name string + header string + want config.ProviderConfigName + ok bool + rawPath string + wantOpen bool + }{ + {name: "provider override parses google", header: "google", want: config.ProviderGoogle, ok: true}, + {name: "provider override rejects unknown", header: "grok", ok: false}, + {name: "openai-compatible chat path is recognized", rawPath: "/v1/chat/completions", wantOpen: true}, + {name: "native ollama path is not openai-compatible", rawPath: "/api/chat", wantOpen: false}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + if tt.header != "" { + req := httptest.NewRequest(http.MethodGet, "/v1/models", nil) + req.Header.Set(providerOverrideHeader, tt.header) + got, ok := parseProviderOverride(req) + if ok != tt.ok { + t.Fatalf("parseProviderOverride() ok = %v, want %v", ok, tt.ok) + } + if got != tt.want { + t.Fatalf("parseProviderOverride() = %q, want %q", got, tt.want) + } + } + if tt.rawPath != "" { + if got := isOpenAICompatiblePath(tt.rawPath); got != tt.wantOpen { + t.Fatalf("isOpenAICompatiblePath(%q) = %v, want %v", tt.rawPath, got, tt.wantOpen) + } + } + }) + } +} + +func TestWriteProviderResolutionError_TableDriven(t *testing.T) { + t.Parallel() + + recorder := httptest.NewRecorder() + writeProviderResolutionError(recorder, "no provider", http.StatusBadRequest) + + if recorder.Code != http.StatusBadRequest { + t.Fatalf("status code = %d, want %d", recorder.Code, http.StatusBadRequest) + } + if !strings.Contains(recorder.Body.String(), "provider_resolution_failed") { + t.Fatalf("body = %q, want provider resolution error code", recorder.Body.String()) + } + if !strings.Contains(recorder.Body.String(), "no provider") { + t.Fatalf("body = %q, want original message", recorder.Body.String()) + } +} + +func TestCompatibilityHelpers_TableDriven(t *testing.T) { + t.Parallel() + + t.Run("extracts message content text", func(t *testing.T) { + t.Parallel() + + got := extractMessageContentText([]any{ + map[string]any{"type": "text", "text": "hello"}, + map[string]any{"type": "image", "text": "ignore"}, + map[string]any{"type": "text", "text": "world"}, + }) + if got != "hello\nworld" { + t.Fatalf("extractMessageContentText() = %q, want %q", got, "hello\nworld") + } + }) + + t.Run("extracts bearer token", func(t *testing.T) { + t.Parallel() + + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.Header.Set("Authorization", "Bearer test-token") + if got := extractBearerToken(req); got != "test-token" { + t.Fatalf("extractBearerToken() = %q, want %q", got, "test-token") + } + }) + + t.Run("copies compatibility headers safely", func(t *testing.T) { + t.Parallel() + + src := http.Header{ + "Authorization": []string{"Bearer token"}, + "X-API-Key": []string{"secret"}, + "X-Oberwatch-Agent": []string{"agent"}, + "X-Custom": []string{"ok"}, + } + dst := make(http.Header) + copyCompatibilityHeaders(dst, src) + if dst.Get("Authorization") != "" || dst.Get("X-API-Key") != "" || dst.Get("X-Oberwatch-Agent") != "" { + t.Fatalf("copyCompatibilityHeaders() leaked protected headers: %#v", dst) + } + if dst.Get("X-Custom") != "ok" { + t.Fatalf("copyCompatibilityHeaders() missing custom header: %#v", dst) + } + }) + + t.Run("maps anthropic stop reasons", func(t *testing.T) { + t.Parallel() + + if got := mapAnthropicStopReason("max_tokens"); got != "length" { + t.Fatalf("mapAnthropicStopReason(max_tokens) = %q, want %q", got, "length") + } + if got := mapAnthropicStopReason("end_turn"); got != "stop" { + t.Fatalf("mapAnthropicStopReason(end_turn) = %q, want %q", got, "stop") + } + }) +} + +func TestProviderRoutingMiddleware_TableDriven(t *testing.T) { + t.Parallel() + + mustParseURL := func(t *testing.T, raw string) *url.URL { + t.Helper() + parsed, err := url.Parse(raw) + if err != nil { + t.Fatalf("url.Parse(%q) error = %v", raw, err) + } + return parsed + } + + //nolint:govet // keep test table fields grouped for readability. + tests := []struct { + name string + path string + body string + header string + readErr error + targets map[config.ProviderConfigName]*url.URL + wantStatus int + wantProvider config.ProviderConfigName + wantCalledNext bool + wantBodySub string + }{ + { + name: "non proxy path passes through untouched", + path: "/", + targets: map[config.ProviderConfigName]*url.URL{config.ProviderOpenAI: mustParseURL(t, "https://api.openai.com")}, + wantStatus: http.StatusNoContent, + wantCalledNext: true, + }, + { + name: "gemini request resolves to google", + path: "/v1/chat/completions", + body: `{"model":"gemini-2.5-flash","messages":[{"role":"user","content":"hi"}]}`, + targets: map[config.ProviderConfigName]*url.URL{config.ProviderGoogle: mustParseURL(t, "https://generativelanguage.googleapis.com")}, + wantStatus: http.StatusNoContent, + wantProvider: config.ProviderGoogle, + wantCalledNext: true, + }, + { + name: "models without header fail", + path: "/v1/models", + targets: map[config.ProviderConfigName]*url.URL{config.ProviderOpenAI: mustParseURL(t, "https://api.openai.com")}, + wantStatus: http.StatusBadRequest, + wantBodySub: providerOverrideHeader, + }, + { + name: "resolved provider must be configured", + path: "/v1/chat/completions", + body: `{"model":"gemini-2.5-flash","messages":[{"role":"user","content":"hi"}]}`, + targets: map[config.ProviderConfigName]*url.URL{config.ProviderOpenAI: mustParseURL(t, "https://api.openai.com")}, + wantStatus: http.StatusBadRequest, + wantBodySub: `provider \"google\" is not configured`, + }, + { + name: "body read error returns config error", + path: "/v1/chat/completions", + readErr: errors.New("boom"), + targets: map[config.ProviderConfigName]*url.URL{config.ProviderOpenAI: mustParseURL(t, "https://api.openai.com")}, + wantStatus: http.StatusInternalServerError, + wantBodySub: "config_error", + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + var called bool + var gotRoute resolvedRoute + handler := providerRoutingMiddleware(config.ProviderOpenAI, tt.targets, nil)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + called = true + route, ok := resolvedRouteFromContext(r.Context()) + if ok { + gotRoute = route + } + w.WriteHeader(http.StatusNoContent) + })) + + req := httptest.NewRequest(http.MethodPost, tt.path, strings.NewReader(tt.body)) + if tt.header != "" { + req.Header.Set(providerOverrideHeader, tt.header) + } + if tt.readErr != nil { + req.Body = errorReadCloser{err: tt.readErr} + req.ContentLength = 10 + } + recorder := httptest.NewRecorder() + + handler.ServeHTTP(recorder, req) + if recorder.Code != tt.wantStatus { + t.Fatalf("status code = %d, want %d", recorder.Code, tt.wantStatus) + } + if called != tt.wantCalledNext { + t.Fatalf("next called = %v, want %v", called, tt.wantCalledNext) } + if tt.wantProvider != "" && gotRoute.provider != tt.wantProvider { + t.Fatalf("resolved provider = %q, want %q", gotRoute.provider, tt.wantProvider) + } + if tt.wantBodySub != "" && !strings.Contains(recorder.Body.String(), tt.wantBodySub) { + t.Fatalf("body = %q, want substring %q", recorder.Body.String(), tt.wantBodySub) + } + }) + } +} + +func TestServeAnthropicOpenAICompat_TableDriven(t *testing.T) { + withHTTPTransport := func(t *testing.T, transport http.RoundTripper) { + t.Helper() + original := http.DefaultClient.Transport + http.DefaultClient.Transport = transport + t.Cleanup(func() { + http.DefaultClient.Transport = original }) } + + t.Run("streaming requests are rejected", func(t *testing.T) { + target, err := url.Parse("https://api.anthropic.com") + if err != nil { + t.Fatalf("url.Parse() error = %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{"model":"claude-haiku-4-5","stream":true,"messages":[{"role":"user","content":"hello"}]}`)) + recorder := httptest.NewRecorder() + + serveAnthropicOpenAICompat(recorder, req, resolvedRoute{provider: config.ProviderAnthropic, anthropicCompat: true}, map[config.ProviderConfigName]*url.URL{config.ProviderAnthropic: target}, Hooks{}) + + if recorder.Code != http.StatusBadRequest { + t.Fatalf("status code = %d, want %d", recorder.Code, http.StatusBadRequest) + } + if !strings.Contains(recorder.Body.String(), "non-streaming") { + t.Fatalf("body = %q, want non-streaming error", recorder.Body.String()) + } + }) + + t.Run("upstream error is passed through", func(t *testing.T) { + target, err := url.Parse("https://api.anthropic.com") + if err != nil { + t.Fatalf("url.Parse() error = %v", err) + } + withHTTPTransport(t, roundTripFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusUnauthorized, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"error":"bad api key"}`)), + }, nil + })) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{"model":"claude-haiku-4-5","messages":[{"role":"user","content":"hello"}]}`)) + req.Header.Set("Authorization", "Bearer bad-key") + recorder := httptest.NewRecorder() + + serveAnthropicOpenAICompat(recorder, req, resolvedRoute{provider: config.ProviderAnthropic, anthropicCompat: true}, map[config.ProviderConfigName]*url.URL{config.ProviderAnthropic: target}, Hooks{}) + + if recorder.Code != http.StatusUnauthorized { + t.Fatalf("status code = %d, want %d", recorder.Code, http.StatusUnauthorized) + } + if !strings.Contains(recorder.Body.String(), "bad api key") { + t.Fatalf("body = %q, want upstream error", recorder.Body.String()) + } + }) + + t.Run("successful translation records spend and sink data", func(t *testing.T) { + target, err := url.Parse("https://api.anthropic.com") + if err != nil { + t.Fatalf("url.Parse() error = %v", err) + } + withHTTPTransport(t, roundTripFunc(func(r *http.Request) (*http.Response, error) { + if got := r.Header.Get("x-api-key"); got != "anthropic-key" { + t.Fatalf("x-api-key = %q, want %q", got, "anthropic-key") + } + if got := r.Header.Get("anthropic-version"); got != "2023-06-01" { + t.Fatalf("anthropic-version = %q, want %q", got, "2023-06-01") + } + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"id":"msg_123","model":"claude-haiku-4-5","content":[{"type":"text","text":"hello back"}],"usage":{"input_tokens":18,"output_tokens":10},"stop_reason":"end_turn"}`)), + }, nil + })) + + cfg := config.DefaultConfig() + manager := budget.NewManager(cfg.Gate, nil) + table := pricing.NewPricingTableFromConfig(cfg.Pricing, nil) + var records []storage.CostRecord + sink := sinkFunc(func(record storage.CostRecord) { + records = append(records, record) + }) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{"model":"claude-haiku-4-5","messages":[{"role":"user","content":"hello"}]}`)) + req.Header.Set("Authorization", "Bearer anthropic-key") + meta := budgetRequestMeta{ + agent: "agent-a", + model: "claude-haiku-4-5", + provider: string(config.ProviderAnthropic), + } + req = req.WithContext(context.WithValue(req.Context(), budgetContextKey{}, meta)) + recorder := httptest.NewRecorder() + + serveAnthropicOpenAICompat(recorder, req, resolvedRoute{provider: config.ProviderAnthropic, anthropicCompat: true}, map[config.ProviderConfigName]*url.URL{config.ProviderAnthropic: target}, Hooks{ + Budget: manager, + Pricing: table, + CostSink: sink, + }) + + if recorder.Code != http.StatusOK { + t.Fatalf("status code = %d, want %d", recorder.Code, http.StatusOK) + } + if !strings.Contains(recorder.Body.String(), `"object":"chat.completion"`) { + t.Fatalf("body = %q, want OpenAI-compatible response", recorder.Body.String()) + } + if spent := manager.Snapshot("agent-a").SpentUSD; spent <= 0 { + t.Fatalf("spent = %v, want > 0", spent) + } + if len(records) != 1 { + t.Fatalf("sink records = %d, want 1", len(records)) + } + }) +} + +func TestProviderRoutingAdditionalBranches_TableDriven(t *testing.T) { + t.Parallel() + + t.Run("resolve provider handles api path override and missing model branches", func(t *testing.T) { + t.Parallel() + + //nolint:govet // keep test table fields grouped for readability. + tests := []struct { + name string + path string + body string + header string + want resolvedRoute + wantErrSub string + }{ + {name: "native ollama api path", path: "/api/chat", want: resolvedRoute{provider: config.ProviderOllama}}, + {name: "provider override on openai-compatible path", path: "/v1/chat/completions", body: `{"model":"gemini-2.5-flash"}`, header: string(config.ProviderGoogle), want: resolvedRoute{provider: config.ProviderGoogle}}, + {name: "anthropic override on chat path enables compatibility", path: "/v1/chat/completions", body: `{"model":"claude-haiku-4-5"}`, header: string(config.ProviderAnthropic), want: resolvedRoute{provider: config.ProviderAnthropic, anthropicCompat: true}}, + {name: "missing model is rejected", path: "/v1/chat/completions", body: `{}`, wantErrSub: "request model is required"}, + {name: "claude embeddings path is rejected", path: "/v1/embeddings", body: `{"model":"claude-haiku-4-5"}`, wantErrSub: "only supports /v1/chat/completions"}, + {name: "empty path normalizes to slash and uses default", path: "", want: resolvedRoute{provider: config.ProviderOpenAI}}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + requestPath := tt.path + if requestPath == "" { + requestPath = "/" + } + req := httptest.NewRequest(http.MethodPost, requestPath, strings.NewReader(tt.body)) + if tt.path == "" { + req.URL.Path = "" + } + if tt.header != "" { + req.Header.Set(providerOverrideHeader, tt.header) + } + got, err := resolveProvider(req, []byte(tt.body), config.ProviderOpenAI) + if tt.wantErrSub != "" { + if err == nil || !strings.Contains(err.Error(), tt.wantErrSub) { + t.Fatalf("resolveProvider() error = %v, want substring %q", err, tt.wantErrSub) + } + return + } + if err != nil { + t.Fatalf("resolveProvider() error = %v", err) + } + if !reflect.DeepEqual(got, tt.want) { + t.Fatalf("resolveProvider() = %#v, want %#v", got, tt.want) + } + }) + } + }) + + t.Run("resolve provider rejects anthropic override on unsupported path", func(t *testing.T) { + t.Parallel() + + req := httptest.NewRequest(http.MethodPost, "/v1/embeddings", strings.NewReader(`{"model":"claude-haiku-4-5"}`)) + req.Header.Set(providerOverrideHeader, string(config.ProviderAnthropic)) + + _, err := resolveProvider(req, []byte(`{"model":"claude-haiku-4-5"}`), config.ProviderOpenAI) + if err == nil || !strings.Contains(err.Error(), "only supports /v1/chat/completions") { + t.Fatalf("resolveProvider() error = %v, want anthropic path restriction", err) + } + }) + + t.Run("resolve provider falls back to default on unknown non-proxy path", func(t *testing.T) { + t.Parallel() + + req := httptest.NewRequest(http.MethodGet, "/internal/healthz", nil) + got, err := resolveProvider(req, nil, config.ProviderCustom) + if err != nil { + t.Fatalf("resolveProvider() error = %v", err) + } + if got.provider != config.ProviderCustom { + t.Fatalf("resolved provider = %q, want %q", got.provider, config.ProviderCustom) + } + }) + + t.Run("resolve provider fails when unknown path has no default", func(t *testing.T) { + t.Parallel() + + req := httptest.NewRequest(http.MethodGet, "/internal/healthz", nil) + _, err := resolveProvider(req, nil, "") + if err == nil || !strings.Contains(err.Error(), "could not resolve provider") { + t.Fatalf("resolveProvider() error = %v, want missing default failure", err) + } + }) + + t.Run("openai to anthropic requires non-system message", func(t *testing.T) { + t.Parallel() + + _, _, _, err := openAIChatToAnthropic([]byte(`{"model":"claude-haiku-4-5","messages":[{"role":"system","content":"only system"}]}`)) + if err == nil || !strings.Contains(err.Error(), "non-system message") { + t.Fatalf("openAIChatToAnthropic() error = %v, want non-system message failure", err) + } + }) + + t.Run("openai to anthropic rejects malformed json", func(t *testing.T) { + t.Parallel() + + _, _, _, err := openAIChatToAnthropic([]byte(`{`)) + if err == nil || !strings.Contains(err.Error(), "decode OpenAI chat completion request") { + t.Fatalf("openAIChatToAnthropic() error = %v, want decode failure", err) + } + }) + + t.Run("openai to anthropic preserves temperature", func(t *testing.T) { + t.Parallel() + + body, _, _, err := openAIChatToAnthropic([]byte(`{"model":"claude-haiku-4-5","temperature":0.25,"messages":[{"role":"user","content":"hello"}]}`)) + if err != nil { + t.Fatalf("openAIChatToAnthropic() error = %v", err) + } + if !strings.Contains(string(body), `"temperature":0.25`) { + t.Fatalf("body = %q, want temperature preserved", string(body)) + } + }) + + t.Run("anthropic response decode failure surfaces", func(t *testing.T) { + t.Parallel() + + _, err := anthropicToOpenAIChatCompletion([]byte(`{`), "claude-haiku-4-5") + if err == nil || !strings.Contains(err.Error(), "decode anthropic response") { + t.Fatalf("anthropicToOpenAIChatCompletion() error = %v, want decode failure", err) + } + }) + + t.Run("copy compatibility response headers skips content length", func(t *testing.T) { + t.Parallel() + + src := http.Header{ + "Content-Length": []string{"99"}, + "X-Upstream": []string{"ok"}, + } + dst := make(http.Header) + copyCompatibilityResponseHeaders(dst, src) + if dst.Get("Content-Length") != "" { + t.Fatalf("copyCompatibilityResponseHeaders() copied Content-Length: %#v", dst) + } + if dst.Get("X-Upstream") != "ok" { + t.Fatalf("copyCompatibilityResponseHeaders() missing X-Upstream header: %#v", dst) + } + }) + + t.Run("serve anthropic compat rejects missing target", func(t *testing.T) { + t.Parallel() + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{"model":"claude-haiku-4-5","messages":[{"role":"user","content":"hello"}]}`)) + recorder := httptest.NewRecorder() + + serveAnthropicOpenAICompat(recorder, req, resolvedRoute{provider: config.ProviderAnthropic, anthropicCompat: true}, map[config.ProviderConfigName]*url.URL{}, Hooks{}) + if recorder.Code != http.StatusBadRequest { + t.Fatalf("status code = %d, want %d", recorder.Code, http.StatusBadRequest) + } + if !strings.Contains(recorder.Body.String(), "not configured") { + t.Fatalf("body = %q, want missing target error", recorder.Body.String()) + } + }) + + t.Run("middleware preserves body via GetBody and resolves route with logger", func(t *testing.T) { + t.Parallel() + + target, err := url.Parse("https://generativelanguage.googleapis.com") + if err != nil { + t.Fatalf("url.Parse() error = %v", err) + } + + var gotBody string + var gotRoute resolvedRoute + handler := providerRoutingMiddleware(config.ProviderOpenAI, map[config.ProviderConfigName]*url.URL{ + config.ProviderGoogle: target, + }, slog.New(slog.NewTextHandler(io.Discard, nil)))(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + route, ok := resolvedRouteFromContext(r.Context()) + if !ok { + t.Fatal("resolved route missing from context") + } + gotRoute = route + reader, err := r.GetBody() + if err != nil { + t.Fatalf("GetBody() error = %v", err) + } + defer func() { _ = reader.Close() }() + body, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("ReadAll() error = %v", err) + } + gotBody = string(body) + w.WriteHeader(http.StatusNoContent) + })) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{"model":"gemini-2.5-flash","messages":[{"role":"user","content":"hello"}]}`)) + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, req) + + if recorder.Code != http.StatusNoContent { + t.Fatalf("status code = %d, want %d", recorder.Code, http.StatusNoContent) + } + if gotRoute.provider != config.ProviderGoogle { + t.Fatalf("resolved provider = %q, want %q", gotRoute.provider, config.ProviderGoogle) + } + if !strings.Contains(gotBody, `"model":"gemini-2.5-flash"`) { + t.Fatalf("GetBody() payload = %q, want original request body", gotBody) + } + }) + + t.Run("compat helpers cover remaining branches", func(t *testing.T) { + t.Parallel() + + if got := extractMessageContentText([]any{"skip", map[string]any{"type": "text", "text": "ok"}}); got != "ok" { + t.Fatalf("extractMessageContentText() = %q, want %q", got, "ok") + } + if got := extractMessageContentText(123); got != "" { + t.Fatalf("extractMessageContentText(non-text) = %q, want empty", got) + } + + req := httptest.NewRequest(http.MethodGet, "/", nil) + if got := extractBearerToken(req); got != "" { + t.Fatalf("extractBearerToken() = %q, want empty", got) + } + req.Header.Set("Authorization", "Basic nope") + if got := extractBearerToken(req); got != "" { + t.Fatalf("extractBearerToken() with basic auth = %q, want empty", got) + } + + dst := make(http.Header) + copyCompatibilityHeaders(dst, http.Header{ + "Host": []string{"example.com"}, + "Content-Length": []string{"123"}, + "X-API-Key": []string{"secret"}, + "X-Custom": []string{"ok"}, + }) + if dst.Get("Host") != "" || dst.Get("Content-Length") != "" || dst.Get("X-API-Key") != "" { + t.Fatalf("copyCompatibilityHeaders() copied restricted header: %#v", dst) + } + if dst.Get("X-Custom") != "ok" { + t.Fatalf("copyCompatibilityHeaders() missing custom header: %#v", dst) + } + + if got := mapAnthropicStopReason("tool_use"); got != "tool_calls" { + t.Fatalf("mapAnthropicStopReason(tool_use) = %q, want %q", got, "tool_calls") + } + }) + + t.Run("anthropic conversion covers non-text response blocks", func(t *testing.T) { + t.Parallel() + + got, err := anthropicToOpenAIChatCompletion([]byte(`{"id":"msg_123","model":"claude-haiku-4-5","content":[{"type":"image","text":"ignore"},{"type":"text","text":"hello"}],"usage":{"input_tokens":18,"output_tokens":10},"stop_reason":"tool_use"}`), "claude-haiku-4-5") + if err != nil { + t.Fatalf("anthropicToOpenAIChatCompletion() error = %v", err) + } + if !strings.Contains(string(got), `"finish_reason":"tool_calls"`) { + t.Fatalf("body = %q, want tool_calls finish reason", string(got)) + } + if !strings.Contains(string(got), `"content":"hello"`) { + t.Fatalf("body = %q, want text block only", string(got)) + } + }) + + t.Run("serve anthropic compat returns config error for unreadable body", func(t *testing.T) { + t.Parallel() + + target, err := url.Parse("https://api.anthropic.com") + if err != nil { + t.Fatalf("url.Parse() error = %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil) + req.Body = errorReadCloser{err: errors.New("boom")} + req.ContentLength = 10 + recorder := httptest.NewRecorder() + + serveAnthropicOpenAICompat(recorder, req, resolvedRoute{provider: config.ProviderAnthropic, anthropicCompat: true}, map[config.ProviderConfigName]*url.URL{config.ProviderAnthropic: target}, Hooks{}) + if recorder.Code != http.StatusInternalServerError { + t.Fatalf("status code = %d, want %d", recorder.Code, http.StatusInternalServerError) + } + if !strings.Contains(recorder.Body.String(), "config_error") { + t.Fatalf("body = %q, want config_error payload", recorder.Body.String()) + } + }) + + t.Run("serve anthropic compat rejects invalid openai payload", func(t *testing.T) { + t.Parallel() + + target, err := url.Parse("https://api.anthropic.com") + if err != nil { + t.Fatalf("url.Parse() error = %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{`)) + recorder := httptest.NewRecorder() + + serveAnthropicOpenAICompat(recorder, req, resolvedRoute{provider: config.ProviderAnthropic, anthropicCompat: true}, map[config.ProviderConfigName]*url.URL{config.ProviderAnthropic: target}, Hooks{}) + if recorder.Code != http.StatusBadRequest { + t.Fatalf("status code = %d, want %d", recorder.Code, http.StatusBadRequest) + } + if !strings.Contains(recorder.Body.String(), "provider_resolution_failed") { + t.Fatalf("body = %q, want provider resolution payload", recorder.Body.String()) + } + }) } func TestStripOberwatchHeaders_TableDriven(t *testing.T) { @@ -659,10 +1479,16 @@ type errorReadCloser struct { type sinkFunc func(storage.CostRecord) +type roundTripFunc func(*http.Request) (*http.Response, error) + func (f sinkFunc) Enqueue(record storage.CostRecord) { f(record) } +func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} + func (e errorReadCloser) Read([]byte) (int, error) { return 0, e.err } diff --git a/oberwatch.example.toml b/oberwatch.example.toml index c98186b..46b3041 100644 --- a/oberwatch.example.toml +++ b/oberwatch.example.toml @@ -51,7 +51,7 @@ tls_key = "" # ----------------------------------------------------------------------------- [upstream] # Default provider used when path-based provider detection is ambiguous. -# Valid values: openai, anthropic, ollama, custom +# Valid values: openai, anthropic, google, ollama, custom # Env: OBERWATCH_UPSTREAM__DEFAULT_PROVIDER default_provider = "openai" @@ -69,6 +69,11 @@ base_url = "https://api.openai.com" # Env: OBERWATCH_UPSTREAM__ANTHROPIC__BASE_URL base_url = "https://api.anthropic.com" +[upstream.google] +# Base URL for Google Gemini OpenAI-compatible API. +# Env: OBERWATCH_UPSTREAM__GOOGLE__BASE_URL +base_url = "https://generativelanguage.googleapis.com/v1beta/openai" + [upstream.ollama] # Base URL for local or remote Ollama server. # Env: OBERWATCH_UPSTREAM__OLLAMA__BASE_URL