Index multi-account routing and probe one catalog per profile

.env.example

@@ -104,6 +104,16 @@ CODEBUDDY2API_MAX_CONCURRENT=64
 # Disable only the declared model-capability preflight; other safeguards and Intl image merging remain.
 # 只关闭声明的模型能力预检；其他保护与国际站图片合并仍然生效。
 # CODEBUDDY2API_MODEL_CAPABILITY_GUARD=true
+# Routing scheduler: indexed reuses per-model shards, shadow records legacy/indexed differences
+# while returning the legacy answer, legacy rescans every credential on each request.
+# 路由调度：indexed 复用按模型分片（默认），shadow 记录新旧差异但仍返回旧结果，legacy 每次全量判断。
+# CODEBUDDY2API_ROUTING_SCHEDULER=indexed
+# Catalog probing: representative probes one account per product per TTL cycle; account refreshes all.
+# 目录探测：representative 每个产品每周期只探测一个代表账号（默认），account 逐个账号刷新。
+# CODEBUDDY2API_CATALOG_PROBE_MODE=representative
+# Bounded background maintenance workers; 1 restores the previous serial behaviour.
+# 后台维护并发数；设为 1 即退化为串行。
+# CODEBUDDY2API_HOUSEKEEP_CONCURRENCY=4
 
 # Retired: this setting only warns; all persistent logs use logs.sqlite3.
 # 已停用：该项只会给出告警；所有持久日志都写入 logs.sqlite3。

app/admin_api.py

@@ -17,6 +17,7 @@
 from starlette.responses import JSONResponse, Response
 
 from . import auth_oauth
+from . import perf_metrics
 from .admin_auth import AdminAuth, AdminMiddleware, COOKIE_NAME, SESSION_TTL, error_response, same_origin
 from .control_store import ConflictError, validate_model
 from .credential_io import CredentialFileError, MAX_CREDENTIAL_BYTES, _valid_name, read_import_file
@@ -338,9 +339,21 @@ def build_models():
                     "model_guard_locked": bool(resolved.get("model_guard", {}).get("locked")),
                     "model_capability_guard": config.get("model_capability_guard", True),
                     "model_capability_guard_locked": bool(resolved.get("model_capability_guard", {}).get("locked")),
+                    "catalog": gateway.catalog_status(),
                 })
         return await run_in_threadpool(build_models)
 
+    @route("POST", "/admin/models/scan")
+    async def models_scan_start(request):
+        """Start one user-triggered full catalog scan; it is never a periodic task."""
+        result = await run_in_threadpool(gateway.start_catalog_scan)
+        event("models.scan")
+        return JSONResponse(result)
+
+    @route("GET", "/admin/models/scan")
+    async def models_scan_state(request):
+        return JSONResponse(gateway.catalog_scan_view())
+
     def checked_rule(source, data, *, creating=False):
         if "custom" in data:
             raise ValueError("custom 是只读字段")
@@ -687,4 +700,14 @@ def build_dashboard():
         result = await run_in_threadpool(build_dashboard)
         return result if isinstance(result, Response) else JSONResponse(result)
 
+    @route("GET", "/admin/performance")
+    async def performance_get(request):
+        """Read-only process counters; no routing state is exposed or mutated."""
+        payload = perf_metrics.snapshot()
+        pool = config.get("cred_pool")
+        scheduler = getattr(pool, "_scheduler", None)
+        payload["scheduler"] = {"mode": config.get("routing_scheduler", "indexed"),
+                                **(scheduler.report() if scheduler is not None else {"shards": [], "mismatches": []})}
+        return JSONResponse(payload)
+
     return auth

app/catalog_index.py

@@ -0,0 +1,77 @@
+"""Read-only indexes over published account catalogs, built once per catalog revision."""
+from __future__ import annotations
+
+from .model_catalog_view import INTERNATIONAL, share_models
+
+
+def account_scope(account, scope="models"):
+    """Merge root candidates into selector models while retaining selector metadata."""
+    picker = account.get("models")
+    if scope == "models" or picker is None:
+        return picker
+    seen = {item.get("id") for item in picker}
+    # Root entries supply missing names without replacing selector metadata.
+    return picker + [item for item in account.get("serves") or [] if item.get("id") not in seen]
+
+
+class CatalogIndex:
+    """Precompute per-account and per-profile catalogs so request paths never rescan accounts.
+
+    ``accounts`` is the published ``account_catalogs`` mapping; the index keeps that same
+    object so a caller can detect a republished catalog by identity.
+    """
+
+    SCOPES = ("models", "serves")
+
+    def __init__(self, accounts, *, revision=0, policy_revision=0, is_enabled=None):
+        self.accounts = accounts if isinstance(accounts, dict) else {}
+        self.revision = revision
+        self.policy_revision = policy_revision
+        self._enabled = is_enabled or (lambda _identity: True)
+        self._account: dict[tuple[int, str], list[dict] | None] = {}
+        self._profile: dict[str, dict[str, tuple[dict, ...]]] = {}
+        self.profiles: tuple[str, ...] = tuple(dict.fromkeys(
+            account.get("profile") for account in self.accounts.values() if account.get("profile")))
+        for scope in self.SCOPES:
+            sources = self._shared_sources(scope)
+            merged: dict[str, list[dict]] = {}
+            for identity, account in self.accounts.items():
+                profile = account.get("profile") if isinstance(account, dict) else None
+                if not profile:
+                    continue
+                items = self._effective(account, scope, sources)
+                if items is None:
+                    continue
+                merged.setdefault(profile, []).extend(items)
+            self._profile[scope] = {name: tuple(items) for name, items in merged.items()}
+
+    def _shared_sources(self, scope):
+        """Collect every enabled international account catalog once per scope."""
+        return [(account["profile"], account_scope(account, scope))
+                for identity, account in self.accounts.items()
+                if isinstance(account, dict) and account.get("profile") in INTERNATIONAL
+                and account.get("models") is not None and self._enabled(identity)]
+
+    def _effective(self, account, scope, sources):
+        native = account_scope(account, scope)
+        if native is None:
+            return None
+        profile = account.get("profile")
+        items = share_models(native, profile, sources) if profile in INTERNATIONAL else native
+        self._account[(id(account), scope)] = items
+        return items
+
+    def account_catalog(self, account, scope="models"):
+        """Return this account's effective catalog, or None when the index never saw it."""
+        if not isinstance(account, dict):
+            return None
+        return self._account.get((id(account), scope))
+
+    def catalog(self, profile, scope="models", *, model_id=None):
+        """Return the merged catalog for one profile, or None when no account supplied one."""
+        items = self._profile.get(scope, {}).get(profile)
+        if items is None:
+            return None
+        if model_id is None:
+            return list(items)
+        return [item for item in items if item.get("id") == model_id]

app/catalog_probe.py

@@ -0,0 +1,160 @@
+"""Representative per-profile catalog probes with rotation, backoff and diff status."""
+from __future__ import annotations
+
+import math
+import random
+import threading
+import time
+
+SCHEMA_VERSION = 1
+PROFILES = ("cn-cli", "cn-work", "intl-cli", "intl-work")
+MAX_PROFILES = 8
+MAX_MODEL_IDS = 2000
+RETRY_BASE_S = 60
+RETRY_MAX_S = 1800
+_MAX_FAILURES = 16
+_FIELDS = {"source", "source_generation", "fetched_at", "retry_at", "failures", "status", "models"}
+
+
+def _time(value):
+    return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) and value >= 0
+
+
+def _token(value):
+    return isinstance(value, str) and 0 < len(value) <= 160 and value.isprintable() and not any(c.isspace() for c in value)
+
+
+def valid_row(row):
+    """Validate one persisted profile probe row without ever touching credential material."""
+    if not isinstance(row, dict) or set(row) - _FIELDS:
+        return False
+    if "source" in row and row["source"] is not None and not _token(row["source"]):
+        return False
+    if not _time(row.get("fetched_at")) or not _time(row.get("retry_at")):
+        return False
+    failures = row.get("failures", 0)
+    if type(failures) is not int or not 0 <= failures <= _MAX_FAILURES:
+        return False
+    if row.get("status") not in ("ok", "error", "fresh"):
+        return False
+    if "source_generation" in row and type(row["source_generation"]) is not int:
+        return False
+    models = row.get("models", [])
+    return (isinstance(models, list) and len(models) <= MAX_MODEL_IDS
+            and all(isinstance(model, str) and 0 < len(model) <= 160 for model in models))
+
+
+class AggregateCatalog:
+    """One representative source per product profile, with rotation and retry backoff."""
+
+    def __init__(self, *, ttl=6 * 3600, store=None, clock=time.time):
+        self.ttl = max(60.0, float(ttl or 0))
+        self._store = store
+        self._clock = clock
+        self._lock = threading.Lock()
+        self._data = self._load()
+
+    def _load(self):
+        raw = self._store.get("catalog_probe") if self._store is not None else None
+        if (not isinstance(raw, dict) or raw.get("version") != SCHEMA_VERSION
+                or not isinstance(raw.get("profiles"), dict)):
+            return {"version": SCHEMA_VERSION, "profiles": {}}
+        profiles = {name: dict(row) for name, row in raw["profiles"].items()
+                    if name in PROFILES and valid_row(row)}
+        return {"version": SCHEMA_VERSION, "profiles": profiles}
+
+    def _save(self):
+        if self._store is not None:
+            self._store.put_cache("catalog_probe", {"version": SCHEMA_VERSION, "profiles": self._data["profiles"]})
+
+    def row(self, profile):
+        """Return a copy of one profile row."""
+        with self._lock:
+            return dict(self._data["profiles"].get(profile) or {})
+
+    def profiles(self):
+        """Return a copy of every profile row for the management view."""
+        with self._lock:
+            return {name: dict(row) for name, row in self._data["profiles"].items()}
+
+    def due(self, profile, now=None):
+        """Whether this profile may probe again; a retry backoff keeps failures quiet."""
+        now = self._clock() if now is None else now
+        row = self.row(profile)
+        return not row or now >= float(row.get("retry_at") or 0.0)
+
+    @staticmethod
+    def _age(candidate, age_of):
+        """Treat an unknown catalog age as infinitely stale so a new account is probed."""
+        value = age_of(candidate)
+        return math.inf if value is None else float(value)
+
+    def choose(self, profile, candidates, *, age_of, now=None):
+        """Select the account to probe: unknown catalogs first, then the source, else the stalest."""
+        now = self._clock() if now is None else now
+        due = [candidate for candidate in candidates if self._age(candidate, age_of) >= self.ttl]
+        if not due:
+            return None
+        # An account without any catalog cannot be routed at all, so it outranks reuse.
+        unknown = [candidate for candidate in due if self._age(candidate, age_of) == math.inf]
+        if unknown:
+            return unknown[0]
+        source = self.row(profile).get("source")
+        for candidate in due:
+            if candidate["account_key"] == source:
+                return candidate
+        # Oldest catalogs first, so rotation eventually covers every account.
+        return max(due, key=lambda candidate: self._age(candidate, age_of))
+
+    def success(self, profile, *, source, generation, models, now=None):
+        """Record one probe result and schedule the next TTL check."""
+        now = self._clock() if now is None else now
+        with self._lock:
+            self._data["profiles"][profile] = {
+                "source": source, "source_generation": int(generation or 0),
+                "fetched_at": now, "retry_at": now + self.ttl, "failures": 0, "status": "ok",
+                "models": sorted({str(model)[:160] for model in models if model})[:MAX_MODEL_IDS]}
+            self._save()
+
+    def fresh(self, profile, now=None):
+        """Record that the representative catalog is still inside its TTL."""
+        now = self._clock() if now is None else now
+        with self._lock:
+            row = dict(self._data["profiles"].get(profile) or {})
+            if not row:
+                return
+            row.update(status="fresh", fetched_at=now, retry_at=now + self.ttl, failures=0)
+            self._data["profiles"][profile] = row
+            self._save()
+
+    def failure(self, profile, *, source=None, now=None):
+        """Back off an unusable probe and keep the previous aggregate for the next attempt."""
+        now = self._clock() if now is None else now
+        with self._lock:
+            row = dict(self._data["profiles"].get(profile) or {})
+            failures = min(int(row.get("failures") or 0) + 1, _MAX_FAILURES)
+            delay = min(RETRY_BASE_S * 2 ** (failures - 1), RETRY_MAX_S)
+            row.update(source=source if source is not None else row.get("source"),
+                       failures=failures, status="error",
+                       # Jitter keeps several failing profiles from retrying in lockstep.
+                       retry_at=now + delay * random.uniform(1.0, 1.25))
+            row.setdefault("fetched_at", 0.0)
+            row.setdefault("models", [])
+            self._data["profiles"][profile] = row
+            self._save()
+
+    def model_ids(self, profile):
+        """Return the model identifiers the last successful probe advertised."""
+        return set(self.row(profile).get("models") or ())
+
+    def view(self, now=None):
+        """Describe probe state for the management API without exposing credentials."""
+        now = self._clock() if now is None else now
+        rows = {}
+        for profile, row in self.profiles().items():
+            rows[profile] = {
+                "source": row.get("source"), "status": row.get("status", "unknown"),
+                "models": len(row.get("models") or []), "failures": row.get("failures", 0),
+                "fetched_at": row.get("fetched_at"),
+                "next_probe_in": max(0, round(float(row.get("retry_at") or 0.0) - now))}
+        return {"ttl_seconds": self.ttl, "profiles": rows}

app/control_store.py

@@ -174,6 +174,11 @@ def snapshot(self):
         # Published dictionaries are never mutated; SQLite writer locks cannot stall routing.
         return copy.deepcopy(self._snapshot)
 
+    @property
+    def revision(self):
+        """Current revision without copying the snapshot, for cache and policy views."""
+        return self._snapshot["revision"]
+
     def _update(self, revision, change):
         with self._lock:
             self._db.execute("BEGIN IMMEDIATE")

app/credits.py

@@ -4,6 +4,7 @@
 import base64
 from copy import deepcopy
 import json
+import itertools
 import math
 import os
 import re
@@ -794,6 +795,22 @@ def fetch_request_usage(access_token: str, days: int = USAGE_MAX_DAYS,
 # Atomic JSON persistence for per-credential credit and check-in state
 # ---------------------------------------------------------------------------
 
+_LEDGER_REVISION = itertools.count(1)
+
+
+def _solvent_row(row: dict | None) -> bool:
+    """Mirror the pool's has-credit test so only eligibility flips bump the routing revision."""
+    credits = (row or {}).get("credits") or {}
+    if not credits:
+        return True
+    if credits.get("unlimited") and credits.get("enterprise"):
+        return True
+    try:
+        return float(credits.get("credits") or 0) > 0
+    except (TypeError, ValueError):
+        return False
+
+
 class CreditLedger:
     """Persist per-credential check-in and balance snapshots for expiry-aware scheduling."""
 
@@ -802,6 +819,10 @@ def __init__(self, path: Path | None = None, *, store=None):
         self.path = Path(store.path if store is not None else path)
         self._lock = threading.Lock()
         self._data: dict = {"version": 1, "creds": {}}
+        # Globally unique so a rebuilt ledger can never reuse the previous one's cache key.
+        self.revision = next(_LEDGER_REVISION)
+        # Bumped only when an account's solvency flips; value churn must not clear shards.
+        self.routing_revision = next(_LEDGER_REVISION)
         self._load()
 
     def _load(self):
@@ -816,6 +837,7 @@ def _load(self):
             self._data = {"version": 1, "creds": {}}
 
     def _save(self):
+        self.revision = next(_LEDGER_REVISION)
         if self._store is not None:
             self._store.put("credits", self._data)
             return
@@ -849,7 +871,8 @@ def bind_identity(self, cred_id: str, identity: str) -> bool:
     def remove(self, cred_id: str):
         """Clear persisted credit, check-in and error state after an identity change."""
         with self._lock:
-            self._data["creds"].pop(cred_id, None)
+            if _solvent_row(self._data["creds"].pop(cred_id, None)) is not True:
+                self.routing_revision = next(_LEDGER_REVISION)
             self._save()
 
     # Check-in state
@@ -878,6 +901,9 @@ def update_travel(self, cred_id: str, result: dict):
 
     def update_credits(self, cred_id: str, result: dict):
         with self._lock:
+            # Snapshot before mutating: _entry may hand back the very row read here.
+            was_solvent = _solvent_row(self._data["creds"].get(cred_id))
+            had_balance_row = bool((self._data["creds"].get(cred_id) or {}).get("credits"))
             e = self._entry(cred_id)
             e["credits"] = {
                 "credits": result.get("credits"),
@@ -893,6 +919,10 @@ def update_credits(self, cred_id: str, result: dict):
                 "unlimited": bool(result.get("unlimited")) and bool(result.get("enterprise")),
             }
             e["error"] = None
+            # A first balance row changes eligibility too: an absent row never matched
+            # _has_credit for intl accounts, so its appearance must move the routing revision.
+            if not had_balance_row or _solvent_row({"credits": e["credits"]}) != was_solvent:
+                self.routing_revision = next(_LEDGER_REVISION)
             self._save()
 
     def note_error(self, cred_id: str, message: str):