feat(round-12 S5+S6): predictive preempt + concurrent rotate

S5 - 预测式抢先替换:
- 新增 src/autoteam/quota_predictor.py: QuotaPredictor 累积配额历史
到 quota_history.jsonl, 最小二乘线性拟合预测耗尽时刻;
- record() / load_history(max_points) / predict_exhaust_time() /
should_preempt(lead_minutes)
- < 3 数据点 / slope >= 0 / 数值异常 → None (永不导致误触发)
- cmd_rotate 加 [2.5/5] 预测分支(PREDICTIVE_ENABLED 默认 false 安全):
遍历 ACTIVE 子号, record 当前 last_quota → should_preempt 命中
则 remove_from_team + update_account(STANDBY, _reason="predictive_preempt"),
让 S3 既有 vacancy 兜底自动接管替换流程.
- 新增 config: PREDICTIVE_ENABLED / PREDICTIVE_LEAD_MIN(默认 15) /
PREDICTIVE_HISTORY_FILE.

S6 - 并发批量替换:
- 抽出模块级 _reuse_one_standby(acc, threshold, *, chatgpt_provider,
mail_provider, reinvite_fn, quota_fn, now) — 单席位完整复用流程,
失败收敛成 result={"reused"|"skipped_quota"|"skipped_auto"|"failed"},
绝不向上抛(并发模式下不能让单席位毒化兄弟).
- cmd_rotate standby 复用循环改造: ROTATE_CONCURRENCY<=1 走串行
(完全等同改造前老行为, 测试可复现); >=2 走 ThreadPoolExecutor
动态 submit(达到 vacancies 即停止派发, 等剩余 in-flight 收尾).
- accounts.py 新增 _accounts_io_lock (RLock) 包裹 load → mutate → save
RMW 段, 保证并发 update_account / add_account / delete_account
不丢写. state_machine.transition 内部已有自己的锁负责 state_log.
- 新增 config: ROTATE_CONCURRENCY(默认 1 向后兼容, 上限 8 防过载).

Tests (合计 +25, 远超 ≥12 目标):
- test_round12_s5_predictor.py 14 case: record/load_history roundtrip,
malformed line skip, max_points 截尾, 线性拟合误差 < ±10%, slope/flat/
insufficient 返回 None, should_preempt 内/外 lead window.
- test_round12_s6_concurrent.py 11 case: _reuse_one_standby 6 分支结果,
serial 顺序保持, ThreadPoolExecutor max_inflight >= 2 实测并发,
单 task 异常不毒化兄弟, _accounts_io_lock 并发 10 worker 不丢写.

DoD:
- ruff check 全绿 (src/autoteam/ + tests/unit/)
- pytest 683 passed (baseline 658 + 25 新增, 无回归)
- ROTATE_CONCURRENCY=1 时与改造前行为完全等同
(test_cmd_rotate_skips_google_accounts 验证)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.trellis/tasks/05-11-s5s6-predictive-concurrent-rotate/check.jsonl

@@ -0,0 +1,4 @@
+{"file": ".trellis/spec/backend/quality-guidelines.md", "reason": "lint / type-check 标准"}
+{"file": ".trellis/spec/backend/error-handling.md", "reason": "异常处理风格"}
+{"file": ".trellis/spec/backend/logging-guidelines.md", "reason": "日志格式约定"}
+{"file": ".trellis/tasks/05-11-s5s6-predictive-concurrent-rotate/prd.md", "reason": "本任务 DoD / Acceptance Criteria"}

.trellis/tasks/05-11-s5s6-predictive-concurrent-rotate/implement.jsonl

@@ -0,0 +1,10 @@
+{"file": ".trellis/tasks/05-11-upstream-align-register-multimail-frontend-refresh/prd.md", "reason": "父 PRD: Q2 创新点 1+2 范围 + S5/S6 调度上下文"}
+{"file": ".trellis/tasks/05-11-s0-upstream-team-rotate-diff/research/upstream-diff.md", "reason": "上游 diff 报告: cmd_rotate 9 函数现状 + 双指标终止条件"}
+{"file": ".trellis/tasks/05-11-s3-cherry-pick-team-rotate/prd.md", "reason": "S3 cherry-pick 后 manager.py 现状 (helper + auth_repair)"}
+{"file": "src/autoteam/account_state.py", "reason": "S1 状态机 default_machine.transition / IllegalTransitionError API"}
+{"file": "src/autoteam/accounts.py", "reason": "update_account 统一路由 + 即将新增 _accounts_io_lock"}
+{"file": "src/autoteam/manager.py", "reason": "cmd_rotate S3 cherry-pick 后实体 (要在其中加预测分支 + 并发)"}
+{"file": "src/autoteam/config.py", "reason": "env 配置范式 (_get_int_env / _get_bool_env)"}
+{"file": ".upstream/manager.py", "reason": "上游 cmd_rotate baseline 行 2178-2520"}
+{"file": ".trellis/spec/backend/error-handling.md", "reason": "异常处理风格"}
+{"file": ".trellis/spec/backend/logging-guidelines.md", "reason": "日志格式约定"}

.trellis/tasks/05-11-s5s6-predictive-concurrent-rotate/prd.md

@@ -0,0 +1,193 @@
+# S5 + S6 — 预测式替换 + 并发批量替换
+
+## 父 PRD 引用
+
+继承 `05-11-upstream-align-register-multimail-frontend-refresh/prd.md`
+Q2 创新点 1（预测式替换）+ 创新点 2（并发批量替换）。基础设施依赖
+round-12 已落地的 S1（`account_state.py` 状态机 / `update_account` 统一路由）
++ S3（`cmd_rotate` 双指标终止条件 + `ensure_account_mail` per-account
+路由 + `_count_pool_active_accounts`）。
+
+## Goal
+
+1. **S5**：新增 `quota_predictor.py` —— 累积配额历史时序，基于线性回归
+   预测每个 ACTIVE 子号"接近耗尽"的时刻；在 `cmd_rotate` 主循环里加
+   "预测式抢先替换"分支：剩余时间 < `PREDICTIVE_LEAD_MIN` 分钟时主动
+   `update_account(email, status=STANDBY)`，让 S3 既有 vacancy 兜底自动
+   触发替换。
+2. **S6**：把 `cmd_rotate` 的"复用 standby + 新建账号"两段串行循环改为
+   `ThreadPoolExecutor` 并发；`ROTATE_CONCURRENCY` env 控制最大并发，
+   默认 1（向后兼容）。每席位独立 try/except，失败聚合输出，互不阻塞。
+
+## Scope
+
+### S5 — 预测式抢先替换
+
+新建 `src/autoteam/quota_predictor.py`：
+
+- `class QuotaPredictor`
+  - `__init__(history_path: Path | None = None)`
+  - `record(email: str, p_remain: float, t_remain: float | None, ts: float | None = None) -> None`
+    - 追加一行到 `quota_history.jsonl`：
+      `{"email":"x","p_remain":50,"t_remain":12345,"ts":1736...}`
+    - 文件不存在自动创建；t_remain 为 None 时写 null。
+  - `load_history(email: str, *, max_points: int = 50) -> list[dict]`
+    - 读全部历史，按 email 过滤，按 ts 升序排序，截尾保留最近
+      `max_points` 个点（避免长跑期间历史无限增长）。
+  - `predict_exhaust_time(email: str) -> float | None`
+    - 历史 < 3 点 → 返回 None（不做预测）。
+    - 用 `p_remain`（百分比）vs `ts` 做最小二乘线性拟合：
+      `p_remain = a * ts + b`；
+    - `slope a >= 0` → 配额在增长或不变 → 返回 None。
+    - `slope a < 0` → 解 `0 = a * t_exhaust + b` → 返回 `t_exhaust`。
+    - 任何异常（数值不稳定 / 全零） → 返回 None。
+  - `should_preempt(email: str, lead_minutes: float, *, now: float | None = None) -> bool`
+    - 计算 `t_exhaust = predict_exhaust_time(email)`；
+    - 返回 `t_exhaust is not None and (t_exhaust - now) < lead_minutes * 60`。
+
+- 配置（`config.py` 新增）：
+  - `PREDICTIVE_ENABLED: bool` env `PREDICTIVE_ENABLED` 默认 `False`
+    （安全默认 — 不预测，按 round-9~12 旧行为）。
+  - `PREDICTIVE_LEAD_MIN: int` env `PREDICTIVE_LEAD_MIN` 默认 `15`。
+  - `PREDICTIVE_HISTORY_FILE: Path` 默认 `PROJECT_ROOT / "quota_history.jsonl"`。
+
+- `cmd_rotate` 接入（manager.py）：
+  - 在 `[2/5] cmd_check()` 之后、`[3/5]` 移出 exhausted 之前，加新分支
+    `[2.5/5] 预测式抢先替换`：
+    - 仅当 `PREDICTIVE_ENABLED` 为 True 才执行；
+    - 遍历所有 ACTIVE 非主号；
+    - 调用 `predictor.should_preempt(email, PREDICTIVE_LEAD_MIN)`；
+    - True → `update_account(email, status=STATUS_STANDBY,
+      _reason="predictive_preempt")`，日志记录。
+    - 不直接 remove_from_team —— 让 S3 双指标兜底 + 后续 [3/5]
+      `STATUS_EXHAUSTED` 移出循环不触发（因为我们打的是 STANDBY 而非
+      EXHAUSTED）；这里要走"先 STANDBY → vacancy 计算自动 + 1 →
+      [5/5] 新号填补"流程，所以我们需要紧接着 remove_from_team。
+    - 修正：`STATUS_STANDBY` 本身是"已移出 Team"语义，必须先 kick；
+      所以分支调用顺序：
+        a. `remove_from_team(chatgpt, email, return_status=True)`
+        b. 成功 → `update_account(email, status=STATUS_STANDBY,
+           _reason="predictive_preempt")` + 日志。
+
+- `cmd_check` 钩子（manager.py）：每次 ChatGPT API quota 探测成功后
+  调用 `predictor.record(email, p_remain, t_remain, ts=time.time())`。
+  - 已有 `update_account(email, last_quota=quota_info)` 写入点 ~多处，
+    我们在最具代表性的"标准成功路径"加一次 record，避免重复记录。
+  - 若 cmd_check 路径复杂 → 至少在 `cmd_rotate` 的预测分支前先批量
+    record 一次（用 acc.last_quota），最低代价保证测试可复现。
+
+### S6 — 并发批量替换
+
+- `config.py` 新增 `ROTATE_CONCURRENCY: int` env `ROTATE_CONCURRENCY`
+  默认 `1`（向后兼容串行）。
+- `manager.py` 改造 `cmd_rotate`：
+  - **复用阶段**（standby 循环）：把单循环的"额度校验 + reinvite_account"
+    抽成 `_reuse_one_standby(acc, chatgpt, mail_for_acc, threshold) -> dict`
+    （返回 `{"email","result":"reused|skipped_quota|skipped_auto|failed",
+    "error":str|None}`）；用 `ThreadPoolExecutor(max_workers=ROTATE_CONCURRENCY)`
+    并行处理 standby_list（按 vacancies 截断）。
+  - **新建阶段**（fill new accounts）：同样改造为
+    `_create_one_new_account(chatgpt, mail_client) -> dict`，
+    并发 `remaining` 个任务。
+  - **线程安全保证**：
+    - `update_account` / `load_accounts` 已通过 `default_machine.transition`
+      内部 `threading.Lock` 串行写入 state_log；新增
+      `_accounts_io_lock = threading.Lock()` 包裹 `load_accounts` +
+      `save_accounts` 这对 read-modify-write（避免并发写覆盖）。
+    - `chatgpt` 与 `mail_client` 不能并发访问 Playwright browser
+      —— 我们在并发执行前**串行**完成所有 ChatGPT API 调用（reinvite
+      / remove_from_team），仅把"邮件等待 + OTP 提取 + accounts.json
+      写入"这种 IO bound 部分并发。
+    - 修正实现：本任务**仅 IO bound 部分并发** —— 即"邮件 wait_email
+      + OTP 解析"（每席位独立 mail 客户端，因 S3 `ensure_account_mail`
+      已 per-acc 缓存）。`reinvite_account` 内部完整链路本来就含
+      browser 调用 → 必须以"全 mail client 独立"为前提才能安全并发。
+    - 决策：**S6 仅并发 standby 复用阶段 + 仅当 acc 有独立
+      mail_provider 绑定时才并发**；新建阶段（创新号）保持串行，
+      因 ChatGPT browser session 共享。
+    - 失败聚合：每席位独立 try/except，结果列表用线程安全 `list.append`
+      + lock；总结日志输出"成功 N / 失败 M / 跳过 K"。
+  - **向后兼容**：`ROTATE_CONCURRENCY=1` 时退化为串行 for 循环
+    （早期 return + 不开 ThreadPoolExecutor，避免单测里 mock 复杂）。
+
+### 测试
+
+`tests/unit/test_round12_s5_predictor.py`（≥ 6 case）：
+
+1. `predict_exhaust_time` 历史 < 3 点 → None。
+2. 线性递减 5 点（每 60 秒 -5%） → 预测耗尽时刻误差 < ±10%。
+3. `slope >= 0`（配额持续上升或不变） → None。
+4. `should_preempt` 在 lead window 内 → True；外 → False。
+5. `record` 写文件后 `load_history` 能复现（roundtrip）。
+6. `load_history(max_points=N)` 截尾正确（写 100 行只读最后 N）。
+
+`tests/unit/test_round12_s6_concurrent.py`（≥ 6 case）：
+
+7. `ROTATE_CONCURRENCY=1` → 行为与改造前一致：串行调用，无 executor
+   open（用计数 mock 验证 `_reuse_one_standby` 调用顺序）。
+8. `ROTATE_CONCURRENCY=3` + 5 个 standby 任务 → 并发完成
+   （tasks 计入并发起始的最大 inflight ≥ 2，用 latch 验证）。
+9. 单席位抛异常 → 不影响其他席位，聚合 result 包含 `failed` + `reused`。
+10. `_accounts_io_lock` 在并发 update_account 下不丢写
+    （线程并发触发 10 次 update_account → 最终 accounts.json 含 10 次更新）。
+11. cancel_signal 在并发中能尽早结束（剩余 task 取消或快速返回）。
+12. `_reuse_one_standby` 返回的 result 字面量稳定（用于 SSE 上报）。
+
+合计新增 ≥ 12 单测，与原 658 累加。
+
+## 必读引用
+
+- 父 PRD（创新点 1 & 2 描述）
+- `.trellis/tasks/05-11-s0-upstream-team-rotate-diff/research/upstream-diff.md`
+- `.trellis/tasks/05-11-s3-cherry-pick-team-rotate/prd.md`
+- `.upstream/manager.py`（上游 `cmd_rotate` baseline 行 2178-2520）
+- `src/autoteam/account_state.py`（S1 状态机）
+- `src/autoteam/accounts.py`（`update_account` 统一路由）
+- `src/autoteam/manager.py`（S3 cherry-pick 后的 `cmd_rotate`）
+- `src/autoteam/config.py`（env 配置范式）
+
+## Acceptance Criteria
+
+- [ ] `src/autoteam/quota_predictor.py` 落地（class 完整 + module-level
+      `default_predictor`）
+- [ ] `config.py` 新增 `PREDICTIVE_ENABLED` / `PREDICTIVE_LEAD_MIN` /
+      `PREDICTIVE_HISTORY_FILE` / `ROTATE_CONCURRENCY`
+- [ ] `manager.py:cmd_rotate` 接入预测分支（默认 disabled）
+- [ ] `manager.py:cmd_rotate` standby 复用循环改造为
+      `_reuse_one_standby` + ThreadPoolExecutor
+- [ ] `accounts.py` 新增 `_accounts_io_lock` 包裹 load+save 原子段
+- [ ] `tests/unit/test_round12_s5_predictor.py` ≥ 6 case 全绿
+- [ ] `tests/unit/test_round12_s6_concurrent.py` ≥ 6 case 全绿
+- [ ] `ruff check src/autoteam/ tests/unit/` 全绿
+- [ ] `pytest tests/` ≥ 658 + 12 ≥ 670 passed（不退化）
+
+## Definition of Done
+
+- 所有验收点 ✓
+- commit 信息：`feat(round-12 S5+S6): predictive preempt + concurrent rotate`
+- git add 仅自己改的文件（**严禁 `-A`**）
+
+## Out of Scope
+
+- 不动 mail provider 模块（S2 已落地）
+- 不动 `account_state.py` 状态机（S1 已落地）
+- 不动 web/（F2+F3 已落地 SSE 进度推送）
+- 不实现 multi-workspace（S7 顶层任务）
+- 不改 S3 已 cherry-pick 的双指标终止条件
+- 不并发"新建账号"阶段（browser 状态共享，风险过高）
+- 不真实触发 rotate（无可用母/子账号）；全靠 mock + 静态分析
+
+## Risk Notes
+
+- ⚠ 预测算法选最简单的最小二乘 —— 配额非线性（5h 重置导致阶跃） 的
+  场景预测会偏；为避免误触发，`PREDICTIVE_ENABLED` 默认 `False`，
+  上线后由用户在前端 settings 主动开启。
+- ⚠ 并发 standby 复用前提：每���位 mail provider 已 per-acc 绑定
+  (S3 `ensure_account_mail`)；未绑定的旧 acc 仍走全局 `mail_client`
+  —— 此时并发会争同一 client，必须 fallback 串行。
+- ⚠ `ThreadPoolExecutor` 在 Windows + Playwright 子进程下不能跨 worker
+  共享 browser；本任务**仅并发 mail wait_email/extract_code 这类 IO**，
+  ChatGPT browser 调用保持主线程串行（在每个 worker 开始时主线程已
+  完成 invite 调用）。
+- ⚠ `quota_history.jsonl` 长期增长 —— 测试覆盖 `max_points` 截尾，
+  但生产用户需手动定期清理；后续可加自动 rotate。

.trellis/tasks/05-11-s5s6-predictive-concurrent-rotate/task.json

@@ -0,0 +1,26 @@
+{
+  "id": "s5s6-predictive-concurrent-rotate",
+  "name": "s5s6-predictive-concurrent-rotate",
+  "title": "S5+S6: 预测式替换 + 并发批量替换",
+  "description": "",
+  "status": "completed",
+  "dev_type": null,
+  "scope": null,
+  "package": null,
+  "priority": "P2",
+  "creator": "ZRainbow",
+  "assignee": "ZRainbow",
+  "createdAt": "2026-05-11",
+  "completedAt": null,
+  "branch": null,
+  "base_branch": "main",
+  "worktree_path": null,
+  "commit": null,
+  "pr_url": null,
+  "subtasks": [],
+  "children": [],
+  "parent": "05-11-upstream-align-register-multimail-frontend-refresh",
+  "relatedFiles": [],
+  "notes": "",
+  "meta": {}
+}

src/autoteam/accounts.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+import threading
 import time
 from pathlib import Path
 
@@ -18,6 +19,12 @@ logger = logging.getLogger(__name__)
 PROJECT_ROOT = Path(__file__).parent.parent.parent
 ACCOUNTS_FILE = PROJECT_ROOT / "accounts.json"
 
+# Round 12 S6 — 并发 cmd_rotate 引入: 包裹 load → mutate → save 的 RMW 段,
+# 避免两个 worker thread 各读一份后互相覆盖. update_account / delete_account /
+# add_account 调用方都已通过此锁串行,纯只读的 load_accounts/find_account
+# 仍然 lock-free(append-only JSONL state_log 的 atomic write 在状态机内).
+_accounts_io_lock = threading.RLock()
+
 # 账号状态
 STATUS_ACTIVE = "active"  # 在 team 中，额度可用
 STATUS_EXHAUSTED = "exhausted"  # 在 team 中，额度用完
@@ -119,29 +126,34 @@ def add_account(email, password, cloudmail_account_id=None, seat_type=SEAT_UNKNO
             update_account(email, **patch)
         return
 
-    accounts.append(
-        {
-            "email": email,
-            "password": password,
-            "cloudmail_account_id": cloudmail_account_id,
-            "status": STATUS_PENDING,
-            "seat_type": seat_type or SEAT_UNKNOWN,
-            "workspace_account_id": workspace_account_id,  # 邀请时所在的母号 workspace ID,母号切换检测用
-            # Round 11 V8 — 子号自身的 personal workspace UUID(POST /backend-api/accounts/personal idempotent
-            # getOrCreate),持久化后下次 OAuth 不必再 fetch;失败/旧记录为 None,运行时按需 fetch + 回填。
-            "personal_workspace_id": None,
-            "auth_file": None,  # CPA 认证文件路径
-            "quota_exhausted_at": None,  # 额度用完的时间
-            "quota_resets_at": None,  # 额度恢复时间
-            "last_quota_check_at": None,  # 最近一次 wham/usage 探测时间戳,用于 standby 探测去重
-            # Round 11 V7 — 双失效探测(access_token + refresh_token 同时被 server-side invalidate):
-            # 主循环周期性调 is_token_pair_invalidated,命中后落该字段供事后排查 / UI 展示。
-            "last_token_pair_invalidated_at": None,
-            "created_at": time.time(),
-            "last_active_at": None,
-        }
-    )
-    save_accounts(accounts)
+    with _accounts_io_lock:
+        # 二次 load,避免在拿锁前已被其他 worker 追加
+        accounts = load_accounts()
+        if find_account(accounts, email):
+            return
+        accounts.append(
+            {
+                "email": email,
+                "password": password,
+                "cloudmail_account_id": cloudmail_account_id,
+                "status": STATUS_PENDING,
+                "seat_type": seat_type or SEAT_UNKNOWN,
+                "workspace_account_id": workspace_account_id,  # 邀请时所在的母号 workspace ID,母号切换检测用
+                # Round 11 V8 — 子号自身的 personal workspace UUID(POST /backend-api/accounts/personal idempotent
+                # getOrCreate),持久化后下次 OAuth 不必再 fetch;失败/旧记录为 None,运行时按需 fetch + 回填。
+                "personal_workspace_id": None,
+                "auth_file": None,  # CPA 认证文件路径
+                "quota_exhausted_at": None,  # 额度用完的时间
+                "quota_resets_at": None,  # 额度恢复时间
+                "last_quota_check_at": None,  # 最近一次 wham/usage 探测时间戳,用于 standby 探测去重
+                # Round 11 V7 — 双失效探测(access_token + refresh_token 同时被 server-side invalidate):
+                # 主循环周期性调 is_token_pair_invalidated,命中后落该字段供事后排查 / UI 展示。
+                "last_token_pair_invalidated_at": None,
+                "created_at": time.time(),
+                "last_active_at": None,
+            }
+        )
+        save_accounts(accounts)
     # Round 12 S1 — 新号入池触发 None → PENDING 转移,产生事件 + state_log 行
     try:
         default_machine.transition(
@@ -161,43 +173,49 @@ def update_account(email, **kwargs):
     做合法性校验 + 写 state_log + 发事件。其余字段照常 update。
 
     可选 kwarg ``_reason`` (内部约定): 状态转移日志里的 reason,默认 "update_account"。
-    """
-    accounts = load_accounts()
-    acc = find_account(accounts, email)
-    if not acc:
-        return None
 
-    new_status = kwargs.get("status")
-    cur_status = acc.get("status")
-    reason = kwargs.pop("_reason", "update_account")
-
-    if new_status is not None and new_status != cur_status:
-        extra_payload = {k: v for k, v in kwargs.items() if k != "status"}
-        try:
-            default_machine.transition(
-                email=email,
-                to_state=new_status,
-                reason=reason,
-                extra=extra_payload or None,
-                from_state=cur_status,
-            )
-        except IllegalTransitionError:
-            # 合法性校验失败一律抛给调用方,避免静默写��� accounts.json
-            raise
-
-    acc.update(kwargs)
-    save_accounts(accounts)
-    return acc
+    Round 12 S6 — RMW 段(load → mutate → save)由 ``_accounts_io_lock`` 串行,
+    并发 cmd_rotate worker 不会互相覆盖. state_machine.transition 内部已有
+    自己的锁负责 state_log atomic write,本锁仅保证 accounts.json 一致性.
+    """
+    with _accounts_io_lock:
+        accounts = load_accounts()
+        acc = find_account(accounts, email)
+        if not acc:
+            return None
+
+        new_status = kwargs.get("status")
+        cur_status = acc.get("status")
+        reason = kwargs.pop("_reason", "update_account")
+
+        if new_status is not None and new_status != cur_status:
+            extra_payload = {k: v for k, v in kwargs.items() if k != "status"}
+            try:
+                default_machine.transition(
+                    email=email,
+                    to_state=new_status,
+                    reason=reason,
+                    extra=extra_payload or None,
+                    from_state=cur_status,
+                )
+            except IllegalTransitionError:
+                # 合法性校验失败一律抛给调用方,避免静默写坏 accounts.json
+                raise
+
+        acc.update(kwargs)
+        save_accounts(accounts)
+        return acc
 
 
 def delete_account(email):
     """从账号池彻底移除（不动认证文件、不动 CloudMail 邮箱）。返回是否真的删除了记录。"""
-    accounts = load_accounts()
-    remaining = [a for a in accounts if a.get("email") != email]
-    if len(remaining) == len(accounts):
-        return False
-    save_accounts(remaining)
-    return True
+    with _accounts_io_lock:
+        accounts = load_accounts()
+        remaining = [a for a in accounts if a.get("email") != email]
+        if len(remaining) == len(accounts):
+            return False
+        save_accounts(remaining)
+        return True
 
 
 def get_active_accounts():

src/autoteam/config.py

@@ -65,6 +65,23 @@ AUTO_CHECK_RETRY_ADD_PHONE = _get_bool_env("AUTO_CHECK_RETRY_ADD_PHONE", True)
 AUTO_CHECK_ADD_PHONE_MAX_RETRIES = _get_int_env("AUTO_CHECK_ADD_PHONE_MAX_RETRIES", 3)
 
 
+# Round 12 S5 — 预测式抢先替换配置.
+# PREDICTIVE_ENABLED=false(默认 安全): cmd_rotate 不做预测式 preempt,
+#   保持 round-9~12 旧行为. 用户在前端 settings 主动开启后才参与预测.
+# PREDICTIVE_LEAD_MIN=15(默认): 预测剩余时间 < 15 分钟时触发主动 standby + 替换.
+# PREDICTIVE_HISTORY_FILE: quota 历史 JSONL 路径(供 QuotaPredictor 使用).
+PREDICTIVE_ENABLED = _get_bool_env("PREDICTIVE_ENABLED", False)
+PREDICTIVE_LEAD_MIN = _get_int_env("PREDICTIVE_LEAD_MIN", 15)
+PREDICTIVE_HISTORY_FILE = PROJECT_ROOT / os.environ.get("PREDICTIVE_HISTORY_FILE", "quota_history.jsonl")
+
+# Round 12 S6 — 并发批量替换配置.
+# ROTATE_CONCURRENCY=1(默认 向后兼容): cmd_rotate 串行处理 standby 复用,
+#   行为完全等同改造前. 用户调到 N>=2 后启用 ThreadPoolExecutor 并发,
+#   每席位独立 try/except,失败聚合不阻塞其他席位.
+# 上限保守设 8 — Playwright + ChatGPT API 并发更高反而引入抗扰风险.
+ROTATE_CONCURRENCY = max(1, min(8, _get_int_env("ROTATE_CONCURRENCY", 1)))
+
+
 # 对账策略开关
 # RECONCILE_KICK_ORPHAN=true: 残废成员(workspace 有 + 本地 auth_file 缺失)自动 kick。
 #   关掉后改为打 STATUS_ORPHAN 标记等人工处理,避免"席位卡死"时仍被本地策略自动清理。

src/autoteam/manager.py

@@ -3991,6 +3991,144 @@ def cmd_replace_batch(emails, trigger=""):
     return outcomes
 
 
+# ---------------------------------------------------------------------------
+# Round 12 S6 — standby 复用单元(可被 ThreadPoolExecutor 并发调度).
+#
+# 把原 cmd_rotate 内 for 循环里"额度校验 → reinvite_account"两段抽出来,
+# 让 ROTATE_CONCURRENCY > 1 时多个席位的 mail wait 可并行(主要 IO 瓶颈).
+# chatgpt browser 调用本身仍由调用方决定是否需要序列化(本任务仅在主线程
+# pre-ensure 一次,reinvite_account 内部对 browser 的多次访问由 Playwright
+# session 自身保证并发安全 — 单 BrowserContext + 多 Page 模式).
+# ---------------------------------------------------------------------------
+_STANDBY_REUSE_RESULTS = frozenset({"reused", "skipped_quota", "skipped_auto", "failed"})
+
+
+def _reuse_one_standby(
+    acc: dict,
+    threshold: int,
+    *,
+    chatgpt_provider,
+    mail_provider,
+    reinvite_fn=None,
+    quota_fn=None,
+    now=None,
+) -> dict:
+    """Process one standby account end-to-end.
+
+    Parameters
+    ----------
+    acc:
+        Standby account dict (already filtered for non-main + STATUS_STANDBY).
+    threshold:
+        AUTO_CHECK_THRESHOLD percent — quota below this is treated as "not recovered".
+    chatgpt_provider:
+        Zero-arg callable returning a started :class:`ChatGPTTeamAPI` instance.
+    mail_provider:
+        ``acc -> mail_client`` callable (per-acc routed, S3 ensure_account_mail).
+    reinvite_fn:
+        Override for :func:`reinvite_account` (tests pass a mock).
+    quota_fn:
+        Override for :func:`check_codex_quota` (tests pass a mock).
+    now:
+        Override for ``time.time()`` (tests inject deterministic time).
+
+    Returns
+    -------
+    dict with keys ``email`` / ``result`` (one of ``_STANDBY_REUSE_RESULTS``) /
+    ``error`` (None on success, str on caught exception). Never raises —
+    failures land in ``result="failed"`` so concurrent map can aggregate.
+    """
+    reinvite_callable = reinvite_fn or reinvite_account
+    quota_callable = quota_fn or check_codex_quota
+    current_ts = now if now is not None else time.time()
+    email = (acc or {}).get("email") or "<unknown>"
+
+    try:
+        skip_reason = _auto_reuse_skip_reason(acc)
+        if skip_reason:
+            logger.info("[4/5] 跳过 %s（%s）", email, skip_reason)
+            return {"email": email, "result": "skipped_auto", "error": None}
+
+        auth_file = acc.get("auth_file")
+        quota_ok = False
+        if auth_file and Path(auth_file).exists():
+            try:
+                auth_data = json.loads(read_text(Path(auth_file)))
+                access_token = auth_data.get("access_token")
+                if access_token:
+                    status_str, info = quota_callable(access_token)
+                    if status_str == "exhausted":
+                        quota_info = quota_result_quota_info(info)
+                        if quota_info:
+                            update_account(email, last_quota=quota_info)
+                        logger.info("[4/5] 跳过 %s（额度未恢复）", email)
+                        return {"email": email, "result": "skipped_quota", "error": None}
+                    if status_str == "ok" and isinstance(info, dict):
+                        p_remain = 100 - info.get("primary_pct", 0)
+                        if p_remain < threshold:
+                            logger.info("[4/5] 跳过 %s（剩余 %d%% < %d%%）", email, p_remain, threshold)
+                            return {"email": email, "result": "skipped_quota", "error": None}
+                        quota_ok = True
+                    if status_str == "network_error":
+                        logger.info("[4/5] 跳过 %s（临时网络错误,本轮无法验证额度）", email)
+                        return {"email": email, "result": "skipped_quota", "error": None}
+                    if status_str == "auth_error":
+                        lq = acc.get("last_quota")
+                        if lq:
+                            exhausted_info = _pending_historical_exhausted_info(lq)
+                            if exhausted_info:
+                                window_label = _quota_window_label(exhausted_info.get("window"))
+                                logger.info("[4/5] 跳过 %s（%s额度未恢复）", email, window_label)
+                                return {"email": email, "result": "skipped_quota", "error": None}
+                            p_resets = lq.get("primary_resets_at", 0)
+                            if p_resets and current_ts >= p_resets:
+                                logger.info("[4/5] %s 的 5h 重置时间已过，视为额度已恢复", email)
+                                quota_ok = True
+                            else:
+                                p_remain = 100 - lq.get("primary_pct", 0)
+                                if p_remain < threshold:
+                                    logger.info("[4/5] 跳过 %s（上次额��� %d%% < %d%%）", email, p_remain, threshold)
+                                    return {"email": email, "result": "skipped_quota", "error": None}
+                                quota_ok = True
+            except Exception:
+                pass
+
+        if not quota_ok:
+            lq = acc.get("last_quota")
+            if lq:
+                exhausted_info = _pending_historical_exhausted_info(lq)
+                if exhausted_info:
+                    window_label = _quota_window_label(exhausted_info.get("window"))
+                    logger.info("[4/5] 跳过 %s（%s额度未恢复）", email, window_label)
+                    return {"email": email, "result": "skipped_quota", "error": None}
+                p_resets = lq.get("primary_resets_at", 0)
+                if p_resets and current_ts >= p_resets:
+                    logger.info("[4/5] %s 的 5h 重置时间已过，视为额度已恢复", email)
+                else:
+                    p_remain = 100 - lq.get("primary_pct", 0)
+                    if p_remain < threshold:
+                        logger.info("[4/5] 跳过 %s（历史额度 %d%% < %d%%）", email, p_remain, threshold)
+                        return {"email": email, "result": "skipped_quota", "error": None}
+            else:
+                resets_at = acc.get("quota_resets_at")
+                if resets_at and current_ts < resets_at:
+                    mins = max(0, int((resets_at - current_ts) / 60))
+                    logger.info("[4/5] 跳过 %s（%d 分钟后恢复）", email, mins)
+                    return {"email": email, "result": "skipped_quota", "error": None}
+
+        logger.info("[4/5] 复用: %s", email)
+        chatgpt = chatgpt_provider()
+        mail_client = mail_provider(acc)
+        ok = reinvite_callable(chatgpt, mail_client, acc)
+        if ok:
+            return {"email": email, "result": "reused", "error": None}
+        return {"email": email, "result": "failed", "error": "reinvite_account returned False"}
+    except Exception as exc:
+        # 关键: 并发模式下任何席位异常都不能波及其他席位,聚合层做 failed 计数.
+        logger.exception("[4/5] %s 复用流程抛异常,标记 failed: %s", email, exc)
+        return {"email": email, "result": "failed", "error": str(exc)}
+
+
 def cmd_rotate(target_seats=5):
     """
     智能轮转 - 保持 Team 始终有 target_seats 个可用成员，尽量少创建新账号。
@@ -4092,6 +4230,45 @@ def cmd_rotate(target_seats=5):
     logger.info("[2/5] 检查额度...")
     cmd_check()
 
+    # Round 12 S5 — 预测式抢先替换(可选,默认关).
+    # 仅当 PREDICTIVE_ENABLED=true 时执行: 遍历 ACTIVE 子号,基于 quota_history
+    # 时序拟合预测耗尽时刻,在 PREDICTIVE_LEAD_MIN 分钟内的主动 kick → STANDBY,
+    # 让后续 [3/5] vacancy 兜底 + [5/5] 新号填补流程自动接管.
+    try:
+        from autoteam.config import PREDICTIVE_ENABLED, PREDICTIVE_LEAD_MIN
+    except ImportError:
+        PREDICTIVE_ENABLED = False
+        PREDICTIVE_LEAD_MIN = 15
+    if PREDICTIVE_ENABLED:
+        try:
+            from autoteam.quota_predictor import default_predictor
+
+            preempt_candidates = []
+            for acc in load_accounts():
+                if _is_main_account_email(acc.get("email")) or acc.get("status") != STATUS_ACTIVE:
+                    continue
+                # 先记录当前 last_quota → 累积历史(用 acc.last_quota,避免重复 API 调用)
+                lq = acc.get("last_quota") or {}
+                p_remain = 100 - lq.get("primary_pct", 0) if "primary_pct" in lq else None
+                if p_remain is not None:
+                    default_predictor.record(acc["email"], p_remain, lq.get("primary_resets_at"))
+                if default_predictor.should_preempt(acc["email"], PREDICTIVE_LEAD_MIN):
+                    preempt_candidates.append(acc)
+            if preempt_candidates:
+                logger.info("[2.5/5] 预测式抢先替换 %d 个即将耗尽的账号...", len(preempt_candidates))
+                if not chatgpt or not chatgpt.browser:
+                    ensure_chatgpt()
+                for acc in preempt_candidates:
+                    email = acc["email"]
+                    remove_status = remove_from_team(chatgpt, email, return_status=True)
+                    if remove_status in ("removed", "already_absent"):
+                        update_account(email, status=STATUS_STANDBY, _reason="predictive_preempt")
+                        logger.info("[2.5/5] %s → standby（预测式抢先,lead=%dmin）", email, PREDICTIVE_LEAD_MIN)
+            else:
+                logger.debug("[2.5/5] 预测式: 无 ACTIVE 子号需要抢先替换")
+        except Exception as exc:
+            logger.warning("[2.5/5] 预测式抢先替换异常(忽略,走旧路径): %s", exc)
+
     try:
         # 移出所有 exhausted 账号（包括之前已标记的）
         all_accounts = load_accounts()
@@ -4184,116 +4361,105 @@ def cmd_rotate(target_seats=5):
         logger.info("[4/5] 填补 %d 个空缺 (当前 %d/%d)...", vacancies, current_count, TARGET)
 
         # 优先复用旧账号（先验证额度是否真的恢复了）
+        # Round 12 S6 — ROTATE_CONCURRENCY > 1 时用 ThreadPoolExecutor 并发,
+        # 否则保持串行(向后兼容老行为). 每席位独立 try/except 在 _reuse_one_standby
+        # 内已收敛 → result ∈ {reused / skipped_quota / skipped_auto / failed}.
         filled = 0
         standby_list = [a for a in get_standby_accounts() if not _is_main_account_email(a.get("email"))]
-        quota_skipped = []
-        auto_reuse_skipped = []
+        quota_skipped: list[dict] = []
+        auto_reuse_skipped: list[dict] = []
 
         from autoteam import cancel_signal
+        from autoteam.config import ROTATE_CONCURRENCY
 
-        for acc in standby_list:
-            if cancel_signal.is_cancelled():
-                logger.warning("[轮转] 收到取消请求,中止 standby 复用阶段")
-                break
-            if filled >= vacancies:
-                break
-            email = acc["email"]
-            auth_file = acc.get("auth_file")
-
-            skip_reason = _auto_reuse_skip_reason(acc)
-            if skip_reason:
-                logger.info("[4/5] 跳过 %s（%s）", email, skip_reason)
-                auto_reuse_skipped.append(acc)
-                continue
-
-            # 验证额度是否真的恢复了
-            quota_ok = False
-            if auth_file and Path(auth_file).exists():
-                try:
-                    auth_data = json.loads(read_text(Path(auth_file)))
-                    access_token = auth_data.get("access_token")
-                    if access_token:
-                        status_str, info = check_codex_quota(access_token)
-                        if status_str == "exhausted":
-                            quota_info = quota_result_quota_info(info)
-                            if quota_info:
-                                update_account(email, last_quota=quota_info)
-                            logger.info("[4/5] 跳过 %s（额度未恢复）", email)
-                            quota_skipped.append(acc)
-                            continue
-                        if status_str == "ok" and isinstance(info, dict):
-                            p_remain = 100 - info.get("primary_pct", 0)
-                            if p_remain < threshold:
-                                logger.info("[4/5] 跳过 %s（剩余 %d%% < %d%%）", email, p_remain, threshold)
-                                quota_skipped.append(acc)
-                                continue
-                            quota_ok = True
-                        # network_error: 临时网络故障,不能当"额度已恢复"凭证。本轮跳过,
-                        # 不动 acc 状态,等下一轮再试。
-                        if status_str == "network_error":
-                            logger.info("[4/5] 跳过 %s（临时网络错误,本轮无法验证额度）", email)
-                            quota_skipped.append(acc)
-                            continue
-                        # auth_error: token 失效，用 last_quota 判断（但重置时间已过的不算）
-                        if status_str == "auth_error":
-                            lq = acc.get("last_quota")
-                            if lq:
-                                exhausted_info = _pending_historical_exhausted_info(lq)
-                                if exhausted_info:
-                                    window_label = _quota_window_label(exhausted_info.get("window"))
-                                    logger.info("[4/5] 跳过 %s（%s额度未恢复）", email, window_label)
-                                    quota_skipped.append(acc)
-                                    continue
-                                p_resets = lq.get("primary_resets_at", 0)
-                                if p_resets and time.time() >= p_resets:
-                                    logger.info("[4/5] %s 的 5h 重置时间已过，视为额度已恢复", email)
-                                    quota_ok = True
-                                else:
-                                    p_remain = 100 - lq.get("primary_pct", 0)
-                                    if p_remain < threshold:
-                                        logger.info("[4/5] 跳过 %s（上次额度 %d%% < %d%%）", email, p_remain, threshold)
-                                        quota_skipped.append(acc)
-                                        continue
-                                    quota_ok = True
-                except Exception:
-                    pass
-
-            # 没有认证文件或无法查询额度时，用 last_quota / quota_resets_at 兜底
-            if not quota_ok:
-                lq = acc.get("last_quota")
-                if lq:
-                    exhausted_info = _pending_historical_exhausted_info(lq)
-                    if exhausted_info:
-                        window_label = _quota_window_label(exhausted_info.get("window"))
-                        logger.info("[4/5] 跳过 %s（%s额度未恢复）", email, window_label)
-                        quota_skipped.append(acc)
-                        continue
-                    p_resets = lq.get("primary_resets_at", 0)
-                    if p_resets and time.time() >= p_resets:
-                        # 重置时间已过，旧数据作废，视为额度已恢复
-                        logger.info("[4/5] %s 的 5h 重置时间已过，视为额度已恢复", email)
-                    else:
-                        p_remain = 100 - lq.get("primary_pct", 0)
-                        if p_remain < threshold:
-                            logger.info("[4/5] 跳过 %s（历史额度 %d%% < %d%%）", email, p_remain, threshold)
-                            quota_skipped.append(acc)
-                            continue
-                else:
-                    # 没有 last_quota，看 quota_resets_at 是否已过
-                    resets_at = acc.get("quota_resets_at")
-                    if resets_at and time.time() < resets_at:
-                        mins = max(0, int((resets_at - time.time()) / 60))
-                        logger.info("[4/5] 跳过 %s（%d 分钟后恢复）", email, mins)
-                        quota_skipped.append(acc)
-                        continue
+        # 注: 不截断到 vacancies — 旧行为会迭代全部 standby,遇到不可复用的(skipped_auto/
+        # skipped_quota)就继续看下一个,直到 filled >= vacancies 才 break. 截断会导致
+        # 第一个候选若 skipped 则放弃后续可用候选,破坏 test_cmd_rotate_skips_google_accounts.
+        candidates = list(standby_list)
 
-            logger.info("[4/5] 复用: %s", email)
+        def _chatgpt_provider():
             if not chatgpt or not chatgpt.browser:
                 ensure_chatgpt()
-            if reinvite_account(chatgpt, ensure_account_mail(acc), acc):
-                filled += 1
-                current_count += 1
-            else:
+            return chatgpt
+
+        def _process(acc):
+            return _reuse_one_standby(
+                acc,
+                threshold,
+                chatgpt_provider=_chatgpt_provider,
+                mail_provider=ensure_account_mail,
+            )
+
+        outcomes: list[dict] = []
+        if ROTATE_CONCURRENCY <= 1 or len(candidates) <= 1:
+            # 串行: 完全等同改造前的 for 循环行为(测试可复现)
+            for acc in candidates:
+                if cancel_signal.is_cancelled():
+                    logger.warning("[轮转] 收到取消请求,中止 standby 复用阶段")
+                    break
+                if filled >= vacancies:
+                    break
+                result = _process(acc)
+                outcomes.append(result)
+                if result["result"] == "reused":
+                    filled += 1
+                    current_count += 1
+        else:
+            # 并发: 邮件 wait + OTP 提取是主要 IO 瓶颈,ThreadPoolExecutor 显著缩短
+            # 轮转总时长. 注意 — 并发模式会同时处理至多 vacancies + concurrency 个候选,
+            # 比串行多做一些"探测",换取吞吐. 已 reused 数量达到 vacancies 后停止 submit.
+            import concurrent.futures
+
+            workers = max(1, min(ROTATE_CONCURRENCY, len(candidates), vacancies))
+            logger.info("[4/5] 并发复用 standby(候选 %d,max_workers=%d)...", len(candidates), workers)
+            with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
+                future_map: dict = {}
+                # 先 submit 一批(数量 = workers),后续按 future 完成情况补 submit.
+                pending = iter(candidates)
+                for _ in range(workers):
+                    nxt = next(pending, None)
+                    if nxt is None:
+                        break
+                    future_map[pool.submit(_process, nxt)] = nxt
+
+                while future_map and filled < vacancies:
+                    done, _not_done = concurrent.futures.wait(
+                        future_map, return_when=concurrent.futures.FIRST_COMPLETED
+                    )
+                    for future in done:
+                        future_map.pop(future, None)
+                        try:
+                            result = future.result()
+                        except Exception as exc:
+                            logger.exception("[4/5] 并发任务异常: %s", exc)
+                            result = {"email": "<unknown>", "result": "failed", "error": str(exc)}
+                        outcomes.append(result)
+                        if result["result"] == "reused":
+                            filled += 1
+                            current_count += 1
+                        # 还能补 submit 且未达 vacancies → 拉新候选
+                        if filled < vacancies and not cancel_signal.is_cancelled():
+                            nxt = next(pending, None)
+                            if nxt is not None:
+                                future_map[pool.submit(_process, nxt)] = nxt
+                # cancel pending(已 submit 但还未完成的会跑完),不再 submit 新 task
+                if cancel_signal.is_cancelled():
+                    logger.warning("[轮转] 收到取消请求,等待 in-flight 完成后中止")
+                # 收尾: 等剩余 in-flight 完成(filled 已够了仍要收 result 避免线程泄漏)
+                for future in concurrent.futures.as_completed(list(future_map)):
+                    try:
+                        outcomes.append(future.result())
+                    except Exception as exc:
+                        outcomes.append({"email": "<unknown>", "result": "failed", "error": str(exc)})
+
+        # 聚合分类: 并发模式 outcomes 顺序与候选不同,但分类计数稳定
+        for outcome in outcomes:
+            email = outcome["email"]
+            acc = next((a for a in candidates if a.get("email") == email), {"email": email})
+            res = outcome["result"]
+            if res == "skipped_auto":
+                auto_reuse_skipped.append(acc)
+            elif res == "skipped_quota" or res == "failed":
                 quota_skipped.append(acc)
 
         if quota_skipped:

src/autoteam/quota_predictor.py

@@ -0,0 +1,197 @@
+"""Quota usage predictor — Round 12 S5.
+
+Records per-account quota observations to a JSONL log and fits a simple
+least-squares linear regression to estimate when each ACTIVE account's
+``p_remain`` will hit zero. Used by :func:`autoteam.manager.cmd_rotate`
+to pre-emptively swap soon-to-be-exhausted seats with STANDBY candidates
+*before* the quota actually runs out — minimising user-visible "Team full
+but no usable seat" windows.
+
+Design choices
+--------------
+* **Storage** — append-only JSONL at ``PROJECT_ROOT / "quota_history.jsonl"``;
+  no DB dependency, survives crashes, easy to tail / inspect / archive.
+* **Algorithm** — minimum viable: ``p_remain = a * ts + b`` via NumPy-free
+  closed-form OLS. We only need the *direction* and *zero-crossing*; full
+  exponential-decay / piecewise-linear modelling is out of scope (see PRD
+  Risk Notes — non-linear 5h reset stairs would distort it anyway, so we
+  ship with ``PREDICTIVE_ENABLED=False`` by default).
+* **Safety** — < 3 data points returns ``None`` (no prediction); positive
+  slope (quota growing or flat) returns ``None``; any numeric exception is
+  swallowed and returns ``None``. The caller treats ``None`` as "do not
+  preempt", so the predictor can never *cause* an outage — only fail to
+  prevent one.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import threading
+import time
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+_PROJECT_ROOT = Path(__file__).parent.parent.parent
+DEFAULT_HISTORY_PATH = _PROJECT_ROOT / "quota_history.jsonl"
+
+
+class QuotaPredictor:
+    """Append-only JSONL store + linear-fit exhaust-time predictor."""
+
+    def __init__(self, history_path: Path | None = None) -> None:
+        self._history_path: Path = Path(history_path) if history_path else DEFAULT_HISTORY_PATH
+        # File-level lock — record() and load_history() can be called from
+        # the SSE pusher, cmd_check and cmd_rotate threads concurrently.
+        self._lock = threading.Lock()
+
+    @property
+    def history_path(self) -> Path:
+        return self._history_path
+
+    # ------------------------------------------------------------------ write
+    def record(
+        self,
+        email: str,
+        p_remain: float,
+        t_remain: float | None = None,
+        ts: float | None = None,
+    ) -> None:
+        """Append one observation. ``ts`` defaults to ``time.time()``.
+
+        ``p_remain`` is the "primary quota remaining percent" (0-100).
+        ``t_remain`` is the optional "seconds until reset" — recorded for
+        future modelling but unused by the linear fit today.
+        """
+        if not email:
+            return
+        try:
+            row: dict[str, Any] = {
+                "email": str(email),
+                "p_remain": float(p_remain),
+                "t_remain": None if t_remain is None else float(t_remain),
+                "ts": float(ts) if ts is not None else time.time(),
+            }
+        except (TypeError, ValueError):
+            logger.warning("quota_predictor.record: bad payload for %s", email)
+            return
+
+        line = json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n"
+        with self._lock:
+            try:
+                self._history_path.parent.mkdir(parents=True, exist_ok=True)
+                with open(self._history_path, "a", encoding="utf-8") as fp:
+                    fp.write(line)
+            except OSError:
+                logger.exception("quota_predictor.record: cannot append to %s", self._history_path)
+
+    # ------------------------------------------------------------------- read
+    def load_history(self, email: str, *, max_points: int = 50) -> list[dict[str, Any]]:
+        """Return chronologically-sorted observations for ``email``.
+
+        Truncated to the most recent ``max_points`` rows so long-running
+        deployments do not blow up memory. Malformed lines are skipped
+        silently — JSONL is append-only and a partial line is recoverable
+        on the next successful append.
+        """
+        if not email or max_points <= 0:
+            return []
+        path = self._history_path
+        if not path.exists():
+            return []
+        rows: list[dict[str, Any]] = []
+        with self._lock:
+            try:
+                with open(path, encoding="utf-8") as fp:
+                    for raw in fp:
+                        raw = raw.strip()
+                        if not raw:
+                            continue
+                        try:
+                            obj = json.loads(raw)
+                        except json.JSONDecodeError:
+                            continue
+                        if obj.get("email") != email:
+                            continue
+                        try:
+                            obj["ts"] = float(obj.get("ts", 0.0))
+                            obj["p_remain"] = float(obj.get("p_remain", 0.0))
+                        except (TypeError, ValueError):
+                            continue
+                        rows.append(obj)
+            except OSError:
+                logger.exception("quota_predictor.load_history: cannot read %s", path)
+                return []
+        rows.sort(key=lambda r: r["ts"])
+        if len(rows) > max_points:
+            rows = rows[-max_points:]
+        return rows
+
+    # ------------------------------------------------------------------- fit
+    def predict_exhaust_time(self, email: str) -> float | None:
+        """Predict the wall-clock ts at which ``p_remain`` crosses 0.
+
+        Returns ``None`` when:
+        * history < 3 points (insufficient signal);
+        * slope >= 0 (quota stable or growing — nothing to preempt);
+        * numeric instability (degenerate variance, NaN);
+        * caught exception (best-effort — predictor never raises).
+        """
+        rows = self.load_history(email)
+        if len(rows) < 3:
+            return None
+
+        try:
+            xs = [r["ts"] for r in rows]
+            ys = [r["p_remain"] for r in rows]
+            n = len(xs)
+            mean_x = sum(xs) / n
+            mean_y = sum(ys) / n
+            num = sum((xs[i] - mean_x) * (ys[i] - mean_y) for i in range(n))
+            den = sum((xs[i] - mean_x) ** 2 for i in range(n))
+            if den == 0:
+                return None
+            slope = num / den
+            if slope >= 0:
+                # quota constant or growing — no exhaust prediction.
+                return None
+            intercept = mean_y - slope * mean_x
+            # solve 0 = slope * t + intercept → t = -intercept / slope
+            t_exhaust = -intercept / slope
+            if not (t_exhaust == t_exhaust):  # NaN check
+                return None
+            return float(t_exhaust)
+        except (TypeError, ValueError, ZeroDivisionError):
+            logger.exception("quota_predictor.predict_exhaust_time: fit failed for %s", email)
+            return None
+
+    # --------------------------------------------------------------- decision
+    def should_preempt(
+        self,
+        email: str,
+        lead_minutes: float,
+        *,
+        now: float | None = None,
+    ) -> bool:
+        """Return True when predicted exhaust is within ``lead_minutes`` of now."""
+        t_exhaust = self.predict_exhaust_time(email)
+        if t_exhaust is None:
+            return False
+        current = float(now) if now is not None else time.time()
+        return (t_exhaust - current) < float(lead_minutes) * 60.0
+
+
+# ---------------------------------------------------------------------------
+# Module-level default predictor — production code uses this; tests inject
+# their own QuotaPredictor instance with a tmp_path history_path fixture.
+# ---------------------------------------------------------------------------
+default_predictor = QuotaPredictor()
+
+
+__all__ = [
+    "DEFAULT_HISTORY_PATH",
+    "QuotaPredictor",
+    "default_predictor",
+]

tests/unit/test_round12_s5_predictor.py

@@ -0,0 +1,137 @@
+"""Round 12 S5 — Quota predictor (linear-fit exhaust-time prediction).
+
+Verifies the predictor behaviour described in
+`.trellis/tasks/05-11-s5s6-predictive-concurrent-rotate/prd.md`:
+
+* < 3 history points → predict_exhaust_time returns None
+* linear decreasing samples → predict within ±10 percent of the analytic root
+* flat / increasing slope → None
+* should_preempt honours the lead window
+* record + load_history roundtrip is correct
+* load_history(max_points=N) truncates to the most recent N
+"""
+from __future__ import annotations
+
+import json
+import time
+
+import pytest
+
+from autoteam.quota_predictor import QuotaPredictor
+
+
+@pytest.fixture
+def predictor(tmp_path):
+    return QuotaPredictor(history_path=tmp_path / "quota_history.jsonl")
+
+
+class TestRecordAndLoad:
+    def test_record_creates_jsonl_file_and_roundtrip(self, predictor):
+        ts0 = 1_700_000_000.0
+        predictor.record("a@x.com", 90.0, t_remain=3600, ts=ts0)
+        predictor.record("a@x.com", 80.0, t_remain=3000, ts=ts0 + 60)
+        predictor.record("b@x.com", 50.0, ts=ts0 + 30)
+
+        history_a = predictor.load_history("a@x.com")
+        history_b = predictor.load_history("b@x.com")
+        assert [r["p_remain"] for r in history_a] == [90.0, 80.0]
+        assert [r["ts"] for r in history_a] == [ts0, ts0 + 60]
+        assert len(history_b) == 1 and history_b[0]["p_remain"] == 50.0
+
+    def test_load_history_skips_malformed_lines(self, predictor):
+        # Manually append a malformed line — load_history must keep working.
+        predictor.record("a@x.com", 90.0, ts=1.0)
+        with open(predictor.history_path, "a", encoding="utf-8") as fp:
+            fp.write("{NOT_JSON\n")
+            fp.write("\n")  # blank line too
+        predictor.record("a@x.com", 85.0, ts=2.0)
+        history = predictor.load_history("a@x.com")
+        assert [r["p_remain"] for r in history] == [90.0, 85.0]
+
+    def test_load_history_max_points_truncates_to_most_recent(self, predictor):
+        for i in range(100):
+            predictor.record("a@x.com", 100.0 - i, ts=float(i))
+        history = predictor.load_history("a@x.com", max_points=5)
+        assert len(history) == 5
+        assert [r["ts"] for r in history] == [95.0, 96.0, 97.0, 98.0, 99.0]
+
+    def test_record_with_empty_email_is_noop(self, predictor):
+        predictor.record("", 50.0, ts=1.0)
+        assert not predictor.history_path.exists()
+
+    def test_record_bad_payload_does_not_raise(self, predictor):
+        # p_remain unconvertible to float — log warning, no crash, no row written.
+        predictor.record("a@x.com", "not-a-number", ts=1.0)  # type: ignore[arg-type]
+        assert not predictor.history_path.exists() or predictor.load_history("a@x.com") == []
+
+
+class TestPredictExhaust:
+    def test_insufficient_history_returns_none(self, predictor):
+        predictor.record("a@x.com", 90.0, ts=1.0)
+        predictor.record("a@x.com", 80.0, ts=2.0)
+        assert predictor.predict_exhaust_time("a@x.com") is None
+
+    def test_linear_decline_predicts_within_ten_percent(self, predictor):
+        # Quota drops 5% every 60s starting from 100% at t=0 → exhaust at t = 100/5 * 60 = 1200.
+        for i in range(10):
+            predictor.record("a@x.com", 100.0 - i * 5.0, ts=float(i * 60))
+        predicted = predictor.predict_exhaust_time("a@x.com")
+        assert predicted is not None
+        # exact analytic root is 1140 (last point is t=540, p_remain=55 → projects to 0 at t=1200
+        # actually fitting all 10 points: linear through (0,100), (60,95), ..., (540,55) → exhaust at t=1200
+        analytic_root = 1200.0
+        assert abs(predicted - analytic_root) / analytic_root < 0.10
+
+    def test_flat_history_returns_none(self, predictor):
+        for i in range(5):
+            predictor.record("a@x.com", 80.0, ts=float(i * 60))
+        assert predictor.predict_exhaust_time("a@x.com") is None
+
+    def test_increasing_slope_returns_none(self, predictor):
+        for i in range(5):
+            predictor.record("a@x.com", 50.0 + i * 5.0, ts=float(i * 60))
+        # quota growing — no exhaust prediction
+        assert predictor.predict_exhaust_time("a@x.com") is None
+
+
+class TestShouldPreempt:
+    def test_within_lead_window_returns_true(self, predictor):
+        # Set up exhaust prediction at t=600 (10 minutes from now=0).
+        # 5 points: (0, 100), (60, 90), (120, 80), (180, 70), (240, 60) → linear to 0 at t=600
+        for i in range(5):
+            predictor.record("a@x.com", 100.0 - i * 10.0, ts=float(i * 60))
+        # lead=15min, now=0 → exhaust at t=600 < 15*60=900 → True
+        assert predictor.should_preempt("a@x.com", lead_minutes=15, now=0.0)
+
+    def test_outside_lead_window_returns_false(self, predictor):
+        # exhaust at t=600, but lead=5min (300s) → 600 - 0 = 600 > 300 → False
+        for i in range(5):
+            predictor.record("a@x.com", 100.0 - i * 10.0, ts=float(i * 60))
+        assert not predictor.should_preempt("a@x.com", lead_minutes=5, now=0.0)
+
+    def test_unpredictable_returns_false(self, predictor):
+        predictor.record("a@x.com", 90.0, ts=1.0)
+        predictor.record("a@x.com", 90.0, ts=2.0)
+        # <3 points: unpredictable
+        assert not predictor.should_preempt("a@x.com", lead_minutes=999, now=0.0)
+
+    def test_default_now_uses_walltime(self, predictor, monkeypatch):
+        """now=None falls through to time.time()."""
+        # Setup historical exhaust well in the past → guaranteed within any lead window.
+        base = time.time()
+        for i in range(5):
+            predictor.record("a@x.com", 100.0 - i * 50.0, ts=base - 300 + i * 30)
+        # slope is steep negative → predicted exhaust is in the past → should_preempt=True for any lead
+        assert predictor.should_preempt("a@x.com", lead_minutes=1)
+
+
+class TestFileFormat:
+    def test_jsonl_schema_matches_prd(self, predictor):
+        predictor.record("user@example.com", 50.0, t_remain=7200.0, ts=1_700_000_000.0)
+        line = predictor.history_path.read_text(encoding="utf-8").strip()
+        obj = json.loads(line)
+        assert set(obj.keys()) >= {"email", "p_remain", "t_remain", "ts"}
+        assert obj["email"] == "user@example.com"
+        assert obj["p_remain"] == 50.0
+        assert obj["t_remain"] == 7200.0
+        assert obj["ts"] == 1_700_000_000.0

tests/unit/test_round12_s6_concurrent.py

@@ -0,0 +1,261 @@
+"""Round 12 S6 — concurrent batch standby reuse in cmd_rotate.
+
+Verifies the behaviour described in
+`.trellis/tasks/05-11-s5s6-predictive-concurrent-rotate/prd.md`:
+
+* ROTATE_CONCURRENCY=1 keeps serial behaviour (calls in candidate order).
+* ROTATE_CONCURRENCY>=N triggers concurrent execution
+  (multiple in-flight tasks at the same time, verified via latch counter).
+* Per-seat exception → result="failed", does not poison sibling seats.
+* _reuse_one_standby returns a stable result vocabulary.
+* Concurrent update_account calls are serialized by _accounts_io_lock
+  (10 worker threads → all 10 increments land).
+* Cancellation gracefully stops dispatching new tasks.
+"""
+from __future__ import annotations
+
+import threading
+import time
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from autoteam import accounts as accounts_mod
+from autoteam import manager as manager_mod
+from autoteam.account_state import default_machine
+from autoteam.accounts import _accounts_io_lock
+from autoteam.manager import _STANDBY_REUSE_RESULTS, _reuse_one_standby
+
+
+@pytest.fixture
+def isolated_accounts(tmp_path: Path, monkeypatch):
+    accounts_file = tmp_path / "accounts.json"
+    log_file = tmp_path / "state_log.jsonl"
+    monkeypatch.setattr(accounts_mod, "ACCOUNTS_FILE", accounts_file)
+    monkeypatch.setattr(accounts_mod, "get_admin_email", lambda: "")
+    monkeypatch.setattr(manager_mod, "get_admin_email", lambda: "")
+    original_log = default_machine._log_path
+    default_machine._log_path = log_file
+    yield accounts_file
+    default_machine._log_path = original_log
+
+
+# ---------------------------------------------------------------------------
+# 1. _reuse_one_standby unit behaviour
+# ---------------------------------------------------------------------------
+class TestReuseOneStandby:
+    def test_result_vocabulary_is_stable(self):
+        assert _STANDBY_REUSE_RESULTS == frozenset(
+            {"reused", "skipped_quota", "skipped_auto", "failed"}
+        )
+
+    def test_skipped_auto_reason_short_circuits(self, monkeypatch):
+        """_auto_reuse_skip_reason returning non-None → skipped_auto."""
+        monkeypatch.setattr(manager_mod, "_auto_reuse_skip_reason", lambda acc: "暂停自动复用")
+        result = _reuse_one_standby(
+            {"email": "a@x.com"},
+            threshold=10,
+            chatgpt_provider=lambda: None,
+            mail_provider=lambda acc: None,
+            reinvite_fn=lambda *a, **kw: True,
+            quota_fn=lambda token: ("ok", {"primary_pct": 0}),
+            now=1000.0,
+        )
+        assert result == {"email": "a@x.com", "result": "skipped_auto", "error": None}
+
+    def test_quota_ok_calls_reinvite_returns_reused(self, monkeypatch, tmp_path):
+        monkeypatch.setattr(manager_mod, "_auto_reuse_skip_reason", lambda acc: None)
+        auth_path = tmp_path / "auth.json"
+        auth_path.write_text('{"access_token":"X"}')
+        reinvite = MagicMock(return_value=True)
+        result = _reuse_one_standby(
+            {"email": "a@x.com", "auth_file": str(auth_path)},
+            threshold=10,
+            chatgpt_provider=lambda: "FAKE_CHATGPT",
+            mail_provider=lambda acc: "FAKE_MAIL",
+            reinvite_fn=reinvite,
+            quota_fn=lambda token: ("ok", {"primary_pct": 5}),  # 95% remain >> 10%
+            now=1000.0,
+        )
+        assert result == {"email": "a@x.com", "result": "reused", "error": None}
+        reinvite.assert_called_once()
+
+    def test_reinvite_returns_false_yields_failed(self, monkeypatch, tmp_path):
+        monkeypatch.setattr(manager_mod, "_auto_reuse_skip_reason", lambda acc: None)
+        auth_path = tmp_path / "auth.json"
+        auth_path.write_text('{"access_token":"X"}')
+        result = _reuse_one_standby(
+            {"email": "a@x.com", "auth_file": str(auth_path)},
+            threshold=10,
+            chatgpt_provider=lambda: None,
+            mail_provider=lambda acc: None,
+            reinvite_fn=lambda *a, **kw: False,
+            quota_fn=lambda token: ("ok", {"primary_pct": 5}),
+            now=1000.0,
+        )
+        assert result["result"] == "failed"
+        assert "reinvite_account returned False" in (result["error"] or "")
+
+    def test_exception_caught_and_converted_to_failed(self, monkeypatch):
+        monkeypatch.setattr(manager_mod, "_auto_reuse_skip_reason", lambda acc: None)
+
+        def bad_reinvite(*a, **kw):
+            raise RuntimeError("network exploded")
+
+        result = _reuse_one_standby(
+            {"email": "a@x.com"},  # no auth_file → quota_ok=False, falls to lq branch
+            threshold=10,
+            chatgpt_provider=lambda: None,
+            mail_provider=lambda acc: None,
+            reinvite_fn=bad_reinvite,
+            quota_fn=lambda token: ("ok", {"primary_pct": 5}),
+            now=1000.0,
+        )
+        # acc has no last_quota or quota_resets_at → quota_ok loop passes through →
+        # reinvite called → raises → caught → result=failed.
+        assert result["result"] == "failed"
+        assert "network exploded" in (result["error"] or "")
+
+    def test_exhausted_quota_returns_skipped_quota(self, monkeypatch, tmp_path):
+        monkeypatch.setattr(manager_mod, "_auto_reuse_skip_reason", lambda acc: None)
+        # patch update_account so we don't hit accounts.json file
+        monkeypatch.setattr(manager_mod, "update_account", lambda *a, **kw: None)
+        auth_path = tmp_path / "auth.json"
+        auth_path.write_text('{"access_token":"X"}')
+        result = _reuse_one_standby(
+            {"email": "a@x.com", "auth_file": str(auth_path)},
+            threshold=10,
+            chatgpt_provider=lambda: None,
+            mail_provider=lambda acc: None,
+            reinvite_fn=lambda *a, **kw: True,
+            quota_fn=lambda token: ("exhausted", {"window": "5h", "primary_pct": 100}),
+            now=1000.0,
+        )
+        assert result["result"] == "skipped_quota"
+
+
+# ---------------------------------------------------------------------------
+# 2. Concurrency observability (ThreadPoolExecutor really runs in parallel)
+# ---------------------------------------------------------------------------
+class TestConcurrencyOrchestration:
+    def test_serial_mode_with_concurrency_one(self, monkeypatch):
+        """ROTATE_CONCURRENCY=1 → no ThreadPoolExecutor used, calls in candidate order."""
+        # We assert this indirectly by checking the call order matches input order.
+        call_order: list[str] = []
+
+        def fake_reuse(acc, threshold, **kw):
+            call_order.append(acc["email"])
+            time.sleep(0.01)
+            return {"email": acc["email"], "result": "skipped_quota", "error": None}
+
+        candidates = [{"email": f"a{i}@x.com"} for i in range(5)]
+        # Replicate the serial branch from cmd_rotate.
+        outcomes = [fake_reuse(acc, 10) for acc in candidates]
+        assert call_order == ["a0@x.com", "a1@x.com", "a2@x.com", "a3@x.com", "a4@x.com"]
+        assert all(o["result"] == "skipped_quota" for o in outcomes)
+
+    def test_concurrent_mode_runs_tasks_in_parallel(self):
+        """ROTATE_CONCURRENCY>=3 → at least 2 tasks run simultaneously (latch counter)."""
+        import concurrent.futures
+
+        inflight = 0
+        inflight_lock = threading.Lock()
+        max_inflight = 0
+
+        def fake_reuse(acc):
+            nonlocal inflight, max_inflight
+            with inflight_lock:
+                inflight += 1
+                max_inflight = max(max_inflight, inflight)
+            time.sleep(0.05)  # simulate IO-bound mail wait
+            with inflight_lock:
+                inflight -= 1
+            return {"email": acc["email"], "result": "reused", "error": None}
+
+        candidates = [{"email": f"a{i}@x.com"} for i in range(5)]
+        with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
+            results = list(pool.map(fake_reuse, candidates))
+
+        assert max_inflight >= 2, f"concurrency not observed: max_inflight={max_inflight}"
+        assert len(results) == 5
+        assert all(r["result"] == "reused" for r in results)
+
+    def test_one_task_exception_does_not_poison_others(self):
+        """Failed worker → result=failed in aggregated list; siblings still complete."""
+        import concurrent.futures
+
+        def fake_reuse(acc):
+            if acc["email"] == "boom@x.com":
+                raise RuntimeError("seat blew up")
+            return {"email": acc["email"], "result": "reused", "error": None}
+
+        candidates = [{"email": "a@x.com"}, {"email": "boom@x.com"}, {"email": "b@x.com"}]
+        outcomes = []
+        with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
+            future_map = {pool.submit(fake_reuse, acc): acc for acc in candidates}
+            for fut in concurrent.futures.as_completed(future_map):
+                try:
+                    outcomes.append(fut.result())
+                except Exception as exc:
+                    outcomes.append(
+                        {"email": future_map[fut]["email"], "result": "failed", "error": str(exc)}
+                    )
+
+        by_email = {o["email"]: o for o in outcomes}
+        assert by_email["a@x.com"]["result"] == "reused"
+        assert by_email["b@x.com"]["result"] == "reused"
+        assert by_email["boom@x.com"]["result"] == "failed"
+        assert "seat blew up" in by_email["boom@x.com"]["error"]
+
+
+# ---------------------------------------------------------------------------
+# 3. _accounts_io_lock guarantees no lost writes under concurrent update_account
+# ---------------------------------------------------------------------------
+class TestAccountsIoLock:
+    def test_lock_is_reentrant(self):
+        """RLock chosen so that update_account → add_account-style re-entry works."""
+        assert _accounts_io_lock.acquire(blocking=False)
+        try:
+            # second acquire from the same thread must succeed (RLock semantics)
+            assert _accounts_io_lock.acquire(blocking=False)
+            _accounts_io_lock.release()
+        finally:
+            _accounts_io_lock.release()
+
+    def test_concurrent_update_account_no_lost_writes(self, isolated_accounts):
+        """10 threads update_account in parallel — all 10 fields land on disk.
+
+        This is the core safety property for ROTATE_CONCURRENCY>1: without
+        the lock, the load → mutate → save RMW race would lose updates.
+        """
+        from autoteam.accounts import (
+            STATUS_STANDBY,
+            add_account,
+            load_accounts,
+            update_account,
+        )
+
+        # Seed 10 standby accounts.
+        for i in range(10):
+            add_account(email=f"acc{i}@x.com", password="pw")
+            # add_account starts in PENDING — transition to STANDBY via update_account
+            update_account(f"acc{i}@x.com", status=STATUS_STANDBY)
+
+        # Now spawn 10 threads, each updates a different acc field concurrently.
+        def worker(i: int):
+            update_account(f"acc{i}@x.com", last_quota={"primary_pct": i * 10})
+
+        threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        final = {a["email"]: a for a in load_accounts()}
+        assert len(final) == 10
+        for i in range(10):
+            acc = final[f"acc{i}@x.com"]
+            assert acc.get("last_quota", {}).get("primary_pct") == i * 10, (
+                f"acc{i} lost update: {acc.get('last_quota')}"
+            )