"""CPA (CLIProxyAPI) 认证文件同步 - 保持本地 codex 认证文件与 CPA 一致""" import base64 import json import logging import time from datetime import datetime from hashlib import md5 from pathlib import Path import requests from autoteam.auth_storage import AUTH_DIR, ensure_auth_dir, ensure_auth_file_permissions from autoteam.config import AUTO_CHECK_TARGET_SEATS, CPA_KEY, CPA_URL from autoteam.textio import write_text logger = logging.getLogger(__name__) def _headers(): return {"Authorization": f"Bearer {CPA_KEY}"} def list_cpa_files(): """获取 CPA 中所有认证文件。远端异常必须显式失败，不能伪装成空列表。""" try: resp = requests.get(f"{CPA_URL}/v0/management/auth-files", headers=_headers(), timeout=10) except requests.RequestException as exc: logger.error("[CPA] 获取文件列表失败: %s", exc) raise RuntimeError(f"[CPA] auth-files list request failed: {exc}") from exc if resp.status_code != 200: logger.error("[CPA] 获取文件列表失败: %d %s", resp.status_code, resp.text[:200]) raise RuntimeError(f"[CPA] auth-files list failed: HTTP {resp.status_code}") try: data = resp.json() except ValueError as exc: logger.error("[CPA] 获取文件列表返回非 JSON 内容: %s", resp.text[:200]) raise RuntimeError("[CPA] auth-files list returned non-JSON response") from exc files = data.get("files", []) if not isinstance(files, list): raise RuntimeError("[CPA] auth-files list response missing files list") return files def upload_to_cpa(filepath): """上传认证文件到 CPA""" filepath = Path(filepath) if not filepath.exists(): logger.warning("[CPA] 文件不存在: %s", filepath) return False with open(filepath, "rb") as f: resp = requests.post( f"{CPA_URL}/v0/management/auth-files", headers=_headers(), files={"file": (filepath.name, f, "application/json")}, timeout=10, ) if resp.status_code == 200: logger.info("[CPA] 已上传: %s", filepath.name) return True else: logger.error("[CPA] 上传失败: %d %s", resp.status_code, resp.text[:200]) return False def delete_from_cpa(name): """从 CPA 删除认证文件""" resp = requests.delete( f"{CPA_URL}/v0/management/auth-files", headers=_headers(), params={"name": name}, timeout=10, ) if resp.status_code == 200: logger.info("[CPA] 已删除: %s", name) return True else: logger.error("[CPA] 删除失败: %d %s", resp.status_code, resp.text[:200]) return False def download_from_cpa(name): """从 CPA 下载认证文件内容。""" resp = requests.get( f"{CPA_URL}/v0/management/auth-files/download", headers=_headers(), params={"name": name}, timeout=10, ) if resp.status_code == 200: return resp.text logger.error("[CPA] 下载失败: %s -> %d %s", name, resp.status_code, resp.text[:200]) return None def _parse_expired_timestamp(value): if isinstance(value, (int, float)): return float(value) if not value: return time.time() + 3600 text = str(value).strip() try: if text.endswith("Z"): return datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp() return datetime.fromisoformat(text).timestamp() except Exception: return time.time() + 3600 def _parse_optional_timestamp(value): if isinstance(value, (int, float)): return float(value) if not value: return 0.0 text = str(value).strip() try: if text.endswith("Z"): return datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp() return datetime.fromisoformat(text).timestamp() except Exception: return 0.0 def _parse_jwt_payload(token): parts = token.split(".") if len(parts) < 2: return {} payload = parts[1] payload += "=" * (-len(payload) % 4) try: return json.loads(base64.urlsafe_b64decode(payload)) except Exception: return {} def _bundle_from_auth_data(auth_data, fallback_name=""): id_token = auth_data.get("id_token", "") claims = _parse_jwt_payload(id_token) if id_token else {} auth_claims = claims.get("https://api.openai.com/auth", {}) if isinstance(claims, dict) else {} plan_type = auth_claims.get("chatgpt_plan_type", "") if not plan_type and "-team" in fallback_name: plan_type = "team" if not plan_type and "-plus" in fallback_name: plan_type = "plus" if not plan_type and "-free" in fallback_name: plan_type = "free" if not plan_type: plan_type = "unknown" return { "id_token": id_token, "access_token": auth_data.get("access_token", ""), "refresh_token": auth_data.get("refresh_token", ""), "account_id": auth_data.get("account_id", ""), "email": auth_data.get("email", ""), "plan_type": plan_type, "expired": _parse_expired_timestamp(auth_data.get("expired")), "last_refresh_ts": _parse_optional_timestamp(auth_data.get("last_refresh")), } def _refresh_account_proxy_url_for_upload(acc: dict, path: Path) -> None: email = str(acc.get("email") or "").strip().lower() if not email or path.name.startswith("codex-main-"): return try: from autoteam.admin_state import get_admin_email if email == (get_admin_email() or "").strip().lower(): return except Exception: pass required = False try: from autoteam import config as runtime_config from autoteam.ipv6_pool import ipv6_pool required = bool(getattr(runtime_config, "AUTOTEAM_IPV6_POOL_REQUIRED", False)) proxy_url = ipv6_pool.ensure(email) or "" if not proxy_url: if required: raise RuntimeError("IPv6 pool is required but no account proxy was assigned") return try: auth_data = json.loads(path.read_text(encoding="utf-8")) except Exception as exc: if required: raise RuntimeError(f"cannot read auth file for required IPv6 proxy: {path.name}") from exc logger.warning("[CPA] 无法读取 auth_file 以刷新 proxy_url，继续原样上传: %s (%s)", path.name, exc) return if auth_data.get("proxy_url") == proxy_url: return auth_data["proxy_url"] = proxy_url write_text(path, json.dumps(auth_data, indent=2, ensure_ascii=False)) logger.info("[CPA] 已刷新待同步凭证 proxy_url: %s", email) except Exception as exc: if required: logger.error("[CPA] IPv6 proxy_url 为必需但刷新失败: %s (%s)", email, exc) raise logger.warning("[CPA] IPv6 proxy_url 刷新失败，继续上传原凭证: %s (%s)", email, exc) def _active_auth_publish_decision(acc: dict, path: Path) -> str: """Return publish/delete_remote/keep_remote for a local active Codex auth file.""" if not path.exists(): return "delete_remote" try: auth_data = json.loads(path.read_text(encoding="utf-8")) except Exception as exc: logger.warning("[CPA] 跳过无法读取的 active 凭证 %s: %s", path.name, exc) return "delete_remote" access_token = auth_data.get("access_token") if not access_token: logger.warning("[CPA] 跳过缺少 access_token 的 active 凭证: %s", path.name) return "delete_remote" try: from autoteam.codex_auth import check_codex_quota quota_status, _info = check_codex_quota(access_token, timeout=8) except Exception as exc: logger.warning("[CPA] active 凭证实时验证异常，保留远端副本等待下轮: %s (%s)", path.name, exc) return "keep_remote" if quota_status == "network_error": logger.warning("[CPA] active 凭证实时验证网络异常，保留远端副本等待下轮: %s", path.name) return "keep_remote" if quota_status == "ok": try: from autoteam.ipv6_pool import ipv6_pool email = (acc.get("email") or auth_data.get("email") or "").strip().lower() proxy_url = ipv6_pool.ensure(email) or "" if proxy_url and auth_data.get("proxy_url") != proxy_url: auth_data["proxy_url"] = proxy_url write_text(path, json.dumps(auth_data, indent=2, ensure_ascii=False)) logger.info("[CPA] 已刷新 active 凭证 proxy_url: %s", email) except Exception as exc: logger.warning("[CPA] active 凭证 IPv6 proxy_url 刷新失败，继续上传原凭证: %s", exc) return "publish" logger.warning("[CPA] 跳过实时验证失败的 active 凭证: %s (%s)", path.name, quota_status) return "delete_remote" def _normalized_auth_path(bundle, main=False): email = bundle.get("email", "") account_id = bundle.get("account_id", "") if main: suffix = account_id or md5(email.encode()).hexdigest()[:8] return AUTH_DIR / f"codex-main-{suffix}.json" plan_type = bundle.get("plan_type", "unknown") hash_id = md5(account_id.encode()).hexdigest()[:8] if account_id else "unknown" return AUTH_DIR / f"codex-{email}-{plan_type}-{hash_id}.json" def _auth_identity(bundle, main=False): if main: return ("main", bundle.get("account_id") or bundle.get("email") or "") return ("codex", (bundle.get("email") or "").lower(), bundle.get("account_id") or "") def _candidate_score(auth_data, bundle, name, main=False): canonical_name = _normalized_auth_path(bundle, main=main).name return ( 1 if name == canonical_name else 0, bundle.get("last_refresh_ts", _parse_optional_timestamp(auth_data.get("last_refresh"))), _parse_expired_timestamp(auth_data.get("expired")), len(auth_data.get("refresh_token") or ""), ) def _write_auth_file(filepath, bundle): ensure_auth_dir() auth_data = { "type": "codex", "id_token": bundle.get("id_token", ""), "access_token": bundle.get("access_token", ""), "refresh_token": bundle.get("refresh_token", ""), "account_id": bundle.get("account_id", ""), "email": bundle.get("email", ""), "expired": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(bundle.get("expired", 0))), "last_refresh": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(bundle.get("last_refresh_ts", time.time()))), } write_text(filepath, json.dumps(auth_data, indent=2)) ensure_auth_file_permissions(filepath) return filepath def _save_normalized_auth_file(bundle, main=False): filepath = _normalized_auth_path(bundle, main=main) if main: for old in AUTH_DIR.glob("codex-main-*.json"): if old != filepath and old.exists(): old.unlink() else: email = bundle.get("email", "") for old in AUTH_DIR.glob(f"codex-{email}-*.json"): if old != filepath and old.exists(): old.unlink() return _write_auth_file(filepath, bundle) def _load_local_best_candidate(identity_key): """读取本地同 identity 的最佳候选认证文件。""" best = None for path in AUTH_DIR.glob("codex-*.json"): if not path.is_file(): continue try: auth_data = json.loads(path.read_text(encoding="utf-8")) except Exception: continue if auth_data.get("type") != "codex": continue main = path.name.startswith("codex-main-") bundle = _bundle_from_auth_data(auth_data, fallback_name=path.name) if _auth_identity(bundle, main=main) != identity_key: continue candidate = { "path": path, "auth_data": auth_data, "bundle": bundle, "main": main, } if best is None or _candidate_score( candidate["auth_data"], candidate["bundle"], candidate["path"].name, candidate["main"] ) > _candidate_score(best["auth_data"], best["bundle"], best["path"].name, best["main"]): best = candidate return best def _cleanup_local_duplicates(accounts=None): """清理本地同账号重复认证文件，只保留一个规范文件。""" grouped = {} for path in AUTH_DIR.glob("codex-*.json"): if not path.is_file(): continue try: auth_data = json.loads(path.read_text(encoding="utf-8")) except Exception: continue if auth_data.get("type") != "codex": continue main = path.name.startswith("codex-main-") bundle = _bundle_from_auth_data(auth_data, fallback_name=path.name) key = _auth_identity(bundle, main=main) grouped.setdefault(key, []).append( { "path": path, "auth_data": auth_data, "bundle": bundle, "main": main, } ) canonical_map = {} removed = 0 for items in grouped.values(): if not items: continue winner = max( items, key=lambda item: _candidate_score(item["auth_data"], item["bundle"], item["path"].name, item["main"]) ) canonical_path = Path(_save_normalized_auth_file(winner["bundle"], main=winner["main"])) canonical_map[_auth_identity(winner["bundle"], main=winner["main"])] = canonical_path for item in items: if item["path"] != canonical_path and item["path"].exists(): item["path"].unlink() removed += 1 if accounts is not None: changed = False for acc in accounts: auth_path = acc.get("auth_file") if not auth_path: continue try: path = Path(auth_path) if not path.exists(): continue auth_data = json.loads(path.read_text(encoding="utf-8")) except Exception: continue bundle = _bundle_from_auth_data(auth_data, fallback_name=path.name) canonical_path = canonical_map.get(_auth_identity(bundle, main=False)) if canonical_path and acc.get("auth_file") != str(canonical_path.resolve()): acc["auth_file"] = str(canonical_path.resolve()) changed = True return removed, changed return removed, False def sync_from_cpa(): """ 从 CPA 反向同步认证文件到本地。 规则： - 下载 CPA 中所有 codex 认证文件到本地 auths/ - 非主号文件会导入/修复到 accounts.json，默认状态为 standby（保守导入） - 不删除本地账号记录，仅补充/更新 auth_file """ from autoteam.accounts import ( STATUS_STANDBY, add_account, find_account, load_accounts, save_accounts, update_account, ) from autoteam.mail import infer_mail_provider_from_email AUTH_DIR.mkdir(exist_ok=True) accounts = load_accounts() changed_accounts = False imported_files = 0 updated_files = 0 added_accounts = 0 updated_accounts = 0 skipped = 0 cpa_duplicates_deleted = 0 local_kept_newer = 0 local_duplicates_deleted, accounts_path_repaired = _cleanup_local_duplicates(accounts) if accounts_path_repaired: save_accounts(accounts) cpa_files = list_cpa_files() if not cpa_files: logger.info("[CPA] 未发现可反向同步的认证文件") return { "downloaded": 0, "updated": 0, "accounts_added": 0, "accounts_updated": 0, "skipped": 0, "cpa_duplicates_deleted": 0, "local_duplicates_deleted": local_duplicates_deleted, "local_kept_newer": 0, "total": 0, } candidates = [] for item in cpa_files: name = (item.get("name") or "").strip() if not name or not name.endswith(".json") or not name.startswith("codex-"): skipped += 1 continue content = download_from_cpa(name) if not content: skipped += 1 continue try: auth_data = json.loads(content) except Exception: logger.warning("[CPA] 跳过无效 JSON: %s", name) skipped += 1 continue if auth_data.get("type") != "codex": logger.info("[CPA] 跳过非 codex 文件: %s", name) skipped += 1 continue bundle = _bundle_from_auth_data(auth_data, fallback_name=name) email = (bundle.get("email") or item.get("email") or "").lower().strip() bundle["email"] = email if not email and not name.startswith("codex-main-"): logger.info("[CPA] 跳过缺少邮箱的文件: %s", name) continue candidates.append( { "name": name, "auth_data": auth_data, "bundle": bundle, "main": name.startswith("codex-main-"), } ) grouped = {} for item in candidates: grouped.setdefault(_auth_identity(item["bundle"], main=item["main"]), []).append(item) for items in grouped.values(): winner = max( items, key=lambda item: _candidate_score(item["auth_data"], item["bundle"], item["name"], main=item["main"]), ) for item in items: if item is winner: continue if delete_from_cpa(item["name"]): cpa_duplicates_deleted += 1 name = winner["name"] bundle = winner["bundle"] email = bundle.get("email", "") identity_key = _auth_identity(bundle, main=winner["main"]) local_best = _load_local_best_candidate(identity_key) cpa_score = _candidate_score(winner["auth_data"], bundle, name, main=winner["main"]) local_score = None if local_best: local_score = _candidate_score( local_best["auth_data"], local_best["bundle"], local_best["path"].name, main=local_best["main"] ) if winner["main"]: if local_best and local_score >= cpa_score: local_kept_newer += 1 normalized_path = local_best["path"] else: normalized_path = _normalized_auth_path(bundle, main=True) existed = normalized_path.exists() previous = None if existed: try: previous = normalized_path.read_text(encoding="utf-8") except Exception: previous = None normalized_path = Path(_save_normalized_auth_file(bundle, main=True)) current = normalized_path.read_text(encoding="utf-8") if not existed: imported_files += 1 elif previous != current: updated_files += 1 if normalized_path.name != name: old_path = AUTH_DIR / name if old_path.exists() and old_path != normalized_path: old_path.unlink() continue if local_best and local_score >= cpa_score: local_kept_newer += 1 normalized_path = local_best["path"] else: normalized_path = _normalized_auth_path(bundle) existed = normalized_path.exists() previous = None if existed: try: previous = normalized_path.read_text(encoding="utf-8") except Exception: previous = None normalized_path = Path(_save_normalized_auth_file(bundle)) current = normalized_path.read_text(encoding="utf-8") if not existed: imported_files += 1 elif previous != current: updated_files += 1 acc = find_account(accounts, email) resolved_path = str(normalized_path.resolve()) inferred_provider = infer_mail_provider_from_email(email) if acc: acc_changed = False if acc.get("auth_file") != resolved_path: acc["auth_file"] = resolved_path acc_changed = True if inferred_provider and not acc.get("mail_provider"): acc["mail_provider"] = inferred_provider acc_changed = True if acc_changed: changed_accounts = True updated_accounts += 1 else: # Round 12 wire-up (M2) — 改用 add_account API 触发 None→PENDING transition, # 紧接着 update_account → STANDBY 一次性带上 auth_file. 这样状态机能落每条 # state_log.jsonl + F2 SSE 推送, 不再静默直 append 旁路. try: add_account(email, "", cloudmail_account_id=None, mail_provider=inferred_provider or None) update_account( email, status=STATUS_STANDBY, auth_file=resolved_path, quota_exhausted_at=None, quota_resets_at=None, _reason="cpa_sync:import_unknown", ) accounts = load_accounts() # 刷新 in-memory snapshot 给后续 _cleanup_local_duplicates 用 except Exception as exc: logger.warning( "[CPA] 反向同步 add_account/transition 抛异常,回退直 append: %s (%s)", email, exc, ) accounts.append( { "email": email, "password": "", "cloudmail_account_id": None, "mail_provider": inferred_provider or "", "status": STATUS_STANDBY, "auth_file": resolved_path, "quota_exhausted_at": None, "quota_resets_at": None, "created_at": time.time(), "last_active_at": None, } ) changed_accounts = True added_accounts += 1 if changed_accounts: save_accounts(accounts) local_duplicates_deleted_after, accounts_path_repaired = _cleanup_local_duplicates(accounts) local_duplicates_deleted += local_duplicates_deleted_after if accounts_path_repaired: save_accounts(accounts) logger.info( "[CPA] 反向同步完成: 新增文件 %d, 更新文件 %d, 新增账号 %d, 更新账号 %d, 保留本地较新 %d, CPA去重 %d, 本地去重 %d, 跳过 %d", imported_files, updated_files, added_accounts, updated_accounts, local_kept_newer, cpa_duplicates_deleted, local_duplicates_deleted, skipped, ) return { "downloaded": imported_files, "updated": updated_files, "accounts_added": added_accounts, "accounts_updated": updated_accounts, "skipped": skipped, "local_kept_newer": local_kept_newer, "cpa_duplicates_deleted": cpa_duplicates_deleted, "local_duplicates_deleted": local_duplicates_deleted, "total": len(cpa_files), } def sync_to_cpa(): """ 同步本地认证文件到 CPA。同步范围：STATUS_ACTIVE（Team 席位）+ STATUS_PERSONAL（免费号）。 - active / personal 有 auth_file → 上传（覆盖） - CPA 有但本地账号状态已不在上述两种（standby / exhausted / pending 等）→ 从 CPA 删除 - 仅清理本地 accounts.json 管理过的邮箱，主号和 CPA 手动上传文件不会被删 """ from autoteam.accounts import STATUS_ACTIVE, STATUS_PERSONAL, is_account_disabled, load_accounts, save_accounts accounts = load_accounts() local_emails = { str(a.get("email") or "").strip().lower() for a in accounts if str(a.get("email") or "").strip() } local_duplicates_deleted, accounts_path_repaired = _cleanup_local_duplicates(accounts) if accounts_path_repaired: save_accounts(accounts) # 修复断裂的 auth_file 路径 changed = False for acc in accounts: auth_path = acc.get("auth_file") if auth_path and not Path(auth_path).exists(): matches = list(AUTH_DIR.glob(f"codex-{acc['email']}-*.json")) if matches: acc["auth_file"] = str(matches[0].resolve()) changed = True if changed: save_accounts(accounts) # 需要同步到 CPA 的账号：active（Team 席位）和 personal（免费号）都要覆盖 # 两种状态在 CPA 端共存但相互隔离：文件名不同、email 域可能不同，Team/Personal 互不干扰 files_to_sync = {} synced_active = 0 synced_personal = 0 disabled_skipped = 0 active_publish_skipped = 0 active_publish_kept_remote = 0 active_publish_delete_remote = 0 for acc in accounts: if is_account_disabled(acc): disabled_skipped += 1 continue status = acc.get("status") if status not in (STATUS_ACTIVE, STATUS_PERSONAL): continue auth_path = acc.get("auth_file") if not auth_path: continue path = Path(auth_path) if not path.exists(): continue if status == STATUS_ACTIVE: decision = _active_auth_publish_decision(acc, path) if decision == "keep_remote": active_publish_kept_remote += 1 continue if decision == "delete_remote": active_publish_delete_remote += 1 continue if decision != "publish": active_publish_skipped += 1 continue _refresh_account_proxy_url_for_upload(acc, path) files_to_sync[path.name] = path if status == STATUS_ACTIVE: synced_active += 1 else: synced_personal += 1 # CPA 认证文件 cpa_files = list_cpa_files() cpa_names = {f["name"]: f for f in cpa_files if f.get("name")} min_active_for_remote_delete = max(1, int(AUTO_CHECK_TARGET_SEATS) - 1) allow_remote_delete = synced_active >= min_active_for_remote_delete try: from autoteam.manager import _is_protected_local_credential_seat except Exception: _is_protected_local_credential_seat = None protected_remote_names = set() protected_remote_emails = set() if _is_protected_local_credential_seat is not None: for acc in accounts: if is_account_disabled(acc): continue email = str(acc.get("email") or "").strip().lower() auth_path_value = acc.get("auth_file") if not email: continue try: if _is_protected_local_credential_seat(acc): protected_remote_emails.add(email) if auth_path_value: protected_remote_names.add(Path(auth_path_value).name) except Exception as exc: logger.warning("[CPA] 判断受保护凭证失败，保留远端副本: %s (%s)", email, exc) protected_remote_emails.add(email) if auth_path_value: protected_remote_names.add(Path(auth_path_value).name) logger.info( "[CPA] 待同步认证文件: %d (Team=%d, Personal=%d), CPA 现有: %d", len(files_to_sync), synced_active, synced_personal, len(cpa_files), ) if not allow_remote_delete: logger.warning( "[CPA] active 凭证不足，跳过本轮远端删除: %d/%d", synced_active, min_active_for_remote_delete, ) # 上传：所有 active + personal 认证文件（覆盖同名文件，确保 token 最新） uploaded = 0 for name, path in files_to_sync.items(): logger.info("[CPA] 上传: %s", name) if upload_to_cpa(path): uploaded += 1 deleted = 0 skipped_remote_delete = 0 skipped_protected = 0 for name, cpa_file in cpa_names.items(): email = cpa_file.get("email", "").lower() if email in local_emails and name not in files_to_sync: if name in protected_remote_names or email in protected_remote_emails: logger.info("[CPA] 保留受保护本地凭证远端副本: %s (%s)", name, email) skipped_protected += 1 continue if not allow_remote_delete: skipped_remote_delete += 1 continue logger.info("[CPA] 删除非 active/personal 文件: %s (%s)", name, email) if delete_from_cpa(name): deleted += 1 if skipped_protected: logger.info("[CPA] 守卫保留 %d 个 CPA 文件(本地仍持有,避免误删 token)", skipped_protected) if skipped_remote_delete: logger.warning("[CPA] 本轮因 active 凭证不足保留远端非 active 文件: %d", skipped_remote_delete) if disabled_skipped: logger.info("[CPA] 跳过 %d 个本地禁用账号", disabled_skipped) logger.info("[CPA] 同步完成: 上传 %d, 删除 %d, 本地去重 %d", uploaded, deleted, local_duplicates_deleted) # 最终状态 final_cpa = list_cpa_files() final_local_managed = [f for f in final_cpa if f.get("email", "").lower() in local_emails] logger.info( "[CPA] CPA 中本地管理: %d, 本地待同步 (Team+Personal): %d", len(final_local_managed), len(files_to_sync), ) return { "ok": True, "uploaded": uploaded, "deleted": deleted, "remote_count_before": len(cpa_files), "remote_managed_after": len(final_local_managed), "synced_active": synced_active, "synced_personal": synced_personal, "disabled_skipped": disabled_skipped, "active_publish": { "skipped_unknown": active_publish_skipped, "kept_remote": active_publish_kept_remote, "delete_remote": active_publish_delete_remote, }, "local_duplicates_deleted": local_duplicates_deleted, "delete_guard": { "allow_remote_delete": allow_remote_delete, "min_active_for_remote_delete": min_active_for_remote_delete, "skipped_remote_delete": skipped_remote_delete, "skipped_protected": skipped_protected, "skipped_protected_delete": skipped_protected, }, } def sync_main_codex_to_cpa(filepath): """同步主号 Codex 认证文件到 CPA。""" filepath = Path(filepath) if not filepath.exists(): raise FileNotFoundError(f"主号认证文件不存在: {filepath}") name = filepath.name existing = {item.get("name"): item for item in list_cpa_files()} for old_name in existing: if old_name and old_name.startswith("codex-main-"): logger.info("[CPA] 删除旧主号文件: %s", old_name) delete_from_cpa(old_name) if not upload_to_cpa(filepath): raise RuntimeError(f"上传主号认证文件失败: {name}") # Round 12 S7 — 主号同步成功后,记录当前 active workspace,便于上游观测 try: active = _get_active_workspace_summary() if active: logger.info( "[CPA] 主号 Codex 已同步 (active workspace: id=%s admin=%s account_id=%s)", active.get("id"), active.get("admin_email"), active.get("account_id"), ) else: logger.info("[CPA] 主号 Codex 已同步: %s", name) except Exception: logger.info("[CPA] 主号 Codex 已同步: %s", name) return {"uploaded": name} def _get_active_workspace_summary(): """Round 12 S7 — return the current active workspace (id/admin/account_id) or None. Best-effort: never raises; pool unavailable returns None. """ try: from autoteam.workspace_pool import default_pool active = default_pool.get_active() if not active: return None return { "id": active.get("id"), "admin_email": active.get("admin_email"), "account_id": active.get("account_id"), } except Exception: return None def get_active_sync_target(): """Round 12 S7 — public helper: which workspace identity should CPA sync to? Returns dict({id, admin_email, account_id}) or None when single-workspace mode (caller falls back to legacy admin_state). """ return _get_active_workspace_summary()