docs: scaffold subscription- and quota-aware routing engine spec

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
 

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.

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",
+]

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",
+]

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())

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)

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"]

uc_routing/failover/circuit.py

@@ -0,0 +1,64 @@
+"""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
+        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)
+        if st == CircuitState.CLOSED:
+            return True
+        if st == CircuitState.OPEN:
+            return False
+        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

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

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

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"]