From a1f9abe6ce90baf16543c54984293638fbc03aa0 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 22 Aug 2026 21:25:48 +0000 Subject: [PATCH 1/2] docs: subscription- and quota-aware routing engine spec + scaffold Add docs/ROUTING_ENGINE_SPEC.md covering provider/account ledger, telemetry, failover, Honcho sync, and Life OS metrics. Add uc_routing/ package with placeholder modules for ledger, telemetry, routing, providers, failover, honcho, life_os, config, and tests. Link the new spec from README.md. No production behavior implemented. Co-Authored-By: Rob --- README.md | 1 + docs/ROUTING_ENGINE_SPEC.md | 821 +++++++++++++++++++++++++++++ uc_routing/README.md | 35 ++ uc_routing/__init__.py | 27 + uc_routing/config/__init__.py | 12 + uc_routing/config/loader.py | 58 ++ uc_routing/config/schema.py | 37 ++ uc_routing/failover/__init__.py | 7 + uc_routing/failover/circuit.py | 58 ++ uc_routing/failover/cooldown.py | 58 ++ uc_routing/failover/health.py | 57 ++ uc_routing/honcho/__init__.py | 6 + uc_routing/honcho/contract.py | 29 + uc_routing/honcho/sync.py | 43 ++ uc_routing/ledger/__init__.py | 5 + uc_routing/ledger/balance.py | 47 ++ uc_routing/ledger/models.py | 105 ++++ uc_routing/ledger/window.py | 91 ++++ uc_routing/life_os/__init__.py | 6 + uc_routing/life_os/api.py | 38 ++ uc_routing/life_os/metrics.py | 42 ++ uc_routing/providers/__init__.py | 7 + uc_routing/providers/adapter.py | 30 ++ uc_routing/providers/registry.py | 58 ++ uc_routing/providers/types.py | 47 ++ uc_routing/routing/__init__.py | 8 + uc_routing/routing/decision.py | 21 + uc_routing/routing/engine.py | 102 ++++ uc_routing/routing/policy.py | 104 ++++ uc_routing/routing/task_tiers.py | 59 +++ uc_routing/telemetry/__init__.py | 7 + uc_routing/telemetry/collector.py | 52 ++ uc_routing/telemetry/cost.py | 43 ++ uc_routing/telemetry/schema.py | 62 +++ uc_routing/tests/__init__.py | 1 + uc_routing/tests/test_failover.py | 36 ++ uc_routing/tests/test_ledger.py | 43 ++ uc_routing/tests/test_routing.py | 84 +++ uc_routing/tests/test_telemetry.py | 40 ++ 39 files changed, 2387 insertions(+) create mode 100644 docs/ROUTING_ENGINE_SPEC.md create mode 100644 uc_routing/README.md create mode 100644 uc_routing/__init__.py create mode 100644 uc_routing/config/__init__.py create mode 100644 uc_routing/config/loader.py create mode 100644 uc_routing/config/schema.py create mode 100644 uc_routing/failover/__init__.py create mode 100644 uc_routing/failover/circuit.py create mode 100644 uc_routing/failover/cooldown.py create mode 100644 uc_routing/failover/health.py create mode 100644 uc_routing/honcho/__init__.py create mode 100644 uc_routing/honcho/contract.py create mode 100644 uc_routing/honcho/sync.py create mode 100644 uc_routing/ledger/__init__.py create mode 100644 uc_routing/ledger/balance.py create mode 100644 uc_routing/ledger/models.py create mode 100644 uc_routing/ledger/window.py create mode 100644 uc_routing/life_os/__init__.py create mode 100644 uc_routing/life_os/api.py create mode 100644 uc_routing/life_os/metrics.py create mode 100644 uc_routing/providers/__init__.py create mode 100644 uc_routing/providers/adapter.py create mode 100644 uc_routing/providers/registry.py create mode 100644 uc_routing/providers/types.py create mode 100644 uc_routing/routing/__init__.py create mode 100644 uc_routing/routing/decision.py create mode 100644 uc_routing/routing/engine.py create mode 100644 uc_routing/routing/policy.py create mode 100644 uc_routing/routing/task_tiers.py create mode 100644 uc_routing/telemetry/__init__.py create mode 100644 uc_routing/telemetry/collector.py create mode 100644 uc_routing/telemetry/cost.py create mode 100644 uc_routing/telemetry/schema.py create mode 100644 uc_routing/tests/__init__.py create mode 100644 uc_routing/tests/test_failover.py create mode 100644 uc_routing/tests/test_ledger.py create mode 100644 uc_routing/tests/test_routing.py create mode 100644 uc_routing/tests/test_telemetry.py diff --git a/README.md b/README.md index 6615968..61b406c 100644 --- a/README.md +++ b/README.md @@ -331,6 +331,7 @@ test → troubleshoot) written for an AI to follow. | [docs/DIRECTIVES.md](docs/DIRECTIVES.md) | Routing directives — pin a request to a model from the prompt (per-role multi-agent workflows) | | [docs/ADD_A_MODEL.md](docs/ADD_A_MODEL.md) | Add any backend to the `/model` menu | | [docs/TROUBLESHOOTING.md](docs/TROUBLESHOOTING.md) | Symptom → cause → fix | +| [docs/ROUTING_ENGINE_SPEC.md](docs/ROUTING_ENGINE_SPEC.md) | Subscription- and quota-aware routing engine (design + scaffold) | ## License diff --git a/docs/ROUTING_ENGINE_SPEC.md b/docs/ROUTING_ENGINE_SPEC.md new file mode 100644 index 0000000..e5db0c8 --- /dev/null +++ b/docs/ROUTING_ENGINE_SPEC.md @@ -0,0 +1,821 @@ +# Subscription- and Quota-Aware AI Model Routing Engine + +**Status:** Design / scaffold +**Scope:** UltraCode-Shim +**Last updated:** 2026-08-22 + +This document specifies the next-generation routing layer for UltraCode-Shim. It supersedes the experimental `Auto Router` (`docs/AUTO_ROUTER.md`) with a ledger-aware engine that selects models based on real account entitlements, provider health, observed latency/quality telemetry, and task tier — while keeping the proxy itself dependency-light and stdlib-first. + +--- + +## 1. Overview + +### 1.1 Goals + +- **Route every request to the cheapest capable backend** while respecting real-time quotas, rate limits, and user preferences. +- **Saturate zero-marginal-cost capacity first** (local RTX 5090 compute) before spending subscription or prepaid quota. +- **Prevent hangs on dead or empty routes** with deterministic timeouts, health checks, and circuit breakers. +- **Account for heterogeneous account types:** fixed-window subscriptions (Claude, ChatGPT, Cursor, Grok), prepaid credit pools (Devin, OpenRouter, API-key gateways), and local compute. +- **Emit fine-grained telemetry** (TTFT, end-to-end latency, token spend, remaining allowances, 429/cooldown events) to drive routing decisions and user dashboards. +- **Synchronize routing state to Honcho** so OnlyTerp's cross-session memory and cost state stays consistent. +- **Surface a Life OS metrics API** so `terpOS` (or any consumer) can display spend, quota, provider health, and task-tier mix. + +### 1.2 Non-Goals + +- This task is **spec and scaffold only**; no full provider integrations, no live inference, and no production enablement. +- We do **not** aim to replace the existing `proxy.py` in one PR; the engine is a parallel package (`uc_routing/`) that can be wired into `proxy.py` in a later migration. +- We do **not** build a new model-quality classifier from scratch in this scaffold; the existing `Auto Router` classifier path remains the default until the engine matures. +- We do **not** store provider secrets in the ledger; secrets continue to live in `config.json` / env vars / keychain and are resolved only inside provider adapters. + +### 1.3 Terminology + +| Term | Meaning | +|------|---------| +| **Account** | A billable identity with a provider (your Claude Pro account, your Devin org, your OpenRouter key, your local Ollama endpoint). | +| **Entitlement** | A unit of capacity attached to an account: a fixed-window subscription allowance, a prepaid credit balance, or a local compute slot. | +| **Provider** | A model-serving endpoint with a well-defined protocol (`anthropic`, `openai_compat`, `codex_oauth`, `cursor_agent`, `local_openai`). | +| **Route** | A concrete binding of a provider + account + model id + optional overrides. Mirrors `routes` in `config.json`. | +| **Task Tier** | A classification of a request by workload shape: `planning`, `heavy_reasoning`, `bulk_context`, `frontend`. | +| **Capability Card** | Structured metadata describing what a model/route can do, its context window, vision/tool support, and latency/cost class. | +| **Ledger** | The in-memory + persisted record of accounts, entitlements, quota usage, cooldowns, and circuit-breaker states. | + +--- + +## 2. Domain Model + +``` +Request + │ + ▼ +[Classifier / Task-Tier Detector] ──▶ TaskTier + Constraints + │ + ▼ +[Routing Engine] + │ reads: Ledger, Capability Index, Health Registry, Telemetry Cache + │ writes: Telemetry, Ledger updates, Honcho sync, Life OS metrics + │ + ▼ +[Provider Adapter] ──▶ chosen Route ──▶ upstream model + │ + ▼ +[Telemetry Collector] ──▶ Ledger update + Honcho + Life OS +``` + +Core entities: + +- `Request`: the inbound conversation/prompt, tools list, image attachments, desired max_tokens, and explicit routing hints (e.g. `[[route:opus]]`). +- `TaskTier`: one of `planning`, `heavy_reasoning`, `bulk_context`, `frontend`, plus a confidence score. +- `CapabilityProfile`: per-route metadata (context window, vision, tool support, reasoning, output quality per tier). +- `Route`: `route_id`, `provider_type`, `upstream`, `model`, `account_id`, `auth_ref`, `headers`, `body_overrides`, `capability_profile`. +- `Account`: `account_id`, `provider`, `account_kind` (`subscription`, `prepaid`, `local`), list of `Entitlement`s, metadata. +- `Entitlement`: `kind` (`fixed_window`, `prepaid`, `local_compute`), `window`, `limit`, `used`, `remaining`, `resets_at`, `cooldown_until`, `cost_per_token` or `cost_per_request`. +- `RoutingDecision`: `request_id`, chosen `route_id`, `tier`, `scores`, `fallback_chain`, `reason`, `estimated_cost`. +- `TelemetryEvent`: per-request metrics including TTFT, latency, tokens, outcome, rate-limit signals, and final ledger deltas. + +--- + +## 3. Provider & Account Ledger + +### 3.1 Provider Taxonomy + +| Provider | Billing model | Quota semantics | Example route type | +|----------|--------------|-------------------|--------------------| +| **Anthropic (Claude)** | Subscription Pro/Max | Messages per N hours, input/output tokens per day, rate limits | `anthropic` passthrough | +| **OpenAI (ChatGPT/Codex)** | ChatGPT Plus / Codex credits | Per-day/hour message limits, rate limits, model-tier quotas | `codex_oauth` or `openai_compat` | +| **Cursor** | Subscription (Composer) | Rate-limited, no published hard cap; treat as soft quota + cooldown | `cursor_agent` | +| **Grok** | X Premium+ / API credits | Subscription window + prepaid API tier | `openai_compat` to xAI | +| **Devin** | Prepaid ACU/org credits | ACU balance consumed per session/operation | adapter bridge | +| **OpenRouter** | Prepaid credits + per-request cost | Balance decrements per token | `openai_compat` | +| **Generic API-key gateways** | Prepaid or metered | Balance or monthly limit | `openai_compat` | +| **Local compute** | Zero marginal | Concurrent slots, vRAM, tokens/sec throughput | `local_openai` / Ollama / vLLM | + +### 3.2 Account Schema + +```python +@dataclass +class Account: + account_id: str # stable, user-defined or derived + provider: str # e.g. "anthropic", "openrouter", "local_rtx5090" + kind: AccountKind # subscription | prepaid | local + display_name: str + secrets_ref: SecretsRef # {"auth": "Bearer ${ANTHROPIC_API_KEY}", ...} + entitlements: List[Entitlement] + metadata: Dict[str, Any] + enabled: bool = True +``` + +`account_id` must be unique per user/org. Multiple routes may share one account (e.g., several OpenRouter model slugs draw from the same OpenRouter balance). + +### 3.3 Subscription Model (Fixed-Window) + +Fixed-window subscriptions carry one or more `WindowedEntitlement`s: + +- `window_type`: `rolling` or `calendar` (hourly, daily, monthly). +- `limit`: max allowed units in the window. +- `used`: units consumed in the current window. +- `resets_at`: timestamp when the window resets. +- `unit`: `requests`, `input_tokens`, `output_tokens`, `messages`. +- `overage_policy`: `block` (hard stop), `warn` (allow but mark), `spill_to` (route to prepaid fallback). + +Examples: +- Claude Pro: 5x messages/4h for Opus; 50x messages/4h for Sonnet. +- ChatGPT Plus: ~40 messages/3h for GPT-4 class. +- Grok: rate-limited requests per minute/hour depending on tier. + +### 3.4 Prepaid Credit Pools + +Prepaid accounts use `BalanceEntitlement`: + +- `currency`: USD, credits, ACU, OR-credits. +- `balance`: remaining units. +- `cost_per_input_token`, `cost_per_output_token`, `cost_per_request`. +- `minimum_balance`: stop routing when balance below this. +- `top_up_url`: optional link for manual top-up. + +The ledger updates `balance` after every request using observed token counts (or, if unavailable, estimated from prompt/completion sizes). + +### 3.5 Zero-Marginal-Cost Local Compute (RTX 5090) + +Local accounts use `LocalComputeEntitlement`: + +- `max_concurrent`: how many requests can run in parallel. +- `current_load`: in-flight count. +- `vram_total_mb`, `vram_reserved_mb`. +- `max_context`: model context length loaded. +- `tokens_per_second`: observed throughput. +- `priority`: routing priority (`always_first`, `tier_based`, `fallback_only`). + +The router prefers local compute when: +- `priority` is `always_first` or the task tier is `bulk_context`/`frontend` and local can satisfy it. +- `current_load < max_concurrent` and available vRAM >= estimated model memory. +- The loaded model's capability profile scores >= the tier threshold. + +--- + +## 4. Capability & Task-Tier Matrix + +### 4.1 Task Tiers + +| Tier | Typical request | Needs | +|------|----------------|-------| +| `planning` | "Design a multi-tenant auth system" | High reasoning, architecture, large context, reliable tool use. | +| `heavy_reasoning` | "Debug this non-deterministic concurrency bug" | Maximum reasoning, deep debugging, long chains, may need vision for screenshots. | +| `bulk_context` | "Summarize these 100 issues" / "Find all usages of X" | Huge context, cheap per token, moderate quality acceptable. | +| `frontend` | "Generate this React component from this screenshot" | Vision, fast, cheap, moderate reasoning; may prefer Claude/GPT-4o class. | + +Tiers can be detected by: +- explicit directive (`[[tier:heavy]]`, `[[route:opus]]`, existing `[[route:...]]` syntax). +- heuristic: prompt length, presence of images, tool count, keywords ("plan", "refactor", "summarize", "find all"). +- a small classifier model (same role as today's `classifier` in `Auto Router`). + +### 4.2 Capability Profile + +Each route has a `CapabilityProfile`: + +```python +@dataclass +class CapabilityProfile: + context_window: int + supports_vision: bool + supports_tools: bool + supports_reasoning_split: bool + scores: Dict[TaskTier, float] # 0.0 .. 1.0 + cost_class: str # "free", "cheap", "mid", "premium" + latency_class: str # "fast", "normal", "slow" + quality_class: str # "local", "good", "great", "frontier" + tags: List[str] # e.g. ["local", "32k-context", "vision"] +``` + +### 4.3 Example Capability/Task-Tier Matrix + +| Route | Context | Vision | Tools | planning | heavy_reasoning | bulk_context | frontend | +|-------|---------|--------|-------|----------|-----------------|--------------|----------| +| `claude-opus` (Anthropic Pro) | 200K/1M | Yes | Yes | 0.95 | 0.98 | 0.85 | 0.90 | +| `claude-sonnet` | 200K | Yes | Yes | 0.90 | 0.90 | 0.90 | 0.90 | +| `gpt-5.5-codex` | 256K | Yes | Yes | 0.92 | 0.95 | 0.80 | 0.88 | +| `claude-mimo` | 1M | No | Yes | 0.75 | 0.65 | 0.95 | 0.60 | +| `claude-minimax-m3` | 1M | No | Yes | 0.80 | 0.75 | 0.95 | 0.55 | +| `local-llama-3.3-70b-rtx5090` | 128K | No | Yes | 0.60 | 0.55 | 0.85 | 0.50 | +| `local-qwen3-32b-rtx5090` | 128K | No | Yes | 0.70 | 0.65 | 0.90 | 0.60 | + +The matrix is user-editable in `config.json`; the scaffold ships sensible defaults. + +--- + +## 5. Routing Decision Algorithm + +### 5.1 Inputs + +- `request`: prompt, tools, images, context length, explicit hints. +- `tier`: detected or pinned task tier. +- `ledger`: accounts, entitlements, cooldowns, circuit states. +- `capability_index`: all routes with `CapabilityProfile`. +- `health_registry`: last health check per route, failure counts. +- `telemetry_cache`: recent latency and cost observations. + +### 5.2 Selection Constraints + +A route is **eligible** only if all of the following hold: + +1. `enabled` in config and account `enabled`. +2. Capability score for `tier` >= `tier_threshold` (config per tier, default 0.70). +3. Context length of route >= estimated prompt length + requested `max_tokens`. +4. If images present, `supports_vision` is true. +5. If tools present, `supports_tools` is true. +6. Account has a usable entitlement: + - fixed-window: `used < limit` and `now < resets_at` and `now > cooldown_until`. + - prepaid: `balance > minimum_balance + estimated_cost`. + - local: `current_load < max_concurrent` and enough vRAM. +7. Circuit breaker is `closed`. + +### 5.3 Cost Function + +For eligible routes, compute an `effective_cost` used for sorting: + +``` +monetary_cost = estimated_input_tokens * cost_per_input_token + + estimated_output_tokens * cost_per_output_token + + cost_per_request + +capacity_pressure = used / limit # for windowed entitlements +opportunity_cost = monetary_cost * (1 + capacity_pressure) + +latency_penalty = observed_p95_latency_ms * latency_weight +quality_bonus = -score * quality_weight # higher score lowers effective cost + +effective_cost = opportunity_cost + latency_penalty + quality_bonus +``` + +Local compute uses `monetary_cost = 0` but may include a small `capacity_pressure` term to avoid overloading the GPU. + +### 5.4 Priority Sorting + +1. **Local zero-marginal-cost routes first**, sorted by capability score (desc), then load (asc). +2. **Subscription routes**, sorted by `effective_cost` (asc). +3. **Prepaid routes**, sorted by `effective_cost` (asc). +4. Any route with explicit `[[route:...]]` directive is promoted to the top of its tier group (it must still be eligible; if not, fail with a clear error). + +### 5.5 Pseudocode + +```python +def select_route(request, ledger, index, health, telemetry) -> RoutingDecision: + request_id = generate_uuid() + tier = detect_tier(request) # may use classifier or hints + + candidates = [] + for route in index.routes: + if not route.enabled: + continue + if not satisfies_constraints(route, request, tier): + continue + account = ledger.account(route.account_id) + if not account.has_capacity_for(route, request): + continue + if health.circuit_is_open(route.route_id): + continue + + score = route.capability.scores.get(tier, 0.0) + if score < TIER_THRESHOLD[tier]: + continue + + cost = estimate_effective_cost(route, account, request, telemetry) + candidates.append((route, score, cost)) + + # Priority: local first, then cost-sorted + candidates.sort(key=lambda rc: ( + 0 if route_is_local(rc[0]) else 1, + -rc[1], # higher capability first (tie-break) + rc[2] # lower effective cost + )) + + if not candidates: + return RoutingDecision( + request_id=request_id, + route_id=None, + tier=tier, + outcome="no_eligible_route", + reason="No route satisfied constraints, quota, health, and circuit-breaker checks." + ) + + # Try primary and failover chain + fallback_chain = [c[0] for c in candidates[:MAX_FAILOVER_DEPTH]] + for route in fallback_chain: + if try_reserve_capacity(route, ledger): + return RoutingDecision( + request_id=request_id, + route_id=route.route_id, + tier=tier, + fallback_chain=fallback_chain, + estimated_cost=cost_for(route, request), + reason=f"capability={route.capability.scores[tier]:.2f}; effective_cost={cost:.4f}" + ) + + return RoutingDecision(request_id=request_id, route_id=None, tier=tier, + outcome="reservation_failed", + reason="All eligible routes failed capacity reservation.") +``` + +### 5.6 Failover on Runtime Errors + +When a call fails: + +- `429` / rate limit: mark route/account cooldown from `Retry-After` or exponential backoff; retry next candidate. +- `5xx` / timeout / connection error: increment failure counter; if threshold reached, open circuit breaker; retry next candidate. +- Empty or invalid response: retry same route up to `MAX_EMPTY_RETRIES`, then fail over. +- No remaining candidates: surface a structured error to the caller with the last error and suggestions (wait, top-up, switch tier). + +A **dead or empty route** (no configured upstream, auth missing, model unavailable) is marked `unhealthy` immediately and skipped at selection time. + +--- + +## 6. Quota-Window Accounting + +### 6.1 Window Types + +| Window type | Description | +|-------------|-------------| +| `sliding` | Most recent `window_size` of activity (e.g., last 4 hours). | +| `calendar` | Resets at fixed boundaries (midnight UTC, top of hour). | +| `session` | Counts within a single Claude Code session. | + +### 6.2 Counter Implementation + +- Maintain an append-only event log of `(timestamp, account_id, route_id, unit, delta)`. +- For sliding windows, keep events in a deque pruned by age. +- For calendar windows, snapshot `used` at reset boundaries. +- All counters are persisted to disk and synced to Honcho for cross-device consistency. + +### 6.3 Over-Usage Protection + +- **Pessimistic reservation**: decrement quota when a request starts; restore on failure. +- **Reconciliation**: after receiving actual usage from provider response headers, adjust the ledger to match. +- **Spill-over**: if a fixed-window entitlement is exhausted but the user has a prepaid account for the same provider/model, the request can spill to prepaid (configurable per account). + +--- + +## 7. Health Checks, Timeouts, and Circuit Breakers + +### 7.1 Health Checks + +Each route has a lightweight health check performed periodically and on-demand: + +- `GET /v1/models` or equivalent cheap endpoint. +- Measures: reachable, latency, auth accepted, model listed, `429`/`403` status. +- Health state: `healthy`, `degraded`, `unhealthy`, `unknown`. + +### 7.2 Timeouts + +| Tier | First-token timeout (TTFT) | Total timeout | +|------|---------------------------|---------------| +| `frontend` | 5s | 30s | +| `planning` | 10s | 120s | +| `heavy_reasoning` | 15s | 300s | +| `bulk_context` | 8s | 180s | + +Per-route overrides are allowed in `config.json`. + +### 7.3 Circuit Breakers + +Per route/account: + +- `failure_threshold`: number of consecutive failures before opening. +- `slow_request_threshold`: number of TTFT violations before marking degraded. +- `open_duration`: seconds the circuit stays open. +- `half_open_max`: number of probe requests allowed while half-open. + +State machine: `closed -> open -> half_open -> closed` or `open`. + +### 7.4 Cooldown Timers + +- Triggered by explicit rate-limit (`429`) or `Retry-After` header. +- Stored in ledger as `cooldown_until` per account/entitlement. +- During cooldown, route is excluded unless user explicitly forces it. + +--- + +## 8. Cost and Quota Telemetry + +### 8.1 Telemetry Event Schema + +```python +@dataclass +class TelemetryEvent: + event_id: str + request_id: str + route_id: str + account_id: str + provider: str + model_id: str + tier: str + timestamp: datetime + ttft_ms: Optional[float] # Time-To-First-Token + e2e_latency_ms: float + prompt_tokens: int + completion_tokens: int + total_tokens: int + estimated_cost: float + actual_cost: Optional[float] + remaining_quota: Optional[float] + rate_limit_headers: Dict[str, str] + outcome: str # success, failure, cached, cancelled + error_code: Optional[str] + error_message: Optional[str] + cached: bool + fallback_index: int + request_size_bytes: int + response_size_bytes: int +``` + +### 8.2 TTFT and End-to-End Latency + +- `ttft_ms` measured from request send to first streamed chunk. +- `e2e_latency_ms` measured from request start to final chunk/complete. +- Latency histograms are kept per route and tier (p50, p95, p99). + +### 8.3 Token Spend and Remaining Allowances + +- Token counts sourced from provider response body `usage` or response headers. +- For `openai_compat`, read `usage.prompt_tokens` / `completion_tokens`. +- For Anthropic passthrough, read `usage.input_tokens` / `output_tokens`. +- If missing, estimate using the same tokenizer approximation used for routing. +- Deduct from the appropriate entitlement and record `remaining_quota`. + +### 8.4 Rate-Limit / 429 Detection + +- Parse `x-ratelimit-*`, `Retry-After`, `x-ratelimit-remaining`, `x-ratelimit-reset`. +- Normalize to a `RateLimitSnapshot`: + - `requests_remaining`, `requests_reset` + - `tokens_remaining`, `tokens_reset` +- Trigger cooldown when `remaining == 0` or `429` returned. + +### 8.5 Cooldown and Backoff Telemetry + +- Record cooldown events: `route_id`, `reason` (`429`, `timeout`, `manual`), `cooldown_until`. +- Exponential backoff per account (1s, 2s, 4s, ... capped at `max_cooldown`). +- Reset on successful health check. + +--- + +## 9. Failover Policy + +### 9.1 Failover Triggers + +- Provider returns HTTP error (`>= 500`, `429`, `403` with auth failure). +- Request exceeds TTFT or total timeout. +- Empty or malformed response after `MAX_EMPTY_RETRIES`. +- Circuit breaker opens during the request. +- Quota exhausted mid-flight (returned by provider). + +### 9.2 Failover Order + +The `fallback_chain` produced by the routing algorithm is the ordered list of candidates. The engine tries each in order, applying backoff/cooldown updates after each failure, until: + +- a candidate succeeds, +- the chain is exhausted, +- or the user cancels. + +### 9.3 Preventing Hangs on Dead/Empty Routes + +- Routes with no `upstream`, missing `model`, or unresolved `${VAR}` are marked `unhealthy` at config load and excluded. +- Selection always computes the full fallback chain before the first outbound request. +- Each attempt has a hard deadline; if the overall request deadline is exceeded, abort and return a structured error. +- No request is allowed to block on a single provider for longer than the per-tier total timeout. + +--- + +## 10. Honcho Synchronization Contract + +### 10.1 What We Sync + +Honcho stores durable, cross-session state for the OnlyTerp user. The engine pushes: + +- `LedgerSnapshot`: accounts, entitlements, cooldowns, circuit breaker states. +- `TelemetryBatch`: recent `TelemetryEvent`s. +- `RoutingDecisionLog`: recent decisions for explainability. + +### 10.2 Sync API (placeholder) + +```python +class HonchoSyncClient: + def __init__(self, app_id: str, user_id: str, base_url: str, api_key_ref: str): ... + def push_snapshot(self, snapshot: LedgerSnapshot, request_id: str) -> None: ... + def push_telemetry(self, batch: List[TelemetryEvent]) -> None: ... + def fetch_latest(self) -> Optional[LedgerSnapshot]: ... +``` + +### 10.3 Consistency Model + +- Sync is **asynchronous and best-effort**; routing continues if Honcho is unreachable. +- Each sync payload carries a `request_id` and monotonic `sequence` number for idempotency. +- Honcho merges snapshots by `sequence` (last-write-wins per account/entitlement). +- Initial load attempts to hydrate the ledger from Honcho at startup; if unavailable, fall back to local disk. + +--- + +## 11. Life OS Metrics/API + +Life OS (terpOS) displays real-time cost/quota/health dashboards. The engine exposes an internal metrics surface that can be consumed by `terpOS` or any authorized client. + +### 11.1 Metrics Endpoints + +``` +GET /life-os/metrics/routing +GET /life-os/metrics/providers +GET /life-os/metrics/cost +GET /life-os/metrics/health +GET /life-os/metrics/tiers +WS /life-os/metrics/stream +``` + +### 11.2 Metrics Payload + +```python +@dataclass +class RoutingMetrics: + window: str # e.g., "1h", "24h", "7d" + total_requests: int + requests_by_tier: Dict[str, int] + requests_by_provider: Dict[str, int] + total_estimated_cost: float + cost_by_provider: Dict[str, float] + avg_ttft_ms: float + p95_ttft_ms: float + avg_e2e_latency_ms: float + p95_e2e_latency_ms: float + quota_remaining: Dict[str, float] + provider_health: Dict[str, str] + active_cooldowns: List[CooldownRecord] +``` + +### 11.3 Push vs Pull + +- **Pull:** `terpOS` polls the metrics endpoints. +- **Push:** optional `UC_LIFE_OS_WEBSOCKET` pushes streaming events when enabled. + +--- + +## 12. Security and Secret Handling + +- Secrets (API keys, OAuth tokens) **never** enter the ledger or telemetry; they live only in `config.json` (gitignored) or environment variables. +- Provider adapters resolve `${VAR}` references at request time and never log resolved values. +- Ledger persistence is encrypted at rest when possible (e.g., via OS keyring or `cryptography` if user opts in); otherwise stored in the user state directory. +- Honzo/Life OS traffic is sent over TLS with authenticated `Authorization` headers. +- The proxy's existing `GUARD_LOCAL` Host-header checks remain active; no new admin endpoints are exposed beyond `/healthz`, `/metrics`, and `/life-os/*`. +- Audit: every `RoutingDecision` and quota mutation is logged with `request_id` for traceability. + +--- + +## 13. Consistency and Idempotency + +### 13.1 Request IDs + +- A single `request_id` is generated at the routing boundary and propagated through provider calls, telemetry, ledger updates, and Honcho sync. +- All downstream events are keyed by `request_id` and `event_id`. + +### 13.2 Ledger Atomicity + +- Quota reservation and spend updates are applied in memory first, then persisted to disk and Honcho asynchronously. +- A crash between in-memory update and persistence is recovered on next startup by replaying the local event log or fetching from Honcho. + +### 13.3 Idempotent Sync + +- Sync payloads include `sequence` and `request_id`. +- Honcho rejects older sequences and deduplicates by `request_id`. + +--- + +## 14. Observability + +### 14.1 Logging + +- Structured logs with `request_id`, `route_id`, `tier`, `outcome`. +- `UC_ROUTING_LOG=1` enables decision logging (same spirit as `UC_ROUTER_LOG`). +- Log levels: `INFO` for routing decisions, `WARN` for quota/cooldown events, `ERROR` for failover exhaustion. + +### 14.2 Metrics + +- Counters: `routing_requests_total`, `routing_failovers_total`, `routing_quota_exhausted_total`. +- Histograms: `routing_latency_ms`, `provider_ttft_ms`, `provider_e2e_latency_ms`. +- Gauges: `provider_health`, `quota_remaining`, `active_cooldowns`, `circuit_breaker_state`. + +### 14.3 Tracing + +- Minimal in-process spans: `classify_tier`, `select_route`, `provider_call`, `update_ledger`, `sync_honcho`. +- Optional OpenTelemetry exporter; core engine remains stdlib-only. + +--- + +## 15. Testing Strategy + +### 15.1 Unit Tests + +- `test_ledger.py`: windowed/prepaid/local entitlement accounting, cooldowns, spill-over. +- `test_routing.py`: constraint satisfaction, cost sorting, tier detection, failover chain ordering. +- `test_failover.py`: circuit breaker transitions, cooldown backoff, timeout handling. +- `test_telemetry.py`: event schema, rate-limit header parsing, token reconciliation. +- `test_honcho_sync.py`: snapshot serialization, idempotent push, fetch fallback. + +### 15.2 Integration Tests + +- Mock provider servers (same pattern as `test_proxy.py`) for Anthropic, OpenAI-compatible, and local backends. +- End-to-end scenarios: + - route a request to local compute first. + - exhaust a fixed-window quota and spill to prepaid. + - trigger 429 and verify cooldown + failover. + - open circuit breaker after repeated failures. + +### 15.3 Property Tests + +- Quota windows: for all window types and sizes, `used` is always <= `limit` after reservation, and resets happen exactly once per window. +- Failover: with at least one healthy route, every request eventually succeeds or returns a structured `no_eligible_route`. +- Cost: the selected route is never more expensive than another eligible route with higher or equal capability and lower actual cost, unless explicitly pinned. + +### 15.4 Load / Concurrency Tests + +- Many concurrent requests hit the router; verify: + - no over-commit of local compute slots. + - ledger counters remain accurate. + - circuit breakers do not flap. + +--- + +## 16. Rollout and Migration Plan + +### Phase 1: Spec and scaffold (this PR) + +- Merge `docs/ROUTING_ENGINE_SPEC.md` and the `uc_routing/` package with placeholder interfaces. +- No changes to `proxy.py` behavior. + +### Phase 2: Ledger and telemetry in memory + +- Implement in-memory `Ledger`, `TelemetryCollector`, and `RoutingEngine`. +- Add unit tests. + +### Phase 3: Provider adapter integration + +- Refactor existing `proxy.py` route dispatch into `uc_routing/providers/adapter.py` wrappers. +- Keep `proxy.py` as the HTTP server and protocol translator; routing decisions move into `uc_routing/`. +- Add integration tests with mock backends. + +### Phase 4: Honcho and Life OS integration + +- Implement `HonchoSyncClient` and `LifeOSMetrics` behind config flags. +- Add persistence and cross-device sync. + +### Phase 5: Gradual enablement + +- Introduce `UC_ROUTING_ENGINE=0/1` (default `0`). +- Run A/B against `Auto Router`. +- When stable, make it the default and deprecate the old `router` block. + +--- + +## 17. Open Questions and Assumptions + +### Assumptions + +1. The engine is implemented in Python 3.8+, using only the standard library for core logic; optional integrations (Honcho SDK, OpenTelemetry) may be installed by the user but are not required. +2. Existing `config.example.json` remains the canonical user-facing config; engine-specific config lives under a new `routing_engine` key. +3. Provider secrets continue to be interpolated from `${VAR}` at request time and are never persisted by the ledger. +4. Local compute is exposed as an OpenAI-compatible HTTP endpoint (Ollama, LM Studio, vLLM, llama.cpp server) on `127.0.0.1`. +5. Honcho and Life OS base URLs and auth are configured under `routing_engine.honcho` and `routing_engine.life_os`. + +### Open Questions + +1. What is the exact Honcho app/user ID scheme and API surface? (The `HonchoSyncClient` is a placeholder until this is finalized.) +2. Should Cursor Composer be modeled as a first-class route with its own quota semantics, or kept as an experimental `cursor_agent` provider? +3. How do we obtain accurate rate-limit headers from Devin and Grok? Do they expose the same `x-ratelimit-*` conventions? +4. Should the classifier for tier detection be a separate cheap model call, or can we reuse the existing `Auto Router` classifier and add a tier output? +5. What is the desired granularity of Life OS dashboards (per minute, per hour, per session)? +6. Should the engine support multi-user/org ledger separation, or is it single-user per proxy instance? +7. Do we need a SQLite backend for the ledger, or is JSON-on-disk + Honcho sufficient for v1? +8. How should we normalize cost across different credit types (USD, OpenRouter credits, ACU)? + +--- + +## 18. Repo Structure Scaffold + +The following package is added under `uc_routing/` and is designed to be wired into `proxy.py` later without disrupting the current `providers/`, `scripts/`, `examples/`, or `docs/` layout. + +``` +UltraCode-Shim/ +├── docs/ +│ └── ROUTING_ENGINE_SPEC.md # this document +├── uc_routing/ +│ ├── __init__.py # public API exports +│ ├── README.md # package overview and integration notes +│ ├── ledger/ +│ │ ├── __init__.py +│ │ ├── models.py # Account, Entitlement, Ledger +│ │ ├── window.py # fixed-window accounting +│ │ └── balance.py # prepaid/local compute accounting +│ ├── telemetry/ +│ │ ├── __init__.py +│ │ ├── schema.py # TelemetryEvent, RateLimitSnapshot +│ │ ├── collector.py # in-memory collector + histograms +│ │ └── cost.py # cost estimation / normalization +│ ├── routing/ +│ │ ├── __init__.py +│ │ ├── engine.py # RoutingEngine.select_route() +│ │ ├── policy.py # selection, sorting, failover policy +│ │ ├── decision.py # RoutingDecision dataclass +│ │ └── task_tiers.py # TaskTier enum + detection +│ ├── providers/ +│ │ ├── __init__.py +│ │ ├── types.py # ProviderType, CapabilityProfile +│ │ ├── adapter.py # Abstract / concrete adapter interface +│ │ └── registry.py # route index from config.json +│ ├── failover/ +│ │ ├── __init__.py +│ │ ├── health.py # health checks and registry +│ │ ├── circuit.py # circuit breaker +│ │ └── cooldown.py # cooldown/backoff timers +│ ├── honcho/ +│ │ ├── __init__.py +│ │ ├── contract.py # data contracts for sync +│ │ └── sync.py # HonchoSyncClient placeholder +│ ├── life_os/ +│ │ ├── __init__.py +│ │ ├── api.py # metrics endpoints / handlers +│ │ └── metrics.py # RoutingMetrics + aggregation +│ ├── config/ +│ │ ├── __init__.py +│ │ ├── schema.py # engine-specific config dataclasses +│ │ └── loader.py # load engine config from config.json +│ └── tests/ +│ ├── __init__.py +│ ├── test_ledger.py +│ ├── test_routing.py +│ ├── test_failover.py +│ └── test_telemetry.py +├── proxy.py # existing; integration point for later +├── test_proxy.py # existing +├── config.example.json # existing; engine section added later +└── README.md # existing +``` + +### Integration Notes + +- `uc_routing` is a sibling to the existing `providers/` directory. The new `uc_routing/providers/adapter.py` will eventually wrap the existing `providers/codex_oauth.py` and `providers/cursor_agent.py` helpers, plus add new adapters for Anthropic passthrough and OpenAI-compatible endpoints. +- `uc_routing/config/loader.py` reads the same `config.json` used by `proxy.py` and expects engine config under a top-level `routing_engine` key. If the key is absent, the engine disables itself and `proxy.py` continues using the existing `Auto Router`. +- The engine is designed to be **feature-flagged** (`UC_ROUTING_ENGINE`) so the existing proxy behavior is unchanged until explicitly enabled. + +--- + +## Appendix A: Sample Config Snippet + +```jsonc +{ + "_routing_engine": "New subscription/quota-aware routing. Set enabled:true to opt in. See docs/ROUTING_ENGINE_SPEC.md.", + "routing_engine": { + "enabled": false, + "tier_thresholds": { + "planning": 0.80, + "heavy_reasoning": 0.90, + "bulk_context": 0.60, + "frontend": 0.70 + }, + "honcho": { + "enabled": false, + "base_url": "${HONCHO_BASE_URL}", + "app_id": "onlyterp-routing", + "api_key_ref": "${HONCHO_API_KEY}" + }, + "life_os": { + "enabled": false, + "base_url": "${LIFE_OS_BASE_URL}", + "push_stream": false + }, + "accounts": [ + { + "account_id": "anthropic-pro", + "provider": "anthropic", + "kind": "subscription", + "entitlements": [ + {"kind": "fixed_window", "unit": "messages", "window_type": "sliding", "window_size_hours": 4, "limit": 5} + ] + }, + { + "account_id": "openrouter-pool", + "provider": "openrouter", + "kind": "prepaid", + "entitlements": [ + {"kind": "prepaid", "currency": "USD", "balance": 25.00, "cost_per_input_token": 0.0000005, "cost_per_output_token": 0.0000015} + ] + }, + { + "account_id": "local-rtx5090", + "provider": "ollama", + "kind": "local", + "entitlements": [ + {"kind": "local_compute", "max_concurrent": 2, "vram_total_mb": 24576, "priority": "always_first"} + ] + } + ] + } +} +``` + +--- + +*End of specification. Implementation placeholders live in `uc_routing/`.* diff --git a/uc_routing/README.md b/uc_routing/README.md new file mode 100644 index 0000000..9aad2c6 --- /dev/null +++ b/uc_routing/README.md @@ -0,0 +1,35 @@ +# `uc_routing` — Subscription- and Quota-Aware Routing Engine + +This package is the implementation scaffold for the routing engine specified in +[`docs/ROUTING_ENGINE_SPEC.md`](../docs/ROUTING_ENGINE_SPEC.md). It is designed to +be wired into `proxy.py` later without disrupting the existing Auto Router until +the engine is stable. + +## Design + +- **Ledger** (`ledger/`) — tracks accounts, subscriptions, prepaid pools, and local compute capacity. +- **Telemetry** (`telemetry/`) — records per-request latency, token spend, rate limits, and quota state. +- **Routing** (`routing/`) — decides which provider/model to use for a given task tier. +- **Provider Adapters** (`providers/`) — wraps the existing `providers/` helpers and OpenAI/Anthropic endpoints. +- **Failover / Health** (`failover/`) — circuit breakers, cooldowns, health checks, and timeout policies. +- **Honcho Sync** (`honcho/`) — pushes ledger and telemetry state to Honcho for cross-device consistency. +- **Life OS Metrics** (`life_os/`) — exposes metrics for the terpOS / Life OS dashboard. +- **Config** (`config/`) — loads engine-specific settings from `config.json`. + +## Status + +Placeholder interfaces only. No production behavior is implemented yet. See the +spec for the full design, algorithms, and rollout plan. + +## Running Tests + +```bash +python3 -m unittest uc_routing.tests +``` + +## Integration + +1. Add a `routing_engine` section to `config.json` (see spec Appendix A). +2. Import and instantiate `RoutingEngine` from `proxy.py`. +3. Route `POST /v1/messages` through `engine.select_route()` and dispatch via `uc_routing.providers` adapters. +4. Enable with `UC_ROUTING_ENGINE=1` once the implementation is complete. diff --git a/uc_routing/__init__.py b/uc_routing/__init__.py new file mode 100644 index 0000000..22b3f0d --- /dev/null +++ b/uc_routing/__init__.py @@ -0,0 +1,27 @@ +"""Subscription- and quota-aware AI model routing engine for UltraCode-Shim. + +This package is a scaffold. Importing it does not enable the engine; wire it into +`proxy.py` and flip `UC_ROUTING_ENGINE=1` when the implementation is ready. +""" + +from .ledger.models import Account, AccountKind, Entitlement, Ledger +from .providers.types import CapabilityProfile, ProviderType, Route +from .routing.decision import RoutingDecision +from .routing.engine import RoutingEngine +from .routing.task_tiers import TaskTier, TaskTierDetector +from .telemetry.schema import TelemetryEvent + +__all__ = [ + "Account", + "AccountKind", + "CapabilityProfile", + "Entitlement", + "Ledger", + "ProviderType", + "Route", + "RoutingDecision", + "RoutingEngine", + "TaskTier", + "TaskTierDetector", + "TelemetryEvent", +] diff --git a/uc_routing/config/__init__.py b/uc_routing/config/__init__.py new file mode 100644 index 0000000..dbc768c --- /dev/null +++ b/uc_routing/config/__init__.py @@ -0,0 +1,12 @@ +"""Engine-specific configuration loading.""" + +from .loader import EngineConfig, load_engine_config +from .schema import HonchoConfig, LifeOSConfig, RoutingEngineConfig + +__all__ = [ + "EngineConfig", + "HonchoConfig", + "LifeOSConfig", + "RoutingEngineConfig", + "load_engine_config", +] diff --git a/uc_routing/config/loader.py b/uc_routing/config/loader.py new file mode 100644 index 0000000..d7dabf3 --- /dev/null +++ b/uc_routing/config/loader.py @@ -0,0 +1,58 @@ +"""Load engine config from `config.json`.""" + +from __future__ import annotations + +import json +import os +from dataclasses import dataclass +from typing import Any, Dict, Optional + +from .schema import HonchoConfig, LifeOSConfig, RoutingEngineConfig + + +@dataclass +class EngineConfig: + """Top-level container returned by the loader.""" + + config: RoutingEngineConfig + source_path: Optional[str] = None + + +def load_engine_config(path: Optional[str] = None) -> EngineConfig: + """Load `routing_engine` section from `config.json`. + + If `path` is omitted, reads `UC_CONFIG` env var, then `config.json`, then + `config.example.json`, mirroring `proxy.py` behavior. + """ + candidates = [ + path, + os.environ.get("UC_CONFIG"), + "config.json", + "config.example.json", + ] + for candidate in candidates: + if not candidate: + continue + if os.path.isfile(candidate): + with open(candidate, "r", encoding="utf-8") as f: + data = json.load(f) + cfg = data.get("routing_engine", {}) + return EngineConfig( + config=RoutingEngineConfig( + enabled=cfg.get("enabled", False), + tier_thresholds=cfg.get( + "tier_thresholds", + { + "planning": 0.80, + "heavy_reasoning": 0.90, + "bulk_context": 0.60, + "frontend": 0.70, + }, + ), + honcho=HonchoConfig(**cfg.get("honcho", {})), + life_os=LifeOSConfig(**cfg.get("life_os", {})), + accounts=cfg.get("accounts", []), + ), + source_path=candidate, + ) + return EngineConfig(config=RoutingEngineConfig()) diff --git a/uc_routing/config/schema.py b/uc_routing/config/schema.py new file mode 100644 index 0000000..8329c06 --- /dev/null +++ b/uc_routing/config/schema.py @@ -0,0 +1,37 @@ +"""Typed configuration schema for the routing engine.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +@dataclass +class HonchoConfig: + enabled: bool = False + base_url: Optional[str] = None + app_id: str = "onlyterp-routing" + api_key_ref: Optional[str] = None + + +@dataclass +class LifeOSConfig: + enabled: bool = False + base_url: Optional[str] = None + push_stream: bool = False + + +@dataclass +class RoutingEngineConfig: + enabled: bool = False + tier_thresholds: Dict[str, float] = field( + default_factory=lambda: { + "planning": 0.80, + "heavy_reasoning": 0.90, + "bulk_context": 0.60, + "frontend": 0.70, + } + ) + honcho: HonchoConfig = field(default_factory=HonchoConfig) + life_os: LifeOSConfig = field(default_factory=LifeOSConfig) + accounts: List[Dict[str, Any]] = field(default_factory=list) diff --git a/uc_routing/failover/__init__.py b/uc_routing/failover/__init__.py new file mode 100644 index 0000000..038ad62 --- /dev/null +++ b/uc_routing/failover/__init__.py @@ -0,0 +1,7 @@ +"""Failover primitives: health checks, circuit breakers, cooldowns.""" + +from .circuit import CircuitBreaker +from .cooldown import CooldownManager +from .health import HealthRegistry, HealthState + +__all__ = ["CircuitBreaker", "CooldownManager", "HealthRegistry", "HealthState"] diff --git a/uc_routing/failover/circuit.py b/uc_routing/failover/circuit.py new file mode 100644 index 0000000..224f344 --- /dev/null +++ b/uc_routing/failover/circuit.py @@ -0,0 +1,58 @@ +"""Circuit breaker state machine.""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Dict + + +class CircuitState(enum.Enum): + CLOSED = "closed" + OPEN = "open" + HALF_OPEN = "half_open" + + +@dataclass +class CircuitBreaker: + """Simple circuit breaker per route/account.""" + + failure_threshold: int = 3 + slow_request_threshold: int = 3 + open_duration_seconds: int = 30 + half_open_max: int = 1 + states: Dict[str, CircuitState] = field(default_factory=dict) + failures: Dict[str, int] = field(default_factory=dict) + opened_at: Dict[str, datetime] = field(default_factory=dict) + half_open_count: Dict[str, int] = field(default_factory=dict) + + def state(self, route_id: str) -> CircuitState: + st = self.states.get(route_id, CircuitState.CLOSED) + if st == CircuitState.OPEN: + opened = self.opened_at.get(route_id) + if opened and datetime.now(timezone.utc) - opened > timedelta( + seconds=self.open_duration_seconds + ): + self.states[route_id] = CircuitState.HALF_OPEN + self.half_open_count[route_id] = 0 + return CircuitState.HALF_OPEN + return st + + def record_success(self, route_id: str) -> None: + self.states[route_id] = CircuitState.CLOSED + self.failures[route_id] = 0 + + def record_failure(self, route_id: str) -> None: + self.failures[route_id] = self.failures.get(route_id, 0) + 1 + if self.failures[route_id] >= self.failure_threshold: + self.states[route_id] = CircuitState.OPEN + self.opened_at[route_id] = datetime.now(timezone.utc) + + def can_try(self, route_id: str) -> bool: + st = self.state(route_id) + if st == CircuitState.CLOSED: + return True + if st == CircuitState.OPEN: + return False + return self.half_open_count.get(route_id, 0) < self.half_open_max diff --git a/uc_routing/failover/cooldown.py b/uc_routing/failover/cooldown.py new file mode 100644 index 0000000..8a90068 --- /dev/null +++ b/uc_routing/failover/cooldown.py @@ -0,0 +1,58 @@ +"""Cooldown / backoff timer management.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Dict, Optional + + +@dataclass +class CooldownRecord: + route_id: str + reason: str + cooldown_until: datetime + retry_after_seconds: Optional[int] = None + + +class CooldownManager: + """Track active cooldowns per route/account and compute exponential backoff.""" + + def __init__(self, max_cooldown_seconds: int = 3600) -> None: + self.cooldowns: Dict[str, CooldownRecord] = {} + self.consecutive_failures: Dict[str, int] = {} + self.max_cooldown_seconds = max_cooldown_seconds + + def set( + self, + route_id: str, + reason: str = "429", + retry_after: Optional[int] = None, + ) -> CooldownRecord: + failures = self.consecutive_failures.get(route_id, 0) + 1 + self.consecutive_failures[route_id] = failures + + if retry_after: + seconds = retry_after + else: + seconds = min(2 ** failures, self.max_cooldown_seconds) + + until = datetime.now(timezone.utc) + timedelta(seconds=seconds) + record = CooldownRecord( + route_id=route_id, + reason=reason, + cooldown_until=until, + retry_after_seconds=seconds, + ) + self.cooldowns[route_id] = record + return record + + def clear(self, route_id: str) -> None: + self.cooldowns.pop(route_id, None) + self.consecutive_failures.pop(route_id, None) + + def is_cooled(self, route_id: str) -> bool: + record = self.cooldowns.get(route_id) + if not record: + return False + return datetime.now(timezone.utc) < record.cooldown_until diff --git a/uc_routing/failover/health.py b/uc_routing/failover/health.py new file mode 100644 index 0000000..3383dbd --- /dev/null +++ b/uc_routing/failover/health.py @@ -0,0 +1,57 @@ +"""Health registry for routes and accounts.""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Dict, Optional + + +class HealthState(enum.Enum): + UNKNOWN = "unknown" + HEALTHY = "healthy" + DEGRADED = "degraded" + UNHEALTHY = "unhealthy" + + +@dataclass +class HealthRecord: + state: HealthState = HealthState.UNKNOWN + last_check: Optional[datetime] = None + last_success: Optional[datetime] = None + last_error: Optional[str] = None + consecutive_failures: int = 0 + + +class HealthRegistry: + """Track per-route health status.""" + + def __init__(self) -> None: + self.records: Dict[str, HealthRecord] = {} + + def record(self, route_id: str) -> HealthRecord: + return self.records.setdefault(route_id, HealthRecord()) + + def mark_healthy(self, route_id: str) -> None: + r = self.record(route_id) + r.state = HealthState.HEALTHY + r.last_check = datetime.now(timezone.utc) + r.last_success = r.last_check + r.consecutive_failures = 0 + + def mark_failure(self, route_id: str, error: Optional[str] = None) -> None: + r = self.record(route_id) + r.last_check = datetime.now(timezone.utc) + r.last_error = error + r.consecutive_failures += 1 + if r.consecutive_failures >= 3: + r.state = HealthState.UNHEALTHY + elif r.consecutive_failures >= 1: + r.state = HealthState.DEGRADED + + def is_unhealthy(self, route_id: str) -> bool: + r = self.records.get(route_id) + if not r: + return False + return r.state == HealthState.UNHEALTHY diff --git a/uc_routing/honcho/__init__.py b/uc_routing/honcho/__init__.py new file mode 100644 index 0000000..11abcd7 --- /dev/null +++ b/uc_routing/honcho/__init__.py @@ -0,0 +1,6 @@ +"""Honcho synchronization contract.""" + +from .contract import LedgerSnapshot, TelemetryBatch +from .sync import HonchoSyncClient + +__all__ = ["HonchoSyncClient", "LedgerSnapshot", "TelemetryBatch"] diff --git a/uc_routing/honcho/contract.py b/uc_routing/honcho/contract.py new file mode 100644 index 0000000..8f9b688 --- /dev/null +++ b/uc_routing/honcho/contract.py @@ -0,0 +1,29 @@ +"""Data contracts for Honcho synchronization.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + + +@dataclass +class LedgerSnapshot: + """Serializable ledger state for cross-device sync.""" + + user_id: str + sequence: int + timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + accounts: List[Dict[str, Any]] = field(default_factory=list) + cooldowns: List[Dict[str, Any]] = field(default_factory=list) + circuit_states: Dict[str, str] = field(default_factory=dict) + + +@dataclass +class TelemetryBatch: + """Batch of telemetry events ready to push to Honcho.""" + + user_id: str + request_id: str + sequence: int + events: List[Dict[str, Any]] = field(default_factory=list) diff --git a/uc_routing/honcho/sync.py b/uc_routing/honcho/sync.py new file mode 100644 index 0000000..37974f2 --- /dev/null +++ b/uc_routing/honcho/sync.py @@ -0,0 +1,43 @@ +"""Honcho sync client placeholder. + +Honcho endpoints and auth are configured under `routing_engine.honcho`. +This class is intentionally minimal; implementation will call the Honcho SDK or +REST API once the contract is finalized. +""" + +from __future__ import annotations + +from typing import Optional + +from .contract import LedgerSnapshot, TelemetryBatch + + +class HonchoSyncClient: + """Push/pull routing state to/from Honcho for cross-session consistency.""" + + def __init__( + self, + app_id: str, + user_id: str, + base_url: str, + api_key_ref: Optional[str] = None, + ) -> None: + self.app_id = app_id + self.user_id = user_id + self.base_url = base_url.rstrip("/") + self.api_key_ref = api_key_ref + + def push_snapshot(self, snapshot: LedgerSnapshot, request_id: str) -> bool: + """Push a ledger snapshot. Returns True on success.""" + # TODO: implement Honcho PUT/POST once API is known. + return False + + def push_telemetry(self, batch: TelemetryBatch) -> bool: + """Push a telemetry batch.""" + # TODO: implement Honcho telemetry ingest. + return False + + def fetch_latest(self) -> Optional[LedgerSnapshot]: + """Fetch the latest ledger snapshot for this user.""" + # TODO: implement Honcho GET. + return None diff --git a/uc_routing/ledger/__init__.py b/uc_routing/ledger/__init__.py new file mode 100644 index 0000000..3e8a1eb --- /dev/null +++ b/uc_routing/ledger/__init__.py @@ -0,0 +1,5 @@ +"""Account ledger for subscription, prepaid, and local compute entitlements.""" + +from .models import Account, AccountKind, Entitlement, EntitlementKind, Ledger + +__all__ = ["Account", "AccountKind", "Entitlement", "EntitlementKind", "Ledger"] diff --git a/uc_routing/ledger/balance.py b/uc_routing/ledger/balance.py new file mode 100644 index 0000000..7eb4f9e --- /dev/null +++ b/uc_routing/ledger/balance.py @@ -0,0 +1,47 @@ +"""Prepaid credit pool and local compute capacity accounting.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + +from .models import Account, Entitlement, EntitlementKind + + +@dataclass +class BalanceSheet: + """Tracks live balances for a prepaid or local account.""" + + account_id: str + input_tokens: int = 0 + output_tokens: int = 0 + requests: int = 0 + estimated_cost: float = 0.0 + + def spend( + self, + input_tokens: int, + output_tokens: int, + cost_per_input: float = 0.0, + cost_per_output: float = 0.0, + cost_per_request: float = 0.0, + ) -> float: + """Record a spend and return the estimated cost.""" + self.input_tokens += input_tokens + self.output_tokens += output_tokens + self.requests += 1 + cost = ( + input_tokens * cost_per_input + + output_tokens * cost_per_output + + cost_per_request + ) + self.estimated_cost += cost + return cost + + def remaining_balance(self, entitlement: Entitlement) -> float: + """Return remaining balance for a prepaid entitlement.""" + if entitlement.kind != EntitlementKind.PREPAID: + return 0.0 + if entitlement.minimum_balance is None: + return entitlement.limit - self.estimated_cost + return (entitlement.limit - entitlement.minimum_balance) - self.estimated_cost diff --git a/uc_routing/ledger/models.py b/uc_routing/ledger/models.py new file mode 100644 index 0000000..efe98e4 --- /dev/null +++ b/uc_routing/ledger/models.py @@ -0,0 +1,105 @@ +"""Core ledger domain models. + +All monetary and token values use float only where unavoidable; most counts are +integers. This module intentionally contains no persistence logic. +""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + + +class AccountKind(enum.Enum): + SUBSCRIPTION = "subscription" + PREPAID = "prepaid" + LOCAL = "local" + + +class EntitlementKind(enum.Enum): + FIXED_WINDOW = "fixed_window" + PREPAID = "prepaid" + LOCAL_COMPUTE = "local_compute" + + +@dataclass +class Entitlement: + """A unit of capacity attached to an account.""" + + kind: EntitlementKind + unit: str # e.g. "messages", "input_tokens", "output_tokens", "USD", "credits" + limit: float # for local compute, limit is max_concurrent slots + used: float = 0.0 + remaining: float = 0.0 + window_type: Optional[str] = None # sliding | calendar | session + window_size_seconds: Optional[int] = None + resets_at: Optional[datetime] = None + cooldown_until: Optional[datetime] = None + cost_per_input_token: Optional[float] = None + cost_per_output_token: Optional[float] = None + cost_per_request: Optional[float] = None + minimum_balance: Optional[float] = None + max_concurrent: Optional[int] = None + current_load: int = 0 + vram_total_mb: Optional[int] = None + vram_reserved_mb: Optional[int] = None + priority: Optional[str] = None # always_first | tier_based | fallback_only + metadata: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class SecretsRef: + """Non-secret references to where credentials live (env vars, keyring).""" + + auth: Optional[str] = None + api_key_ref: Optional[str] = None + oauth_token_ref: Optional[str] = None + + +@dataclass +class Account: + """A billable identity with one or more entitlements.""" + + account_id: str + provider: str + kind: AccountKind + display_name: str + secrets_ref: SecretsRef = field(default_factory=SecretsRef) + entitlements: List[Entitlement] = field(default_factory=list) + enabled: bool = True + metadata: Dict[str, Any] = field(default_factory=dict) + + def has_capacity_for(self, unit: str, amount: float = 1.0) -> bool: + """Return True if any entitlement can satisfy `amount` of `unit`.""" + now = datetime.now(timezone.utc) + for ent in self.entitlements: + if ent.cooldown_until and ent.cooldown_until > now: + continue + if ent.unit != unit: + continue + if ent.kind == EntitlementKind.FIXED_WINDOW: + if ent.resets_at and ent.resets_at <= now: + return True + return (ent.used + amount) <= ent.limit + if ent.kind in (EntitlementKind.PREPAID, EntitlementKind.LOCAL_COMPUTE): + return (ent.used + amount) <= ent.limit + return False + + +@dataclass +class Ledger: + """In-memory collection of accounts and their live state.""" + + accounts: Dict[str, Account] = field(default_factory=dict) + events: List[Dict[str, Any]] = field(default_factory=list) + + def account(self, account_id: str) -> Optional[Account]: + return self.accounts.get(account_id) + + def add_account(self, account: Account) -> None: + self.accounts[account.account_id] = account + + def record(self, event: Dict[str, Any]) -> None: + self.events.append(event) diff --git a/uc_routing/ledger/window.py b/uc_routing/ledger/window.py new file mode 100644 index 0000000..b203dda --- /dev/null +++ b/uc_routing/ledger/window.py @@ -0,0 +1,91 @@ +"""Fixed-window and sliding-window entitlement accounting.""" + +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Dict, List, Optional + + +@dataclass +class WindowUsageEvent: + """One reservation or spend against an entitlement window.""" + + timestamp: datetime + account_id: str + entitlement_id: str + route_id: str + unit: str + delta: float + + +class WindowTracker: + """Track recent usage for sliding/calendar windows. + + This is a placeholder implementation using an in-memory deque. A production + version should persist events and prune by the configured window. + """ + + def __init__(self) -> None: + self.events: deque = deque() + + def add(self, event: WindowUsageEvent) -> None: + self.events.append(event) + + def prune(self, now: Optional[datetime] = None) -> None: + now = now or datetime.now(timezone.utc) + cutoff = now - timedelta(hours=24 * 365) + while self.events and self.events[0].timestamp < cutoff: + self.events.popleft() + + def sum_for( + self, + account_id: str, + entitlement_id: str, + window_seconds: int, + now: Optional[datetime] = None, + ) -> float: + """Return total `delta` within the last `window_seconds`.""" + now = now or datetime.now(timezone.utc) + cutoff = now - timedelta(seconds=window_seconds) + total = 0.0 + for ev in self.events: + if ev.timestamp < cutoff: + continue + if ev.account_id == account_id and ev.entitlement_id == entitlement_id: + total += ev.delta + return total + + +@dataclass +class WindowedEntitlementLedger: + """Container for window trackers keyed by account + entitlement.""" + + trackers: Dict[str, WindowTracker] = field(default_factory=dict) + + def _key(self, account_id: str, entitlement_id: str) -> str: + return f"{account_id}::{entitlement_id}" + + def reserve( + self, + account_id: str, + entitlement_id: str, + route_id: str, + unit: str, + delta: float, + ) -> bool: + """Record a usage reservation and return success.""" + key = self._key(account_id, entitlement_id) + tracker = self.trackers.setdefault(key, WindowTracker()) + tracker.add( + WindowUsageEvent( + timestamp=datetime.now(timezone.utc), + account_id=account_id, + entitlement_id=entitlement_id, + route_id=route_id, + unit=unit, + delta=delta, + ) + ) + return True diff --git a/uc_routing/life_os/__init__.py b/uc_routing/life_os/__init__.py new file mode 100644 index 0000000..62379cc --- /dev/null +++ b/uc_routing/life_os/__init__.py @@ -0,0 +1,6 @@ +"""Life OS / terpOS metrics surface.""" + +from .api import MetricsHandler +from .metrics import RoutingMetrics + +__all__ = ["MetricsHandler", "RoutingMetrics"] diff --git a/uc_routing/life_os/api.py b/uc_routing/life_os/api.py new file mode 100644 index 0000000..563183a --- /dev/null +++ b/uc_routing/life_os/api.py @@ -0,0 +1,38 @@ +"""HTTP endpoint handlers for Life OS metrics. + +These are intended to be mounted into `proxy.py` under `/life-os/*` paths when +`UC_ROUTING_ENGINE=1` and `routing_engine.life_os.enabled` are true. +""" + +import json +from http.server import BaseHTTPRequestHandler +from typing import Any, Dict + +from .metrics import RoutingMetrics + + +class MetricsHandler: + """Serve routing metrics as JSON.""" + + def __init__(self, metrics: RoutingMetrics) -> None: + self.metrics = metrics + + def get_routing(self) -> Dict[str, Any]: + return self.metrics.to_dict() + + def get_providers(self) -> Dict[str, Any]: + return { + "provider_health": self.metrics.provider_health, + "cost_by_provider": self.metrics.cost_by_provider, + "requests_by_provider": self.metrics.requests_by_provider, + } + + def get_health(self) -> Dict[str, Any]: + return {"provider_health": self.metrics.provider_health} + + def get_cost(self) -> Dict[str, Any]: + return { + "window": self.metrics.window, + "total_estimated_cost": self.metrics.total_estimated_cost, + "cost_by_provider": self.metrics.cost_by_provider, + } diff --git a/uc_routing/life_os/metrics.py b/uc_routing/life_os/metrics.py new file mode 100644 index 0000000..36fa536 --- /dev/null +++ b/uc_routing/life_os/metrics.py @@ -0,0 +1,42 @@ +"""Aggregated metrics for the Life OS dashboard.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +@dataclass +class RoutingMetrics: + """Metrics payload consumed by terpOS / Life OS.""" + + window: str = "24h" + total_requests: int = 0 + requests_by_tier: Dict[str, int] = field(default_factory=dict) + requests_by_provider: Dict[str, int] = field(default_factory=dict) + total_estimated_cost: float = 0.0 + cost_by_provider: Dict[str, float] = field(default_factory=dict) + avg_ttft_ms: Optional[float] = None + p95_ttft_ms: Optional[float] = None + avg_e2e_latency_ms: Optional[float] = None + p95_e2e_latency_ms: Optional[float] = None + quota_remaining: Dict[str, float] = field(default_factory=dict) + provider_health: Dict[str, str] = field(default_factory=dict) + active_cooldowns: List[Dict[str, Any]] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + return { + "window": self.window, + "total_requests": self.total_requests, + "requests_by_tier": self.requests_by_tier, + "requests_by_provider": self.requests_by_provider, + "total_estimated_cost": self.total_estimated_cost, + "cost_by_provider": self.cost_by_provider, + "avg_ttft_ms": self.avg_ttft_ms, + "p95_ttft_ms": self.p95_ttft_ms, + "avg_e2e_latency_ms": self.avg_e2e_latency_ms, + "p95_e2e_latency_ms": self.p95_e2e_latency_ms, + "quota_remaining": self.quota_remaining, + "provider_health": self.provider_health, + "active_cooldowns": self.active_cooldowns, + } diff --git a/uc_routing/providers/__init__.py b/uc_routing/providers/__init__.py new file mode 100644 index 0000000..1ecb85b --- /dev/null +++ b/uc_routing/providers/__init__.py @@ -0,0 +1,7 @@ +"""Provider adapters and route registry.""" + +from .adapter import ProviderAdapter +from .registry import RouteRegistry +from .types import CapabilityProfile, ProviderType, Route + +__all__ = ["CapabilityProfile", "ProviderAdapter", "ProviderType", "Route", "RouteRegistry"] diff --git a/uc_routing/providers/adapter.py b/uc_routing/providers/adapter.py new file mode 100644 index 0000000..f16defe --- /dev/null +++ b/uc_routing/providers/adapter.py @@ -0,0 +1,30 @@ +"""Abstract provider adapter. + +Concrete adapters will wrap the existing providers/ helpers and translate +between Anthropic Messages API, OpenAI Chat Completions, and any special routes +like codex_oauth / cursor_agent. +""" + +from abc import ABC, abstractmethod +from typing import Any, Dict, Generator, Optional + +from ..telemetry.schema import TelemetryEvent + + +class ProviderAdapter(ABC): + """Issue a request to a provider and return a response stream/event.""" + + @abstractmethod + def send( + self, + route_id: str, + request_body: Dict[str, Any], + timeout: Optional[float] = None, + ) -> Generator[Dict[str, Any], None, None]: + """Yield streamed chunks and a final usage dict.""" + raise NotImplementedError + + @abstractmethod + def health_check(self, route_id: str) -> Dict[str, Any]: + """Return a lightweight health status dict.""" + raise NotImplementedError diff --git a/uc_routing/providers/registry.py b/uc_routing/providers/registry.py new file mode 100644 index 0000000..fd82bc4 --- /dev/null +++ b/uc_routing/providers/registry.py @@ -0,0 +1,58 @@ +"""Route registry built from config.json.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +from .types import CapabilityProfile, ProviderType, Route + + +@dataclass +class RouteRegistry: + """Index of configured routes.""" + + routes: Dict[str, Route] = field(default_factory=dict) + + def add(self, route: Route) -> None: + self.routes[route.route_id] = route + + def get(self, route_id: str) -> Optional[Route]: + return self.routes.get(route_id) + + @classmethod + def from_config(cls, config: Dict[str, Any]) -> RouteRegistry: + """Build a registry from the `routing_engine` config section. + + This is a scaffold; the full loader lives in `uc_routing.config.loader`. + """ + registry = cls() + engine_cfg = config.get("routing_engine", {}) + accounts = {a["account_id"]: a for a in engine_cfg.get("accounts", [])} + + for route_id, route_cfg in config.get("routes", {}).items(): + account_id = route_cfg.get("account_id") or route_id + account = accounts.get(account_id, {}) + route_type = route_cfg.get("type") + if route_type is None: + route_type = "anthropic" # config convention: missing type == real Claude + provider_type = ProviderType(route_type) + registry.add( + Route( + route_id=route_id, + provider_type=provider_type, + account_id=account_id, + model=route_cfg.get("model", route_id), + upstream=route_cfg.get("upstream"), + auth_ref=route_cfg.get("auth"), + headers=route_cfg.get("headers", {}), + body_overrides=route_cfg.get("body", {}), + capability=CapabilityProfile( + context_window=route_cfg.get("context_window", 128000), + supports_vision=route_cfg.get("supports_vision", False), + supports_tools=route_cfg.get("supports_tools", True), + ), + enabled=route_cfg.get("enabled", True), + ) + ) + return registry diff --git a/uc_routing/providers/types.py b/uc_routing/providers/types.py new file mode 100644 index 0000000..1c841d2 --- /dev/null +++ b/uc_routing/providers/types.py @@ -0,0 +1,47 @@ +"""Provider and route domain types.""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass, field +from typing import Any, Dict, Optional + + +class ProviderType(enum.Enum): + ANTHROPIC = "anthropic" + OPENAI_COMPAT = "openai_compat" + CODEX_OAUTH = "codex_oauth" + CURSOR_AGENT = "cursor_agent" + LOCAL_OPENAI = "local_openai" + + +@dataclass +class CapabilityProfile: + """Structured capability metadata for a route/model.""" + + context_window: int = 8192 + supports_vision: bool = False + supports_tools: bool = True + supports_reasoning_split: bool = False + scores: Dict[str, float] = field(default_factory=dict) + cost_class: str = "mid" # free | cheap | mid | premium + latency_class: str = "normal" # fast | normal | slow + quality_class: str = "good" # local | good | great | frontier + tags: list = field(default_factory=list) + + +@dataclass +class Route: + """A concrete provider + account + model binding.""" + + route_id: str + provider_type: ProviderType + account_id: str + model: str + upstream: Optional[str] = None + auth_ref: Optional[str] = None + headers: Dict[str, str] = field(default_factory=dict) + body_overrides: Dict[str, Any] = field(default_factory=dict) + capability: Optional[CapabilityProfile] = None + enabled: bool = True + max_output_tokens: Optional[int] = None diff --git a/uc_routing/routing/__init__.py b/uc_routing/routing/__init__.py new file mode 100644 index 0000000..ec829a7 --- /dev/null +++ b/uc_routing/routing/__init__.py @@ -0,0 +1,8 @@ +"""Core routing engine: task tier detection, decision, selection, failover.""" + +from .decision import RoutingDecision +from .engine import RoutingEngine +from .policy import SelectionPolicy +from .task_tiers import TaskTier, TaskTierDetector + +__all__ = ["RoutingDecision", "RoutingEngine", "SelectionPolicy", "TaskTier", "TaskTierDetector"] diff --git a/uc_routing/routing/decision.py b/uc_routing/routing/decision.py new file mode 100644 index 0000000..623432a --- /dev/null +++ b/uc_routing/routing/decision.py @@ -0,0 +1,21 @@ +"""Routing decision record.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +@dataclass +class RoutingDecision: + """The outcome of a routing decision.""" + + request_id: str + route_id: Optional[str] + tier: str + estimated_cost: float = 0.0 + fallback_chain: List[str] = field(default_factory=list) + scores: Dict[str, float] = field(default_factory=dict) + outcome: str = "selected" # selected | no_eligible_route | reservation_failed + reason: str = "" + metadata: Dict[str, Any] = field(default_factory=dict) diff --git a/uc_routing/routing/engine.py b/uc_routing/routing/engine.py new file mode 100644 index 0000000..530052d --- /dev/null +++ b/uc_routing/routing/engine.py @@ -0,0 +1,102 @@ +"""Routing engine: the main select_route entry point.""" + +from __future__ import annotations + +import uuid +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +from ..failover.health import HealthRegistry +from ..ledger.models import Ledger +from ..providers.registry import RouteRegistry +from ..telemetry.collector import TelemetryCollector +from .decision import RoutingDecision +from .policy import SelectionPolicy +from .task_tiers import TaskTierDetector + + +@dataclass +class RoutingEngine: + """Selects a route for each request based on ledger, capability, and health.""" + + ledger: Ledger + registry: RouteRegistry + health: HealthRegistry + telemetry: TelemetryCollector + detector: TaskTierDetector = field(default_factory=TaskTierDetector) + policy: SelectionPolicy = field( + default_factory=lambda: SelectionPolicy( + tier_thresholds={ + "planning": 0.80, + "heavy_reasoning": 0.90, + "bulk_context": 0.60, + "frontend": 0.70, + } + ) + ) + max_failover_depth: int = 3 + + def select_route( + self, + prompt: str, + prompt_tokens: int = 0, + max_tokens: int = 4096, + has_images: bool = False, + has_tools: bool = False, + explicit_tier: Optional[str] = None, + ) -> RoutingDecision: + """Return a routing decision for a single request. + + This is a scaffold: the cost/health/failover logic is wired but the + provider dispatch path is intentionally not implemented. + """ + request_id = str(uuid.uuid4()) + tier = self.detector.detect( + prompt, has_images, 0, prompt_tokens, explicit_tier + ).value + + candidates: List[Any] = [] + for route in self.registry.routes.values(): + account = self.ledger.account(route.account_id) + if account is None: + continue + if not self.policy.is_eligible( + route, account, tier, prompt_tokens, max_tokens, has_images, has_tools + ): + continue + if self.health.is_unhealthy(route.route_id): + continue + candidates.append( + ( + route, + account, + self.policy.sort_key( + route, account, tier, prompt_tokens, max_tokens + ), + ) + ) + + if not candidates: + return RoutingDecision( + request_id=request_id, + route_id=None, + tier=tier, + outcome="no_eligible_route", + reason="No route satisfied capability, quota, and health constraints.", + ) + + candidates.sort(key=lambda x: x[2]) + fallback_chain = [c[0].route_id for c in candidates[: self.max_failover_depth]] + chosen = candidates[0] + + return RoutingDecision( + request_id=request_id, + route_id=chosen[0].route_id, + tier=tier, + estimated_cost=self.policy.effective_cost( + chosen[0], chosen[1], prompt_tokens, max_tokens + ), + fallback_chain=fallback_chain, + scores=chosen[0].capability.scores if chosen[0].capability else {}, + reason=f"local={chosen[2][0]==0}; score={-chosen[2][1]:.2f}; cost={chosen[2][2]:.6f}", + ) diff --git a/uc_routing/routing/policy.py b/uc_routing/routing/policy.py new file mode 100644 index 0000000..3360bdf --- /dev/null +++ b/uc_routing/routing/policy.py @@ -0,0 +1,104 @@ +"""Selection policy, cost function, and failover ordering.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, Tuple + +from ..ledger.models import Account, EntitlementKind, Ledger +from ..providers.types import CapabilityProfile, Route +from ..telemetry.cost import CostNormalizer +from .decision import RoutingDecision + + +@dataclass +class SelectionPolicy: + """Encapsulates the rules for choosing and ordering candidate routes.""" + + tier_thresholds: Dict[str, float] + cost_normalizer: CostNormalizer = CostNormalizer() + + def is_eligible( + self, + route: Route, + account: Account, + tier: str, + prompt_tokens: int, + max_tokens: int, + has_images: bool, + has_tools: bool, + ) -> bool: + """Return True if route satisfies capability and quota constraints.""" + if not route.enabled or not account.enabled: + return False + profile = route.capability + if profile is None: + return False + score = profile.scores.get(tier, 0.0) + if score < self.tier_thresholds.get(tier, 0.7): + return False + if prompt_tokens + max_tokens > profile.context_window: + return False + if has_images and not profile.supports_vision: + return False + if has_tools and not profile.supports_tools: + return False + return account.has_capacity_for("requests") + + def effective_cost( + self, + route: Route, + account: Account, + prompt_tokens: int, + completion_tokens: int, + ) -> float: + """Compute a comparable effective cost for sorting. + + Local accounts are zero monetary cost. Subscriptions carry an opportunity + cost as their window fills up. Prepaid accounts are direct cost. + """ + entitlement = self._primary_entitlement(account) + cost_per_input = 0.0 + cost_per_output = 0.0 + cost_per_request = 0.0 + pressure = 0.0 + + if entitlement: + cost_per_input = entitlement.cost_per_input_token or 0.0 + cost_per_output = entitlement.cost_per_output_token or 0.0 + cost_per_request = entitlement.cost_per_request or 0.0 + if entitlement.limit > 0: + pressure = entitlement.used / entitlement.limit + + estimate = self.cost_normalizer.estimate( + prompt_tokens, completion_tokens, cost_per_input, cost_per_output, cost_per_request + ) + if self._is_local(account): + return estimate.total + (pressure * 0.001) # tiny load penalty + return estimate.total * (1.0 + pressure) + + def _primary_entitlement(self, account: Account) -> Optional[Any]: + for ent in account.entitlements: + return ent + return None + + def _is_local(self, account: Account) -> bool: + return account.kind.value == "local" + + def sort_key( + self, + route: Route, + account: Account, + tier: str, + prompt_tokens: int, + completion_tokens: int, + ) -> Tuple[int, float, float]: + """Return a tuple for sorting candidates. + + Order: local first (0), then non-local (1); then capability score desc + (negated), then effective cost asc. + """ + score = route.capability.scores.get(tier, 0.0) if route.capability else 0.0 + cost = self.effective_cost(route, account, prompt_tokens, completion_tokens) + local_order = 0 if self._is_local(account) else 1 + return (local_order, -score, cost) diff --git a/uc_routing/routing/task_tiers.py b/uc_routing/routing/task_tiers.py new file mode 100644 index 0000000..f78fadf --- /dev/null +++ b/uc_routing/routing/task_tiers.py @@ -0,0 +1,59 @@ +"""Task tier classification.""" + +from __future__ import annotations + +import enum +import re +from dataclasses import dataclass +from typing import Any, Dict, List, Optional + + +class TaskTier(enum.Enum): + PLANNING = "planning" + HEAVY_REASONING = "heavy_reasoning" + BULK_CONTEXT = "bulk_context" + FRONTEND = "frontend" + + +@dataclass +class TaskTierDetector: + """Detect task tier from request content and constraints. + + This placeholder uses heuristics. A production implementation may call a small + classifier model and/or honor explicit routing directives. + """ + + default_tier: TaskTier = TaskTier.HEAVY_REASONING + + def detect( + self, + prompt: str, + has_images: bool = False, + tool_count: int = 0, + estimated_tokens: int = 0, + explicit_tier: Optional[str] = None, + ) -> TaskTier: + if explicit_tier: + try: + return TaskTier(explicit_tier) + except ValueError: + pass + + text = prompt.lower() + + if has_images: + return TaskTier.FRONTEND + + bulk_keywords = ["summarize", "find all", "search", "bulk", "every", "all usages"] + if any(k in text for k in bulk_keywords) or estimated_tokens > 16000: + return TaskTier.BULK_CONTEXT + + planning_keywords = ["design", "plan", "architecture", "roadmap", "strategy"] + if any(k in text for k in planning_keywords): + return TaskTier.PLANNING + + heavy_keywords = ["debug", "reason", "prove", "concurrency", "algorithm"] + if any(k in text for k in heavy_keywords) or tool_count > 5: + return TaskTier.HEAVY_REASONING + + return self.default_tier diff --git a/uc_routing/telemetry/__init__.py b/uc_routing/telemetry/__init__.py new file mode 100644 index 0000000..8c7477b --- /dev/null +++ b/uc_routing/telemetry/__init__.py @@ -0,0 +1,7 @@ +"""Telemetry collection for latency, token spend, quotas, and rate limits.""" + +from .collector import TelemetryCollector +from .cost import CostNormalizer +from .schema import RateLimitSnapshot, TelemetryEvent + +__all__ = ["CostNormalizer", "RateLimitSnapshot", "TelemetryCollector", "TelemetryEvent"] diff --git a/uc_routing/telemetry/collector.py b/uc_routing/telemetry/collector.py new file mode 100644 index 0000000..0cf0b4f --- /dev/null +++ b/uc_routing/telemetry/collector.py @@ -0,0 +1,52 @@ +"""In-memory telemetry collector and simple histograms.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, List, Optional + +from .schema import TelemetryEvent + + +@dataclass +class LatencyHistogram: + """Simple latency statistics (placeholder for p50/p95).""" + + values: List[float] = field(default_factory=list) + + def add(self, value: float) -> None: + self.values.append(value) + + def percentile(self, p: float) -> Optional[float]: + if not self.values: + return None + sorted_vals = sorted(self.values) + idx = int(round(p / 100.0 * (len(sorted_vals) - 1))) + return sorted_vals[idx] + + +class TelemetryCollector: + """Collect events and compute per-route latency summaries.""" + + def __init__(self) -> None: + self.events: List[TelemetryEvent] = [] + self.latency_by_route: Dict[str, LatencyHistogram] = {} + + def record(self, event: TelemetryEvent) -> None: + self.events.append(event) + hist = self.latency_by_route.setdefault(event.route_id, LatencyHistogram()) + if event.e2e_latency_ms is not None: + hist.add(event.e2e_latency_ms) + + def recent_events(self, count: int = 100) -> List[TelemetryEvent]: + return self.events[-count:] + + def p95_latency_ms(self, route_id: str) -> Optional[float]: + hist = self.latency_by_route.get(route_id) + return hist.percentile(95) if hist else None + + def flush(self) -> List[TelemetryEvent]: + """Return and clear in-memory events (push to persistence/Honcho).""" + events = self.events + self.events = [] + return events diff --git a/uc_routing/telemetry/cost.py b/uc_routing/telemetry/cost.py new file mode 100644 index 0000000..a295eae --- /dev/null +++ b/uc_routing/telemetry/cost.py @@ -0,0 +1,43 @@ +"""Cost estimation and normalization across providers.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + + +@dataclass +class CostEstimate: + input_cost: float + output_cost: float + request_cost: float + total: float + + +class CostNormalizer: + """Normalize provider-specific cost into a comparable unit. + + For v1 all costs are USD-equivalent. Future versions may add credit-type + conversion tables (ACU, OR credits, Devin session credits). + """ + + def estimate( + self, + prompt_tokens: int, + completion_tokens: int, + cost_per_input: float = 0.0, + cost_per_output: float = 0.0, + cost_per_request: float = 0.0, + ) -> CostEstimate: + input_cost = prompt_tokens * cost_per_input + output_cost = completion_tokens * cost_per_output + return CostEstimate( + input_cost=input_cost, + output_cost=output_cost, + request_cost=cost_per_request, + total=input_cost + output_cost + cost_per_request, + ) + + def compare(self, a: CostEstimate, b: CostEstimate) -> float: + """Return a - b; negative means a is cheaper.""" + return a.total - b.total diff --git a/uc_routing/telemetry/schema.py b/uc_routing/telemetry/schema.py new file mode 100644 index 0000000..df33283 --- /dev/null +++ b/uc_routing/telemetry/schema.py @@ -0,0 +1,62 @@ +"""Telemetry event and rate-limit schemas.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + + +@dataclass +class RateLimitSnapshot: + """Normalized rate-limit state from a provider response.""" + + requests_remaining: Optional[int] = None + requests_reset: Optional[datetime] = None + tokens_remaining: Optional[float] = None + tokens_reset: Optional[datetime] = None + retry_after_seconds: Optional[int] = None + raw_headers: Dict[str, str] = field(default_factory=dict) + + +@dataclass +class TelemetryEvent: + """One routing/provider request plus its outcome.""" + + event_id: str + request_id: str + route_id: str + account_id: str + provider: str + model_id: str + tier: str + timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + ttft_ms: Optional[float] = None + e2e_latency_ms: Optional[float] = None + prompt_tokens: int = 0 + completion_tokens: int = 0 + total_tokens: int = 0 + estimated_cost: float = 0.0 + actual_cost: Optional[float] = None + remaining_quota: Optional[float] = None + rate_limit: RateLimitSnapshot = field(default_factory=RateLimitSnapshot) + outcome: str = "unknown" # success | failure | cached | cancelled + error_code: Optional[str] = None + error_message: Optional[str] = None + cached: bool = False + fallback_index: int = 0 + request_size_bytes: int = 0 + response_size_bytes: int = 0 + metadata: Dict[str, Any] = field(default_factory=dict) + + def token_cost_estimate( + self, + cost_per_input: float = 0.0, + cost_per_output: float = 0.0, + cost_per_request: float = 0.0, + ) -> float: + return ( + self.prompt_tokens * cost_per_input + + self.completion_tokens * cost_per_output + + cost_per_request + ) diff --git a/uc_routing/tests/__init__.py b/uc_routing/tests/__init__.py new file mode 100644 index 0000000..f30854b --- /dev/null +++ b/uc_routing/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for the routing engine scaffold.""" diff --git a/uc_routing/tests/test_failover.py b/uc_routing/tests/test_failover.py new file mode 100644 index 0000000..a406614 --- /dev/null +++ b/uc_routing/tests/test_failover.py @@ -0,0 +1,36 @@ +"""Unit tests for failover primitives.""" + +import unittest + +from uc_routing.failover.circuit import CircuitBreaker, CircuitState +from uc_routing.failover.cooldown import CooldownManager + + +class TestCircuitBreaker(unittest.TestCase): + def test_opens_after_failures(self): + cb = CircuitBreaker(failure_threshold=3) + self.assertEqual(cb.state("r1"), CircuitState.CLOSED) + cb.record_failure("r1") + cb.record_failure("r1") + self.assertEqual(cb.state("r1"), CircuitState.CLOSED) + cb.record_failure("r1") + self.assertEqual(cb.state("r1"), CircuitState.OPEN) + self.assertFalse(cb.can_try("r1")) + + def test_success_closes(self): + cb = CircuitBreaker(failure_threshold=3) + cb.record_failure("r1") + cb.record_success("r1") + self.assertEqual(cb.state("r1"), CircuitState.CLOSED) + + +class TestCooldown(unittest.TestCase): + def test_cooldown_active(self): + cm = CooldownManager() + cm.set("r1", "429", retry_after=60) + self.assertTrue(cm.is_cooled("r1")) + self.assertFalse(cm.is_cooled("r2")) + + +if __name__ == "__main__": + unittest.main() diff --git a/uc_routing/tests/test_ledger.py b/uc_routing/tests/test_ledger.py new file mode 100644 index 0000000..7322f3d --- /dev/null +++ b/uc_routing/tests/test_ledger.py @@ -0,0 +1,43 @@ +"""Unit tests for the ledger.""" + +import unittest + +from uc_routing.ledger.models import Account, AccountKind, Entitlement, EntitlementKind + + +class TestLedger(unittest.TestCase): + def test_account_has_capacity_for_window(self): + ent = Entitlement( + kind=EntitlementKind.FIXED_WINDOW, + unit="messages", + limit=5, + used=0, + ) + account = Account( + account_id="test-anthropic", + provider="anthropic", + kind=AccountKind.SUBSCRIPTION, + display_name="Test", + entitlements=[ent], + ) + self.assertTrue(account.has_capacity_for("messages", 1)) + + def test_account_exhausted_window(self): + ent = Entitlement( + kind=EntitlementKind.FIXED_WINDOW, + unit="messages", + limit=5, + used=5, + ) + account = Account( + account_id="test-anthropic", + provider="anthropic", + kind=AccountKind.SUBSCRIPTION, + display_name="Test", + entitlements=[ent], + ) + self.assertFalse(account.has_capacity_for("messages", 1)) + + +if __name__ == "__main__": + unittest.main() diff --git a/uc_routing/tests/test_routing.py b/uc_routing/tests/test_routing.py new file mode 100644 index 0000000..838e6f8 --- /dev/null +++ b/uc_routing/tests/test_routing.py @@ -0,0 +1,84 @@ +"""Unit tests for the routing engine.""" + +import unittest + +from uc_routing.ledger.models import Account, AccountKind, Entitlement, EntitlementKind, Ledger +from uc_routing.providers.types import CapabilityProfile, ProviderType, Route +from uc_routing.providers.registry import RouteRegistry +from uc_routing.failover.health import HealthRegistry +from uc_routing.routing.engine import RoutingEngine +from uc_routing.telemetry.collector import TelemetryCollector + + +class TestRoutingEngine(unittest.TestCase): + def test_select_route_prefers_local(self): + local_route = Route( + route_id="local-rtx5090", + provider_type=ProviderType.LOCAL_OPENAI, + account_id="local-rtx5090", + model="llama-3.3-70b", + capability=CapabilityProfile( + context_window=128000, + scores={"heavy_reasoning": 0.65, "bulk_context": 0.85}, + ), + ) + cloud_route = Route( + route_id="claude-opus", + provider_type=ProviderType.ANTHROPIC, + account_id="anthropic-pro", + model="claude-opus-4-8", + capability=CapabilityProfile( + context_window=200000, + scores={"heavy_reasoning": 0.98, "bulk_context": 0.85}, + ), + ) + registry = RouteRegistry() + registry.add(local_route) + registry.add(cloud_route) + + ledger = Ledger() + ledger.add_account( + Account( + account_id="local-rtx5090", + provider="ollama", + kind=AccountKind.LOCAL, + display_name="Local RTX 5090", + entitlements=[ + Entitlement( + kind=EntitlementKind.LOCAL_COMPUTE, + unit="requests", + limit=2, + used=0, + ) + ], + ) + ) + ledger.add_account( + Account( + account_id="anthropic-pro", + provider="anthropic", + kind=AccountKind.SUBSCRIPTION, + display_name="Anthropic Pro", + entitlements=[ + Entitlement( + kind=EntitlementKind.FIXED_WINDOW, + unit="requests", + limit=100, + used=0, + ) + ], + ) + ) + + engine = RoutingEngine( + ledger=ledger, + registry=registry, + health=HealthRegistry(), + telemetry=TelemetryCollector(), + ) + decision = engine.select_route("summarize all recent issues", prompt_tokens=500) + self.assertEqual(decision.route_id, "local-rtx5090") + + +if __name__ == "__main__": + unittest.main() diff --git a/uc_routing/tests/test_telemetry.py b/uc_routing/tests/test_telemetry.py new file mode 100644 index 0000000..ff0e793 --- /dev/null +++ b/uc_routing/tests/test_telemetry.py @@ -0,0 +1,40 @@ +"""Unit tests for telemetry schema and cost estimation.""" + +import unittest + +from uc_routing.telemetry.cost import CostNormalizer +from uc_routing.telemetry.schema import RateLimitSnapshot, TelemetryEvent + + +class TestTelemetry(unittest.TestCase): + def test_cost_normalizer(self): + cn = CostNormalizer() + estimate = cn.estimate( + prompt_tokens=1000, + completion_tokens=500, + cost_per_input=1e-6, + cost_per_output=2e-6, + cost_per_request=0.005, + ) + self.assertAlmostEqual(estimate.input_cost, 0.001) + self.assertAlmostEqual(estimate.output_cost, 0.001) + self.assertAlmostEqual(estimate.total, 0.007) + + def test_telemetry_event(self): + event = TelemetryEvent( + event_id="e1", + request_id="r1", + route_id="local-rtx5090", + account_id="local-rtx5090", + provider="ollama", + model_id="llama-3.3-70b", + tier="bulk_context", + prompt_tokens=1000, + completion_tokens=500, + ) + self.assertEqual(event.outcome, "unknown") + self.assertIsInstance(event.rate_limit, RateLimitSnapshot) + + +if __name__ == "__main__": + unittest.main() From e5c37ddc0dba4b5a5262a434456fc29fee421f7c Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 22 Aug 2026 21:30:18 +0000 Subject: [PATCH 2/2] fix(routing): address review feedback on registry, capacity, and circuit breaker Co-Authored-By: Rob --- uc_routing/failover/circuit.py | 8 ++++++- uc_routing/ledger/models.py | 37 ++++++++++++++++++++++++++++++++ uc_routing/providers/registry.py | 5 +++++ uc_routing/routing/policy.py | 2 +- 4 files changed, 50 insertions(+), 2 deletions(-) diff --git a/uc_routing/failover/circuit.py b/uc_routing/failover/circuit.py index 224f344..8cbac79 100644 --- a/uc_routing/failover/circuit.py +++ b/uc_routing/failover/circuit.py @@ -42,12 +42,14 @@ def state(self, route_id: str) -> CircuitState: def record_success(self, route_id: str) -> None: self.states[route_id] = CircuitState.CLOSED self.failures[route_id] = 0 + self.half_open_count[route_id] = 0 def record_failure(self, route_id: str) -> None: self.failures[route_id] = self.failures.get(route_id, 0) + 1 if self.failures[route_id] >= self.failure_threshold: self.states[route_id] = CircuitState.OPEN self.opened_at[route_id] = datetime.now(timezone.utc) + self.half_open_count[route_id] = 0 def can_try(self, route_id: str) -> bool: st = self.state(route_id) @@ -55,4 +57,8 @@ def can_try(self, route_id: str) -> bool: return True if st == CircuitState.OPEN: return False - return self.half_open_count.get(route_id, 0) < self.half_open_max + count = self.half_open_count.get(route_id, 0) + if count < self.half_open_max: + self.half_open_count[route_id] = count + 1 + return True + return False diff --git a/uc_routing/ledger/models.py b/uc_routing/ledger/models.py index efe98e4..fa611e2 100644 --- a/uc_routing/ledger/models.py +++ b/uc_routing/ledger/models.py @@ -87,6 +87,43 @@ def has_capacity_for(self, unit: str, amount: float = 1.0) -> bool: return (ent.used + amount) <= ent.limit return False + def can_fulfill_request( + self, prompt_tokens: int = 0, completion_tokens: int = 0, amount: float = 1.0 + ) -> bool: + """Return True if any entitlement can satisfy a request of this size. + + This is the high-level capacity check used by the routing policy. It + derives the relevant unit from each entitlement's kind and unit rather + than requiring callers to know whether a subscription counts messages, + a prepaid pool counts USD/credits, or a local GPU counts concurrent slots. + """ + now = datetime.now(timezone.utc) + token_need = float(prompt_tokens + completion_tokens) + for ent in self.entitlements: + if ent.cooldown_until and ent.cooldown_until > now: + continue + if ent.resets_at and ent.resets_at <= now: + return True + # Generic request quota used by simple/local placeholders. + if ent.unit == "requests": + if (ent.used + amount) <= ent.limit: + return True + continue + if ent.kind == EntitlementKind.FIXED_WINDOW: + if ent.unit == "messages" and (ent.used + amount) <= ent.limit: + return True + if ent.unit in ("input_tokens", "output_tokens", "tokens") and ( + ent.used + token_need + ) <= ent.limit: + return True + elif ent.kind == EntitlementKind.PREPAID: + if ent.unit in ("USD", "credits") and ent.remaining > 0: + return True + elif ent.kind == EntitlementKind.LOCAL_COMPUTE: + if (ent.current_load + amount) <= ent.limit: + return True + return False + @dataclass class Ledger: diff --git a/uc_routing/providers/registry.py b/uc_routing/providers/registry.py index fd82bc4..13fb77c 100644 --- a/uc_routing/providers/registry.py +++ b/uc_routing/providers/registry.py @@ -36,6 +36,11 @@ def from_config(cls, config: Dict[str, Any]) -> RouteRegistry: route_type = route_cfg.get("type") if route_type is None: route_type = "anthropic" # config convention: missing type == real Claude + if route_type == "auto": + continue # the Auto Router pseudo-backend is not a real provider route + valid_provider_types = {m.value for m in ProviderType} + if route_type not in valid_provider_types: + continue # skip unknown / future route types until explicitly supported provider_type = ProviderType(route_type) registry.add( Route( diff --git a/uc_routing/routing/policy.py b/uc_routing/routing/policy.py index 3360bdf..85347d7 100644 --- a/uc_routing/routing/policy.py +++ b/uc_routing/routing/policy.py @@ -43,7 +43,7 @@ def is_eligible( return False if has_tools and not profile.supports_tools: return False - return account.has_capacity_for("requests") + return account.can_fulfill_request(prompt_tokens, max_tokens) def effective_cost( self,