fix: harden seat rotation and direct signup race

.env.example

@@ -86,6 +86,8 @@ AUTO_CHECK_THRESHOLD=10      # 额度低于此百分比触发轮转，默认 10%
 AUTO_CHECK_MIN_LOW=2         # 至少几个账号低于阈值才触发，默认 2
 AUTO_CHECK_RETRY_ADD_PHONE=true          # 是否自动重试 add_phone（手机号验证）
 AUTO_CHECK_ADD_PHONE_MAX_RETRIES=3        # add_phone 最大自动重试次数
+ROTATE_SKIP_REUSE=true                    # 轮换默认不复用旧/失败子号，满员时先移旧号再创建新号
+ROTATE_MAX_DURATION=1500                  # 单次 rotate 最长运行秒数，超时优雅退出交给下轮巡检
 
 # 多 Team 母号并行调度（每个 Team 仍保持 1 母 + 最多 2 子）
 MULTI_MASTER_MAX_OWNER_WORKERS=2          # 同时处理几个母号

.trellis/spec/backend/free-registration-hardening.md

@@ -13,10 +13,10 @@
 - `TaskParams.leave_workspace: bool = False`
 - `TaskParams.target: int = 3`
 - `post_fill(params: TaskParams = TaskParams())`
-- `cmd_fill(target=3, leave_workspace=False)`
+- `cmd_fill(target=3, leave_workspace=False, *, post_sync=True, print_status=True, direct_parallel=None)`
 - `_cmd_fill_personal(count)`
-- `create_new_account(chatgpt_api, mail_client=None, *, leave_workspace=False, out_outcome=None, acc=None, path_rotator=None)`
-- `create_account_direct(mail_client=None, *, leave_workspace=False, out_outcome=None, acc=None, path_rotator=None)`
+- `create_new_account(chatgpt_api, mail_client=None, *, leave_workspace=False, out_outcome=None, acc=None, path_rotator=None, parallel=None)`
+- `create_account_direct(mail_client=None, *, leave_workspace=False, out_outcome=None, acc=None, path_rotator=None, parallel=None)`
 - `_run_post_register_oauth(email, password, mail_client, leave_workspace=False, out_outcome=None, chatgpt_session_token=None, signup_profile=None)`
 - `generate_signup_profile(*, today: date | None = None, rng: random.Random | random.SystemRandom | None = None) -> SignupProfile`
 
@@ -105,3 +105,99 @@ in_team_local = _count_local_team_seat_accounts(load_accounts())
 ```
 
 Keep the API entrypoint and manager entrypoint aligned so unsafe fill-personal work is rejected before starting browser or mail-provider operations.
+
+## Scenario: Direct Signup Race and Managed Child Validation
+
+### 1. Scope / Trigger
+
+- Trigger: any change to direct signup, `DIRECT_REGISTER_PARALLEL`, multi-master owner fill budgets, Team fill/rotate child creation, or auto-check decisions when local usable children are below target.
+- Goal: improve throughput without breaking the `1 owner + 2 managed children = 3 seats` contract or accepting a newly created child before remote/auth/quota validation.
+
+### 2. Signatures
+
+- `DIRECT_REGISTER_PARALLEL: int`, clamped to `1..4`.
+- `AUTOTEAM_REGISTER_PARALLEL_MEMORY_WARN_RATIO`, default `0.72`.
+- `AUTOTEAM_REGISTER_PARALLEL_MAX_BROWSER_LIVE`, default `4`.
+- `_direct_register_parallel_size() -> int`.
+- `_cap_direct_register_parallel(requested: int) -> int`.
+- `_attempt_chatgpt_signup_only(mail_client, *, acc=None, out_outcome=None) -> dict`.
+- `_race_chatgpt_signup(mail_client_factory, *, parallel: int, acc=None, out_outcome=None) -> dict`.
+- `create_account_direct(..., parallel=None)`.
+- `create_new_account(..., parallel=None)`.
+- `cmd_fill(..., direct_parallel=None)`.
+- `_validate_managed_account_operational(email, *, threshold: int, stage_label="[轮转验收]", chatgpt_api=None) -> bool`.
+
+### 3. Contracts
+
+- `parallel=None` means read `DIRECT_REGISTER_PARALLEL` and then apply local runtime downgrades.
+- Direct signup race must run independent signup-only workers. Only the winner is persisted with `add_account()` and passed into `_run_post_register_oauth()`.
+- Loser workers that successfully reached Team must be removed from Team, have their temporary mailbox discarded, and have account-scoped IPv6 proxy state released.
+- The same `SignupProfile` used by a winning signup worker must be passed into post-registration OAuth.
+- `cmd_fill(..., direct_parallel=N)` must pass `N` into `create_new_account(..., parallel=N)`. Multi-master worker budgets must not remain display-only metadata.
+- New managed children created by fill/rotate must pass remote member presence, local auth file, and Codex quota checks before counting as filled.
+- If a new child fails validation, release the remote Team seat and mark the local row back to standby with a diagnostic reason.
+- `ROTATE_SKIP_REUSE=true` means pending invites and standby reuse are skipped for automated fill/rotate; replacement must be remove-before-create under full Team.
+- Auto-check cooldown/full-Team logic must still trigger `auto-fill` when local usable active children are below target and replaceable blockers are present.
+
+### 4. Validation & Error Matrix
+
+| Condition | Required behavior |
+| --- | --- |
+| `DIRECT_REGISTER_PARALLEL <= 1` | Run the existing serial direct signup path |
+| Runtime memory ratio exceeds warn threshold | Downgrade direct signup race to `1` |
+| Browser process count exceeds max live threshold | Downgrade direct signup race to `1` |
+| A race worker fails registration | Clean up its temporary mailbox through the existing failure path |
+| A non-winning race worker succeeds | Remove it from Team, delete its mailbox, and release its proxy |
+| Winning child lacks local auth file | Do not count it as filled; release/mark standby |
+| Winning child is absent from remote Team member list | Do not count it as filled; release/mark standby |
+| Winning child quota is below threshold or auth fails | Do not count it as filled; release/mark standby |
+| Team is full and a replaceable blocker exists | Remove the blocker first, wait for observed capacity, then create the replacement |
+| Team is full and no blocker exists | Do not create before remove |
+
+### 5. Good/Base/Bad Cases
+
+- Good: `DIRECT_REGISTER_PARALLEL=3` starts three signup-only attempts, persists one winner, and reports race counts in `out_outcome`.
+- Good: multi-master owner fill computes a direct parallel budget and the worker calls `cmd_fill(..., direct_parallel=budget)`.
+- Base: serial `parallel=1` keeps the previous duplicate-swap and add-phone behavior.
+- Bad: creating a second child before removing an unusable full-Team blocker.
+- Bad: counting a newly created child as successful before remote/auth/quota validation.
+- Bad: only displaying `direct_register_parallel` in task metadata while `create_account_direct()` still runs serially.
+
+### 6. Tests Required
+
+- `tests/unit/test_free_registration_hardening.py`
+  - direct signup race starts multiple signup workers and persists only the winner.
+  - high memory or high browser-live runtime snapshots downgrade parallel to `1`.
+- `tests/unit/test_multi_master.py`
+  - owner worker passes `direct_parallel` into `cmd_fill`.
+- `tests/unit/test_manager_fill.py`
+  - `cmd_fill` passes direct parallel into `create_new_account` and releases a child that fails validation.
+- `tests/unit/test_manager_rotate.py`
+  - replaceable blocker reasons are concrete.
+  - full-Team replacement removes the blocker before creating a child.
+- `tests/unit/test_api_status.py`
+  - auto-check cooldown/full-Team logic still starts `auto-fill` when replaceable blockers exist.
+
+### 7. Wrong vs Correct
+
+#### Wrong
+
+```python
+result = create_new_account(chatgpt, mail_client)
+if result:
+    current_count += 1
+```
+
+This treats "created" as "operational" and can leave a dead child occupying one of the two managed seats.
+
+#### Correct
+
+```python
+created_email = create_new_account(chatgpt, mail_client, parallel=direct_parallel)
+if created_email and _validate_managed_account_operational(created_email, threshold=threshold, chatgpt_api=chatgpt):
+    current_count += 1
+else:
+    remove_from_team(chatgpt, created_email, return_status=True)
+```
+
+Creation, Team membership, local auth, and quota are separate facts. A child becomes usable only after all are validated.

.trellis/spec/backend/runtime-docker-hardening.md

@@ -182,7 +182,7 @@ Use `require_browser=True` for any path that depends on a real browser context.
 - `build_multi_master_status(accounts=None, pool=None) -> dict`
 - `resolve_worker_budget(owner_count, *, requested_owner_workers=None, requested_direct_parallel=None, runtime_snapshot=None) -> dict`
 - `run_multi_master_fill(target_seats=3, *, owner_workers=None, direct_parallel=None, workspace_ids=None, dry_run=False, post_sync=True, pool=None, worker=None) -> dict`
-- `cmd_fill(target=3, leave_workspace=False, *, post_sync=True, print_status=True)`
+- `cmd_fill(target=3, leave_workspace=False, *, post_sync=True, print_status=True, direct_parallel=None)`
 - `POST /api/tasks/multi-master/fill`
 
 ### 3. Contracts
@@ -193,8 +193,8 @@ Use `require_browser=True` for any path that depends on a real browser context.
 - Per-owner workers must use `temporary_admin_state(...)` so `ChatGPTTeamAPI.start()` reads the owner-local session/account/workspace data without mutating global `state.json`.
 - Owner worker failures are isolated. One failed owner becomes one failed result row and must not prevent other submitted owners from completing.
 - `session_token` may be persisted in the local workspace pool for imported owners, but API status and task results must expose only `session_present`, never the raw token.
-- `MULTI_MASTER_MAX_OWNER_WORKERS`, `MULTI_MASTER_BROWSER_BUDGET`, `MULTI_MASTER_MEMORY_DOWNGRADE_RATIO`, and `DIRECT_REGISTER_PARALLEL` are the scheduling budget inputs. The effective direct-registration race is a budget value until the direct-signup race is explicitly wired into `manager.py`; do not claim single-account race behavior from the scheduler alone.
-- When owner workers call `cmd_fill`, they must set `post_sync=False` and `print_status=False`. The parent multi-master task may run CPA sync once after all owners finish; a post-sync failure should be reported in `post_sync` without converting completed owner work into failed owner work.
+- `MULTI_MASTER_MAX_OWNER_WORKERS`, `MULTI_MASTER_BROWSER_BUDGET`, `MULTI_MASTER_MEMORY_DOWNGRADE_RATIO`, and `DIRECT_REGISTER_PARALLEL` are the scheduling budget inputs. `run_multi_master_fill()` must pass the resolved `direct_register_parallel` into `cmd_fill(..., direct_parallel=...)`, and `cmd_fill()` must pass it into `create_new_account(..., parallel=...)`.
+- When owner workers call `cmd_fill`, they must set `post_sync=False`, `print_status=False`, and the budgeted `direct_parallel` value. The parent multi-master task may run CPA sync once after all owners finish; a post-sync failure should be reported in `post_sync` without converting completed owner work into failed owner work.
 - `/api/status.multi_master` is additive. It must not remove or rename existing status fields and must not raise a 500 if multi-master diagnostics fail.
 
 ### 4. Validation & Error Matrix
@@ -206,6 +206,7 @@ Use `require_browser=True` for any path that depends on a real browser context.
 | Owner row has no session token | Mark it non-runnable in status; execution may fail only that owner |
 | Runtime memory ratio >= `MULTI_MASTER_MEMORY_DOWNGRADE_RATIO` | Force `owner_workers=1` and `direct_register_parallel=1` with reason `memory_high` |
 | `owner_workers * direct_register_parallel` exceeds `MULTI_MASTER_BROWSER_BUDGET` | Clip both values to stay within the global browser budget |
+| Owner worker starts `cmd_fill` | Pass the resolved `direct_parallel`; do not leave it as display-only metadata |
 | One owner worker raises | Record `last_error` for that workspace and return overall `partial_failed` when other owners complete |
 | All owner workers raise | Return overall `failed` |
 | Parent post-sync raises | Return `post_sync.ok=false`; keep per-owner results intact |
@@ -214,6 +215,7 @@ Use `require_browser=True` for any path that depends on a real browser context.
 ### 5. Good/Base/Bad Cases
 
 - Good: two imported owners marked `parallel=true` are filled inside one `multi-master-fill` task, each owner uses its own temporary admin state, and final `post_sync` runs once.
+- Good: a multi-master dry-run/result reports the same `direct_register_parallel` that execution passes into `cmd_fill`.
 - Base: a single-owner install with no parallel rows still reports compatible status and can dry-run against the active workspace.
 - Bad: increasing `target` above `3` to gain throughput, because this breaks the Team-seat contract.
 - Bad: calling `cmd_fill()` concurrently with its default `post_sync=True`, because each worker would race remote CPA sync and make delete-guard behavior harder to reason about.
@@ -226,7 +228,7 @@ Use `require_browser=True` for any path that depends on a real browser context.
   - `WorkspacePool.upsert` persists owner metadata and keeps the active-workspace invariant.
   - `build_multi_master_status` groups accounts by `workspace_account_id` and omits session tokens.
   - `resolve_worker_budget` downgrades on high memory and clips by browser budget.
-  - `run_multi_master_fill` isolates owner failures, records `last_error`, suppresses worker-local sync/status, and reports parent `post_sync`.
+  - `run_multi_master_fill` isolates owner failures, records `last_error`, suppresses worker-local sync/status, passes `direct_parallel` into `cmd_fill`, and reports parent `post_sync`.
   - API dry-run returns a plan and passes request parameters through.
 - Existing single-owner regressions must still pass:
   - `tests/unit/test_round12_s7_workspace_pool.py`

.trellis/tasks/05-19-autoteam1-seat-rotation-hardening-migration/check.jsonl

@@ -0,0 +1,5 @@
+{"file": ".trellis/spec/backend/index.md", "reason": "backend quality verification entrypoint"}
+{"file": ".trellis/spec/backend/free-registration-hardening.md", "reason": "verify direct signup race, child validation, and Team seat contracts"}
+{"file": ".trellis/spec/backend/runtime-docker-hardening.md", "reason": "verify multi-master budget propagation and runtime downgrade behavior"}
+{"file": ".trellis/spec/backend/account-disable-cpa-sync.md", "reason": "verify disabled/protected credential and CPA guard behavior was not weakened"}
+{"file": ".trellis/tasks/05-19-autoteam1-seat-rotation-hardening-migration/research/current-vs-autoteam1-seat-rotation.md", "reason": "verify implementation matches the migration slice and avoids whole-file copying"}

.trellis/tasks/05-19-autoteam1-seat-rotation-hardening-migration/implement.jsonl

@@ -0,0 +1,6 @@
+{"file": ".trellis/spec/backend/index.md", "reason": "backend task entrypoint and applicable spec index"}
+{"file": ".trellis/spec/backend/free-registration-hardening.md", "reason": "direct signup race, fill-personal, Team seat, SignupProfile, and managed child validation contracts"}
+{"file": ".trellis/spec/backend/runtime-docker-hardening.md", "reason": "multi-master direct parallel budget, runtime downgrade, Playwright/browser budget, and status contracts"}
+{"file": ".trellis/spec/backend/account-disable-cpa-sync.md", "reason": "disabled accounts and CPA delete guard must remain excluded from automated rotation/reuse/sync"}
+{"file": ".trellis/spec/guides/index.md", "reason": "shared code-reuse and cross-layer thinking triggers for config/API/manager changes"}
+{"file": ".trellis/tasks/05-19-autoteam1-seat-rotation-hardening-migration/research/current-vs-autoteam1-seat-rotation.md", "reason": "current vs autoteam-1 migration comparison and selected Approach B context"}

.trellis/tasks/05-19-autoteam1-seat-rotation-hardening-migration/prd.md

@@ -0,0 +1,147 @@
+# autoteam-1 seat rotation hardening migration
+
+## Goal
+
+将 `D:\Desktop\autoteam-1\AutoTeam` 中围绕 Team 席位轮换、注册链路、运行态校验、CPA 同步防护和资源治理的一系列优化，经过差异审计后迁移到当前项目 `D:\Desktop\AutoTeam`。目标不是机械照搬，而是吸收目标仓库已验证有效的设计，并按当前项目现有状态机、注册栈、多母号调度和运行约束做适配性改善。
+
+## What I Already Know
+
+* 用户明确要求走 `$trellis-brainstorm`，本轮先规划和收敛 PRD，不直接改业务代码。
+* 当前项目已有大量相关 in-progress Trellis 任务；本任务必须避免把无关 WIP 混入实现。
+* 当前仓库已吸收不少目标仓能力：3-seat clamp、runtime validation、`runtime_resources`、IPv6 pool/proxy、CLIProxy health、Docker/runtime hardening、multi-master scaffold。
+* 目标仓最新值得迁移的剩余能力集中在：`ROTATE_SKIP_REUSE` 默认跳过旧号复用、replaceable pool blocker 判定、满员时 remove-before-create、每个新 child 的 operational validation、更细的 auto-check action 分流、rotate heartbeat / duration fuse、以及 direct signup race 的真实接入。
+* 当前项目和目标项目已经明显分叉。`api.py`、`manager.py`、`cpa_sync.py` 和 `test_api_status.py` 仍有大差异，不能整文件覆盖。
+
+## Requirements
+
+* 以当前项目为主线迁移，不覆盖现有 `RegisterPathRotator`、多 provider mail fallback、SignupProfile 复用、Playwright cleanup、multi-master scaffold、CPA delete guard 和前端状态面板。
+* 保持单 Team 硬约束：`1 owner + 2 managed children = 3 seats`，不得通过 create-before-remove 或临时超席位绕过。
+* 将目标仓的“不可用但占席 child”概念适配到当前 `STATUS_AUTH_INVALID` / account-state 语义，识别 missing-auth、auth-error、auth-retry paused、exhausted、auth-invalid 等 blocker。
+* 当 Team 已满但存在可替换 blocker 或确认不可用 child 时，必须先远端移除旧 child，确认 capacity 后再创建 replacement。
+* 新建 child 不能只以“创建成功”算成功；需验证远端 member 存在、本地 auth 文件存在、Codex quota/auth 可用，再计入可用 pool。
+* 自动巡检需要区分至少四类动作：低额度轮换、真实 seat shortage 补位、auth repair、超员 cleanup；cooldown 只限制低额度抖动，不能阻止真实缺席补位或确认无额度占席替换。
+* 长任务需要可观测进度：rotate/fill 关键阶段应有 heartbeat 或等价 progress 标记；单次 rotate 需要 duration fuse，避免和 watchdog/recovery 周期互相撞车。
+* CPA/CLIProxyAPI 同步必须继续采用 delete guard。degraded pool、active list 不完整、远端列表读取失败时不得删除远端 auth。
+* 若纳入 direct signup race，必须受当前 multi-master browser budget 约束，并保留当前注册路径轮换与 provider fallback。
+
+## Acceptance Criteria
+
+* [x] 研究记录明确标注当前已吸收、仍缺失、需适配、不能覆盖的目标仓能力。
+* [x] 实现前选定 MVP slice，并把不进入本轮的迁移项列入 Out of Scope。
+* [x] 若进入实现，新增/更新测试覆盖：3-seat cap、remove-before-create、protected local credential guard、replaceable blocker reason、auto-check shortage/auth-repair/cleanup/cooldown 分流、runtime validation degraded/failed 语义。
+* [x] 若实现 child validation，测试必须证明新 child 未通过远端/auth/quota 验收时不会被计入完成，并会释放或标记为待处理。
+* [x] 若实现 direct signup race，测试必须证明 `DIRECT_REGISTER_PARALLEL` 被实际传入注册路径，且高内存/低 budget 时自动降级。
+* [x] 运行 `ruff`、相关 `pytest`，并在需要时用本地 API/status/task 证据确认运行态字段。
+* [x] 不修改、提交或回滚与本任务无关的既有未提交 WIP。
+
+## Research References
+
+* [`research/current-vs-autoteam1-seat-rotation.md`](research/current-vs-autoteam1-seat-rotation.md) — re-verified current vs target migration notes and recommended slices.
+
+## Technical Approach
+
+### Approach A: Safe rotation core (Recommended)
+
+先迁移目标仓的轮换安全核心：blocker 分类、remove-before-create、managed child validation、auto-check action 分流、cooldown 语义、heartbeat 和 duration fuse。保留现有注册栈和 multi-master scaffold，不在同一 slice 接入 direct signup race。
+
+Pros:
+
+* 最贴近用户说的“轮换席位等一系列优化”。
+* 风险集中在后端轮换/巡检，可用单元测试和局部 API 验证闭环。
+* 避免同时引入 direct registration 并发导致 Playwright 资源风险放大。
+
+Cons:
+
+* 注册成功率/速度提升主要来自更准确的轮换和验证，不会立即获得目标仓 direct signup race 的吞吐收益。
+
+### Approach B: Rotation core + direct signup race
+
+在 Approach A 基础上同时迁移目标仓 `_race_chatgpt_signup` / `DIRECT_REGISTER_PARALLEL`，并接入当前 `RegisterPathRotator`、`cmd_fill` 和 multi-master browser budget。
+
+Pros:
+
+* 一次性覆盖轮换准确性和单账号注册吞吐。
+* 与多母号并行调度的最终方向更接近。
+
+Cons:
+
+* 风险显著更高：Playwright 并发、邮箱 provider 限速、注册失败分类、全局任务锁都要一起验证。
+* 更容易和现有 `05-18-multi-master-parallel-registration` 任务边界重叠。
+
+### Approach C: Audit-only decomposition
+
+本任务只产出迁移审计和拆分子任务，不进入实现。随后把 rotation core、direct signup race、multi-master parity 分别建子任务执行。
+
+Pros:
+
+* 最稳，不碰当前脏工作区。
+* 适合用户希望先梳理多个 in-progress 任务关系时使用。
+
+Cons:
+
+* 不能立即改善当前运行态轮换效率和准确性。
+
+## Decision (ADR-lite)
+
+Context: 当前项目已经吸收了目标仓的一批低耦合能力，但仍缺目标仓最新的“占席 blocker → 先移后补 → 子号验收 → 细分巡检动作”闭环。整文件复制会破坏当前 Round 11/12 注册栈和多母号架构。
+
+Decision: 选择 Approach B：Rotation core + direct signup race。
+
+Consequences: 本轮同时解决席位轮换准确性和单账号注册吞吐。实现必须把 direct signup race 接入当前 `RegisterPathRotator` / `cmd_fill` / multi-master browser budget，而不能只暴露配置或结果字段。风险集中在 Playwright 并发、邮箱 provider 限速、注册失败归因和测试矩阵扩大，需要更宽的单元测试和至少一次资源预算降级验证。
+
+## Out of Scope
+
+* 本 brainstorm 阶段不直接修改业务代码。
+* 不提高单 Team seat cap，不做 create-before-remove。
+* 不整文件覆盖目标仓 `api.py`、`manager.py`、`cpa_sync.py`。
+* 不使用会上传或删除远端 auth 文件的 `/api/sync` 作为 smoke test。
+* 不重启或扰动当前 live container，除非后续实现阶段明确需要并经过验证窗口。
+* 不自动创建新的 Team owner 母号；多母号并行仅复用当前已有/imported owner scaffold。
+
+## Expansion Sweep
+
+Future evolution:
+
+* 本轮 rotation core 后，可以把 direct signup race 作为独立 slice 接到 multi-master browser budget。
+* 后续多母号并行可复用同一 blocker/validation contract，但按 owner 分组运行。
+
+Related scenarios:
+
+* API `/api/status`、任务 `validation`、前端 PoolPage/TeamPage 需要继续展示同一套健康语义。
+* CPA/CLIProxyAPI sync 和 account cleanup 必须共享 protected credential / delete guard 规则。
+
+Failure and edge cases:
+
+* Team count probe unknown、remote remove delayed、new child auth missing、quota API network_error、auth repair throttled、protected local credential、disabled account、external remote member。
+* Windows 本地环境与 Docker runtime 对 Playwright/browser 资源限制不同，实现需保留降级和可观测日志。
+
+## Technical Notes
+
+* Task dir: `.trellis/tasks/05-19-autoteam1-seat-rotation-hardening-migration`
+* Current repo: `D:\Desktop\AutoTeam`
+* Target repo: `D:\Desktop\autoteam-1\AutoTeam`
+* GitNexus 当前未索引 AutoTeam；Serena MCP 在本会话返回参数错误，本轮已记录回退到本地只读搜索。
+* External search was used only to validate general SaaS lifecycle direction: remove/deprovision before replacement, preflight quota, idempotent retry, background reconciliation, audit logging.
+
+## Implementation Notes
+
+Date: 2026-05-19
+
+Implemented Approach B in the current repo without whole-file copying from the target repo:
+
+* Added `ROTATE_SKIP_REUSE` and `ROTATE_MAX_DURATION` config/env examples.
+* Added replaceable pool blocker classification for `missing_auth`, `auth_error`, `auth_retry_*`, `auth_invalid`, `quota_exhausted`, and `orphan`, while preserving disabled/protected local credential seats.
+* Wired remove-before-create semantics into `cmd_rotate`: replaceable blockers are removed first, remote capacity is polled, then replacements are created.
+* Added managed child operational validation after fill/rotate creation. A new child must have local auth, remote Team membership, and Codex quota above threshold before it counts as filled; failed validation releases/marks the child standby.
+* Wired direct signup race through `create_account_direct(..., parallel=...)`, `create_new_account(..., parallel=...)`, `cmd_fill(..., direct_parallel=...)`, and multi-master owner workers. Race workers run signup-only attempts; only the winner is persisted/OAuth'd.
+* Added runtime downgrade for direct race based on memory ratio and live browser process count.
+* Updated auto-check so cooldown/full-Team observations do not block rotation when replaceable local blockers exist.
+* Updated backend specs for the direct signup race, child validation, and multi-master budget propagation contracts.
+
+Verification:
+
+* `python -m py_compile src/autoteam/manager.py src/autoteam/multi_master.py src/autoteam/config.py`
+* `python -m ruff check src/autoteam/api.py src/autoteam/manager.py src/autoteam/multi_master.py src/autoteam/config.py tests/unit/test_api_status.py tests/unit/test_manager_rotate.py tests/unit/test_manager_fill.py tests/unit/test_free_registration_hardening.py tests/unit/test_multi_master.py tests/unit/test_reconcile_anomalies.py`
+* `pytest -q tests/unit` -> `826 passed, 1 warning`
+
+Runtime note: no live Docker container was restarted and `/api/sync` was not used as a smoke test.

.trellis/tasks/05-19-autoteam1-seat-rotation-hardening-migration/research/current-vs-autoteam1-seat-rotation.md

@@ -0,0 +1,113 @@
+# Current vs autoteam-1 seat rotation migration notes
+
+Date: 2026-05-19
+
+## Scope
+
+This note re-checks the current `D:\Desktop\AutoTeam` tree against the target `D:\Desktop\autoteam-1\AutoTeam` tree for the user's requested migration around seat rotation and related hardening. Older Trellis audit files are useful history but some of their conclusions are stale because current `main` has since absorbed IPv6 pool, CLIProxy health, runtime resources, multi-master scaffolding, and other features.
+
+## Re-verified current baseline
+
+* Current repo recent commits include `ead7e93 feat: add multi-master Team fill scheduler`, `5aff1af fix: 对齐注册轮转链路和 CloudMail 配置提示`, `8543971 feat: add ipv6 proxy pool and status surface`, and `a011d67 fix: harden Playwright and HTTP transport cleanup`.
+* Target repo recent commits include `276ed0b fix: harden quota blocker rotation`, `f120570 fix rotate post sync and cliproxy health`, `6f4d106 fix invite blank page recovery`, `15e50d6 Improve registration credential sync and diagnostics`, `d952ce6 fix rotation cooldown safety boundary`, `2fb3031 fix rotation validation and ipv6 pool status`, and `7d46b46 fix autoteam seat rotation runtime hardening`.
+* `git diff --no-index --stat` still shows large divergence in `api.py`, `manager.py`, `cpa_sync.py`, and `tests/unit/test_api_status.py`, so broad copying would be unsafe.
+* GitNexus does not currently index either AutoTeam tree. Serena MCP returned parameter errors in this session. Code evidence here comes from local read-only searches and targeted file reads.
+
+## Already absorbed in current repo
+
+* 3-seat clamp exists: current `src/autoteam/config.py` clamps `AUTO_CHECK_TARGET_SEATS` to `1..3`; current `manager.py` has `TEAM_SEATS_MAX = 3` semantics via `_clamp_team_target_seats`.
+* Runtime validation and status exposure exist: current `api.py` has `_log_task_runtime_validation`, `_rotation_validation_cooldown`, `/api/status.runtime_resources`, `/api/status.ipv6_pool`, and `/api/status.cliproxy`.
+* IPv6 pool/proxy exists in current repo: `src/autoteam/ipv6_pool.py`, `src/autoteam/ipv6_proxy.py`, `manager._ensure_account_ipv6_proxy`, `chatgpt_api` integration, `cpa_sync` proxy refresh, `tests/unit/test_ipv6_pool.py`.
+* CLIProxyAPI read-only health exists: `src/autoteam/cliproxy_health.py`, `tests/unit/test_cliproxy_health.py`, and `web/src/components/PoolPage.vue` references `status.cliproxy`.
+* Docker/runtime hardening has a completed local audit under `.trellis/tasks/05-15-autoteam1-hardening-docker-apply/completion-audit-2026-05-15.md`.
+* Multi-master scaffolding exists in current repo: `src/autoteam/multi_master.py`, `tests/unit/test_multi_master.py`, and a dry-run API path. It groups owners and budgets owner workers, but it does not by itself solve the single-owner rotate blocker semantics below.
+
+## Remaining target behaviors not fully applied
+
+### 1. Default skip-reuse / new-account replacement policy
+
+Target repo has `ROTATE_SKIP_REUSE=true` in config and uses it throughout `manager.cmd_rotate`, `cmd_check`, and `create_new_account` to avoid reusing old/disabled/retired child accounts. Current repo does not define `ROTATE_SKIP_REUSE`; current `cmd_rotate` still prioritizes standby reuse and describes itself as "尽量少创建新账号". This is a real policy difference.
+
+Migration risk: current repo also has Round 12 provider fallback and `RegisterPathRotator`. A target-style skip-reuse policy should be added as an opt-in/defaulted config and threaded through current reuse paths rather than replacing the current manager wholesale.
+
+### 2. Replaceable pool blocker semantics
+
+Target repo has:
+
+* `manager._account_auth_state_blocks_pool_use`
+* `manager._is_pool_active_account_usable`
+* `manager._replaceable_pool_blocker_reason`
+* `manager._is_replaceable_pool_blocker`
+* `api._collect_auto_check_state` that classifies active/auth-pending/missing-auth/auth-error/low-quota candidates before choosing rotate, auth repair, cleanup, or no-op.
+
+Current repo search found no equivalent symbols. Current auto-check mostly counts `STATUS_ACTIVE` rows with existing auth files, handles basic cooldown, and triggers `auto-fill`; it does not classify local occupied-but-unusable Team seats as replaceable blockers.
+
+Migration value: this is likely the most important remaining target improvement for rotation accuracy. It prevents "Team full but local usable pool short" from being treated as healthy.
+
+### 3. Remove-before-create replacement under full Team
+
+Target `cmd_rotate` contains `attempt_remove_then_create(...)`, `_wait_for_remote_capacity_after_removal(...)`, and managed-account post-create validation. When the Team is already full and a child must be replaced, it removes a concrete old child, waits for remote capacity, then creates the replacement. Current `cmd_rotate` removes exhausted rows before creating, but lacks target's explicit full-Team remove-first transaction for local blockers and confirmed unusable low-quota seats.
+
+Migration value: aligns with the hard 3-seat cap and avoids create-before-remove behavior. This also matches external SaaS lifecycle guidance: deprovision/confirm capacity before provisioning replacements, keep operations idempotent, and record audit outcomes.
+
+### 4. Managed account operational validation
+
+Target repo has `manager._validate_managed_account_operational(...)` and tests requiring a new managed child to be present remotely and have usable Codex auth/quota. Current repo has post-task validation in `api.py`, but current `cmd_rotate` does not validate each replacement child before treating the vacancy as filled.
+
+Migration value: catches degraded successes earlier and allows immediate release/discard of a newly created but unusable child.
+
+### 5. More nuanced auto-check actions
+
+Target auto-check can trigger:
+
+* `auto-rotate` for real seat shortage, confirmed low/exhausted blockers, or local_pool_blocker.
+* `auto-auth-repair` when remote Team is full but local auth is missing/invalid and retry is not throttled.
+* `auto-cleanup` when remote Team exceeds target.
+* Cooldown bypass for real Team shortage or confirmed exhausted remote blocker, but cooldown delay for low-quota churn when capacity is otherwise healthy.
+
+Current tests cover only two cooldown cases around real shortage vs full Team. Target `tests/unit/test_api_status.py` has many more cases around degraded cooldown, unknown probes, auth repair, cleanup, and blocker rotation.
+
+Migration value: improves both efficiency and accuracy without changing the 3-seat contract.
+
+### 6. Task progress heartbeat and rotate duration fuse
+
+Target `api.py` exposes `bump_task_progress(...)`, and target `cmd_rotate` uses `ROTATE_MAX_DURATION` plus `_deadline_exceeded(...)`. Current repo search found no `bump_task_progress`, `ROTATE_MAX_DURATION`, or deadline helper.
+
+Migration value: long registration/rotation jobs can show healthy forward progress to watchdog-style observers and stop before colliding with the next recovery loop.
+
+### 7. Direct registration race is still not wired into single-owner registration
+
+Current `.env.example` contains `DIRECT_REGISTER_PARALLEL=1`, and `multi_master.resolve_worker_budget(...)` budgets `direct_register_parallel`, but current `manager.py` does not contain target's `_direct_register_parallel_size`, `_cap_direct_register_parallel`, or `_race_chatgpt_signup`. The current multi-master worker passes the direct-parallel value in result metadata but calls `cmd_fill(...)` without passing it into actual account creation.
+
+Migration value: if the project wants target's direct signup race, it must be wired into current `create_account_direct` / `cmd_fill` path while preserving `RegisterPathRotator`, provider fallback, SignupProfile reuse, Playwright cleanup, and global browser budget.
+
+## Areas that should not be blindly overwritten
+
+* Current repo uses `STATUS_AUTH_INVALID = "auth_invalid"` and an account-state machine where AUTH_PENDING maps to auth invalid semantics. Target repo uses `STATUS_AUTH_PENDING = "auth_pending"`. Migration must map concepts, not copy constants.
+* Current repo has newer `RegisterPathRotator`, multi-provider mail fallback, multi-master scaffolding, frontend status display, and stronger CPA live quota decision logic. Target `manager.py`, `api.py`, or `cpa_sync.py` should not be copied wholesale.
+* CPA sync in current repo already has a local `_active_auth_publish_decision` with live quota checks, disabled-account skip, personal credential handling, IPv6 proxy refresh, and delete guard. Target ideas should be compared at behavior level only.
+
+## External practice check
+
+Grok Search query on SaaS seat rotation and lifecycle automation returned consistent general guidance:
+
+* Deprovision or deactivate the old user/seat first, confirm capacity, then provision the replacement.
+* Preflight current seat count and quota before adding.
+* Make operations retry-safe and idempotent.
+* Use background jobs plus full/incremental reconciliation, with audit logs and failure alerts.
+
+These are compatible with the target repo's remove-before-create and validation direction, but current implementation details must still be governed by local code.
+
+## Recommended migration slices
+
+### Slice A: Safe rotation core (recommended MVP)
+
+Implement target-equivalent blocker classification, remove-before-create, child validation, cooldown semantics, rotate heartbeat, and duration fuse in current code. Preserve current registration provider/fallback stack and multi-master scaffolding.
+
+### Slice B: Direct signup race integration
+
+After Slice A, wire target direct registration race into current `create_account_direct` / `RegisterPathRotator` path and multi-master browser budget. This has higher Playwright resource risk.
+
+### Slice C: Broad target parity
+
+Attempt to converge all target repo rotation/registration behaviors in one larger implementation. This is highest risk because current repo has newer divergent architecture and many active Trellis tasks.

.trellis/tasks/05-19-autoteam1-seat-rotation-hardening-migration/task.json

@@ -0,0 +1,26 @@
+{
+  "id": "autoteam1-seat-rotation-hardening-migration",
+  "name": "autoteam1-seat-rotation-hardening-migration",
+  "title": "brainstorm: migrate autoteam-1 seat rotation hardening",
+  "description": "",
+  "status": "in_progress",
+  "dev_type": null,
+  "scope": null,
+  "package": null,
+  "priority": "P2",
+  "creator": "ZRainbow",
+  "assignee": "ZRainbow",
+  "createdAt": "2026-05-19",
+  "completedAt": null,
+  "branch": null,
+  "base_branch": "main",
+  "worktree_path": null,
+  "commit": null,
+  "pr_url": null,
+  "subtasks": [],
+  "children": [],
+  "parent": null,
+  "relatedFiles": [],
+  "notes": "",
+  "meta": {}
+}

src/autoteam/api.py

@@ -3592,6 +3592,22 @@ def _auto_check_loop():
                 and a.get("auth_file")
                 and Path(a["auth_file"]).exists()
             ]
+            try:
+                from autoteam.manager import _replaceable_pool_blocker_reason
+
+                replaceable_blockers = [
+                    {
+                        "email": a.get("email"),
+                        "reason": reason,
+                    }
+                    for a in accounts
+                    if not _is_main_account_email(a.get("email"))
+                    and not is_account_disabled(a)
+                    and (reason := _replaceable_pool_blocker_reason(a))
+                ]
+            except Exception as exc:
+                logger.warning("[巡检] 本地占席 blocker 分类失败: %s", exc)
+                replaceable_blockers = []
 
             # Watchdog:active 账号数 < 子号目标时自动补位。
             # 之前的 `if not active: continue` 在 active 全 kick 进 standby
@@ -3615,10 +3631,20 @@ def _auto_check_loop():
                     # 否则 cooldown 会把实际席位缺口拖到下一轮甚至 4h backoff 之后。
                     actual_team_count = _auto_check_team_member_count(timeout_seconds=30, retries=2)
                     if actual_team_count >= target_seats:
-                        logger.info(
-                            "[巡检] auto-fill 冷却中且 Team 实际成员数=%d 已满；本轮只保留低额度替换检查",
-                            actual_team_count,
-                        )
+                        if replaceable_blockers:
+                            logger.warning(
+                                "[巡检] auto-fill 冷却中但 Team 已满且存在 %d 个不可用占席子号，继续触发轮转: %s",
+                                len(replaceable_blockers),
+                                ", ".join(
+                                    f"{item['email']}({item['reason']})" for item in replaceable_blockers[:5]
+                                ),
+                            )
+                            should_start_auto_fill = True
+                        else:
+                            logger.info(
+                                "[巡检] auto-fill 冷却中且 Team 实际成员数=%d 已满；本轮只保留低额度替换检查",
+                                actual_team_count,
+                            )
                     elif actual_team_count >= 0:
                         logger.info(
                             "[巡检] auto-fill 冷却中但 Team 实际成员不足（%d/%d），继续补位",
@@ -3667,13 +3693,23 @@ def _auto_check_loop():
 
                     actual_team_count = _auto_check_team_member_count(timeout_seconds=30, retries=2)
                     if actual_team_count >= target_seats:
-                        logger.info(
-                            "[巡检] 本地 active=%d < %d,但 Team 实际成员数=%d 已满；先跳过 auto-fill,等待同步/对账稳定",
-                            len(active),
-                            sub_account_target,
-                            actual_team_count,
-                        )
-                        continue
+                        if replaceable_blockers:
+                            logger.warning(
+                                "[巡检] Team 已满但本地 active=%d/%d 且存在不可用占席子号，触发 auto-fill 修复: %s",
+                                len(active),
+                                sub_account_target,
+                                ", ".join(
+                                    f"{item['email']}({item['reason']})" for item in replaceable_blockers[:5]
+                                ),
+                            )
+                        else:
+                            logger.info(
+                                "[巡检] 本地 active=%d < %d,但 Team 实际成员数=%d 已满；先跳过 auto-fill,等待同步/对账稳定",
+                                len(active),
+                                sub_account_target,
+                                actual_team_count,
+                            )
+                            continue
                     should_start_auto_fill = True
 
                 if should_start_auto_fill:

src/autoteam/config.py

@@ -106,6 +106,10 @@ SUB2API_OVERWRITE_ACCOUNT_SETTINGS = _get_bool_env("SUB2API_OVERWRITE_ACCOUNT_SE
 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)
 
+# 默认不复用旧/失败/退役子号。Team 满员需要替换时必须先移出旧 child，再创建新 child。
+ROTATE_SKIP_REUSE = _get_bool_env("ROTATE_SKIP_REUSE", True)
+ROTATE_MAX_DURATION = max(60, _get_int_env("ROTATE_MAX_DURATION", 1500))
+
 
 # Round 12 S5 — 预测式抢先替换配置.
 # PREDICTIVE_ENABLED=false(默认 安全): cmd_rotate 不做预测式 preempt,

src/autoteam/manager.py

@@ -42,6 +42,7 @@ from autoteam.accounts import (
     delete_account,
     find_account,
     get_standby_accounts,
+    is_account_disabled,
     is_supported_plan,
     load_accounts,
     save_accounts,
@@ -349,14 +350,83 @@ def _count_pool_active_accounts(accounts: list[dict] | None = None, *, require_a
     accounts = accounts if accounts is not None else load_accounts()
     count = 0
     for acc in accounts:
-        if _is_main_account_email(acc.get("email")) or acc.get("status") != STATUS_ACTIVE:
-            continue
-        if require_auth and not _has_auth_file(acc):
+        if require_auth:
+            if not _is_pool_active_account_usable(acc, require_auth=True):
+                continue
+        elif _is_main_account_email(acc.get("email")) or is_account_disabled(acc) or acc.get("status") != STATUS_ACTIVE:
             continue
         count += 1
     return count
 
 
+def _account_auth_state_blocks_pool_use(acc: dict | None, *, now: float | None = None) -> bool:
+    """Return True when saved Codex auth is known or likely unusable."""
+    acc = acc or {}
+    if acc.get("auth_retry_paused"):
+        return True
+    if acc.get("auth_last_error"):
+        return True
+    retry_after = acc.get("auth_retry_after")
+    if retry_after:
+        try:
+            return float(retry_after) > (time.time() if now is None else now)
+        except (TypeError, ValueError):
+            return True
+    return False
+
+
+def _is_pool_active_account_usable(acc: dict | None, *, require_auth: bool = True) -> bool:
+    """Check whether a child account should count as an actually usable pool seat."""
+    acc = acc or {}
+    if _is_main_account_email(acc.get("email")) or is_account_disabled(acc):
+        return False
+    if acc.get("status") != STATUS_ACTIVE:
+        return False
+    if require_auth and not _has_auth_file(acc):
+        return False
+    if require_auth and _account_auth_state_blocks_pool_use(acc):
+        return False
+    return True
+
+
+def _replaceable_pool_blocker_reason(acc: dict | None, *, missing_auth: bool = False) -> str | None:
+    """Return the concrete reason a child seat blocks the usable pool but may be replaced."""
+    acc = acc or {}
+    if _is_main_account_email(acc.get("email")) or is_account_disabled(acc):
+        return None
+    if _is_protected_local_credential_seat(acc):
+        return None
+    status = acc.get("status")
+    if status == STATUS_EXHAUSTED:
+        return "quota_exhausted"
+    if status == STATUS_AUTH_INVALID:
+        return "auth_invalid"
+    if status == STATUS_ORPHAN:
+        return "orphan"
+    if status != STATUS_ACTIVE:
+        return None
+    if missing_auth:
+        return "auth_error"
+    if not _has_auth_file(acc):
+        return "missing_auth"
+    if acc.get("auth_retry_paused"):
+        return "auth_retry_paused"
+    if acc.get("auth_last_error"):
+        return "auth_error"
+    retry_after = acc.get("auth_retry_after")
+    if retry_after:
+        try:
+            if float(retry_after) > time.time():
+                return "auth_retry_after"
+        except (TypeError, ValueError):
+            return "auth_retry_after_invalid"
+    return None
+
+
+def _is_replaceable_pool_blocker(acc: dict | None, *, missing_auth: bool = False) -> bool:
+    return _replaceable_pool_blocker_reason(acc, missing_auth=missing_auth) is not None
+
+
 def _count_local_team_seat_accounts(accounts: list[dict] | None = None) -> int:
     """统计本地"占着 Team 席位"的非主号账号(上游 `.upstream/manager.py:175`).
 
@@ -368,7 +438,9 @@ def _count_local_team_seat_accounts(accounts: list[dict] | None = None) -> int:
     return sum(
         1
         for acc in accounts
-        if not _is_main_account_email(acc.get("email")) and acc.get("status") in seat_statuses
+        if not _is_main_account_email(acc.get("email"))
+        and not is_account_disabled(acc)
+        and acc.get("status") in seat_statuses
     )
 
 
@@ -2216,6 +2288,112 @@ def remove_from_team(chatgpt_api, email, *, return_status=False, lookup_retries=
         return "failed" if return_status else False
 
 
+def _wait_for_remote_capacity_after_removal(
+    chatgpt_api,
+    *,
+    target: int,
+    removed_email: str,
+    timeout: int = 24,
+    poll_interval: float = 3.0,
+    stage_label: str = "[Team]",
+) -> tuple[int, bool]:
+    """Poll Team member count after a removal before creating a replacement."""
+    deadline = time.time() + max(1, int(timeout))
+    latest_count = -1
+    while time.time() < deadline:
+        latest_count = get_team_member_count(chatgpt_api)
+        if latest_count >= 0 and latest_count < target:
+            logger.info("%s 已释放 %s 的席位，当前成员 %d/%d", stage_label, removed_email, latest_count, target)
+            return latest_count, True
+        logger.info(
+            "%s 等待 %s 的远端席位释放，当前成员=%s/%d",
+            stage_label,
+            removed_email,
+            latest_count if latest_count >= 0 else "unknown",
+            target,
+        )
+        time.sleep(max(0.5, float(poll_interval)))
+    return latest_count, False
+
+
+def _get_remote_team_member(email: str | None, chatgpt_api=None) -> dict | None:
+    """Return remote Team member by email, or None when absent/unknown."""
+    email_l = _normalized_email(email)
+    if not email_l:
+        return None
+    owns_api = chatgpt_api is None
+    api = chatgpt_api or ChatGPTTeamAPI()
+    try:
+        if owns_api or not _chatgpt_session_ready(api):
+            api.start()
+        members, _invites = fetch_team_state(api)
+        for member in members:
+            if _normalized_email(member.get("email")) == email_l:
+                return member
+        return None
+    finally:
+        if owns_api:
+            try:
+                api.stop()
+            except Exception:
+                pass
+
+
+def _validate_managed_account_operational(
+    email: str,
+    *,
+    threshold: int,
+    stage_label: str = "[轮转验收]",
+    chatgpt_api=None,
+) -> bool:
+    """Validate a newly managed child is remotely present and has usable Codex auth/quota."""
+    acc = find_account(load_accounts(), email)
+    if not acc:
+        logger.warning("%s %s 本地账号不存在", stage_label, email)
+        return False
+    if is_account_disabled(acc):
+        logger.warning("%s %s 已禁用，不能计入可用 child", stage_label, email)
+        return False
+    if acc.get("status") != STATUS_ACTIVE:
+        logger.warning("%s %s status=%s，不能计入可用 child", stage_label, email, acc.get("status"))
+        return False
+    if not _has_auth_file(acc):
+        logger.warning("%s %s 缺少可用 auth_file", stage_label, email)
+        return False
+
+    if chatgpt_api is not None:
+        try:
+            member = _get_remote_team_member(email, chatgpt_api=chatgpt_api)
+        except Exception as exc:
+            logger.warning("%s 查询远端成员失败: %s (%s)", stage_label, email, exc)
+            member = None
+        if not member:
+            logger.warning("%s %s 未出现在远端 Team 成员列表", stage_label, email)
+            return False
+
+    try:
+        auth_data = json.loads(read_text(Path(acc["auth_file"])))
+        access_token = auth_data.get("access_token")
+        if not access_token:
+            logger.warning("%s %s auth_file 缺少 access_token", stage_label, email)
+            return False
+        status_str, info = check_codex_quota(access_token)
+    except Exception as exc:
+        logger.warning("%s %s quota 验证异常: %s", stage_label, email, exc)
+        return False
+
+    if status_str != "ok" or not isinstance(info, dict):
+        logger.warning("%s %s quota 验证失败: %s", stage_label, email, status_str)
+        return False
+    primary_remaining = 100 - int(info.get("primary_pct", 100) or 100)
+    if primary_remaining < int(threshold):
+        logger.warning("%s %s quota 剩余 %d%% < %d%%", stage_label, email, primary_remaining, threshold)
+        return False
+    update_account(email, last_quota=info, last_active_at=time.time())
+    logger.info("%s %s 运行态验收通过，5h 剩余 %d%%", stage_label, email, primary_remaining)
+    return True
+
+
 def _kick_team_seat_after_oauth_failure(email: str, *, reason: str) -> None:
     """OAuth 失败时同步 KICK ws,消除"workspace 有 + 本地 auth 缺失"残废延迟。
 
@@ -3694,59 +3872,97 @@ def _extract_session_token_from_context(context):
     return session_token or None
 
 
-def create_account_direct(mail_client=None, *, leave_workspace=False, out_outcome=None, acc=None, path_rotator=None):
-    """
-    直接注册模式（域名已配置自动加入 workspace，不需要邀请）。
-    流程：创建邮箱 → 注册 ChatGPT → 自动加入 workspace → Codex 登录
-    leave_workspace: 加入 workspace 后是否立即退出，转为 personal 模式跑 OAuth。
-    out_outcome:     可选 dict，函数会把最终结局（success/phone_blocked/duplicate_exhausted/register_failed/...）
-                     + 统计信息（register_attempts / duplicate_swaps / last_email / reason）写入，供上游汇总。
+def _direct_register_float_env(name: str, default: float) -> float:
+    try:
+        return float(os.environ.get(name, str(default)))
+    except (TypeError, ValueError):
+        return default
 
-    捕获 RegisterBlocked：
-    - is_phone=True:     当前邮箱已暴露给 OpenAI，立即删邮箱、整个账号放弃（return None）
-    - is_duplicate=True: 换个临时邮箱继续尝试，独立计数不消耗 register_attempts
-    - 其他异常:          归入现有 retry 计数
 
-    Round 12 S4: `mail_client` 改为可选(默认 None) — None 时按 `acc`(可选)走
-    `_get_mail_client_for_account(acc)` 路由;旧调用方显式传 mail_client 时完全
-    保留旧行为(向后兼容)。register-level 多 provider rotation 由 `MAIL_PROVIDER_CHAIN`
-    env 驱动:配置 chain 时 `get_mail_client()` 返回 `FallbackMailProvider`,
-    自动在 mail-API 级失败时降级;邮箱-粒度的 provider 切换通过 `RegisterPathRotator`
-    包装(参见 `autoteam.mail.register_dual_path`),业务调用方可自行编排。
+def _direct_register_int_env(name: str, default: int) -> int:
+    try:
+        return int(os.environ.get(name, str(default)))
+    except (TypeError, ValueError):
+        return default
 
-    Round 12 wire-up (M3): `path_rotator` 可选 — 传入 `RegisterPathRotator` 实例时
-    启用邮箱-粒度的 provider 切换(OTP_TIMEOUT / DOMAIN_REJECTED / INVITE_LINK_MISSING
-    自动切下一 provider 重试整个注册流程);None 时走旧逻辑(单 provider + retry 3 次).
-    """
-    from autoteam.invite import RegisterBlocked
 
-    # Round 12 wire-up M3 — 邮箱-粒度 provider 切换。
-    if path_rotator is not None:
-        return _create_account_direct_via_rotator(
-            path_rotator,
-            leave_workspace=leave_workspace,
-            out_outcome=out_outcome,
-            acc=acc,
+def _cap_direct_register_parallel(requested: int) -> int:
+    requested = max(1, min(4, int(requested or 1)))
+    if requested <= 1:
+        return 1
+
+    try:
+        from autoteam.runtime_resources import collect_runtime_resource_snapshot
+
+        snapshot = collect_runtime_resource_snapshot()
+    except Exception as exc:
+        logger.debug("[直接注册] 读取资源快照失败，保留并行=%d: %s", requested, exc)
+        return requested
+
+    memory_ratio = snapshot.get("cgroup_memory_usage_ratio")
+    browser_live = int(snapshot.get("browser_process_live") or 0)
+    warn_ratio = _direct_register_float_env("AUTOTEAM_REGISTER_PARALLEL_MEMORY_WARN_RATIO", 0.72)
+    max_browser_live = _direct_register_int_env("AUTOTEAM_REGISTER_PARALLEL_MAX_BROWSER_LIVE", 4)
+
+    if memory_ratio is not None and float(memory_ratio) >= warn_ratio:
+        logger.warning(
+            "[直接注册] 内存占用 %.1f%% >= %.1f%%，并行注册从 %d 降级为 1，避免 OOM",
+            float(memory_ratio) * 100,
+            warn_ratio * 100,
+            requested,
         )
+        return 1
+
+    if browser_live >= max_browser_live:
+        logger.warning(
+            "[直接注册] 浏览器进程数 %d >= %d，并行注册从 %d 降级为 1，避免进程堆积",
+            browser_live,
+            max_browser_live,
+            requested,
+        )
+        return 1
+
+    return requested
+
+
+def _direct_register_parallel_size() -> int:
+    """读取并行尝试数（DIRECT_REGISTER_PARALLEL），范围 [1, 4]，再按本机资源降级。"""
+    try:
+        from autoteam.config import DIRECT_REGISTER_PARALLEL
+
+        raw = int(DIRECT_REGISTER_PARALLEL)
+    except Exception:
+        try:
+            raw = int(os.environ.get("DIRECT_REGISTER_PARALLEL", "1"))
+        except Exception:
+            raw = 1
+    return _cap_direct_register_parallel(raw)
 
-    mail_client = _resolve_mail_client_or_default(mail_client, acc=acc)
 
+def _attempt_chatgpt_signup_only(mail_client, *, acc=None, out_outcome=None) -> dict:
+    """Run direct ChatGPT signup only; caller decides whether to persist/OAuth the winner."""
+    from autoteam.invite import RegisterBlocked
+
+    mail_client = _resolve_mail_client_or_default(mail_client, acc=acc)
     account_id, email = mail_client.create_temp_email()
     password = random_password()
     signup_profile = generate_signup_profile()
     auth_proxy_url = ""
     playwright_proxy_url = ""
+    local_outcome: dict = {}
 
     def _record_outcome(status, **extra):
+        local_outcome.clear()
+        local_outcome.update(
+            status=status,
+            last_email=email,
+            register_attempts=register_attempts,
+            duplicate_swaps=duplicate_swaps,
+            **extra,
+        )
         if out_outcome is not None:
             out_outcome.clear()
-            out_outcome.update(
-                status=status,
-                last_email=email,
-                register_attempts=register_attempts,
-                duplicate_swaps=duplicate_swaps,
-                **extra,
-            )
+            out_outcome.update(local_outcome)
 
     def _discard_email(reason):
         try:
@@ -3773,6 +3989,20 @@ def create_account_direct(mail_client=None, *, leave_workspace=False, out_outcom
             _record_outcome("ipv6_proxy_unavailable", reason=str(exc))
             return False
 
+    def _build_result(success_value: bool) -> dict:
+        return {
+            "success": bool(success_value),
+            "email": email,
+            "password": password,
+            "account_id": account_id,
+            "session_token": session_token,
+            "signup_profile": signup_profile,
+            "auth_proxy_url": auth_proxy_url,
+            "playwright_proxy_url": playwright_proxy_url,
+            "mail_client": mail_client,
+            "outcome": dict(local_outcome),
+        }
+
     # 注册失败（非 duplicate）最多重试 3 次；duplicate 额外独立上限，防止 CloudMail 异常导致无限换邮箱
     success = False
     session_token = None  # Round 11 四轮 — 注册成功后从 chatgpt.com 抽出���,透给 personal OAuth 跳过 /log-in
@@ -3781,7 +4011,7 @@ def create_account_direct(mail_client=None, *, leave_workspace=False, out_outcom
     register_attempts = 0
     duplicate_swaps = 0
     if not _assign_proxy_or_fail("ipv6_proxy_required_unavailable"):
-        return None
+        return _build_result(False)
 
     while register_attempts < MAX_REGISTER_ATTEMPTS:
         logger.info(
@@ -3826,7 +4056,7 @@ def create_account_direct(mail_client=None, *, leave_workspace=False, out_outcom
                     duplicate_swaps=duplicate_swaps,
                 )
                 _record_outcome("phone_blocked", reason=f"add-phone 手机验证 step={blocked.step}", step=blocked.step)
-                return None
+                return _build_result(False)
             if blocked.is_duplicate:
                 # 邮箱重复 → 换一个全新的临时邮箱再来，不计入 register_attempts
                 duplicate_swaps += 1
@@ -3843,14 +4073,14 @@ def create_account_direct(mail_client=None, *, leave_workspace=False, out_outcom
                         "duplicate_exhausted",
                         reason=f"duplicate 换邮箱 {duplicate_swaps} 次仍失败",
                     )
-                    return None
+                    return _build_result(False)
                 _discard_email("duplicate")
                 account_id, email = mail_client.create_temp_email()
                 password = random_password()
                 signup_profile = generate_signup_profile()
                 logger.info("[直接注册] 已换新临时邮箱: %s", email)
                 if not _assign_proxy_or_fail("ipv6_proxy_required_unavailable_after_duplicate"):
-                    return None
+                    return _build_result(False)
                 continue
             # 其他阻断按普通失败处理
             success = False
@@ -3904,14 +4134,182 @@ def create_account_direct(mail_client=None, *, leave_workspace=False, out_outcom
             duplicate_swaps=duplicate_swaps,
         )
         _record_outcome("register_failed", reason=f"注册 {register_attempts} 次均未进入 Team")
+        return _build_result(False)
+
+    _record_outcome("success", reason="")
+    return _build_result(True)
+
+
+def _direct_signup_mail_client_factory(base_mail_client, acc=None):
+    def _factory(idx: int):
+        if idx <= 0:
+            return base_mail_client
+        try:
+            client = type(base_mail_client)()
+        except Exception:
+            client = _resolve_mail_client_or_default(None, acc=acc)
+        login = getattr(client, "login", None)
+        if callable(login):
+            login()
+        return client
+
+    return _factory
+
+
+def _discard_direct_signup_loser(outcome: dict) -> None:
+    email = outcome.get("email")
+    account_id = outcome.get("account_id")
+    mail_client = outcome.get("mail_client")
+    if email:
+        chatgpt = None
+        try:
+            chatgpt = ChatGPTTeamAPI()
+            chatgpt.start()
+            remove_from_team(chatgpt, email, return_status=True)
+        except Exception as exc:
+            logger.warning("[直接注册] 清理并行 loser Team 席位失败: %s (%s)", email, exc)
+        finally:
+            if chatgpt is not None:
+                try:
+                    chatgpt.stop()
+                except Exception:
+                    pass
+        _release_account_ipv6_proxy(email)
+    if account_id is not None and mail_client is not None:
+        try:
+            mail_client.delete_account(account_id)
+            logger.info("[直接注册] 已丢弃并行 loser 临时邮箱: %s", email)
+        except Exception as exc:
+            logger.warning("[直接注册] 丢弃并行 loser 临时邮箱失败: %s (%s)", email, exc)
+
+
+def _race_chatgpt_signup(mail_client_factory, *, parallel: int, acc=None, out_outcome=None) -> dict:
+    """Run direct signup race and return the first successful signup-only outcome."""
+    import concurrent.futures
+
+    parallel = _cap_direct_register_parallel(parallel)
+    if parallel <= 1:
+        return _attempt_chatgpt_signup_only(mail_client_factory(0), acc=acc, out_outcome=out_outcome)
+
+    logger.info("[直接注册] 启动 %d 个并行注册尝试", parallel)
+    winner: dict | None = None
+    failures: list[dict] = []
+    losers: list[dict] = []
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=parallel, thread_name_prefix="direct-signup") as pool:
+        future_map = {
+            pool.submit(
+                _attempt_chatgpt_signup_only,
+                mail_client_factory(idx),
+                acc=acc,
+            ): idx
+            for idx in range(parallel)
+        }
+        for future in concurrent.futures.as_completed(future_map):
+            try:
+                outcome = future.result()
+            except Exception as exc:
+                logger.warning("[直接注册] 并行 worker 异常: %s", exc)
+                failures.append({"success": False, "outcome": {"status": "exception", "reason": str(exc)}})
+                continue
+            if outcome.get("success") and winner is None:
+                winner = outcome
+                logger.info("[直接注册] 并行获胜: %s", outcome.get("email"))
+            elif outcome.get("success"):
+                losers.append(outcome)
+            else:
+                failures.append(outcome)
+
+    for loser in losers:
+        _discard_direct_signup_loser(loser)
+
+    selected = winner or (failures[-1] if failures else {"success": False, "outcome": {"status": "register_failed"}})
+    if out_outcome is not None:
+        out_outcome.clear()
+        out_outcome.update(selected.get("outcome") or {})
+        out_outcome["direct_register_parallel"] = parallel
+        out_outcome["direct_register_failures"] = len(failures)
+        out_outcome["direct_register_losers"] = len(losers)
+    return selected
+
+
+def create_account_direct(
+    mail_client=None,
+    *,
+    leave_workspace=False,
+    out_outcome=None,
+    acc=None,
+    path_rotator=None,
+    parallel: int | None = None,
+):
+    """
+    直接注册模式（域名已配置自动加入 workspace，不需要邀请）。
+    流程：创建邮箱 → 注册 ChatGPT → 自动加入 workspace → Codex 登录
+    leave_workspace: 加入 workspace 后是否立即退出，转为 personal 模式跑 OAuth。
+    out_outcome:     可选 dict，函数会把最终结局（success/phone_blocked/duplicate_exhausted/register_failed/...）
+                     + 统计信息（register_attempts / duplicate_swaps / last_email / reason）写入，供上游汇总。
+    parallel:        direct signup race 并行度；None 时读取 DIRECT_REGISTER_PARALLEL 并按资源预算降级。
+
+    捕获 RegisterBlocked：
+    - is_phone=True:     当前邮箱已暴露给 OpenAI，立即删邮箱、整个账号放弃（return None）
+    - is_duplicate=True: 换个临时邮箱继续尝试，独立计数不消耗 register_attempts
+    - 其他异常:          归入现有 retry 计数
+
+    Round 12 S4: `mail_client` 改为可选(默认 None) — None 时按 `acc`(可选)走
+    `_get_mail_client_for_account(acc)` 路由;旧调用方显式传 mail_client 时完全
+    保留旧行为(向后兼容)。register-level 多 provider rotation 由 `MAIL_PROVIDER_CHAIN`
+    env 驱动:配置 chain 时 `get_mail_client()` 返回 `FallbackMailProvider`,
+    自动在 mail-API 级失败时降级;邮箱-粒度的 provider 切换通过 `RegisterPathRotator`
+    包装(参见 `autoteam.mail.register_dual_path`),业务调用方可自行编排。
+
+    Round 12 wire-up (M3): `path_rotator` 可选 — 传入 `RegisterPathRotator` 实例时
+    启用邮箱-粒度的 provider 切换(OTP_TIMEOUT / DOMAIN_REJECTED / INVITE_LINK_MISSING
+    自动切下一 provider 重试整个注册流程);None 时走旧逻辑(单 provider + retry 3 次).
+    """
+    # Round 12 wire-up M3 — 邮箱-粒度 provider 切换。
+    if path_rotator is not None:
+        return _create_account_direct_via_rotator(
+            path_rotator,
+            leave_workspace=leave_workspace,
+            out_outcome=out_outcome,
+            acc=acc,
+            parallel=parallel,
+        )
+
+    mail_client = _resolve_mail_client_or_default(mail_client, acc=acc)
+    if parallel is None:
+        parallel = _direct_register_parallel_size()
+    else:
+        parallel = _cap_direct_register_parallel(parallel)
+
+    if parallel <= 1:
+        signup = _attempt_chatgpt_signup_only(mail_client, acc=acc, out_outcome=out_outcome)
+    else:
+        signup = _race_chatgpt_signup(
+            _direct_signup_mail_client_factory(mail_client, acc=acc),
+            parallel=parallel,
+            acc=acc,
+            out_outcome=out_outcome,
+        )
+
+    if not signup.get("success"):
         return None
 
+    email = signup["email"]
+    password = signup["password"]
+    account_id = signup["account_id"]
+    session_token = signup.get("session_token")
+    signup_profile = signup.get("signup_profile")
+    auth_proxy_url = signup.get("auth_proxy_url") or ""
+    playwright_proxy_url = signup.get("playwright_proxy_url") or ""
+    winner_mail_client = signup.get("mail_client") or mail_client
+
     add_account(email, password, cloudmail_account_id=account_id, workspace_account_id=get_chatgpt_account_id() or None)
 
     post_result = _run_post_register_oauth(
         email,
         password,
-        mail_client,
+        winner_mail_client,
         leave_workspace=leave_workspace,
         out_outcome=out_outcome,
         chatgpt_session_token=session_token,
@@ -3924,7 +4322,9 @@ def create_account_direct(mail_client=None, *, leave_workspace=False, out_outcom
     return post_result
 
 
-def _create_account_direct_via_rotator(path_rotator, *, leave_workspace=False, out_outcome=None, acc=None):
+def _create_account_direct_via_rotator(
+    path_rotator, *, leave_workspace=False, out_outcome=None, acc=None, parallel: int | None = None
+):
     """Round 12 wire-up (M3) — 用 RegisterPathRotator 编排邮箱-粒度的 provider 切换。
 
     流程:
@@ -3955,6 +4355,7 @@ def _create_account_direct_via_rotator(path_rotator, *, leave_workspace=False, o
             out_outcome=local_outcome,
             acc=acc,
             path_rotator=None,  # 避免递归
+            parallel=parallel,
         )
         if result_email is None:
             status = local_outcome.get("status") or "register_failed"
@@ -4003,7 +4404,16 @@ def _create_account_direct_via_rotator(path_rotator, *, leave_workspace=False, o
         return None
 
 
-def create_new_account(chatgpt_api, mail_client=None, *, leave_workspace=False, out_outcome=None, acc=None, path_rotator=None):
+def create_new_account(
+    chatgpt_api,
+    mail_client=None,
+    *,
+    leave_workspace=False,
+    out_outcome=None,
+    acc=None,
+    path_rotator=None,
+    parallel: int | None = None,
+):
     """
     创建新账号。优先用直接注册模式（域名自动加入 workspace）。
     chatgpt_api 可为 None（直接注册不需要）。
@@ -4014,8 +4424,12 @@ def create_new_account(chatgpt_api, mail_client=None, *, leave_workspace=False,
     """
     # 先检查 pending invites
     mail_client = _resolve_mail_client_or_default(mail_client, acc=acc)
+    try:
+        from autoteam.config import ROTATE_SKIP_REUSE
+    except Exception:
+        ROTATE_SKIP_REUSE = False
 
-    if chatgpt_api and _chatgpt_session_ready(chatgpt_api):
+    if chatgpt_api and _chatgpt_session_ready(chatgpt_api) and not ROTATE_SKIP_REUSE:
         logger.info("[创建] 先检查 pending invites...")
         completed = _check_pending_invites(
             chatgpt_api,
@@ -4026,6 +4440,8 @@ def create_new_account(chatgpt_api, mail_client=None, *, leave_workspace=False,
         if completed:
             logger.info("[创建] 从 pending invites 完成了 %d 个账号", len(completed))
             return completed[0]
+    elif chatgpt_api and _chatgpt_session_ready(chatgpt_api):
+        logger.info("[创建] ROTATE_SKIP_REUSE 启用：跳过历史 pending invite，只创建新账号")
 
     # 直接注册模式（不需要邀请）
     logger.info("[创建] 使用直接注册模式...")
@@ -4037,6 +4453,7 @@ def create_new_account(chatgpt_api, mail_client=None, *, leave_workspace=False,
         out_outcome=out_outcome,
         acc=acc,
         path_rotator=path_rotator,
+        parallel=parallel,
     )
 
 
@@ -4689,8 +5106,10 @@ def cmd_rotate(target_seats=3, force_auth_repair=False, background_post_sync=Fal
     """
     TARGET = _clamp_team_target_seats(target_seats)
     ACTIVE_TARGET = _pool_active_target(TARGET)
+    started_at = time.time()
 
-    from autoteam.config import AUTO_CHECK_THRESHOLD
+    from autoteam.config import AUTO_CHECK_THRESHOLD, ROTATE_MAX_DURATION, ROTATE_SKIP_REUSE
+    rotate_deadline = started_at + float(ROTATE_MAX_DURATION)
 
     try:
         from autoteam.api import _auto_check_config
@@ -4699,6 +5118,28 @@ def cmd_rotate(target_seats=3, force_auth_repair=False, background_post_sync=Fal
     except ImportError:
         threshold = AUTO_CHECK_THRESHOLD
 
+    def _bump(stage: str) -> None:
+        try:
+            from autoteam.api import bump_task_progress
+
+            bump_task_progress(stage)
+        except Exception:
+            pass
+
+    def _deadline_exceeded(stage: str) -> bool:
+        if time.time() < rotate_deadline:
+            return False
+        logger.warning(
+            "[轮转] 总时长熔断触发 (%s)，已用 %ds，优雅退出本轮 rotate",
+            stage,
+            int(time.time() - started_at),
+        )
+        return True
+
+    skip_reuse = bool(ROTATE_SKIP_REUSE)
+    if skip_reuse:
+        logger.info("[轮转] ROTATE_SKIP_REUSE 启用：跳过旧号复用，优先释放不可用占席子号后创建新号")
+
     chatgpt = None
     mail_client = None
     # 按账号 mail_provider 复用 mail client(同 provider 只 login 一次,避免 N 个号 N 次握手)
@@ -4766,9 +5207,11 @@ def cmd_rotate(target_seats=3, force_auth_repair=False, background_post_sync=Fal
             return ensure_mail()
 
     logger.info("[1/5] 同步 Team 状态...")
+    _bump("rotate:sync_team")
     sync_account_states()
 
     logger.info("[2/5] 检查额度...")
+    _bump("rotate:check_quota")
     cmd_check()
 
     # Round 12 S5 — 预测式抢先替换(可选,默认关).
@@ -4814,29 +5257,41 @@ def cmd_rotate(target_seats=3, force_auth_repair=False, background_post_sync=Fal
         # 移出所有 exhausted 账号（包括之前已标记的）
         all_accounts = load_accounts()
         all_exhausted = [
-            a for a in all_accounts if a["status"] == STATUS_EXHAUSTED and not _is_main_account_email(a.get("email"))
+            a
+            for a in all_accounts
+            if _is_replaceable_pool_blocker(a)
         ]
         initial_api_count = -1
         removed_now = 0
         already_absent_count = 0
 
         if all_exhausted:
-            logger.info("[3/5] 移出 %d 个额度用完的账号...", len(all_exhausted))
+            logger.info("[3/5] 移出 %d 个不可用占席账号...", len(all_exhausted))
             ensure_chatgpt()
             initial_api_count = get_team_member_count(chatgpt)
             for acc in all_exhausted:
+                if _deadline_exceeded("rotate:remove_blockers"):
+                    break
                 email = acc["email"]
+                reason = _replaceable_pool_blocker_reason(acc) or "replaceable_pool_blocker"
                 if not _chatgpt_session_ready(chatgpt):
                     chatgpt.start()
                 remove_status = remove_from_team(chatgpt, email, return_status=True)
                 if remove_status in ("removed", "already_absent"):
-                    update_account(email, status=STATUS_STANDBY)
+                    update_account(email, status=STATUS_STANDBY, _reason=reason)
                     if remove_status == "removed":
                         removed_now += 1
-                        logger.info("[3/5] %s → standby（已从 Team 移出）", email)
+                        logger.info("[3/5] %s → standby（已从 Team 移出）| reason=%s", email, reason)
+                        _wait_for_remote_capacity_after_removal(
+                            chatgpt,
+                            target=TARGET,
+                            removed_email=email,
+                            timeout=24,
+                            stage_label="[3/5]",
+                        )
                     else:
                         already_absent_count += 1
-                        logger.info("[3/5] %s → standby（远端已不存在）", email)
+                        logger.info("[3/5] %s → standby（远端已不存在）| reason=%s", email, reason)
         else:
             logger.info("[3/5] 无需移出账号")
         if not chatgpt or not _chatgpt_session_ready(chatgpt):
@@ -4880,7 +5335,12 @@ def cmd_rotate(target_seats=3, force_auth_repair=False, background_post_sync=Fal
                 # 只移除本地管理的账号，优先移除额度最低的
                 all_accs = load_accounts()
                 local_active = [
-                    a for a in all_accs if a["status"] == STATUS_ACTIVE and not _is_main_account_email(a.get("email"))
+                    a
+                    for a in all_accs
+                    if a["status"] == STATUS_ACTIVE
+                    and not _is_main_account_email(a.get("email"))
+                    and not is_account_disabled(a)
+                    and not _is_protected_local_credential_seat(a)
                 ]
                 # 按额度排序，额度低的优先移除
                 local_active.sort(key=lambda a: 100 - (a.get("last_quota") or {}).get("primary_pct", 0))
@@ -4916,7 +5376,11 @@ def cmd_rotate(target_seats=3, force_auth_repair=False, background_post_sync=Fal
         # 注: 不截断到 vacancies — 旧行为会迭代全部 standby,遇到不可复用的(skipped_auto/
         # skipped_quota)就继续看下一个,直到 filled >= vacancies 才 break. 截断会导致
         # 第一个候选若 skipped 则放弃后续可用候选,破坏 test_cmd_rotate_skips_google_accounts.
-        candidates = list(standby_list)
+        if skip_reuse:
+            logger.info("[4/5] ROTATE_SKIP_REUSE 启用：跳过 standby 复用，直接创建新账号补位")
+            candidates = []
+        else:
+            candidates = list(standby_list)
 
         def _chatgpt_provider():
             if not chatgpt or not _chatgpt_session_ready(chatgpt):
@@ -4958,6 +5422,8 @@ def cmd_rotate(target_seats=3, force_auth_repair=False, background_post_sync=Fal
                 if cancel_signal.is_cancelled():
                     logger.warning("[轮转] 收到取消请求,中止 standby 复用阶段")
                     break
+                if _deadline_exceeded("rotate:standby_reuse"):
+                    break
                 if filled >= vacancies:
                     break
                 result = _process(acc)
@@ -5038,11 +5504,29 @@ def cmd_rotate(target_seats=3, force_auth_repair=False, background_post_sync=Fal
                 if cancel_signal.is_cancelled():
                     logger.warning("[轮转] 收到取消请求,已创建 %d/%d 个新号", i, remaining)
                     break
+                if _deadline_exceeded("rotate:create_new"):
+                    break
                 logger.info("[5/5] 创建第 %d/%d 个...", i + 1, remaining)
                 if not chatgpt or not _chatgpt_session_ready(chatgpt):
                     ensure_chatgpt()
-                if create_new_account(chatgpt, ensure_mail()):
+                created_email = create_new_account(chatgpt, ensure_mail())
+                if created_email and (
+                    not isinstance(created_email, str)
+                    or _validate_managed_account_operational(
+                        created_email,
+                        threshold=threshold,
+                        stage_label="[轮转验收]",
+                        chatgpt_api=chatgpt,
+                    )
+                ):
                     current_count += 1
+                elif created_email:
+                    logger.warning("[5/5] 新���号未通过运行验收，释放席位并丢弃: %s", created_email)
+                    if not _chatgpt_session_ready(chatgpt):
+                        chatgpt.start()
+                    remove_status = remove_from_team(chatgpt, created_email, return_status=True)
+                    if remove_status in ("removed", "already_absent"):
+                        update_account(created_email, status=STATUS_STANDBY, _reason="new_account_not_ready")
 
         if not chatgpt or not _chatgpt_session_ready(chatgpt):
             ensure_chatgpt()
@@ -5346,7 +5830,7 @@ def get_team_member_count(chatgpt_api):
     return len(members)
 
 
-def cmd_fill(target=3, leave_workspace=False, *, post_sync=True, print_status=True):
+def cmd_fill(target=3, leave_workspace=False, *, post_sync=True, print_status=True, direct_parallel: int | None = None):
     """
     补位流程。
     leave_workspace=False: 补满 Team 席位到 target（原行为），优先复用 standby 旧号
@@ -5357,6 +5841,7 @@ def cmd_fill(target=3, leave_workspace=False, *, post_sync=True, print_status=Tr
     if leave_workspace:
         return _cmd_fill_personal(target)
     target = _clamp_team_target_seats(target)
+    from autoteam.config import AUTO_CHECK_THRESHOLD, ROTATE_SKIP_REUSE
 
     chatgpt = ChatGPTTeamAPI()
     chatgpt.start()
@@ -5377,11 +5862,17 @@ def cmd_fill(target=3, leave_workspace=False, *, post_sync=True, print_status=Tr
             return
 
         logger.info("[填充] 需要添加 %d 个账号", need)
-        standby_list = [
-            a
-            for a in get_standby_accounts()
-            if a.get("_quota_recovered") and not _is_main_account_email(a.get("email"))
-        ]
+        if ROTATE_SKIP_REUSE:
+            logger.info("[填充] ROTATE_SKIP_REUSE 启用：不复用旧账号，只创建新账号补位")
+            standby_list = []
+        else:
+            standby_list = [
+                a
+                for a in get_standby_accounts()
+                if a.get("_quota_recovered")
+                and not _is_main_account_email(a.get("email"))
+                and not is_account_disabled(a)
+            ]
         standby_index = 0
 
         from autoteam import cancel_signal
@@ -5416,7 +5907,27 @@ def cmd_fill(target=3, leave_workspace=False, *, post_sync=True, print_status=Tr
                 logger.info("[填充] 创建新账号...")
                 if not _chatgpt_session_ready(chatgpt):
                     chatgpt.start()
-                added = create_new_account(chatgpt, mail_client)
+                if direct_parallel is None:
+                    created_email = create_new_account(chatgpt, mail_client)
+                else:
+                    created_email = create_new_account(chatgpt, mail_client, parallel=direct_parallel)
+                if created_email and (
+                    not isinstance(created_email, str)
+                    or _validate_managed_account_operational(
+                        created_email,
+                        threshold=AUTO_CHECK_THRESHOLD,
+                        stage_label="[填充验收]",
+                        chatgpt_api=chatgpt,
+                    )
+                ):
+                    added = created_email
+                elif created_email:
+                    logger.warning("[填充] 新账号未通过运行验收，释放席位并丢弃: %s", created_email)
+                    if not _chatgpt_session_ready(chatgpt):
+                        chatgpt.start()
+                    remove_status = remove_from_team(chatgpt, created_email, return_status=True)
+                    if remove_status in ("removed", "already_absent"):
+                        update_account(created_email, status=STATUS_STANDBY, _reason="fill_new_account_not_ready")
 
             if not added:
                 logger.warning("[填充] 本轮补位失败，第 %d/%d 个空缺仍未填上", i + 1, need)

src/autoteam/multi_master.py

@@ -240,7 +240,13 @@ def _run_fill_for_owner(owner: dict[str, Any], target_seats: int, direct_paralle
         account_id=owner.get("account_id") or "",
         workspace_name=owner.get("workspace_name") or "",
     ):
-        cmd_fill(target_seats, leave_workspace=False, post_sync=False, print_status=False)
+        cmd_fill(
+            target_seats,
+            leave_workspace=False,
+            post_sync=False,
+            print_status=False,
+            direct_parallel=direct_parallel,
+        )
     return {
         "workspace_id": owner.get("id"),
         "admin_email": owner.get("admin_email"),

tests/unit/test_api_status.py

@@ -249,6 +249,48 @@ def test_auto_check_cooldown_keeps_full_team_from_refilling(tmp_path, monkeypatc
     assert started == []
 
 
+def test_auto_check_cooldown_allows_full_team_blocker_replacement(monkeypatch):
+    started = []
+
+    def fake_start_task(command, func, params, *args, **kwargs):
+        started.append((command, params, args, kwargs))
+
+    monkeypatch.setattr(api, "_auto_fill_last_trigger_ts", time.time())
+    monkeypatch.setattr(api, "_auto_check_config", {"interval": 0, "target_seats": 3, "threshold": 10, "min_low": 1})
+    monkeypatch.setattr(api, "log_runtime_resource_snapshot", lambda *args, **kwargs: {})
+    monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False)
+    monkeypatch.setattr(api, "_auto_check_team_member_count", lambda *args, **kwargs: 3)
+    monkeypatch.setattr(api, "_start_task", fake_start_task)
+    monkeypatch.setattr(
+        "autoteam.accounts.load_accounts",
+        lambda: [
+            {"email": "healthy@example.com", "status": "active", "auth_file": ""},
+            {"email": "blocked@example.com", "status": "auth_invalid", "auth_file": ""},
+        ],
+    )
+
+    stop_event = threading.Event()
+    restart_event = threading.Event()
+    wait_calls = {"count": 0}
+
+    def fake_wait(_seconds):
+        wait_calls["count"] += 1
+        return wait_calls["count"] > 1
+
+    monkeypatch.setattr(stop_event, "wait", fake_wait)
+    monkeypatch.setattr(api, "_auto_check_stop", stop_event)
+    monkeypatch.setattr(api, "_auto_check_restart", restart_event)
+
+    api._auto_check_loop()
+
+    assert len(started) == 1
+    command, params, args, kwargs = started[0]
+    assert command == "auto-fill"
+    assert params == {"target_seats": 3}
+    assert args == (3,)
+    assert kwargs == {"background_post_sync": True}
+
+
 def test_sanitize_account_keeps_exportable_main_account_active_without_live_quota(tmp_path, monkeypatch):
     main_email = "owner@example.com"
     auth_file = tmp_path / "codex-main.json"

tests/unit/test_free_registration_hardening.py

@@ -226,6 +226,81 @@ def test_create_account_direct_reuses_signup_profile_for_register_and_oauth(monk
     assert captured["oauth"]["leave_workspace"] is True
 
 
+def test_create_account_direct_races_signup_workers_and_uses_winner(monkeypatch):
+    calls = []
+    added = []
+    out = {}
+
+    class _RaceMailClient:
+        counter = 0
+        provider_name = "race"
+
+        def __init__(self):
+            type(self).counter += 1
+            self.idx = type(self).counter
+
+        def login(self):
+            return None
+
+        def create_temp_email(self):
+            return f"mail-{self.idx}", f"user-{self.idx}@example.com"
+
+        def delete_account(self, *_args, **_kwargs):
+            return None
+
+    def fake_register_once(_mail_client, email, password, *, cloudmail_account_id=None, signup_profile=None):
+        calls.append(
+            {
+                "email": email,
+                "password": password,
+                "cloudmail_account_id": cloudmail_account_id,
+                "signup_profile": signup_profile,
+            }
+        )
+        return email == "user-2@example.com", f"token-for-{email}"
+
+    monkeypatch.setattr(manager_mod, "_ensure_account_ipv6_proxy", lambda _email: ("", ""))
+    monkeypatch.setattr(manager_mod, "_release_account_ipv6_proxy", lambda _email: None)
+    monkeypatch.setattr(manager_mod, "_register_direct_once", fake_register_once)
+    monkeypatch.setattr(manager_mod, "_is_email_in_team", lambda _email: False)
+    monkeypatch.setattr(manager_mod, "time", manager_mod.time)
+    monkeypatch.setattr(manager_mod.time, "sleep", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(manager_mod, "add_account", lambda *args, **kwargs: added.append((args, kwargs)))
+    monkeypatch.setattr(manager_mod, "get_chatgpt_account_id", lambda: "workspace-account")
+    monkeypatch.setattr(
+        manager_mod,
+        "_run_post_register_oauth",
+        lambda email, *_args, **_kwargs: email,
+    )
+
+    result = manager_mod.create_account_direct(
+        mail_client=_RaceMailClient(),
+        out_outcome=out,
+        parallel=3,
+    )
+
+    assert result == "user-2@example.com"
+    assert sorted({call["email"] for call in calls}) == [
+        "user-1@example.com",
+        "user-2@example.com",
+        "user-3@example.com",
+    ]
+    assert sum(1 for call in calls if call["email"] == "user-2@example.com") == 1
+    assert added[0][0][0] == "user-2@example.com"
+    assert out["direct_register_parallel"] == 3
+    assert out["direct_register_failures"] == 2
+
+
+def test_direct_register_parallel_downgrades_when_memory_high(monkeypatch):
+    monkeypatch.setenv("AUTOTEAM_REGISTER_PARALLEL_MEMORY_WARN_RATIO", "0.70")
+    monkeypatch.setattr(
+        "autoteam.runtime_resources.collect_runtime_resource_snapshot",
+        lambda: {"cgroup_memory_usage_ratio": 0.95, "browser_process_live": 0},
+    )
+
+    assert manager_mod._cap_direct_register_parallel(4) == 1
+
+
 def test_run_post_register_oauth_passes_signup_profile_to_team_oauth(monkeypatch):
     profile = SignupProfile(
         full_name="Mia Wilson",

tests/unit/test_manager_fill.py

@@ -22,10 +22,13 @@ class _FakeMailClient:
 
 
 def test_cmd_fill_tries_other_reusable_accounts_before_creating_new(monkeypatch):
+    import autoteam.config as config
+
     chatgpt = _FakeChatGPT()
     count_values = iter([2, 3])
     events = []
 
+    monkeypatch.setattr(config, "ROTATE_SKIP_REUSE", False)
     monkeypatch.setattr(manager, "ChatGPTTeamAPI", lambda: chatgpt)
     monkeypatch.setattr(manager, "CloudMailClient", lambda: _FakeMailClient())
     monkeypatch.setattr(manager, "get_team_member_count", lambda _chatgpt: next(count_values))
@@ -63,10 +66,13 @@ def test_cmd_fill_tries_other_reusable_accounts_before_creating_new(monkeypatch)
 
 
 def test_cmd_fill_skips_google_accounts_during_auto_reuse(monkeypatch):
+    import autoteam.config as config
+
     chatgpt = _FakeChatGPT()
     count_values = iter([2, 3])
     events = []
 
+    monkeypatch.setattr(config, "ROTATE_SKIP_REUSE", False)
     monkeypatch.setattr(manager, "ChatGPTTeamAPI", lambda: chatgpt)
     monkeypatch.setattr(manager, "CloudMailClient", lambda: _FakeMailClient())
     monkeypatch.setattr(manager, "get_team_member_count", lambda _chatgpt: next(count_values))
@@ -109,3 +115,41 @@ def test_auto_reuse_skip_reason_detects_google_provider_and_gmail():
         == "Google 登录账号暂不支持自动复用"
     )
     assert manager._auto_reuse_skip_reason({"email": "user@example.com"}) is None
+
+
+def test_cmd_fill_passes_direct_parallel_and_releases_failed_validation(monkeypatch):
+    import autoteam.config as config
+
+    chatgpt = _FakeChatGPT()
+    counts = iter([2, 2])
+    events = []
+
+    monkeypatch.setattr(config, "ROTATE_SKIP_REUSE", True)
+    monkeypatch.setattr(manager, "ChatGPTTeamAPI", lambda: chatgpt)
+    monkeypatch.setattr(manager, "CloudMailClient", lambda: _FakeMailClient())
+    monkeypatch.setattr(manager, "get_team_member_count", lambda _chatgpt: next(counts))
+    monkeypatch.setattr(manager, "get_standby_accounts", lambda: [])
+    monkeypatch.setattr(
+        manager,
+        "create_new_account",
+        lambda _chatgpt, _mail, *, parallel=None: events.append(("create", parallel)) or "new@example.com",
+    )
+    monkeypatch.setattr(manager, "_validate_managed_account_operational", lambda *_args, **_kwargs: False)
+    monkeypatch.setattr(
+        manager,
+        "remove_from_team",
+        lambda _chatgpt, email, *, return_status=False, **_kwargs: events.append(("remove", email)) or "removed",
+    )
+    monkeypatch.setattr(
+        manager,
+        "update_account",
+        lambda email, **kwargs: events.append(("update", email, kwargs.get("status"), kwargs.get("_reason"))),
+    )
+    monkeypatch.setattr(manager, "sync_to_cpa", lambda: events.append(("sync", None)))
+    monkeypatch.setattr(manager, "cmd_status", lambda: events.append(("status", None)))
+
+    manager.cmd_fill(target=3, direct_parallel=3)
+
+    assert ("create", 3) in events
+    assert ("remove", "new@example.com") in events
+    assert ("update", "new@example.com", manager.STATUS_STANDBY, "fill_new_account_not_ready") in events

tests/unit/test_manager_rotate.py

@@ -22,10 +22,13 @@ class _FakeMailClient:
 
 
 def test_cmd_rotate_skips_google_accounts_during_auto_reuse(monkeypatch):
+    import autoteam.config as config
+
     chatgpt = _FakeChatGPT()
     count_values = iter([2, 3])
     events = []
 
+    monkeypatch.setattr(config, "ROTATE_SKIP_REUSE", False)
     monkeypatch.setattr(manager, "sync_account_states", lambda: events.append(("sync_account_states", None)))
     monkeypatch.setattr(manager, "cmd_check", lambda: events.append(("cmd_check", None)))
     monkeypatch.setattr(manager, "ChatGPTTeamAPI", lambda: chatgpt)
@@ -98,3 +101,76 @@ def test_cmd_rotate_can_defer_final_sync_when_running_as_api_task(monkeypatch):
         ("cmd_check", None),
         ("schedule_post_sync", "[轮转]"),
     ]
+
+
+def test_replaceable_pool_blocker_reason_reports_concrete_evidence():
+    assert (
+        manager._replaceable_pool_blocker_reason(
+            {"email": "missing@example.com", "status": manager.STATUS_ACTIVE, "auth_file": ""}
+        )
+        == "missing_auth"
+    )
+    assert (
+        manager._replaceable_pool_blocker_reason(
+            {"email": "invalid@example.com", "status": manager.STATUS_AUTH_INVALID}
+        )
+        == "auth_invalid"
+    )
+    assert (
+        manager._replaceable_pool_blocker_reason(
+            {"email": "exhausted@example.com", "status": manager.STATUS_EXHAUSTED}
+        )
+        == "quota_exhausted"
+    )
+
+
+def test_cmd_rotate_removes_replaceable_blocker_before_creating_replacement(monkeypatch):
+    import autoteam.config as config
+
+    chatgpt = _FakeChatGPT()
+    accounts = [{"email": "blocked@example.com", "status": manager.STATUS_ACTIVE, "auth_file": ""}]
+    counts = iter([3, 2, 2, 3])
+    events = []
+
+    monkeypatch.setattr(config, "ROTATE_SKIP_REUSE", True)
+    monkeypatch.setattr(manager, "sync_account_states", lambda: events.append(("sync_account_states", None)))
+    monkeypatch.setattr(manager, "cmd_check", lambda: events.append(("cmd_check", None)))
+    monkeypatch.setattr(manager, "ChatGPTTeamAPI", lambda: chatgpt)
+    monkeypatch.setattr(manager, "CloudMailClient", lambda: _FakeMailClient())
+    monkeypatch.setattr(manager, "load_accounts", lambda: accounts)
+    monkeypatch.setattr(manager, "get_team_member_count", lambda _chatgpt: next(counts))
+    monkeypatch.setattr(manager, "get_standby_accounts", lambda: [])
+    monkeypatch.setattr(manager, "time", manager.time)
+    monkeypatch.setattr(manager.time, "sleep", lambda *_args, **_kwargs: None)
+
+    def fake_update(email, **kwargs):
+        events.append(("update", email, kwargs.get("status"), kwargs.get("_reason")))
+        for acc in accounts:
+            if acc["email"] == email:
+                acc.update(kwargs)
+
+    def fake_remove(_chatgpt, email, *, return_status=False, **_kwargs):
+        events.append(("remove", email))
+        return "removed" if return_status else True
+
+    def fake_create(_chatgpt, _mail):
+        events.append(("create", None))
+        accounts.append(
+            {
+                "email": "new@example.com",
+                "status": manager.STATUS_ACTIVE,
+                "auth_file": "auth.json",
+            }
+        )
+        return "new@example.com"
+
+    monkeypatch.setattr(manager, "update_account", fake_update)
+    monkeypatch.setattr(manager, "remove_from_team", fake_remove)
+    monkeypatch.setattr(manager, "create_new_account", fake_create)
+    monkeypatch.setattr(manager, "_validate_managed_account_operational", lambda *_args, **_kwargs: True)
+    monkeypatch.setattr(manager, "sync_to_cpa", lambda: events.append(("sync_to_cpa", None)))
+
+    manager.cmd_rotate(target_seats=3)
+
+    assert events.index(("remove", "blocked@example.com")) < events.index(("create", None))
+    assert ("update", "blocked@example.com", manager.STATUS_STANDBY, "missing_auth") in events

tests/unit/test_multi_master.py

@@ -129,7 +129,7 @@ def test_run_multi_master_fill_isolates_owner_failures(tmp_path, monkeypatch):
     seen = []
     lock = threading.Lock()
 
-    def fake_cmd_fill(target, leave_workspace=False, post_sync=True, print_status=True):
+    def fake_cmd_fill(target, leave_workspace=False, post_sync=True, print_status=True, direct_parallel=None):
         with lock:
             seen.append(
                 (
@@ -139,6 +139,7 @@ def test_run_multi_master_fill_isolates_owner_failures(tmp_path, monkeypatch):
                     leave_workspace,
                     post_sync,
                     print_status,
+                    direct_parallel,
                 )
             )
         if admin_state.get_admin_email() == "owner-b@example.com":
@@ -157,8 +158,8 @@ def test_run_multi_master_fill_isolates_owner_failures(tmp_path, monkeypatch):
 
     assert result["status"] == "partial_failed"
     assert {row[0] for row in seen} == {"owner-a@example.com", "owner-b@example.com"}
-    assert ("owner-a@example.com", UUID_A, 3, False, False, False) in seen
-    assert ("owner-b@example.com", UUID_B, 3, False, False, False) in seen
+    assert ("owner-a@example.com", UUID_A, 3, False, False, False, 1) in seen
+    assert ("owner-b@example.com", UUID_B, 3, False, False, False, 1) in seen
     assert result["post_sync"] == {"skipped": True}
     owner_b = next(owner for owner in result["owners"] if owner["admin_email"] == "owner-b@example.com")
     assert owner_b["status"] == "failed"

tests/unit/test_reconcile_anomalies.py

@@ -242,7 +242,7 @@ def test_reconcile_dry_run_includes_over_cap_predictions(tmp_path, monkeypatch):
     """HIGH-1:dry_run 必须**预测**第二轮 over-cap 受害者填进 over_cap_kicked,
     但不能 GET /users 第二次,也不能调 remove_from_team / update_account。
     """
-    # 5 子号 + 1 主号 → 超员 1
+    # 5 子号 + 1 主号；当前硬约束为 1 母 + 2 子，因此超员 3。
     auth_ok = tmp_path / "codex-good@example.com-team-1.json"
     auth_ok.write_text("{}", encoding="utf-8")
 
@@ -287,11 +287,11 @@ def test_reconcile_dry_run_includes_over_cap_predictions(tmp_path, monkeypatch):
     user_get_calls = [c for c in fetch_calls if c[0] == "GET" and c[1].endswith("/users")]
     assert len(user_get_calls) == 1, f"dry_run should not refetch /users, got {fetch_calls}"
 
-    # 必须有 1 个 over_cap_kicked 预测项
-    assert len(result["over_cap_kicked"]) == 1, f"expected 1 prediction, got {result['over_cap_kicked']}"
+    # 必须预测踢到 2 个子号硬上限以内。
+    assert len(result["over_cap_kicked"]) == 3, f"expected 3 predictions, got {result['over_cap_kicked']}"
     # 受害者按 _priority 升序,active 按 p_remain (100-primary_pct) 升序 → primary_pct 最高的先 kick
     # u5 primary_pct=50 → p_remain=50 是最低 remain → 第一个被 kick
-    assert result["over_cap_kicked"] == ["u5@example.com"]
+    assert result["over_cap_kicked"] == ["u5@example.com", "u4@example.com", "u3@example.com"]
 
 
 def test_reconcile_priority_keeps_ghost_when_kick_disabled(tmp_path, monkeypatch):
@@ -301,7 +301,7 @@ def test_reconcile_priority_keeps_ghost_when_kick_disabled(tmp_path, monkeypatch
     auth_ok = tmp_path / "codex-keep@example.com-team-1.json"
     auth_ok.write_text("{}", encoding="utf-8")
 
-    # 4 个 active 账号 + 1 个 ghost(本地无记录) = 5,超员 1
+    # 4 个 active 账号 + 1 个 ghost(本地无记录) = 5；ghost 保留时需从本地 active 中踢 3 个。
     accounts = [
         {
             "email": f"u{i}@example.com",
@@ -343,13 +343,13 @@ def test_reconcile_priority_keeps_ghost_when_kick_disabled(tmp_path, monkeypatch
     assert "ghost-extra@example.com" in result["ghost_seen"]
     assert "ghost-extra@example.com" not in result["ghost_kicked"]
 
-    # 第二轮 over-cap 1 个:必须挑 u1-u4 中的一个 active(p_remain=50, priority=(5,50))
+    # 第二轮 over-cap:必须挑 u1-u4 中的 active(p_remain=50, priority=(5,50))
     # 而不是 ghost-extra(被强制排到 (99, 0) 末尾)
     assert "ghost-extra@example.com" not in result["over_cap_kicked"], (
         f"ghost must not be over-cap kicked when RECONCILE_KICK_GHOST=False: {result['over_cap_kicked']}"
     )
-    assert len(result["over_cap_kicked"]) == 1
-    assert result["over_cap_kicked"][0] in {f"u{i}@example.com" for i in range(1, 5)}
+    assert len(result["over_cap_kicked"]) == 3
+    assert set(result["over_cap_kicked"]).issubset({f"u{i}@example.com" for i in range(1, 5)})
 
 
 def test_reconcile_find_team_auth_file_rejects_personal_plan(tmp_path, monkeypatch):