src/autoteam/workspace_pool.py

"""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: 健康检查 hot-path(`mark_unhealthy` / `mark_healthy`)在收到非法
       workspace_id 时**抛 KeyError**;`register` 在重复 / 非法 tier 时
       **抛 ValueError**. 这与 master_health.py M-I1 的语义对齐 ——
       输入验证错误抛(快失败,便于调用方排查),probe / 重试 hot-path
       通过 `apply_pool_health_signal()` 用 broad try/except 包裹 raise,
       不会让异常蔓延到顶层流程. 永远不会因为正常运行时状态(读盘失败 /
       schema 漂移 / 并发写覆盖)抛异常.
"""
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,
            "workspace_name": str(st.get("workspace_name") or ""),
            "session_token": str(st.get("session_token") or ""),
            "enabled": True,
            "parallel": False,
            "last_error": "",
            "last_run_ts": None,
            "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,
        *,
        workspace_name: str = "",
        session_token: str = "",
        enabled: bool = True,
        parallel: bool = False,
    ) -> 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,
                "workspace_name": workspace_name or "",
                "session_token": session_token or "",
                "enabled": bool(enabled),
                "parallel": bool(parallel),
                "last_error": "",
                "last_run_ts": None,
                "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 upsert(
        self,
        workspace_id: str,
        admin_email: str,
        account_id: str,
        tier: str = TIER_WARM,
        *,
        workspace_name: str = "",
        session_token: str = "",
        enabled: bool = True,
        parallel: bool = False,
    ) -> dict:
        """Create or update a workspace owner row without dropping runtime state."""
        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()
            now = time.time()
            target = None
            for row in doc["workspaces"]:
                if row.get("id") == workspace_id:
                    target = row
                    break

            if target is None:
                target_tier = tier
                if doc.get("active") is None and not any(r.get("tier") == TIER_ACTIVE for r in doc["workspaces"]):
                    target_tier = TIER_ACTIVE
                if target_tier == TIER_ACTIVE:
                    for row in doc["workspaces"]:
                        if row.get("tier") == TIER_ACTIVE:
                            _append_transition(row, TIER_ACTIVE, TIER_COLD, "demote_on_upsert_active", now)
                            row["tier"] = TIER_COLD
                target = {
                    "id": workspace_id,
                    "admin_email": admin_email,
                    "account_id": account_id,
                    "workspace_name": workspace_name or "",
                    "session_token": session_token or "",
                    "enabled": bool(enabled),
                    "parallel": bool(parallel),
                    "last_error": "",
                    "last_run_ts": None,
                    "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": "upsert_register"},
                    ],
                }
                if target_tier == TIER_ACTIVE:
                    doc["active"] = workspace_id
                doc["workspaces"].append(target)
                events.append({"type": "registered", "workspace_id": workspace_id, "tier": target_tier, "ts": now})
            else:
                if tier == TIER_ACTIVE and target.get("tier") != TIER_ACTIVE:
                    for row in doc["workspaces"]:
                        if row.get("tier") == TIER_ACTIVE:
                            _append_transition(row, TIER_ACTIVE, TIER_COLD, "demote_on_upsert_active", now)
                            row["tier"] = TIER_COLD
                    _append_transition(target, target.get("tier"), TIER_ACTIVE, "upsert_set_active", now)
                    target["tier"] = TIER_ACTIVE
                    doc["active"] = workspace_id
                target.update({
                    "admin_email": admin_email,
                    "account_id": account_id,
                    "workspace_name": workspace_name or target.get("workspace_name") or "",
                    "session_token": session_token or target.get("session_token") or "",
                    "enabled": bool(enabled),
                    "parallel": bool(parallel),
                })
                _append_transition(target, target.get("tier"), target.get("tier"), "upsert_update", now)
                events.append({"type": "updated", "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

    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

    def record_run_result(self, workspace_id: str, *, last_error: str = "", last_run_ts: float | None = None) -> dict:
        """Persist lightweight per-workspace scheduler telemetry."""
        if not workspace_id:
            raise ValueError("workspace_id required")
        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")
            target["last_error"] = last_error or ""
            target["last_run_ts"] = float(last_run_ts or time.time())
            self._save_raw(doc)
            snapshot = dict(target) | {"transition_log": list(target.get("transition_log") or [])}
        self._publish({
            "type": "run_result",
            "workspace_id": workspace_id,
            "last_error": snapshot.get("last_error", ""),
            "ts": snapshot.get("last_run_ts"),
        })
        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 ""),
        "workspace_name": str(row.get("workspace_name") or ""),
        "session_token": str(row.get("session_token") or ""),
        "enabled": row.get("enabled") is not False,
        "parallel": bool(row.get("parallel")),
        "last_error": str(row.get("last_error") or ""),
        "last_run_ts": row.get("last_run_ts"),
        "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",
]