From 485e632412ad9b0acac5466d1b59b3bd1c33cbd5 Mon Sep 17 00:00:00 2001 From: ck <13679491312@189.cn> Date: Mon, 6 Jul 2026 17:17:23 +0800 Subject: [PATCH 1/5] Add image auto-register retry deployment docs --- .gitignore | 3 + README.md | 6 +- docs/deployment.md | 208 ++++++++++++++++++++++++ docs/server-deployment-notes.md | 33 ++++ services/account_service.py | 198 ++++++++++++++++++++-- test/test_account_image_capabilities.py | 98 +++++++++++ 6 files changed, 530 insertions(+), 16 deletions(-) create mode 100644 docs/deployment.md create mode 100644 docs/server-deployment-notes.md diff --git a/.gitignore b/.gitignore index 0187587..12d93ff 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,6 @@ AGENTS.md # Git cache git_cache/ + +# Agent process docs +.ckcoding/ diff --git a/README.md b/README.md index 5dcbf33..5816749 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,11 @@ ## 快速开始 +当前服务器部署信息见： + +- [部署文档](docs/deployment.md) +- [服务器部署说明](docs/server-deployment-notes.md) + 已发布镜像支持 `linux/amd64` 与 `linux/arm64`，在 x86 服务器和 Apple Silicon / ARM Linux 设备上都会自动拉取匹配架构的版本。 ### Docker 运行 @@ -403,4 +408,3 @@ curl http://localhost:8000/v1/responses \ ## 友情链接 - [LINUX DO - 新的理想型社区](https://linux.do/) - diff --git a/docs/deployment.md b/docs/deployment.md new file mode 100644 index 0000000..41875b8 --- /dev/null +++ b/docs/deployment.md @@ -0,0 +1,208 @@ +# 部署文档 + +本文记录当前服务器上的两个 ChatGPT2API 部署实例，以及本仓库对应的部署方式。 + +## 服务器信息 + +| 项 | 值 | +| --- | --- | +| 服务器 IP | `8.220.241.231` | +| SSH 用户 | `root` | +| SSH 密码 | `Chai962464` | +| 管理后台 / API Bearer Key | `Chai962464` | + +## 当前部署矩阵 + +| 端口 | 地址 | 项目 | 服务器目录 | 容器名 | 镜像 | 说明 | +| --- | --- | --- | --- | --- | --- | --- | +| 80 | `http://8.220.241.231` | `https://github.com/RemotePinee/ChatGPT2API` | `/opt/remote-chatgpt2api` | `remote-chatgpt2api` | `remote-chatgpt2api:local` | 本仓库，已加入图片账号失败后自动注册重试 | +| 666 | `http://8.220.241.231:666` | `https://github.com/ckcoding/chatgpt2api` | `/root/chatgpt2api-mod` | `chatgpt2api-mod` | `chatgpt2api:local` | 之前部署的修改版项目 | + +## 本仓库部署实例 + +本仓库部署在服务器 80 端口： + +```text +URL: http://8.220.241.231 +API: http://8.220.241.231/v1 +Web: http://8.220.241.231 +Authorization: Bearer Chai962464 +``` + +服务器目录： + +```bash +cd /opt/remote-chatgpt2api +``` + +启动和重建： + +```bash +docker-compose -f /opt/remote-chatgpt2api/docker-compose.yml build app +docker-compose -f /opt/remote-chatgpt2api/docker-compose.yml up -d app +``` + +查看状态： + +```bash +docker ps --format "table {{.Names}}\t{{.Image}}\t{{.Ports}}\t{{.Status}}" +docker logs -f remote-chatgpt2api +``` + +## 80 端口配置 + +主配置文件： + +```text +/opt/remote-chatgpt2api/config.json +``` + +关键配置： + +```json +{ + "auth-key": "Chai962464", + "base_url": "http://8.220.241.231", + "image_account_concurrency": 3, + "auto_remove_invalid_accounts": true, + "auto_remove_rate_limited_accounts": false +} +``` + +注册邮箱配置文件： + +```text +/opt/remote-chatgpt2api/data/register.json +``` + +邮箱配置： + +```json +{ + "mail": { + "request_timeout": 15, + "wait_timeout": 30, + "wait_interval": 3, + "providers": [ + { + "enable": true, + "type": "gptmail", + "api_key": "sk-72hz1Sy4MB5V", + "domain": [], + "default_domain": "" + } + ] + }, + "proxy": "", + "total": 2, + "threads": 1, + "fixed_password": "", + "mode": "quota", + "target_quota": 10, + "target_available": 10, + "check_interval": 5, + "enabled": true +} +``` + +## 666 端口配置 + +666 是之前部署的 `ckcoding/chatgpt2api` 修改版，不是本仓库。 + +```text +URL: http://8.220.241.231:666 +API: http://8.220.241.231:666/v1 +Web: http://8.220.241.231:666 +Authorization: Bearer Chai962464 +``` + +服务器目录： + +```text +/root/chatgpt2api-mod +``` + +关键配置： + +```json +{ + "auth-key": "Chai962464", + "refresh_account_interval_minute": 5, + "image_account_concurrency": 3, + "image_parallel_generation": true, + "auto_remove_invalid_accounts": true, + "auto_remove_rate_limited_accounts": false +} +``` + +注册邮箱配置： + +```json +{ + "mail": { + "request_timeout": 30, + "wait_timeout": 30, + "wait_interval": 2, + "api_use_register_proxy": true, + "providers": [ + { + "enable": true, + "type": "gptmail", + "api_base": "", + "admin_email": "", + "admin_password": "", + "domain": [], + "subdomain": [], + "email_prefix": "", + "api_key": "sk-72hz1Sy4MB5V", + "default_domain": "" + } + ] + }, + "proxy": "", + "total": 10, + "threads": 1, + "mode": "quota", + "target_quota": 10, + "target_available": 10, + "check_interval": 5, + "enabled": false +} +``` + +## 定时更新 + +服务器 crontab： + +```cron +0 * * * * /opt/remote-chatgpt2api/auto-update.sh +``` + +要求：以后的定时任务只更新 `https://github.com/RemotePinee/ChatGPT2API` 对应的 80 端口部署。 + +`/opt/remote-chatgpt2api/auto-update.sh` 只操作： + +- `/opt/remote-chatgpt2api` +- 容器 `remote-chatgpt2api` +- 镜像 `remote-chatgpt2api:local` +- 端口 `80` + +它不操作 666 的 `/root/chatgpt2api-mod`、`chatgpt2api-mod` 或 `chatgpt2api:local`。 + +## 本次变更说明 + +本仓库 80 端口部署加入了图片账号失败后的自动注册重试： + +- 没有可用图片账号时，不再秒返回 `no available image quota`。 +- 会调用注册机 `openai_register.worker()` 自动注册账号。 +- 注册失败会最多重试 3 次，或受 `register.json.total` 限制。 +- 注册成功且新账号进入号池后，原图片请求继续拿新账号生成。 +- 缺少邮箱 provider 时记录 `自动注册跳过`。 +- 注册补号、成功、失败、跳过、耗尽都会写入账号业务日志。 + +本地验证命令： + +```bash +uv run python -m unittest test.test_account_image_capabilities +uv run python -m compileall services/account_service.py test/test_account_image_capabilities.py +``` diff --git a/docs/server-deployment-notes.md b/docs/server-deployment-notes.md new file mode 100644 index 0000000..579d682 --- /dev/null +++ b/docs/server-deployment-notes.md @@ -0,0 +1,33 @@ +# 服务器部署说明 + +当前服务器 `8.220.241.231` 同时跑两个项目： + +| 项 | 80 端口 | 666 端口 | +| --- | --- | --- | +| URL | `http://8.220.241.231` | `http://8.220.241.231:666` | +| GitHub 项目 | `https://github.com/RemotePinee/ChatGPT2API` | `https://github.com/ckcoding/chatgpt2api` | +| 服务器路径 | `/opt/remote-chatgpt2api` | `/root/chatgpt2api-mod` | +| 容器 | `remote-chatgpt2api` | `chatgpt2api-mod` | +| 镜像 | `remote-chatgpt2api:local` | `chatgpt2api:local` | +| 后台密钥 | `Chai962464` | `Chai962464` | +| 邮箱 provider | `gptmail` | `gptmail` | +| 邮箱 API key | `sk-72hz1Sy4MB5V` | `sk-72hz1Sy4MB5V` | + +SSH 登录： + +```bash +ssh root@8.220.241.231 +# password: Chai962464 +``` + +检查两个容器： + +```bash +docker ps --format "table {{.Names}}\t{{.Image}}\t{{.Ports}}\t{{.Status}}" +``` + +当前约定： + +- 80 端口是 RemotePinee/ChatGPT2API，部署自动注册重试修复。 +- 666 端口是之前修改版 chatgpt2api，继续保留给旧部署使用。 +- 定时任务只更新 80 端口的 RemotePinee/ChatGPT2API。 diff --git a/services/account_service.py b/services/account_service.py index 8acf126..79a4d7a 100644 --- a/services/account_service.py +++ b/services/account_service.py @@ -2,8 +2,9 @@ from concurrent.futures import ThreadPoolExecutor, as_completed import json +import time from threading import Condition, Lock -from typing import Any +from typing import Any, ClassVar from datetime import datetime, timedelta, timezone from services.config import config @@ -18,6 +19,11 @@ class AccountService: """账号池服务，使用 token -> account 的 dict 保存账号。""" + _AUTO_REGISTER_MAX_ATTEMPTS = 3 + _auto_registering: ClassVar[bool] = False + _register_done_condition: ClassVar[Condition] = Condition() + _last_auto_register_result: ClassVar[dict[str, Any]] = {} + def __init__(self, storage_backend: StorageBackend): self.storage = storage_backend self._lock = Lock() @@ -263,6 +269,159 @@ def _list_available_candidate_tokens( if int(self._image_inflight.get(token, 0)) < max_concurrency ] + @staticmethod + def _image_quota_error_message(plan_type: str | None = None, source_type: str | None = None) -> str: + label = plan_type or source_type or "" + if not label: + return "no available image quota" + return f"no available {label} image quota".replace(" ", " ").strip() + + @staticmethod + def _has_register_provider(reg_config: dict) -> bool: + if not isinstance(reg_config, dict): + return False + mail_config = reg_config.get("mail") if isinstance(reg_config.get("mail"), dict) else {} + providers = mail_config.get("providers") if isinstance(mail_config.get("providers"), list) else [] + has_enabled_provider = any( + bool(provider.get("enable")) + for provider in providers + if isinstance(provider, dict) + ) + outlook_config = mail_config.get("outlook") if isinstance(mail_config.get("outlook"), dict) else {} + has_outlook_pools = bool(outlook_config.get("pools")) + return has_enabled_provider or has_outlook_pools + + @classmethod + def _auto_register_attempt_limit(cls, reg_config: dict) -> int: + if not isinstance(reg_config, dict): + return cls._AUTO_REGISTER_MAX_ATTEMPTS + try: + total = int(reg_config.get("total") or cls._AUTO_REGISTER_MAX_ATTEMPTS) + except Exception: + total = cls._AUTO_REGISTER_MAX_ATTEMPTS + return max(0, min(cls._AUTO_REGISTER_MAX_ATTEMPTS, total)) + + @staticmethod + def _auto_register_log_detail( + event: str, + reason: str, + index: int, + plan_type: str | None = None, + source_type: str | None = None, + plan_types: set[str] | tuple[str, ...] | None = None, + ) -> dict[str, Any]: + return { + "event": event, + "reason": reason, + "index": index, + "plan_type": plan_type or "", + "source_type": source_type or "", + "plan_types": sorted(str(item) for item in (plan_types or [])), + } + + def _try_trigger_auto_register( + self, + event: str, + reason: str, + index: int = 0, + plan_type: str | None = None, + source_type: str | None = None, + plan_types: set[str] | tuple[str, ...] | None = None, + ) -> dict[str, Any]: + try: + from services.register import openai_register + except Exception as exc: + detail = self._auto_register_log_detail(event, reason, index, plan_type, source_type, plan_types) + detail["error"] = str(exc)[:500] + log_service.add(LOG_TYPE_ACCOUNT, "自动注册失败", detail) + return {"ok": False, "index": index, "error": str(exc)} + + reg_config = getattr(openai_register, "config", {}) + if not self._has_register_provider(reg_config): + detail = self._auto_register_log_detail(event, reason, index, plan_type, source_type, plan_types) + detail["skipped"] = "no_provider" + log_service.add(LOG_TYPE_ACCOUNT, "自动注册跳过", detail) + return {"ok": False, "index": index, "skipped": "no_provider"} + + with self._register_done_condition: + if self._auto_registering: + while self._auto_registering: + self._register_done_condition.wait(timeout=1.0) + result = dict(self._last_auto_register_result or {}) + result["waited"] = True + return result + AccountService._auto_registering = True + + result: dict[str, Any] = {"ok": False, "index": index} + try: + detail = self._auto_register_log_detail(event, reason, index, plan_type, source_type, plan_types) + log_service.add(LOG_TYPE_ACCOUNT, "自动注册补充", detail) + if hasattr(openai_register, "stats") and hasattr(openai_register, "stats_lock"): + with openai_register.stats_lock: + openai_register.stats["start_time"] = time.time() + worker_result = openai_register.worker(index) + result = dict(worker_result or {}) + ok = bool(result.get("ok")) + detail = self._auto_register_log_detail(event, reason, index, plan_type, source_type, plan_types) + if not ok: + detail["error"] = str(result.get("error") or "register worker failed")[:500] + log_service.add(LOG_TYPE_ACCOUNT, "自动注册成功" if ok else "自动注册失败", detail) + return result + except Exception as exc: + result = {"ok": False, "index": index, "error": str(exc)} + detail = self._auto_register_log_detail(event, reason, index, plan_type, source_type, plan_types) + detail["error"] = str(exc)[:500] + log_service.add(LOG_TYPE_ACCOUNT, "自动注册失败", detail) + return result + finally: + with self._register_done_condition: + AccountService._last_auto_register_result = dict(result) + AccountService._auto_registering = False + self._register_done_condition.notify_all() + + def _try_auto_register_until_image_account( + self, + excluded_tokens: set[str] | None = None, + plan_type: str | None = None, + source_type: str | None = None, + plan_types: set[str] | tuple[str, ...] | None = None, + event: str = "get_available_access_token", + reason: str = "no_available_image_quota", + ) -> bool: + try: + from services.register import openai_register + reg_config = getattr(openai_register, "config", {}) + except Exception: + reg_config = {} + + attempt_limit = self._auto_register_attempt_limit(reg_config) + if attempt_limit <= 0: + detail = self._auto_register_log_detail(event, reason, 0, plan_type, source_type, plan_types) + detail["skipped"] = "attempt_limit" + log_service.add(LOG_TYPE_ACCOUNT, "自动注册跳过", detail) + return False + + for index in range(attempt_limit): + result = self._try_trigger_auto_register( + event, + reason, + index=index, + plan_type=plan_type, + source_type=source_type, + plan_types=plan_types, + ) + with self._image_slot_condition: + if self._list_ready_candidate_tokens(excluded_tokens, plan_type, source_type, plan_types): + self._image_slot_condition.notify_all() + return True + if result.get("skipped") == "no_provider": + return False + + detail = self._auto_register_log_detail(event, reason, attempt_limit, plan_type, source_type, plan_types) + detail["attempts"] = attempt_limit + log_service.add(LOG_TYPE_ACCOUNT, "自动注册补充耗尽", detail) + return False + def _acquire_next_candidate_token( self, excluded_tokens: set[str] | None = None, @@ -270,20 +429,28 @@ def _acquire_next_candidate_token( source_type: str | None = None, plan_types: set[str] | tuple[str, ...] | None = None, ) -> str: - with self._image_slot_condition: - while True: - if not self._list_ready_candidate_tokens(excluded_tokens, plan_type, source_type, plan_types): - raise RuntimeError( - f"no available {plan_type or source_type or ''} image quota".replace(" ", " ").strip() - if plan_type or source_type else "no available image quota" - ) - tokens = self._list_available_candidate_tokens(excluded_tokens, plan_type, source_type, plan_types) - if tokens: - access_token = tokens[self._index % len(tokens)] - self._index += 1 - self._image_inflight[access_token] = int(self._image_inflight.get(access_token, 0)) + 1 - return access_token - self._image_slot_condition.wait(timeout=1.0) + while True: + with self._image_slot_condition: + while True: + if not self._list_ready_candidate_tokens(excluded_tokens, plan_type, source_type, plan_types): + break + tokens = self._list_available_candidate_tokens(excluded_tokens, plan_type, source_type, plan_types) + if tokens: + access_token = tokens[self._index % len(tokens)] + self._index += 1 + self._image_inflight[access_token] = int(self._image_inflight.get(access_token, 0)) + 1 + return access_token + self._image_slot_condition.wait(timeout=1.0) + if self._try_auto_register_until_image_account( + excluded_tokens=excluded_tokens, + plan_type=plan_type, + source_type=source_type, + plan_types=plan_types, + event="get_available_access_token", + reason="no_ready_candidate", + ): + continue + raise RuntimeError(self._image_quota_error_message(plan_type, source_type)) def release_image_slot(self, access_token: str) -> None: if not access_token: @@ -517,6 +684,7 @@ def _add_account_payloads(self, payloads: list[dict]) -> dict: if account is not None: self._accounts[access_token] = account self._save_accounts() + self._image_slot_condition.notify_all() items = [dict(item) for item in self._accounts.values()] log_service.add(LOG_TYPE_ACCOUNT, f"新增 {added} 个账号，跳过 {skipped} 个", {"added": added, "skipped": skipped}) diff --git a/test/test_account_image_capabilities.py b/test/test_account_image_capabilities.py index 5c7c4cd..26e34d3 100644 --- a/test/test_account_image_capabilities.py +++ b/test/test_account_image_capabilities.py @@ -5,18 +5,28 @@ import unittest from datetime import datetime, timedelta, timezone from pathlib import Path +from unittest.mock import patch os.environ.setdefault("CHATGPT2API_AUTH_KEY", "test-auth") from services.account_service import AccountService from services.auth_service import AuthService from services.cpa_service import _account_payload_from_auth_file +from services.register import openai_register from services.sub2api_service import _account_payload_from_remote from services.storage.json_storage import JSONStorageBackend from utils.helper import anonymize_token class AccountCapabilityTests(unittest.TestCase): + def _reset_auto_register_state(self) -> None: + condition = getattr(AccountService, "_register_done_condition", None) + if condition is None: + return + with condition: + AccountService._auto_registering = False + AccountService._last_auto_register_result = {} + def test_unknown_quota_accounts_are_available_only_when_not_throttled(self) -> None: self.assertFalse( AccountService._is_image_account_available( @@ -219,6 +229,94 @@ def test_mark_image_rate_limited_uses_codex_reset_headers(self) -> None: self.assertLessEqual(reset_at, after + timedelta(seconds=60)) self.assertFalse(AccountService._is_image_account_available(updated)) + def test_auto_register_retries_until_new_image_account_is_available(self) -> None: + self._reset_auto_register_state() + with tempfile.TemporaryDirectory() as tmp_dir: + service = AccountService(JSONStorageBackend(Path(tmp_dir) / "accounts.json")) + worker_calls: list[int] = [] + + def fake_worker(index: int) -> dict: + worker_calls.append(index) + if len(worker_calls) == 1: + return {"ok": False, "index": index, "error": "temporary register failure"} + service.add_account_items([{ + "access_token": "token-new", + "status": "正常", + "quota": 1, + "type": "free", + }]) + return {"ok": True, "index": index, "result": {"access_token": "token-new"}} + + with patch.dict(openai_register.config, { + "mail": {"providers": [{"type": "mock", "enable": True}]}, + "total": 2, + }, clear=False), \ + patch.object(openai_register, "worker", side_effect=fake_worker), \ + patch.object(service, "fetch_remote_info", side_effect=lambda token, _event: service.get_account(token)): + token = service.get_available_access_token() + + self.assertEqual(token, "token-new") + self.assertEqual(worker_calls, [0, 1]) + + def test_auto_register_logs_skip_when_provider_is_missing(self) -> None: + self._reset_auto_register_state() + with tempfile.TemporaryDirectory() as tmp_dir: + service = AccountService(JSONStorageBackend(Path(tmp_dir) / "accounts.json")) + + with patch.dict(openai_register.config, { + "mail": {"providers": []}, + "total": 2, + }, clear=False), \ + patch.object(openai_register, "worker") as worker, \ + patch("services.account_service.log_service.add") as log_add: + with self.assertRaisesRegex(RuntimeError, "no available image quota"): + service.get_available_access_token() + + worker.assert_not_called() + self.assertTrue( + any(len(call.args) >= 2 and call.args[1] == "自动注册跳过" for call in log_add.call_args_list) + ) + + def test_failed_existing_candidate_triggers_auto_register_retry(self) -> None: + self._reset_auto_register_state() + with tempfile.TemporaryDirectory() as tmp_dir: + service = AccountService(JSONStorageBackend(Path(tmp_dir) / "accounts.json")) + service.add_account_items([{ + "access_token": "token-old", + "status": "正常", + "quota": 1, + "type": "free", + }]) + worker_calls: list[int] = [] + + def fake_worker(index: int) -> dict: + worker_calls.append(index) + service.add_account_items([{ + "access_token": "token-new", + "status": "正常", + "quota": 1, + "type": "free", + }]) + return {"ok": True, "index": index, "result": {"access_token": "token-new"}} + + def fake_fetch(token: str, _event: str) -> dict: + if token == "token-old": + raise RuntimeError("upstream verification failed") + account = service.get_account(token) + assert account is not None + return account + + with patch.dict(openai_register.config, { + "mail": {"providers": [{"type": "mock", "enable": True}]}, + "total": 1, + }, clear=False), \ + patch.object(openai_register, "worker", side_effect=fake_worker), \ + patch.object(service, "fetch_remote_info", side_effect=fake_fetch): + token = service.get_available_access_token() + + self.assertEqual(token, "token-new") + self.assertEqual(worker_calls, [0]) + class TokenLogTests(unittest.TestCase): def test_anonymize_token_hides_raw_value(self) -> None: From f84a8fa95c2dbddb191689df2291116f0fb0cc53 Mon Sep 17 00:00:00 2001 From: ck <13679491312@189.cn> Date: Mon, 6 Jul 2026 17:27:30 +0800 Subject: [PATCH 2/5] Retry text streams after auto-registering invalid tokens --- docs/deployment.md | 5 +- services/account_service.py | 88 ++++++++++++++++++++++--- services/protocol/conversation.py | 29 ++++++-- test/test_account_image_capabilities.py | 49 ++++++++++++++ 4 files changed, 156 insertions(+), 15 deletions(-) diff --git a/docs/deployment.md b/docs/deployment.md index 41875b8..25e068b 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -191,12 +191,13 @@ Authorization: Bearer Chai962464 ## 本次变更说明 -本仓库 80 端口部署加入了图片账号失败后的自动注册重试： +本仓库 80 端口部署加入了图片账号失败和文本 token 失效后的自动注册重试： - 没有可用图片账号时，不再秒返回 `no available image quota`。 - 会调用注册机 `openai_register.worker()` 自动注册账号。 - 注册失败会最多重试 3 次，或受 `register.json.total` 限制。 - 注册成功且新账号进入号池后，原图片请求继续拿新账号生成。 +- `/api/chat/stream`、`/v1/chat/completions` 等文本通道遇到 `token_invalidated` / `token_revoked` 且尚未输出内容时，会删除失效账号、同步注册新账号，并用新账号重试当前请求。 - 缺少邮箱 provider 时记录 `自动注册跳过`。 - 注册补号、成功、失败、跳过、耗尽都会写入账号业务日志。 @@ -204,5 +205,5 @@ Authorization: Bearer Chai962464 ```bash uv run python -m unittest test.test_account_image_capabilities -uv run python -m compileall services/account_service.py test/test_account_image_capabilities.py +uv run python -m compileall services/account_service.py services/protocol/conversation.py test/test_account_image_capabilities.py ``` diff --git a/services/account_service.py b/services/account_service.py index 79a4d7a..8527c85 100644 --- a/services/account_service.py +++ b/services/account_service.py @@ -422,6 +422,63 @@ def _try_auto_register_until_image_account( log_service.add(LOG_TYPE_ACCOUNT, "自动注册补充耗尽", detail) return False + def _list_ready_text_candidate_tokens( + self, + excluded_tokens: set[str] | None = None, + plan_type: str | None = None, + plan_types: set[str] | tuple[str, ...] | None = None, + ) -> list[str]: + excluded = set(excluded_tokens or set()) + return [ + token + for account in self._accounts.values() + if account.get("status") not in {"禁用", "异常"} + and self._account_matches_plan_type(account, plan_type) + and self._account_matches_any_plan_type(account, plan_types) + and (token := account.get("access_token") or "") + and token not in excluded + ] + + def _try_auto_register_until_text_account( + self, + excluded_tokens: set[str] | None = None, + plan_type: str | None = None, + plan_types: set[str] | tuple[str, ...] | None = None, + event: str = "get_text_access_token", + reason: str = "no_available_text_account", + ) -> bool: + try: + from services.register import openai_register + reg_config = getattr(openai_register, "config", {}) + except Exception: + reg_config = {} + + attempt_limit = self._auto_register_attempt_limit(reg_config) + if attempt_limit <= 0: + detail = self._auto_register_log_detail(event, reason, 0, plan_type, None, plan_types) + detail["skipped"] = "attempt_limit" + log_service.add(LOG_TYPE_ACCOUNT, "自动注册跳过", detail) + return False + + for index in range(attempt_limit): + result = self._try_trigger_auto_register( + event, + reason, + index=index, + plan_type=plan_type, + plan_types=plan_types, + ) + with self._lock: + if self._list_ready_text_candidate_tokens(excluded_tokens, plan_type, plan_types): + return True + if result.get("skipped") == "no_provider": + return False + + detail = self._auto_register_log_detail(event, reason, attempt_limit, plan_type, None, plan_types) + detail["attempts"] = attempt_limit + log_service.add(LOG_TYPE_ACCOUNT, "自动注册补充耗尽", detail) + return False + def _acquire_next_candidate_token( self, excluded_tokens: set[str] | None = None, @@ -522,21 +579,34 @@ def get_text_access_token( ) -> str: excluded = set(excluded_tokens or set()) with self._lock: - candidates = [ - token - for account in self._accounts.values() - if account.get("status") not in {"禁用", "异常"} - and self._account_matches_plan_type(account, plan_type) - and self._account_matches_any_plan_type(account, plan_types) - and (token := account.get("access_token") or "") - and token not in excluded - ] + candidates = self._list_ready_text_candidate_tokens(excluded, plan_type, plan_types) if not candidates: return "" access_token = candidates[self._index % len(candidates)] self._index += 1 return access_token + def get_text_access_token_with_auto_register( + self, + excluded_tokens: set[str] | None = None, + plan_type: str | None = None, + plan_types: set[str] | tuple[str, ...] | None = None, + event: str = "get_text_access_token", + reason: str = "no_available_text_account", + ) -> str: + token = self.get_text_access_token(excluded_tokens, plan_type, plan_types) + if token: + return token + if not self._try_auto_register_until_text_account( + excluded_tokens=excluded_tokens, + plan_type=plan_type, + plan_types=plan_types, + event=event, + reason=reason, + ): + return "" + return self.get_text_access_token(excluded_tokens, plan_type, plan_types) + def mark_text_used(self, access_token: str) -> None: if not access_token: return diff --git a/services/protocol/conversation.py b/services/protocol/conversation.py index 70b2125..753aca6 100644 --- a/services/protocol/conversation.py +++ b/services/protocol/conversation.py @@ -678,7 +678,10 @@ def conversation_events( def text_backend() -> OpenAIBackendAPI: - return OpenAIBackendAPI(access_token=account_service.get_text_access_token()) + return OpenAIBackendAPI(access_token=account_service.get_text_access_token_with_auto_register( + event="text_backend", + reason="no_text_account_before_request", + )) def stream_text_deltas(backend: OpenAIBackendAPI, request: ConversationRequest) -> Iterator[str]: @@ -705,7 +708,11 @@ def stream_text_deltas(backend: OpenAIBackendAPI, request: ConversationRequest) error_message = str(exc) if token and not emitted and is_token_invalid_error(error_message): account_service.remove_invalid_token(token, "text_stream") - token = account_service.get_text_access_token(attempted_tokens) + token = account_service.get_text_access_token_with_auto_register( + attempted_tokens, + event="text_stream", + reason="token_invalidated_before_text_output", + ) if token: continue raise @@ -1157,9 +1164,17 @@ def stream_chat_events( ): token = preferred_token if not token: - token = account_service.get_text_access_token(attempted, plan_type=plan_type, plan_types=plan_types) + token = account_service.get_text_access_token_with_auto_register( + attempted, + plan_type=plan_type, + plan_types=plan_types, + event="chat_stream", + reason="no_text_account_before_chat_stream", + ) if plan_type and not token: raise RuntimeError(f"no available {plan_type} text account") + if not token: + raise RuntimeError("no available text account") emitted = False while True: if token and token in attempted: @@ -1187,7 +1202,13 @@ def stream_chat_events( error_message = str(exc) if token and not emitted and is_token_invalid_error(error_message): account_service.remove_invalid_token(token, "chat_stream") - token = account_service.get_text_access_token(attempted, plan_type=plan_type, plan_types=plan_types) + token = account_service.get_text_access_token_with_auto_register( + attempted, + plan_type=plan_type, + plan_types=plan_types, + event="chat_stream", + reason="token_invalidated_before_chat_output", + ) if plan_type and not token: raise RuntimeError(f"no available {plan_type} text account") if token: diff --git a/test/test_account_image_capabilities.py b/test/test_account_image_capabilities.py index 26e34d3..f3ffa0c 100644 --- a/test/test_account_image_capabilities.py +++ b/test/test_account_image_capabilities.py @@ -12,6 +12,7 @@ from services.account_service import AccountService from services.auth_service import AuthService from services.cpa_service import _account_payload_from_auth_file +from services.protocol import conversation from services.register import openai_register from services.sub2api_service import _account_payload_from_remote from services.storage.json_storage import JSONStorageBackend @@ -317,6 +318,54 @@ def fake_fetch(token: str, _event: str) -> dict: self.assertEqual(token, "token-new") self.assertEqual(worker_calls, [0]) + def test_chat_stream_token_invalidated_auto_registers_and_retries(self) -> None: + self._reset_auto_register_state() + with tempfile.TemporaryDirectory() as tmp_dir: + service = AccountService(JSONStorageBackend(Path(tmp_dir) / "accounts.json")) + service.add_account_items([{ + "access_token": "token-old", + "status": "正常", + "type": "free", + }]) + worker_calls: list[int] = [] + + def fake_worker(index: int) -> dict: + worker_calls.append(index) + service.add_account_items([{ + "access_token": "token-new", + "status": "正常", + "type": "free", + }]) + return {"ok": True, "index": index, "result": {"access_token": "token-new"}} + + class FakeBackend: + def __init__(self, access_token: str = "") -> None: + self.access_token = access_token + + def stream_conversation(self, **_kwargs): + if self.access_token == "token-old": + raise RuntimeError( + "auth_chat_requirements failed: status=401, " + "body={'error': {'code': 'token_invalidated'}}" + ) + return iter([{"type": "conversation.delta", "delta": "ok"}]) + + with patch.dict(openai_register.config, { + "mail": {"providers": [{"type": "mock", "enable": True}]}, + "total": 1, + }, clear=False), \ + patch.object(openai_register, "worker", side_effect=fake_worker), \ + patch.object(conversation, "account_service", service), \ + patch.object(conversation, "OpenAIBackendAPI", FakeBackend), \ + patch.object(conversation, "iter_conversation_payloads", side_effect=lambda payloads, *_args: payloads): + events = list(conversation.stream_chat_events(conversation.ConversationRequest( + model="auto", + messages=[{"role": "user", "content": "你好"}], + ))) + + self.assertEqual(worker_calls, [0]) + self.assertEqual(events, [{"type": "conversation.delta", "delta": "ok", "account_token": "token-new"}]) + class TokenLogTests(unittest.TestCase): def test_anonymize_token_hides_raw_value(self) -> None: From b5008f146f5b614f3f1dac99a426713815dc708d Mon Sep 17 00:00:00 2001 From: ck <13679491312@189.cn> Date: Mon, 6 Jul 2026 17:36:01 +0800 Subject: [PATCH 3/5] Auto-register Anthropic messages text accounts --- services/protocol/anthropic_v1_messages.py | 5 +++- test/test_account_image_capabilities.py | 31 +++++++++++++++++++++- 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/services/protocol/anthropic_v1_messages.py b/services/protocol/anthropic_v1_messages.py index a5113b0..dee7daf 100644 --- a/services/protocol/anthropic_v1_messages.py +++ b/services/protocol/anthropic_v1_messages.py @@ -110,7 +110,10 @@ def preprocess_payload(payload: dict[str, object], text_mapper: Callable[[str], def message_request(body: dict[str, Any]) -> MessageRequest: payload = preprocess_payload(dict(body)) return MessageRequest( - backend=OpenAIBackendAPI(access_token=account_service.get_text_access_token()), + backend=OpenAIBackendAPI(access_token=account_service.get_text_access_token_with_auto_register( + event="anthropic_messages", + reason="no_text_account_before_messages_request", + )), messages=normalize_messages(payload.get("messages"), payload.get("system")), model=str(payload.get("model") or "auto").strip() or "auto", tools=payload.get("tools"), diff --git a/test/test_account_image_capabilities.py b/test/test_account_image_capabilities.py index f3ffa0c..46e10bd 100644 --- a/test/test_account_image_capabilities.py +++ b/test/test_account_image_capabilities.py @@ -12,7 +12,7 @@ from services.account_service import AccountService from services.auth_service import AuthService from services.cpa_service import _account_payload_from_auth_file -from services.protocol import conversation +from services.protocol import anthropic_v1_messages, conversation from services.register import openai_register from services.sub2api_service import _account_payload_from_remote from services.storage.json_storage import JSONStorageBackend @@ -366,6 +366,35 @@ def stream_conversation(self, **_kwargs): self.assertEqual(worker_calls, [0]) self.assertEqual(events, [{"type": "conversation.delta", "delta": "ok", "account_token": "token-new"}]) + def test_anthropic_messages_auto_registers_when_text_account_missing(self) -> None: + self._reset_auto_register_state() + with tempfile.TemporaryDirectory() as tmp_dir: + service = AccountService(JSONStorageBackend(Path(tmp_dir) / "accounts.json")) + worker_calls: list[int] = [] + + def fake_worker(index: int) -> dict: + worker_calls.append(index) + service.add_account_items([{ + "access_token": "token-new", + "status": "正常", + "type": "free", + }]) + return {"ok": True, "index": index, "result": {"access_token": "token-new"}} + + with patch.dict(openai_register.config, { + "mail": {"providers": [{"type": "mock", "enable": True}]}, + "total": 1, + }, clear=False), \ + patch.object(openai_register, "worker", side_effect=fake_worker), \ + patch.object(anthropic_v1_messages, "account_service", service): + request = anthropic_v1_messages.message_request({ + "model": "auto", + "messages": [{"role": "user", "content": "hello"}], + }) + + self.assertEqual(worker_calls, [0]) + self.assertEqual(request.backend.access_token, "token-new") + class TokenLogTests(unittest.TestCase): def test_anonymize_token_hides_raw_value(self) -> None: From 1e44bba85d377924239aa669ab6e76e3e1819aeb Mon Sep 17 00:00:00 2001 From: ck <13679491312@189.cn> Date: Mon, 6 Jul 2026 17:43:33 +0800 Subject: [PATCH 4/5] Cover auto-register protocol entrypoints --- test/test_account_image_capabilities.py | 196 +++++++++++++++++++++++- 1 file changed, 195 insertions(+), 1 deletion(-) diff --git a/test/test_account_image_capabilities.py b/test/test_account_image_capabilities.py index 46e10bd..abef133 100644 --- a/test/test_account_image_capabilities.py +++ b/test/test_account_image_capabilities.py @@ -12,7 +12,7 @@ from services.account_service import AccountService from services.auth_service import AuthService from services.cpa_service import _account_payload_from_auth_file -from services.protocol import anthropic_v1_messages, conversation +from services.protocol import anthropic_v1_messages, conversation, openai_v1_chat_complete, openai_v1_response from services.register import openai_register from services.sub2api_service import _account_payload_from_remote from services.storage.json_storage import JSONStorageBackend @@ -366,6 +366,85 @@ def stream_conversation(self, **_kwargs): self.assertEqual(worker_calls, [0]) self.assertEqual(events, [{"type": "conversation.delta", "delta": "ok", "account_token": "token-new"}]) + def test_openai_chat_completion_auto_registers_when_text_account_missing(self) -> None: + self._reset_auto_register_state() + with tempfile.TemporaryDirectory() as tmp_dir: + service = AccountService(JSONStorageBackend(Path(tmp_dir) / "accounts.json")) + worker_calls: list[int] = [] + + def fake_worker(index: int) -> dict: + worker_calls.append(index) + service.add_account_items([{ + "access_token": "token-new", + "status": "正常", + "type": "free", + }]) + return {"ok": True, "index": index, "result": {"access_token": "token-new"}} + + class FakeBackend: + def __init__(self, access_token: str = "") -> None: + self.access_token = access_token + + def stream_conversation(self, **_kwargs): + return iter([{"type": "conversation.delta", "delta": f"ok:{self.access_token}"}]) + + with patch.dict(openai_register.config, { + "mail": {"providers": [{"type": "mock", "enable": True}]}, + "total": 1, + }, clear=False), \ + patch.object(openai_register, "worker", side_effect=fake_worker), \ + patch.object(conversation, "account_service", service), \ + patch.object(conversation, "OpenAIBackendAPI", FakeBackend), \ + patch.object(conversation, "iter_conversation_payloads", side_effect=lambda payloads, *_args: payloads), \ + patch.object(openai_v1_chat_complete, "count_message_tokens", return_value=1), \ + patch.object(openai_v1_chat_complete, "count_text_tokens", return_value=1): + response = openai_v1_chat_complete.handle({ + "model": "auto", + "messages": [{"role": "user", "content": "hello"}], + }) + + self.assertEqual(worker_calls, [0]) + self.assertEqual(response["choices"][0]["message"]["content"], "ok:token-new") + + def test_openai_responses_auto_registers_when_text_account_missing(self) -> None: + self._reset_auto_register_state() + with tempfile.TemporaryDirectory() as tmp_dir: + service = AccountService(JSONStorageBackend(Path(tmp_dir) / "accounts.json")) + worker_calls: list[int] = [] + + def fake_worker(index: int) -> dict: + worker_calls.append(index) + service.add_account_items([{ + "access_token": "token-new", + "status": "正常", + "type": "free", + }]) + return {"ok": True, "index": index, "result": {"access_token": "token-new"}} + + class FakeBackend: + def __init__(self, access_token: str = "") -> None: + self.access_token = access_token + + def stream_conversation(self, **_kwargs): + return iter([{"type": "conversation.delta", "delta": f"ok:{self.access_token}"}]) + + with patch.dict(openai_register.config, { + "mail": {"providers": [{"type": "mock", "enable": True}]}, + "total": 1, + }, clear=False), \ + patch.object(openai_register, "worker", side_effect=fake_worker), \ + patch.object(conversation, "account_service", service), \ + patch.object(conversation, "OpenAIBackendAPI", FakeBackend), \ + patch.object(conversation, "iter_conversation_payloads", side_effect=lambda payloads, *_args: payloads): + response = openai_v1_response.handle({ + "model": "auto", + "input": "hello", + }) + + self.assertEqual(worker_calls, [0]) + content = response["output"][0]["content"][0]["text"] + self.assertEqual(content, "ok:token-new") + def test_anthropic_messages_auto_registers_when_text_account_missing(self) -> None: self._reset_auto_register_state() with tempfile.TemporaryDirectory() as tmp_dir: @@ -395,6 +474,121 @@ def fake_worker(index: int) -> dict: self.assertEqual(worker_calls, [0]) self.assertEqual(request.backend.access_token, "token-new") + def test_anthropic_messages_token_invalidated_auto_registers_and_retries(self) -> None: + self._reset_auto_register_state() + with tempfile.TemporaryDirectory() as tmp_dir: + service = AccountService(JSONStorageBackend(Path(tmp_dir) / "accounts.json")) + service.add_account_items([{ + "access_token": "token-old", + "status": "正常", + "type": "free", + }]) + worker_calls: list[int] = [] + + def fake_worker(index: int) -> dict: + worker_calls.append(index) + service.add_account_items([{ + "access_token": "token-new", + "status": "正常", + "type": "free", + }]) + return {"ok": True, "index": index, "result": {"access_token": "token-new"}} + + class FakeBackend: + def __init__(self, access_token: str = "") -> None: + self.access_token = access_token + + def stream_conversation(self, **_kwargs): + if self.access_token == "token-old": + raise RuntimeError( + "auth_chat_requirements failed: status=401, " + "body={'error': {'code': 'token_invalidated'}}" + ) + return iter([{"type": "conversation.delta", "delta": "ok"}]) + + with patch.dict(openai_register.config, { + "mail": {"providers": [{"type": "mock", "enable": True}]}, + "total": 1, + }, clear=False), \ + patch.object(openai_register, "worker", side_effect=fake_worker), \ + patch.object(anthropic_v1_messages, "account_service", service), \ + patch.object(anthropic_v1_messages, "OpenAIBackendAPI", FakeBackend), \ + patch.object(conversation, "account_service", service), \ + patch.object(conversation, "OpenAIBackendAPI", FakeBackend), \ + patch.object(conversation, "iter_conversation_payloads", side_effect=lambda payloads, *_args: payloads), \ + patch.object(anthropic_v1_messages, "count_message_tokens", return_value=1), \ + patch.object(anthropic_v1_messages, "count_text_tokens", return_value=1): + response = anthropic_v1_messages.handle({ + "model": "auto", + "messages": [{"role": "user", "content": "hello"}], + }) + + self.assertEqual(worker_calls, [0]) + self.assertEqual(response["content"], [{"type": "text", "text": "ok"}]) + + def test_image_stream_invalid_token_auto_registers_and_retries(self) -> None: + self._reset_auto_register_state() + with tempfile.TemporaryDirectory() as tmp_dir: + service = AccountService(JSONStorageBackend(Path(tmp_dir) / "accounts.json")) + service.add_account_items([{ + "access_token": "token-old", + "status": "正常", + "quota": 1, + "type": "free", + }]) + worker_calls: list[int] = [] + stream_calls: list[str] = [] + + def fake_worker(index: int) -> dict: + worker_calls.append(index) + service.add_account_items([{ + "access_token": "token-new", + "status": "正常", + "quota": 1, + "type": "free", + }]) + return {"ok": True, "index": index, "result": {"access_token": "token-new"}} + + class FakeBackend: + def __init__(self, access_token: str = "") -> None: + self.access_token = access_token + + def fake_stream_image_outputs(backend, request, index=1, total=1): + stream_calls.append(backend.access_token) + if backend.access_token == "token-old": + raise RuntimeError( + "auth_chat_requirements failed: status=401, " + "body={'error': {'code': 'token_invalidated'}}" + ) + yield conversation.ImageOutput( + kind="result", + model=request.model, + index=index, + total=total, + data=[{"url": "http://example.test/image.png"}], + ) + + with patch.dict(openai_register.config, { + "mail": {"providers": [{"type": "mock", "enable": True}]}, + "total": 1, + }, clear=False), \ + patch.object(openai_register, "worker", side_effect=fake_worker), \ + patch.object(service, "fetch_remote_info", side_effect=lambda token, _event: service.get_account(token)), \ + patch.object(conversation, "account_service", service), \ + patch.object(conversation, "OpenAIBackendAPI", FakeBackend), \ + patch.object(conversation, "stream_image_outputs", side_effect=fake_stream_image_outputs): + result = conversation.collect_image_outputs(conversation.stream_image_outputs_with_pool( + conversation.ConversationRequest( + prompt="cat", + model="gpt-image-2", + response_format="url", + ) + )) + + self.assertEqual(worker_calls, [0]) + self.assertEqual(stream_calls, ["token-old", "token-new"]) + self.assertEqual(result["data"], [{"url": "http://example.test/image.png"}]) + class TokenLogTests(unittest.TestCase): def test_anonymize_token_hides_raw_value(self) -> None: From cc9b040d94f065954dc80d41e5698ecac51d63b8 Mon Sep 17 00:00:00 2001 From: ck <13679491312@189.cn> Date: Mon, 6 Jul 2026 17:52:22 +0800 Subject: [PATCH 5/5] Retry image generation when streams return no result --- services/protocol/conversation.py | 20 ++++++-- test/test_account_image_capabilities.py | 62 +++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 3 deletions(-) diff --git a/services/protocol/conversation.py b/services/protocol/conversation.py index 753aca6..946aea3 100644 --- a/services/protocol/conversation.py +++ b/services/protocol/conversation.py @@ -987,6 +987,7 @@ def stream_image_outputs_with_pool(request: ConversationRequest) -> Iterator[Ima raise ImageGenerationError("unsupported image model,supported models: " + ", ".join(sorted(IMAGE_MODELS))) emitted = False + emitted_result = False last_error = "" normalized_resolution = normalize_image_resolution(request.resolution) high_resolution_requested = normalized_resolution in {"2k", "4k"} @@ -996,6 +997,7 @@ def stream_image_outputs_with_pool(request: ConversationRequest) -> Iterator[Ima try: for output in codex_outputs: emitted = True + emitted_result = emitted_result or output.kind == "result" yield output if used_codex: continue @@ -1074,7 +1076,7 @@ def stream_image_outputs_with_pool(request: ConversationRequest) -> Iterator[Ima "token": anonymize_token(token), }) except RuntimeError as exc: - if emitted: + if emitted_result: return raise ImageGenerationError(str(exc) or "image generation failed") from exc @@ -1092,13 +1094,25 @@ def stream_image_outputs_with_pool(request: ConversationRequest) -> Iterator[Ima code="content_policy_violation", ) emitted = True + emitted_result = emitted_result or output.kind == "result" emitted_for_token = True returned_message = output.kind == "message" returned_result = returned_result or output.kind == "result" yield output - if returned_message or not returned_result: + if returned_message: account_service.mark_image_result(token, False) return + if not returned_result: + last_error = "image stream returned no image data" + account_service.mark_image_result(token, False) + logger.warning({ + "event": "image_stream_no_result", + "model": request.model, + "requested_resolution": request.resolution or "", + "requested_size": request.size or "", + "token": anonymize_token(token), + }) + continue account_service.mark_image_result(token, True) break except ImageGenerationError as exc: @@ -1130,7 +1144,7 @@ def stream_image_outputs_with_pool(request: ConversationRequest) -> Iterator[Ima account_service.mark_image_result(token, False) raise ImageGenerationError(image_stream_error_message(last_error)) from exc - if not emitted: + if not emitted_result: raise ImageGenerationError(image_stream_error_message(last_error)) diff --git a/test/test_account_image_capabilities.py b/test/test_account_image_capabilities.py index abef133..686e56c 100644 --- a/test/test_account_image_capabilities.py +++ b/test/test_account_image_capabilities.py @@ -589,6 +589,68 @@ def fake_stream_image_outputs(backend, request, index=1, total=1): self.assertEqual(stream_calls, ["token-old", "token-new"]) self.assertEqual(result["data"], [{"url": "http://example.test/image.png"}]) + def test_image_stream_no_result_auto_registers_and_retries(self) -> None: + self._reset_auto_register_state() + with tempfile.TemporaryDirectory() as tmp_dir: + service = AccountService(JSONStorageBackend(Path(tmp_dir) / "accounts.json")) + service.add_account_items([{ + "access_token": "token-old", + "status": "正常", + "quota": 1, + "type": "free", + }]) + worker_calls: list[int] = [] + stream_calls: list[str] = [] + + def fake_worker(index: int) -> dict: + worker_calls.append(index) + service.add_account_items([{ + "access_token": "token-new", + "status": "正常", + "quota": 1, + "type": "free", + }]) + return {"ok": True, "index": index, "result": {"access_token": "token-new"}} + + class FakeBackend: + def __init__(self, access_token: str = "") -> None: + self.access_token = access_token + + def fake_stream_image_outputs(backend, request, index=1, total=1): + stream_calls.append(backend.access_token) + if backend.access_token == "token-old": + if False: + yield None + return + yield conversation.ImageOutput( + kind="result", + model=request.model, + index=index, + total=total, + data=[{"url": "http://example.test/image.png"}], + ) + + with patch.dict(openai_register.config, { + "mail": {"providers": [{"type": "mock", "enable": True}]}, + "total": 1, + }, clear=False), \ + patch.object(openai_register, "worker", side_effect=fake_worker), \ + patch.object(service, "fetch_remote_info", side_effect=lambda token, _event: service.get_account(token)), \ + patch.object(conversation, "account_service", service), \ + patch.object(conversation, "OpenAIBackendAPI", FakeBackend), \ + patch.object(conversation, "stream_image_outputs", side_effect=fake_stream_image_outputs): + result = conversation.collect_image_outputs(conversation.stream_image_outputs_with_pool( + conversation.ConversationRequest( + prompt="cat", + model="gpt-image-2", + response_format="url", + ) + )) + + self.assertEqual(worker_calls, [0]) + self.assertEqual(stream_calls, ["token-old", "token-new"]) + self.assertEqual(result["data"], [{"url": "http://example.test/image.png"}]) + class TokenLogTests(unittest.TestCase): def test_anonymize_token_hides_raw_value(self) -> None: