feat(ops): 执行日志与 cookie 失效遥测入库

新增 PostgreSQL 运维审计层，本地 run.json 仍是权威源，数据库做跨机器可查、
可长期留存的关系型镜像。

数据面：
- ops_workflow_runs / ops_workflow_run_steps：每次运行与每次步骤尝试一行，
  保留状态、模式、耗时、exit_code、错误与 journal 路径
- ops_account_cookie_events：登录态观测事件流，只存 cookie 指纹，不落明文值
- ops_account_cookie_lifetimes / ops_account_cookie_stats：一份 cookie 的实际
  寿命与账号级平均/中位/最短/最长，用于发现失效规律；同时记录 cookie 自带
  的声明过期时间，可与实际寿命对比区分自然到期与服务端提前吊销

采集点：
- RunIndex.index_journal 镜像运行记录，覆盖 CLI、控制台与调度器全部路径
- CookieGate 的 gate 判定与 AccountKeepaliveManager 的周期探活（默认 10 分钟）
  共同提供 cookie 观测，探活决定失效时间的精度

可用性：
- 新增 gyxx audit runs|cookies|cookie-stats|schema，支持 --json
- 首次写入自动幂等建表，建表用 advisory lock 串行化，避免并发 DDL 死锁
- 审计写入永不抛出异常：连续失败熔断冷却并落 logs/ops-audit 日志，
  数据库不可用不影响工作流执行

src/gyxx_flow/accounts.py

@@ -223,6 +223,8 @@ class AccountKeepaliveManager:
         self._next_due: dict[str, float] = {}
         self._locks = LockManager(settings.data_root / "state" / "locks")
         self._login_runner = login_runner or self._run_configured_login
+        self._audit_store: Any | None = None
+        self._audit_seen: dict[str, tuple[str, float]] = {}
 
     def tick(self) -> list[dict[str, object]]:
         """Run due account refreshes once; never starts work in dry-run mode."""
@@ -586,8 +588,63 @@ class AccountKeepaliveManager:
         if result.get("status") == "success":
             payload["last_success_at"] = timestamp
         atomic_write_json(path, payload)
+        self._audit_keepalive(account, result)
         return dict(result)
 
+    _AUDIT_STATE_BY_PROBE = {
+        "success": "healthy",
+        "expired": "expired",
+        "not_logged_in": "not_logged_in",
+        "busy": "busy",
+        "disabled": "disabled",
+    }
+
+    def _audit_keepalive(self, account: Any, result: dict[str, object]) -> None:
+        """Record probe outcomes so the real cookie lifetime becomes measurable.
+
+        The resident scheduler probes every ``keepalive_interval_seconds``
+        (10 minutes by default), which is what turns "logins seem to last about
+        a week" into a measured distribution.  Repeated identical outcomes
+        within an hour collapse into one row, so the event table tracks state
+        changes rather than polling noise.
+        """
+
+        try:
+            from gyxx_flow.ops.audit import snapshot_for_account
+
+            probe = str(result.get("status", "") or "error")
+            state = self._AUDIT_STATE_BY_PROBE.get(probe, "probe_error")
+            key = str(account.account_id)
+            previous = self._audit_seen.get(key)
+            if previous is not None:
+                previous_state, previous_at = previous
+                if previous_state == state and (self._clock() - previous_at) < 3600:
+                    return
+            store = self._audit()
+            if store is None or not store.enabled:
+                return
+            if store.record_cookie_event(
+                account_id=key,
+                state=state,
+                snapshot=snapshot_for_account(account),
+                probe_status=probe,
+                probe_source="keepalive",
+                detail=str(result.get("error_type", "") or ""),
+            ):
+                self._audit_seen[key] = (state, self._clock())
+        except Exception:  # noqa: BLE001 - telemetry must never break keepalive
+            return
+
+    def _audit(self) -> Any | None:
+        if self._audit_store is None:
+            try:
+                from gyxx_flow.ops.audit import AuditStore
+
+                self._audit_store = AuditStore(self.settings.data_root)
+            except Exception:  # noqa: BLE001
+                return None
+        return self._audit_store
+
 
 def _vault_cookie_valid(account: Any, store: BrowserCookieStore) -> bool:
     if not account.cookie_file.exists():

src/gyxx_flow/cli.py

@@ -229,6 +229,36 @@ def build_parser() -> argparse.ArgumentParser:
     status_parser = acceptance_commands.add_parser("status")
     status_parser.add_argument("--json", action="store_true")
     status_parser.add_argument("--output", type=Path)
+
+    audit_parser = commands.add_parser(
+        "audit", help="query the PostgreSQL operational audit tables"
+    )
+    audit_commands = audit_parser.add_subparsers(dest="audit_command", required=True)
+
+    audit_runs = audit_commands.add_parser("runs", help="recent workflow runs")
+    audit_runs.add_argument("--workflow")
+    audit_runs.add_argument("--status")
+    audit_runs.add_argument("--date", dest="business_date")
+    audit_runs.add_argument("--limit", type=int, default=30)
+    audit_runs.add_argument("--json", action="store_true")
+
+    audit_cookies = audit_commands.add_parser(
+        "cookies", help="recent cookie observations"
+    )
+    audit_cookies.add_argument("--account")
+    audit_cookies.add_argument("--state")
+    audit_cookies.add_argument("--limit", type=int, default=30)
+    audit_cookies.add_argument("--json", action="store_true")
+
+    audit_stats = audit_commands.add_parser(
+        "cookie-stats", help="measured cookie lifetime per account"
+    )
+    audit_stats.add_argument("--account")
+    audit_stats.add_argument("--json", action="store_true")
+
+    audit_commands.add_parser(
+        "schema", help="print the DDL that creates the audit tables"
+    )
     return parser
 
 
@@ -304,6 +334,8 @@ def main(
             and arguments.acceptance_command == "status"
         ):
             return _acceptance_status_command(arguments, resolved_settings, output)
+        if arguments.command == "audit":
+            return _audit_command(arguments, resolved_settings, output)
         if arguments.command == "scripts" and arguments.scripts_command == "list":
             return _list_scripts(arguments, output)
         if arguments.command == "scripts" and arguments.scripts_command == "run":
@@ -549,6 +581,152 @@ def _acceptance_status_command(
     return EXIT_SUCCESS if report.is_complete else EXIT_ACCEPTANCE_INCOMPLETE
 
 
+def _format_seconds(value: object) -> str:
+    """Human-readable duration for the audit tables (None -> '-')."""
+
+    if value is None:
+        return "-"
+    try:
+        seconds = int(value)
+    except (TypeError, ValueError):
+        return "-"
+    if seconds >= 86400:
+        return f"{seconds / 86400:.1f}d"
+    if seconds >= 3600:
+        return f"{seconds / 3600:.1f}h"
+    if seconds >= 60:
+        return f"{seconds / 60:.0f}m"
+    return f"{seconds}s"
+
+
+def _format_timestamp(value: object) -> str:
+    if value is None:
+        return "-"
+    text = value.isoformat() if hasattr(value, "isoformat") else str(value)
+    return text.replace("T", " ")[:19]
+
+
+def _audit_command(
+    arguments: argparse.Namespace, settings: Settings, output: TextIO
+) -> int:
+    from gyxx_flow.ops.audit import SCHEMA_SQL, AuditStore
+
+    command = arguments.audit_command
+    if command == "schema":
+        output.write(SCHEMA_SQL.strip() + "\n")
+        return EXIT_SUCCESS
+
+    store = AuditStore(settings.data_root)
+    if not store.enabled:
+        output.write("审计已关闭（GYXX_OPS_AUDIT=off）\n")
+        return EXIT_CONFIGURATION
+
+    if command == "runs":
+        rows = store.query_runs(
+            workflow_id=arguments.workflow,
+            status=arguments.status,
+            business_date=arguments.business_date,
+            limit=arguments.limit,
+        )
+    elif command == "cookies":
+        rows = store.query_cookie_events(
+            account_id=arguments.account,
+            state=arguments.state,
+            limit=arguments.limit,
+        )
+    elif command == "cookie-stats":
+        rows = (
+            store.query_cookie_lifetimes(account_id=arguments.account)
+            if arguments.account
+            else store.query_cookie_stats()
+        )
+    else:  # pragma: no cover - argparse 已限制取值
+        raise CliConfigurationError(f"unsupported audit command: {command}")
+
+    if store.last_error and not rows:
+        output.write(f"审计库不可用：{store.last_error}\n")
+        return EXIT_CONFIGURATION
+
+    if getattr(arguments, "json", False):
+        _write_json(output, {"rows": rows})
+        return EXIT_SUCCESS
+
+    if not rows:
+        output.write("暂无审计记录\n")
+        return EXIT_SUCCESS
+
+    if command == "runs":
+        output.write(
+            f"{'started':19s}  {'workflow':40s} {'date':10s} {'mode':7s} "
+            f"{'status':8s} {'dur':>6s}  steps\n"
+        )
+        for row in rows:
+            steps = (
+                f"{row.get('step_success', 0)}/{row.get('step_total', 0)}"
+                f" failed={row.get('step_failed', 0)}"
+            )
+            output.write(
+                f"{_format_timestamp(row.get('started_at')):19s}  "
+                f"{str(row.get('workflow_id', ''))[:40]:40s} "
+                f"{_format_timestamp(row.get('business_date'))[:10]:10s} "
+                f"{str(row.get('mode', '')):7s} "
+                f"{str(row.get('status', '')):8s} "
+                f"{_format_seconds(row.get('duration_seconds')):>6s}  {steps}\n"
+            )
+            if row.get("error"):
+                output.write(f"{'':21s}error: {str(row['error'])[:160]}\n")
+        return EXIT_SUCCESS
+
+    if command == "cookies":
+        output.write(
+            f"{'observed':19s}  {'account':24s} {'state':14s} {'probe':14s} "
+            f"{'fingerprint':12s}  expires_at\n"
+        )
+        for row in rows:
+            output.write(
+                f"{_format_timestamp(row.get('observed_at')):19s}  "
+                f"{str(row.get('account_id', ''))[:24]:24s} "
+                f"{str(row.get('state', '')):14s} "
+                f"{str(row.get('probe_status', '')):14s} "
+                f"{str(row.get('fingerprint', ''))[:12]:12s}  "
+                f"{_format_timestamp(row.get('declared_expires_at'))}\n"
+            )
+        return EXIT_SUCCESS
+
+    # cookie-stats / 单账号 lifetime
+    if arguments.account:
+        output.write(
+            f"{'fingerprint':12s} {'first_seen':19s} {'last_healthy':19s} "
+            f"{'failed_at':19s} {'lifetime':>9s} {'declared':19s}\n"
+        )
+        for row in rows:
+            output.write(
+                f"{str(row.get('fingerprint', ''))[:12]:12s} "
+                f"{_format_timestamp(row.get('first_seen_at')):19s} "
+                f"{_format_timestamp(row.get('last_healthy_at')):19s} "
+                f"{_format_timestamp(row.get('first_failed_at')):19s} "
+                f"{_format_seconds(row.get('lifetime_seconds')):>9s} "
+                f"{_format_timestamp(row.get('declared_expires_at')):19s}\n"
+            )
+        return EXIT_SUCCESS
+
+    output.write(
+        f"{'account':24s} {'sessions':>8s} {'expired':>7s} {'avg':>7s} "
+        f"{'median':>7s} {'min':>7s} {'max':>7s}  last_failed\n"
+    )
+    for row in rows:
+        output.write(
+            f"{str(row.get('account_id', ''))[:24]:24s} "
+            f"{row.get('sessions', 0):>8d} {row.get('expired_sessions', 0):>7d} "
+            f"{_format_seconds(row.get('avg_lifetime_seconds')):>7s} "
+            f"{_format_seconds(row.get('median_lifetime_seconds')):>7s} "
+            f"{_format_seconds(row.get('min_lifetime_seconds')):>7s} "
+            f"{_format_seconds(row.get('max_lifetime_seconds')):>7s}  "
+            f"{_format_timestamp(row.get('last_failed_at'))}\n"
+        )
+    return EXIT_SUCCESS
+
+
 def _list_workflows(
     arguments: argparse.Namespace,
     registry: WorkflowRegistry,

src/gyxx_flow/cookie_gate.py

@@ -147,6 +147,7 @@ class CookieGate:
         self._now = now or (lambda: datetime.now(timezone.utc))
         self._sender = sender or send_lark_bot_message
         self._workflow_catalog: WorkflowCatalog | None = None
+        self._audit_store: Any | None = None
 
     # ------------------------------------------------------------------
     # Account resolution
@@ -236,7 +237,7 @@ class CookieGate:
         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")
+            self._write_state(account, STATE_HEALTHY, probe="fresh-keepalive", context=context)
             return GateVerdict(
                 account_id=account_id,
                 mode=mode,
@@ -269,6 +270,7 @@ class CookieGate:
                 STATE_HEALTHY,
                 probe="success",
                 recovered=recovered,
+                context=context,
             )
             message = (
                 f"账号 {account_id} 登录态有效（自动账密重登成功）"
@@ -367,6 +369,20 @@ class CookieGate:
                 state.get("pending_reruns"), context, _now_iso(self._now)
             )
         self._save_state(account, state)
+        try:
+            keepalive_status = _read_keepalive_status(account) or {}
+        except Exception:  # noqa: BLE001 - state file is advisory
+            keepalive_status = {}
+        self._audit_cookie(
+            account,
+            "not_logged_in"
+            if str(keepalive_status.get("status", "")) == "not_logged_in"
+            else "expired",
+            probe_status=str(keepalive_status.get("status", "expired")),
+            probe_source="gate",
+            detail=detail,
+            context=context,
+        )
         message = _expiry_message(account_id, mode, context, detail)
         if notified:
             message += " 已通过飞书通知登录负责人。"
@@ -387,7 +403,17 @@ class CookieGate:
         *,
         probe: str,
         recovered: bool = False,
+        context: RunContext | None = None,
     ) -> None:
+        if state_name == STATE_HEALTHY:
+            # 健康心跳要落在每一次判定上，才能算准「最后一次可用」的时间。
+            self._audit_cookie(
+                account,
+                "healthy",
+                probe_status=probe,
+                probe_source="gate",
+                context=context,
+            )
         state = self._read_state(account)
         if (
             state_name == STATE_HEALTHY
@@ -431,6 +457,64 @@ class CookieGate:
         except OSError:  # pragma: no cover - state file is advisory only
             pass
 
+    # ------------------------------------------------------------------
+    # Cookie telemetry (best-effort; never blocks or fails the gate)
+    # ------------------------------------------------------------------
+
+    def _audit(self) -> Any | None:
+        if self._audit_store is None:
+            try:
+                from gyxx_flow.ops.audit import AuditStore
+
+                self._audit_store = AuditStore(
+                    self.settings.data_root,
+                    environment=self._environment,
+                )
+            except Exception:  # noqa: BLE001 - telemetry is optional
+                return None
+        return self._audit_store
+
+    def _audit_cookie(
+        self,
+        account: Any,
+        state: str,
+        *,
+        probe_status: str,
+        probe_source: str,
+        detail: str = "",
+        context: RunContext | None = None,
+    ) -> None:
+        """Append one cookie observation.
+
+        Stores a value-free fingerprint plus the earliest declared expiry of the
+        required cookies, so failure patterns (how long a login really lasts,
+        and whether it dies before or after the declared expiry) can be derived
+        from history instead of guessed.  Errors are swallowed: gating must
+        never depend on the audit database.
+        """
+
+        try:
+            store = self._audit()
+            if store is None or not store.enabled:
+                return
+            store.record_cookie_event(
+                account_id=str(getattr(account, "account_id", "") or ""),
+                state=state,
+                snapshot=self._cookie_snapshot(account),
+                probe_status=probe_status,
+                probe_source=probe_source,
+                detail=detail,
+                workflow_id=str(getattr(context, "workflow_id", "") or ""),
+                run_id=str(getattr(context, "run_id", "") or ""),
+            )
+        except Exception:  # noqa: BLE001 - telemetry is advisory only
+            return
+
+    def _cookie_snapshot(self, account: Any) -> Any | None:
+        from gyxx_flow.ops.audit import snapshot_for_account
+
+        return snapshot_for_account(account)
+
     def _within_notify_interval(self, state: Mapping[str, object]) -> bool:
         raw = state.get("last_notified_at")
         if not isinstance(raw, str):

src/gyxx_flow/ops/AUDIT.md

@@ -0,0 +1,117 @@
+# 运维审计库（ops_* 表）
+
+工作流执行历史与登录 cookie 遥测的 PostgreSQL 落库。本地 `var/runs/**/run.json`
+仍是权威源，这里做的是**跨机器可查、可长期留存的关系型镜像**。
+
+## 为什么需要
+
+- 本地 journal 只在部署机磁盘上：换机、扩容、清理后历史就断了。
+- `var/logs` 是文本，做不了「近 30 天哪些工作流失败最多」这类查询。
+- cookie 状态原先只存在 `keepalive_status.json`（只有最新值），无法回答
+  「登录态到底能活几天」。
+
+## 表与视图
+
+| 对象 | 粒度 | 用途 |
+|---|---|---|
+| `ops_workflow_runs` | 一次运行一行 | 状态、模式、耗时、步骤计数、错误、journal 路径、主机 |
+| `ops_workflow_run_steps` | 一次尝试一行 | 含重试；`exit_code`、错误、耗时 |
+| `ops_account_cookie_events` | 一次观测一行 | 事件流；只存指纹，不存 cookie 明文 |
+| `ops_account_cookie_lifetimes`（视图） | 一份 cookie 一行 | 从首次出现到失效的寿命 |
+| `ops_account_cookie_stats`（视图） | 一个账号一行 | 平均 / 中位 / 最短 / 最长寿命 |
+
+## 自动建表
+
+首次写入时执行 `SCHEMA_SQL`（幂等 DDL），无需手工迁移。也可以用：
+
+```bash
+uv run gyxx audit schema > 001_ops_audit.sql   # 导出 DDL 交 DBA 预建
+psql "$GYXX_POSTGRES_DSN" -f 001_ops_audit.sql
+```
+
+## 查询
+
+```bash
+uv run gyxx audit runs --workflow content.metrics.self_operated.daily --limit 20
+uv run gyxx audit runs --status failed --limit 50
+uv run gyxx audit cookies --account douyin-shop --limit 50
+uv run gyxx audit cookie-stats              # 各账号失效规律
+uv run gyxx audit cookie-stats --account douyin-shop   # 单账号每次的寿命明细
+```
+
+都支持 `--json`，方便接到面板或告警脚本。
+
+直接 SQL 也一样：
+
+```sql
+-- 最近 7 天失败最多的 10 个工作流
+SELECT workflow_id, COUNT(*) AS failures
+FROM ops_workflow_runs
+WHERE status = 'failed' AND started_at > NOW() - INTERVAL '7 days'
+GROUP BY workflow_id ORDER BY failures DESC LIMIT 10;
+
+-- 每次失败卡在哪个步骤
+SELECT r.workflow_id, s.step_id, COUNT(*) AS n
+FROM ops_workflow_run_steps s
+JOIN ops_workflow_runs r USING (run_id)
+WHERE s.status = 'failed' AND r.started_at > NOW() - INTERVAL '7 days'
+GROUP BY 1, 2 ORDER BY n DESC;
+
+-- 哪个账号的登录态最短命（预警阈值参考）
+SELECT account_id, sessions, expired_sessions,
+       avg_lifetime_seconds / 86400.0 AS avg_days,
+       min_lifetime_seconds / 86400.0 AS min_days,
+       last_failed_at
+FROM ops_account_cookie_stats ORDER BY min_lifetime_seconds NULLS LAST;
+```
+
+## 怎么读「失效规律」
+
+一次 cookie 代由 `fingerprint` 标识（需求域名/名称/值哈希排序后取摘要，与文件顺序无关）：
+
+- `first_healthy_at` → `first_failed_at` = `lifetime_seconds`：**实际可用时长**
+- `healthy_to_failure_seconds`：最后一次探活成功到发现失效，反映**探测精度**
+- `declared_expires_at`：cookie 自带的最早过期时间
+
+把 `lifetime_seconds` 和 `declared_expires_at` 对比，能区分三种情况：
+
+1. 实际寿命接近声明寿命 → 平台按声明失效，可按 `declared_expires_at` 提前预警；
+2. 实际远短于声明 → 服务端提前吊销（常见于多端登录、风控）；
+3. `declared_expires_at` 为空（session cookie）→ 只能靠观测，`min_lifetime_seconds`
+   就是安全刷新周期的上界。
+
+`fingerprint` 变化表示换了一份 cookie（重新登录），不算失效，也不会把两段寿命混在一起。
+
+## 采集点
+
+| 来源 | 触发 | `probe_source` |
+|---|---|---|
+| `CookieGate.check` | 每次工作流运行前的门禁判定 | `gate` |
+| `AccountKeepaliveManager` | 常驻调度器每 `keepalive_interval_seconds`（默认 600s） | `keepalive` |
+| `RunIndex.index_journal` | 每次运行开始/结束/异常 | — |
+
+keepalive 是主要的精度来源：默认 10 分钟一次，同类状态一小时内只落一行，
+状态翻转立即落行，所以「首次失效时间」误差不超过一个探活周期。
+
+## 容错
+
+审计**永不阻断业务**：
+
+- 数据库不可达 / 缺 DSN / 建表失败 → 写入返回 `False`，调用方继续执行；
+- 连续失败 3 次后熔断冷却（默认 300s），避免每个工作流都卡在连接超时；
+- 失败记录写到 `<GYXX_DATA_ROOT>/logs/ops-audit/audit-failures.log`。
+
+## 开关
+
+| 变量 | 默认 | 说明 |
+|---|---|---|
+| `GYXX_OPS_AUDIT` | 开 | 设 `off` 关闭全部审计写入 |
+| `GYXX_OPS_AUDIT_HOST` | 主机名 | 多机部署时写入手工标识 |
+| `GYXX_OPS_AUDIT_FAILURE_COOLDOWN` | 300 | 熔断冷却秒数 |
+
+## 不存什么
+
+- **cookie 值**：只存哈希指纹；排查具体 cookie 请走 vault 文件。
+- **日志正文**：`ops_workflow_runs` 只存 `journal_path` 指向本地文件；
+  完整堆栈仍在 `var/logs/console-restarts/`。
+- **凭据**：DSN 只从 `GYXX_POSTGRES_DSN`（或 `PG_*`）读，不落表。

src/gyxx_flow/ops/__init__.py

@@ -1,5 +1,6 @@
-"""Operational run indexing and durable outbox services."""
+"""Operational run indexing, audit telemetry and durable outbox services."""
 
+from .audit import AuditStore, CookieSnapshot, summarize_cookies
 from .effects import (
     EffectAlreadyApplied,
     EffectClaim,
@@ -10,6 +11,8 @@ from .index import RunIndex, RunRecord
 from .outbox import IdempotencyConflict, Outbox, OutboxMessage
 
 __all__ = [
+    "AuditStore",
+    "CookieSnapshot",
     "IdempotencyConflict",
     "EffectAlreadyApplied",
     "EffectClaim",
@@ -19,4 +22,5 @@ __all__ = [
     "OutboxMessage",
     "RunIndex",
     "RunRecord",
+    "summarize_cookies",
 ]

src/gyxx_flow/ops/audit.py

@@ -0,0 +1,901 @@
+"""PostgreSQL-backed operational audit: workflow run history and cookie telemetry.
+
+设计原则
+--------
+1. **可观测性不拖累业务**：审计写入永远不向调用方抛异常。数据库不可达时
+   只记录一条 failure 日志并熔断冷却，工作流照常执行。
+2. **本地 journal 仍是权威源**：本模块是镜像/汇总，供跨机器查询与长期留存；
+   `var/runs/**/run.json` 不会被替代。
+3. **不落明文凭据**：cookie 只存指纹（域名+名称+值哈希），值本身永不入库。
+
+环境开关
+--------
+``GYXX_OPS_AUDIT=off``                 关闭审计（默认开启）
+``GYXX_OPS_AUDIT_HOST``                覆盖主机标识（默认 socket.gethostname()）
+``GYXX_OPS_AUDIT_FAILURE_COOLDOWN``    熔断冷却秒数（默认 300）
+``GYXX_POSTGRES_DSN``                  与动态配置共用同一 DSN
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import socket
+from contextlib import contextmanager
+from dataclasses import dataclass
+from datetime import date, datetime, timezone
+from pathlib import Path
+from typing import Any, Iterator, Mapping, Sequence
+
+SCHEMA_VERSION = 1
+
+OPS_AUDIT_ENV = "GYXX_OPS_AUDIT"
+OPS_AUDIT_HOST_ENV = "GYXX_OPS_AUDIT_HOST"
+OPS_AUDIT_COOLDOWN_ENV = "GYXX_OPS_AUDIT_FAILURE_COOLDOWN"
+
+_DEFAULT_COOLDOWN_SECONDS = 300
+_MAX_CONSECUTIVE_FAILURES = 3
+_MAX_TEXT_CHARS = 4000
+_SCHEMA_LOCK_KEY = "gyxx:ops-audit:migration"
+
+HEALTHY_STATES = frozenset({"healthy"})
+FAILED_STATES = frozenset({"expired", "not_logged_in"})
+_COOKIE_STATES = frozenset(
+    {"healthy", "expired", "not_logged_in", "probe_error", "busy", "disabled"}
+)
+
+SCHEMA_SQL = """
+CREATE TABLE IF NOT EXISTS ops_workflow_runs (
+    run_id            TEXT PRIMARY KEY,
+    workflow_id       TEXT NOT NULL,
+    business_date     DATE NOT NULL,
+    mode              TEXT NOT NULL DEFAULT 'unknown',
+    shadow            BOOLEAN NOT NULL DEFAULT FALSE,
+    status            TEXT NOT NULL,
+    started_at        TIMESTAMPTZ,
+    ended_at          TIMESTAMPTZ,
+    duration_seconds  INTEGER,
+    error             TEXT,
+    step_total        INTEGER NOT NULL DEFAULT 0,
+    step_success      INTEGER NOT NULL DEFAULT 0,
+    step_failed       INTEGER NOT NULL DEFAULT 0,
+    step_skipped      INTEGER NOT NULL DEFAULT 0,
+    step_running      INTEGER NOT NULL DEFAULT 0,
+    journal_path      TEXT NOT NULL DEFAULT '',
+    host              TEXT NOT NULL DEFAULT '',
+    indexed_at        TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS ops_workflow_runs_workflow_idx
+    ON ops_workflow_runs (workflow_id, started_at DESC);
+CREATE INDEX IF NOT EXISTS ops_workflow_runs_status_idx
+    ON ops_workflow_runs (status, started_at DESC);
+CREATE INDEX IF NOT EXISTS ops_workflow_runs_business_date_idx
+    ON ops_workflow_runs (business_date, workflow_id);
+
+CREATE TABLE IF NOT EXISTS ops_workflow_run_steps (
+    run_id           TEXT NOT NULL
+                     REFERENCES ops_workflow_runs (run_id) ON DELETE CASCADE,
+    step_id          TEXT NOT NULL,
+    attempt          INTEGER NOT NULL DEFAULT 1,
+    status           TEXT NOT NULL DEFAULT '',
+    exit_code        INTEGER,
+    started_at       TIMESTAMPTZ,
+    ended_at         TIMESTAMPTZ,
+    duration_seconds INTEGER,
+    error            TEXT,
+    PRIMARY KEY (run_id, step_id, attempt)
+);
+
+CREATE INDEX IF NOT EXISTS ops_workflow_run_steps_step_idx
+    ON ops_workflow_run_steps (step_id, status);
+
+CREATE TABLE IF NOT EXISTS ops_account_cookie_events (
+    id                  BIGSERIAL PRIMARY KEY,
+    account_id          TEXT NOT NULL,
+    observed_at         TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    state               TEXT NOT NULL,
+    probe_status        TEXT NOT NULL DEFAULT '',
+    probe_source        TEXT NOT NULL DEFAULT '',
+    cookie_count        INTEGER,
+    required_count      INTEGER,
+    fingerprint         TEXT NOT NULL DEFAULT '',
+    declared_expires_at TIMESTAMPTZ,
+    detail              TEXT NOT NULL DEFAULT '',
+    workflow_id         TEXT NOT NULL DEFAULT '',
+    run_id              TEXT NOT NULL DEFAULT '',
+    host                TEXT NOT NULL DEFAULT ''
+);
+
+CREATE INDEX IF NOT EXISTS ops_account_cookie_events_account_idx
+    ON ops_account_cookie_events (account_id, observed_at DESC);
+CREATE INDEX IF NOT EXISTS ops_account_cookie_events_state_idx
+    ON ops_account_cookie_events (state, observed_at DESC);
+CREATE INDEX IF NOT EXISTS ops_account_cookie_events_fingerprint_idx
+    ON ops_account_cookie_events (account_id, fingerprint);
+
+-- 一段「cookie 代」的生命周期：同一 fingerprint 从首次出现到失效。
+-- first_failed_at 只取最后一次 healthy 之后的失败，避免恢复后再次失败时口径前移。
+CREATE OR REPLACE VIEW ops_account_cookie_lifetimes AS
+WITH per_fingerprint AS (
+    SELECT
+        account_id,
+        fingerprint,
+        MIN(observed_at) AS first_seen_at,
+        MIN(observed_at) FILTER (WHERE state = 'healthy') AS first_healthy_at,
+        MAX(observed_at) FILTER (WHERE state = 'healthy') AS last_healthy_at,
+        MAX(declared_expires_at) AS declared_expires_at
+    FROM ops_account_cookie_events
+    WHERE fingerprint <> ''
+    GROUP BY account_id, fingerprint
+)
+SELECT
+    p.account_id,
+    p.fingerprint,
+    p.first_seen_at,
+    p.first_healthy_at,
+    p.last_healthy_at,
+    p.declared_expires_at,
+    f.first_failed_at,
+    (f.first_failed_at IS NOT NULL) AS expired,
+    CASE
+        WHEN p.first_healthy_at IS NOT NULL AND p.last_healthy_at IS NOT NULL
+        THEN EXTRACT(EPOCH FROM (p.last_healthy_at - p.first_healthy_at))::BIGINT
+    END AS healthy_span_seconds,
+    CASE
+        WHEN p.first_healthy_at IS NOT NULL AND f.first_failed_at IS NOT NULL
+        THEN EXTRACT(EPOCH FROM (f.first_failed_at - p.first_healthy_at))::BIGINT
+    END AS lifetime_seconds,
+    CASE
+        WHEN p.last_healthy_at IS NOT NULL AND f.first_failed_at IS NOT NULL
+        THEN EXTRACT(EPOCH FROM (f.first_failed_at - p.last_healthy_at))::BIGINT
+    END AS healthy_to_failure_seconds
+FROM per_fingerprint p
+LEFT JOIN LATERAL (
+    SELECT MIN(e.observed_at) AS first_failed_at
+    FROM ops_account_cookie_events e
+    WHERE e.account_id = p.account_id
+      AND e.fingerprint = p.fingerprint
+      AND e.state IN ('expired', 'not_logged_in')
+      AND (p.last_healthy_at IS NULL OR e.observed_at > p.last_healthy_at)
+) f ON TRUE;
+
+-- 按账号汇总的失效规律：平均/中位/最短/最长寿命，直接支撑「多久失效一次」。
+CREATE OR REPLACE VIEW ops_account_cookie_stats AS
+SELECT
+    account_id,
+    COUNT(*) AS sessions,
+    COUNT(*) FILTER (WHERE expired) AS expired_sessions,
+    AVG(lifetime_seconds)::BIGINT AS avg_lifetime_seconds,
+    PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY lifetime_seconds)::BIGINT
+        AS median_lifetime_seconds,
+    MIN(lifetime_seconds) AS min_lifetime_seconds,
+    MAX(lifetime_seconds) AS max_lifetime_seconds,
+    MAX(first_failed_at) AS last_failed_at,
+    MAX(last_healthy_at) AS last_healthy_at
+FROM ops_account_cookie_lifetimes
+WHERE lifetime_seconds IS NOT NULL
+GROUP BY account_id;
+
+CREATE TABLE IF NOT EXISTS ops_audit_schema (
+    schema_version INTEGER PRIMARY KEY,
+    applied_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+"""
+
+
+# ----------------------------------------------------------------------
+# 纯函数：cookie 快照
+# ----------------------------------------------------------------------
+
+
+@dataclass(frozen=True, slots=True)
+class CookieSnapshot:
+    """一次静态检查得到的 cookie 画像，不含任何明文值。"""
+
+    cookie_count: int
+    required_count: int
+    fingerprint: str
+    declared_expires_at: datetime | None
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "cookie_count": self.cookie_count,
+            "required_count": self.required_count,
+            "fingerprint": self.fingerprint,
+            "declared_expires_at": (
+                self.declared_expires_at.isoformat()
+                if self.declared_expires_at is not None
+                else None
+            ),
+        }
+
+
+def _domain_matches(domain: str, required_domains: Sequence[str]) -> bool:
+    normalized = str(domain or "").lstrip(".").casefold()
+    if not normalized:
+        return False
+    return any(
+        normalized == required.casefold() or normalized.endswith("." + required.casefold())
+        for required in required_domains
+    )
+
+
+def _expires_at(cookie: Mapping[str, Any]) -> datetime | None:
+    raw = cookie.get("expires")
+    if raw is None or isinstance(raw, bool):
+        return None
+    try:
+        value = float(raw)
+    except (TypeError, ValueError):
+        return None
+    # Playwright 用 -1 表示 session cookie；0 与负数都视为「不声明」。
+    if value <= 0:
+        return None
+    try:
+        return datetime.fromtimestamp(value, tz=timezone.utc)
+    except (OverflowError, OSError, ValueError):
+        return None
+
+
+def summarize_cookies(
+    cookies: Sequence[Mapping[str, Any]],
+    *,
+    required_domains: Sequence[str] = (),
+    required_names: Sequence[str] = (),
+) -> CookieSnapshot:
+    """Build a value-free snapshot: counts, fingerprint and earliest declared expiry.
+
+    指纹只看「必需 cookie」的域名/名称/值哈希，因此登录态真正换了一份 cookie
+    时指纹会变，而无关 cookie 的增删不会造成抖动。
+    """
+
+    required_domains = tuple(item for item in required_domains if item)
+    required_names = tuple(item for item in required_names if item)
+    components: list[str] = []
+    required_count = 0
+    earliest: datetime | None = None
+
+    for cookie in cookies:
+        if not isinstance(cookie, Mapping):
+            continue
+        name = str(cookie.get("name", ""))
+        domain = str(cookie.get("domain", ""))
+        is_required = bool(required_domains) and _domain_matches(domain, required_domains)
+        if required_names:
+            is_required = is_required and name in required_names
+        elif not required_domains:
+            is_required = True
+        if not is_required:
+            continue
+        required_count += 1
+        value = str(cookie.get("value", ""))
+        value_hash = hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest()[:16]
+        components.append(f"{domain.lstrip('.')}|{name}|{value_hash}")
+        expiry = _expires_at(cookie)
+        if expiry is not None and (earliest is None or expiry < earliest):
+            earliest = expiry
+
+    # 浏览器写入 cookie 的顺序并不稳定；排序后哈希才能保证同一份登录态得到同一指纹。
+    digest = hashlib.sha256()
+    for component in sorted(components):
+        digest.update(f"{component}\n".encode("utf-8"))
+
+    return CookieSnapshot(
+        cookie_count=len(cookies),
+        required_count=required_count,
+        fingerprint=digest.hexdigest()[:32] if required_count else "",
+        declared_expires_at=earliest,
+    )
+
+
+def snapshot_for_account(account: Any) -> CookieSnapshot | None:
+    """Read an account's cookie vault and summarise it, or ``None`` when unreadable.
+
+    Shared by the gate and the keepalive manager so both record the exact same
+    fingerprint for the same vault contents.
+    """
+
+    cookie_file = getattr(account, "cookie_file", None)
+    if cookie_file is None:
+        return None
+    try:
+        from gyxx_flow.adapters.browser import BrowserCookieStore
+
+        store = BrowserCookieStore(
+            cookie_file,
+            getattr(account, "storage_state_file", cookie_file),
+        )
+        cookies = store.load_cookies()
+    except Exception:  # noqa: BLE001 - a missing/corrupt vault is itself a signal
+        return None
+    return summarize_cookies(
+        cookies,
+        required_domains=tuple(getattr(account, "required_cookie_domains", ()) or ()),
+        required_names=tuple(getattr(account, "required_cookie_names", ()) or ()),
+    )
+
+
+# ----------------------------------------------------------------------
+# 时间与文本工具
+# ----------------------------------------------------------------------
+
+
+def _parse_datetime(value: Any) -> datetime | None:
+    if isinstance(value, datetime):
+        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
+    if not isinstance(value, str) or not value.strip():
+        return None
+    text = value.strip().replace("Z", "+00:00")
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+
+
+def _parse_date(value: Any) -> date | None:
+    if isinstance(value, date) and not isinstance(value, datetime):
+        return value
+    if not isinstance(value, str):
+        return None
+    try:
+        return date.fromisoformat(value)
+    except ValueError:
+        return None
+
+
+def _seconds_between(start: datetime | None, end: datetime | None) -> int | None:
+    if start is None or end is None:
+        return None
+    delta = (end - start).total_seconds()
+    return max(0, int(delta))
+
+
+def _clip(value: Any, limit: int = _MAX_TEXT_CHARS) -> str:
+    if value is None:
+        return ""
+    text = value if isinstance(value, str) else str(value)
+    return text[:limit]
+
+
+def split_step_key(key: str) -> tuple[str, int]:
+    """``sync.attempt-2`` -> ``("sync", 2)``; 无法解析时原样当作 step_id。"""
+
+    if ".attempt-" in key:
+        name, _, raw = key.rpartition(".attempt-")
+        try:
+            attempt = int(raw)
+        except ValueError:
+            return key, 1
+        if name and attempt > 0:
+            return name, attempt
+    return key, 1
+
+
+# ----------------------------------------------------------------------
+# 审计存储
+# ----------------------------------------------------------------------
+
+
+class AuditStore:
+    """Idempotent writer for the ``ops_*`` tables.
+
+    所有公开写入方法都返回 bool，并且**永不抛出**异常；数据库问题只会导致本次
+    审计丢失，不会影响工作流执行。
+    """
+
+    def __init__(
+        self,
+        data_root: Path | str,
+        *,
+        environment: Mapping[str, str] | None = None,
+        connect: Any | None = None,
+        clock: Any | None = None,
+    ) -> None:
+        self._data_root = Path(data_root).expanduser().resolve()
+        self._environment = dict(os.environ if environment is None else environment)
+        self._connect_impl = connect
+        self._clock = clock or (lambda: datetime.now(timezone.utc))
+        self._schema_ready = False
+        self._consecutive_failures = 0
+        self._cooldown_until = 0.0
+        self._last_error = ""
+        self._failure_log = self._data_root / "logs" / "ops-audit" / "audit-failures.log"
+
+    # -- 状态 ---------------------------------------------------------
+
+    @property
+    def enabled(self) -> bool:
+        raw = str(self._environment.get(OPS_AUDIT_ENV, "")).strip().lower()
+        return raw not in {"0", "off", "false", "no", "disabled"}
+
+    @property
+    def last_error(self) -> str:
+        return self._last_error
+
+    def _host(self) -> str:
+        override = str(self._environment.get(OPS_AUDIT_HOST_ENV, "")).strip()
+        if override:
+            return override
+        try:
+            return socket.gethostname()[:128]
+        except OSError:  # pragma: no cover - 极端环境
+            return ""
+
+    def _cooldown_seconds(self) -> int:
+        raw = str(self._environment.get(OPS_AUDIT_COOLDOWN_ENV, "")).strip()
+        try:
+            return max(0, int(raw))
+        except ValueError:
+            return _DEFAULT_COOLDOWN_SECONDS
+
+    def _in_cooldown(self) -> bool:
+        import time
+
+        return time.monotonic() < self._cooldown_until
+
+    # -- 连接 ---------------------------------------------------------
+
+    @contextmanager
+    def _connection(self) -> Iterator[Any]:
+        if self._connect_impl is not None:
+            conn = self._connect_impl()
+            try:
+                yield conn
+            finally:
+                close = getattr(conn, "close", None)
+                if callable(close):
+                    close()
+            return
+
+        import psycopg  # 延迟导入：未安装 psycopg 时整个模块仍可导入
+        from psycopg.rows import dict_row
+
+        dsn, kwargs = database_kwargs(self._environment)
+        if dsn:
+            conn = psycopg.connect(dsn, row_factory=dict_row)
+        else:
+            conn = psycopg.connect(**kwargs, row_factory=dict_row)
+        try:
+            yield conn
+        finally:
+            conn.close()
+
+    def _ensure_schema(self, conn: Any) -> None:
+        """Create the audit objects once, serialising the first DDL pass.
+
+        Several workflows can start at the same moment; running CREATE/COMMENT
+        concurrently lets sessions take DDL locks in different orders and can
+        deadlock.  The advisory lock is only taken when the schema is missing,
+        so steady-state processes skip the round trip entirely.
+        """
+
+        if self._schema_ready:
+            return
+        with conn.cursor() as cursor:
+            cursor.execute(
+                "SELECT 1 FROM pg_class "
+                "WHERE relname = 'ops_audit_schema' AND relkind = 'r'"
+            )
+            initialised = cursor.fetchone() is not None
+        if not initialised:
+            with conn.cursor() as cursor:
+                cursor.execute(
+                    "SELECT pg_advisory_xact_lock(hashtext(%s))",
+                    (_SCHEMA_LOCK_KEY,),
+                )
+                cursor.execute(SCHEMA_SQL)
+                cursor.execute(
+                    "INSERT INTO ops_audit_schema (schema_version) VALUES (%s) "
+                    "ON CONFLICT (schema_version) DO NOTHING",
+                    (SCHEMA_VERSION,),
+                )
+            conn.commit()
+        self._schema_ready = True
+
+    # -- 失败处理 -----------------------------------------------------
+
+    def _record_failure(self, operation: str, exc: BaseException) -> None:
+        import time
+
+        self._consecutive_failures += 1
+        self._last_error = f"{type(exc).__name__}: {exc}"[:300]
+        if self._consecutive_failures >= _MAX_CONSECUTIVE_FAILURES:
+            self._cooldown_until = time.monotonic() + self._cooldown_seconds()
+        try:
+            self._failure_log.parent.mkdir(parents=True, exist_ok=True)
+            payload = {
+                "at": self._clock().isoformat(timespec="seconds"),
+                "operation": operation,
+                "error": self._last_error,
+                "consecutive_failures": self._consecutive_failures,
+            }
+            with self._failure_log.open("a", encoding="utf-8") as stream:
+                stream.write(json.dumps(payload, ensure_ascii=False) + "\n")
+        except OSError:
+            pass
+
+    def _run_guarded(self, operation: str, action: Any) -> bool:
+        if not self.enabled or self._in_cooldown():
+            return False
+        try:
+            with self._connection() as conn:
+                self._ensure_schema(conn)
+                action(conn)
+            self._consecutive_failures = 0
+            return True
+        except Exception as exc:  # noqa: BLE001 - 审计绝不影响业务
+            self._record_failure(operation, exc)
+            return False
+
+    # -- 运行记录 -----------------------------------------------------
+
+    def record_run(
+        self,
+        record: Mapping[str, Any],
+        steps: Mapping[str, Any] | None = None,
+    ) -> bool:
+        """Upsert one run and its steps from a journal payload or RunRecord dict."""
+
+        workflow_id = _clip(record.get("workflow_id"), 200)
+        run_id = _clip(record.get("run_id"), 200)
+        business_date = _parse_date(record.get("business_date"))
+        if not workflow_id or not run_id or business_date is None:
+            return False
+        status = _clip(record.get("status"), 32) or "unknown"
+        mode = _clip(record.get("mode"), 32) or "unknown"
+        started_at = _parse_datetime(record.get("started_at"))
+        ended_at = _parse_datetime(record.get("ended_at"))
+        step_rows = self._step_rows(run_id, steps)
+        counts = self._step_counts(steps, record)
+
+        def action(conn: Any) -> None:
+            with conn.cursor() as cursor:
+                cursor.execute(
+                    """
+                    INSERT INTO ops_workflow_runs (
+                        run_id, workflow_id, business_date, mode, shadow, status,
+                        started_at, ended_at, duration_seconds, error,
+                        step_total, step_success, step_failed, step_skipped,
+                        step_running, journal_path, host, indexed_at
+                    ) VALUES (
+                        %(run_id)s, %(workflow_id)s, %(business_date)s, %(mode)s,
+                        %(shadow)s, %(status)s, %(started_at)s, %(ended_at)s,
+                        %(duration_seconds)s, %(error)s, %(step_total)s,
+                        %(step_success)s, %(step_failed)s, %(step_skipped)s,
+                        %(step_running)s, %(journal_path)s, %(host)s, NOW()
+                    )
+                    ON CONFLICT (run_id) DO UPDATE SET
+                        status           = EXCLUDED.status,
+                        ended_at         = EXCLUDED.ended_at,
+                        duration_seconds = EXCLUDED.duration_seconds,
+                        error            = EXCLUDED.error,
+                        step_total       = EXCLUDED.step_total,
+                        step_success     = EXCLUDED.step_success,
+                        step_failed      = EXCLUDED.step_failed,
+                        step_skipped     = EXCLUDED.step_skipped,
+                        step_running     = EXCLUDED.step_running,
+                        journal_path     = EXCLUDED.journal_path,
+                        indexed_at       = NOW()
+                    """,
+                    {
+                        "run_id": run_id,
+                        "workflow_id": workflow_id,
+                        "business_date": business_date,
+                        "mode": mode,
+                        "shadow": bool(record.get("shadow", False)),
+                        "status": status,
+                        "started_at": started_at,
+                        "ended_at": ended_at,
+                        "duration_seconds": _seconds_between(started_at, ended_at),
+                        "error": _clip(record.get("error")),
+                        "step_total": counts["total"],
+                        "step_success": counts["success"],
+                        "step_failed": counts["failed"],
+                        "step_skipped": counts["skipped"],
+                        "step_running": counts["running"],
+                        "journal_path": _clip(record.get("journal_path"), 1000),
+                        "host": self._host(),
+                    },
+                )
+                if step_rows:
+                    cursor.executemany(
+                        """
+                        INSERT INTO ops_workflow_run_steps (
+                            run_id, step_id, attempt, status, exit_code,
+                            started_at, ended_at, duration_seconds, error
+                        ) VALUES (
+                            %(run_id)s, %(step_id)s, %(attempt)s, %(status)s,
+                            %(exit_code)s, %(started_at)s, %(ended_at)s,
+                            %(duration_seconds)s, %(error)s
+                        )
+                        ON CONFLICT (run_id, step_id, attempt) DO UPDATE SET
+                            status           = EXCLUDED.status,
+                            exit_code        = EXCLUDED.exit_code,
+                            started_at       = COALESCE(
+                                ops_workflow_run_steps.started_at,
+                                EXCLUDED.started_at
+                            ),
+                            ended_at         = EXCLUDED.ended_at,
+                            duration_seconds = EXCLUDED.duration_seconds,
+                            error            = EXCLUDED.error
+                        """,
+                        step_rows,
+                    )
+            conn.commit()
+
+        return self._run_guarded("record_run", action)
+
+    @staticmethod
+    def _step_rows(
+        run_id: str, steps: Mapping[str, Any] | None
+    ) -> list[dict[str, Any]]:
+        if not isinstance(steps, Mapping):
+            return []
+        rows: list[dict[str, Any]] = []
+        for key, raw in steps.items():
+            if not isinstance(raw, Mapping):
+                continue
+            step_id, attempt = split_step_key(str(key))
+            attempt_value = raw.get("attempt")
+            if isinstance(attempt_value, int) and not isinstance(attempt_value, bool):
+                attempt = max(1, attempt_value)
+            started_at = _parse_datetime(raw.get("started_at"))
+            ended_at = _parse_datetime(raw.get("ended_at"))
+            exit_code = raw.get("exit_code")
+            rows.append(
+                {
+                    "run_id": run_id,
+                    "step_id": _clip(step_id, 200),
+                    "attempt": attempt,
+                    "status": _clip(raw.get("status"), 32),
+                    "exit_code": exit_code if isinstance(exit_code, int) else None,
+                    "started_at": started_at,
+                    "ended_at": ended_at,
+                    "duration_seconds": _seconds_between(started_at, ended_at),
+                    "error": _clip(raw.get("error")),
+                }
+            )
+        return rows
+
+    @staticmethod
+    def _step_counts(
+        steps: Mapping[str, Any] | None, record: Mapping[str, Any]
+    ) -> dict[str, int]:
+        counts = {"total": 0, "success": 0, "failed": 0, "skipped": 0, "running": 0}
+        if isinstance(steps, Mapping) and steps:
+            for raw in steps.values():
+                if not isinstance(raw, Mapping):
+                    continue
+                status = str(raw.get("status", ""))
+                counts["total"] += 1
+                if status in counts:
+                    counts[status] += 1
+            return counts
+        # 回退到 RunRecord 的 step_counts 投影
+        projection = record.get("step_counts")
+        if isinstance(projection, Mapping):
+            for name in ("success", "failed", "skipped", "running"):
+                value = projection.get(name)
+                if isinstance(value, int) and not isinstance(value, bool):
+                    counts[name] = max(0, value)
+            counts["total"] = counts["success"] + counts["failed"] + counts["skipped"] + counts["running"]
+        return counts
+
+    # -- cookie 事件 ---------------------------------------------------
+
+    def record_cookie_event(
+        self,
+        *,
+        account_id: str,
+        state: str,
+        snapshot: CookieSnapshot | None = None,
+        probe_status: str = "",
+        probe_source: str = "",
+        detail: str = "",
+        workflow_id: str = "",
+        run_id: str = "",
+        observed_at: datetime | None = None,
+    ) -> bool:
+        """Append one observation. 状态必须是白名单值，避免脏数据污染统计。"""
+
+        account_id = _clip(account_id, 200)
+        if not account_id:
+            return False
+        state = str(state or "").strip()
+        if state not in _COOKIE_STATES:
+            raise ValueError(f"invalid cookie state: {state!r}")
+
+        def action(conn: Any) -> None:
+            with conn.cursor() as cursor:
+                cursor.execute(
+                    """
+                    INSERT INTO ops_account_cookie_events (
+                        account_id, observed_at, state, probe_status, probe_source,
+                        cookie_count, required_count, fingerprint, declared_expires_at,
+                        detail, workflow_id, run_id, host
+                    ) VALUES (
+                        %(account_id)s, %(observed_at)s, %(state)s, %(probe_status)s,
+                        %(probe_source)s, %(cookie_count)s, %(required_count)s,
+                        %(fingerprint)s, %(declared_expires_at)s, %(detail)s,
+                        %(workflow_id)s, %(run_id)s, %(host)s
+                    )
+                    """,
+                    {
+                        "account_id": account_id,
+                        "observed_at": observed_at or self._clock(),
+                        "state": state,
+                        "probe_status": _clip(probe_status, 64),
+                        "probe_source": _clip(probe_source, 64),
+                        "cookie_count": snapshot.cookie_count if snapshot else None,
+                        "required_count": snapshot.required_count if snapshot else None,
+                        "fingerprint": snapshot.fingerprint if snapshot else "",
+                        "declared_expires_at": (
+                            snapshot.declared_expires_at if snapshot else None
+                        ),
+                        "detail": _clip(detail),
+                        "workflow_id": _clip(workflow_id, 200),
+                        "run_id": _clip(run_id, 200),
+                        "host": self._host(),
+                    },
+                )
+            conn.commit()
+
+        return self._run_guarded("record_cookie_event", action)
+
+    # -- 查询（供 CLI） ------------------------------------------------
+
+    def query_runs(
+        self,
+        *,
+        workflow_id: str | None = None,
+        status: str | None = None,
+        business_date: str | None = None,
+        limit: int = 50,
+    ) -> list[dict[str, Any]]:
+        conditions: list[str] = []
+        params: dict[str, Any] = {"limit": max(1, min(int(limit), 500))}
+        if workflow_id:
+            conditions.append("workflow_id = %(workflow_id)s")
+            params["workflow_id"] = workflow_id
+        if status:
+            conditions.append("status = %(status)s")
+            params["status"] = status
+        if business_date:
+            parsed = _parse_date(business_date)
+            if parsed is not None:
+                conditions.append("business_date = %(business_date)s")
+                params["business_date"] = parsed
+        where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
+        sql = f"""
+            SELECT * FROM ops_workflow_runs
+            {where}
+            ORDER BY started_at DESC NULLS LAST, run_id DESC
+            LIMIT %(limit)s
+        """
+        return self._fetch(sql, params)
+
+    def query_steps(self, run_id: str) -> list[dict[str, Any]]:
+        return self._fetch(
+            """
+            SELECT * FROM ops_workflow_run_steps
+            WHERE run_id = %(run_id)s
+            ORDER BY step_id, attempt
+            """,
+            {"run_id": run_id},
+        )
+
+    def query_cookie_events(
+        self,
+        *,
+        account_id: str | None = None,
+        state: str | None = None,
+        limit: int = 50,
+    ) -> list[dict[str, Any]]:
+        conditions: list[str] = []
+        params: dict[str, Any] = {"limit": max(1, min(int(limit), 500))}
+        if account_id:
+            conditions.append("account_id = %(account_id)s")
+            params["account_id"] = account_id
+        if state:
+            conditions.append("state = %(state)s")
+            params["state"] = state
+        where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
+        sql = f"""
+            SELECT * FROM ops_account_cookie_events
+            {where}
+            ORDER BY observed_at DESC
+            LIMIT %(limit)s
+        """
+        return self._fetch(sql, params)
+
+    def query_cookie_lifetimes(self, *, account_id: str | None = None) -> list[dict[str, Any]]:
+        if account_id:
+            return self._fetch(
+                """
+                SELECT * FROM ops_account_cookie_lifetimes
+                WHERE account_id = %(account_id)s
+                ORDER BY first_seen_at DESC
+                """,
+                {"account_id": account_id},
+            )
+        return self._fetch(
+            "SELECT * FROM ops_account_cookie_lifetimes ORDER BY first_seen_at DESC",
+            {},
+        )
+
+    def query_cookie_stats(self) -> list[dict[str, Any]]:
+        return self._fetch(
+            "SELECT * FROM ops_account_cookie_stats ORDER BY account_id",
+            {},
+        )
+
+    def _fetch(self, sql: str, params: Mapping[str, Any]) -> list[dict[str, Any]]:
+        if not self.enabled or self._in_cooldown():
+            return []
+        try:
+            with self._connection() as conn:
+                self._ensure_schema(conn)
+                with conn.cursor() as cursor:
+                    cursor.execute(sql, params)
+                    rows = cursor.fetchall()
+            self._consecutive_failures = 0
+            return [dict(row) for row in rows]
+        except Exception as exc:  # noqa: BLE001
+            self._record_failure("query", exc)
+            return []
+
+
+# ----------------------------------------------------------------------
+# 连接参数（与 PostgresWorkflowConfigStore 同口径）
+# ----------------------------------------------------------------------
+
+
+def database_kwargs(environment: Mapping[str, str]) -> tuple[str | None, dict[str, Any]]:
+    values = _apply_runtime_policy(environment)
+    dsn = str(values.get("GYXX_POSTGRES_DSN", "")).strip()
+    if dsn:
+        return dsn, {}
+    required = ("PG_HOST", "PG_PORT", "PG_DB", "PG_USER", "PG_PASSWORD")
+    missing = [name for name in required if not str(values.get(name, "")).strip()]
+    if missing:
+        raise RuntimeError("审计数据库配置缺失: " + ", ".join(missing))
+    try:
+        port = int(values["PG_PORT"])
+    except (TypeError, ValueError) as exc:
+        raise RuntimeError("PG_PORT 必须是整数") from exc
+    return None, {
+        "host": str(values["PG_HOST"]).strip(),
+        "port": port,
+        "dbname": str(values["PG_DB"]).strip(),
+        "user": str(values["PG_USER"]).strip(),
+        "password": str(values["PG_PASSWORD"]),
+    }
+
+
+def _apply_runtime_policy(environment: Mapping[str, str]) -> Mapping[str, str]:
+    """复用 RuntimeServicePolicy，保证与控制台/工作流看到同一个数据库地址。"""
+
+    try:
+        from gyxx_flow.adapters import RuntimeServicePolicy
+    except Exception:  # pragma: no cover - 循环导入兜底
+        return environment
+    try:
+        return RuntimeServicePolicy().apply(dict(environment))
+    except Exception:  # pragma: no cover
+        return environment
+
+
+def default_store(data_root: Path | str | None = None) -> AuditStore:
+    """Best-effort store resolved from the standard data root."""
+
+    if data_root is None:
+        try:
+            from gyxx_flow.core.config import Settings
+
+            data_root = Settings.from_env().data_root
+        except Exception:  # pragma: no cover
+            data_root = Path.cwd() / "var"
+    return AuditStore(data_root)

src/gyxx_flow/ops/index.py

@@ -86,6 +86,7 @@ class RunIndex:
         resource = f"ops:run-index:{_digest(record.run_id)}"
         with self._locks.acquire(resource, owner=_owner(), timeout_seconds=30):
             atomic_write_json(destination, record.to_dict())
+        _mirror_run_to_audit(self._data_root, record, raw)
         return record
 
     def get(self, run_id: str) -> RunRecord | None:
@@ -255,3 +256,42 @@ def _digest(value: str) -> str:
     import hashlib
 
     return hashlib.sha256(value.encode("utf-8")).hexdigest()[:32]
+
+
+_AUDIT_STORES: dict[str, Any] = {}
+
+
+def _audit_store_for(data_root: Path) -> Any:
+    """Reuse one store per data root so schema bootstrap and the circuit breaker
+    survive the short-lived ``RunIndex`` instances created per run."""
+
+    key = str(data_root)
+    store = _AUDIT_STORES.get(key)
+    if store is None:
+        from gyxx_flow.ops.audit import AuditStore
+
+        store = AuditStore(data_root)
+        _AUDIT_STORES[key] = store
+    return store
+
+
+def _mirror_run_to_audit(
+    data_root: Path,
+    record: RunRecord,
+    journal_payload: Any,
+) -> None:
+    """Best-effort mirror of a run into the PostgreSQL audit tables.
+
+    Auditing is strictly additive: any failure here (missing DSN, unreachable
+    database, unexpected schema state) is swallowed so that journal indexing and
+    the workflow itself are never affected.
+    """
+
+    try:
+        store = _audit_store_for(data_root)
+        if not store.enabled:
+            return
+        steps = journal_payload.get("steps") if isinstance(journal_payload, dict) else None
+        store.record_run(record.to_dict(), steps)
+    except Exception:  # noqa: BLE001 - audit mirroring must never raise
+        return

src/gyxx_flow/ops/migrations/001_ops_audit.sql

@@ -0,0 +1,152 @@
+-- Migration 001: operational audit tables (workflow runs + login cookie telemetry).
+--
+-- 本文件由 src/gyxx_flow/ops/audit.py 的 SCHEMA_SQL 导出，请勿手工编辑；
+-- 运行时首次写入会自动执行同一份 DDL（幂等），此处仅供 DBA 审阅或预建。
+--
+-- 用法:  psql "$GYXX_POSTGRES_DSN" -f 001_ops_audit.sql
+--
+-- 说明:
+--   ops_workflow_runs            一次工作流运行一行，run.json 的关系型投影
+--   ops_workflow_run_steps       每次尝试一行（含重试），保留 exit_code 与错误
+--   ops_account_cookie_events    cookie 观测事件流，只存指纹不存明文值
+--   ops_account_cookie_lifetimes 一处登录态从首次出现到失效（视图）
+--   ops_account_cookie_stats     每账号失效规律汇总（视图）
+--
+-- 日志正文仍留在 var/logs 下，数据库只存结构化元数据与指向路径。
+
+CREATE TABLE IF NOT EXISTS ops_workflow_runs (
+    run_id            TEXT PRIMARY KEY,
+    workflow_id       TEXT NOT NULL,
+    business_date     DATE NOT NULL,
+    mode              TEXT NOT NULL DEFAULT 'unknown',
+    shadow            BOOLEAN NOT NULL DEFAULT FALSE,
+    status            TEXT NOT NULL,
+    started_at        TIMESTAMPTZ,
+    ended_at          TIMESTAMPTZ,
+    duration_seconds  INTEGER,
+    error             TEXT,
+    step_total        INTEGER NOT NULL DEFAULT 0,
+    step_success      INTEGER NOT NULL DEFAULT 0,
+    step_failed       INTEGER NOT NULL DEFAULT 0,
+    step_skipped      INTEGER NOT NULL DEFAULT 0,
+    step_running      INTEGER NOT NULL DEFAULT 0,
+    journal_path      TEXT NOT NULL DEFAULT '',
+    host              TEXT NOT NULL DEFAULT '',
+    indexed_at        TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS ops_workflow_runs_workflow_idx
+    ON ops_workflow_runs (workflow_id, started_at DESC);
+CREATE INDEX IF NOT EXISTS ops_workflow_runs_status_idx
+    ON ops_workflow_runs (status, started_at DESC);
+CREATE INDEX IF NOT EXISTS ops_workflow_runs_business_date_idx
+    ON ops_workflow_runs (business_date, workflow_id);
+
+CREATE TABLE IF NOT EXISTS ops_workflow_run_steps (
+    run_id           TEXT NOT NULL
+                     REFERENCES ops_workflow_runs (run_id) ON DELETE CASCADE,
+    step_id          TEXT NOT NULL,
+    attempt          INTEGER NOT NULL DEFAULT 1,
+    status           TEXT NOT NULL DEFAULT '',
+    exit_code        INTEGER,
+    started_at       TIMESTAMPTZ,
+    ended_at         TIMESTAMPTZ,
+    duration_seconds INTEGER,
+    error            TEXT,
+    PRIMARY KEY (run_id, step_id, attempt)
+);
+
+CREATE INDEX IF NOT EXISTS ops_workflow_run_steps_step_idx
+    ON ops_workflow_run_steps (step_id, status);
+
+CREATE TABLE IF NOT EXISTS ops_account_cookie_events (
+    id                  BIGSERIAL PRIMARY KEY,
+    account_id          TEXT NOT NULL,
+    observed_at         TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    state               TEXT NOT NULL,
+    probe_status        TEXT NOT NULL DEFAULT '',
+    probe_source        TEXT NOT NULL DEFAULT '',
+    cookie_count        INTEGER,
+    required_count      INTEGER,
+    fingerprint         TEXT NOT NULL DEFAULT '',
+    declared_expires_at TIMESTAMPTZ,
+    detail              TEXT NOT NULL DEFAULT '',
+    workflow_id         TEXT NOT NULL DEFAULT '',
+    run_id              TEXT NOT NULL DEFAULT '',
+    host                TEXT NOT NULL DEFAULT ''
+);
+
+CREATE INDEX IF NOT EXISTS ops_account_cookie_events_account_idx
+    ON ops_account_cookie_events (account_id, observed_at DESC);
+CREATE INDEX IF NOT EXISTS ops_account_cookie_events_state_idx
+    ON ops_account_cookie_events (state, observed_at DESC);
+CREATE INDEX IF NOT EXISTS ops_account_cookie_events_fingerprint_idx
+    ON ops_account_cookie_events (account_id, fingerprint);
+
+-- 一段「cookie 代」的生命周期：同一 fingerprint 从首次出现到失效。
+-- first_failed_at 只取最后一次 healthy 之后的失败，避免恢复后再次失败时口径前移。
+CREATE OR REPLACE VIEW ops_account_cookie_lifetimes AS
+WITH per_fingerprint AS (
+    SELECT
+        account_id,
+        fingerprint,
+        MIN(observed_at) AS first_seen_at,
+        MIN(observed_at) FILTER (WHERE state = 'healthy') AS first_healthy_at,
+        MAX(observed_at) FILTER (WHERE state = 'healthy') AS last_healthy_at,
+        MAX(declared_expires_at) AS declared_expires_at
+    FROM ops_account_cookie_events
+    WHERE fingerprint <> ''
+    GROUP BY account_id, fingerprint
+)
+SELECT
+    p.account_id,
+    p.fingerprint,
+    p.first_seen_at,
+    p.first_healthy_at,
+    p.last_healthy_at,
+    p.declared_expires_at,
+    f.first_failed_at,
+    (f.first_failed_at IS NOT NULL) AS expired,
+    CASE
+        WHEN p.first_healthy_at IS NOT NULL AND p.last_healthy_at IS NOT NULL
+        THEN EXTRACT(EPOCH FROM (p.last_healthy_at - p.first_healthy_at))::BIGINT
+    END AS healthy_span_seconds,
+    CASE
+        WHEN p.first_healthy_at IS NOT NULL AND f.first_failed_at IS NOT NULL
+        THEN EXTRACT(EPOCH FROM (f.first_failed_at - p.first_healthy_at))::BIGINT
+    END AS lifetime_seconds,
+    CASE
+        WHEN p.last_healthy_at IS NOT NULL AND f.first_failed_at IS NOT NULL
+        THEN EXTRACT(EPOCH FROM (f.first_failed_at - p.last_healthy_at))::BIGINT
+    END AS healthy_to_failure_seconds
+FROM per_fingerprint p
+LEFT JOIN LATERAL (
+    SELECT MIN(e.observed_at) AS first_failed_at
+    FROM ops_account_cookie_events e
+    WHERE e.account_id = p.account_id
+      AND e.fingerprint = p.fingerprint
+      AND e.state IN ('expired', 'not_logged_in')
+      AND (p.last_healthy_at IS NULL OR e.observed_at > p.last_healthy_at)
+) f ON TRUE;
+
+-- 按账号汇总的失效规律：平均/中位/最短/最长寿命，直接支撑「多久失效一次」。
+CREATE OR REPLACE VIEW ops_account_cookie_stats AS
+SELECT
+    account_id,
+    COUNT(*) AS sessions,
+    COUNT(*) FILTER (WHERE expired) AS expired_sessions,
+    AVG(lifetime_seconds)::BIGINT AS avg_lifetime_seconds,
+    PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY lifetime_seconds)::BIGINT
+        AS median_lifetime_seconds,
+    MIN(lifetime_seconds) AS min_lifetime_seconds,
+    MAX(lifetime_seconds) AS max_lifetime_seconds,
+    MAX(first_failed_at) AS last_failed_at,
+    MAX(last_healthy_at) AS last_healthy_at
+FROM ops_account_cookie_lifetimes
+WHERE lifetime_seconds IS NOT NULL
+GROUP BY account_id;
+
+CREATE TABLE IF NOT EXISTS ops_audit_schema (
+    schema_version INTEGER PRIMARY KEY,
+    applied_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);

tests/test_ops_audit.py

@@ -0,0 +1,359 @@
+from __future__ import annotations
+
+import json
+from datetime import date, datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from gyxx_flow.ops.audit import (
+    SCHEMA_SQL,
+    AuditStore,
+    split_step_key,
+    summarize_cookies,
+)
+from gyxx_flow.ops.index import RunIndex
+
+
+DOUYIN_COOKIES: list[dict[str, Any]] = [
+    {
+        "name": "sessionid",
+        "value": "AAA",
+        "domain": ".jinritemai.com",
+        "expires": 1766000000.0,
+    },
+    {
+        "name": "sid_tt",
+        "value": "BBB",
+        "domain": ".jinritemai.com",
+        "expires": 1766100000.0,
+    },
+    {
+        "name": "unrelated",
+        "value": "CCC",
+        "domain": ".example.com",
+        "expires": 1700000000.0,
+    },
+    {
+        "name": "session",
+        "value": "DDD",
+        "domain": ".jinritemai.com",
+        "expires": -1,
+    },
+]
+
+
+class _FakeCursor:
+    def __init__(self, log: list[Any], rows: list[dict[str, Any]]) -> None:
+        self._log = log
+        self._rows = rows
+
+    def __enter__(self) -> "_FakeCursor":
+        return self
+
+    def __exit__(self, *_: object) -> bool:
+        return False
+
+    def execute(self, sql: str, params: Any = None) -> None:
+        self._log.append(("execute", sql, params))
+
+    def executemany(self, sql: str, rows: Any) -> None:
+        self._log.append(("executemany", sql, list(rows)))
+
+    def fetchall(self) -> list[dict[str, Any]]:
+        return self._rows
+
+    def fetchone(self) -> dict[str, Any] | None:
+        # 默认模拟「ops_audit_schema 尚不存在」，从而走首次建表分支。
+        return None
+
+    def close(self) -> None:
+        return None
+
+
+class _FakeConnection:
+    def __init__(self, log: list[Any], rows: list[dict[str, Any]] | None = None) -> None:
+        self._log = log
+        self._rows = rows or []
+
+    def cursor(self) -> _FakeCursor:
+        return _FakeCursor(self._log, self._rows)
+
+    def commit(self) -> None:
+        self._log.append(("commit",))
+
+    def close(self) -> None:
+        self._log.append(("close",))
+
+
+def _store_with_fake(
+    tmp_path: Path,
+    *,
+    environment: dict[str, str] | None = None,
+    rows: list[dict[str, Any]] | None = None,
+) -> tuple[AuditStore, list[Any]]:
+    log: list[Any] = []
+    store = AuditStore(
+        tmp_path,
+        environment=environment or {"GYXX_POSTGRES_DSN": "postgresql://fake/audit"},
+        connect=lambda: _FakeConnection(log, rows),
+    )
+    return store, log
+
+
+# ----------------------------------------------------------------------
+# cookie 快照
+# ----------------------------------------------------------------------
+
+
+def test_summarize_cookies_keeps_only_required_cookies() -> None:
+    snapshot = summarize_cookies(
+        DOUYIN_COOKIES,
+        required_domains=("jinritemai.com",),
+        required_names=("sessionid", "sid_tt", "session"),
+    )
+
+    assert snapshot.cookie_count == 4
+    assert snapshot.required_count == 3
+    # 最早的必需 cookie 过期时间：1766000000，而不是无关 cookie 的 1700000000
+    assert snapshot.declared_expires_at == datetime.fromtimestamp(
+        1766000000.0, tz=timezone.utc
+    )
+
+
+def test_summarize_cookies_ignores_session_cookies_for_expiry() -> None:
+    snapshot = summarize_cookies(
+        DOUYIN_COOKIES,
+        required_domains=("jinritemai.com",),
+        required_names=("session",),
+    )
+
+    assert snapshot.required_count == 1
+    assert snapshot.declared_expires_at is None
+
+
+def test_summarize_cookies_fingerprint_tracks_value_changes_only() -> None:
+    baseline = summarize_cookies(DOUYIN_COOKIES, required_domains=("jinritemai.com",))
+    reordered = summarize_cookies(
+        list(reversed(DOUYIN_COOKIES)), required_domains=("jinritemai.com",)
+    )
+    replaced = [dict(cookie) for cookie in DOUYIN_COOKIES]
+    replaced[0]["value"] = "ROTATED"
+
+    assert baseline.fingerprint  # 非空
+    assert baseline.fingerprint == reordered.fingerprint
+    assert baseline.fingerprint != summarize_cookies(
+        replaced, required_domains=("jinritemai.com",)
+    ).fingerprint
+
+
+def test_summarize_cookies_without_requirements_counts_everything() -> None:
+    snapshot = summarize_cookies(DOUYIN_COOKIES)
+
+    assert snapshot.required_count == 4
+    assert snapshot.declared_expires_at == datetime.fromtimestamp(
+        1700000000.0, tz=timezone.utc
+    )
+
+
+def test_summarize_cookies_has_no_plaintext_value_in_fingerprint() -> None:
+    snapshot = summarize_cookies(DOUYIN_COOKIES, required_domains=("jinritemai.com",))
+
+    assert "AAA" not in snapshot.fingerprint
+    assert len(snapshot.fingerprint) == 32
+
+
+def test_split_step_key_handles_retry_suffix() -> None:
+    assert split_step_key("sync.attempt-2") == ("sync", 2)
+    assert split_step_key("collect_collaborators") == ("collect_collaborators", 1)
+    assert split_step_key("a.attempt-x") == ("a.attempt-x", 1)
+
+
+# ----------------------------------------------------------------------
+# 写入装配
+# ----------------------------------------------------------------------
+
+
+def test_record_run_upserts_run_and_steps(tmp_path: Path) -> None:
+    store, log = _store_with_fake(tmp_path)
+
+    ok = store.record_run(
+        {
+            "workflow_id": "content.metrics.daily",
+            "run_id": "content.metrics.daily__20261018__abc",
+            "business_date": "2026-10-18",
+            "status": "success",
+            "mode": "dry_run",
+            "shadow": False,
+            "started_at": "2026-10-19T01:41:43+00:00",
+            "ended_at": "2026-10-19T01:41:48+00:00",
+            "error": None,
+            "journal_path": "runs/content.metrics.daily/run.json",
+        },
+        {
+            "sync.attempt-1": {
+                "status": "success",
+                "exit_code": 0,
+                "started_at": "2026-10-19T01:41:44+00:00",
+                "ended_at": "2026-10-19T01:41:46+00:00",
+            },
+            "collect.attempt-1": {"status": "failed", "exit_code": 4, "error": "boom"},
+        },
+    )
+
+    assert ok is True
+    statements = [entry for entry in log if entry[0] == "execute"]
+    assert any("CREATE TABLE IF NOT EXISTS ops_workflow_runs" in item[1] for item in statements)
+
+    insert = next(item for item in statements if "INSERT INTO ops_workflow_runs" in item[1])
+    params = insert[2]
+    assert params["business_date"] == date(2026, 10, 18)
+    assert params["duration_seconds"] == 5
+    assert params["step_total"] == 2
+    assert params["step_success"] == 1
+    assert params["step_failed"] == 1
+
+    batch = next(item for item in log if item[0] == "executemany")
+    rows = {item["step_id"]: item for item in batch[2]}
+    assert rows["sync"]["attempt"] == 1
+    assert rows["sync"]["duration_seconds"] == 2
+    assert rows["collect"]["exit_code"] == 4
+    assert rows["collect"]["error"] == "boom"
+
+
+def test_record_run_rejects_incomplete_payload(tmp_path: Path) -> None:
+    store, log = _store_with_fake(tmp_path)
+
+    assert store.record_run({"workflow_id": "x"}) is False
+    assert not [entry for entry in log if entry[0] == "execute"]
+
+
+def test_record_cookie_event_writes_value_free_snapshot(tmp_path: Path) -> None:
+    store, log = _store_with_fake(tmp_path)
+    snapshot = summarize_cookies(DOUYIN_COOKIES, required_domains=("jinritemai.com",))
+
+    ok = store.record_cookie_event(
+        account_id="douyin-shop",
+        state="healthy",
+        snapshot=snapshot,
+        probe_status="success",
+        probe_source="keepalive",
+        workflow_id="content.metrics.daily",
+        run_id="run-1",
+    )
+
+    assert ok is True
+    insert = next(
+        item
+        for item in log
+        if item[0] == "execute" and "INSERT INTO ops_account_cookie_events" in item[1]
+    )
+    params = insert[2]
+    assert params["account_id"] == "douyin-shop"
+    assert params["state"] == "healthy"
+    assert params["fingerprint"] == snapshot.fingerprint
+    assert params["declared_expires_at"] == snapshot.declared_expires_at
+    # 明文值不得出现在任何参数里
+    assert "AAA" not in json.dumps(params, default=str)
+
+
+def test_record_cookie_event_rejects_unknown_state(tmp_path: Path) -> None:
+    store, _ = _store_with_fake(tmp_path)
+
+    with pytest.raises(ValueError):
+        store.record_cookie_event(account_id="a", state="definitely-not-valid")
+
+
+# ----------------------------------------------------------------------
+# 降级与熔断
+# ----------------------------------------------------------------------
+
+
+def test_missing_dsn_never_raises_and_logs_failure(tmp_path: Path) -> None:
+    store = AuditStore(tmp_path, environment={})
+
+    assert store.record_run(
+        {
+            "workflow_id": "content.metrics.daily",
+            "run_id": "r1",
+            "business_date": "2026-10-18",
+            "status": "success",
+        }
+    ) is False
+    assert "审计数据库配置缺失" in store.last_error
+    failure_log = tmp_path / "logs" / "ops-audit" / "audit-failures.log"
+    assert failure_log.exists()
+    assert json.loads(failure_log.read_text(encoding="utf-8").splitlines()[0])["operation"] == (
+        "record_run"
+    )
+
+
+def test_audit_can_be_disabled_without_touching_the_database(tmp_path: Path) -> None:
+    store, log = _store_with_fake(
+        tmp_path, environment={"GYXX_POSTGRES_DSN": "postgresql://fake", "GYXX_OPS_AUDIT": "off"}
+    )
+
+    assert store.enabled is False
+    assert store.record_cookie_event(account_id="a", state="healthy") is False
+    assert log == []
+
+
+def test_consecutive_failures_open_the_circuit(tmp_path: Path) -> None:
+    attempts = 0
+
+    def _boom() -> Any:
+        nonlocal attempts
+        attempts += 1
+        raise OSError("database unreachable")
+
+    store = AuditStore(
+        tmp_path,
+        environment={"GYXX_POSTGRES_DSN": "postgresql://fake", "GYXX_OPS_AUDIT_FAILURE_COOLDOWN": "60"},
+        connect=_boom,
+    )
+    payload = {
+        "workflow_id": "w",
+        "run_id": "r",
+        "business_date": "2026-10-18",
+        "status": "success",
+    }
+
+    for _ in range(3):
+        store.record_run(payload)
+    assert attempts == 3
+
+    # 熔断生效后不再尝试连接，避免每个工作流都卡在 TCP 超时上
+    assert store.record_run(payload) is False
+    assert attempts == 3
+
+
+# ----------------------------------------------------------------------
+# 与运行索引的集成
+# ----------------------------------------------------------------------
+
+
+def test_run_index_still_works_when_audit_is_unavailable(tmp_path: Path) -> None:
+    from gyxx_flow.core.context import RunContext
+    from gyxx_flow.core.layout import DataLayout
+    from gyxx_flow.core.records import RunJournal
+
+    layout = DataLayout(tmp_path)
+    context = RunContext.create("content.metrics.daily", "2026-10-18")
+    journal = RunJournal.create(layout, context, mode="dry_run")
+    journal.finalize("success")
+
+    record = RunIndex(tmp_path).index_journal(journal)
+
+    assert record.workflow_id == "content.metrics.daily"
+    assert record.status == "success"
+    # 运行索引文件仍然写出，审计失败不影响主链路
+    assert (tmp_path / "state" / "ops" / "run-index" / f"{record.run_id}.json").exists()
+
+
+def test_schema_sql_defines_both_audit_surfaces() -> None:
+    assert "ops_workflow_runs" in SCHEMA_SQL
+    assert "ops_workflow_run_steps" in SCHEMA_SQL
+    assert "ops_account_cookie_events" in SCHEMA_SQL
+    assert "ops_account_cookie_lifetimes" in SCHEMA_SQL
+    assert "ops_account_cookie_stats" in SCHEMA_SQL