Fix international model catalog aggregation

app/acp_client.py

@@ -16,7 +16,7 @@
 import json
 import select
 import time
-from http.client import HTTPConnection, HTTPSConnection, HTTPSConnection as _HTTPS
+from http.client import HTTPConnection, HTTPSConnection as _HTTPS
 from urllib.parse import urlsplit
 
 from .credits import BROWSER_UA

app/credits.py

@@ -59,6 +59,15 @@ class _PersonalCatalogUnavailable(RuntimeError):
     """Signal that an older or non-SaaS product has no personal model route."""
 
 
+class ModelScopes(dict):
+    """Carry catalog completeness without changing the mapping contract."""
+
+    def __init__(self, values, *, complete=True, error=None):
+        super().__init__(values)
+        self.complete = bool(complete)
+        self.error = error
+
+
 # ---------------------------------------------------------------------------
 # Billing host selection
 # ---------------------------------------------------------------------------
@@ -749,21 +758,28 @@ def fetch_personal_model_scopes(access_token: str, user_agent: str = "", *, doma
     return _parse_model_scopes(payload, profile_product(profile), "个人模型表接口")
 
 
-def fetch_model_scopes(access_token: str, user_agent: str = "", *, domain: str = "",
-                       uid: str = "", enterprise_id: str = "", prefer_personal: bool = False) -> dict[str, list[dict]]:
-    """Fetch selector and account-root catalogs without assuming every candidate is servable."""
+def _merge_model_scopes(primary: dict[str, list[dict]], supplemental: dict[str, list[dict]]) -> dict[str, list[dict]]:
+    """Append models absent from the primary catalog while preserving its declarations."""
+    merged = {}
+    for scope in ("picker", "account"):
+        rows = list(primary.get(scope) or [])
+        seen = {item.get("id") for item in rows if isinstance(item, dict) and item.get("id")}
+        rows.extend(item for item in supplemental.get(scope) or []
+                    if isinstance(item, dict) and item.get("id") not in seen)
+        merged[scope] = rows
+    return merged
+
+
+def _fetch_config_model_scopes(access_token: str, user_agent: str = "", *, domain: str = "",
+                               uid: str = "", enterprise_id: str = "") -> dict[str, list[dict]]:
+    """Fetch the legacy v3 model catalog and split its picker/root scopes."""
     auth = {"accessToken": access_token, "domain": domain}
     profile = profile_for_auth(auth)
-    if prefer_personal and not enterprise_id:
-        try:
-            return fetch_personal_model_scopes(access_token, user_agent, domain=domain, uid=uid)
-        except _PersonalCatalogUnavailable:
-            # Older deployments and private endpoints retain the established /v3/config path.
-            pass
     headers = catalog_headers(auth, {"uid": uid, "enterpriseId": enterprise_id}, user_agent=user_agent)
     try:
         with httpx.Client() as client:
-            response = client.get(PROFILE_ENDPOINTS[profile] + CONFIG_PATH, headers=headers, timeout=REQUEST_TIMEOUT)
+            response = client.get(PROFILE_ENDPOINTS[profile] + CONFIG_PATH, headers=headers,
+                                  timeout=REQUEST_TIMEOUT)
     except httpx.HTTPError:
         raise RuntimeError("模型配置接口网络错误") from None
     if response.status_code == 401:
@@ -777,6 +793,27 @@ def fetch_model_scopes(access_token: str, user_agent: str = "", *, domain: str =
     return _parse_model_scopes(payload, profile_product(profile), "模型配置接口")
 
 
+def fetch_model_scopes(access_token: str, user_agent: str = "", *, domain: str = "",
+                       uid: str = "", enterprise_id: str = "", prefer_personal: bool = False) -> dict[str, list[dict]]:
+    """Fetch selector and account-root catalogs, supplementing personal data with v3."""
+    if prefer_personal and not enterprise_id:
+        try:
+            personal = fetch_personal_model_scopes(access_token, user_agent, domain=domain, uid=uid)
+        except _PersonalCatalogUnavailable:
+            # Older deployments and private endpoints retain the established /v3/config path.
+            personal = None
+        if personal is not None:
+            try:
+                legacy = _fetch_config_model_scopes(access_token, user_agent, domain=domain,
+                                                    uid=uid, enterprise_id=enterprise_id)
+            except (AuthExpiredError, RuntimeError, ValueError) as error:
+                # Personal data remains usable, but the caller must not mark this refresh complete.
+                return ModelScopes(personal, complete=False, error=error)
+            return ModelScopes(_merge_model_scopes(personal, legacy))
+    return _fetch_config_model_scopes(access_token, user_agent, domain=domain,
+                                      uid=uid, enterprise_id=enterprise_id)
+
+
 def fetch_model_catalog(access_token: str, user_agent: str = "", *, domain: str = "",
                         uid: str = "", enterprise_id: str = "", prefer_personal: bool = False) -> list[dict]:
     """Return selector models for this credential's product."""
@@ -1139,6 +1176,7 @@ def __init__(self, path: Path | None = None, ttl: float = 6 * 3600, *, store=Non
         self.path = Path(store.path if store is not None else path)
         self.ttl = max(60.0, float(ttl or 0))
         self._lock = threading.Lock()
+        self._write_sequences: dict[str, int] = {}
         self._data: dict = {"version": self.SCHEMA_VERSION, "groups": {}}
         self._load()
 
@@ -1160,6 +1198,7 @@ def _load(self):
                     if not isinstance(ts, (int, float)) or not math.isfinite(ts):
                         ts = 0
                     groups[group] = {"models": entry["models"], "fetched_at": ts,
+                                     "complete": entry.get("complete", True) is True,
                                      "version": (self.SCHEMA_VERSION
                                                  if d["version"] == self.SCHEMA_VERSION
                                                  and entry.get("version") == self.SCHEMA_VERSION
@@ -1196,7 +1235,8 @@ def fresh(self, group: str) -> bool:
         with self._lock:
             g = self._data["groups"].get(group) or {}
             age = time.time() - float(g.get("fetched_at") or 0)
-            return g.get("version") == self.SCHEMA_VERSION and 0 <= age < self.ttl
+            return (g.get("version") == self.SCHEMA_VERSION and g.get("complete", True) is True
+                    and 0 <= age < self.ttl)
 
     def models(self, group: str) -> list[dict]:
         with self._lock:
@@ -1208,14 +1248,24 @@ def age(self, group: str) -> float | None:
             g = self._data["groups"].get(group)
             return time.time() - float(g.get("fetched_at") or 0) if g is not None else None
 
-    def put(self, group: str, models: list[dict], serves: list[dict] | None = None):
+    def put(self, group: str, models: list[dict], serves: list[dict] | None = None, *, complete=True, sequence=None):
         with self._lock:
+            current = self._data["groups"].get(group) or {}
+            current_sequence = self._write_sequences.get(group, 0)
+            if sequence is not None and sequence < current_sequence:
+                return False
+            if sequence is not None:
+                self._write_sequences[group] = sequence
+            if complete is not True and current.get("complete", True) is True and current.get("models") is not None:
+                return False
             entry = {"models": deepcopy(models), "fetched_at": time.time(),
+                     "complete": complete is True,
                      "version": self.SCHEMA_VERSION}
             if serves is not None:
                 entry["serves"] = deepcopy(serves)
             self._data["groups"][group] = entry
             self._save()
+            return True
 
     def serves(self, group: str) -> list[dict]:
         """Return root catalog candidates, or an empty list when legacy caches omit them."""

app/desensitize.py

@@ -583,7 +583,7 @@ def apply_system_prompt_mode(body: dict, profile: str, mode: str) -> dict:
         print(f"{changed} | 原文: {s}")
         if d != s:
             print(f"     | 脱敏: {d}")
-            print(f"     | 可见字符相同，差异为零宽空格 U+200B")
+            print("     | 可见字符相同，差异为零宽空格 U+200B")
     print()
     print("=== messages 脱敏（只处理 system）===")
     msgs = [

app/observability.py

@@ -13,7 +13,6 @@
 from dataclasses import dataclass, field
 import json
 import time
-from typing import Any
 
 from app.audit_store import AuditStore, METRICS, number, safe_agent, safe_attempt, safe_label
 from app.request_context import PATHS, current_context, ensure_context

converter.py

@@ -67,13 +67,14 @@ def apply_system_prompt_mode(body, profile, mode):
 
 from app import auth_oauth
 from app.settings import SCHEMA
-from app import trial_rewards
+# Re-exported for app.runtime_management and the tests, which reach it as ``gateway.trial_rewards``.
+from app import trial_rewards  # noqa: F401
 from app import buddy, daily_chat, growth_chains, growth_tasks, checkin as checkin_service, model_policy, travel
 from app.credential_cooldowns import CredentialCooldowns
 from app.model_blocks import ModelBlocks
 from app.usage_snapshots import UsageSnapshots
 from app.client_hangup import ClientHungUp, await_or_hangup
-from app.observability import (AuditMiddleware, observe_recovery, observe_route, observe_stream_mode,
+from app.observability import (observe_recovery, observe_route, observe_stream_mode,
                                observe_responses_projection, observe_usage, observe_attempt,
                                observe_failure, observe_failure_seq, observe_discard,
                                observe_nested)
@@ -106,16 +107,16 @@ def apply_system_prompt_mode(body, profile, mode):
 from app.safe_logging import format_log_body, sanitize_log_text
 from app import perf_metrics
 from app.startup import capture_launch, load_startup_env, restart_process, run_server
-from app.site_routing import (DOMESTIC, INTERNATIONAL, PROFILE_ENDPOINTS, site_for_auth, site_for_headers,
+from app.site_routing import (INTERNATIONAL, PROFILE_ENDPOINTS, site_for_headers,
                               profile_for_auth, profile_for_headers, profile_region, profile_product,
                               profile_site, chat_url_for_headers, search_url_for_headers, fetch_url_for_headers,
                               refresh_url_for_auth, media_url_for_headers, embedding_url_for_headers,
                               paid_embedding_url_for_headers, rerank_url_for_headers)
 from app import dsml, mcp_server, server_tools, web_fetch, web_search, media_generation, media_store, embeddings, rerank
 from app.inbound_limits import anthropic_path
 from app.client_profiles import (CLI_VERSION, CLI_USER_AGENT, credential_headers, credential_refresh_headers,
-                                 catalog_cache_key, account_key, canonical_uid, enterprise_id_of, identity_email,
-                                 is_enterprise_account, normalize_uid)
+                                 catalog_cache_key, account_key, canonical_uid, enterprise_id_of,
+                                 is_enterprise_account)
 try:
     from app import credits as credits_mod
 except ImportError:  # Disable credit maintenance when its module is unavailable.
@@ -1924,6 +1925,8 @@ def _reconcile_loop(pool: CredentialPool):
 _endpoint_limits: dict = {}
 _ENDPOINT_LIMIT_LOCK = threading.Lock()  # Workers initialize semaphores concurrently.
 _published_catalog_signature: str | None = None  # Last published catalog content; skips no-op bumps.
+_catalog_sequence_lock = threading.Lock()
+_catalog_sequences: dict[str, int] = {}
 
 
 def _housekeep_concurrency() -> int:
@@ -2270,6 +2273,13 @@ def _account_catalog_key(profile, headers):
                                                   headers.get("X-Enterprise-Id")))
 
 
+def _next_catalog_sequence(key: str) -> int:
+    with _catalog_sequence_lock:
+        sequence = _catalog_sequences.get(key, 0) + 1
+        _catalog_sequences[key] = sequence
+        return sequence
+
+
 def _catalog_model_signature(model: dict, region: str):
     """Return the conservative signature used for automatic same-cost variant grouping."""
     if not isinstance(model, dict) or model.get("supportsToolCall") is not True:
@@ -2331,23 +2341,28 @@ def _build_catalog_route_groups(accounts: dict) -> tuple[dict, dict]:
 
 
 def _fetch_catalog(headers):
-    """Fetch both catalog scopes for one account token; the only upstream call involved."""
+    """Fetch both catalog scopes and report whether personal/v3 supplementation completed."""
     with _endpoint_semaphore("models"):
         scopes = credits_mod.fetch_model_scopes(
             _bearer_token(headers), domain=headers.get("X-Domain", ""),
             uid=headers.get("X-User-Id", ""), enterprise_id=headers.get("X-Enterprise-Id", ""),
             prefer_personal=True)
-    return scopes["picker"], scopes["account"]
+    return scopes["picker"], scopes["account"], getattr(scopes, "complete", True), getattr(scopes, "error", None)
 
 
-def _store_catalog(pool, cache, entry, generation, key, models, serves):
-    """Publish one fetched catalog only while the credential lease is still current."""
+def _store_catalog(pool, cache, entry, generation, key, models, serves, *, complete=True, sequence=None):
+    """Publish a catalog only while the credential lease and fetch sequence are current."""
+    stored = False
     def publish():
-        cache.put(key, models, serves=serves)
-        for current in pool._entries:
-            if current["cm"] is entry["cm"]:
-                current["catalog_dirty"] = False
-    return pool.apply_if_current(entry["cm"], generation, publish)
+        nonlocal stored
+        stored = cache.put(key, models, serves=serves, complete=complete, sequence=sequence)
+        if stored:
+            for current in pool._entries:
+                if current["cm"] is entry["cm"]:
+                    current["catalog_dirty"] = not complete
+    if not pool.apply_if_current(entry["cm"], generation, publish):
+        return False
+    return stored
 
 
 def _sync_one_account_catalog(pool, ledger, entry, generation, headers, profile, failed, cache):
@@ -2357,13 +2372,19 @@ def _sync_one_account_catalog(pool, ledger, entry, generation, headers, profile,
     key = _account_catalog_key(profile, headers)
     if cache.fresh(key) and not entry.get("catalog_dirty"):
         return None
+    sequence = _next_catalog_sequence(key)
     perf_metrics.inc("catalog_fetch_count")
     try:
-        models, serves = _fetch_catalog(headers)
-        if _store_catalog(pool, cache, entry, generation, key, models, serves):
+        models, serves, complete, supplemental_error = _fetch_catalog(headers)
+        if supplemental_error is not None:
+            _log(f"[models] {profile} 补充目录失败（保留个人目录）：{_network_error_text(supplemental_error)}")
+        if _store_catalog(pool, cache, entry, generation, key, models, serves,
+                          complete=complete, sequence=sequence):
             _log(f"[models] {profile} 模型表已刷新: 选择器 {len(models)} 个，"
-                 f"账号根表 {len(serves)} 个")
-            return True
+                 f"账号根表 {len(serves)} 个" + ("（待补充 v3）" if not complete else ""))
+            if not complete:
+                failed.add(entry["id"])
+            return complete
         failed.add(entry["id"])
     except Exception as error:
         perf_metrics.inc("catalog_fetch_failures")
@@ -2463,10 +2484,18 @@ def _probe_profile_catalog(pool, ledger, refs, failed, cache, aggregate, profile
         return
     entry, generation, headers = target["entry"], target["generation"], target["headers"]
     key = _account_catalog_key(profile, headers)
+    sequence = _next_catalog_sequence(key)
     perf_metrics.inc("catalog_fetch_count")
     try:
-        models, serves = _fetch_catalog(headers)
-        if not _store_catalog(pool, cache, entry, generation, key, models, serves):
+        models, serves, complete, supplemental_error = _fetch_catalog(headers)
+        if supplemental_error is not None:
+            _log(f"[models] {profile} 补充目录失败（保留个人目录）：{_network_error_text(supplemental_error)}")
+        if not _store_catalog(pool, cache, entry, generation, key, models, serves,
+                              complete=complete, sequence=sequence):
+            aggregate.failure(profile, source=entry.get("account_key"))
+            failed.add(entry["id"])
+            return
+        if not complete:
             aggregate.failure(profile, source=entry.get("account_key"))
             failed.add(entry["id"])
             return
@@ -2538,10 +2567,17 @@ def _scan_one_catalog(pool, cache, entry):
         generation = cm._generation
         profile = profile_for_headers(headers)
     key = _account_catalog_key(profile, headers)
+    sequence = _next_catalog_sequence(key)
     previous = {item.get("id") for item in (cache.models(key) or [])}
-    models, serves = _fetch_catalog(headers)
-    if not _store_catalog(pool, cache, entry, generation, key, models, serves):
+    models, serves, complete, supplemental_error = _fetch_catalog(headers)
+    if supplemental_error is not None:
+        _log(f"[models] {profile} 补充目录失败（保留个人目录）：{_network_error_text(supplemental_error)}")
+    if not _store_catalog(pool, cache, entry, generation, key, models, serves,
+                          complete=complete, sequence=sequence):
         raise RuntimeError("扫描期间凭证已变化，已跳过该账号")
+    if not complete:
+        # The personal catalog is stored, but the scan must report the missing v3 supplement.
+        raise RuntimeError(f"补充目录失败: {_network_error_text(supplemental_error)}")
     current = {item.get("id") for item in models}
     # A first fetch has no baseline to diff against; only a change to a known catalog counts.
     differences = len(previous ^ current) if previous else 0
@@ -4090,7 +4126,7 @@ def _guard_request_size(body: dict) -> int:
                 raise HTTPException(status_code=413, detail={"error": {
                     "message": f"处理后的请求体超过网关上限 {limit} 字节，请缩短历史或压缩图片",
                     "type": "invalid_request_error", "code": "request_too_large", "max_bytes": limit}})
-    except (ValueError, UnicodeError) as error:
+    except (ValueError, UnicodeError):
         raise HTTPException(status_code=400, detail={"error": {
             "message": "请求体包含无法序列化的 JSON 值", "type": "invalid_request_error"}}) from None
     return size
@@ -7454,7 +7490,7 @@ def main():
     sys.stderr.write("按 Ctrl+C 退出。\n\n")
 
     # Record service startup.
-    _log(f"==== converter 启动 ====")
+    _log("==== converter 启动 ====")
 
     CONFIG["launch"] = launch
     try: