feat: implement PRD-00009 MT-1 internal email pool with full P0 contract (claim/release/complete/stats, ownership checks, lease expiry, cooldown recovery, 38 pool tests)

outlook_web/app.py

@@ -36,6 +36,7 @@ def create_app(*, autostart_scheduler: Optional[bool] = None):
             groups,
             oauth,
             pages,
+            pool,
             scheduler,
             settings,
             system,
@@ -88,7 +89,9 @@ def create_app(*, autostart_scheduler: Optional[bool] = None):
         # ProxyFix 中间件（仅在配置启用时应用）
         # 注意：启用前必须配置 TRUSTED_PROXIES 环境变量
         if config.get_proxy_fix_enabled():
-            app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)
+            app.wsgi_app = ProxyFix(
+                app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1
+            )
 
         # DB teardown（请求结束释放连接）
         register_db(app)
@@ -98,7 +101,11 @@ def create_app(*, autostart_scheduler: Optional[bool] = None):
         import sys
 
         _log_handler = logging.StreamHandler(sys.stderr)
-        _log_handler.setFormatter(logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s", datefmt="%H:%M:%S"))
+        _log_handler.setFormatter(
+            logging.Formatter(
+                "%(asctime)s %(name)s %(levelname)s %(message)s", datefmt="%H:%M:%S"
+            )
+        )
         _ow_logger = logging.getLogger("outlook_web")
         if not _ow_logger.handlers:
             _ow_logger.addHandler(_log_handler)
@@ -125,6 +132,7 @@ def create_app(*, autostart_scheduler: Optional[bool] = None):
         app.register_blueprint(scheduler.create_blueprint())
         app.register_blueprint(system.create_blueprint())
         app.register_blueprint(audit.create_blueprint())
+        app.register_blueprint(pool.create_blueprint())
 
         # 打印初始化信息
         print("=" * 60)
@@ -141,11 +149,15 @@ def create_app(*, autostart_scheduler: Optional[bool] = None):
         from outlook_web.services import scheduler as scheduler_service
 
         if scheduler_service.should_autostart_scheduler():
-            scheduler_service.init_scheduler(_APP_INSTANCE, graph_service.test_refresh_token_with_rotation)
+            scheduler_service.init_scheduler(
+                _APP_INSTANCE, graph_service.test_refresh_token_with_rotation
+            )
     elif autostart_scheduler:
         from outlook_web.services import graph as graph_service
         from outlook_web.services import scheduler as scheduler_service
 
-        scheduler_service.init_scheduler(_APP_INSTANCE, graph_service.test_refresh_token_with_rotation)
+        scheduler_service.init_scheduler(
+            _APP_INSTANCE, graph_service.test_refresh_token_with_rotation
+        )
 
     return _APP_INSTANCE

outlook_web/controllers/pool.py

@@ -0,0 +1,133 @@
+"""
+邮箱池控制器（PRD-00009 MT-1）
+
+职责：
+- 解析 HTTP 请求参数
+- 调用 services/pool.py 的业务方法
+- 将 service 层业务结果映射为 FD 统一响应格式
+- 将 PoolServiceError 映射为统一错误响应
+
+响应格式（FD-00009 §6）：
+- 成功：{"success": true, "data": {...}}
+- 失败：{"success": false, "error": "error_code", "message": "错误说明"}
+
+HTTP 状态码映射（TD-00009 §6.3）：
+- 参数错误 → 400
+- 归属/凭据错误 → 403
+- 状态冲突 → 409
+- 业务失败（如 no_available_account） → 200 + success=false
+"""
+
+from __future__ import annotations
+
+from flask import jsonify, request
+
+from outlook_web.services.pool import (
+    PoolServiceError,
+    claim_random,
+    complete_claim,
+    get_pool_stats,
+    release_claim,
+)
+
+
+def _success_response(data: dict, http_status: int = 200):
+    """统一成功响应：{"success": true, "data": {...}}"""
+    return jsonify({"success": True, "data": data}), http_status
+
+
+def _error_response(error_code: str, message: str, http_status: int = 400):
+    """统一失败响应：{"success": false, "error": "...", "message": "..."}"""
+    return jsonify(
+        {"success": False, "error": error_code, "message": message}
+    ), http_status
+
+
+def api_claim_random():
+    body = request.get_json(silent=True) or {}
+    caller_id = body.get("caller_id", "")
+    task_id = body.get("task_id", "")
+    provider = body.get("provider")
+
+    try:
+        account = claim_random(
+            caller_id=caller_id,
+            task_id=task_id,
+            provider=provider,
+        )
+        return _success_response(
+            {
+                "account_id": account["id"],
+                "email": account["email"],
+                "claim_token": account["claim_token"],
+                "lease_expires_at": account["lease_expires_at"],
+            }
+        )
+    except PoolServiceError as exc:
+        return _error_response(exc.error_code, str(exc), exc.http_status)
+
+
+def api_claim_release():
+    body = request.get_json(silent=True) or {}
+    account_id = body.get("account_id")
+    claim_token = body.get("claim_token", "")
+    caller_id = body.get("caller_id", "")
+    task_id = body.get("task_id", "")
+    reason = body.get("reason")
+
+    if account_id is None:
+        return _error_response("account_id_missing", "account_id 不能为空", 400)
+    try:
+        account_id = int(account_id)
+    except (TypeError, ValueError):
+        return _error_response("account_id_invalid", "account_id 必须为整数", 400)
+
+    try:
+        release_claim(
+            account_id=account_id,
+            claim_token=claim_token,
+            caller_id=caller_id,
+            task_id=task_id,
+            reason=reason,
+        )
+        return _success_response({"account_id": account_id, "pool_status": "available"})
+    except PoolServiceError as exc:
+        return _error_response(exc.error_code, str(exc), exc.http_status)
+
+
+def api_claim_complete():
+    body = request.get_json(silent=True) or {}
+    account_id = body.get("account_id")
+    claim_token = body.get("claim_token", "")
+    caller_id = body.get("caller_id", "")
+    task_id = body.get("task_id", "")
+    result = body.get("result", "")
+    detail = body.get("detail")
+
+    if account_id is None:
+        return _error_response("account_id_missing", "account_id 不能为空", 400)
+    try:
+        account_id = int(account_id)
+    except (TypeError, ValueError):
+        return _error_response("account_id_invalid", "account_id 必须为整数", 400)
+
+    try:
+        new_status = complete_claim(
+            account_id=account_id,
+            claim_token=claim_token,
+            caller_id=caller_id,
+            task_id=task_id,
+            result=result,
+            detail=detail,
+        )
+        return _success_response({"account_id": account_id, "pool_status": new_status})
+    except PoolServiceError as exc:
+        return _error_response(exc.error_code, str(exc), exc.http_status)
+
+
+def api_pool_stats():
+    try:
+        stats = get_pool_stats()
+        return _success_response(stats)
+    except Exception as exc:
+        return _error_response("internal_error", str(exc), 500)

outlook_web/db.py

@@ -24,7 +24,8 @@ from outlook_web.security.crypto import (
 # v8：PRD-00008 P1 — 上游真实探测结果缓存表
 # v9：PRD-00008 P2 — 多 API Key 表
 # v10：PRD-00008 P2 — 调用方日级使用统计表
-DB_SCHEMA_VERSION = 10
+# v11：PRD-00009 MT-1 — 邮箱池字段（pool_status/claimed_by/...）+ account_claim_logs 表 + pool settings
+DB_SCHEMA_VERSION = 11
 DB_SCHEMA_VERSION_KEY = "db_schema_version"
 DB_SCHEMA_LAST_UPGRADE_TRACE_ID_KEY = "db_schema_last_upgrade_trace_id"
 DB_SCHEMA_LAST_UPGRADE_ERROR_KEY = "db_schema_last_upgrade_error"
@@ -124,7 +125,9 @@ def init_db(database_path: Optional[str] = None):
             """)
 
         # 在锁内读取当前 schema 版本（保证一致性）
-        row = cursor.execute("SELECT value FROM settings WHERE key = ?", (DB_SCHEMA_VERSION_KEY,)).fetchone()
+        row = cursor.execute(
+            "SELECT value FROM settings WHERE key = ?", (DB_SCHEMA_VERSION_KEY,)
+        ).fetchone()
         current_version = int(row["value"]) if row and row["value"] is not None else 0
 
         upgrading = current_version < DB_SCHEMA_VERSION
@@ -133,7 +136,9 @@ def init_db(database_path: Optional[str] = None):
             if db_existed:
                 try:
                     print("=" * 60)
-                    print(f"[升级提示] 检测到数据库需要升级：v{current_version} -> v{DB_SCHEMA_VERSION}")
+                    print(
+                        f"[升级提示] 检测到数据库需要升级：v{current_version} -> v{DB_SCHEMA_VERSION}"
+                    )
                     print(f"[升级提示] 强烈建议先备份数据库文件：{path}")
                     print(f'[升级提示] 示例：cp "{path}" "{path}.backup"')
                     print(f"[升级提示] trace_id={migration_trace_id}")
@@ -347,25 +352,39 @@ def init_db(database_path: Optional[str] = None):
         if "remark" not in columns:
             cursor.execute("ALTER TABLE accounts ADD COLUMN remark TEXT")
         if "status" not in columns:
-            cursor.execute("ALTER TABLE accounts ADD COLUMN status TEXT DEFAULT 'active'")
+            cursor.execute(
+                "ALTER TABLE accounts ADD COLUMN status TEXT DEFAULT 'active'"
+            )
         if "updated_at" not in columns:
-            cursor.execute("ALTER TABLE accounts ADD COLUMN updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP")
+            cursor.execute(
+                "ALTER TABLE accounts ADD COLUMN updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
+            )
         if "last_refresh_at" not in columns:
             cursor.execute("ALTER TABLE accounts ADD COLUMN last_refresh_at TIMESTAMP")
         if "account_type" not in columns:
-            cursor.execute("ALTER TABLE accounts ADD COLUMN account_type TEXT DEFAULT 'outlook'")
+            cursor.execute(
+                "ALTER TABLE accounts ADD COLUMN account_type TEXT DEFAULT 'outlook'"
+            )
         if "provider" not in columns:
-            cursor.execute("ALTER TABLE accounts ADD COLUMN provider TEXT DEFAULT 'outlook'")
+            cursor.execute(
+                "ALTER TABLE accounts ADD COLUMN provider TEXT DEFAULT 'outlook'"
+            )
         if "imap_host" not in columns:
             cursor.execute("ALTER TABLE accounts ADD COLUMN imap_host TEXT")
         if "imap_port" not in columns:
-            cursor.execute("ALTER TABLE accounts ADD COLUMN imap_port INTEGER DEFAULT 993")
+            cursor.execute(
+                "ALTER TABLE accounts ADD COLUMN imap_port INTEGER DEFAULT 993"
+            )
         if "imap_password" not in columns:
             cursor.execute("ALTER TABLE accounts ADD COLUMN imap_password TEXT")
         if "telegram_push_enabled" not in columns:
-            cursor.execute("ALTER TABLE accounts ADD COLUMN telegram_push_enabled INTEGER NOT NULL DEFAULT 0")
+            cursor.execute(
+                "ALTER TABLE accounts ADD COLUMN telegram_push_enabled INTEGER NOT NULL DEFAULT 0"
+            )
         if "telegram_last_checked_at" not in columns:
-            cursor.execute("ALTER TABLE accounts ADD COLUMN telegram_last_checked_at TEXT DEFAULT NULL")
+            cursor.execute(
+                "ALTER TABLE accounts ADD COLUMN telegram_last_checked_at TEXT DEFAULT NULL"
+            )
 
         cursor.execute("PRAGMA table_info(groups)")
         group_columns = [col[1] for col in cursor.fetchall()]
@@ -640,6 +659,64 @@ def init_db(database_path: Optional[str] = None):
             ON external_api_consumer_usage_daily(consumer_key, usage_date)
             """)
 
+        # v11: 邮箱池字段 + account_claim_logs 表（PRD-00009 MT-1）
+        cursor.execute("PRAGMA table_info(accounts)")
+        accounts_columns_v11 = [col[1] for col in cursor.fetchall()]
+        for col_def in [
+            ("pool_status", "TEXT DEFAULT NULL"),
+            ("claimed_by", "TEXT DEFAULT NULL"),
+            ("claimed_at", "TEXT DEFAULT NULL"),
+            ("lease_expires_at", "TEXT DEFAULT NULL"),
+            ("claim_token", "TEXT DEFAULT NULL"),
+            ("last_claimed_at", "TEXT DEFAULT NULL"),
+            ("last_result", "TEXT DEFAULT NULL"),
+            ("last_result_detail", "TEXT DEFAULT NULL"),
+            ("success_count", "INTEGER DEFAULT 0"),
+            ("fail_count", "INTEGER DEFAULT 0"),
+        ]:
+            if col_def[0] not in accounts_columns_v11:
+                cursor.execute(
+                    f"ALTER TABLE accounts ADD COLUMN {col_def[0]} {col_def[1]}"
+                )
+
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS account_claim_logs (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                account_id INTEGER NOT NULL,
+                claim_token TEXT NOT NULL,
+                caller_id TEXT NOT NULL,
+                task_id TEXT NOT NULL,
+                action TEXT NOT NULL,
+                result TEXT DEFAULT NULL,
+                detail TEXT DEFAULT NULL,
+                created_at TEXT NOT NULL,
+                FOREIGN KEY (account_id) REFERENCES accounts(id)
+            )
+            """)
+        cursor.execute("""
+            CREATE INDEX IF NOT EXISTS idx_claim_logs_account_id
+            ON account_claim_logs(account_id)
+            """)
+        cursor.execute("""
+            CREATE INDEX IF NOT EXISTS idx_claim_logs_task_id
+            ON account_claim_logs(task_id)
+            """)
+        cursor.execute("""
+            CREATE INDEX IF NOT EXISTS idx_claim_logs_claim_token
+            ON account_claim_logs(claim_token)
+            """)
+        cursor.execute("""
+            CREATE INDEX IF NOT EXISTS idx_accounts_pool_status
+            ON accounts(pool_status)
+            """)
+
+        cursor.execute(
+            "INSERT OR IGNORE INTO settings (key, value) VALUES ('pool_cooldown_seconds', '86400')"
+        )
+        cursor.execute(
+            "INSERT OR IGNORE INTO settings (key, value) VALUES ('pool_default_lease_seconds', '600')"
+        )
+
         # 迁移现有明文数据为加密数据
         migrate_sensitive_data(conn)
 

outlook_web/repositories/pool.py

@@ -0,0 +1,282 @@
+from __future__ import annotations
+
+import secrets
+from datetime import datetime, timedelta, timezone
+from typing import Dict, List, Optional
+
+import sqlite3
+
+
+RESULT_TO_POOL_STATUS: Dict[str, str] = {
+    "success": "used",
+    "verification_timeout": "cooldown",
+    "provider_blocked": "frozen",
+    "credential_invalid": "retired",
+    "network_error": "available",
+}
+
+
+def _utcnow() -> datetime:
+    return datetime.now(timezone.utc).replace(tzinfo=None)
+
+
+def claim_atomic(
+    conn: sqlite3.Connection,
+    caller_id: str,
+    task_id: str,
+    lease_seconds: int,
+    provider: Optional[str] = None,
+    group_id: Optional[int] = None,
+    tags: Optional[List[str]] = None,
+    exclude_recent_minutes: Optional[int] = None,
+) -> Optional[dict]:
+    sql = """
+        SELECT a.* FROM accounts a
+        WHERE a.pool_status = 'available'
+        AND a.status = 'active'
+    """
+    params: list = []
+
+    if provider:
+        sql += " AND a.provider = ?"
+        params.append(provider)
+
+    if group_id is not None:
+        sql += " AND a.group_id = ?"
+        params.append(group_id)
+
+    if tags:
+        for tag_name in tags:
+            sql += """
+                AND EXISTS (
+                    SELECT 1 FROM account_tags at2
+                    JOIN tags t2 ON at2.tag_id = t2.id
+                    WHERE at2.account_id = a.id AND t2.name = ?
+                )
+            """
+            params.append(tag_name)
+
+    if exclude_recent_minutes and exclude_recent_minutes > 0:
+        cutoff = (
+            _utcnow() - timedelta(minutes=exclude_recent_minutes)
+        ).isoformat() + "Z"
+        sql += " AND (a.last_claimed_at IS NULL OR a.last_claimed_at < ?)"
+        params.append(cutoff)
+
+    sql += " ORDER BY RANDOM() LIMIT 1"
+
+    conn.execute("BEGIN IMMEDIATE")
+    account = conn.execute(sql, params).fetchone()
+
+    if account is None:
+        conn.execute("ROLLBACK")
+        return None
+
+    now_str = _utcnow().isoformat() + "Z"
+    lease_expires_at_str = (
+        _utcnow() + timedelta(seconds=lease_seconds)
+    ).isoformat() + "Z"
+    token = "clm_" + secrets.token_urlsafe(9)
+
+    conn.execute(
+        """
+        UPDATE accounts SET
+            pool_status = 'claimed',
+            claimed_by = ?,
+            claimed_at = ?,
+            lease_expires_at = ?,
+            claim_token = ?,
+            last_claimed_at = ?,
+            updated_at = ?
+        WHERE id = ?
+        """,
+        (
+            f"{caller_id}:{task_id}",
+            now_str,
+            lease_expires_at_str,
+            token,
+            now_str,
+            now_str,
+            account["id"],
+        ),
+    )
+    conn.execute(
+        """
+        INSERT INTO account_claim_logs
+            (account_id, claim_token, caller_id, task_id, action, result, detail, created_at)
+        VALUES (?, ?, ?, ?, 'claim', NULL, NULL, ?)
+        """,
+        (account["id"], token, caller_id, task_id, now_str),
+    )
+    conn.execute("COMMIT")
+    return dict(account) | {
+        "claim_token": token,
+        "lease_expires_at": lease_expires_at_str,
+    }
+
+
+def release(
+    conn: sqlite3.Connection,
+    account_id: int,
+    claim_token: str,
+    caller_id: str,
+    task_id: str,
+    reason: Optional[str],
+) -> None:
+    now_str = _utcnow().isoformat() + "Z"
+    conn.execute("BEGIN IMMEDIATE")
+    conn.execute(
+        """
+        UPDATE accounts SET
+            pool_status = 'available',
+            claimed_by = NULL,
+            claimed_at = NULL,
+            lease_expires_at = NULL,
+            claim_token = NULL,
+            updated_at = ?
+        WHERE id = ?
+        """,
+        (now_str, account_id),
+    )
+    conn.execute(
+        """
+        INSERT INTO account_claim_logs
+            (account_id, claim_token, caller_id, task_id, action, result, detail, created_at)
+        VALUES (?, ?, ?, ?, 'release', 'manual_release', ?, ?)
+        """,
+        (account_id, claim_token, caller_id, task_id, reason, now_str),
+    )
+    conn.execute("COMMIT")
+
+
+def complete(
+    conn: sqlite3.Connection,
+    account_id: int,
+    claim_token: str,
+    caller_id: str,
+    task_id: str,
+    result: str,
+    detail: Optional[str],
+) -> str:
+    new_pool_status = RESULT_TO_POOL_STATUS[result]
+    is_success = result == "success"
+    now_str = _utcnow().isoformat() + "Z"
+
+    conn.execute("BEGIN IMMEDIATE")
+    conn.execute(
+        """
+        UPDATE accounts SET
+            pool_status = ?,
+            claimed_by = NULL,
+            claimed_at = NULL,
+            lease_expires_at = NULL,
+            claim_token = NULL,
+            last_result = ?,
+            last_result_detail = ?,
+            success_count = success_count + ?,
+            fail_count = fail_count + ?,
+            updated_at = ?
+        WHERE id = ?
+        """,
+        (
+            new_pool_status,
+            result,
+            detail,
+            1 if is_success else 0,
+            0 if is_success else 1,
+            now_str,
+            account_id,
+        ),
+    )
+    conn.execute(
+        """
+        INSERT INTO account_claim_logs
+            (account_id, claim_token, caller_id, task_id, action, result, detail, created_at)
+        VALUES (?, ?, ?, ?, 'complete', ?, ?, ?)
+        """,
+        (account_id, claim_token, caller_id, task_id, result, detail, now_str),
+    )
+    conn.execute("COMMIT")
+    return new_pool_status
+
+
+def expire_stale_claims(conn: sqlite3.Connection) -> int:
+    now_str = _utcnow().isoformat() + "Z"
+    expired = conn.execute(
+        """
+        SELECT id, claim_token, claimed_by FROM accounts
+        WHERE pool_status = 'claimed' AND lease_expires_at < ?
+        """,
+        (now_str,),
+    ).fetchall()
+
+    for account in expired:
+        parts = (account["claimed_by"] or ":").split(":", 1)
+        caller_id = parts[0]
+        task_id = parts[1] if len(parts) > 1 else ""
+
+        conn.execute("BEGIN IMMEDIATE")
+        conn.execute(
+            """
+            UPDATE accounts SET
+                pool_status = 'cooldown',
+                claimed_by = NULL,
+                claimed_at = NULL,
+                lease_expires_at = NULL,
+                claim_token = NULL,
+                fail_count = fail_count + 1,
+                last_result = 'lease_expired',
+                updated_at = ?
+            WHERE id = ?
+            """,
+            (now_str, account["id"]),
+        )
+        conn.execute(
+            """
+            INSERT INTO account_claim_logs
+                (account_id, claim_token, caller_id, task_id, action, result, detail, created_at)
+            VALUES (?, ?, ?, ?, 'expire', 'lease_expired', 'lease timeout, auto moved to cooldown', ?)
+            """,
+            (account["id"], account["claim_token"], caller_id, task_id, now_str),
+        )
+        conn.execute("COMMIT")
+
+    return len(expired)
+
+
+def recover_cooldown(conn: sqlite3.Connection, cooldown_seconds: int) -> int:
+    cutoff_str = (_utcnow() - timedelta(seconds=cooldown_seconds)).isoformat() + "Z"
+    now_str = _utcnow().isoformat() + "Z"
+    cursor = conn.execute(
+        """
+        UPDATE accounts SET pool_status = 'available', updated_at = ?
+        WHERE pool_status = 'cooldown' AND updated_at < ?
+        """,
+        (now_str, cutoff_str),
+    )
+    conn.commit()
+    return cursor.rowcount
+
+
+def get_stats(conn: sqlite3.Connection) -> dict:
+    rows = conn.execute(
+        """
+        SELECT pool_status, COUNT(*) as cnt FROM accounts
+        GROUP BY pool_status
+        """
+    ).fetchall()
+    pool_counts: dict = {
+        "available": 0,
+        "claimed": 0,
+        "used": 0,
+        "cooldown": 0,
+        "frozen": 0,
+        "retired": 0,
+        "not_in_pool": 0,
+    }
+    for row in rows:
+        key = row["pool_status"] if row["pool_status"] is not None else "not_in_pool"
+        if key in pool_counts:
+            pool_counts[key] = row["cnt"]
+
+    return {"pool_counts": pool_counts}

outlook_web/routes/pool.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from flask import Blueprint
+
+from outlook_web.controllers import pool as pool_controller
+
+
+def create_blueprint() -> Blueprint:
+    bp = Blueprint("pool", __name__)
+    bp.add_url_rule(
+        "/api/pool/claim-random",
+        view_func=pool_controller.api_claim_random,
+        methods=["POST"],
+    )
+    bp.add_url_rule(
+        "/api/pool/claim-release",
+        view_func=pool_controller.api_claim_release,
+        methods=["POST"],
+    )
+    bp.add_url_rule(
+        "/api/pool/claim-complete",
+        view_func=pool_controller.api_claim_complete,
+        methods=["POST"],
+    )
+    bp.add_url_rule(
+        "/api/pool/stats",
+        view_func=pool_controller.api_pool_stats,
+        methods=["GET"],
+    )
+    return bp

outlook_web/services/pool.py

@@ -0,0 +1,225 @@
+"""
+邮箱池服务层（PRD-00009 MT-1）
+
+职责：
+- 输入校验（caller_id / task_id / lease_seconds / result / detail 长度）
+- 读取 settings（在 Flask app_context 下用 get_db，或直接接受 conn）
+- 调用 repositories/pool.py 的原子操作
+- 将 repository 层的异常转换为业务错误码
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from outlook_web.db import create_sqlite_connection
+from outlook_web.repositories import pool as pool_repo
+
+CALLER_ID_MAX_LEN = 64
+TASK_ID_MAX_LEN = 128
+REASON_MAX_LEN = 256
+DETAIL_MAX_LEN = 512
+
+VALID_RESULTS = set(pool_repo.RESULT_TO_POOL_STATUS.keys())
+
+
+class PoolServiceError(Exception):
+    """业务错误，包含 HTTP 状态码和错误码。"""
+
+    def __init__(self, message: str, error_code: str, http_status: int = 400):
+        super().__init__(message)
+        self.error_code = error_code
+        self.http_status = http_status
+
+
+def _validate_caller_id(caller_id: str) -> None:
+    if not caller_id or not caller_id.strip():
+        raise PoolServiceError("caller_id 不能为空", "caller_id_empty")
+    if len(caller_id) > CALLER_ID_MAX_LEN:
+        raise PoolServiceError(
+            f"caller_id 超过最大长度 {CALLER_ID_MAX_LEN}", "caller_id_too_long"
+        )
+
+
+def _validate_task_id(task_id: str) -> None:
+    if not task_id or not task_id.strip():
+        raise PoolServiceError("task_id 不能为空", "task_id_empty")
+    if len(task_id) > TASK_ID_MAX_LEN:
+        raise PoolServiceError(
+            f"task_id 超过最大长度 {TASK_ID_MAX_LEN}", "task_id_too_long"
+        )
+
+
+def _validate_lease_seconds(lease_seconds: int, max_lease: int = 3600) -> None:
+    if lease_seconds <= 0:
+        raise PoolServiceError("lease_seconds 必须大于 0", "lease_seconds_invalid")
+    if lease_seconds > max_lease:
+        raise PoolServiceError(
+            f"lease_seconds 不能超过 {max_lease} 秒", "lease_seconds_too_large"
+        )
+
+
+def _read_settings_via_conn(conn) -> dict:
+    """在独立连接场景下直接从 settings 表读取池相关配置。"""
+    rows = conn.execute(
+        "SELECT key, value FROM settings WHERE key IN (?, ?)",
+        ("pool_cooldown_seconds", "pool_default_lease_seconds"),
+    ).fetchall()
+    result = {"pool_cooldown_seconds": 86400, "pool_default_lease_seconds": 600}
+    for row in rows:
+        try:
+            result[row["key"]] = int(row["value"])
+        except (TypeError, ValueError):
+            pass
+    return result
+
+
+def claim_random(
+    *,
+    caller_id: str,
+    task_id: str,
+    provider: Optional[str] = None,
+) -> dict:
+    _validate_caller_id(caller_id)
+    _validate_task_id(task_id)
+
+    conn = create_sqlite_connection()
+    try:
+        settings = _read_settings_via_conn(conn)
+        default_lease = settings["pool_default_lease_seconds"]
+        _validate_lease_seconds(default_lease)
+
+        account = pool_repo.claim_atomic(
+            conn,
+            caller_id=caller_id,
+            task_id=task_id,
+            lease_seconds=default_lease,
+            provider=provider,
+        )
+        if account is None:
+            raise PoolServiceError(
+                "池中没有符合条件的可用邮箱", "no_available_account", http_status=200
+            )
+        return account
+    finally:
+        conn.close()
+
+
+def release_claim(
+    *,
+    account_id: int,
+    claim_token: str,
+    caller_id: str,
+    task_id: str,
+    reason: Optional[str] = None,
+) -> None:
+    """释放已领取的邮箱账号（不计入成功/失败统计，直接回 available）。"""
+    _validate_caller_id(caller_id)
+    _validate_task_id(task_id)
+    if not claim_token or not claim_token.strip():
+        raise PoolServiceError("claim_token 不能为空", "claim_token_empty")
+    if reason and len(reason) > REASON_MAX_LEN:
+        raise PoolServiceError(
+            f"reason 超过最大长度 {REASON_MAX_LEN}", "reason_too_long"
+        )
+
+    conn = create_sqlite_connection()
+    try:
+        row = conn.execute(
+            "SELECT id, claim_token, claimed_by, pool_status FROM accounts WHERE id = ?",
+            (account_id,),
+        ).fetchone()
+        if row is None:
+            raise PoolServiceError("账号不存在", "account_not_found", http_status=400)
+        if row["pool_status"] != "claimed":
+            raise PoolServiceError(
+                f"账号当前状态为 '{row['pool_status']}'，无法 release",
+                "not_claimed",
+                http_status=409,
+            )
+        if row["claim_token"] != claim_token:
+            raise PoolServiceError(
+                "claim_token 不匹配", "token_mismatch", http_status=403
+            )
+        expected_claimed_by = f"{caller_id}:{task_id}"
+        if row["claimed_by"] != expected_claimed_by:
+            raise PoolServiceError(
+                "caller_id 或 task_id 与领取记录不一致",
+                "caller_mismatch",
+                http_status=403,
+            )
+
+        pool_repo.release(conn, account_id, claim_token, caller_id, task_id, reason)
+    finally:
+        conn.close()
+
+
+def complete_claim(
+    *,
+    account_id: int,
+    claim_token: str,
+    caller_id: str,
+    task_id: str,
+    result: str,
+    detail: Optional[str] = None,
+) -> str:
+    """
+    标记领取结果并驱动状态机流转。
+
+    返回账号的新 pool_status。
+    """
+    _validate_caller_id(caller_id)
+    _validate_task_id(task_id)
+    if not claim_token or not claim_token.strip():
+        raise PoolServiceError("claim_token 不能为空", "claim_token_empty")
+    if result not in VALID_RESULTS:
+        raise PoolServiceError(
+            f"result 必须是 {sorted(VALID_RESULTS)} 之一",
+            "invalid_result",
+        )
+    if detail and len(detail) > DETAIL_MAX_LEN:
+        raise PoolServiceError(
+            f"detail 超过最大长度 {DETAIL_MAX_LEN}", "detail_too_long"
+        )
+
+    conn = create_sqlite_connection()
+    try:
+        row = conn.execute(
+            "SELECT id, claim_token, claimed_by, pool_status FROM accounts WHERE id = ?",
+            (account_id,),
+        ).fetchone()
+        if row is None:
+            raise PoolServiceError("账号不存在", "account_not_found", http_status=400)
+        if row["pool_status"] != "claimed":
+            raise PoolServiceError(
+                f"账号当前状态为 '{row['pool_status']}'，无法 complete",
+                "not_claimed",
+                http_status=409,
+            )
+        if row["claim_token"] != claim_token:
+            raise PoolServiceError(
+                "claim_token 不匹配", "token_mismatch", http_status=403
+            )
+        expected_claimed_by = f"{caller_id}:{task_id}"
+        if row["claimed_by"] != expected_claimed_by:
+            raise PoolServiceError(
+                "caller_id 或 task_id 与领取记录不一致",
+                "caller_mismatch",
+                http_status=403,
+            )
+
+        new_status = pool_repo.complete(
+            conn, account_id, claim_token, caller_id, task_id, result, detail
+        )
+        return new_status
+    finally:
+        conn.close()
+
+
+def get_pool_stats() -> dict:
+    """返回池状态统计（不修改任何数据）。"""
+    conn = create_sqlite_connection()
+    try:
+        return pool_repo.get_stats(conn)
+    finally:
+        conn.close()

outlook_web/services/scheduler.py

@@ -97,7 +97,10 @@ def _configure_telegram_push_job(scheduler, app) -> None:
 
 def _configure_probe_poll_job(scheduler, app) -> None:
     """P2: 注册异步探测轮询 Job。每 5 秒执行一次，处理 pending 探测并清理过期记录。"""
-    from outlook_web.services.external_api import cleanup_expired_probes, poll_pending_probes
+    from outlook_web.services.external_api import (
+        cleanup_expired_probes,
+        poll_pending_probes,
+    )
 
     try:
         scheduler.remove_job("external_probe_poll")
@@ -125,20 +128,94 @@ def _configure_probe_poll_job(scheduler, app) -> None:
     print("✓ 对外 API 异步探测 Job 已配置（轮询间隔：5 秒）")
 
 
+def _configure_pool_maintenance_jobs(scheduler) -> None:
+    try:
+        scheduler.remove_job("pool_expire_stale_claims")
+    except Exception:
+        pass
+    try:
+        scheduler.remove_job("pool_recover_cooldown")
+    except Exception:
+        pass
+
+    def _expire_task():
+        try:
+            from outlook_web.db import create_sqlite_connection
+            from outlook_web.repositories import pool as pool_repo
+
+            conn = create_sqlite_connection()
+            try:
+                expired = pool_repo.expire_stale_claims(conn)
+                if expired:
+                    print(f"[pool] 过期 claims 回收：{expired} 条")
+            finally:
+                conn.close()
+        except Exception:
+            pass
+
+    def _recover_task():
+        try:
+            from outlook_web.db import create_sqlite_connection
+            from outlook_web.repositories import pool as pool_repo
+
+            conn = create_sqlite_connection()
+            try:
+                row = conn.execute(
+                    "SELECT value FROM settings WHERE key = 'pool_cooldown_seconds'"
+                ).fetchone()
+                cooldown_seconds = int(row["value"]) if row and row["value"] else 86400
+                recovered = pool_repo.recover_cooldown(conn, cooldown_seconds)
+                if recovered:
+                    print(f"[pool] cooldown 恢复：{recovered} 条")
+            finally:
+                conn.close()
+        except Exception:
+            pass
+
+    scheduler.add_job(
+        func=_expire_task,
+        trigger="interval",
+        seconds=60,
+        id="pool_expire_stale_claims",
+        name="邮箱池过期 Claim 回收",
+        replace_existing=True,
+        max_instances=1,
+        coalesce=True,
+        misfire_grace_time=60,
+    )
+    scheduler.add_job(
+        func=_recover_task,
+        trigger="interval",
+        seconds=300,
+        id="pool_recover_cooldown",
+        name="邮箱池 Cooldown 恢复",
+        replace_existing=True,
+        max_instances=1,
+        coalesce=True,
+        misfire_grace_time=120,
+    )
+    print("✓ 邮箱池维护 Job 已配置（expire: 60s，recover: 300s）")
+
+
 def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None:
-    """根据当前 settings 重新配置定时刷新 Job（配置变更即时生效）"""
     try:
         from apscheduler.triggers.cron import CronTrigger
     except Exception:
         return
 
     with app.app_context():
-        enable_scheduled = settings_repo.get_setting("enable_scheduled_refresh", "true").lower() == "true"
-        use_cron = settings_repo.get_setting("use_cron_schedule", "false").lower() == "true"
-        refresh_interval_days = int(settings_repo.get_setting("refresh_interval_days", "30"))
+        enable_scheduled = (
+            settings_repo.get_setting("enable_scheduled_refresh", "true").lower()
+            == "true"
+        )
+        use_cron = (
+            settings_repo.get_setting("use_cron_schedule", "false").lower() == "true"
+        )
+        refresh_interval_days = int(
+            settings_repo.get_setting("refresh_interval_days", "30")
+        )
         cron_expr = settings_repo.get_setting("refresh_cron", "0 2 * * *")
 
-    # 心跳 Job：始终存在（可服务/可刷新可验证）
     try:
         scheduler.remove_job("scheduler_heartbeat")
     except Exception:
@@ -155,11 +232,9 @@ def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None:
         misfire_grace_time=60,
     )
 
-    # Telegram 推送 Job
     _configure_telegram_push_job(scheduler, app)
-
-    # P2: 异步探测轮询 Job（每 5 秒检查一次 pending 探测）
     _configure_probe_poll_job(scheduler, app)
+    _configure_pool_maintenance_jobs(scheduler)
 
     # 刷新 Job：根据 enable_scheduled 决定是否启用
     try:
@@ -184,7 +259,9 @@ def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None:
             if len(parts) != 5:
                 raise ValueError("Cron 表达式格式错误，应为 5 段")
             minute, hour, day, month, day_of_week = parts
-            trigger = CronTrigger(minute=minute, hour=hour, day=day, month=month, day_of_week=day_of_week)
+            trigger = CronTrigger(
+                minute=minute, hour=hour, day=day, month=month, day_of_week=day_of_week
+            )
             scheduler.add_job(
                 func=scheduled_task,
                 trigger=trigger,
@@ -210,7 +287,9 @@ def configure_scheduler_jobs(scheduler, app, test_refresh_token) -> None:
         coalesce=True,
         misfire_grace_time=600,
     )
-    print(f"✓ 定时任务已配置：每天凌晨 2:00 检查刷新（周期：{refresh_interval_days} 天）")
+    print(
+        f"✓ 定时任务已配置：每天凌晨 2:00 检查刷新（周期：{refresh_interval_days} 天）"
+    )
 
 
 def init_scheduler(app, test_refresh_token):
@@ -253,9 +332,17 @@ def scheduled_refresh_task(app, test_refresh_token):
 
     try:
         with app.app_context():
-            enable_scheduled = settings_repo.get_setting("enable_scheduled_refresh", "true").lower() == "true"
-            use_cron = settings_repo.get_setting("use_cron_schedule", "false").lower() == "true"
-            refresh_interval_days = int(settings_repo.get_setting("refresh_interval_days", "30"))
+            enable_scheduled = (
+                settings_repo.get_setting("enable_scheduled_refresh", "true").lower()
+                == "true"
+            )
+            use_cron = (
+                settings_repo.get_setting("use_cron_schedule", "false").lower()
+                == "true"
+            )
+            refresh_interval_days = int(
+                settings_repo.get_setting("refresh_interval_days", "30")
+            )
             delay_seconds = int(settings_repo.get_setting("refresh_delay_seconds", "5"))
 
         run_id = create_refresh_run(conn, "scheduled", trace_id, total=0)
@@ -297,13 +384,19 @@ def scheduled_refresh_task(app, test_refresh_token):
                         return
 
         # 获取刷新间隔配置
-        delay_row = conn.execute("SELECT value FROM settings WHERE key = 'refresh_delay_seconds'").fetchone()
+        delay_row = conn.execute(
+            "SELECT value FROM settings WHERE key = 'refresh_delay_seconds'"
+        ).fetchone()
         delay_seconds = int(delay_row["value"]) if delay_row else 5
 
         # 清理超过半年的刷新记录
         try:
-            conn.execute("DELETE FROM account_refresh_logs WHERE created_at < datetime('now', '-6 months')")
-            conn.execute("DELETE FROM distributed_locks WHERE expires_at < ?", (time.time(),))
+            conn.execute(
+                "DELETE FROM account_refresh_logs WHERE created_at < datetime('now', '-6 months')"
+            )
+            conn.execute(
+                "DELETE FROM distributed_locks WHERE expires_at < ?", (time.time(),)
+            )
             conn.commit()
         except Exception:
             pass
@@ -326,9 +419,13 @@ def scheduled_refresh_task(app, test_refresh_token):
         ttl_seconds = max(60 * 60 * 2, estimated)
         ttl_seconds = min(ttl_seconds, 60 * 60 * 24)
 
-        ok, lock_info = acquire_distributed_lock(conn, REFRESH_LOCK_NAME, lock_owner_id, ttl_seconds)
+        ok, lock_info = acquire_distributed_lock(
+            conn, REFRESH_LOCK_NAME, lock_owner_id, ttl_seconds
+        )
         if not ok:
-            finish_refresh_run(conn, run_id, "skipped", total, 0, 0, "刷新任务冲突：已有刷新在执行")
+            finish_refresh_run(
+                conn, run_id, "skipped", total, 0, 0, "刷新任务冲突：已有刷新在执行"
+            )
             return
         lock_acquired = True
 
@@ -343,7 +440,11 @@ def scheduled_refresh_task(app, test_refresh_token):
 
             # 解密 refresh_token
             try:
-                refresh_token = decrypt_data(encrypted_refresh_token) if encrypted_refresh_token else encrypted_refresh_token
+                refresh_token = (
+                    decrypt_data(encrypted_refresh_token)
+                    if encrypted_refresh_token
+                    else encrypted_refresh_token
+                )
             except Exception as e:
                 failed_count += 1
                 error_msg = f"解密 token 失败: {str(e)}"
@@ -372,13 +473,17 @@ def scheduled_refresh_task(app, test_refresh_token):
             group_id = account["group_id"]
             if group_id:
                 try:
-                    group_row = conn.execute("SELECT proxy_url FROM groups WHERE id = ?", (group_id,)).fetchone()
+                    group_row = conn.execute(
+                        "SELECT proxy_url FROM groups WHERE id = ?", (group_id,)
+                    ).fetchone()
                     if group_row:
                         proxy_url = group_row["proxy_url"] or ""
                 except Exception:
                     proxy_url = ""
 
-            success, error_msg, new_refresh_token = test_refresh_token(client_id, refresh_token, proxy_url)
+            success, error_msg, new_refresh_token = test_refresh_token(
+                client_id, refresh_token, proxy_url
+            )
 
             try:
                 conn.execute(
@@ -398,7 +503,11 @@ def scheduled_refresh_task(app, test_refresh_token):
 
                 if success:
                     # refresh token 可能滚动更新：保存新的 refresh_token（加密存储）
-                    if isinstance(new_refresh_token, str) and new_refresh_token.strip() and new_refresh_token != refresh_token:
+                    if (
+                        isinstance(new_refresh_token, str)
+                        and new_refresh_token.strip()
+                        and new_refresh_token != refresh_token
+                    ):
                         conn.execute(
                             """
                             UPDATE accounts
@@ -461,7 +570,9 @@ def should_autostart_scheduler() -> bool:
         return False
 
     # Flask CLI (flask run) + reloader：父进程不启动
-    if config.env_true("FLASK_RUN_FROM_CLI", False) and not config.env_true("WERKZEUG_RUN_MAIN", False):
+    if config.env_true("FLASK_RUN_FROM_CLI", False) and not config.env_true(
+        "WERKZEUG_RUN_MAIN", False
+    ):
         return False
 
     return True