feat(round-12 S7): multi team workspace pool + auto failover

新增 src/autoteam/workspace_pool.py — WorkspacePool 持久化 K 个 team workspace
互为冷备(active=1, warm/cold N-1)。主 workspace 连续探测失败 ≥ MASTER_HEALTH_FAIL_THRESHOLD
(默认 3)秒切到 warm。register / set_active / mark_unhealthy / mark_healthy 全部
追加 transition_log,与 S1 default_machine 风格一致(原子写 + .bak 回滚 + lock-外
事件总线)。

admin_state.get_admin_email / get_chatgpt_account_id 改为路由 active workspace
的 admin_email / account_id;pool 空时回退原 state.json 读路径,单 workspace 模
式向后兼容。master_health 新增 apply_pool_health_signal helper:probe healthy
→ mark_healthy 重置 fail_count;subscription_cancelled / auth_invalid /
workspace_missing / role_not_owner → mark_unhealthy 累计触发故障切换;
network_error 视为瞬时不计入。cpa_sync 新增 get_active_sync_target helper +
sync_main_codex_to_cpa 上报当前 active workspace identity。

测试 35 case 覆盖:基础注册/激活/idempotent/重复 id 拒绝、set_active 切换/降级、
mark_unhealthy 阈值前不切换 + 阈值触发 failover + 无候选 → active=None +
force_failover、failover 候选选最旧 warm、mark_healthy 重置、transition_log
完整、原子写 .bak 回滚(模拟 os.replace 失败)、admin_state 路由 + 兜底、
master_health 集成 4 路径、subscriber 异常吞掉、5-thread 并发注册仅 1 active、
schema 不兼容 + JSON 损坏 + 字段非法值 normalize、cpa_sync get_active_sync_target。

向后兼容:workspaces.json 不存在时从 state.json (admin_email + UUID account_id)
自动 seed 为单一 active workspace;现有调用方零改动。

DoD:
- ruff check src/ tests/ 全绿
- pytest 699 passed (664 baseline + 35 新增,无退化)
- 所有改动文件已提交,严格 git add 文件名,未用 -A

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

.trellis/tasks/05-11-s7-multi-workspace-pool/check.jsonl

@@ -0,0 +1,5 @@
+{"file": ".trellis/tasks/05-11-s7-multi-workspace-pool/prd.md", "reason": "DoD + 不变量 I1~I7 验收清单"}
+{"file": "src/autoteam/workspace_pool.py", "reason": "新增核心模块 — 验证 register / set_active / mark_unhealthy / 自动 failover / 原子写 / 向后兼容 seed"}
+{"file": "src/autoteam/admin_state.py", "reason": "路由改造 — 验证 pool 为空时回退 state.json，签名不变"}
+{"file": "src/autoteam/master_health.py", "reason": "失败阈值累计 + 调 pool.mark_unhealthy 触发切换"}
+{"file": "tests/unit/test_round12_s7_workspace_pool.py", "reason": "≥ 15 case：注册/切换/故障阈值/原子写回滚/向后兼容/边界"}

.trellis/tasks/05-11-s7-multi-workspace-pool/implement.jsonl

@@ -0,0 +1,5 @@
+{"file": ".trellis/tasks/05-11-upstream-align-register-multimail-frontend-refresh/prd.md", "reason": "父 PRD — Q2 创新方向 5 多 workspace 池化 scope + 子任务依赖图 + round-12 波次范围"}
+{"file": "src/autoteam/admin_state.py", "reason": "改造目标 1：get_admin_email / get_chatgpt_account_id 路由到 WorkspacePool.get_active()，向后兼容回退 state.json"}
+{"file": "src/autoteam/master_health.py", "reason": "改造目标 2：失败路径（subscription_cancelled / auth_invalid / network_error）调 pool.mark_unhealthy 触发故障切换"}
+{"file": "src/autoteam/cpa_sync.py", "reason": "改造目标 3：sync_to_cpa / sync_main_codex_to_cpa 同步目标跟随 active workspace（感知失败切换后立即新 workspace 推送）"}
+{"file": "src/autoteam/account_state.py", "reason": "S1 状态机参考实现 — atomic 写盘 + .bak 回滚 + subscribe 事件总线 + IllegalTransitionError 风格对齐"}

.trellis/tasks/05-11-s7-multi-workspace-pool/prd.md

@@ -0,0 +1,123 @@
+# S7 — 多 team workspace 池化（冷备 + 秒级故障切换）
+
+> **Round 12 子任务**，父任务 `05-11-upstream-align-register-multimail-frontend-refresh` 中
+> 「Q2 创新方向 5 — 多 team workspace 池化」的实施单元。
+
+## 1. 背景
+
+当前 AutoTeam 仅维持一个 active team workspace（admin_state.json 单 admin email + account_id）。
+单 workspace 故障（被封 / 订阅 cancel / Cloudflare 全域风控）整套自动化即停摆。
+
+S7 引入 workspace 池：注册 K 个 workspace（active=1 + warm/cold N-1 互为冷备），主 workspace
+连续探测失败超阈值时秒切到下一个 warm，cold 升 warm，整体不停摆。
+
+## 2. 范围
+
+### MVP（本任务）
+
+1. 新建 `src/autoteam/workspace_pool.py`（持久化 `workspaces.json` + `WorkspacePool`）
+2. `admin_state.get_admin_email/get_chatgpt_account_id` 路由到 `WorkspacePool.get_active()`
+3. `master_health` 失败路径调 `pool.mark_unhealthy()` 触发自动切换（阈值 env `MASTER_HEALTH_FAIL_THRESHOLD=3`）
+4. `cpa_sync` 同步目标改为 active workspace
+5. 测试 ≥ 15 case（mock 故障注入；禁止真实 OpenAI 调用）
+
+### 非范围
+
+- 不实现真实 OpenAI workspace 创建 API（user 自建 workspace 后手动 register）
+- 不动 mail provider / account_state / manager.py 主流程
+- 不做 active workspace 内 accounts 数据隔离（accounts.json 仍是全局视图）
+- 不动 web 前端（后续基于 SSE 流展示）
+
+## 3. 数据 schema
+
+### `workspaces.json`
+
+```json
+{
+  "schema_version": 1,
+  "active": "ws-aaa",
+  "workspaces": [
+    {
+      "id": "ws-aaa",
+      "admin_email": "ad1@example.com",
+      "account_id": "00000000-0000-0000-0000-000000000001",
+      "tier": "active",
+      "status": "healthy",
+      "fail_count": 0,
+      "last_check_ts": 1715300000.0,
+      "registered_at": 1715200000.0,
+      "transition_log": [
+        {"ts": 1715200000.0, "from": null, "to": "warm", "reason": "register"},
+        {"ts": 1715200100.0, "from": "warm", "to": "active", "reason": "promoted_no_active"}
+      ]
+    }
+  ]
+}
+```
+
+### tier / status 取值
+
+- **tier**: `active` | `warm` | `cold` （任一时刻 `active` 至多 1 个）
+- **status**: `healthy` | `unhealthy` | `unknown`
+
+## 4. 不变量（强制执行）
+
+- **I1**: 任意时刻 workspaces[*].tier == "active" 至多 1 个
+- **I2**: `mark_unhealthy` 命中 active 且 fail_count ≥ threshold → 自动 promote 一个 warm/cold 至 active；原 active 降为 cold + status=unhealthy
+- **I3**: 写盘原子（snapshot `.bak` → write `.tmp` → `os.replace`），失败回滚
+- **I4**: 单 workspace 模式向后兼容：`workspaces.json` 不存在时，从 `state.json` 读 admin_email/account_id 自动 seed 为单一 active workspace；外部 API（`get_admin_email` 等）签名不变
+- **I5**: `register / set_active / mark_unhealthy` 全部追加 `transition_log` 条目
+- **I6**: 与 S1 default_machine 风格一致（lock 内修改、lock 外发布事件）
+- **I7**: 永不抛未捕获异常（与 master_health.py M-I1 看齐）
+
+## 5. 关键 API
+
+```python
+class WorkspacePool:
+    def register(self, workspace_id: str, admin_email: str, account_id: str, tier: str = "warm") -> dict
+    def get_active(self) -> dict | None         # 当前 active 单条；可能 None
+    def set_active(self, workspace_id: str) -> dict
+    def mark_unhealthy(self, workspace_id: str, reason: str) -> dict | None
+    def list_all(self) -> list[dict]
+    def get(self, workspace_id: str) -> dict | None
+```
+
+模块级单例 `default_pool: WorkspacePool` 用于生产路径。
+
+## 6. 故障切换流程
+
+```
+master_health probe FAIL
+     │
+     ▼
+default_pool.mark_unhealthy(active_ws_id, reason)
+     │ fail_count += 1
+     ▼
+fail_count >= MASTER_HEALTH_FAIL_THRESHOLD ?
+     ├── No  → 仅累计，保持 active
+     └── Yes → 1) 当前 active.tier := cold, status := unhealthy
+              2) 选 warm（最早注册）→ tier := active, status := unknown
+              3) 若无 warm → 选 cold healthy → tier := active
+              4) 若无可升者 → active = None，记 transition_log {reason: no_failover_candidate}
+              5) 写盘 + 转移日志
+```
+
+切换写 `state_log.jsonl`（与 default_machine.subscribe 复用渠道，前端 SSE 流即可感知）。
+
+## 7. 验收（DoD）
+
+- ruff check 全绿
+- pytest 不退化 + 新增 ≥ 15 case 全绿
+- commit `feat(round-12 S7): multi team workspace pool + auto failover`
+- 改动文件：
+  - 新增 `src/autoteam/workspace_pool.py`
+  - 新增 `tests/unit/test_round12_s7_workspace_pool.py`
+  - 改 `src/autoteam/admin_state.py`（路由）
+  - 改 `src/autoteam/master_health.py`（失败路径调 pool）
+  - 改 `src/autoteam/cpa_sync.py`（active workspace 切换感知）
+
+## 8. 风险
+
+- **R1**: 现有调用方 `get_admin_email/get_chatgpt_account_id` 极多，若 workspace_pool 退路不严密会破坏所有路径 → 强制兜底（pool 空时回退 state.json）
+- **R2**: master_health 是热路径，增加 pool I/O 不能阻塞 → 写盘异步友好（已在 lock 内但写盘耗时 ms 级，可接受）
+- **R3**: 用户当前无 team 母号子号，无法 e2e；MVP 只做单测 + mock 故障注入

.trellis/tasks/05-11-s7-multi-workspace-pool/task.json

@@ -0,0 +1,26 @@
+{
+  "id": "s7-multi-workspace-pool",
+  "name": "s7-multi-workspace-pool",
+  "title": "S7: 多 team workspace 池化",
+  "description": "",
+  "status": "in_progress",
+  "dev_type": null,
+  "scope": null,
+  "package": null,
+  "priority": "P2",
+  "creator": "ZRainbow",
+  "assignee": "ZRainbow",
+  "createdAt": "2026-05-11",
+  "completedAt": null,
+  "branch": null,
+  "base_branch": "main",
+  "worktree_path": null,
+  "commit": null,
+  "pr_url": null,
+  "subtasks": [],
+  "children": [],
+  "parent": "05-11-upstream-align-register-multimail-frontend-refresh",
+  "relatedFiles": [],
+  "notes": "",
+  "meta": {}
+}

src/autoteam/admin_state.py

@@ -111,6 +111,19 @@ def clear_admin_state():
 
 
 def get_admin_email():
+    """Round 12 S7 — route through WorkspacePool active workspace.
+
+    Backwards-compat: if the pool is empty / module unavailable, fall back
+    to the legacy state.json read so single-workspace installs are unaffected.
+    """
+    try:
+        from autoteam.workspace_pool import default_pool
+
+        active = default_pool.get_active()
+        if active and active.get("admin_email"):
+            return active["admin_email"]
+    except Exception:
+        pass
     return load_admin_state().get("email", "")
 
 
@@ -126,6 +139,21 @@ def _is_valid_uuid(value: str) -> bool:
 
 
 def get_chatgpt_account_id():
+    """Round 12 S7 — route through WorkspacePool active workspace.
+
+    Backwards-compat: pool empty → legacy state.json read → CHATGPT_ACCOUNT_ID env.
+    Pool ID still must pass UUID validation (filters legacy `user-xxx` ids).
+    """
+    try:
+        from autoteam.workspace_pool import default_pool
+
+        active = default_pool.get_active()
+        if active:
+            pool_aid = (active.get("account_id") or "").strip()
+            if pool_aid and _is_valid_uuid(pool_aid):
+                return pool_aid
+    except Exception:
+        pass
     state = load_admin_state()
     state_id = state.get("account_id", "")
     # state.json 里的值必须是 UUID 格式才有效（user-xxx 是 user ID 不是 account ID）

src/autoteam/cpa_sync.py

@@ -653,5 +653,45 @@ def sync_main_codex_to_cpa(filepath):
     if not upload_to_cpa(filepath):
         raise RuntimeError(f"上传主号认证文件失败: {name}")
 
-    logger.info("[CPA] 主号 Codex 已同步: %s", name)
+    # Round 12 S7 — 主号同步成功后,记录当前 active workspace,便于上游观测
+    try:
+        active = _get_active_workspace_summary()
+        if active:
+            logger.info(
+                "[CPA] 主号 Codex 已同步 (active workspace: id=%s admin=%s account_id=%s)",
+                active.get("id"), active.get("admin_email"), active.get("account_id"),
+            )
+        else:
+            logger.info("[CPA] 主号 Codex 已同步: %s", name)
+    except Exception:
+        logger.info("[CPA] 主号 Codex 已同步: %s", name)
     return {"uploaded": name}
+
+
+def _get_active_workspace_summary():
+    """Round 12 S7 — return the current active workspace (id/admin/account_id) or None.
+
+    Best-effort: never raises; pool unavailable returns None.
+    """
+    try:
+        from autoteam.workspace_pool import default_pool
+
+        active = default_pool.get_active()
+        if not active:
+            return None
+        return {
+            "id": active.get("id"),
+            "admin_email": active.get("admin_email"),
+            "account_id": active.get("account_id"),
+        }
+    except Exception:
+        return None
+
+
+def get_active_sync_target():
+    """Round 12 S7 — public helper: which workspace identity should CPA sync to?
+
+    Returns dict({id, admin_email, account_id}) or None when single-workspace
+    mode (caller falls back to legacy admin_state).
+    """
+    return _get_active_workspace_summary()

src/autoteam/master_health.py

@@ -511,6 +511,62 @@ def is_master_subscription_healthy(
     return healthy, reason, evidence
 
 
+# ---------------------------------------------------------------------------
+# Round 12 S7 — WorkspacePool failover hook
+# ---------------------------------------------------------------------------
+
+# reasons that should NOT trip workspace failover (transient / classification noise)
+_TRANSIENT_FAIL_REASONS = ("network_error",)
+# reasons that SHOULD trip failover (real master-side failure)
+_HARD_FAIL_REASONS = ("auth_invalid", "subscription_cancelled", "workspace_missing", "role_not_owner")
+
+
+def apply_pool_health_signal(
+    healthy: bool,
+    reason: str,
+    evidence: dict | None = None,
+    *,
+    pool=None,
+) -> dict | None:
+    """Round 12 S7 — feed a master_health probe result into the WorkspacePool.
+
+    Call this from any caller of :func:`is_master_subscription_healthy` to
+    drive auto failover. Behavior:
+
+    - healthy=True (active or subscription_grace) → pool.mark_healthy(active_id)
+    - healthy=False AND reason in _HARD_FAIL_REASONS → pool.mark_unhealthy(...)
+      which accumulates fail_count + (when ≥ threshold) auto-promotes a
+      warm/cold candidate.
+    - healthy=False AND reason in _TRANSIENT_FAIL_REASONS → no-op (don't
+      flap on intermittent network glitches).
+
+    Returns the **post-call** active workspace snapshot (None if pool empty
+    or failed). Never raises — M-I1 alignment.
+
+    ``pool`` defaults to ``workspace_pool.default_pool``; tests can inject
+    their own.
+    """
+    try:
+        if pool is None:
+            from autoteam.workspace_pool import default_pool as pool_obj
+            pool = pool_obj
+        active = pool.get_active()
+        if active is None:
+            return None
+        ws_id = active.get("id")
+        if not ws_id:
+            return None
+        if healthy and reason in ("active", "subscription_grace"):
+            return pool.mark_healthy(ws_id)
+        if (not healthy) and reason in _HARD_FAIL_REASONS:
+            return pool.mark_unhealthy(ws_id, reason)
+        # transient or unknown → leave state alone but bubble snapshot
+        return active
+    except Exception as exc:
+        logger.warning("[master_health] apply_pool_health_signal failed: %s", exc)
+        return None
+
+
 # ---------------------------------------------------------------------------
 # Round 9 SPEC v1.1 §11~§12 — Retroactive helper + grace 期 JWT 解析
 # ---------------------------------------------------------------------------

src/autoteam/workspace_pool.py

@@ -0,0 +1,683 @@
+"""Multi team workspace pool — Round 12 S7。
+
+维持 K 个 team workspace 互为冷备:任意时刻仅 1 个 active(主热路径用),
+其余作 warm/cold 备份;主 workspace 连续探测失败超阈值时秒切。
+
+不变量(I1~I7,详见 .trellis/tasks/05-11-s7-multi-workspace-pool/prd.md):
+
+- I1: workspaces[*].tier == "active" 至多 1 个
+- I2: mark_unhealthy 命中 active 且 fail_count >= threshold → 自动 promote warm/cold
+- I3: 写盘原子(snapshot .bak → tmp → os.replace),失败回滚
+- I4: 单 workspace 模式向后兼容 — workspaces.json 不存在时 seed 自 state.json
+- I5: register / set_active / mark_unhealthy / promote 均追加 transition_log
+- I6: 与 default_machine 风格一致(lock 内修改、lock 外发布事件)
+- I7: 永不抛未捕获异常(与 master_health.py M-I1 看齐)
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+import shutil
+import threading
+import time
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+DEFAULT_POOL_FILE = PROJECT_ROOT / "workspaces.json"
+POOL_FILE_MODE = 0o666
+
+SCHEMA_VERSION = 1
+
+TIER_ACTIVE = "active"
+TIER_WARM = "warm"
+TIER_COLD = "cold"
+_VALID_TIERS = (TIER_ACTIVE, TIER_WARM, TIER_COLD)
+
+STATUS_HEALTHY = "healthy"
+STATUS_UNHEALTHY = "unhealthy"
+STATUS_UNKNOWN = "unknown"
+_VALID_STATUSES = (STATUS_HEALTHY, STATUS_UNHEALTHY, STATUS_UNKNOWN)
+
+
+def _default_fail_threshold() -> int:
+    raw = os.environ.get("MASTER_HEALTH_FAIL_THRESHOLD", "3")
+    try:
+        v = int(raw)
+        return v if v > 0 else 3
+    except Exception:
+        return 3
+
+
+PoolSubscriber = Callable[[dict], None]
+
+
+class WorkspacePool:
+    """Persistent multi-workspace pool with auto failover.
+
+    The on-disk schema is documented in prd.md §3. The instance owns a
+    threading lock — all reads / writes serialize on it, mirroring
+    :class:`autoteam.account_state.StateMachine`.
+    """
+
+    def __init__(
+        self,
+        path: Path | None = None,
+        *,
+        fail_threshold: int | None = None,
+    ) -> None:
+        self._path: Path = Path(path) if path else DEFAULT_POOL_FILE
+        self._lock = threading.Lock()
+        self._subscribers: list[PoolSubscriber] = []
+        self._fail_threshold = fail_threshold if fail_threshold is not None else _default_fail_threshold()
+
+    # ---------------------------------------------------------------- config
+    @property
+    def path(self) -> Path:
+        return self._path
+
+    @property
+    def fail_threshold(self) -> int:
+        return self._fail_threshold
+
+    def set_fail_threshold(self, threshold: int) -> None:
+        if not isinstance(threshold, int) or threshold <= 0:
+            raise ValueError("fail_threshold must be a positive int")
+        self._fail_threshold = threshold
+
+    # ----------------------------------------------------------- event bus
+    def subscribe(self, callback: PoolSubscriber) -> PoolSubscriber:
+        if not callable(callback):
+            raise TypeError("subscribe expects a callable")
+        with self._lock:
+            if callback not in self._subscribers:
+                self._subscribers.append(callback)
+        return callback
+
+    def unsubscribe(self, callback: PoolSubscriber) -> bool:
+        with self._lock:
+            try:
+                self._subscribers.remove(callback)
+                return True
+            except ValueError:
+                return False
+
+    def _publish(self, event: dict) -> None:
+        with self._lock:
+            subs = list(self._subscribers)
+        for cb in subs:
+            try:
+                cb(event)
+            except Exception:
+                logger.exception("workspace_pool subscriber %r raised on event %s", cb, event)
+
+    # ---------------------------------------------------------------- I/O
+    def _load_raw(self) -> dict:
+        """Load on-disk pool. Auto-seed from state.json on missing file (I4)."""
+        if not self._path.exists():
+            return self._seed_from_admin_state()
+        try:
+            text = self._path.read_text(encoding="utf-8").strip()
+        except Exception as exc:
+            logger.warning("[workspace_pool] read failed: %s — treating as empty", exc)
+            return _empty_doc()
+        if not text:
+            return _empty_doc()
+        try:
+            data = json.loads(text)
+        except Exception as exc:
+            logger.warning("[workspace_pool] json parse failed: %s — treating as empty", exc)
+            return _empty_doc()
+        if not isinstance(data, dict):
+            return _empty_doc()
+        if data.get("schema_version") != SCHEMA_VERSION:
+            # forward-incompatible / legacy → discard, will re-seed on next register
+            logger.warning(
+                "[workspace_pool] schema_version mismatch (got %r, expect %d) — discard",
+                data.get("schema_version"),
+                SCHEMA_VERSION,
+            )
+            return _empty_doc()
+        ws = data.get("workspaces")
+        if not isinstance(ws, list):
+            ws = []
+        # normalize each row
+        norm_ws: list[dict] = []
+        for row in ws:
+            n = _normalize_row(row)
+            if n is not None:
+                norm_ws.append(n)
+        active = data.get("active") or None
+        return {
+            "schema_version": SCHEMA_VERSION,
+            "active": active if isinstance(active, str) else None,
+            "workspaces": norm_ws,
+        }
+
+    def _seed_from_admin_state(self) -> dict:
+        """I4 — single-workspace backwards compat.
+
+        On first read with no workspaces.json, attempt to seed a single
+        workspace from state.json (admin_email + account_id). Empty state
+        produces an empty doc (no workspaces). Never raises.
+        """
+        try:
+            from autoteam.admin_state import load_admin_state  # late import to break cycle
+            st = load_admin_state() or {}
+        except Exception as exc:
+            logger.warning("[workspace_pool] seed: cannot read admin_state: %s", exc)
+            return _empty_doc()
+        admin_email = (st.get("email") or "").strip()
+        account_id = (st.get("account_id") or "").strip()
+        if not admin_email or not account_id:
+            # nothing to seed — fresh install
+            return _empty_doc()
+        ws_id = f"ws-{account_id}"
+        now = time.time()
+        row = {
+            "id": ws_id,
+            "admin_email": admin_email,
+            "account_id": account_id,
+            "tier": TIER_ACTIVE,
+            "status": STATUS_UNKNOWN,
+            "fail_count": 0,
+            "last_check_ts": None,
+            "registered_at": now,
+            "transition_log": [
+                {"ts": now, "from": None, "to": TIER_ACTIVE, "reason": "seed_from_admin_state"},
+            ],
+        }
+        return {
+            "schema_version": SCHEMA_VERSION,
+            "active": ws_id,
+            "workspaces": [row],
+        }
+
+    def _save_raw(self, doc: dict) -> None:
+        """I3 — atomic write with .bak rollback."""
+        path = self._path
+        path.parent.mkdir(parents=True, exist_ok=True)
+        bak = path.with_suffix(path.suffix + ".bak")
+        tmp = path.with_suffix(path.suffix + ".tmp")
+        payload = json.dumps(doc, indent=2, ensure_ascii=False).encode("utf-8")
+        had_snapshot = False
+        if path.exists():
+            try:
+                shutil.copy2(path, bak)
+                had_snapshot = True
+            except OSError as exc:
+                logger.warning("[workspace_pool] snapshot failed %s -> %s: %s", path, bak, exc)
+        try:
+            with open(tmp, "wb") as fp:
+                fp.write(payload)
+                fp.flush()
+                try:
+                    os.fsync(fp.fileno())
+                except OSError:
+                    pass
+            os.replace(tmp, path)
+        except OSError:
+            logger.exception("[workspace_pool] write failed for %s, attempting rollback", path)
+            if had_snapshot and bak.exists():
+                try:
+                    os.replace(bak, path)
+                except OSError:
+                    logger.exception("[workspace_pool] rollback failed for %s", path)
+            if tmp.exists():
+                try:
+                    tmp.unlink()
+                except OSError:
+                    pass
+            raise
+        else:
+            try:
+                os.chmod(path, POOL_FILE_MODE)
+            except OSError:
+                pass
+            if had_snapshot and bak.exists():
+                try:
+                    bak.unlink()
+                except OSError:
+                    logger.warning("[workspace_pool] cannot delete .bak %s", bak)
+
+    # ---------------------------------------------------------- public API
+
+    def list_all(self) -> list[dict]:
+        """Return all workspace rows (deep copy, safe to mutate)."""
+        with self._lock:
+            doc = self._load_raw()
+        return [dict(row) | {"transition_log": list(row.get("transition_log") or [])}
+                for row in doc["workspaces"]]
+
+    def get(self, workspace_id: str) -> dict | None:
+        """Return one workspace by id (deep copy) or None."""
+        if not workspace_id:
+            return None
+        with self._lock:
+            doc = self._load_raw()
+        for row in doc["workspaces"]:
+            if row.get("id") == workspace_id:
+                return dict(row) | {"transition_log": list(row.get("transition_log") or [])}
+        return None
+
+    def get_active(self) -> dict | None:
+        """Return the current active workspace (or None if pool empty)."""
+        with self._lock:
+            doc = self._load_raw()
+        active_id = doc.get("active")
+        if not active_id:
+            return None
+        for row in doc["workspaces"]:
+            if row.get("id") == active_id and row.get("tier") == TIER_ACTIVE:
+                return dict(row) | {"transition_log": list(row.get("transition_log") or [])}
+        return None
+
+    def register(
+        self,
+        workspace_id: str,
+        admin_email: str,
+        account_id: str,
+        tier: str = TIER_WARM,
+    ) -> dict:
+        """Register a new workspace. If id exists, raise ValueError.
+
+        - tier=active is allowed but will demote any prior active to cold
+          (I1 — at most one active).
+        - First registration with no active in pool auto-promotes (regardless
+          of supplied tier) so single-workspace fresh installs work.
+        """
+        if not workspace_id or not isinstance(workspace_id, str):
+            raise ValueError("workspace_id required")
+        if not admin_email or not isinstance(admin_email, str):
+            raise ValueError("admin_email required")
+        if not account_id or not isinstance(account_id, str):
+            raise ValueError("account_id required")
+        if tier not in _VALID_TIERS:
+            raise ValueError(f"tier must be one of {_VALID_TIERS!r}")
+
+        events: list[dict] = []
+        with self._lock:
+            doc = self._load_raw()
+            for row in doc["workspaces"]:
+                if row.get("id") == workspace_id:
+                    raise ValueError(f"workspace_id {workspace_id!r} already registered")
+            now = time.time()
+            target_tier = tier
+            # auto-promote if pool is empty / has no active
+            if doc.get("active") is None and not any(r.get("tier") == TIER_ACTIVE for r in doc["workspaces"]):
+                target_tier = TIER_ACTIVE
+
+            new_row = {
+                "id": workspace_id,
+                "admin_email": admin_email,
+                "account_id": account_id,
+                "tier": target_tier,
+                "status": STATUS_UNKNOWN,
+                "fail_count": 0,
+                "last_check_ts": None,
+                "registered_at": now,
+                "transition_log": [
+                    {"ts": now, "from": None, "to": target_tier, "reason": "register"},
+                ],
+            }
+
+            if target_tier == TIER_ACTIVE:
+                # I1 — demote any other active row
+                prior_active_id = doc.get("active")
+                for row in doc["workspaces"]:
+                    if row.get("tier") == TIER_ACTIVE:
+                        _append_transition(row, TIER_ACTIVE, TIER_COLD, "demote_on_register_active", now)
+                        row["tier"] = TIER_COLD
+                doc["active"] = workspace_id
+                if prior_active_id and prior_active_id != workspace_id:
+                    events.append({
+                        "type": "demoted",
+                        "workspace_id": prior_active_id,
+                        "reason": "register_active",
+                        "ts": now,
+                    })
+
+            doc["workspaces"].append(new_row)
+            self._save_raw(doc)
+            events.append({
+                "type": "registered",
+                "workspace_id": workspace_id,
+                "tier": target_tier,
+                "ts": now,
+            })
+            snapshot = dict(new_row) | {"transition_log": list(new_row["transition_log"])}
+
+        for ev in events:
+            self._publish(ev)
+        return snapshot
+
+    def set_active(self, workspace_id: str) -> dict:
+        """Promote ``workspace_id`` to active; demote any prior active to cold."""
+        if not workspace_id:
+            raise ValueError("workspace_id required")
+        events: list[dict] = []
+        with self._lock:
+            doc = self._load_raw()
+            target = None
+            for row in doc["workspaces"]:
+                if row.get("id") == workspace_id:
+                    target = row
+                    break
+            if target is None:
+                raise KeyError(f"workspace_id {workspace_id!r} not found")
+            now = time.time()
+            prior_active_id = doc.get("active")
+            if target.get("tier") == TIER_ACTIVE and prior_active_id == workspace_id:
+                # already active — short-circuit but still touch last_check_ts? no, idempotent no-op
+                snapshot = dict(target) | {"transition_log": list(target.get("transition_log") or [])}
+                return snapshot
+            for row in doc["workspaces"]:
+                if row is target:
+                    continue
+                if row.get("tier") == TIER_ACTIVE:
+                    _append_transition(row, TIER_ACTIVE, TIER_COLD, "demote_on_set_active", now)
+                    row["tier"] = TIER_COLD
+                    events.append({
+                        "type": "demoted",
+                        "workspace_id": row.get("id"),
+                        "reason": "set_active",
+                        "ts": now,
+                    })
+            _append_transition(target, target.get("tier"), TIER_ACTIVE, "set_active", now)
+            target["tier"] = TIER_ACTIVE
+            target["status"] = STATUS_UNKNOWN  # reset; next probe sets healthy/unhealthy
+            target["fail_count"] = 0
+            doc["active"] = workspace_id
+            self._save_raw(doc)
+            events.append({
+                "type": "promoted",
+                "workspace_id": workspace_id,
+                "reason": "set_active",
+                "ts": now,
+            })
+            snapshot = dict(target) | {"transition_log": list(target["transition_log"])}
+
+        for ev in events:
+            self._publish(ev)
+        return snapshot
+
+    def mark_unhealthy(
+        self,
+        workspace_id: str,
+        reason: str,
+        *,
+        force_failover: bool = False,
+    ) -> dict | None:
+        """Mark a workspace unhealthy. If active and fail_count >= threshold,
+        auto-promote a warm/cold healthy candidate (I2).
+
+        Returns the (possibly newly-promoted) active workspace snapshot, or
+        None if the pool ended up with no active candidate.
+        """
+        if not workspace_id:
+            raise ValueError("workspace_id required")
+        events: list[dict] = []
+        with self._lock:
+            doc = self._load_raw()
+            target = None
+            for row in doc["workspaces"]:
+                if row.get("id") == workspace_id:
+                    target = row
+                    break
+            if target is None:
+                raise KeyError(f"workspace_id {workspace_id!r} not found")
+
+            now = time.time()
+            prev_status = target.get("status")
+            target["fail_count"] = int(target.get("fail_count") or 0) + 1
+            target["status"] = STATUS_UNHEALTHY
+            target["last_check_ts"] = now
+            _append_transition(
+                target,
+                prev_status,
+                STATUS_UNHEALTHY,
+                f"mark_unhealthy:{reason}",
+                now,
+                kind="status",
+                extra={"fail_count": target["fail_count"]},
+            )
+            events.append({
+                "type": "marked_unhealthy",
+                "workspace_id": workspace_id,
+                "reason": reason,
+                "fail_count": target["fail_count"],
+                "ts": now,
+            })
+
+            should_failover = (
+                target.get("tier") == TIER_ACTIVE
+                and (force_failover or target["fail_count"] >= self._fail_threshold)
+            )
+            promoted_id: str | None = None
+            if should_failover:
+                # demote current active → cold
+                _append_transition(target, TIER_ACTIVE, TIER_COLD, "failover_demote", now)
+                target["tier"] = TIER_COLD
+                if doc.get("active") == workspace_id:
+                    doc["active"] = None
+                # pick promotion candidate
+                candidate = _pick_failover_candidate(doc["workspaces"])
+                if candidate is not None:
+                    _append_transition(
+                        candidate,
+                        candidate.get("tier"),
+                        TIER_ACTIVE,
+                        "failover_promote",
+                        now,
+                    )
+                    candidate["tier"] = TIER_ACTIVE
+                    candidate["status"] = STATUS_UNKNOWN
+                    candidate["fail_count"] = 0
+                    doc["active"] = candidate.get("id")
+                    promoted_id = candidate.get("id")
+                    events.append({
+                        "type": "promoted",
+                        "workspace_id": promoted_id,
+                        "reason": "failover",
+                        "ts": now,
+                    })
+                else:
+                    events.append({
+                        "type": "no_failover_candidate",
+                        "workspace_id": workspace_id,
+                        "ts": now,
+                    })
+
+            self._save_raw(doc)
+            new_active = self._active_row(doc)
+            snapshot = (
+                dict(new_active) | {"transition_log": list(new_active.get("transition_log") or [])}
+                if new_active else None
+            )
+
+        for ev in events:
+            self._publish(ev)
+        return snapshot
+
+    def mark_healthy(self, workspace_id: str) -> dict | None:
+        """Reset fail_count to 0 + status=healthy on a successful probe.
+
+        Idempotent. Useful for the master_health success path so that
+        intermittent failures don't accumulate to threshold.
+        """
+        if not workspace_id:
+            raise ValueError("workspace_id required")
+        events: list[dict] = []
+        with self._lock:
+            doc = self._load_raw()
+            target = None
+            for row in doc["workspaces"]:
+                if row.get("id") == workspace_id:
+                    target = row
+                    break
+            if target is None:
+                raise KeyError(f"workspace_id {workspace_id!r} not found")
+            now = time.time()
+            prev_status = target.get("status")
+            prev_fail = int(target.get("fail_count") or 0)
+            target["fail_count"] = 0
+            target["status"] = STATUS_HEALTHY
+            target["last_check_ts"] = now
+            if prev_status != STATUS_HEALTHY or prev_fail != 0:
+                _append_transition(
+                    target,
+                    prev_status,
+                    STATUS_HEALTHY,
+                    "mark_healthy",
+                    now,
+                    kind="status",
+                    extra={"prev_fail_count": prev_fail},
+                )
+                events.append({
+                    "type": "marked_healthy",
+                    "workspace_id": workspace_id,
+                    "ts": now,
+                })
+            self._save_raw(doc)
+            snapshot = dict(target) | {"transition_log": list(target.get("transition_log") or [])}
+        for ev in events:
+            self._publish(ev)
+        return snapshot
+
+    # --------------------------------------------------------------- helpers
+    def _active_row(self, doc: dict) -> dict | None:
+        active_id = doc.get("active")
+        if not active_id:
+            return None
+        for row in doc["workspaces"]:
+            if row.get("id") == active_id:
+                return row
+        return None
+
+    def reset(self) -> None:
+        """Test helper — wipe the pool file. Production code should not call."""
+        with self._lock:
+            try:
+                if self._path.exists():
+                    self._path.unlink()
+            except OSError as exc:
+                logger.warning("[workspace_pool] reset failed: %s", exc)
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+def _empty_doc() -> dict:
+    return {"schema_version": SCHEMA_VERSION, "active": None, "workspaces": []}
+
+
+def _normalize_row(row: Any) -> dict | None:
+    if not isinstance(row, dict):
+        return None
+    rid = row.get("id")
+    if not isinstance(rid, str) or not rid:
+        return None
+    tier = row.get("tier")
+    if tier not in _VALID_TIERS:
+        tier = TIER_WARM
+    status = row.get("status")
+    if status not in _VALID_STATUSES:
+        status = STATUS_UNKNOWN
+    log = row.get("transition_log")
+    if not isinstance(log, list):
+        log = []
+    fail_count = row.get("fail_count")
+    try:
+        fail_count = int(fail_count or 0)
+    except (TypeError, ValueError):
+        fail_count = 0
+    return {
+        "id": rid,
+        "admin_email": str(row.get("admin_email") or ""),
+        "account_id": str(row.get("account_id") or ""),
+        "tier": tier,
+        "status": status,
+        "fail_count": fail_count,
+        "last_check_ts": row.get("last_check_ts"),
+        "registered_at": row.get("registered_at"),
+        "transition_log": list(log),
+    }
+
+
+def _append_transition(
+    row: dict,
+    from_value: Any,
+    to_value: Any,
+    reason: str,
+    ts: float,
+    *,
+    kind: str = "tier",
+    extra: dict | None = None,
+) -> None:
+    entry: dict = {
+        "ts": ts,
+        "kind": kind,
+        "from": from_value,
+        "to": to_value,
+        "reason": reason,
+    }
+    if extra:
+        entry["extra"] = extra
+    log = row.get("transition_log")
+    if not isinstance(log, list):
+        log = []
+        row["transition_log"] = log
+    log.append(entry)
+
+
+def _pick_failover_candidate(workspaces: list[dict]) -> dict | None:
+    """Pick best workspace to promote to active.
+
+    Order:
+      1. tier=warm AND status != unhealthy, sorted by registered_at asc (oldest first)
+      2. tier=cold AND status == healthy, sorted by registered_at asc
+      3. tier=warm regardless of status (last resort), oldest first
+      4. None
+    """
+    warm_healthy = [w for w in workspaces if w.get("tier") == TIER_WARM and w.get("status") != STATUS_UNHEALTHY]
+    if warm_healthy:
+        warm_healthy.sort(key=lambda w: w.get("registered_at") or 0)
+        return warm_healthy[0]
+    cold_healthy = [w for w in workspaces if w.get("tier") == TIER_COLD and w.get("status") == STATUS_HEALTHY]
+    if cold_healthy:
+        cold_healthy.sort(key=lambda w: w.get("registered_at") or 0)
+        return cold_healthy[0]
+    warm_any = [w for w in workspaces if w.get("tier") == TIER_WARM]
+    if warm_any:
+        warm_any.sort(key=lambda w: w.get("registered_at") or 0)
+        return warm_any[0]
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Module-level default pool
+# ---------------------------------------------------------------------------
+default_pool = WorkspacePool()
+
+
+def get_default_pool() -> WorkspacePool:
+    return default_pool
+
+
+__all__ = [
+    "DEFAULT_POOL_FILE",
+    "SCHEMA_VERSION",
+    "STATUS_HEALTHY",
+    "STATUS_UNHEALTHY",
+    "STATUS_UNKNOWN",
+    "TIER_ACTIVE",
+    "TIER_COLD",
+    "TIER_WARM",
+    "WorkspacePool",
+    "default_pool",
+    "get_default_pool",
+]

tests/unit/test_round12_s7_workspace_pool.py

@@ -0,0 +1,483 @@
+"""Round 12 S7 — multi team workspace pool unit tests.
+
+Goals:
+- 注册 / 切换 / 故障阈值 / 自动 promote
+- 原子写 + .bak 回滚
+- 单 workspace 模式 backwards-compat (admin_state.json seed)
+- master_health 失败信号 → pool failover 集成
+- admin_state 路由 (active workspace > state.json fallback)
+
+Mock-only — 不触达真实 OpenAI workspace API。
+"""
+from __future__ import annotations
+
+import json
+import os
+import threading
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from autoteam.workspace_pool import (
+    SCHEMA_VERSION,
+    STATUS_HEALTHY,
+    STATUS_UNHEALTHY,
+    STATUS_UNKNOWN,
+    TIER_ACTIVE,
+    TIER_COLD,
+    TIER_WARM,
+    WorkspacePool,
+)
+
+UUID_A = "00000000-0000-0000-0000-00000000aaaa"
+UUID_B = "00000000-0000-0000-0000-00000000bbbb"
+UUID_C = "00000000-0000-0000-0000-00000000cccc"
+
+
+@pytest.fixture
+def pool_path(tmp_path: Path) -> Path:
+    return tmp_path / "workspaces.json"
+
+
+@pytest.fixture(autouse=True)
+def _isolate_admin_state(monkeypatch):
+    """Default: seed from admin_state returns empty so tests don't pick up the
+    real on-disk state.json. Tests that exercise the seed path override this
+    explicitly via their own monkeypatch."""
+    monkeypatch.setattr(
+        "autoteam.admin_state.load_admin_state",
+        lambda: {"email": "", "account_id": "", "session_token": ""},
+    )
+
+
+@pytest.fixture
+def pool(pool_path: Path) -> WorkspacePool:
+    return WorkspacePool(path=pool_path, fail_threshold=3)
+
+
+# ---------------------------------------------------------------------------
+# 1. register / get_active basics
+# ---------------------------------------------------------------------------
+def test_register_first_auto_promotes_to_active(pool: WorkspacePool):
+    snap = pool.register("ws-a", "ad1@example.com", UUID_A, tier=TIER_WARM)
+    # supplied tier was warm but pool was empty → auto-promote to active
+    assert snap["tier"] == TIER_ACTIVE
+    assert pool.get_active()["id"] == "ws-a"
+    assert len(pool.list_all()) == 1
+
+
+def test_register_second_keeps_warm(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    snap = pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    assert snap["tier"] == TIER_WARM
+    assert pool.get_active()["id"] == "ws-a"
+
+
+def test_register_active_demotes_prior_active(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_ACTIVE)
+    # only one active at a time
+    actives = [r for r in pool.list_all() if r["tier"] == TIER_ACTIVE]
+    assert len(actives) == 1
+    assert actives[0]["id"] == "ws-b"
+    # previous active is demoted to cold
+    a = pool.get("ws-a")
+    assert a["tier"] == TIER_COLD
+
+
+def test_register_duplicate_id_raises(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    with pytest.raises(ValueError):
+        pool.register("ws-a", "ad1b@example.com", UUID_B)
+
+
+def test_register_invalid_tier_raises(pool: WorkspacePool):
+    with pytest.raises(ValueError):
+        pool.register("ws-a", "ad1@example.com", UUID_A, tier="hot")
+
+
+# ---------------------------------------------------------------------------
+# 2. set_active / list_all
+# ---------------------------------------------------------------------------
+def test_set_active_swaps_and_demotes(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    pool.set_active("ws-b")
+    assert pool.get_active()["id"] == "ws-b"
+    assert pool.get("ws-a")["tier"] == TIER_COLD
+    assert pool.get("ws-b")["tier"] == TIER_ACTIVE
+
+
+def test_set_active_unknown_id_raises(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    with pytest.raises(KeyError):
+        pool.set_active("ws-zzz")
+
+
+def test_set_active_idempotent_when_already_active(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    snap = pool.set_active("ws-a")
+    assert snap["tier"] == TIER_ACTIVE
+
+
+# ---------------------------------------------------------------------------
+# 3. mark_unhealthy + failover
+# ---------------------------------------------------------------------------
+def test_mark_unhealthy_below_threshold_no_failover(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    snap = pool.mark_unhealthy("ws-a", "subscription_cancelled")
+    # only 1 fail (threshold=3) → no failover, ws-a still active
+    assert snap["id"] == "ws-a"
+    a = pool.get("ws-a")
+    assert a["fail_count"] == 1
+    assert a["status"] == STATUS_UNHEALTHY
+
+
+def test_mark_unhealthy_at_threshold_triggers_failover(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    pool.mark_unhealthy("ws-a", "auth_invalid")
+    pool.mark_unhealthy("ws-a", "auth_invalid")
+    snap = pool.mark_unhealthy("ws-a", "auth_invalid")
+    # 3rd fail at threshold=3 → failover to ws-b
+    assert snap is not None
+    assert snap["id"] == "ws-b"
+    assert pool.get_active()["id"] == "ws-b"
+    a = pool.get("ws-a")
+    assert a["tier"] == TIER_COLD
+
+
+def test_mark_unhealthy_no_warm_no_failover_candidate(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    # only ws-a (active), no warm/cold candidate
+    pool.set_fail_threshold(1)
+    snap = pool.mark_unhealthy("ws-a", "auth_invalid")
+    # no candidate → active becomes None
+    assert snap is None
+    assert pool.get_active() is None
+    a = pool.get("ws-a")
+    assert a["tier"] == TIER_COLD
+
+
+def test_mark_unhealthy_force_failover_immediate(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    snap = pool.mark_unhealthy("ws-a", "manual_force", force_failover=True)
+    assert snap["id"] == "ws-b"
+
+
+def test_failover_picks_oldest_warm_first(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    pool.register("ws-c", "ad3@example.com", UUID_C, tier=TIER_WARM)
+    # Set deterministic registered_at: ws-b oldest warm, ws-c younger
+    raw = json.loads(pool.path.read_text(encoding="utf-8"))
+    for row in raw["workspaces"]:
+        if row["id"] == "ws-b":
+            row["registered_at"] = 1000.0
+        elif row["id"] == "ws-c":
+            row["registered_at"] = 2000.0
+    pool.path.write_text(json.dumps(raw, indent=2), encoding="utf-8")
+
+    pool.set_fail_threshold(1)
+    snap = pool.mark_unhealthy("ws-a", "auth_invalid")
+    assert snap["id"] == "ws-b"  # oldest warm wins
+
+
+def test_mark_healthy_resets_fail_count(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.mark_unhealthy("ws-a", "subscription_cancelled")
+    pool.mark_unhealthy("ws-a", "subscription_cancelled")
+    pool.mark_healthy("ws-a")
+    a = pool.get("ws-a")
+    assert a["fail_count"] == 0
+    assert a["status"] == STATUS_HEALTHY
+
+
+# ---------------------------------------------------------------------------
+# 4. transition_log
+# ---------------------------------------------------------------------------
+def test_transition_log_register_promote_demote(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    pool.set_active("ws-b")
+    a = pool.get("ws-a")
+    reasons_a = [e["reason"] for e in a["transition_log"]]
+    assert "register" in reasons_a
+    assert "demote_on_set_active" in reasons_a
+    b = pool.get("ws-b")
+    reasons_b = [e["reason"] for e in b["transition_log"]]
+    assert "register" in reasons_b
+    assert "set_active" in reasons_b
+
+
+def test_transition_log_failover_writes_two_rows(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    pool.set_fail_threshold(1)
+    pool.mark_unhealthy("ws-a", "auth_invalid")
+    a = pool.get("ws-a")
+    b = pool.get("ws-b")
+    assert any(e["reason"] == "failover_demote" for e in a["transition_log"])
+    assert any(e["reason"] == "failover_promote" for e in b["transition_log"])
+
+
+# ---------------------------------------------------------------------------
+# 5. atomic write + .bak rollback (I3)
+# ---------------------------------------------------------------------------
+def test_atomic_write_replaces_file(pool: WorkspacePool):
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    raw = json.loads(pool.path.read_text(encoding="utf-8"))
+    assert raw["schema_version"] == SCHEMA_VERSION
+    assert raw["active"] == "ws-a"
+    # no leftover .tmp / .bak
+    assert not pool.path.with_suffix(pool.path.suffix + ".tmp").exists()
+    assert not pool.path.with_suffix(pool.path.suffix + ".bak").exists()
+
+
+def test_atomic_write_rolls_back_on_os_replace_failure(pool_path: Path):
+    pool = WorkspacePool(path=pool_path)
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    original = pool_path.read_text(encoding="utf-8")
+
+    real_replace = os.replace
+
+    def boom(src, dst):
+        if str(dst) == str(pool_path):
+            raise OSError("simulated mid-write failure")
+        return real_replace(src, dst)
+
+    with patch("autoteam.workspace_pool.os.replace", side_effect=boom):
+        with pytest.raises(OSError):
+            pool.register("ws-b", "ad2@example.com", UUID_B)
+
+    # original content preserved (rollback restored .bak)
+    assert pool_path.read_text(encoding="utf-8") == original
+
+
+# ---------------------------------------------------------------------------
+# 6. backwards-compat seed from admin_state.json (I4)
+# ---------------------------------------------------------------------------
+def test_seed_from_admin_state_when_pool_missing(monkeypatch, pool_path: Path):
+    fake_state = {
+        "email": "legacy_admin@example.com",
+        "account_id": UUID_A,
+        "session_token": "tok",
+    }
+    monkeypatch.setattr("autoteam.admin_state.load_admin_state", lambda: fake_state)
+    pool = WorkspacePool(path=pool_path)
+    active = pool.get_active()
+    assert active is not None
+    assert active["admin_email"] == "legacy_admin@example.com"
+    assert active["account_id"] == UUID_A
+    assert active["tier"] == TIER_ACTIVE
+
+
+def test_seed_skipped_when_state_empty(monkeypatch, pool_path: Path):
+    monkeypatch.setattr("autoteam.admin_state.load_admin_state", lambda: {"email": "", "account_id": ""})
+    pool = WorkspacePool(path=pool_path)
+    assert pool.get_active() is None
+    assert pool.list_all() == []
+
+
+# ---------------------------------------------------------------------------
+# 7. admin_state routing
+# ---------------------------------------------------------------------------
+def test_admin_state_get_admin_email_routes_to_pool(monkeypatch, pool_path: Path):
+    pool = WorkspacePool(path=pool_path)
+    pool.register("ws-x", "pool_admin@example.com", UUID_A)
+    monkeypatch.setattr("autoteam.workspace_pool.default_pool", pool)
+    # state.json should NOT be consulted when pool active
+    monkeypatch.setattr("autoteam.admin_state.load_admin_state",
+                        lambda: {"email": "stale@example.com", "account_id": ""})
+    from autoteam.admin_state import get_admin_email, get_chatgpt_account_id
+    assert get_admin_email() == "pool_admin@example.com"
+    assert get_chatgpt_account_id() == UUID_A
+
+
+def test_admin_state_falls_back_to_state_json_when_pool_empty(monkeypatch, pool_path: Path):
+    pool = WorkspacePool(path=pool_path)
+    # empty pool (no register, and state seed returns empty)
+    monkeypatch.setattr("autoteam.admin_state.load_admin_state",
+                        lambda: {"email": "fallback@example.com", "account_id": UUID_B})
+    monkeypatch.setattr("autoteam.workspace_pool.default_pool", pool)
+    # also: pool.get_active will seed from state, so make seed return empty
+    monkeypatch.setattr(
+        WorkspacePool, "_seed_from_admin_state",
+        lambda self: {"schema_version": SCHEMA_VERSION, "active": None, "workspaces": []},
+    )
+    from autoteam.admin_state import get_admin_email, get_chatgpt_account_id
+    assert get_admin_email() == "fallback@example.com"
+    assert get_chatgpt_account_id() == UUID_B
+
+
+# ---------------------------------------------------------------------------
+# 8. master_health pool integration
+# ---------------------------------------------------------------------------
+def test_apply_pool_health_signal_healthy_marks_healthy(monkeypatch, pool_path: Path):
+    pool = WorkspacePool(path=pool_path)
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.mark_unhealthy("ws-a", "subscription_cancelled")  # fail_count=1
+    from autoteam.master_health import apply_pool_health_signal
+    snap = apply_pool_health_signal(True, "active", pool=pool)
+    assert snap is not None
+    a = pool.get("ws-a")
+    assert a["fail_count"] == 0
+    assert a["status"] == STATUS_HEALTHY
+
+
+def test_apply_pool_health_signal_subscription_cancelled_drives_failover(pool_path: Path):
+    pool = WorkspacePool(path=pool_path, fail_threshold=2)
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    from autoteam.master_health import apply_pool_health_signal
+    apply_pool_health_signal(False, "subscription_cancelled", pool=pool)
+    snap = apply_pool_health_signal(False, "subscription_cancelled", pool=pool)
+    # threshold=2 reached → failover to ws-b
+    assert snap is not None
+    assert snap["id"] == "ws-b"
+
+
+def test_apply_pool_health_signal_network_error_no_failover(pool_path: Path):
+    pool = WorkspacePool(path=pool_path, fail_threshold=1)
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    from autoteam.master_health import apply_pool_health_signal
+    snap = apply_pool_health_signal(False, "network_error", pool=pool)
+    # network_error is transient — don't trip
+    assert snap["id"] == "ws-a"
+    a = pool.get("ws-a")
+    assert a["fail_count"] == 0  # untouched
+
+
+def test_apply_pool_health_signal_empty_pool_returns_none(pool_path: Path):
+    pool = WorkspacePool(path=pool_path)
+    from autoteam.master_health import apply_pool_health_signal
+    assert apply_pool_health_signal(False, "auth_invalid", pool=pool) is None
+
+
+# ---------------------------------------------------------------------------
+# 9. event subscribers
+# ---------------------------------------------------------------------------
+def test_subscriber_receives_register_event(pool: WorkspacePool):
+    events: list[dict] = []
+    pool.subscribe(events.append)
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    types = [e["type"] for e in events]
+    assert "registered" in types
+
+
+def test_subscriber_receives_failover_promote_event(pool: WorkspacePool):
+    events: list[dict] = []
+    pool.subscribe(events.append)
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    pool.register("ws-b", "ad2@example.com", UUID_B, tier=TIER_WARM)
+    pool.set_fail_threshold(1)
+    pool.mark_unhealthy("ws-a", "auth_invalid")
+    types = [e["type"] for e in events]
+    assert "promoted" in types
+    assert "marked_unhealthy" in types
+
+
+def test_subscriber_exceptions_swallowed(pool: WorkspacePool):
+    def bad(_e):
+        raise RuntimeError("boom")
+    pool.subscribe(bad)
+    # must not raise
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+
+
+# ---------------------------------------------------------------------------
+# 10. concurrency safety (lock)
+# ---------------------------------------------------------------------------
+def test_concurrent_registers_no_duplicate_active(pool: WorkspacePool):
+    barrier = threading.Barrier(5)
+    errors: list[Exception] = []
+
+    def worker(i):
+        try:
+            barrier.wait(timeout=2)
+            pool.register(f"ws-{i}", f"ad{i}@example.com", f"00000000-0000-0000-0000-{i:012d}")
+        except Exception as exc:
+            errors.append(exc)
+
+    threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    assert errors == []
+    actives = [r for r in pool.list_all() if r["tier"] == TIER_ACTIVE]
+    assert len(actives) == 1  # I1 — exactly one active
+
+
+# ---------------------------------------------------------------------------
+# 11. schema migration / corrupt-file resilience
+# ---------------------------------------------------------------------------
+def test_corrupt_json_treated_as_empty(pool_path: Path):
+    pool_path.write_text("{not valid json", encoding="utf-8")
+    pool = WorkspacePool(path=pool_path)
+    assert pool.list_all() == []
+    # subsequent register works (file is overwritten cleanly)
+    pool.register("ws-a", "ad1@example.com", UUID_A)
+    assert pool.get_active()["id"] == "ws-a"
+
+
+def test_unknown_schema_version_discarded(pool_path: Path):
+    pool_path.write_text(
+        json.dumps({"schema_version": 999, "active": "ws-x", "workspaces": []}),
+        encoding="utf-8",
+    )
+    pool = WorkspacePool(path=pool_path)
+    assert pool.get_active() is None
+
+
+def test_normalize_drops_invalid_tier(pool_path: Path):
+    pool_path.write_text(
+        json.dumps({
+            "schema_version": SCHEMA_VERSION,
+            "active": "ws-x",
+            "workspaces": [{
+                "id": "ws-x",
+                "admin_email": "a@e.com",
+                "account_id": UUID_A,
+                "tier": "hot",  # invalid
+                "status": "weird",  # invalid
+                "fail_count": "not-an-int",
+                "transition_log": "not-a-list",
+            }],
+        }),
+        encoding="utf-8",
+    )
+    pool = WorkspacePool(path=pool_path)
+    rows = pool.list_all()
+    assert len(rows) == 1
+    assert rows[0]["tier"] == TIER_WARM
+    assert rows[0]["status"] == STATUS_UNKNOWN
+    assert rows[0]["fail_count"] == 0
+    assert rows[0]["transition_log"] == []
+
+
+# ---------------------------------------------------------------------------
+# 12. cpa_sync helper
+# ---------------------------------------------------------------------------
+def test_cpa_sync_get_active_sync_target_returns_pool_summary(monkeypatch, pool_path: Path):
+    pool = WorkspacePool(path=pool_path)
+    pool.register("ws-a", "cpa_admin@example.com", UUID_A)
+    monkeypatch.setattr("autoteam.workspace_pool.default_pool", pool)
+    from autoteam.cpa_sync import get_active_sync_target
+    target = get_active_sync_target()
+    assert target == {"id": "ws-a", "admin_email": "cpa_admin@example.com", "account_id": UUID_A}
+
+
+def test_cpa_sync_get_active_sync_target_none_when_empty(monkeypatch, pool_path: Path):
+    pool = WorkspacePool(path=pool_path)
+    monkeypatch.setattr("autoteam.workspace_pool.default_pool", pool)
+    monkeypatch.setattr(
+        WorkspacePool, "_seed_from_admin_state",
+        lambda self: {"schema_version": SCHEMA_VERSION, "active": None, "workspaces": []},
+    )
+    from autoteam.cpa_sync import get_active_sync_target
+    assert get_active_sync_target() is None