feat(cookie-gate): 补充 cookie_gate 网关模块与 deploy/watchdog 保活脚本

cookie_gate.py(账号三级门控/阻断记录 pending_reruns/恢复重跑管理器)为此前
接线提交遗漏的模块文件；deploy/watchdog 为控制台与调度器的进程外保活脚本。

deploy/watchdog/gyxx_keepalive.py

@@ -0,0 +1,336 @@
+# -*- coding: utf-8 -*-
+"""控制台与调度器的进程外保活（登录自启动，随用户会话常驻）。
+
+职责（不承载任何业务调度规则，业务调度仍由 `gyxx schedule run` 唯一负责）：
+1. 控制台：探测 http://127.0.0.1:8765/api/health——进程不在则拉起；
+   进程在但探测连续失败（wedged）则结束该进程树后拉起。控制台进程内
+   另有看门狗（console.py `_console_watchdog`）自检失败会主动退出，此处
+   负责把退出后的空窗补上，两层互为补充。
+2. 调度器：读 `var/state/scheduler/service.lock` 的 PID 判断存活，锁主
+   已死则按 install.ps1 同款参数拉起 `python -m gyxx_flow schedule run`；
+   调度器自身单实例锁保证不会双跑。
+3. 节流：每个目标有冷却间隔；短时间反复重启触发熔断，只记录不再拉起，
+   避免故障风暴。
+
+用法：
+  gyxx_keepalive.py            # 常驻循环（Startup 的 vbs 以 pythonw 启动）
+  gyxx_keepalive.py --once     # 单轮探测+处置后退出，供验收/手动巡检
+
+日志：var/logs/keepalive/keepalive.log（追加）。
+自身单实例：var/state/keepalive.lock（PID 存活检测，重复启动直接退出）。
+"""
+
+from __future__ import annotations
+
+import argparse
+import ctypes
+import http.client
+import json
+import os
+import subprocess
+import sys
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+ENV_FILES = [
+    Path(os.getenv("GYXX_RUNTIME_ENV_FILE", r"D:\product-collector-analyze-flow\.env")),
+    Path(
+        os.getenv(
+            "GYXX_RUNTIME_JUSHUITAN_ENV_FILE",
+            r"D:\product-collector-analyze-flow\private-secrets\jushuitan.env",
+        )
+    ),
+]
+CONSOLE_HOST = "127.0.0.1"
+CONSOLE_PORT = int(os.getenv("GYXX_CONSOLE_PORT", "8765"))
+LOOP_INTERVAL_SECONDS = float(os.getenv("GYXX_KEEPALIVE_INTERVAL_SECONDS", "30"))
+PROBE_TIMEOUT_SECONDS = 5.0
+PROBE_FAILURE_LIMIT = 3
+RESTART_COOLDOWN_SECONDS = float(os.getenv("GYXX_KEEPALIVE_COOLDOWN_SECONDS", "120"))
+TRIP_MAX_RESTARTS = 5
+TRIP_WINDOW_SECONDS = 30 * 60
+TRIP_PAUSE_SECONDS = 60 * 60
+
+DATA_ROOT = PROJECT_ROOT / "var"
+LOG_PATH = DATA_ROOT / "logs" / "keepalive" / "keepalive.log"
+LOCK_PATH = DATA_ROOT / "state" / "keepalive.lock"
+CONSOLE_LOG_DIR = DATA_ROOT / "logs" / "console-restarts"
+SCHEDULER_LOG_DIR = DATA_ROOT / "logs" / "scheduler-service"
+SCHEDULER_SERVICE_LOCK = DATA_ROOT / "state" / "scheduler" / "service.lock"
+
+
+def _log(event: str, detail: str = "") -> None:
+    stamp = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S")
+    line = f"{stamp} {event} {detail}".rstrip()
+    LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
+    with LOG_PATH.open("a", encoding="utf-8", newline="\n") as stream:
+        stream.write(line + "\n")
+    print(line, flush=True)
+
+
+def _pid_alive(pid: int) -> bool:
+    if pid <= 0:
+        return False
+    kernel32 = ctypes.windll.kernel32
+    handle = kernel32.OpenProcess(0x1000, False, pid)  # PROCESS_QUERY_LIMITED_INFORMATION
+    if not handle:
+        return False
+    try:
+        exit_code = ctypes.c_ulong()
+        if kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)):
+            return exit_code.value == 259  # STILL_ACTIVE
+        return False
+    finally:
+        kernel32.CloseHandle(handle)
+
+
+def _acquire_self_lock() -> bool:
+    LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
+    if LOCK_PATH.is_file():
+        try:
+            existing = json.loads(LOCK_PATH.read_text(encoding="utf-8"))
+            if _pid_alive(int(existing.get("pid", 0))):
+                return False
+        except Exception:
+            pass
+    LOCK_PATH.write_text(
+        json.dumps({"pid": os.getpid(), "started_at": datetime.now(timezone.utc).isoformat()}),
+        encoding="utf-8",
+    )
+    return True
+
+
+def _console_env_files() -> list[Path]:
+    return [path for path in ENV_FILES if path.is_file()]
+
+
+def _spawn(argv: list[str], log_dir: Path, log_prefix: str) -> int:
+    log_dir.mkdir(parents=True, exist_ok=True)
+    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    log_path = log_dir / f"{log_prefix}-{stamp}.log"
+    kwargs: dict[str, object] = {}
+    if os.name == "nt":
+        kwargs["creationflags"] = (
+            subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP
+        )
+    else:
+        kwargs["start_new_session"] = True
+    with log_path.open("a", encoding="utf-8", newline="\n") as stream:
+        process = subprocess.Popen(
+            argv,
+            cwd=str(PROJECT_ROOT),
+            env=os.environ,
+            stdin=subprocess.DEVNULL,
+            stdout=stream,
+            stderr=subprocess.STDOUT,
+            shell=False,
+            **kwargs,
+        )
+    return process.pid
+
+
+def _console_argv() -> list[str]:
+    argv = [
+        str(PROJECT_ROOT / ".venv" / "Scripts" / "python.exe"),
+        "-m",
+        "gyxx_flow",
+        "console",
+        "--host",
+        CONSOLE_HOST,
+        "--port",
+        str(CONSOLE_PORT),
+    ]
+    for env_file in _console_env_files():
+        argv.extend(("--env-file", str(env_file)))
+    return argv
+
+
+def _scheduler_argv() -> list[str]:
+    argv = [
+        str(PROJECT_ROOT / ".venv" / "Scripts" / "python.exe"),
+        "-m",
+        "gyxx_flow",
+        "schedule",
+        "run",
+    ]
+    for env_file in _console_env_files():
+        argv.extend(("--env-file", str(env_file)))
+    return argv
+
+
+def _probe_console() -> bool:
+    try:
+        connection = http.client.HTTPConnection(CONSOLE_HOST, CONSOLE_PORT, timeout=PROBE_TIMEOUT_SECONDS)
+        try:
+            connection.request("GET", "/api/health")
+            response = connection.getresponse()
+            response.read()
+            return response.status == 200
+        finally:
+            connection.close()
+    except Exception:
+        return False
+
+
+def _port_owner_pid(port: int) -> int | None:
+    try:
+        output = subprocess.run(
+            ["netstat", "-ano", "-p", "tcp"],
+            capture_output=True,
+            text=True,
+            timeout=10,
+        ).stdout
+    except Exception:
+        return None
+    needle = f"{CONSOLE_HOST}:{port}"
+    for line in output.splitlines():
+        parts = line.split()
+        if (
+            len(parts) >= 5
+            and parts[0].upper() == "TCP"
+            and parts[1] == needle
+            and parts[3].upper() == "LISTENING"
+        ):
+            try:
+                return int(parts[4])
+            except ValueError:
+                return None
+    return None
+
+
+def _kill_pid_tree(pid: int) -> None:
+    if pid <= 0 or pid == os.getpid():
+        return
+    subprocess.run(
+        ["taskkill", "/PID", str(pid), "/T", "/F"],
+        capture_output=True,
+        timeout=30,
+    )
+
+
+def _scheduler_lock_pid() -> int | None:
+    try:
+        payload = json.loads(SCHEDULER_SERVICE_LOCK.read_text(encoding="utf-8"))
+        return int(payload.get("pid", 0)) or None
+    except Exception:
+        return None
+
+
+class _TargetState:
+    """记录单目标的连续失败计数、冷却与熔断状态。"""
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.failures = 0
+        self.last_restart_at = 0.0
+        self.recent_restarts: list[float] = []
+        self.tripped_at: float | None = None
+
+    def tripped(self, now: float) -> bool:
+        if self.tripped_at is None:
+            return False
+        if now - self.tripped_at >= TRIP_PAUSE_SECONDS:
+            self.tripped_at = None
+            self.recent_restarts.clear()
+            _log("keepalive-trip-reset", f"target={self.name} 熔断窗口结束恢复守护")
+            return False
+        return True
+
+    def may_restart(self, now: float) -> bool:
+        if self.tripped(now):
+            return False
+        if now - self.last_restart_at < RESTART_COOLDOWN_SECONDS:
+            return False
+        self.recent_restarts = [
+            stamp for stamp in self.recent_restarts if now - stamp < TRIP_WINDOW_SECONDS
+        ]
+        if len(self.recent_restarts) >= TRIP_MAX_RESTARTS:
+            self.tripped_at = now
+            _log(
+                "keepalive-tripped",
+                f"target={self.name} {TRIP_MAX_RESTARTS} 次重启/{TRIP_WINDOW_SECONDS // 60}分钟 "
+                f"暂停拉起{TRIP_PAUSE_SECONDS // 60}分钟，请人工检查日志",
+            )
+            return False
+        return True
+
+    def record_restart(self, now: float) -> None:
+        self.last_restart_at = now
+        self.recent_restarts.append(now)
+
+
+def run_cycle(console: _TargetState, scheduler: _TargetState) -> None:
+    now = time.time()
+
+    # ---- 控制台 ----
+    if _probe_console():
+        console.failures = 0
+    else:
+        console.failures += 1
+        owner = _port_owner_pid(CONSOLE_PORT)
+        _log(
+            "console-probe-failed",
+            f"{console.failures}/{PROBE_FAILURE_LIMIT} port_owner_pid={owner}",
+        )
+        if console.failures >= PROBE_FAILURE_LIMIT and console.may_restart(now):
+            if owner is not None and _pid_alive(owner):
+                _log("console-kill-wedged", f"pid={owner} 探测持续失败，结束进程树")
+                _kill_pid_tree(owner)
+                time.sleep(3)
+            pid = _spawn(_console_argv(), CONSOLE_LOG_DIR, "keepalive-console")
+            console.record_restart(time.time())
+            console.failures = 0
+            _log("console-restarted", f"pid={pid}")
+
+    # ---- 调度器 ----
+    scheduler_pid = _scheduler_lock_pid()
+    scheduler_alive = scheduler_pid is not None and _pid_alive(scheduler_pid)
+    if not scheduler_alive and scheduler.may_restart(now):
+        pid = _spawn(_scheduler_argv(), SCHEDULER_LOG_DIR, "keepalive-start")
+        scheduler.record_restart(time.time())
+        _log(
+            "scheduler-restarted",
+            f"pid={pid} prev_lock_pid={scheduler_pid}",
+        )
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="GYXX Flow 控制台/调度器保活")
+    parser.add_argument("--once", action="store_true", help="单轮探测+处置后退出")
+    arguments = parser.parse_args(argv)
+
+    if not _acquire_self_lock():
+        _log("keepalive-duplicate-exit", "已有保活进程在运行")
+        return 0
+
+    console = _TargetState("console")
+    scheduler = _TargetState("scheduler")
+    missing_env = [str(path) for path in ENV_FILES if not path.is_file()]
+    _log(
+        "keepalive-start",
+        f"pid={os.getpid()} once={arguments.once}"
+        + (f" missing_env={missing_env}" if missing_env else ""),
+    )
+    try:
+        while True:
+            run_cycle(console, scheduler)
+            if arguments.once:
+                _log("keepalive-once-done", "")
+                return 0
+            time.sleep(LOOP_INTERVAL_SECONDS)
+    except KeyboardInterrupt:
+        return 0
+    finally:
+        try:
+            if LOCK_PATH.is_file():
+                payload = json.loads(LOCK_PATH.read_text(encoding="utf-8"))
+                if int(payload.get("pid", 0)) == os.getpid():
+                    LOCK_PATH.unlink()
+        except Exception:
+            pass
+
+
+if __name__ == "__main__":
+    sys.exit(main())

deploy/watchdog/start-keepalive.vbs

@@ -0,0 +1,5 @@
+' GYXX Flow 保活登录自启动：无窗口启动 deploy/watchdog/gyxx_keepalive.py，
+' 由它负责拉起控制台（含进程内自检看门狗）与调度器。
+' 安装：复制到 %APPDATA%\Microsoft\Windows\Start Menu\Programs\Startup\
+CreateObject("WScript.Shell").Run _
+    """D:\gyxx-flow\.venv\Scripts\pythonw.exe"" ""D:\gyxx-flow\deploy\watchdog\gyxx_keepalive.py""", 0, False

src/gyxx_flow/cookie_gate.py

@@ -0,0 +1,882 @@
+"""Per-account cookie health gate injected before workflow collection steps.
+
+Every runtime account a workflow depends on is checked before the collection
+steps that use it may run.  Accounts fall into three gate modes derived from
+``config/runtime-bindings.json``:
+
+* ``blocking`` -- QR-code / SMS-only accounts (no credential environment and
+  no configured login command).  The gate verifies the login state and, when
+  it has expired, fails the gate step so every dependent step is skipped with
+  ``dependency-failed`` and sends a Feishu alert asking the login owner to
+  re-login interactively via ``gyxx accounts login <account_id>``.
+* ``recover`` -- accounts with a vault-managed credential login command
+  (``login_mode`` ``C``).  The keepalive manager re-runs the credential login
+  automatically; only a failed recovery blocks the workflow and alerts.
+* ``defer`` -- accounts whose member scripts embed username/password login
+  flows.  These keep their script-level login behaviour unchanged and are not
+  gated at all.
+
+The per-account decision is persisted as a small state machine under
+``state/accounts/<id>/cookie_state.json`` (``healthy`` <-> ``awaiting_login``)
+so repeat failures alert at most once per notify interval instead of once per
+workflow run.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from collections.abc import Callable
+from dataclasses import dataclass, replace
+from datetime import date, datetime, timezone
+from pathlib import Path
+from typing import Any, Mapping
+
+from gyxx_flow.accounts import (
+    AccountKeepaliveManager,
+    _keepalive_requires_relogin,
+    _read_keepalive_status,
+    _vault_cookie_valid,
+)
+from gyxx_flow.adapters.browser import BrowserCookieStore
+from gyxx_flow.adapters.feishu_messages import send_lark_bot_message
+from gyxx_flow.adapters.integration import RuntimeIntegrationCatalog
+from gyxx_flow.catalog import WorkflowCatalog
+from gyxx_flow.core.artifacts import atomic_write_json
+from gyxx_flow.core.config import Settings
+from gyxx_flow.core.context import RunContext
+from gyxx_flow.notification_routing import (
+    ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID,
+    ANALYZER_NOTIFICATION_APP_PROFILE,
+    NotificationRoutingStore,
+)
+from gyxx_flow.workflow.model import StepDefinition, WorkflowDefinition
+from gyxx_flow.workflow.steps import StepExecution
+
+GATE_MODE_BLOCKING = "blocking"
+GATE_MODE_RECOVER = "recover"
+GATE_MODE_DEFER = "defer"
+
+# 飞书通知的默认接收人：按手机号在 config/notification-routing.json 的
+# people 段解析 open_id；解析失败时回退到同名 analyzer 常量收件人。
+DEFAULT_LOGIN_ALERT_MOBILE = "+8613663785145"
+
+GYXX_COOKIE_GATE = "GYXX_COOKIE_GATE"
+GYXX_COOKIE_GATE_MAX_STATUS_AGE_SECONDS = "GYXX_COOKIE_GATE_MAX_STATUS_AGE_SECONDS"
+GYXX_LOGIN_ALERT_OPEN_IDS = "GYXX_LOGIN_ALERT_OPEN_IDS"
+GYXX_LOGIN_ALERT_APP_PROFILE = "GYXX_LOGIN_ALERT_APP_PROFILE"
+GYXX_LOGIN_ALERT_NOTIFY_INTERVAL_SECONDS = "GYXX_LOGIN_ALERT_NOTIFY_INTERVAL_SECONDS"
+
+_GATE_STEP_PREFIX = "cookie-gate"
+_GATE_STATE_FILENAME = "cookie_state.json"
+_GATE_TIMEOUT_SECONDS = 1800.0
+_DEFAULT_MAX_STATUS_AGE_SECONDS = 3600.0
+_MIN_MAX_STATUS_AGE_SECONDS = 300.0
+_DEFAULT_NOTIFY_INTERVAL_SECONDS = 1800.0
+_DISABLED_VALUES = frozenset({"0", "false", "off", "no", "disabled"})
+# 自动重跑：待重跑清单上限、健康探活节流、同账号重跑启动冷却。
+_MAX_PENDING_RERUNS = 20
+_RECOVERY_PROBE_INTERVAL_SECONDS = 300.0
+_RECOVERY_LAUNCH_COOLDOWN_SECONDS = 300.0
+_RECOVERY_HISTORY_LIMIT = 20
+
+STATE_HEALTHY = "healthy"
+STATE_AWAITING_LOGIN = "awaiting_login"
+
+
+@dataclass(frozen=True, slots=True)
+class GateVerdict:
+    """One account's gate decision, recorded in the run journal on failure."""
+
+    account_id: str
+    mode: str
+    blocked: bool
+    message: str
+    recovered: bool = False
+    notified: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class CookieGateStepAction:
+    """ExecutableStep that turns one account verdict into a step outcome."""
+
+    gate: "CookieGate"
+    account_id: str
+
+    def execute(
+        self,
+        *,
+        context: RunContext,
+        timeout_seconds: float | None,
+        dry_run: bool,
+    ) -> StepExecution:
+        if dry_run:  # pragma: no cover - the engine skips dry-run steps first
+            return StepExecution(exit_code=0, skipped=True, reason="dry-run")
+        verdict = self.gate.check(self.account_id, context=context)
+        if verdict.blocked:
+            return StepExecution(
+                exit_code=1,
+                error=verdict.message,
+                reason="cookie-expired",
+            )
+        return StepExecution(exit_code=0)
+
+
+class CookieGate:
+    """Resolve, probe, and gate the accounts one workflow depends on."""
+
+    def __init__(
+        self,
+        settings: Settings,
+        *,
+        catalog: RuntimeIntegrationCatalog | None = None,
+        keepalive: AccountKeepaliveManager | None = None,
+        environment: Mapping[str, str] | None = None,
+        now: Callable[[], datetime] | None = None,
+        sender: Callable[..., dict[str, Any]] | None = None,
+    ) -> None:  # noqa: D107 - simple wiring
+        self.settings = settings
+        self.catalog = catalog or RuntimeIntegrationCatalog.load_default(
+            project_root=settings.project_root,
+            data_root=settings.data_root,
+        )
+        self.keepalive = keepalive or AccountKeepaliveManager(
+            settings, catalog=self.catalog
+        )
+        self._environment = dict(os.environ if environment is None else environment)
+        self._now = now or (lambda: datetime.now(timezone.utc))
+        self._sender = sender or send_lark_bot_message
+        self._workflow_catalog: WorkflowCatalog | None = None
+
+    # ------------------------------------------------------------------
+    # Account resolution
+    # ------------------------------------------------------------------
+
+    def gate_mode(self, account_id: str) -> str:
+        account = self.catalog.account_for(account_id)
+        if str(account.login_command_id or "").strip():
+            return GATE_MODE_RECOVER
+        if account.credential_env_names or account.login_mode == "B":
+            return GATE_MODE_DEFER
+        return GATE_MODE_BLOCKING
+
+    def collect_workflow_accounts(
+        self,
+        workflow: WorkflowDefinition,
+    ) -> dict[str, set[str]]:
+        """Map each gated account to the step ids that depend on it.
+
+        Accounts resolve two ways: directly, from each step binding's
+        ``account``; and declaratively, from the catalog workflow's
+        ``required_accounts`` (used by orchestrator entries whose own binding
+        has no account while the spawned child collectors do).
+        """
+
+        accounts: dict[str, set[str]] = {}
+        for step in workflow.steps:
+            account_id = self._account_for_step(step)
+            if account_id:
+                accounts.setdefault(account_id, set()).add(step.step_id)
+        for account_id in self._declared_required_accounts(workflow.workflow_id):
+            self.catalog.account_for(account_id)  # unknown id: loud config error
+            accounts.setdefault(account_id, set()).update(
+                step.step_id for step in workflow.steps
+            )
+        # Defer-mode accounts keep their script-level credential login flows
+        # and are intentionally not gated.
+        return {
+            account_id: step_ids
+            for account_id, step_ids in accounts.items()
+            if self.gate_mode(account_id) != GATE_MODE_DEFER
+        }
+
+    def _account_for_step(self, step: StepDefinition) -> str | None:
+        action = step.action
+        entry = getattr(action, "entry", None)
+        module_id = getattr(entry, "module", "")
+        command_entry = getattr(action, "command_entry", None) or getattr(
+            entry, "entry", ""
+        )
+        if not module_id or not command_entry:
+            return None
+        try:
+            binding = self.catalog.binding_for(f"{module_id}:{command_entry}")
+        except Exception:  # noqa: BLE001 - unbound entries run ungated
+            return None
+        return binding.account or None
+
+    def workflow_catalog(self) -> WorkflowCatalog:
+        """Load (once) the catalog used for declared account and rerun lookups."""
+
+        if self._workflow_catalog is None:
+            try:
+                self._workflow_catalog = WorkflowCatalog.load(
+                    self.settings.project_root / "config"
+                )
+            except Exception:  # noqa: BLE001 - catalog issues surface elsewhere
+                self._workflow_catalog = WorkflowCatalog(
+                    timezone="UTC", workflows=(), schedules=()
+                )
+        return self._workflow_catalog
+
+    def _declared_required_accounts(self, workflow_id: str) -> tuple[str, ...]:
+        for entry in self.workflow_catalog().workflows:
+            if entry.workflow_id == workflow_id:
+                return tuple(entry.required_accounts)
+        return ()
+
+    # ------------------------------------------------------------------
+    # State machine
+    # ------------------------------------------------------------------
+
+    def check(self, account_id: str, *, context: RunContext) -> GateVerdict:
+        account = self.catalog.account_for(account_id)
+        mode = self.gate_mode(account_id)
+        store = BrowserCookieStore(account.cookie_file, account.storage_state_file)
+        static_valid = _vault_cookie_valid(account, store)
+        status = _read_keepalive_status(account)
+        if static_valid and self._keepalive_success_is_fresh(status):
+            self._write_state(account, STATE_HEALTHY, probe="fresh-keepalive")
+            return GateVerdict(
+                account_id=account_id,
+                mode=mode,
+                blocked=False,
+                message=f"账号 {account_id} 登录态有效（keepalive 近期验证通过）",
+            )
+        if self._acceptance_enabled():
+            # 验收模式不开浏览器：按静态证据判定，且不投递飞书通知。
+            if static_valid and not _keepalive_requires_relogin(status):
+                return GateVerdict(
+                    account_id=account_id,
+                    mode=mode,
+                    blocked=False,
+                    message=f"账号 {account_id} 静态 cookie 校验通过（验收模式）",
+                )
+            return self._expire(
+                account,
+                mode,
+                context,
+                detail="验收模式下静态 cookie 校验未通过",
+                notify=False,
+            )
+
+        outcome = self.keepalive.refresh_account(account_id)
+        probe_status = str(outcome.get("status", "error"))
+        if probe_status == "success":
+            recovered = bool(outcome.get("login_recovered"))
+            self._write_state(
+                account,
+                STATE_HEALTHY,
+                probe="success",
+                recovered=recovered,
+            )
+            message = (
+                f"账号 {account_id} 登录态有效（自动账密重登成功）"
+                if recovered
+                else f"账号 {account_id} 登录态有效（探活通过）"
+            )
+            return GateVerdict(
+                account_id=account_id,
+                mode=mode,
+                blocked=False,
+                message=message,
+                recovered=recovered,
+            )
+        if probe_status == "busy":
+            if static_valid and not _keepalive_requires_relogin(status):
+                return GateVerdict(
+                    account_id=account_id,
+                    mode=mode,
+                    blocked=False,
+                    message=f"账号 {account_id} 探活锁忙，静态 cookie 校验通过",
+                )
+            return self._expire(
+                account,
+                mode,
+                context,
+                detail="探活锁被占用且静态 cookie 校验未通过",
+            )
+        if probe_status in {"expired", "not_logged_in"}:
+            evidence = _recovery_evidence(_read_keepalive_status(account))
+            detail = "cookie 已过期，无法正常采集"
+            if probe_status == "not_logged_in" and evidence:
+                detail = f"登录态缺失；{evidence}"
+            elif evidence:
+                detail = f"cookie 已过期；{evidence}"
+            return self._expire(account, mode, context, detail=detail)
+        # timeout / error / http_error / disabled：探测结果不可信时回退静态证据。
+        if static_valid and not _keepalive_requires_relogin(status):
+            return GateVerdict(
+                account_id=account_id,
+                mode=mode,
+                blocked=False,
+                message=f"账号 {account_id} 探活异常（{probe_status}），静态 cookie 校验通过",
+            )
+        return self._expire(
+            account,
+            mode,
+            context,
+            detail=f"探活异常（{probe_status}）且静态 cookie 校验未通过",
+        )
+
+    # ------------------------------------------------------------------
+    # Expiry, notification, persisted state
+    # ------------------------------------------------------------------
+
+    def _expire(
+        self,
+        account: Any,
+        mode: str,
+        context: RunContext,
+        *,
+        detail: str,
+        notify: bool | None = None,
+    ) -> GateVerdict:
+        account_id = str(account.account_id)
+        if notify is None:
+            notify = not context.shadow and not self._acceptance_enabled()
+        state = self._read_state(account)
+        notified = False
+        notification_note = ""
+        if notify and not self._within_notify_interval(state):
+            recipients = self._alert_recipients()
+            message = _expiry_message(account_id, mode, context, detail)
+            for open_id, profile in recipients:
+                try:
+                    self._sender(
+                        user_id=open_id,
+                        text=message,
+                        idempotency_key=f"cookie-gate:{account_id}:{context.run_id}",
+                        profile=profile,
+                    )
+                    notified = True
+                except Exception as exc:  # noqa: BLE001 - alerting must not mask the verdict
+                    notification_note = f"{type(exc).__name__}: {exc}"[:300]
+            if recipients and notified:
+                state["last_notified_at"] = _now_iso(self._now)
+                state.pop("notification_error", None)
+            elif notification_note:
+                state["notification_error"] = notification_note
+        state["state"] = STATE_AWAITING_LOGIN
+        state["updated_at"] = _now_iso(self._now)
+        state["last_probe_status"] = "expired"
+        state["last_blocked_run"] = context.run_id
+        if not context.shadow and not self._acceptance_enabled():
+            # 记录待自动重跑清单：登录态恢复后由调度器维护循环按原业务日期重跑。
+            state["pending_reruns"] = _record_pending_rerun(
+                state.get("pending_reruns"), context, _now_iso(self._now)
+            )
+        self._save_state(account, state)
+        message = _expiry_message(account_id, mode, context, detail)
+        if notified:
+            message += " 已通过飞书通知登录负责人。"
+        elif notification_note:
+            message += f" 飞书通知发送失败（{notification_note}）。"
+        return GateVerdict(
+            account_id=account_id,
+            mode=mode,
+            blocked=True,
+            message=message,
+            notified=notified,
+        )
+
+    def _write_state(
+        self,
+        account: Any,
+        state_name: str,
+        *,
+        probe: str,
+        recovered: bool = False,
+    ) -> None:
+        state = self._read_state(account)
+        if (
+            state_name == STATE_HEALTHY
+            and state.get("state") == STATE_HEALTHY
+            and state.get("last_probe_status") in {"fresh-keepalive", "success"}
+            and not recovered
+        ):
+            return  # 持续健康时避免每次运行都重写状态文件
+        state["state"] = state_name
+        state["updated_at"] = _now_iso(self._now)
+        state["last_probe_status"] = probe
+        if state_name == STATE_HEALTHY:
+            state["last_healthy_at"] = _now_iso(self._now)
+            if recovered:
+                state["last_recovered_at"] = _now_iso(self._now)
+            state.pop("notification_error", None)
+        self._save_state(account, state)
+
+    def _state_path(self, account: Any) -> Path:
+        return account.cookie_file.with_name(_GATE_STATE_FILENAME)
+
+    def _read_state(self, account: Any) -> dict[str, object]:
+        path = self._state_path(account)
+        if not path.exists():
+            return {"account_id": str(account.account_id)}
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            return {"account_id": str(account.account_id)}
+        if not isinstance(payload, dict):
+            return {"account_id": str(account.account_id)}
+        return dict(payload)
+
+    def _save_state(self, account: Any, state: Mapping[str, object]) -> None:
+        payload = {
+            "account_id": str(account.account_id),
+            **state,
+        }
+        try:
+            atomic_write_json(self._state_path(account), payload)
+        except OSError:  # pragma: no cover - state file is advisory only
+            pass
+
+    def _within_notify_interval(self, state: Mapping[str, object]) -> bool:
+        raw = state.get("last_notified_at")
+        if not isinstance(raw, str):
+            return False
+        try:
+            last = datetime.fromisoformat(raw)
+        except ValueError:
+            return False
+        now = self._now()
+        if last.tzinfo is None:
+            last = last.replace(tzinfo=timezone.utc)
+        interval = _env_int(
+            self._environment,
+            GYXX_LOGIN_ALERT_NOTIFY_INTERVAL_SECONDS,
+            _DEFAULT_NOTIFY_INTERVAL_SECONDS,
+            minimum=0,
+        )
+        return (now - last).total_seconds() < interval
+
+    def _keepalive_success_is_fresh(self, status: Mapping[str, object] | None) -> bool:
+        if not isinstance(status, Mapping) or status.get("status") != "success":
+            return False
+        raw = status.get("last_success_at")
+        if not isinstance(raw, str):
+            return False
+        try:
+            last = datetime.fromisoformat(raw)
+        except ValueError:
+            return False
+        now = self._now()
+        if last.tzinfo is None:
+            last = last.replace(tzinfo=timezone.utc)
+        max_age = _env_float(
+            self._environment,
+            GYXX_COOKIE_GATE_MAX_STATUS_AGE_SECONDS,
+            _DEFAULT_MAX_STATUS_AGE_SECONDS,
+            minimum=_MIN_MAX_STATUS_AGE_SECONDS,
+        )
+        return (now - last).total_seconds() <= max_age
+
+    def _alert_recipients(self) -> list[tuple[str, str]]:
+        profile = (
+            self._environment.get(GYXX_LOGIN_ALERT_APP_PROFILE, "").strip()
+            or ANALYZER_NOTIFICATION_APP_PROFILE
+        )
+        override = [
+            value.strip()
+            for value in self._environment.get(GYXX_LOGIN_ALERT_OPEN_IDS, "").split(",")
+            if value.strip()
+        ]
+        if override:
+            return [(open_id, profile) for open_id in override]
+        try:
+            payload, _revision = NotificationRoutingStore(self.settings).snapshot()
+            for person in payload["people"].values():
+                if person.get("mobile") == DEFAULT_LOGIN_ALERT_MOBILE:
+                    binding = person.get("bindings", {}).get(profile)
+                    if binding:
+                        return [(str(binding["open_id"]), profile)]
+        except Exception:  # noqa: BLE001 - fall back to the constant recipient
+            pass
+        return [(ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID, profile)]
+
+    def _acceptance_enabled(self) -> bool:
+        value = self._environment.get("GYXX_WORKFLOW_ACCEPTANCE", "")
+        return bool(value and value.strip().casefold() in {"1", "true", "yes", "on"})
+
+
+def gate_enabled(environment: Mapping[str, str] | None = None) -> bool:
+    values = os.environ if environment is None else environment
+    return values.get(GYXX_COOKIE_GATE, "").strip().casefold() not in _DISABLED_VALUES
+
+
+def apply_cookie_gates(
+    workflow: WorkflowDefinition,
+    *,
+    settings: Settings,
+    environment: Mapping[str, str] | None = None,
+    gate: CookieGate | None = None,
+) -> WorkflowDefinition:
+    """Inject one critical gate step per gated account in front of its users.
+
+    Gate steps are graph sources; each step that uses an account gains the
+    account's gate as an extra dependency.  A failed gate therefore skips its
+    dependents with ``dependency-failed`` through the regular engine, while
+    unrelated branches keep running.  Operator slices (``--rerun-step`` /
+    ``--resume-from``) are produced before this injection and stay gate-free.
+    """
+
+    if not workflow.steps:
+        return workflow
+    if gate is None:
+        try:
+            gate = CookieGate(
+                settings,
+                environment=os.environ if environment is None else environment,
+            )
+        except Exception:  # noqa: BLE001 - 无集成配置时不门控；步骤执行仍会显式报错
+            return workflow
+    account_steps = gate.collect_workflow_accounts(workflow)
+    if not account_steps:
+        return workflow
+    existing_ids = {step.step_id for step in workflow.steps}
+    gate_steps: list[StepDefinition] = []
+    for account_id in sorted(account_steps):
+        step_id = f"{_GATE_STEP_PREFIX}.{account_id}"
+        if step_id in existing_ids:
+            continue
+        gate_steps.append(
+            StepDefinition(
+                step_id,
+                CookieGateStepAction(gate, account_id),
+                timeout_seconds=_GATE_TIMEOUT_SECONDS,
+                max_attempts=1,
+            )
+        )
+    if not gate_steps:
+        return workflow
+    patched: list[StepDefinition] = []
+    for step in workflow.steps:
+        dependencies = list(dict.fromkeys(step.depends_on))
+        changed = False
+        for account_id, step_ids in account_steps.items():
+            gate_id = f"{_GATE_STEP_PREFIX}.{account_id}"
+            if step.step_id in step_ids and gate_id not in dependencies:
+                dependencies.append(gate_id)
+                changed = True
+        if changed:
+            step = replace(step, depends_on=tuple(dependencies))
+        patched.append(step)
+    return WorkflowDefinition(
+        workflow.workflow_id,
+        (*gate_steps, *patched),
+    )
+
+
+class CookieGateRecoveryManager:
+    """Re-run workflows blocked by expired cookies once the login recovers.
+
+    Runs inside the resident scheduler's maintenance loop.  For every account
+    with a non-empty ``pending_reruns`` list, it re-verifies the login state
+    (fresh keepalive success first, then a throttled live probe); on recovery
+    it relaunches each recorded workflow with its original business date
+    through the scheduler's own launcher, clears the pending list, and sends a
+    Feishu confirmation to the login owner.
+    """
+
+    def __init__(
+        self,
+        gate: CookieGate,
+        *,
+        launcher: Any,
+        probe_interval_seconds: float = _RECOVERY_PROBE_INTERVAL_SECONDS,
+        launch_cooldown_seconds: float = _RECOVERY_LAUNCH_COOLDOWN_SECONDS,
+    ) -> None:
+        self.gate = gate
+        self._launcher = launcher
+        self._probe_interval = probe_interval_seconds
+        self._launch_cooldown = launch_cooldown_seconds
+
+    def tick(self) -> list[dict[str, object]]:
+        """Attempt recovery reruns once; returns redacted outcomes per account."""
+
+        outcomes: list[dict[str, object]] = []
+        for account in self.gate.catalog.accounts:
+            state = self.gate._read_state(account)
+            pending = state.get("pending_reruns")
+            if not isinstance(pending, list) or not pending:
+                continue
+            now = self.gate._now()
+            if not self._launch_cooldown_passed(state, now):
+                continue
+            if not self._account_recovered(account, state, now):
+                continue
+            launched, remaining = self._launch_pending(account, pending, now)
+            state["pending_reruns"] = remaining
+            state["updated_at"] = _now_iso(self.gate._now)
+            state["last_recovered_at"] = state["updated_at"]
+            if not remaining:
+                state["state"] = STATE_HEALTHY
+                state["last_probe_status"] = "recovered-rerun"
+            if launched:
+                state["last_rerun_launch_at"] = state["updated_at"]
+                history = [
+                    item
+                    for item in state.get("rerun_history", [])
+                    if isinstance(item, dict)
+                ]
+                history.extend(launched)
+                state["rerun_history"] = history[-_RECOVERY_HISTORY_LIMIT:]
+            self.gate._save_state(account, state)
+            if launched:
+                self._notify_recovered(str(account.account_id), launched)
+            outcomes.append(
+                {
+                    "account_id": str(account.account_id),
+                    "launched": launched,
+                    "pending": len(remaining),
+                }
+            )
+        return outcomes
+
+    # ------------------------------------------------------------------
+    # internals
+    # ------------------------------------------------------------------
+
+    def _launch_cooldown_passed(
+        self, state: Mapping[str, object], now: datetime
+    ) -> bool:
+        raw = state.get("last_rerun_launch_at")
+        if not isinstance(raw, str):
+            return True
+        try:
+            last = datetime.fromisoformat(raw)
+        except ValueError:
+            return True
+        if last.tzinfo is None:
+            last = last.replace(tzinfo=timezone.utc)
+        return (now - last).total_seconds() >= self._launch_cooldown
+
+    def _account_recovered(
+        self,
+        account: Any,
+        state: dict[str, object],
+        now: datetime,
+    ) -> bool:
+        """Fresh keepalive success first; otherwise one throttled live probe."""
+
+        store = BrowserCookieStore(account.cookie_file, account.storage_state_file)
+        if _vault_cookie_valid(account, store) and self.gate._keepalive_success_is_fresh(
+            _read_keepalive_status(account)
+        ):
+            return True
+        raw = state.get("last_recovery_probe_at")
+        if isinstance(raw, str):
+            try:
+                last_probe = datetime.fromisoformat(raw)
+            except ValueError:
+                last_probe = None
+            if last_probe is not None:
+                if last_probe.tzinfo is None:
+                    last_probe = last_probe.replace(tzinfo=timezone.utc)
+                if (now - last_probe).total_seconds() < self._probe_interval:
+                    return False
+        state["last_recovery_probe_at"] = _now_iso(self.gate._now)
+        self.gate._save_state(account, state)
+        try:
+            outcome = self.gate.keepalive.refresh_account(str(account.account_id))
+        except Exception:  # noqa: BLE001 - 恢复探活失败留给下一轮
+            return False
+        return str(outcome.get("status", "error")) == "success"
+
+    def _launch_pending(
+        self,
+        account: Any,
+        pending: list[object],
+        now: datetime,
+    ) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
+        known_ids = {
+            entry.workflow_id for entry in self.gate.workflow_catalog().workflows
+        }
+        launched: list[dict[str, object]] = []
+        remaining: list[dict[str, object]] = []
+        for item in pending:
+            if not isinstance(item, dict):
+                continue  # 控制台脚本等临时运行 id：不在 catalog 中，直接丢弃
+            workflow_id = item.get("workflow_id")
+            business_date = item.get("business_date")
+            if not isinstance(workflow_id, str) or not isinstance(business_date, str):
+                continue
+            if workflow_id not in known_ids:
+                continue
+            try:
+                process = self._launcher.launch(
+                    workflow_id, now, date.fromisoformat(business_date)
+                )
+            except Exception as exc:  # noqa: BLE001 - 保留 pending，下一轮重试
+                remaining.append(
+                    {
+                        "workflow_id": workflow_id,
+                        "business_date": business_date,
+                        "error": f"{type(exc).__name__}: {exc}"[:200],
+                    }
+                )
+                continue
+            close = getattr(self._launcher, "close", None)
+            if callable(close):
+                try:
+                    close(process)
+                except Exception:  # noqa: BLE001 - 关闭日志句柄失败不影响重跑
+                    pass
+            launched.append(
+                {
+                    "workflow_id": workflow_id,
+                    "business_date": business_date,
+                    "pid": int(getattr(process, "pid", 0) or 0),
+                    "launched_at": _now_iso(self.gate._now),
+                }
+            )
+        return launched, remaining
+
+    def _notify_recovered(
+        self,
+        account_id: str,
+        launched: list[dict[str, object]],
+    ) -> None:
+        lines = [
+            f"- {item['workflow_id']}（业务日期 {item['business_date']}）"
+            for item in launched
+        ]
+        message = (
+            "【GYXX 登录态已恢复】账号 "
+            f"{account_id} 登录态已恢复，已自动重跑被阻断的工作流：\n"
+            + "\n".join(lines)
+            + "\n如本次重跑仍有失败，请查看运行日志或工作流控制台。"
+        )
+        key = "-".join(
+            [account_id, *(str(item["workflow_id"]) for item in launched)]
+        )
+        for open_id, profile in self.gate._alert_recipients():
+            try:
+                self.gate._sender(
+                    user_id=open_id,
+                    text=message,
+                    idempotency_key=f"cookie-recovery:{key}",
+                    profile=profile,
+                )
+            except Exception:  # noqa: BLE001 - 确认通知失败不影响已发起的重跑
+                continue
+
+
+def _record_pending_rerun(
+    current: object,
+    context: RunContext,
+    blocked_at: str,
+) -> list[dict[str, object]]:
+    """Append one blocked run to the pending list, deduped and bounded."""
+
+    pending: list[dict[str, object]] = [
+        dict(item)
+        for item in (current if isinstance(current, list) else [])
+        if isinstance(item, dict)
+    ]
+    entry = {
+        "workflow_id": context.workflow_id,
+        "business_date": context.business_date.isoformat(),
+        "blocked_run_id": context.run_id,
+        "blocked_at": blocked_at,
+    }
+    pending = [
+        item
+        for item in pending
+        if not (
+            item.get("workflow_id") == entry["workflow_id"]
+            and item.get("business_date") == entry["business_date"]
+        )
+    ]
+    pending.append(entry)
+    return pending[-_MAX_PENDING_RERUNS:]
+
+
+def _recovery_evidence(status: Mapping[str, object] | None) -> str:
+    if not isinstance(status, Mapping):
+        return ""
+    if not status.get("login_attempted"):
+        return ""
+    reason = status.get("login_reason")
+    error_type = status.get("login_error_type")
+    exit_code = status.get("login_exit_code")
+    parts = ["自动账密重登已尝试但未成功"]
+    if isinstance(reason, str) and reason:
+        parts.append(reason)
+    if isinstance(error_type, str) and error_type:
+        parts.append(f"错误={error_type}")
+    if isinstance(exit_code, int) and not isinstance(exit_code, bool):
+        parts.append(f"退出码={exit_code}")
+    return "（" + "，".join(parts) + "）"
+
+
+def _expiry_message(
+    account_id: str,
+    mode: str,
+    context: RunContext,
+    detail: str,
+) -> str:
+    mode_note = (
+        "账号密码自动重登失败，需人工处理（如验证码）"
+        if mode == GATE_MODE_RECOVER
+        else "该平台仅支持扫码/验证码登录，无法自动恢复"
+    )
+    return (
+        "【GYXX 登录态失效】工作流被阻断，需要人工重新登录\n"
+        f"工作流: {context.workflow_id}（业务日期 {context.business_date.isoformat()}）\n"
+        f"账号: {account_id}\n"
+        f"原因: {detail}；{mode_note}\n"
+        f"请在服务器上执行: uv run gyxx accounts login {account_id}\n"
+        "（将打开浏览器窗口，完成扫码或验证码登录后自动保存并同步到全部成员脚本；"
+        "受影响的采集步骤本次已跳过，登录完成后下次运行自动恢复）"
+    )
+
+
+def _now_iso(now: Callable[[], datetime]) -> str:
+    return now().astimezone(timezone.utc).isoformat(timespec="seconds")
+
+
+def _env_int(
+    environment: Mapping[str, str],
+    name: str,
+    default: int,
+    *,
+    minimum: int,
+) -> int:
+    raw = environment.get(name, "").strip()
+    try:
+        value = int(raw)
+    except (TypeError, ValueError):
+        return default
+    return max(minimum, value)
+
+
+def _env_float(
+    environment: Mapping[str, str],
+    name: str,
+    default: float,
+    *,
+    minimum: float,
+) -> float:
+    raw = environment.get(name, "").strip()
+    try:
+        value = float(raw)
+    except (TypeError, ValueError):
+        return default
+    return max(minimum, value)
+
+
+__all__ = [
+    "DEFAULT_LOGIN_ALERT_MOBILE",
+    "GATE_MODE_BLOCKING",
+    "GATE_MODE_DEFER",
+    "GATE_MODE_RECOVER",
+    "GYXX_COOKIE_GATE",
+    "GYXX_LOGIN_ALERT_OPEN_IDS",
+    "CookieGate",
+    "CookieGateRecoveryManager",
+    "CookieGateStepAction",
+    "GateVerdict",
+    "apply_cookie_gates",
+    "gate_enabled",
+]