ADR-240 D1+D2: BackendRaceChain + CostAwareRouter (dual-provider autonomy) v0.78.0

graqle/__version__.py

@@ -5,4 +5,4 @@
 # constraints: none
 # ── /graqle:intelligence ──
 
-__version__ = "0.77.0"
+__version__ = "0.78.0"

graqle/backends/__init__.py

@@ -22,6 +22,12 @@ def __getattr__(name: str):
     if name == "GeminiBackend":
         from graqle.backends.gemini import GeminiBackend
         return GeminiBackend
+    if name == "BackendFallbackChain":
+        from graqle.backends.fallback import BackendFallbackChain
+        return BackendFallbackChain
+    if name == "BackendRaceChain":  # ADR-240 D1
+        from graqle.backends.race import BackendRaceChain
+        return BackendRaceChain
     if name in ("create_provider_backend", "PROVIDER_PRESETS",
                 "get_provider_names"):
         from graqle.backends import providers

graqle/backends/race.py

@@ -0,0 +1,143 @@
+"""BackendRaceChain — race multiple backends, take the first (or best-of-N).
+
+ADR-240 D1: the SDK had FALLBACK (sequential) and PARALLEL fan-out (BackendPool,
+wait-for-ALL) but no RACE / first-to-finish / best-of-N primitive. This adds one.
+
+Mirrors BackendFallbackChain exactly (subclasses BaseBackend, takes
+list[BaseBackend], empty→ValueError, all-fail→RuntimeError) so it is a drop-in
+ModelBackend anywhere a backend is expected — the agent behind AutonomousExecutor,
+a TaskRouter target, a BackendPool panelist, or nested in a BackendFallbackChain.
+
+Two modes:
+  - scorer=None (default): FIRST-TO-FINISH — return the first successful result,
+    cancel the losers (stops paying for slower providers).
+  - scorer set:           BEST-OF-N — await all, return the highest-scored result
+    (deterministic tie-break: score, then original backend order).
+
+Uses ONLY BaseBackend.generate — no per-provider logic. Generic asyncio hedge
+pattern; contains none of the patented debate/clearance machinery.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Callable
+
+from graqle.backends.base import BaseBackend, GenerateResult
+
+logger = logging.getLogger("graqle.backends.race")
+
+
+class BackendRaceChain(BaseBackend):
+    """Race backends concurrently; return the first success or the best-of-N.
+
+    Usage:
+        # first-to-finish (hedged request) — take whichever provider answers first
+        race = BackendRaceChain([AnthropicBackend(...), OpenAIBackend(...)])
+        result = await race.generate("prompt")
+
+        # best-of-N — run both, keep the higher-scored answer
+        race = BackendRaceChain([a, b], scorer=lambda r: len(r.text))
+        result = await race.generate("prompt")
+    """
+
+    def __init__(
+        self,
+        backends: list[BaseBackend],
+        *,
+        timeout_s: float | None = None,
+        scorer: Callable[[GenerateResult], float] | None = None,
+    ) -> None:
+        if not backends:
+            raise ValueError("BackendRaceChain requires at least one backend")
+        self._backends = backends
+        self._timeout_s = timeout_s
+        self._scorer = scorer
+        self._last_used: str = ""
+        self._last_cost_per_1k: float = backends[0].cost_per_1k_tokens
+
+    async def generate(
+        self,
+        prompt: str,
+        *,
+        max_tokens: int = 512,
+        temperature: float = 0.3,
+        stop: list[str] | None = None,
+    ) -> GenerateResult:
+        async def _run(b: BaseBackend) -> tuple[str, float, GenerateResult]:
+            coro = b.generate(
+                prompt, max_tokens=max_tokens, temperature=temperature, stop=stop
+            )
+            if self._timeout_s is not None:
+                coro = asyncio.wait_for(coro, timeout=self._timeout_s)
+            return b.name, b.cost_per_1k_tokens, await coro
+
+        tasks = {asyncio.ensure_future(_run(b)) for b in self._backends}
+
+        if self._scorer is None:
+            # ── FIRST-TO-FINISH ─────────────────────────────────────────────
+            errors: list[BaseException] = []
+            pending = tasks
+            while pending:
+                done, pending = await asyncio.wait(
+                    pending, return_when=asyncio.FIRST_COMPLETED
+                )
+                for t in done:
+                    exc = t.exception()
+                    if exc is None:
+                        name, cpk, result = t.result()
+                        self._last_used, self._last_cost_per_1k = name, cpk
+                        await self._cancel(pending)  # stop the losers
+                        if errors:
+                            logger.info(
+                                "Race won by %s after %d failure(s)", name, len(errors)
+                            )
+                        return result
+                    errors.append(exc)  # this one failed — keep waiting for a winner
+            self._raise_all_failed(errors)
+
+        # ── BEST-OF-N ───────────────────────────────────────────────────────
+        done, _ = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
+        scored: list[tuple[float, str, float, GenerateResult]] = []
+        errors = []
+        for t in done:
+            exc = t.exception()
+            if exc is None:
+                name, cpk, result = t.result()
+                scored.append((self._scorer(result), name, cpk, result))
+            else:
+                errors.append(exc)
+        if not scored:
+            self._raise_all_failed(errors)
+        # deterministic tie-break: highest score, then earliest in the original list
+        order = {b.name: i for i, b in enumerate(self._backends)}
+        best = max(scored, key=lambda s: (s[0], -order.get(s[1], 0)))
+        _, self._last_used, self._last_cost_per_1k, result = best
+        return result
+
+    async def _cancel(self, pending: set) -> None:
+        for t in pending:
+            t.cancel()
+        if pending:
+            # drain cancellations so no "Task was destroyed but pending" warnings
+            await asyncio.gather(*pending, return_exceptions=True)
+
+    def _raise_all_failed(self, errors: list[BaseException]) -> None:
+        summary = "; ".join(f"{type(e).__name__}: {e}" for e in errors)
+        raise RuntimeError(
+            f"All {len(self._backends)} backends failed: {summary}"
+        )
+
+    @property
+    def name(self) -> str:
+        mode = "race" if self._scorer is None else "best_of"
+        return f"{mode}:[{' | '.join(b.name for b in self._backends)}]"
+
+    @property
+    def cost_per_1k_tokens(self) -> float:
+        # cost of the winning backend; before any run, defaults to the first.
+        return self._last_cost_per_1k
+
+    @property
+    def last_used(self) -> str:
+        return self._last_used

graqle/routing.py

@@ -35,6 +35,7 @@
 import logging
 import os
 from dataclasses import dataclass, field
+from enum import Enum
 from typing import Any
 
 logger = logging.getLogger("graqle.routing")
@@ -489,3 +490,94 @@ def to_config(self) -> dict[str, Any]:
         if self.default_model:
             config["default_model"] = self.default_model
         return config
+
+
+# ---------------------------------------------------------------------------
+# ADR-240 D2 — CostAwareRouter: automatic cost/latency selection over a
+# candidate set given a task-difficulty signal. TaskRouter (above) stays
+# ADVISORY/opt-in and is UNTOUCHED — this is a separate, additive runtime
+# selector. Provider-agnostic: it ranks candidates by their RELATIVE
+# cost_per_1k_tokens (the one capability signal every BaseBackend exposes),
+# so no per-model capability database is needed. The CALLER supplies the
+# difficulty (from task length, plan-vs-code step type, CR-B09 retry flags).
+# ---------------------------------------------------------------------------
+
+class Difficulty(str, Enum):
+    """Task difficulty tier the caller passes to CostAwareRouter.select()."""
+
+    SIMPLE = "simple"      # cheap/fast tier is enough
+    MODERATE = "moderate"
+    HARD = "hard"          # force a strong (pricier) candidate
+
+
+_DIFFICULTY_MIN_TIER = {
+    Difficulty.SIMPLE: 0,
+    Difficulty.MODERATE: 1,
+    Difficulty.HARD: 2,
+}
+
+
+def _assign_tiers(ranked: list) -> dict[str, int]:
+    """Assign 0..2 capability tiers by terciles of the cost-ascending order.
+
+    Cheapest third → tier 0 (weakest), dearest third → tier 2 (strongest).
+    Deterministic given the candidate list.
+    """
+    n = len(ranked)
+    tiers: dict[str, int] = {}
+    for i, b in enumerate(ranked):
+        if n <= 1:
+            tiers[b.name] = 0
+        else:
+            # position 0..n-1 → tier 0/1/2 by thirds
+            frac = i / (n - 1)
+            tiers[b.name] = 0 if frac < 1 / 3 else (1 if frac < 2 / 3 else 2)
+    return tiers
+
+
+@dataclass
+class CostAwareRouter:
+    """Auto-select the cheapest backend that meets a task's difficulty.
+
+    select() never returns None: HARD with only cheap candidates still returns
+    the strongest available, and ceilings never eliminate the last option.
+    Tie-break is deterministic (cost, latency_hint, name) so it is unit-testable
+    with MockBackend.
+    """
+
+    max_cost_per_1k: float | None = None
+    max_latency_ms: float | None = None
+
+    def select(
+        self,
+        candidates: list,
+        difficulty: Difficulty,
+        *,
+        latency_hint: dict[str, float] | None = None,
+    ) -> Any:
+        if not candidates:
+            raise ValueError("CostAwareRouter.select requires at least one candidate")
+
+        # 1. rank by cost ascending (stable on name) → assign capability tiers
+        ranked = sorted(candidates, key=lambda b: (b.cost_per_1k_tokens, b.name))
+        tier_of = _assign_tiers(ranked)
+        min_tier = _DIFFICULTY_MIN_TIER[difficulty]
+
+        # 2. eligible = candidates whose tier meets the difficulty; if none
+        #    qualifies, take the strongest (dearest) available.
+        eligible = [b for b in ranked if tier_of[b.name] >= min_tier]
+        if not eligible:
+            eligible = [ranked[-1]]
+
+        # 3. apply optional cost ceiling, but never empty the set
+        if self.max_cost_per_1k is not None:
+            filtered = [b for b in eligible if b.cost_per_1k_tokens <= self.max_cost_per_1k]
+            eligible = filtered or eligible
+
+        # 4. pick cheapest meeting difficulty; deterministic tie-break
+        hints = latency_hint or {}
+
+        def _key(b: Any) -> tuple:
+            return (b.cost_per_1k_tokens, hints.get(b.name, float("inf")), b.name)
+
+        return min(eligible, key=_key)

pyproject.toml

@@ -4,12 +4,12 @@ build-backend = "hatchling.build"
 
 [project]
 name = "graqle"
-# V-ADR239-STAGE2: native version bump (G4 config, CG-14 gate can't resolve cross-repo
-# relative path — established native-edit precedent per prior V-CR-G1-NATIVE bump).
-# 0.77.0 = ADR-239 Stage 2: AutonomousExecutor CheckpointProtocol + run_tests (one
-# engine for git AND cloud consumers; backward-compatible). 0.76.0 = ADR-225 G1
-# multi-tenant memory service. 0.75.1 = ADR-220-A R2 Studio fail-closed.
-version = "0.77.0"
+# V-ADR240-DELTA: native version bump (G4 config, CG-14 native-edit precedent).
+# 0.78.0 = ADR-240 D1+D2: BackendRaceChain (first-to-finish/best-of-N) +
+# CostAwareRouter (auto cost/latency selection) — dual-provider autonomy;
+# additive, backward-compatible, no IP concern. 0.77.0 = ADR-239 Stage 2
+# (CheckpointProtocol + run_tests). 0.76.0 = ADR-225 G1 multi-tenant memory.
+version = "0.78.0"
 description = "Give your AI tools architecture-aware reasoning. Build a knowledge graph from any codebase — dependency analysis, impact analysis, governed AI answers with confidence scores. Works with Claude Code, Cursor, VS Code Copilot. 14 LLM backends, fully offline capable."
 readme = "README.md"
 license = {text = "Apache-2.0"}