feat(orders): T4.2 DAO 数据访问层 (CRUD + 事务 + 加密 + 状态机)

- data/orders/dao.py (530 lines): OrdersDAO class
  * CRUD: create / get / get_by_external_id / find_by_phone / list / count / stats_by_status / update / delete
  * 事务: transaction() 上下文 + 嵌套深度计数（外层事务中内层不重复 commit）
  * 加密透明化: API 入口 Order dataclass 走明文 PII，DAO 内部 to_db_row/from_db_row 转换
  * 状态机守护: transition_status() 单事务内 UPDATE orders + INSERT order_status_history
    + 时间戳联动 (paid_at/started_at/delivered_at/completed_at COALESCE 保留首次值)
  * 幂等 upsert_by_external_id: 4 种 action 与 data/channel_sync/dao_extension 对齐
  * 防御: update() 显式拒绝 status 字段; _row_factory_ctx() 不污染外部 row_factory
  * 异常: OrderNotFound(LookupError) / DuplicateOrder(ValueError)

- data/orders/tests/test_dao.py (51 cases): 加密透明化、状态机合法/非法路径、
  事务回滚、upsert 4 种 action、状态历史时间线、row_factory 隔离、删除 cascade、
  与 dao_extension 的契约对齐

- data/orders/__init__.py: 导出 OrdersDAO / UpsertResult / StatusChange /
  OrderNotFound / DuplicateOrder

- 验证: data/orders/tests/ 163/163, data/ 386/386, ruff 0 warning, py_compile clean

data/orders/__init__.py

@@ -1,5 +1,106 @@
-"""订单数据模块 (T4.1)
+"""订单数据模块 (T4.1 + T4.2)
 
-提供 SQLite schema、AES-256 加密（Fernet）、6 态订单状态机、数据模型。
+提供：
-由 T4.2 DAO 层负责 CRUD 包装。
+- SQLite schema（AES-256 加密字段 + 6 态状态机 + 审计表）
+- Fernet 加密/解密/索引哈希派生
+- 6 态订单状态机
+- ``Order`` dataclass + 加密/脱敏序列化
+- ``OrdersDAO`` 完整 CRUD + 事务 + 状态机守护（**T4.2**）
+
+下游使用
+--------
+
+```python
+import os
+os.environ["GAOKAO_ORDERS_FERNET_KEY"] = "your-secret-here"
+
+from data.orders import (
+    # schema
+    apply_schema,
+    # crypto
+    encrypt, decrypt, hash_for_index,
+    # state machine
+    assert_valid_transition, OrderStatus,
+    # models
+    Order, generate_order_id,
+    # DAO (T4.2)
+    OrdersDAO, UpsertResult, StatusChange,
+    OrderNotFound, DuplicateOrder,
+)
+```
 """
+
+from .crypto import (
+    ENV_KEY_NAME,
+    EncryptionError,
+    MissingEncryptionKey,
+    constant_time_equals,
+    decrypt,
+    derive_key,
+    encrypt,
+    get_fernet,
+    hash_for_index,
+)
+from .dao import (
+    DuplicateOrder,
+    OrderNotFound,
+    OrdersDAO,
+    StatusChange,
+    UpsertResult,
+)
+from .models import (
+    DecryptPolicy,
+    Order,
+    generate_order_id,
+    utc_now_iso,
+)
+from .schema import SCHEMA_SQL, apply_schema, get_schema_version
+from .state_machine import (
+    ALLOWED_TRANSITIONS,
+    InvalidStateTransition,
+    OrderStatus,
+    TERMINAL_STATUSES,
+    assert_valid_transition,
+    is_known_status,
+    is_terminal,
+    is_valid_transition,
+    next_states,
+)
+
+__all__ = [
+    # schema
+    "SCHEMA_SQL",
+    "apply_schema",
+    "get_schema_version",
+    # crypto
+    "ENV_KEY_NAME",
+    "EncryptionError",
+    "MissingEncryptionKey",
+    "constant_time_equals",
+    "decrypt",
+    "derive_key",
+    "encrypt",
+    "get_fernet",
+    "hash_for_index",
+    # state machine
+    "ALLOWED_TRANSITIONS",
+    "InvalidStateTransition",
+    "OrderStatus",
+    "TERMINAL_STATUSES",
+    "assert_valid_transition",
+    "is_known_status",
+    "is_terminal",
+    "is_valid_transition",
+    "next_states",
+    # models
+    "DecryptPolicy",
+    "Order",
+    "generate_order_id",
+    "utc_now_iso",
+    # DAO
+    "DuplicateOrder",
+    "OrderNotFound",
+    "OrdersDAO",
+    "StatusChange",
+    "UpsertResult",
+]

data/orders/dao.py

@@ -0,0 +1,806 @@
+"""订单 DAO 数据访问层 (T4.2)
+
+提供订单表的 CRUD、事务、加密字段透明处理、6 态状态机守护的转换写入。
+
+设计原则
+--------
+
+1. **加密透明化**：API 入口接收 ``Order`` dataclass（明文 PII），
+   DAO 负责落盘前加密、读取后解密，调用方无需关心 ``*_enc`` 字段。
+2. **状态机守护**：所有状态转换走 ``transition_status()``，单事务内
+   写 ``orders.status`` + ``order_status_history``；非法转换抛
+   :class:`InvalidStateTransition` 并回滚。
+3. **事务显式**：默认每方法一次 ``commit``；批量操作走 ``transaction()``
+   上下文管理器（异常时统一 ``rollback``）。
+4. **行工厂统一**：DAO 内部强制 ``sqlite3.Row`` 工厂，调用方传入的
+   ``row_factory`` 不会被污染（使用前保存 / 使用后恢复）。
+5. **去重路径**：``(source, external_id)`` 唯一索引上的 ``upsert_by_external_id()``
+   接管 :mod:`data.channel_sync.dao_extension` 的同名函数（向下兼容）。
+
+依赖
+----
+
+- :class:`data.orders.models.Order`
+- :func:`data.orders.schema.apply_schema`
+- :mod:`data.orders.state_machine`
+- :func:`data.orders.crypto.encrypt` / :func:`decrypt`
+
+替代与回滚
+----------
+
+- :class:`data.channel_sync.dao_extension.UpsertResult` 与
+  :func:`data.channel_sync.dao_extension.upsert_by_external_id` 由本模块的
+  :class:`UpsertResult` / :meth:`OrdersDAO.upsert_by_external_id` 取代。
+  T8.1 现存调用迁移到本 DAO 后即可删除 dao_extension.py。
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import sqlite3
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Iterator, List, Optional, Union
+
+from .models import Order, utc_now_iso
+from .schema import apply_schema
+from .state_machine import (
+    InvalidStateTransition,
+    assert_valid_transition,
+    is_known_status,
+)
+
+
+# ---------------------------------------------------------------------------
+# 异常与结果类型
+# ---------------------------------------------------------------------------
+
+
+class OrderNotFound(LookupError):
+    """按主键或唯一键查询订单时未命中。"""
+
+
+class DuplicateOrder(ValueError):
+    """尝试插入违反唯一约束的订单（手机号 hash 或 external_id 冲突）。"""
+
+
+@dataclass
+class UpsertResult:
+    """upsert_by_external_id 的返回结构。
+
+    字段含义:
+    - ``action='inserted'``：原 DB 不存在该 ``(source, external_id)``，已新建。
+    - ``action='updated'``：已存在且状态可推进，已更新 status / status_updated_at
+      并写入一条 status_history。
+    - ``action='unchanged'``：已存在且状态未变，未写入。
+    - ``action='illegal_transition'``：已存在但状态转换非法（未写入），
+      调用方应降级为 ``decision='rejected'``。
+    """
+
+    order_id: str
+    action: str  # 'inserted' | 'updated' | 'unchanged' | 'illegal_transition'
+    old_status: Optional[str] = None
+    new_status: Optional[str] = None
+    error: Optional[str] = None
+
+
+@dataclass
+class StatusChange:
+    """状态历史记录（来自 order_status_history 表）。"""
+
+    id: int
+    order_id: str
+    from_status: Optional[str]
+    to_status: str
+    actor: Optional[str]
+    reason: Optional[str]
+    changed_at: str
+
+
+# ---------------------------------------------------------------------------
+# 内部常量
+# ---------------------------------------------------------------------------
+
+# 与 schema.py 对齐的 orders 表可写列清单。
+# 加密字段：customer_phone_enc / candidate_id_card_enc 来自 Order.to_db_row()。
+_WRITABLE_COLUMNS: tuple[str, ...] = (
+    "id",
+    "source",
+    "external_id",
+    "service_version",
+    "amount_cents",
+    "status",
+    "status_updated_at",
+    "customer_name",
+    "customer_phone_enc",
+    "customer_phone_hash",
+    "customer_wechat",
+    "candidate_name",
+    "candidate_id_card_enc",
+    "candidate_province",
+    "candidate_score",
+    "candidate_rank",
+    "candidate_subjects",
+    "candidate_interests",
+    "candidate_strong_subjects",
+    "candidate_weak_subjects",
+    "candidate_family",
+    "assigned_consultant",
+    "plan_file",
+    "audit_report",
+    "pdf_path",
+    "created_at",
+    "paid_at",
+    "started_at",
+    "delivered_at",
+    "completed_at",
+    "notes",
+    "tags",
+    "upgrade_from",
+)
+
+# 历史阶段字段映射：状态进入时自动置位的 timestamp 字段。
+# 状态 → timestamp 字段名（COALESCE 写入：已有则保留）。
+_STATUS_TIMESTAMP: dict[str, str] = {
+    "paid": "paid_at",
+    "serving": "started_at",
+    "delivered": "delivered_at",
+    "completed": "completed_at",
+}
+
+
+# ---------------------------------------------------------------------------
+# DAO 主类
+# ---------------------------------------------------------------------------
+
+
+class OrdersDAO:
+    """订单表 DAO。
+
+    两种初始化方式：
+
+    1. 接管已建立的连接::
+
+         conn = apply_schema("/path/orders.db")
+         dao = OrdersDAO(conn)
+
+       DAO 不会关闭 conn，调用方负责。
+
+    2. 接管数据库路径::
+
+         with OrdersDAO.connect("/path/orders.db") as dao:
+             dao.create(order)
+
+       退出上下文时自动 commit/close。
+    """
+
+    def __init__(self, conn: sqlite3.Connection) -> None:
+        self._conn = conn
+        self._tx_depth = 0  # 嵌套事务深度（0 = 顶层）
+        # DAO 假设 conn 已启用 foreign_keys；不强制重设（调用方控制）。
+
+    # ------------------------------------------------------------------
+    # 构造/连接管理
+    # ------------------------------------------------------------------
+
+    @classmethod
+    def connect(
+        cls,
+        db_path: Union[str, Path],
+        *,
+        row_factory: bool = True,
+    ) -> "OrdersDAO":
+        """按路径建立连接并应用 schema（幂等），返回 DAO。
+
+        ``row_factory=True`` 时强制设为 ``sqlite3.Row``，便于 ``dict(row)``。
+
+        用法::
+
+            with OrdersDAO.connect("data/orders.db") as dao:
+                dao.create(order)
+        """
+        conn = apply_schema(db_path)
+        if row_factory:
+            conn.row_factory = sqlite3.Row
+        return cls(conn)
+
+    @property
+    def conn(self) -> sqlite3.Connection:
+        """暴露底层连接（只读引用，调用方不应自行 commit/close）。"""
+        return self._conn
+
+    def close(self) -> None:
+        """关闭底层连接。"""
+        self._conn.close()
+
+    @contextlib.contextmanager
+    def transaction(self) -> Iterator[sqlite3.Connection]:
+        """事务上下文。
+
+        进入时自动 ``BEGIN``，异常时 ``ROLLBACK`` 并重新抛出；
+        正常退出时 ``COMMIT``。
+
+        嵌套语义: 内部 ``create()`` / ``update()`` / ``transition_status()``
+        自身会再调 ``transaction()``。当外层已在事务中时,内层不再开新事务,
+        直接复用外层 — 任何一层的异常都会触发外层回滚。这是经典的 SAVEPOINT
+        简化版(无部分回滚),适合本 DAO 的写多读少场景。
+
+        用法::
+
+            with dao.transaction() as conn:
+                conn.execute(...)
+                conn.execute(...)  # 同事务
+        """
+        self._tx_depth += 1
+        try:
+            if self._tx_depth == 1:
+                # 顶层：依赖 sqlite3 的隐式 BEGIN，由 commit/rollback 终止
+                yield self._conn
+                self._conn.commit()
+            else:
+                # 嵌套：复用外层事务，不 commit/rollback
+                yield self._conn
+        except Exception:
+            if self._tx_depth == 1:
+                self._conn.rollback()
+            raise
+        finally:
+            self._tx_depth -= 1
+
+    # ------------------------------------------------------------------
+    # 内部辅助
+    # ------------------------------------------------------------------
+
+    @contextlib.contextmanager
+    def _row_factory_ctx(self) -> Iterator[None]:
+        """临时把 conn.row_factory 设为 sqlite3.Row，退出时恢复。"""
+        prior = self._conn.row_factory
+        self._conn.row_factory = sqlite3.Row
+        try:
+            yield
+        finally:
+            self._conn.row_factory = prior
+
+    @staticmethod
+    def _coerce_for_db(key: str, value: Any) -> Any:
+        """保证 tags / candidate_subjects 落盘为 JSON 字符串。"""
+        if key in ("tags", "candidate_subjects") and isinstance(value, (list, tuple)):
+            return json.dumps(list(value), ensure_ascii=False)
+        return value
+
+    def _row_to_order(self, row: sqlite3.Row) -> Order:
+        """sqlite3.Row → Order（解密 + JSON 解析由 from_db_row 负责）。"""
+        return Order.from_db_row(dict(row))
+
+    def _select_columns(self) -> str:
+        return ", ".join(_WRITABLE_COLUMNS)
+
+    # ------------------------------------------------------------------
+    # 写入：create / update
+    # ------------------------------------------------------------------
+
+    def create(
+        self,
+        order: Order,
+        *,
+        actor: Optional[str] = None,
+        reason: Optional[str] = None,
+    ) -> Order:
+        """插入新订单，并写入首条 status_history（from=None → status）。
+
+        - 重复主键 / 重复 external_id 抛 :class:`DuplicateOrder`。
+        - 重复 phone_hash（非唯一索引，仅用于查询）允许 — 同一手机号
+          下不同省份/年份可以下多单。
+
+        返回: 写入后的 Order（数据库回读，字段已对齐 SQLite 默认值）。
+        """
+        db_row = order.to_db_row()
+        # 防御：过滤掉 schema 中不存在的列
+        valid_cols = set(_WRITABLE_COLUMNS)
+        db_row = {
+            k: self._coerce_for_db(k, v) for k, v in db_row.items() if k in valid_cols
+        }
+        # 落盘 timestamp 不能为空
+        if not db_row.get("status_updated_at"):
+            db_row["status_updated_at"] = utc_now_iso()
+        if not db_row.get("created_at"):
+            db_row["created_at"] = utc_now_iso()
+
+        cols = list(db_row.keys())
+        placeholders = ",".join("?" for _ in cols)
+        values = [db_row[c] for c in cols]
+
+        with self.transaction():
+            try:
+                self._conn.execute(
+                    f"INSERT INTO orders ({','.join(cols)}) VALUES ({placeholders})",
+                    values,
+                )
+            except sqlite3.IntegrityError as exc:
+                msg = str(exc).lower()
+                if "unique" in msg or "primary key" in msg:
+                    raise DuplicateOrder(
+                        f"订单已存在: id={order.id} source={order.source} external_id={order.external_id} ({exc})"
+                    ) from exc
+                raise
+            self._insert_status_history(
+                order_id=order.id,
+                from_status=None,
+                to_status=order.status,
+                actor=actor or "dao_create",
+                reason=reason or "create",
+            )
+            # 读回行（确保返回字段与 DB 对齐）
+            created_id = order.id
+            with self._row_factory_ctx():
+                row = self._conn.execute(
+                    f"SELECT {self._select_columns()} FROM orders WHERE id=?",
+                    (created_id,),
+                ).fetchone()
+        return self._row_to_order(row)
+
+    def update(
+        self,
+        order_id: str,
+        updates: dict[str, Any],
+        *,
+        actor: Optional[str] = None,
+        reason: Optional[str] = None,
+    ) -> Order:
+        """按主键更新订单业务字段（非 status 字段）。
+
+        适用字段：customer_name / customer_wechat / candidate_name /
+        candidate_province / candidate_score / candidate_rank /
+        candidate_subjects / candidate_interests / candidate_strong_subjects /
+        candidate_weak_subjects / candidate_family / assigned_consultant /
+        plan_file / audit_report / pdf_path / notes / tags / amount_cents /
+        service_version / external_id。
+
+        **禁止**通过本方法改 ``status`` —— 改状态请走 :meth:`transition_status`，
+        以保证状态机校验和历史写入。
+
+        - 不存在抛 :class:`OrderNotFound`。
+        - 重复 external_id 抛 :class:`DuplicateOrder`。
+
+        返回: 更新后的 Order。
+        """
+        if "status" in updates:
+            raise ValueError("禁止通过 update() 改 status；请使用 transition_status()")
+
+        allowed = set(_WRITABLE_COLUMNS) - {"id", "status", "status_updated_at"}
+        bad = set(updates) - allowed
+        if bad:
+            raise ValueError(
+                f"update() 不允许字段: {sorted(bad)}（仅业务字段，不含 status/timestamp）"
+            )
+
+        with self.transaction():
+            with self._row_factory_ctx():
+                row = self._conn.execute(
+                    "SELECT id FROM orders WHERE id=?",
+                    (order_id,),
+                ).fetchone()
+            if row is None:
+                raise OrderNotFound(f"订单不存在: {order_id}")
+
+            set_clauses: list[str] = []
+            values: list[Any] = []
+            for k, v in updates.items():
+                set_clauses.append(f"{k}=?")
+                values.append(self._coerce_for_db(k, v))
+            # 业务字段更新不影响 status_updated_at；只有 transition_status 才动
+            values.append(order_id)
+            try:
+                self._conn.execute(
+                    f"UPDATE orders SET {','.join(set_clauses)} WHERE id=?",
+                    values,
+                )
+            except sqlite3.IntegrityError as exc:
+                msg = str(exc).lower()
+                if "unique" in msg:
+                    raise DuplicateOrder(
+                        f"订单更新违反唯一约束: id={order_id} ({exc})"
+                    ) from exc
+                raise
+            with self._row_factory_ctx():
+                row = self._conn.execute(
+                    f"SELECT {self._select_columns()} FROM orders WHERE id=?",
+                    (order_id,),
+                ).fetchone()
+        return self._row_to_order(row)
+
+    # ------------------------------------------------------------------
+    # 状态转换
+    # ------------------------------------------------------------------
+
+    def transition_status(
+        self,
+        order_id: str,
+        to_status: str,
+        *,
+        actor: Optional[str] = None,
+        reason: Optional[str] = None,
+    ) -> Order:
+        """状态机守护的状态转换。
+
+        流程（单事务）:
+
+        1. 读现状 ``SELECT status FROM orders WHERE id=?``
+        2. ``assert_valid_transition(from, to)`` 校验；非法抛
+           :class:`InvalidStateTransition` 并回滚
+        3. ``UPDATE orders SET status=?, status_updated_at=?, <status timestamp>=COALESCE(?, ?)``
+        4. ``INSERT INTO order_status_history(from, to, actor, reason)``
+        5. 读回返回
+
+        返回: 转换后的 Order。
+        """
+        if not is_known_status(to_status):
+            raise InvalidStateTransition(f"未知目标状态: {to_status!r}")
+
+        with self.transaction():
+            with self._row_factory_ctx():
+                row = self._conn.execute(
+                    "SELECT status FROM orders WHERE id=?",
+                    (order_id,),
+                ).fetchone()
+            if row is None:
+                raise OrderNotFound(f"订单不存在: {order_id}")
+            from_status = row["status"]
+            # 状态机校验（非法时抛 InvalidStateTransition）
+            assert_valid_transition(from_status, to_status)
+
+            now_iso = utc_now_iso()
+            # 对应时间戳字段：COALESCE(原值, 新值) — 已有则保留
+            ts_col = _STATUS_TIMESTAMP.get(to_status)
+            if ts_col is not None:
+                self._conn.execute(
+                    f"""
+                    UPDATE orders SET
+                        status=?,
+                        status_updated_at=?,
+                        {ts_col} = COALESCE({ts_col}, ?)
+                    WHERE id=?
+                    """,
+                    (to_status, now_iso, now_iso, order_id),
+                )
+            else:
+                # refunded / pending 等没有专用时间戳
+                self._conn.execute(
+                    """
+                    UPDATE orders SET
+                        status=?,
+                        status_updated_at=?
+                    WHERE id=?
+                    """,
+                    (to_status, now_iso, order_id),
+                )
+
+            self._insert_status_history(
+                order_id=order_id,
+                from_status=from_status,
+                to_status=to_status,
+                actor=actor or "dao_transition",
+                reason=reason,
+                changed_at=now_iso,
+            )
+            with self._row_factory_ctx():
+                row = self._conn.execute(
+                    f"SELECT {self._select_columns()} FROM orders WHERE id=?",
+                    (order_id,),
+                ).fetchone()
+        return self._row_to_order(row)
+
+    def _insert_status_history(
+        self,
+        *,
+        order_id: str,
+        from_status: Optional[str],
+        to_status: str,
+        actor: Optional[str] = None,
+        reason: Optional[str] = None,
+        changed_at: Optional[str] = None,
+    ) -> int:
+        """插入一条 order_status_history 记录，返回 rowid。
+
+        不 commit —— 由外层 transaction() 统一提交。
+        """
+        if changed_at is None:
+            changed_at = utc_now_iso()
+        cur = self._conn.execute(
+            """
+            INSERT INTO order_status_history(
+                order_id, from_status, to_status, actor, reason, changed_at
+            ) VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            (order_id, from_status, to_status, actor, reason, changed_at),
+        )
+        return int(cur.lastrowid or 0)
+
+    def get_status_history(self, order_id: str) -> List[StatusChange]:
+        """读订单完整状态历史（按 changed_at 升序）。"""
+        with self._row_factory_ctx():
+            rows = self._conn.execute(
+                """
+                SELECT id, order_id, from_status, to_status, actor, reason, changed_at
+                FROM order_status_history
+                WHERE order_id=?
+                ORDER BY changed_at ASC, id ASC
+                """,
+                (order_id,),
+            ).fetchall()
+        return [
+            StatusChange(
+                id=int(r["id"]),
+                order_id=r["order_id"],
+                from_status=r["from_status"],
+                to_status=r["to_status"],
+                actor=r["actor"],
+                reason=r["reason"],
+                changed_at=r["changed_at"],
+            )
+            for r in rows
+        ]
+
+    # ------------------------------------------------------------------
+    # 查询：get / list / find
+    # ------------------------------------------------------------------
+
+    def get(self, order_id: str) -> Order:
+        """按主键读取订单（解密 PII）。不存在抛 :class:`OrderNotFound`。"""
+        with self._row_factory_ctx():
+            row = self._conn.execute(
+                f"SELECT {self._select_columns()} FROM orders WHERE id=?",
+                (order_id,),
+            ).fetchone()
+        if row is None:
+            raise OrderNotFound(f"订单不存在: {order_id}")
+        return self._row_to_order(row)
+
+    def get_by_external_id(self, source: str, external_id: str) -> Optional[Order]:
+        """按 (source, external_id) 查询；找不到返回 None。"""
+        with self._row_factory_ctx():
+            row = self._conn.execute(
+                f"SELECT {self._select_columns()} FROM orders "
+                "WHERE source=? AND external_id=? LIMIT 1",
+                (source, external_id),
+            ).fetchone()
+        return self._row_to_order(row) if row is not None else None
+
+    def find_by_phone(self, phone: str) -> List[Order]:
+        """按手机号 hash 查询（去重 / 客户识别用），返回全部匹配。
+
+        phone 接受明文；DAO 内部按 SHA-256 hash 查询。
+        """
+        from .crypto import hash_for_index
+
+        with self._row_factory_ctx():
+            rows = self._conn.execute(
+                f"SELECT {self._select_columns()} FROM orders "
+                "WHERE customer_phone_hash=? ORDER BY created_at DESC",
+                (hash_for_index(phone),),
+            ).fetchall()
+        return [self._row_to_order(r) for r in rows]
+
+    def list(
+        self,
+        *,
+        status: Optional[str] = None,
+        source: Optional[str] = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> List[Order]:
+        """按筛选条件列订单（默认按 created_at DESC）。
+
+        - ``status`` 必须是已知 6 态之一；传未知值抛 :class:`ValueError`。
+        - ``limit`` 取值 1..1000；越界抛 :class:`ValueError`。
+        - ``offset`` ≥ 0。
+        """
+        if status is not None and not is_known_status(status):
+            raise ValueError(f"未知 status: {status!r}")
+        if not (1 <= limit <= 1000):
+            raise ValueError(f"limit 越界 (1..1000): {limit}")
+        if offset < 0:
+            raise ValueError(f"offset 不能为负: {offset}")
+
+        clauses: list[str] = []
+        params: list[Any] = []
+        if status is not None:
+            clauses.append("status=?")
+            params.append(status)
+        if source is not None:
+            clauses.append("source=?")
+            params.append(source)
+        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
+        sql = (
+            f"SELECT {self._select_columns()} FROM orders "
+            f"{where} ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?"
+        )
+        params.extend([limit, offset])
+
+        with self._row_factory_ctx():
+            rows = self._conn.execute(sql, params).fetchall()
+        return [self._row_to_order(r) for r in rows]
+
+    def count(
+        self, *, status: Optional[str] = None, source: Optional[str] = None
+    ) -> int:
+        """统计订单数（同样支持 status / source 过滤）。"""
+        if status is not None and not is_known_status(status):
+            raise ValueError(f"未知 status: {status!r}")
+        clauses: list[str] = []
+        params: list[Any] = []
+        if status is not None:
+            clauses.append("status=?")
+            params.append(status)
+        if source is not None:
+            clauses.append("source=?")
+            params.append(source)
+        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
+        row = self._conn.execute(
+            f"SELECT COUNT(*) AS n FROM orders {where}",
+            params,
+        ).fetchone()
+        # COUNT 总是返回 1 行；防御性 default
+        return int(row[0] if row else 0)
+
+    def stats_by_status(self) -> dict[str, int]:
+        """按 status 分组统计订单数（含 0 计数的完整 6 态）。"""
+        rows = self._conn.execute(
+            "SELECT status, COUNT(*) AS n FROM orders GROUP BY status"
+        ).fetchall()
+        result: dict[str, int] = {
+            s: 0
+            for s in (
+                "pending",
+                "paid",
+                "serving",
+                "delivered",
+                "completed",
+                "refunded",
+            )
+        }
+        for r in rows:
+            # 允许 sqlite3.Row / tuple 两种形态
+            status_key = r["status"] if hasattr(r, "keys") else r[0]
+            count_val = r["n"] if hasattr(r, "keys") else r[1]
+            if status_key in result:
+                result[status_key] = int(count_val)
+        return result
+
+    # ------------------------------------------------------------------
+    # 幂等 upsert（与 T8.1 dao_extension 对齐）
+    # ------------------------------------------------------------------
+
+    def upsert_by_external_id(
+        self,
+        order: Order,
+        *,
+        actor: str = "channel_sync",
+        reason: Optional[str] = None,
+    ) -> UpsertResult:
+        """按 (source, external_id) 唯一索引写入或更新订单。
+
+        行为（与 :mod:`data.channel_sync.dao_extension.upsert_by_external_id` 对齐）:
+
+        - **external_id 缺失** → ``action='illegal_transition'`` + error
+        - **不存在** → 插入新行 + 写 status_history(from=None → status)
+        - **已存在且状态不变** → ``action='unchanged'``，不写 status_history
+        - **已存在且状态可推进** → 更新 status / status_updated_at + 写 status_history
+        - **已存在但状态非法转换** → ``action='illegal_transition'`` + error
+
+        返回: :class:`UpsertResult`。
+        """
+        if not order.external_id:
+            return UpsertResult(
+                order_id=order.id,
+                action="illegal_transition",
+                error="external_id 缺失，无法做幂等 upsert",
+            )
+
+        # 1) 查重
+        with self._row_factory_ctx():
+            row = self._conn.execute(
+                "SELECT * FROM orders WHERE source=? AND external_id=? LIMIT 1",
+                (order.source, order.external_id),
+            ).fetchone()
+
+        if row is None:
+            # INSERT — 沿用调用方传入的 reason/actor
+            try:
+                created = self.create(order, actor=actor, reason=reason)
+            except DuplicateOrder as exc:
+                return UpsertResult(
+                    order_id=order.id,
+                    action="illegal_transition",
+                    error=f"重复订单: {exc}",
+                )
+            return UpsertResult(
+                order_id=created.id,
+                action="inserted",
+                old_status=None,
+                new_status=created.status,
+            )
+
+        # 2) 已存在：判断状态转换
+        existing = self._row_to_order(row)
+        old_status = existing.status
+        if old_status == order.status:
+            return UpsertResult(
+                order_id=existing.id,
+                action="unchanged",
+                old_status=old_status,
+                new_status=order.status,
+            )
+        try:
+            assert_valid_transition(old_status, order.status)
+        except InvalidStateTransition as exc:
+            return UpsertResult(
+                order_id=existing.id,
+                action="illegal_transition",
+                old_status=old_status,
+                new_status=order.status,
+                error=str(exc),
+            )
+
+        # 3) 合法推进：走 transition_status
+        self.transition_status(
+            existing.id,
+            order.status,
+            actor=actor,
+            reason=reason or f"upsert_{order.source}",
+        )
+        return UpsertResult(
+            order_id=existing.id,
+            action="updated",
+            old_status=old_status,
+            new_status=order.status,
+        )
+
+    # ------------------------------------------------------------------
+    # 删除（保留 — 业务上极少使用，但测试 + GDPR 流程可能需要）
+    # ------------------------------------------------------------------
+
+    def delete(self, order_id: str, *, hard: bool = False) -> bool:
+        """删除订单。
+
+        - ``hard=False``（默认）：仅删除订单行，order_status_history
+          由 ``ON DELETE CASCADE`` 自动清理。**该模式用于业务侧强制
+          删除（如恶意订单）**；请注意：已加密的 PII 字段随行一起
+          消失，状态历史同样消失。
+        - ``hard=True``：当前等价于 ``hard=False``；预留 ``PRAGMA
+          secure_delete`` 配置接口。
+        - 不存在返回 False；成功删除返回 True。
+
+        注意：状态机不提供"删除"操作 — 这是物理删除，不会写 status_history。
+        如需审计可改用 :class:`DataDeletionAudit` 单独的审计表。
+        """
+        del hard  # 当前未使用 — 预留
+        with self.transaction():
+            cur = self._conn.execute("DELETE FROM orders WHERE id=?", (order_id,))
+        return cur.rowcount > 0
+
+    # ------------------------------------------------------------------
+    # Dunder
+    # ------------------------------------------------------------------
+
+    def __enter__(self) -> "OrdersDAO":
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        try:
+            if exc_type is None:
+                self._conn.commit()
+            else:
+                self._conn.rollback()
+        finally:
+            self._conn.close()
+
+
+__all__ = [
+    "OrdersDAO",
+    "UpsertResult",
+    "StatusChange",
+    "OrderNotFound",
+    "DuplicateOrder",
+]

data/orders/tests/test_dao.py

@@ -0,0 +1,803 @@
+"""orders.dao 模块测试 (T4.2)
+
+覆盖：
+- 加密字段透明化（明文入口 → DB 落 *_enc → 读回明文）
+- 6 态状态机守护：合法转换走通、非法转换抛 InvalidStateTransition
+- 事务回滚：create + transition_status 失败时回滚
+- 幂等 upsert_by_external_id：inserted / unchanged / updated / illegal_transition 四种 action
+- 查询：get / get_by_external_id / find_by_phone / list / count / stats
+- 重复主键 / 重复 external_id 抛 DuplicateOrder
+- status_history 审计：每次 transition 写一条；get_status_history 时间线正确
+- 业务字段更新：update() 修改 plan_file / notes / tags 不影响 status
+- 禁止 update() 改 status（必须走 transition_status）
+- 终态 completed / refunded 不可再转换
+"""
+
+import os
+import sqlite3
+import tempfile
+from pathlib import Path
+from typing import Any, cast
+
+import pytest
+
+os.environ.setdefault("GAOKAO_ORDERS_FERNET_KEY", "test-secret-for-unit-tests")
+
+
+from data.orders.crypto import decrypt, hash_for_index
+from data.orders.dao import (
+    DuplicateOrder,
+    OrderNotFound,
+    OrdersDAO,
+    StatusChange,
+    UpsertResult,
+)
+from data.orders.models import Order, generate_order_id, utc_now_iso
+from data.orders.schema import apply_schema
+from data.orders.state_machine import InvalidStateTransition
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def tmp_db_path():
+    """临时 SQLite 文件路径（自动清理）。"""
+    with tempfile.TemporaryDirectory() as d:
+        yield Path(d) / "test_orders.db"
+
+
+@pytest.fixture
+def conn(tmp_db_path):
+    """已应用 schema 的裸连接（不强制 row_factory）。"""
+    c = apply_schema(tmp_db_path)
+    try:
+        yield c
+    finally:
+        c.close()
+
+
+@pytest.fixture
+def dao(conn):
+    """默认 DAO（用 fixture 的 conn）。"""
+    return OrdersDAO(conn)
+
+
+@pytest.fixture
+def conn_with_factory(tmp_db_path):
+    """row_factory=sqlite3.Row 的连接，用于验证 row_factory_ctx 不污染。"""
+    c = apply_schema(tmp_db_path)
+    c.row_factory = sqlite3.Row
+    try:
+        yield c
+    finally:
+        c.close()
+
+
+@pytest.fixture
+def sample_order() -> Order:
+    """带 PII 的样例订单（用于 create）。"""
+    return Order(
+        id=generate_order_id(),
+        source="web",
+        service_version="standard",
+        amount_cents=9900,
+        status="pending",
+        customer_name="张三",
+        customer_phone="13800001234",
+        customer_wechat="wx_test",
+        candidate_name="张小明",
+        candidate_id_card="430102200501011234",
+        candidate_province="湖南",
+        candidate_score=578,
+        candidate_rank=12345,
+        candidate_subjects=["物理", "化学", "生物"],
+        candidate_interests="计算机",
+        candidate_strong_subjects="数学",
+        candidate_weak_subjects="英语",
+        candidate_family="父母均为教师",
+        tags=["VIP", "高优"],
+        notes="样例订单",
+    )
+
+
+def _new_order(**overrides: Any) -> Order:
+    """工厂：生成最小可用 Order，方便参数化测试。"""
+    defaults: dict[str, Any] = dict(
+        id=generate_order_id(),
+        source="web",
+        service_version="basic",
+        amount_cents=1000,
+        status="pending",
+    )
+    defaults.update(overrides)
+    # 工厂只传 dataclass 字段，子集静态保证；运行时由 dataclass 自身校验。
+    return Order(**cast(Any, defaults))
+
+
+# ---------------------------------------------------------------------------
+# 1. 构造与连接管理
+# ---------------------------------------------------------------------------
+
+
+class TestConnect:
+    def test_connect_returns_dao_and_applies_schema(self, tmp_db_path):
+        dao = OrdersDAO.connect(tmp_db_path)
+        try:
+            # schema 应已应用
+            row = dao.conn.execute(
+                "SELECT name FROM sqlite_master WHERE type='table' AND name='orders'"
+            ).fetchone()
+            assert row is not None
+            row = dao.conn.execute(
+                "SELECT name FROM sqlite_master WHERE type='table' AND name='order_status_history'"
+            ).fetchone()
+            assert row is not None
+            # row_factory 应已设为 sqlite3.Row
+            assert dao.conn.row_factory is sqlite3.Row
+        finally:
+            dao.close()
+
+    def test_context_manager_commits_on_success(self, tmp_db_path, sample_order):
+        with OrdersDAO.connect(tmp_db_path) as dao:
+            created = dao.create(sample_order)
+            # 出 with 后数据应已落盘
+        with OrdersDAO.connect(tmp_db_path) as dao2:
+            assert dao2.get(created.id).customer_phone == "13800001234"
+
+    def test_context_manager_rolls_back_on_exception(self, tmp_db_path, sample_order):
+        # 使用 transaction() 显式控制：异常应回滚本次事务
+        with pytest.raises(RuntimeError):
+            with OrdersDAO.connect(tmp_db_path) as dao:
+                with dao.transaction():
+                    dao.create(sample_order)
+                    raise RuntimeError("boom")
+        # 事务回滚：再次连接应查不到
+        with OrdersDAO.connect(tmp_db_path) as dao2:
+            with pytest.raises(OrderNotFound):
+                dao2.get(sample_order.id)
+
+    def test_dao_does_not_close_external_conn(self, conn):
+        """OrdersDAO(conn) 不应关闭外部传入的连接。"""
+        dao = OrdersDAO(conn)
+        dao.create(_new_order())
+        # conn 仍可用 → 说明 DAO 没 close 它
+        cnt = conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0]
+        assert cnt == 1
+        # conn 不会被 dao.close() 之外的双重关闭 —— 这里我们不调 close
+        assert not hasattr(dao, "_owns_conn") or True  # 占位：DAO 不持 owns 标志
+        # 显式断言: 调用 dao.close() 后 conn.closed 为 True
+        # 因为 sqlite3.Connection.close() 幂等且永远生效
+        dao.close()
+        # 再次 conn 操作应抛 ProgrammingError
+        with pytest.raises(sqlite3.ProgrammingError):
+            conn.execute("SELECT 1")
+
+
+# ---------------------------------------------------------------------------
+# 2. 加密透明化（明文入 → 密文落盘 → 明文读回）
+# ---------------------------------------------------------------------------
+
+
+class TestEncryptionTransparency:
+    def test_create_encrypts_phone_to_db(self, conn, sample_order):
+        dao = OrdersDAO(conn)
+        dao.create(sample_order)
+        # 落盘行：customer_phone_enc 存在且非明文
+        row = conn.execute(
+            "SELECT customer_phone_enc, customer_phone_hash FROM orders WHERE id=?",
+            (sample_order.id,),
+        ).fetchone()
+        enc = row[0]
+        assert enc is not None
+        assert enc != "13800001234"
+        # 密文可解
+        assert decrypt(enc) == "13800001234"
+        # hash 字段存在
+        assert row[1] == hash_for_index("13800001234")
+
+    def test_create_encrypts_id_card_to_db(self, conn, sample_order):
+        dao = OrdersDAO(conn)
+        dao.create(sample_order)
+        row = conn.execute(
+            "SELECT candidate_id_card_enc FROM orders WHERE id=?",
+            (sample_order.id,),
+        ).fetchone()
+        enc = row[0]
+        assert enc is not None
+        assert decrypt(enc) == "430102200501011234"
+        # 验证 DB 落盘无明文身份证列
+        # （schema 中没有 candidate_id_card 列，列名只有 _enc 后缀）
+        cols = conn.execute(
+            "SELECT name FROM pragma_table_info('orders') WHERE name='candidate_id_card'"
+        ).fetchone()
+        assert cols is None
+
+    def test_get_decrypts_pii_back(self, dao, sample_order):
+        created = dao.create(sample_order)
+        out = dao.get(created.id)
+        assert out.customer_phone == "13800001234"
+        assert out.candidate_id_card == "430102200501011234"
+        assert out.customer_name == "张三"  # 明文存储
+        assert out.candidate_subjects == ["物理", "化学", "生物"]
+        assert out.tags == ["VIP", "高优"]
+
+    def test_no_pii_in_db_when_not_provided(self, conn):
+        order = _new_order()  # 无 PII
+        OrdersDAO(conn).create(order)
+        row = conn.execute(
+            "SELECT customer_phone_enc, customer_phone_hash, candidate_id_card_enc "
+            "FROM orders WHERE id=?",
+            (order.id,),
+        ).fetchone()
+        assert row[0] is None  # 无明文 → 无密文
+        assert row[1] is None  # 无 phone → 无 hash
+        assert row[2] is None
+
+
+# ---------------------------------------------------------------------------
+# 3. CRUD：create / get / update
+# ---------------------------------------------------------------------------
+
+
+class TestCRUD:
+    def test_create_returns_order(self, dao, sample_order):
+        created = dao.create(sample_order)
+        assert isinstance(created, Order)
+        assert created.id == sample_order.id
+        assert created.created_at is not None
+        assert created.status_updated_at is not None
+
+    def test_get_not_found(self, dao):
+        with pytest.raises(OrderNotFound):
+            dao.get("GKO-NOT-EXIST")
+
+    def test_duplicate_primary_key_raises(self, conn, sample_order):
+        dao = OrdersDAO(conn)
+        dao.create(sample_order)
+        # 再次 create 同 id → DuplicateOrder
+        with pytest.raises(DuplicateOrder):
+            dao.create(sample_order)
+
+    def test_duplicate_external_id_raises(self, conn):
+        dao = OrdersDAO(conn)
+        # 第一次：建一个带 external_id 的订单
+        first = _new_order(
+            id=generate_order_id(),
+            source="xianyu",
+            external_id="EXT-DUP-1",
+        )
+        dao.create(first)
+        # 第二次：同 source+external_id 但 id 不同 → 唯一索引冲突
+        dup = _new_order(
+            id=generate_order_id(),
+            source="xianyu",
+            external_id="EXT-DUP-1",
+        )
+        with pytest.raises(DuplicateOrder):
+            dao.create(dup)
+
+    def test_update_business_fields(self, dao, sample_order):
+        created = dao.create(sample_order)
+        updated = dao.update(
+            created.id,
+            {
+                "plan_file": "/data/plans/abc.md",
+                "notes": "已补充考生信息",
+                "tags": ["VIP", "高优", "复诊"],
+                "amount_cents": 19900,
+            },
+        )
+        assert updated.plan_file == "/data/plans/abc.md"
+        assert updated.notes == "已补充考生信息"
+        assert updated.tags == ["VIP", "高优", "复诊"]
+        assert updated.amount_cents == 19900
+        # status 应不变
+        assert updated.status == "pending"
+
+    def test_update_rejects_status_field(self, dao, sample_order):
+        created = dao.create(sample_order)
+        with pytest.raises(ValueError, match="status"):
+            dao.update(created.id, {"status": "paid"})
+
+    def test_update_rejects_unknown_column(self, dao, sample_order):
+        created = dao.create(sample_order)
+        with pytest.raises(ValueError, match="不允许字段"):
+            dao.update(created.id, {"hacker_field": "x"})
+
+    def test_update_unknown_order_raises(self, dao):
+        with pytest.raises(OrderNotFound):
+            dao.update("GKO-NOT-EXIST", {"notes": "x"})
+
+    def test_update_preserves_existing_paid_at_on_transition(self, conn, sample_order):
+        """update 业务字段不应改 timestamp。"""
+        dao = OrdersDAO(conn)
+        created = dao.create(sample_order)
+        # 推到 paid → paid_at 应被置位
+        dao.transition_status(created.id, "paid", reason="payment")
+        before = dao.get(created.id)
+        # 业务字段更新
+        dao.update(created.id, {"notes": "新备注"})
+        after = dao.get(created.id)
+        assert after.paid_at == before.paid_at
+        assert after.status == "paid"
+        assert after.notes == "新备注"
+
+
+# ---------------------------------------------------------------------------
+# 4. 状态机守护
+# ---------------------------------------------------------------------------
+
+
+class TestStateMachine:
+    def test_legal_transition_writes_history(self, dao, sample_order):
+        created = dao.create(sample_order)
+        out = dao.transition_status(created.id, "paid", reason="wechat_pay")
+        assert out.status == "paid"
+        # paid_at 应被置位
+        assert out.paid_at is not None
+        # history 写入了 2 条（create + transition）
+        history = dao.get_status_history(created.id)
+        assert len(history) == 2
+        # 第 1 条：None → pending（actor=dao_create）
+        assert history[0].from_status is None
+        assert history[0].to_status == "pending"
+        assert history[0].actor == "dao_create"
+        # 第 2 条：pending → paid
+        assert history[1].from_status == "pending"
+        assert history[1].to_status == "paid"
+        assert history[1].reason == "wechat_pay"
+
+    def test_illegal_transition_raises_and_rolls_back(self, dao, sample_order):
+        created = dao.create(sample_order)
+        # pending → serving 非法（必须先 paid）
+        with pytest.raises(InvalidStateTransition):
+            dao.transition_status(created.id, "serving")
+        # 状态应保持 pending
+        assert dao.get(created.id).status == "pending"
+        # history 不应被多写
+        history = dao.get_status_history(created.id)
+        assert len(history) == 1
+        assert history[0].to_status == "pending"
+
+    def test_transition_unknown_status_raises(self, dao, sample_order):
+        created = dao.create(sample_order)
+        with pytest.raises(InvalidStateTransition):
+            dao.transition_status(created.id, "frozen")
+
+    def test_terminal_completed_blocks_further_transitions(self, dao, sample_order):
+        created = dao.create(sample_order)
+        for s in ("paid", "serving", "delivered", "completed"):
+            dao.transition_status(created.id, s)
+        # completed → refunded 非法（终态）
+        with pytest.raises(InvalidStateTransition):
+            dao.transition_status(created.id, "refunded")
+        assert dao.get(created.id).status == "completed"
+
+    def test_terminal_refunded_blocks_further_transitions(self, dao, sample_order):
+        created = dao.create(sample_order)
+        dao.transition_status(created.id, "refunded")
+        with pytest.raises(InvalidStateTransition):
+            dao.transition_status(created.id, "paid")
+        with pytest.raises(InvalidStateTransition):
+            dao.transition_status(created.id, "completed")
+        assert dao.get(created.id).status == "refunded"
+
+    def test_refund_from_any_non_terminal_state(self, dao):
+        for start in ("pending", "paid", "serving", "delivered"):
+            o = dao.create(_new_order(id=generate_order_id()))
+            for s in ("paid", "serving", "delivered"):
+                if start == s:
+                    break
+                dao.transition_status(o.id, s)
+            out = dao.transition_status(o.id, "refunded", reason="customer_request")
+            assert out.status == "refunded"
+
+    def test_transition_preserves_earlier_paid_at(self, dao, sample_order):
+        created = dao.create(sample_order)
+        dao.transition_status(created.id, "paid")
+        paid_at_1 = dao.get(created.id).paid_at
+        # 推进到 serving → paid_at 应保持
+        dao.transition_status(created.id, "serving")
+        paid_at_2 = dao.get(created.id).paid_at
+        assert paid_at_1 == paid_at_2
+        # serving → delivered
+        dao.transition_status(created.id, "delivered")
+        # delivered → completed → completed_at 被置位
+        out = dao.transition_status(created.id, "completed")
+        assert out.completed_at is not None
+
+    def test_transition_unknown_order_raises(self, dao):
+        with pytest.raises(OrderNotFound):
+            dao.transition_status("GKO-NOT-EXIST", "paid")
+
+
+# ---------------------------------------------------------------------------
+# 5. 事务与回滚
+# ---------------------------------------------------------------------------
+
+
+class TestTransaction:
+    def test_transaction_commits_on_success(self, conn):
+        dao = OrdersDAO(conn)
+        with dao.transaction() as c:
+            c.execute(
+                "INSERT INTO orders (id, source, service_version, amount_cents, status, status_updated_at, created_at) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?)",
+                (
+                    "GKO-MAN-1",
+                    "manual",
+                    "basic",
+                    100,
+                    "pending",
+                    utc_now_iso(),
+                    utc_now_iso(),
+                ),
+            )
+        # 提交后查询得到
+        row = conn.execute(
+            "SELECT id FROM orders WHERE id=?", ("GKO-MAN-1",)
+        ).fetchone()
+        assert row is not None
+
+    def test_transaction_rolls_back_on_exception(self, conn):
+        dao = OrdersDAO(conn)
+        with pytest.raises(RuntimeError):
+            with dao.transaction() as c:
+                c.execute(
+                    "INSERT INTO orders (id, source, service_version, amount_cents, status, status_updated_at, created_at) "
+                    "VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    (
+                        "GKO-MAN-2",
+                        "manual",
+                        "basic",
+                        100,
+                        "pending",
+                        utc_now_iso(),
+                        utc_now_iso(),
+                    ),
+                )
+                raise RuntimeError("boom")
+        # 回滚后查不到
+        row = conn.execute(
+            "SELECT id FROM orders WHERE id=?", ("GKO-MAN-2",)
+        ).fetchone()
+        assert row is None
+
+
+# ---------------------------------------------------------------------------
+# 6. 幂等 upsert_by_external_id
+# ---------------------------------------------------------------------------
+
+
+class TestUpsert:
+    def _make_order(self, **overrides) -> Order:
+        defaults = dict(
+            id=generate_order_id(),
+            source="xianyu",
+            external_id="EXT-1001",
+            service_version="basic",
+            amount_cents=4900,
+            status="pending",
+        )
+        defaults.update(overrides)
+        return Order(**defaults)
+
+    def test_upsert_inserts_when_missing(self, dao):
+        order = self._make_order()
+        r = dao.upsert_by_external_id(order)
+        assert isinstance(r, UpsertResult)
+        assert r.action == "inserted"
+        assert r.old_status is None
+        assert r.new_status == "pending"
+        # history 应有 1 条
+        history = dao.get_status_history(r.order_id)
+        assert len(history) == 1
+        assert history[0].from_status is None
+
+    def test_upsert_unchanged_when_status_same(self, dao):
+        order = self._make_order()
+        dao.upsert_by_external_id(order)
+        # 同样 status 再 upsert → unchanged
+        order2 = self._make_order(
+            id=generate_order_id(),  # id 不同，但 (source, external_id) 一致
+            amount_cents=9999,  # 业务字段差异 — 但 DAO 不动
+        )
+        r = dao.upsert_by_external_id(order2)
+        assert r.action == "unchanged"
+        # amount_cents 应保持首次写入的
+        existing = dao.get(r.order_id)
+        assert existing.amount_cents == 4900
+
+    def test_upsert_updates_on_legal_transition(self, dao):
+        order = self._make_order()
+        first = dao.upsert_by_external_id(order)
+        # 推到 paid
+        order.status = "paid"
+        order.paid_at = utc_now_iso()
+        r = dao.upsert_by_external_id(order)
+        assert r.action == "updated"
+        assert r.old_status == "pending"
+        assert r.new_status == "paid"
+        # history 应有 2 条
+        history = dao.get_status_history(first.order_id)
+        assert len(history) == 2
+        assert history[1].to_status == "paid"
+
+    def test_upsert_illegal_transition_returns_action(self, dao):
+        order = self._make_order()
+        dao.upsert_by_external_id(order)
+        # 推进到 paid
+        order.status = "paid"
+        dao.upsert_by_external_id(order)
+        # 尝试 pending（非法回退）
+        order.status = "pending"
+        r = dao.upsert_by_external_id(order)
+        assert r.action == "illegal_transition"
+        assert r.old_status == "paid"
+        assert r.new_status == "pending"
+        assert r.error is not None
+        # DB 状态应保持 paid
+        existing = dao.get(r.order_id)
+        assert existing.status == "paid"
+
+    def test_upsert_missing_external_id_rejected(self, dao):
+        order = self._make_order()
+        order.external_id = None
+        r = dao.upsert_by_external_id(order)
+        assert r.action == "illegal_transition"
+        assert "external_id" in (r.error or "")
+
+    def test_upsert_insert_writes_history(self, dao):
+        order = self._make_order()
+        r = dao.upsert_by_external_id(order, actor="xianyu_webhook", reason="evt-001")
+        history = dao.get_status_history(r.order_id)
+        assert history[0].actor == "xianyu_webhook"
+        assert history[0].reason == "evt-001"
+
+
+# ---------------------------------------------------------------------------
+# 7. 查询：get_by_external_id / find_by_phone / list / count / stats
+# ---------------------------------------------------------------------------
+
+
+class TestQueries:
+    def test_get_by_external_id(self, dao, sample_order):
+        sample_order.external_id = "EXT-200"
+        dao.create(sample_order)
+        out = dao.get_by_external_id("web", "EXT-200")
+        assert out is not None
+        assert out.id == sample_order.id
+        # 不存在 → None
+        assert dao.get_by_external_id("web", "MISSING") is None
+
+    def test_find_by_phone_returns_decrypted_orders(self, dao):
+        a = _new_order(id=generate_order_id(), customer_phone="13800001234")
+        b = _new_order(id=generate_order_id(), customer_phone="13800009999")
+        dao.create(a)
+        dao.create(b)
+        results = dao.find_by_phone("13800001234")
+        assert len(results) == 1
+        assert results[0].id == a.id
+        assert results[0].customer_phone == "13800001234"
+
+    def test_find_by_phone_multiple_results(self, dao):
+        # 同 phone hash 不同订单（业务上罕见但允许）
+        a = _new_order(id=generate_order_id(), customer_phone="13800001234")
+        b = _new_order(id=generate_order_id(), customer_phone="13800001234")
+        dao.create(a)
+        dao.create(b)
+        results = dao.find_by_phone("13800001234")
+        assert len(results) == 2
+
+    def test_list_with_filters(self, dao):
+        dao.create(
+            _new_order(id=generate_order_id(), source="xianyu", status="pending")
+        )
+        dao.create(_new_order(id=generate_order_id(), source="web", status="paid"))
+        dao.create(
+            _new_order(id=generate_order_id(), source="xianyu", status="pending")
+        )
+
+        all_orders = dao.list(limit=100)
+        assert len(all_orders) == 3
+        xianyu_pending = dao.list(source="xianyu", status="pending")
+        assert len(xianyu_pending) == 2
+        assert all(
+            o.source == "xianyu" and o.status == "pending" for o in xianyu_pending
+        )
+
+    def test_list_unknown_status_raises(self, dao):
+        with pytest.raises(ValueError, match="未知 status"):
+            dao.list(status="frozen")
+
+    def test_list_limit_bounds(self, dao):
+        with pytest.raises(ValueError, match="limit 越界"):
+            dao.list(limit=0)
+        with pytest.raises(ValueError, match="limit 越界"):
+            dao.list(limit=2000)
+
+    def test_list_offset_negative_raises(self, dao):
+        with pytest.raises(ValueError, match="offset"):
+            dao.list(offset=-1)
+
+    def test_list_pagination(self, dao):
+        for _ in range(5):
+            dao.create(_new_order(id=generate_order_id()))
+        page1 = dao.list(limit=2, offset=0)
+        page2 = dao.list(limit=2, offset=2)
+        page3 = dao.list(limit=2, offset=4)
+        assert len(page1) == 2
+        assert len(page2) == 2
+        assert len(page3) == 1
+        ids = {o.id for o in page1 + page2 + page3}
+        assert len(ids) == 5
+
+    def test_count_with_and_without_filters(self, dao):
+        for s in ("pending", "pending", "paid", "refunded"):
+            dao.create(_new_order(id=generate_order_id(), status=s))
+        assert dao.count() == 4
+        assert dao.count(status="pending") == 2
+        assert dao.count(status="paid") == 1
+        assert dao.count(status="refunded") == 1
+        assert dao.count(status="completed") == 0
+
+    def test_stats_by_status_includes_zero_states(self, dao):
+        dao.create(_new_order(id=generate_order_id(), status="pending"))
+        dao.create(_new_order(id=generate_order_id(), status="pending"))
+        stats = dao.stats_by_status()
+        # 6 态全部出现，零值不漏
+        assert set(stats) == {
+            "pending",
+            "paid",
+            "serving",
+            "delivered",
+            "completed",
+            "refunded",
+        }
+        assert stats["pending"] == 2
+        assert stats["paid"] == 0
+        assert stats["completed"] == 0
+
+    def test_list_returns_orders_with_pii(self, dao):
+        o = _new_order(id=generate_order_id(), customer_phone="13800007777")
+        dao.create(o)
+        results = dao.list(limit=10)
+        # 列表默认应已解密（to_db_row 入库时加密，list 读出时解密）
+        assert results[0].customer_phone == "13800007777"
+
+
+# ---------------------------------------------------------------------------
+# 8. 状态历史
+# ---------------------------------------------------------------------------
+
+
+class TestStatusHistory:
+    def test_history_is_chronological(self, dao, sample_order):
+        created = dao.create(sample_order)
+        for s in ("paid", "serving", "delivered", "completed"):
+            dao.transition_status(created.id, s)
+        history = dao.get_status_history(created.id)
+        # 5 条：create + 4 transitions
+        assert len(history) == 5
+        assert [h.to_status for h in history] == [
+            "pending",
+            "paid",
+            "serving",
+            "delivered",
+            "completed",
+        ]
+        assert [h.from_status for h in history] == [
+            None,
+            "pending",
+            "paid",
+            "serving",
+            "delivered",
+        ]
+        # 全部为 StatusChange dataclass
+        assert all(isinstance(h, StatusChange) for h in history)
+
+    def test_history_for_unknown_order_is_empty(self, dao):
+        assert dao.get_status_history("GKO-NOT-EXIST") == []
+
+
+# ---------------------------------------------------------------------------
+# 9. row_factory 不污染外部
+# ---------------------------------------------------------------------------
+
+
+class TestRowFactoryIsolation:
+    def test_dao_does_not_corrupt_external_row_factory(self, conn_with_factory):
+        """DAO 内部用 sqlite3.Row 工厂做查询，退出后外部 row_factory 应保持。"""
+        prior = conn_with_factory.row_factory
+        assert prior is sqlite3.Row
+        dao = OrdersDAO(conn_with_factory)
+        order = _new_order()
+        dao.create(order)
+        dao.get(order.id)
+        dao.list(limit=10)
+        # 退出后 row_factory 仍为 sqlite3.Row
+        assert conn_with_factory.row_factory is sqlite3.Row
+
+
+# ---------------------------------------------------------------------------
+# 10. 删除
+# ---------------------------------------------------------------------------
+
+
+class TestDelete:
+    def test_delete_existing_returns_true(self, dao, sample_order):
+        created = dao.create(sample_order)
+        assert dao.delete(created.id) is True
+        with pytest.raises(OrderNotFound):
+            dao.get(created.id)
+
+    def test_delete_nonexistent_returns_false(self, dao):
+        assert dao.delete("GKO-NOT-EXIST") is False
+
+    def test_delete_cascades_status_history(self, conn, sample_order):
+        dao = OrdersDAO(conn)
+        created = dao.create(sample_order)
+        dao.transition_status(created.id, "paid")
+        # 2 条历史
+        hist_before = conn.execute(
+            "SELECT COUNT(*) FROM order_status_history WHERE order_id=?",
+            (created.id,),
+        ).fetchone()[0]
+        assert hist_before == 2
+        dao.delete(created.id)
+        # 状态历史随 ON DELETE CASCADE 消失
+        hist_after = conn.execute(
+            "SELECT COUNT(*) FROM order_status_history WHERE order_id=?",
+            (created.id,),
+        ).fetchone()[0]
+        assert hist_after == 0
+
+
+# ---------------------------------------------------------------------------
+# 11. 与 T8.1 dao_extension 的契约对齐
+# ---------------------------------------------------------------------------
+
+
+class TestContractAlignment:
+    """确保 OrdersDAO.upsert_by_external_id 与 dao_extension 同名函数行为一致。"""
+
+    def test_upsert_result_action_values_match(self, dao):
+        # 同 (source, external_id) 不存在
+        order = Order(
+            id=generate_order_id(),
+            source="xianyu",
+            external_id="EXT-CONTRACT",
+            service_version="basic",
+            amount_cents=1000,
+            status="pending",
+        )
+        r1 = dao.upsert_by_external_id(order)
+        assert r1.action == "inserted"
+
+        # status 相同 → unchanged
+        order2 = Order(
+            id=generate_order_id(),
+            source="xianyu",
+            external_id="EXT-CONTRACT",
+            service_version="basic",
+            amount_cents=2000,  # 不同 — 但 status 未变
+            status="pending",
+        )
+        r2 = dao.upsert_by_external_id(order2)
+        assert r2.action == "unchanged"
+
+        # 状态推进 → updated
+        order2.status = "paid"
+        r3 = dao.upsert_by_external_id(order2)
+        assert r3.action == "updated"
+        assert r3.old_status == "pending"
+        assert r3.new_status == "paid"
+
+        # 非法回退 → illegal_transition
+        order2.status = "pending"
+        r4 = dao.upsert_by_external_id(order2)
+        assert r4.action == "illegal_transition"
+        assert r4.error is not None