fix(events): S2.1 修掉 emit() 的 naive 时区偏移，并打通台账列与审计关联

外部审计对 S1+S2 提出 3 条，全部复核成立，本提交一并修掉。第 1 条是真缺陷，
且已经写进数据了。

## 1. ⚠️ emit() 把 naive datetime 当成「本地时间」（数据正确性）

`emit()` 原来写 `(occurred_at or now).astimezone(UTC)`。对 **naive** 值，Python 会
把它当作**服务器本地时间**再换算 —— 本机 UTC+8，事件时间整整早 8 小时。

naive 的来源很具体：收集器用**裸 SQL** 读 `douyu_tasks`，绕过 `UTCDateTime`
装饰器，拿到的就是 naive。而这条路正是「充值/兑换/积分/开通」台账。

仓库早有既定约定（写在 `database.UTCDateTime` 里）：**naive → 视为已是 UTC，原样用**。
`scheduler._as_utc` / `yyb_service._as_utc` / `audit.py` 都遵守了，**只有 emit() 漏了**。
现抽成 `account_events_service.to_naive_utc()` 统一收口，并补齐文档。

**数据库铁证**（修复前，本机开发库）：
```
事件 occurred_at        源 douyu_tasks.finished_at    相差
2026-09-30 20:23:36  →  2026-10-01 04:23:36          8 小时
```
已用 `UPDATE ... JOIN douyu_tasks` 把那 3 条按源时间对齐（相差 0 秒），
修复 SQL 记在账号事件设计文档 §11.2 第 5 条，供曾开启过收集器的环境使用。
该作业默认 `enabled=false`、需人工开启，故生产大概率无受影响数据。

## 2. 回归用例：这个 bug 在 UTC 机器上**看不见**

`astimezone(UTC)` 在 TZ=UTC 下恰好是恒等变换 —— 旧实现与正确实现结果相同。
也就是说单写一条普通用例，在 CI/容器里会**假绿**，只在开发机现形。

所以用例显式把进程时区切到 UTC+8（`_force_non_utc_timezone`）。已验证：
**在 TZ=UTC 下临时还原旧实现，该用例照样红**。另补一条 aware 路径的用例，
防止「为了修 naive 把换算弄坏」。

顺带补了收集器的端到端用例：**事件 occurred_at 必须等于源任务 finished_at**。
旧用例只断言 written/watermark 数量、从不看时间，这正是 bug 一路绿灯的原因。

## 3. emit() 支持 S1 建的台账列（此前是「空壳」）

迁移 0042 给 `account_events` 加了 `category` / `subject_type` / `subject_key` /
`actor_type` / `actor_id` 与复合索引 `ix_account_events_ledger`，但 `emit()` 一直没
暴露它们 —— 列建好了没有写入方。现补齐参数与 INSERT，并加用例锁住「能写」。
S3 的 `status_changed`（`category="lifecycle"`）与 S5 的台账查询都要靠它。

## 4. 清 CK 的审计补 account_id

`DELETE /cookies/{id}` 与 `/cookies/batch` 的审计只写了 `target`（自由文本，
账号改名/重名就关联不上）。S1 已给 `audit_logs` 加了 `account_id`，这里补上写入
（批量时仅在只涉及一个账号时填），并加用例断言审计能反查到账号。
这是 S6「审计进账号时间线」的前置。

## 验证

- 新增/修改 5 条用例，先红后绿；**已验证在 TZ=UTC 下旧实现照样被这些用例抓到**
- `TZ=Asia/Shanghai`：**373 passed / 6 skipped**（原 368 + 5 条新用例）
- `TZ=UTC`：369 passed / 4 failed —— 那 4 条是 `test_huya_web_cookie_fields.py`
  的 qimei 用例，**在 main 上同样失败**（qimei 按 SDK 语义编码本地时间，
  与本次改动无关，也未修）
- ruff check / format 全绿；前端 `vite build` 通过（本次未动前端）

docs/设计-登录任务与CK解耦.md

@@ -624,17 +624,15 @@ C1（加列+回填，无行为变化）
 
 解耦（C1~C5）是**前置**：凭证状态不回到账号上，后面的「状态机」「账号页」都是空中楼阁。
 
-| 阶段 | 内容 | 依赖 |
-| --- | --- | --- |
-| **S1 = C1~C3** | CK / 健康 / 重登 状态搬到 `accounts`（含作业改按账号调度） | —— |
-| **S2 = C4~C5** | Cookie 页与删除护栏改为账号级 | S1 |
-| **S3** | `AccountOut` 补全运营状态字段；账号列表页加「CK / 健康 / 积分 / 开通」列 | S1、S2 |
-| **S4** | 账号详情页补全生命周期（状态机、画像、开通时间、累计统计）—— 已有 `AccountDetailPage` 底座 | S3 |
-| **S5** | 补 ③ 生命周期 `status` + 停用/归档 + 导入批次 | S1 |
-| **S6** | 打通 `audit_logs.account_id`，让「分配/改标签/删除」进时间线 | S1 |
+> ⚠️ **阶段编号以 [设计-账号全生命周期](设计-账号全生命周期.md) §6 为准（S1~S8）。**
+> 本文早期版本在这里另写过一套 S1~S6，两套编号含义不同，已废弃 —— 保留本节只为说明依赖关系。
 
-> S3/S4 是用户感知最强的部分（「账号管理终于能看全」），但**必须等 S1**——
-> 否则账号页展示的 CK 状态仍然来自登录记录，删一条记录页面就变了。
+| 依赖关系 | 说明 |
+| --- | --- |
+| **解耦（本文 C1~C5 = 生命周期设计 S1~S2）** | 一切的前置。不做，账号页展示的 CK 状态仍然来自登录记录，删一条记录页面就变了 |
+| **状态收口（S3）** | 依赖解耦；产出「可信的 `status`」 |
+| **账号页重构（S6）** | 依赖 S3（要显示状态）+ S5（要显示台账） |
+| **清理旧列（S8）** | 最后一步，不可逆，前置每日备份 |
 
 ### 13.6 待确认
 

docs/设计-账号详情页与事件历史.md

@@ -332,7 +332,28 @@ class AccountEvent(Base):
 3. **`delete_account` 必须连带清事件**：`account_events` 有指向 `accounts` 的外键，不清就删不掉账号。
    这是本次改动**引入的**问题，已修并有用例；
 4. **测试数据的字符集**：用 `docker exec mysql < file` 灌中文时没带 `--default-character-set=utf8mb4`，
-   数据被双重编码（`套装` → `套装`）。排查时先看 `HEX()` 就能立刻分清「存坏了」还是「读坏了」。
+   数据被双重编码（`套装` → `套装`）。排查时先看 `HEX()` 就能立刻分清「存坏了」还是「读坏了」；
+5. **⚠️ naive datetime 被当成「本地时间」（2026-10-04 审计发现并修复）**：
+   `emit()` 原来写的是 `(occurred_at or now).astimezone(UTC)`。对 **naive** 值，Python 会把它
+   当作**服务器本地时间**再换算 —— 开发机 UTC+8，事件时间整整**早 8 小时**。
+   naive 的来源很常见：**裸 SQL** 读 DATETIME 列（收集器读 `douyu_tasks.finished_at`）
+   会绕过 `UTCDateTime` 装饰器，拿到的就是 naive。
+   仓库的既定约定是「**naive = 已经是 UTC，原样用**」，`scheduler._as_utc` /
+   `yyb_service._as_utc` / `audit.py` 都遵守了，只有 `emit()` 漏了。
+   已抽成 `account_events_service.to_naive_utc()` 统一收口。
+
+   **这个 bug 在 TZ=UTC 的机器上完全不可见**（`astimezone(UTC)` 恰好是恒等变换），
+   所以只在开发机上现形、CI/容器里全绿 —— 回归用例必须**显式把进程时区切到非 UTC**
+   才有意义（见 `tests/test_account_events.py` 的 `_force_non_utc_timezone`）。
+
+   **若某套环境曾开启过收集器**，用这段按源任务时间对齐（幂等）：
+
+   ```sql
+   UPDATE account_events e
+   JOIN douyu_tasks t ON t.id = CAST(SUBSTRING(e.source_ref, 12) AS UNSIGNED)
+   SET e.occurred_at = COALESCE(t.finished_at, t.created_at)
+   WHERE e.source = 'task' AND e.source_ref LIKE 'douyu_task:%';
+   ```
 
 ### 11.3 本地验证结果
 

tests/test_account_events.py

@@ -7,6 +7,10 @@
 3. **水位线只进不退**：解析失败要跳过而不是卡住，否则一行坏数据会让事件流永久停摆。
 """
 
+import os
+import time
+from collections.abc import Iterator
+from contextlib import contextmanager
 from datetime import UTC, datetime
 
 from sqlalchemy.orm import sessionmaker
@@ -18,6 +22,27 @@ from web.backend.services import account_events_service as svc
 from web.backend.services.automation.jobs import account_events_collect as collect_job
 
 
+@contextmanager
+def _force_non_utc_timezone(name: str = "Asia/Shanghai") -> Iterator[None]:
+    """临时把进程时区切到非 UTC，让「naive 被当成本地时间」这类 bug 必然暴露。
+
+    ⚠️ 不加这一段，时区相关的回归在 **UTC 机器上（CI / 生产容器）会假绿**：
+    旧实现 `naive.astimezone(UTC)` 在 TZ=UTC 下恰好等于正确结果，
+    只有开发机（UTC+8）才现形 —— 而「只在某些机器上失败」的测试等于没有测试。
+    """
+    previous = os.environ.get("TZ")
+    os.environ["TZ"] = name
+    time.tzset()
+    try:
+        yield
+    finally:
+        if previous is None:
+            os.environ.pop("TZ", None)
+        else:
+            os.environ["TZ"] = previous
+        time.tzset()
+
+
 class _Base:
     def setup_method(self):
         self.engine = get_test_engine()
@@ -52,6 +77,136 @@ class _Base:
 
 
 class TestEmit(_Base):
+    def test_naive_occurred_at_is_treated_as_utc(self):
+        """⚠️ 回归：naive datetime 必须按「已经是 UTC」处理，不能当本地时间。
+
+        仓库的既定约定写在 `database.UTCDateTime` 里：
+        **写：aware → 转 UTC 去 tzinfo；naive → 视为已是 UTC，原样存**。
+        `scheduler._as_utc` / `yyb_service._as_utc` / `audit.py` 都遵守了这条。
+
+        但 `emit()` 曾经直接 `.astimezone(UTC)` —— 对 naive 值，Python 会把它
+        当作**服务器本地时间**再换算。本机是 UTC+8 时，事件时间会被整整扣掉
+        8 小时（实测库里 `account_events.occurred_at` 比源 `douyu_tasks.finished_at`
+        早 8 小时）。而 `occurred_at` 是「业务发生时间」，错了整条时间线都是错的。
+
+        naive 值的来源很具体：收集器用**裸 SQL** 读 `douyu_tasks`，绕过了
+        `UTCDateTime` 装饰器，拿到的就是 naive。
+
+        用例显式切到 UTC+8 执行：在 UTC 机器上旧实现与正确实现结果相同，
+        不切时区就会假绿。
+        """
+        with _force_non_utc_timezone():
+            # 故意造 naive 值：本用例就是要验证 naive 的处理
+            naive_probe = datetime(2026, 10, 3, 11, 46, 7)  # noqa: DTZ001
+            assert naive_probe.astimezone(UTC).replace(tzinfo=None) != naive_probe, (
+                "时区未生效，本用例失去意义"
+            )
+            db = self._fresh()
+            try:
+                # 模拟从 DATETIME 列读出来的裸值（语义是 UTC，但没有 tzinfo）
+                naive_utc = datetime(2026, 10, 3, 11, 46, 7)  # noqa: DTZ001
+                assert naive_utc.tzinfo is None
+                ok = svc.emit(
+                    db,
+                    account_id=self.account_id,
+                    event_type=svc.EVENT_RECHARGE,
+                    title="充值 98",
+                    amount=98,
+                    occurred_at=naive_utc,
+                    source="task",
+                    source_ref="naive-utc-regression",
+                )
+                assert ok
+                db.commit()
+            finally:
+                db.close()
+
+        db = self._fresh()
+        try:
+            event = (
+                db.query(AccountEvent)
+                .filter(AccountEvent.source_ref == "naive-utc-regression")
+                .one()
+            )
+            # 存进去的应当是同一个「墙上时间」，不能被时区换算挪动
+            assert event.occurred_at == datetime(2026, 10, 3, 11, 46, 7, tzinfo=UTC)
+        finally:
+            db.close()
+
+    def test_aware_occurred_at_is_converted_to_utc(self):
+        """aware 值仍要正常换算（别为了修 naive 把这条路弄坏）。"""
+        from datetime import timedelta, timezone
+
+        db = self._fresh()
+        try:
+            ok = svc.emit(
+                db,
+                account_id=self.account_id,
+                event_type=svc.EVENT_RECHARGE,
+                title="充值 1",
+                occurred_at=datetime(
+                    2026, 10, 3, 19, 46, 7, tzinfo=timezone(timedelta(hours=8))
+                ),
+                source="task",
+                source_ref="aware-utc-regression",
+            )
+            assert ok
+            db.commit()
+        finally:
+            db.close()
+
+        db = self._fresh()
+        try:
+            event = (
+                db.query(AccountEvent)
+                .filter(AccountEvent.source_ref == "aware-utc-regression")
+                .one()
+            )
+            assert event.occurred_at == datetime(2026, 10, 3, 11, 46, 7, tzinfo=UTC)
+        finally:
+            db.close()
+
+    def test_stores_category_subject_and_actor(self):
+        """台账检索列（category / subject_* / actor_*）要真的能写进去。
+
+        S1 的迁移 0042 建了这 5 列与复合索引 `ix_account_events_ledger`，但 `emit()`
+        一直没有暴露它们 —— 列建好了却没有写入方（和 `status` 一样的「空壳」）。
+        这条用例锁住「能写」，S3/S5 才能靠索引做「该账号该商品最近一次兑换」这类查询。
+        """
+        db = self._fresh()
+        try:
+            ok = svc.emit(
+                db,
+                account_id=self.account_id,
+                event_type=svc.EVENT_EXCHANGE,
+                title="兑换 兔兔白日梦",
+                category=svc.CATEGORY_BUSINESS,
+                subject_type="commodity",
+                subject_key="20260120QYOOB_bag10",
+                actor_type="system",
+                actor_id=7,
+                source_ref="ledger-columns-regression",
+            )
+            assert ok
+            db.commit()
+        finally:
+            db.close()
+
+        db = self._fresh()
+        try:
+            event = (
+                db.query(AccountEvent)
+                .filter(AccountEvent.source_ref == "ledger-columns-regression")
+                .one()
+            )
+            assert event.category == "business"
+            assert event.subject_type == "commodity"
+            assert event.subject_key == "20260120QYOOB_bag10"
+            assert event.actor_type == "system"
+            assert event.actor_id == 7
+        finally:
+            db.close()
+
     def test_writes_event(self):
         db = self._fresh()
         try:
@@ -397,6 +552,49 @@ class TestCollector(_Base):
         assert result["detail"]["watermark"] == good_id
         assert good_id > bad_id
 
+    def test_collected_occurred_at_matches_source_task_time(self):
+        """⚠️ 端到端回归：收集器写的事件时间必须等于源任务的业务时间。
+
+        这条用例存在的理由：收集器用**裸 SQL** 读 `douyu_tasks`，拿到的是 naive
+        datetime（绕过 `UTCDateTime`）。旧实现把它当本地时间换算，导致事件时间
+        整整早 8 小时（本机 UTC+8，库里实测过）——而「什么时候兑换/充值了」
+        正是这张表存在的意义。
+
+        旧用例只断言 written / watermark 的数量，从不看时间，所以这个 bug
+        一路绿灯溜了过去。
+        """
+        from web.backend.models import DouyuTask
+
+        task_id = self._add_task(
+            "create_gold_qr",
+            {"buy_num": 98, "order_id": "o-time", "supplier_order": {}},
+        )
+        # 给源任务一个明确的业务完成时间（UTC 语义）
+        source = self.session.query(DouyuTask).filter(DouyuTask.id == task_id).one()
+        source.finished_at = datetime(2026, 10, 3, 11, 46, 7)  # noqa: DTZ001
+        self.session.commit()
+
+        job = self._make_job(watermark=task_id - 1)
+        # 显式切到 UTC+8：这个 bug 在 UTC 机器上不可见，不切时区会假绿。
+        with _force_non_utc_timezone():
+            db = self._fresh()
+            try:
+                result = collect_job.run_account_events_collect(db, job)
+            finally:
+                db.close()
+        assert result["detail"]["written"] == 1
+
+        db = self._fresh()
+        try:
+            event = (
+                db.query(AccountEvent)
+                .filter(AccountEvent.source_ref == f"douyu_task:{task_id}")
+                .one()
+            )
+            assert event.occurred_at == datetime(2026, 10, 3, 11, 46, 7, tzinfo=UTC)
+        finally:
+            db.close()
+
 
 class TestAccountApi(_Base):
     """账号详情接口：概览统计 + 时间线分页 + 删除账号时清理事件。"""

tests/test_cookie_operations.py

@@ -11,9 +11,10 @@ from unittest.mock import patch
 import pytest
 
 from tests.support.db import get_test_engine, make_test_session
-from web.backend.models import Account, LoginTask, ProxyConfig, User
+from web.backend.models import Account, AuditLog, LoginTask, ProxyConfig, User
 from web.backend.routers.cookies import (
     check_cookie_operations,
+    delete_cookie,
     get_cookie,
     list_cookie_operation_tags,
     list_cookie_operations,
@@ -345,3 +346,31 @@ class TestCookieOperation:
         assert result["success"]
         selected_accounts = start_relogin.call_args.args[0]
         assert [account.id for account in selected_accounts] == [self.owned_account.id]
+
+    def test_clearing_cookie_writes_audit_linked_to_account(self):
+        """清 CK 是**不可逆**操作，审计必须能反查到账号。
+
+        S1 给 `audit_logs` 加了 `account_id`（此前 `target` 是自由文本，账号改名
+        或重名就关联不上）。这条用例锁住「清 CK 的审计真的写了 account_id」。
+        """
+        admin = User(username="ck-admin", password_hash="hash", role="super_admin")
+        self.session.add(admin)
+        self.session.commit()
+        target_id = self.owned_account.id
+
+        delete_cookie(target_id, db=self.session, current=admin)
+
+        entry = (
+            self.session.query(AuditLog)
+            .filter(AuditLog.action == "cookie:delete")
+            .order_by(AuditLog.id.desc())
+            .first()
+        )
+        assert entry is not None
+        assert entry.account_id == target_id
+        assert entry.target == "owned-account"
+        # 只清了入参那个账号，另一个不受影响
+        self.session.refresh(self.owned_account)
+        self.session.refresh(self.other_account)
+        assert self.owned_account.cookie == ""
+        assert self.other_account.cookie == "other-secret"

web/backend/routers/cookies.py

@@ -784,12 +784,15 @@ def delete_cookies_batch(
         acc.ck_check_status = ""
         acc.ck_check_result = None
         acc.ck_checked_at = None
+    # 批量清 CK 涉及多个账号，只能落到 target 文本里；单个账号时补 account_id，
+    # 便于审计页/账号时间线按索引反查。
     db.add(
         AuditLog(
             user_id=current.id,
             username=current.username,
             action="cookie:delete",
             target=f"清除 {len(accounts)} 个账号的 CK",
+            account_id=accounts[0].id if len(accounts) == 1 else None,
             detail="不可逆：CK 已从账号上清空",
         )
     )
@@ -819,6 +822,7 @@ def delete_cookie(
     )
     if not acc:
         raise HTTPException(status_code=404, detail="账号不存在或没有 CK")
+    account_id = acc.id
     username = acc.username
     acc.cookie = ""
     acc.cookie_updated_at = datetime.now(UTC)
@@ -831,6 +835,8 @@ def delete_cookie(
             username=current.username,
             action="cookie:delete",
             target=username,
+            # 关联到账号：S1 给审计表加了 account_id，这里才是它第一个真正的写入方
+            account_id=account_id,
             detail="不可逆：CK 已从该账号清空",
         )
     )

web/backend/services/account_events_service.py

@@ -52,6 +52,32 @@ EVENT_TAG_CHANGED = "tag_changed"  # 标签变更
 INCOME_TYPES = (EVENT_RECHARGE,)
 OUTCOME_TYPES = (EVENT_EXCHANGE,)
 
+# 粗分类：决定时间线怎么分组、以及哪些事件算「业务台账」。
+CATEGORY_LIFECYCLE = "lifecycle"  # 生命周期状态变迁
+CATEGORY_AUTH = "auth"  # 登录与 CK
+CATEGORY_BUSINESS = "business"  # 充值 / 兑换 / 积分 / 开通
+CATEGORY_OPS = "ops"  # 人工操作（分配 / 标签 / 停用）
+
+
+def to_naive_utc(value: datetime) -> datetime:
+    """把时间归一成 **naive UTC**（本服务所有时间列的落库形态）。
+
+    约定与 `database.UTCDateTime` 一致，也是全仓通行写法：
+
+    - **naive → 视为「已经是 UTC」，原样使用**；
+    - aware → 换算到 UTC 后去掉 tzinfo。
+
+    ⚠️ 绝对不能直接写 `value.astimezone(UTC)`：对 naive 值，Python 会把它当作
+    **服务器本地时间**再换算。本机 UTC+8 时事件时间会被整整扣掉 8 小时，
+    而 naive 值的来源很常见 —— 用**裸 SQL** 读 DATETIME 列（如收集器读
+    `douyu_tasks.finished_at`）拿到的就是 naive，因为它绕过了 `UTCDateTime`
+    装饰器。仓库里 `scheduler._as_utc` / `yyb_service._as_utc` / `audit.py`
+    都遵守了这条约定，本函数与它们同口径。
+    """
+    if value.tzinfo is None:
+        return value
+    return value.astimezone(UTC).replace(tzinfo=None)
+
 
 def _now() -> datetime:
     return datetime.now(UTC)
@@ -69,11 +95,23 @@ def emit(
     detail: dict[str, Any] | None = None,
     source: str = "manual",
     source_ref: str | None = None,
+    category: str = "",
+    subject_type: str = "",
+    subject_key: str = "",
+    actor_type: str = "",
+    actor_id: int | None = None,
 ) -> bool:
     """记录一条账号事件。返回是否真的写入（幂等命中/异常时为 False）。
 
     **绝不抛异常**：事件是旁路，不能因为记不上一条时间线就让登录或兑换失败。
     用 SAVEPOINT 包住，失败只回滚这一条，不影响调用方已经做过的事。
+
+    `occurred_at` 是**业务发生时间**，可以是 aware，也可以是 naive
+    （naive 按约定视为已是 UTC，见 `to_naive_utc`）。
+
+    `category` / `subject_*` / `actor_*` 支撑台账检索：
+    例如「该账号该商品最近一次兑换」走
+    `(account_id, event_type, subject_key, occurred_at)` 复合索引。
     """
     if not account_id or not event_type:
         return False
@@ -81,27 +119,34 @@ def emit(
     sql = text(
         """
         INSERT INTO account_events (
-            account_id, event_type, occurred_at, title,
-            amount, balance_after, detail, source, source_ref, created_at
+            account_id, event_type, category, occurred_at, title,
+            amount, balance_after, subject_type, subject_key,
+            detail, source, actor_type, actor_id, source_ref, created_at
         ) VALUES (
-            :account_id, :event_type, :occurred_at, :title,
-            :amount, :balance_after, :detail, :source, :source_ref, :created_at
+            :account_id, :event_type, :category, :occurred_at, :title,
+            :amount, :balance_after, :subject_type, :subject_key,
+            :detail, :source, :actor_type, :actor_id, :source_ref, :created_at
         )
         """
     )
     params = {
         "account_id": account_id,
         "event_type": event_type,
-        "occurred_at": (occurred_at or _now()).astimezone(UTC).replace(tzinfo=None),
+        "category": (category or "")[:16],
+        "occurred_at": to_naive_utc(occurred_at or _now()),
         "title": (title or "")[:256],
         "amount": amount,
         "balance_after": balance_after,
+        "subject_type": (subject_type or "")[:16],
+        "subject_key": (subject_key or "")[:96],
         "detail": json.dumps(detail, ensure_ascii=False, default=str)
         if detail
         else None,
         "source": source[:24],
+        "actor_type": (actor_type or "")[:16],
+        "actor_id": actor_id,
         "source_ref": (source_ref or None),
-        "created_at": _now().replace(tzinfo=None),
+        "created_at": to_naive_utc(_now()),
     }
 
     try: