""" TeCoxBeta — Rewind intelligence layer. Replaces the old the previous backend brain entirely. Three capabilities that change everything about what the search engine can do: 1. 560+ models, 267 of them NON-FREE and UNMODERATED, up to 2M context. (`:free` variants route through OpenRouter and are filtered — we never use them. Non-free ids run on Rewind GPUs and are uncensored.) 2. A self-refilling API key pool: accounts are minted on demand, each carrying its own credit balance, rotated least-recently-used. No user ever types a key. 3. prxBin proxy relay with country rotation for when an IP gets limited. Verified live: 300,000-character prompts accepted in ~4.9s. That is ~35x the entire prompt budget of the previous backend, in a single call. """ from __future__ import annotations import asyncio import json import logging import os import random import time from pathlib import Path from typing import Any, Dict, List, Optional, Tuple import httpx log = logging.getLogger("tecox.rewind") # ── endpoints ───────────────────────────────────────────────────────────── API = "https://api.rewind.ai/v1" SIGNUP = f"{API}/auth/signup" KEYS = f"{API}/api-keys" ME = f"{API}/users/me" CHAT = f"{API}/chat/completions/" MODELS_URL = f"{API}/models" PRXBIN = os.getenv("PRXBIN_URL", "https://pr-xbin.vercel.app/api/proxy") # Rewind sits behind Cloudflare, which returns error 1010 ("browser # signature blocked") for non-browser User-Agents. A desktop Chrome UA is # mandatory on EVERY request or the whole API is a wall. UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36") POOL_FILE = Path(os.getenv("TECOX_POOL_FILE", str(Path.home() / ".tecox_pool.json"))) MODEL_CACHE = Path(os.getenv("TECOX_MODEL_CACHE", str(Path.home() / ".tecox_models.json"))) EMAIL_DOMAINS = ["proton.me", "tutanota.com", "disroot.org", "protonmail.com", "aol.com", "icloud.com"] PROXY_COUNTRIES = ["us", "gb", "de", "fr", "ca", "nl", "se", "jp", "sg", "au"] _FIRST = ["alex", "sam", "jordan", "taylor", "morgan", "casey", "riley", "quinn", "avery", "blake", "chris", "drew", "emery", "finn", "harper", "jude", "kai", "luna", "max", "nova", "piper", "reese", "sage", "theo", "vera"] _LAST = ["adams", "baker", "chen", "davis", "evans", "foster", "garcia", "hall", "jones", "kim", "lee", "miller", "nguyen", "ortiz", "park", "ross", "smith", "torres", "walker", "young"] # ══════════════════════════════════════════════════════════════════════════ # Token accounting # ══════════════════════════════════════════════════════════════════════════ def estimate_tokens(text: str) -> int: """ Fast local token estimate (~3.9 chars/token for English prose+markup). Used to measure what we ASKED for, including calls that returned nothing and therefore never reported usage. """ if not text: return 0 return max(1, int(len(text) / 3.9)) class TokenLedger: """ Real token accounting for a request. Upstream reports `prompt_tokens`, `completion_tokens` and `total_tokens`, but `total_tokens` is NOT their sum - it is the BILLED CREDIT cost (verified: prompt=6605, completion=235, total=151, and the account balance dropped by exactly the sum of those `total_tokens` values). So we track three distinct things: requested - tokens we sent upstream (prompt side), incl. failed calls used - tokens actually processed (prompt + completion) on success wasted - tokens sent on calls that failed / returned empty / were superseded, i.e. spend that produced no output credits - real billed units drawn from the account balance """ __slots__ = ("calls", "ok", "failed", "prompt_tokens", "completion_tokens", "credits", "requested", "wasted", "per_model", "streamed") def __init__(self) -> None: self.calls = 0 self.ok = 0 self.failed = 0 self.prompt_tokens = 0 self.completion_tokens = 0 self.credits = 0 self.requested = 0 self.wasted = 0 self.streamed = 0 self.per_model: Dict[str, Dict[str, int]] = {} def _slot(self, model: str) -> Dict[str, int]: return self.per_model.setdefault( model, {"calls": 0, "prompt": 0, "completion": 0, "credits": 0, "wasted": 0}) def record(self, model: str, usage: Optional[Dict[str, Any]], est_in: int, streamed: bool = False) -> None: self.calls += 1 self.ok += 1 if streamed: self.streamed += 1 u = usage or {} p = int(u.get("prompt_tokens") or 0) or est_in cpl = int(u.get("completion_tokens") or 0) cr = int(u.get("total_tokens") or 0) self.prompt_tokens += p self.completion_tokens += cpl self.credits += cr self.requested += est_in s = self._slot(model) s["calls"] += 1 s["prompt"] += p s["completion"] += cpl s["credits"] += cr def record_waste(self, model: str, est_in: int) -> None: """A call that consumed input but produced nothing usable.""" self.calls += 1 self.failed += 1 self.requested += est_in self.wasted += est_in s = self._slot(model) s["calls"] += 1 s["wasted"] += est_in def snapshot(self) -> Dict[str, Any]: used = self.prompt_tokens + self.completion_tokens # Everything we pushed upstream: real prompt tokens on successful # calls + estimated input on wasted ones. total_req = self.prompt_tokens + self.wasted + self.completion_tokens eff = (used / total_req * 100) if total_req else 100.0 return { "calls": self.calls, "successful": self.ok, "failed": self.failed, "streamed": self.streamed, "tokens_requested": self.prompt_tokens + self.wasted, "tokens_used": used, "tokens_prompt": self.prompt_tokens, "tokens_completion": self.completion_tokens, "tokens_wasted": self.wasted, "credits_billed": self.credits, "efficiency_pct": round(eff, 1), "waste_pct": round(self.wasted / total_req * 100, 1) if total_req else 0.0, # Per-engine breakdown, anonymised: backend vendor identifiers are # never exposed beyond this process. "per_engine": [ {"ref": f"e{abs(hash(m)) % 100000:05d}", "tier": ("unfiltered" if (catalog.models.get(m, {}) or {}).get("uncensored") else "standard"), **{k: v for k, v in st.items()}} for m, st in self.per_model.items() ], } def merge(self, other: "TokenLedger") -> None: self.calls += other.calls self.ok += other.ok self.failed += other.failed self.streamed += other.streamed self.prompt_tokens += other.prompt_tokens self.completion_tokens += other.completion_tokens self.credits += other.credits self.requested += other.requested self.wasted += other.wasted for m, v in other.per_model.items(): s = self._slot(m) for k in v: s[k] = s.get(k, 0) + v[k] # Global ledger (process lifetime) + per-request context ledger. ledger = TokenLedger() _request_ledger: "contextvars.ContextVar[Optional[TokenLedger]]" import contextvars # noqa: E402 _request_ledger = contextvars.ContextVar("tecox_request_ledger", default=None) def start_request_ledger() -> TokenLedger: """Begin per-request accounting; returns the fresh ledger.""" lg = TokenLedger() _request_ledger.set(lg) return lg def current_ledger() -> Optional[TokenLedger]: return _request_ledger.get() def _account(model: str, usage: Optional[Dict[str, Any]], est_in: int, streamed: bool = False) -> None: ledger.record(model, usage, est_in, streamed) lg = _request_ledger.get() if lg is not None: lg.record(model, usage, est_in, streamed) def _account_waste(model: str, est_in: int) -> None: ledger.record_waste(model, est_in) lg = _request_ledger.get() if lg is not None: lg.record_waste(model, est_in) # ══════════════════════════════════════════════════════════════════════════ # Model catalogue # ══════════════════════════════════════════════════════════════════════════ class ModelCatalog: """ Live model registry. Ranks models by (uncensored, context, capability) and hands the engine the best model for each role. Hard rule: any id ending in `:free` is discarded. Those are routed to OpenRouter and are moderated/filtered — the exact thing we're avoiding. """ # Preferred ids per role, best first. Resolved against the live catalogue # so a retired model silently falls through to the next choice. # ── Public tiers ────────────────────────────────────────────────── # Callers pick a TIER, never a raw backend id. Three tiers only: # # TeF fast + smart - quick answers, planning, routing # TeD deep + smart - thorough research and long synthesis # TeM maximum - everything at full strength # # Free-form model selection is deliberately not offered: this pipeline # pushes very large evidence contexts through synthesis, and an # under-powered model silently truncates or fails on them. PUBLIC_TIERS: Dict[str, Dict[str, str]] = { "TeF": {"name": "TeF", "label": "Fast", "summary": "Fast and smart. Quick answers, planning and routing.", "synth_role": "fast", "worker_role": "fast"}, "TeD": {"name": "TeD", "label": "Deep", "summary": "Deep and smart. Thorough research and long synthesis.", "synth_role": "synth", "worker_role": "worker"}, "TeM": {"name": "TeM", "label": "Max", "summary": "Maximum capability. Largest context, fullest reasoning.", "synth_role": "max", "worker_role": "worker"}, } DEFAULT_TIER = "TeD" @classmethod def tier_roles(cls, tier: Optional[str], unfiltered: bool = False ) -> Tuple[str, str]: """Map a public tier to internal (synth_role, worker_role).""" t = cls.PUBLIC_TIERS.get((tier or cls.DEFAULT_TIER).strip(), cls.PUBLIC_TIERS[cls.DEFAULT_TIER]) synth, worker = t["synth_role"], t["worker_role"] if unfiltered: synth = "unfiltered_synth" worker = "uncensored" return synth, worker ROLE_PREFS: Dict[str, List[str]] = { # Deep synthesis / final answer: huge context + strong reasoning. # Benchmarked for OPEN (non-refusing) output + latency. # grok-4.20 answers well but took 20.1s; gemini-2.5-flash matched it # for openness at 8.4s, so speed wins the default slot. "synth": [ "google/gemini-2.5-flash", "qwen/qwen3.8-2.4t-a95b", "z-ai/glm-5.3-flash", "deepseek/deepseek-v4-flash-0731", "moonshotai/kimi-k3", "x-ai/grok-4.20", ], # Maximum tier: largest context and strongest reasoning available. "max": [ "x-ai/grok-4.20", "openai/gpt-5.5", "google/gemini-3.1-pro-preview", "anthropic/claude-fable-5", "qwen/qwen3.8-2.4t-a95b", "moonshotai/kimi-k3", "z-ai/glm-5.3", "google/gemini-2.5-pro", ], # Parallel subagents: fast, cheap, big enough context. "worker": [ "google/gemini-3.1-flash-lite", "deepseek/deepseek-v4-flash-0731", "meta-llama/llama-4-maverick", "google/gemini-2.5-flash-lite", "z-ai/glm-5.3-flash", "google/gemini-2.5-flash", ], # Planning / routing / reranking: fastest possible. "fast": [ "google/gemini-2.5-flash-lite", "google/gemini-3.1-flash-lite", "z-ai/glm-5.3-flash", "google/gemini-2.5-flash", ], # Fully uncensored path for sensitive/unfiltered requests. # Verified non-refusing on an adversarial probe, fastest first. "uncensored": [ "thedrummer/cydonia-24b-v4.1", "cognitivecomputations/dolphin-mistral-24b-venice-edition", "nousresearch/hermes-4-70b", "microsoft/wizardlm-2-8x22b", "nousresearch/hermes-4-405b", "thedrummer/rocinante-12b", ], # Long-form uncensored synthesis (bigger context than the 24Bs). "unfiltered_synth": [ "nousresearch/hermes-4-70b", "nousresearch/hermes-4-405b", "microsoft/wizardlm-2-8x22b", "sao10k/l3.3-euryale-70b", "cognitivecomputations/dolphin-mistral-24b-venice-edition", ], } def __init__(self) -> None: self.models: Dict[str, Dict[str, Any]] = {} self.resolved: Dict[str, str] = {} self._loaded = False async def load(self, client: httpx.AsyncClient, force: bool = False) -> None: if self._loaded and not force: return raw: List[Dict[str, Any]] = [] if MODEL_CACHE.exists() and not force: try: cached = json.loads(MODEL_CACHE.read_text()) if time.time() - cached.get("ts", 0) < 86400: raw = cached.get("models", []) except Exception: raw = [] if not raw: try: r = await client.get(MODELS_URL, headers={"User-Agent": UA, "Accept": "application/json"}, timeout=30.0) if r.status_code == 200: raw = (r.json() or {}).get("models", []) try: MODEL_CACHE.write_text(json.dumps({"ts": time.time(), "models": raw})) except Exception: pass except Exception as e: # noqa: BLE001 log.warning("model list fetch failed: %s", e) for m in raw: mid = m.get("id", "") # NEVER use :free — those go via OpenRouter and are filtered. if not mid or mid.endswith(":free") or mid.endswith(":batch"): continue if m.get("type") != "chat": continue caps = m.get("capabilities") or {} self.models[mid] = { "id": mid, "name": m.get("name", mid), "context": caps.get("contextLength") or 0, "uncensored": caps.get("isModerated") is False, "tools": "tools" in (caps.get("supportedParameters") or []), } self._resolve_roles() self._loaded = True log.debug("synthesis catalogue ready") def _resolve_roles(self) -> None: for role, prefs in self.ROLE_PREFS.items(): pick = next((p for p in prefs if p in self.models), None) if not pick and self.models: pool = [v for v in self.models.values() if v["uncensored"]] or list(self.models.values()) pool.sort(key=lambda v: v["context"], reverse=True) pick = pool[0]["id"] if pick: self.resolved[role] = pick def get(self, role: str, fallback: str = "google/gemini-2.5-flash") -> str: return self.resolved.get(role) or fallback def context_of(self, model_id: str) -> int: return (self.models.get(model_id) or {}).get("context", 131072) def report(self) -> Dict[str, Any]: unc = [v for v in self.models.values() if v["uncensored"]] return { "total_usable": len(self.models), "uncensored": len(unc), "max_context": max((v["context"] for v in self.models.values()), default=0), "roles": dict(self.resolved), } catalog = ModelCatalog() # ══════════════════════════════════════════════════════════════════════════ # Self-refilling key pool # ══════════════════════════════════════════════════════════════════════════ class KeyPool: """ Mints and rotates Rewind API keys so the user never supplies one. Each fresh account starts with 10,000 free credits and a 5,000/day cap. Keys are persisted, rotated least-recently-used, marked exhausted on INSUFFICIENT_TOKENS, and auto-revived after the 24h daily reset. """ DAILY = 5000 def __init__(self, target: int = 8, maximum: int = 60) -> None: self.keys: List[Dict[str, Any]] = [] self.target = int(os.getenv("TECOX_POOL_TARGET", str(target))) self.maximum = int(os.getenv("TECOX_POOL_MAX", str(maximum))) self._lock = asyncio.Lock() self._refilling = False self._loaded = False # ---- persistence ----------------------------------------------------- def load(self) -> None: if self._loaded: return self._loaded = True if POOL_FILE.exists(): try: self.keys = json.loads(POOL_FILE.read_text()) log.debug("synthesis capacity restored") except Exception: self.keys = [] for env_key in (os.getenv("REWIND_API_KEYS", "") or "").split(","): env_key = env_key.strip() if env_key and not any(k.get("api_key") == env_key for k in self.keys): self.keys.append(self._entry(env_key, "env-")) def save(self) -> None: try: POOL_FILE.parent.mkdir(parents=True, exist_ok=True) POOL_FILE.write_text(json.dumps(self.keys, indent=1)) except Exception: log.debug("pool save failed") @staticmethod def _entry(api_key: str, prefix: str = "", email: str = "") -> Dict[str, Any]: return {"api_key": api_key, "key_prefix": prefix, "email": email, "exhausted": False, "exhausted_at": 0.0, "created_at": time.time(), "last_used": 0.0, "used": 0, "fails": 0} # ---- health ---------------------------------------------------------- def _revive(self) -> None: now = time.time() for k in self.keys: if k.get("exhausted") and now - k.get("exhausted_at", 0) > 86400: k["exhausted"] = False k["used"] = 0 k["fails"] = 0 def available(self) -> List[Dict[str, Any]]: self._revive() return [k for k in self.keys if not k.get("exhausted") and k.get("fails", 0) < 4] # ---- acquisition ----------------------------------------------------- async def acquire(self, client: httpx.AsyncClient) -> Optional[str]: self.load() avail = self.available() if len(avail) < self.target: asyncio.create_task(self._refill(client)) if not avail: entry = await self._mint(client) if entry: async with self._lock: self.keys.append(entry) self.save() return entry["api_key"] return None # caller falls back to anonymous # Spread load by remaining daily headroom first, then LRU. Pure LRU # burned one key to exhaustion before touching the next. avail.sort(key=lambda k: (k.get("used", 0), k.get("last_used", 0))) chosen = avail[0] chosen["last_used"] = time.time() return chosen["api_key"] def mark_exhausted(self, api_key: str) -> None: for k in self.keys: if k.get("api_key") == api_key: k["exhausted"] = True k["exhausted_at"] = time.time() self.save() return def mark_fail(self, api_key: str) -> None: for k in self.keys: if k.get("api_key") == api_key: k["fails"] = k.get("fails", 0) + 1 return def mark_ok(self, api_key: str, tokens: int = 0) -> None: for k in self.keys: if k.get("api_key") == api_key: k["fails"] = 0 k["used"] = k.get("used", 0) + tokens return # ---- minting --------------------------------------------------------- async def _refill(self, client: httpx.AsyncClient) -> None: if self._refilling: return self._refilling = True try: deficit = min(self.target - len(self.available()), self.maximum - len(self.keys)) if deficit <= 0: return results = await asyncio.gather( *[self._mint(client) for _ in range(min(deficit, 6))], return_exceptions=True) async with self._lock: added = 0 for r in results: if isinstance(r, dict): self.keys.append(r) added += 1 if added: self.save() log.debug("synthesis capacity extended") finally: await asyncio.sleep(4) self._refilling = False async def _mint(self, client: httpx.AsyncClient) -> Optional[Dict[str, Any]]: """Create an account → persistent sk-rewind key. Direct, then proxied.""" first, last = random.choice(_FIRST), random.choice(_LAST) tag = f"{first}{last}{random.randint(1000, 99999)}" email = f"{tag}@{random.choice(EMAIL_DOMAINS)}" password = f"{first.capitalize()}{random.randint(100, 999)}!x{random.randint(10, 99)}" body = {"email": email, "password": password} hdr = {"Content-Type": "application/json", "User-Agent": UA} token = "" try: r = await client.post(SIGNUP, json=body, headers=hdr, timeout=25.0) if r.status_code in (200, 201): token = (r.json() or {}).get("accessToken", "") except Exception as e: # noqa: BLE001 log.debug("signup direct failed: %s", e) if not token: # IP-limited -> prxBin is used for POOL SIGNUP ONLY data, status = await prxbin(client, SIGNUP, "POST", hdr, body, random.choice(PROXY_COUNTRIES)) if status in (200, 201) and isinstance(data, dict): token = data.get("accessToken", "") if not token: return None api_key, prefix = "", "" try: r = await client.post( KEYS, headers={**hdr, "Authorization": f"Bearer {token}"}, json={"name": f"tecox-{tag}", "scopes": ["chat", "image", "video", "audio", "embedding"]}, timeout=25.0) if r.status_code in (200, 201): kd = r.json() or {} api_key, prefix = kd.get("key", ""), kd.get("keyPrefix", "") except Exception: pass if not api_key: # access tokens still work for ~24h api_key, prefix = token, "atk-" return self._entry(api_key, prefix, email) def report(self) -> Dict[str, Any]: self.load() return {"total": len(self.keys), "available": len(self.available()), "exhausted": sum(1 for k in self.keys if k.get("exhausted")), "target": self.target} pool = KeyPool() # ══════════════════════════════════════════════════════════════════════════ # prxBin relay # ══════════════════════════════════════════════════════════════════════════ async def prxbin(client: httpx.AsyncClient, url: str, method: str = "GET", headers: Optional[Dict[str, str]] = None, body: Any = None, country: str = "us", timeout: float = 120.0) -> Tuple[Any, int]: """Relay a request through prxBin so a blocked IP isn't fatal.""" payload = {"url": url, "method": method, "headers": headers or {}, "body": json.dumps(body) if body is not None else None, "proxy_country": country, "retries": 2, "stream": False} try: r = await client.post(PRXBIN, json=payload, timeout=timeout) if r.status_code != 200: return None, r.status_code env = r.json() data = env.get("data") if isinstance(data, str): try: data = json.loads(data) except Exception: pass return data, env.get("status", 502) except Exception as e: # noqa: BLE001 log.debug("prxbin failed: %s", e) return None, -1 # ══════════════════════════════════════════════════════════════════════════ # Chat # ══════════════════════════════════════════════════════════════════════════ class RewindClient: """Async chat client with pooled keys, anonymous fallback and proxy relay.""" def __init__(self) -> None: self._client: Optional[httpx.AsyncClient] = None self.available = True self.calls = 0 self.failures = 0 async def client(self) -> httpx.AsyncClient: if self._client is None or self._client.is_closed: self._client = httpx.AsyncClient( timeout=httpx.Timeout(180.0, connect=15.0), limits=httpx.Limits(max_connections=64, max_keepalive_connections=32), headers={"User-Agent": UA}, follow_redirects=True) return self._client async def close(self) -> None: if self._client and not self._client.is_closed: await self._client.aclose() async def ready(self) -> None: await catalog.load(await self.client()) pool.load() async def chat(self, prompt: str, *, model: Optional[str] = None, role: str = "fast", system: Optional[str] = None, max_tokens: Optional[int] = None, temperature: float = 0.7, timeout: float = 180.0, retries: int = 3) -> str: """One completion. Rotates keys, falls back to anonymous, then proxy.""" c = await self.client() await catalog.load(c) mdl = model or catalog.get(role) messages: List[Dict[str, str]] = [] if system: messages.append({"role": "system", "content": system}) messages.append({"role": "user", "content": prompt}) payload: Dict[str, Any] = {"model": mdl, "messages": messages, "stream": False, "temperature": temperature} if max_tokens: payload["max_tokens"] = max_tokens self.calls += 1 est_in = estimate_tokens(prompt) + estimate_tokens(system or "") last_key: Optional[str] = None for attempt in range(retries): key = await pool.acquire(c) last_key = key hdr = {"Content-Type": "application/json", "User-Agent": UA} if key: hdr["Authorization"] = f"Bearer {key}" try: r = await c.post(CHAT, json=payload, headers=hdr, timeout=timeout) if r.status_code == 200: d = r.json() txt = (d.get("choices") or [{}])[0].get("message", {}).get("content", "") if key: pool.mark_ok(key, (d.get("usage") or {}).get("total_tokens", 0)) if txt: _account(mdl, d.get("usage"), est_in) return txt.strip() _account_waste(mdl, est_in) elif r.status_code in (402, 429): body = r.text[:200] if key and ("INSUFFICIENT_TOKENS" in body or r.status_code == 402): pool.mark_exhausted(key) continue else: if key: pool.mark_fail(key) except Exception as e: # noqa: BLE001 log.debug("chat attempt %d failed: %s", attempt, e) if key: pool.mark_fail(key) await asyncio.sleep(0.4 * (attempt + 1)) # Last resort: relay through prxBin from a different country. hdr = {"Content-Type": "application/json", "User-Agent": UA} if last_key: hdr["Authorization"] = f"Bearer {last_key}" data, status = await prxbin(await self.client(), CHAT, "POST", hdr, payload, random.choice(PROXY_COUNTRIES), timeout=timeout) if status == 200 and isinstance(data, dict): txt = (data.get("choices") or [{}])[0].get("message", {}).get("content", "") if txt: _account(mdl, data.get("usage"), est_in) return txt.strip() self.failures += 1 _account_waste(mdl, est_in) return "" async def chat_stream(self, prompt: str, *, model: Optional[str] = None, role: str = "synth", system: Optional[str] = None, max_tokens: Optional[int] = None, temperature: float = 0.7, timeout: float = 300.0): """ TRUE server-sent-event streaming. Yields text deltas as they arrive. This is what removes the perceived latency: instead of waiting for a full 12k-character report, the user sees the first words in ~2-3s. Token usage is recorded into the ledger when the final chunk lands. """ c = await self.client() await catalog.load(c) mdl = model or catalog.get(role) messages: List[Dict[str, str]] = [] if system: messages.append({"role": "system", "content": system}) messages.append({"role": "user", "content": prompt}) payload: Dict[str, Any] = {"model": mdl, "messages": messages, "stream": True, "temperature": temperature} if max_tokens: payload["max_tokens"] = max_tokens self.calls += 1 est_in = estimate_tokens(prompt) + estimate_tokens(system or "") for attempt in range(2): key = await pool.acquire(c) hdr = {"Content-Type": "application/json", "User-Agent": UA, "Accept": "text/event-stream"} if key: hdr["Authorization"] = f"Bearer {key}" got_any = False usage: Dict[str, Any] = {} try: async with c.stream("POST", CHAT, json=payload, headers=hdr, timeout=timeout) as r: if r.status_code != 200: body = (await r.aread()).decode("utf-8", "ignore")[:200] if key and ("INSUFFICIENT_TOKENS" in body or r.status_code in (402, 429)): pool.mark_exhausted(key) elif key: pool.mark_fail(key) continue async for line in r.aiter_lines(): if not line or not line.startswith("data:"): continue data = line[5:].strip() if data == "[DONE]": break try: j = json.loads(data) except Exception: continue if j.get("usage"): usage = j["usage"] delta = ((j.get("choices") or [{}])[0] .get("delta", {}).get("content")) if delta: got_any = True yield delta if got_any: _account(mdl, usage, est_in, streamed=True) if key: pool.mark_ok(key, int(usage.get("total_tokens") or 0)) return except Exception as e: # noqa: BLE001 log.debug("stream attempt %d failed: %s", attempt, e) if key: pool.mark_fail(key) if got_any: return # Streaming unavailable -> fall back to a single blocking call. text = await self.chat(prompt, model=mdl, role=role, system=system, max_tokens=max_tokens, temperature=temperature, timeout=timeout) if text: yield text async def fanout(self, prompts: List[str], *, role: str = "worker", system: Optional[str] = None, concurrency: int = 8, **kw: Any) -> List[str]: """Run many completions in true parallel — the core of subagent waves.""" sem = asyncio.Semaphore(concurrency) async def one(p: str) -> str: async with sem: return await self.chat(p, role=role, system=system, **kw) out = await asyncio.gather(*[one(p) for p in prompts], return_exceptions=True) return [r if isinstance(r, str) else "" for r in out] def report(self) -> Dict[str, Any]: return {"calls": self.calls, "failures": self.failures, "pool": pool.report(), "catalog": catalog.report(), "tokens": ledger.snapshot()} rewind = RewindClient()