feat(config,db): 配置优先级开关 + MySQL 自动建库；MySQL 路径实测通过

起因：想在 MySQL 上验一遍真链路，却发现做不到——项目 .env 里的
DB_DIALECT=sqlite 把我传的 DB_DIALECT=mysql 盖掉了，于是它去抢 8080 撞车。
冲突检测正确地报出来了，但「.env 一味优先」让跑变体实例变得不可能。

改动：
- DEEP_AGENT_ENV_PRIORITY = dotenv(默认) | process | none，收敛掉原来的
  DEEP_AGENT_SKIP_DOTENV（同类语义两个开关是祸根）。默认仍是 dotenv 优先，
  因为「被污染的 shell 静默改配置」比「有意的覆盖不生效」难查得多；但日志
  现在会直接告诉你该加什么前缀。
- MySQL 连接遇到 1049(Unknown database) 自动建库后再连。骨架不该要求使用者
  先手工跑一次 CREATE DATABASE——那是报错信息也不提示的一步。
- tests/conftest.py 改用 setdefault，使同一套测试可直接跑在 MySQL 上：
  DB_DIALECT=mysql DB_PORT=3399 ... python -m pytest
- scripts/http_e2e.py 支持 BASE 环境变量，便于对变体实例做专项验证。

验证（起了临时 MySQL 8.0.11 实例，独立 datadir 与端口，未触碰原有 3306）：
- pytest 143/143 在 MySQL 上全绿，143/143 在 sqlite 上全绿
- http_e2e 49/49 在 MySQL 档服务上全过
- 确认 datetime(6) 微秒精度、消息顺序、计费四种 kind、Decimal 读回均正常
- 验完已停止临时实例并删除 159MB 数据目录

.env.example

@@ -1,6 +1,17 @@
 # deep_agent 2.0 环境变量样例
 # 复制为 .env 后按需修改。所有变量均有代码内默认值（默认值 = 可跑的最小配置）。
 
+# ============================================================
+# 配置优先级
+# ============================================================
+# 本 .env 与「进程环境变量」冲突时谁生效：
+#   dotenv （默认）本文件优先，防被污染的 shell 静默改掉项目配置
+#   process        进程环境优先，用于跑变体实例（不同端口/库/方言）：
+#                  DEEP_AGENT_ENV_PRIORITY=process DB_DIALECT=mysql PORT=8081 python main.py
+#   none           完全忽略本文件，只认进程环境与代码默认值（测试/CI）
+# 无论哪一档，冲突都会打 WARNING 并出现在 /health 的 config_conflicts
+DEEP_AGENT_ENV_PRIORITY=dotenv
+
 # ============================================================
 # 模型（主 agent 与摘要模型）
 # ============================================================

AGENTS.md

@@ -76,14 +76,29 @@ deepagents 的 SummarizationMiddleware 会把被淘汰的历史**追加落盘**
 
 这些都在开发过程中真实触发过，症状与原因距离很远，所以单独记下来。
 
-### 坑 1：进程环境变量污染 `.env`
+### 坑 1：环境变量与 `.env` 的优先级
 
-机器上存在全局 `OPENAI_API_KEY` 时，`load_dotenv(override=False)` 会让
-**进程环境胜出、项目 .env 被静默忽略**。症状是「密钥明明配对了却报 401，
-且报错里显示的密钥后几位和 .env 里的不一样」。
+机器上存在全局 `OPENAI_API_KEY`（实测遇到过一把无效的）时，进程环境会盖掉
+项目 `.env`。症状是「密钥明明配对了却报 401，且报错里显示的密钥后几位和
+`.env` 里的不一样」。
 
-`config.py` 的对策：项目 `.env` 优先（`override=True`），且冲突时打
-WARNING 并把冲突键暴露给 `/health` 的 `config_conflicts`。
+`config.py` 把它做成**显式可配**的 `DEEP_AGENT_ENV_PRIORITY`：
+
+- `dotenv`（默认）：`.env` 优先。防「被污染的 shell 静默改掉项目配置」
+- `process`：进程环境优先。**跑变体实例必须用这档**，例如
+  `DEEP_AGENT_ENV_PRIORITY=process DB_DIALECT=mysql PORT=8081 python main.py`
+- `none`：完全忽略 `.env`（测试/CI 用）
+
+为什么默认不是 `process`：两种失败模式代价不对等。`dotenv` 的代价是「有意的
+环境变量覆盖不生效」——加个前缀就绕过，且启动日志会告诉你；`process` 的代价是
+「本机某个无关的全局变量静默改掉项目配置」——表现为身份诡异的 401 或连错库。
+
+**无论哪一档，冲突都会打 WARNING 并把键名暴露给 `/health` 的
+`config_conflicts`**，包括「生效值长什么样」（脱敏）。这条可见性是本机制的核心
+价值：它把「配置没生效」从玄学变成一行日志。
+
+注意 `process` 档会放行**所有**环境变量，包括那把无效的全局密钥——所以跑变体
+实例时要连密钥一起显式传入，不要只传你要改的那几项。
 
 ### 坑 2：`MemorySaver` 之外，checkpointer 必须显式声明上下文窗口
 

README.md

@@ -21,6 +21,7 @@ pip install -r requirements.txt
 cp .env.example .env
 #    至少填 OPENAI_API_KEY 与 OPENAI_BASE_URL
 #    想用真实 MySQL：把 DB_DIALECT 改成 mysql 并填 DB_* 连接信息
+#    （库不存在时启动会自动建库，不用手工 CREATE DATABASE）
 
 # 3) 跑起来
 python main.py
@@ -31,7 +32,11 @@ python main.py
 ```
 
 开箱配置走 **SQLite 开发档**（`DB_DIALECT=sqlite`），无需任何数据库服务即可
-跑起来。生产/联调改 `DB_DIALECT=mysql` 并执行 `db/mysql_schema.sql`。
+跑起来。生产/联调改 `DB_DIALECT=mysql`。
+
+> **不能与 1.5 共用同一个 Python 环境**：2.0 绑定 langchain 1.3 / langgraph 1.2 /
+> deepagents 0.6.12，与 1.5 声明的 `langchain>=0.3` 系列不同代，装在一起必然冲突。
+> 用独立 venv。
 
 ---
 
@@ -258,18 +263,19 @@ agent 启动时只读 frontmatter，按需才 `read_file` 读全文（渐进披
 
 | 变量 | 说明 |
 |---|---|
+| `DEEP_AGENT_ENV_PRIORITY` | `.env` 与进程环境谁优先。`dotenv`（默认）/ `process` / `none`。**跑变体实例用 `process`**：`DEEP_AGENT_ENV_PRIORITY=process DB_DIALECT=mysql PORT=8081 python main.py` |
 | `DB_DIALECT` | `mysql`（生产）/ `sqlite`（本地开发与测试，无需数据库服务） |
+| `DB_NAME` | MySQL 库名；库不存在时启动会自动建库，不需要手工 `CREATE DATABASE` |
 | `CONTEXT_LIMIT` | 声明的上下文窗口，**压缩的唯一调节旋钮**。0 = 用模型登记值 |
 | `CTX_COMPACT_ENABLED` | 关掉会同时失去历史落盘能力，长会话会超窗报错 |
 | `CHECKPOINTER` | `sqlite`（默认，落盘）/ `memory`（重启丢图状态） |
 | `HITL_TOOLS` | 追加需要人工审批的工具名，逗号分隔。如 `calculate` 便于自测中断闭环 |
 | `BILLING_ALLOW_UNPRICED` | 模型未定价时是否按 0 计费。**默认 0 = 直接报错**（fail-closed） |
-| `DEEP_AGENT_SKIP_DOTENV` | 测试/CI 用：完全跳过 `.env` |
 | `TFAGENTS_DAILY_TOKEN_LIMIT` | 每用户每日 token 限额，0 = 不限 |
 
-**配置冲突会被大声报出来**：若机器上有同名进程环境变量且值与 `.env` 不同，
-启动日志会打 WARNING，`/health` 的 `config_conflicts` 也能看到。项目 `.env`
-优先——这条反直觉但必要，原因见 `config.py` 的模块 docstring。
+**配置冲突会被大声报出来**：若 `.env` 与进程环境里有同名但不同的值，启动
+日志会打 WARNING（含脱敏后的生效值），`/health` 的 `config_conflicts` 也能
+看到。这解决的是一个很费时间的现象：改了配置却不生效，却没有任何提示。
 
 ---
 

config.py

@@ -6,18 +6,30 @@
 必须等于「什么都不配也能跑起来」的那套（沿用 1.5 的工程纪律：默认 = 可用的
 最小配置，任何新开关的默认值都不应改变既有行为）。
 
-## 关于环境变量污染（踩过的坑，别改回去）
-
-机器上若存在全局的 ``OPENAI_API_KEY``，``load_dotenv(override=False)`` 会让
-**进程环境胜出、项目 .env 被静默忽略**。表现是：配好的密钥明明是对的，服务
-却报 401，且报错里显示的密钥后几位和 .env 里的对不上——足以浪费半天。
-
-所以这里的策略是：
-1. **项目 .env 优先**（``override=True``）。.env 是本项目对外声明的配置面
-   （有 .env.example 为证），比机器上某个无关的全局变量更具体。
-2. 若没有 .env 文件，行为与常规一致：纯用进程环境变量（容器/CI 场景不受影响）。
-3. 两者**都存在且值不同**时，记一条 WARNING 并把冲突键名暴露给 /health，
-   让「我改了配置为什么不生效」有迹可循。
+## 关于环境变量与 .env 的优先级（踩过的坑，别改回去）
+
+机器上若存在全局的 ``OPENAI_API_KEY``（实测遇到过一把无效的），进程环境会
+**静默盖掉项目 .env**。症状是：配好的密钥明明是对的，服务却报 401，且报错
+里显示的密钥后几位和 .env 里的对不上——足以浪费半天。
+
+所以优先级做成**显式可配**，而不是隐含假定某一方赢：
+
+===================  ====================================================
+``DEEP_AGENT_ENV_PRIORITY``  行为
+===================  ====================================================
+``dotenv``（默认）   项目 .env 优先。防「被污染的 shell 静默改掉项目配置」
+``process``          进程环境优先。用于**跑变体实例**：不同端口、不同库、
+                     不同方言（``DB_DIALECT=mysql PORT=8081 python main.py``）
+``none``             完全忽略 .env，只认进程环境与代码默认值（测试/CI 用）
+===================  ====================================================
+
+无论哪一档，**两侧值不同都会打 WARNING 并把键名暴露给 /health**，
+让「我改了配置为什么不生效」有迹可循。
+
+为什么默认不是 ``process``：两种失败模式的代价不对等。``dotenv`` 的代价是
+「有意的环境变量覆盖不生效」（一条命令行加个前缀就绕过了，且启动日志会告诉
+你），``process`` 的代价是「本机某个无关的全局变量静默改掉项目配置」（表现为
+身份诡异的 401 或连错数据库）。
 """
 
 from __future__ import annotations
@@ -32,14 +44,14 @@ from dotenv import dotenv_values, load_dotenv
 ROOT = Path(__file__).resolve().parent
 
 _ENV_PATH = ROOT / ".env"
-# 测试/CI 用：设 DEEP_AGENT_SKIP_DOTENV=1 可完全跳过 .env，只认进程环境变量
-# 与代码默认值。没有这个开关，测试就会受开发者本机 .env 的内容影响——
-# 同一套测试在不同机器上跑出不同结果，是最难查的一类失败。
-_SKIP_DOTENV = os.getenv("DEEP_AGENT_SKIP_DOTENV", "").strip().lower() in {"1", "true", "yes", "on"}
+# dotenv（默认）| process | none
+_ENV_PRIORITY = (os.getenv("DEEP_AGENT_ENV_PRIORITY", "dotenv") or "dotenv").strip().lower()
+if _ENV_PRIORITY not in {"dotenv", "process", "none"}:
+    _ENV_PRIORITY = "dotenv"
 
 # 直接读文件（不经 os.environ），用于检测冲突
 _FILE_VALUES: dict[str, str] = (
-    {} if _SKIP_DOTENV else {k: v for k, v in dotenv_values(_ENV_PATH).items() if v}
+    {} if _ENV_PRIORITY == "none" else {k: v for k, v in dotenv_values(_ENV_PATH).items() if v}
 )
 # 冲突记录：键名 → 说明。供 log_config_sources() 与 /health 使用
 CONFLICTS: dict[str, str] = {}
@@ -47,11 +59,15 @@ CONFLICTS: dict[str, str] = {}
 for _key, _file_value in _FILE_VALUES.items():
     _process_value = os.environ.get(_key)
     if _process_value and _process_value != _file_value:
-        CONFLICTS[_key] = "项目 .env 覆盖了同名进程环境变量"
+        CONFLICTS[_key] = (
+            "项目 .env 覆盖了同名进程环境变量"
+            if _ENV_PRIORITY == "dotenv"
+            else "进程环境变量覆盖了项目 .env"
+        )
 
-# .env 优先：见模块 docstring 的「环境变量污染」一节
-if not _SKIP_DOTENV:
-    load_dotenv(_ENV_PATH, override=True)
+# 见模块 docstring 的「环境变量与 .env 的优先级」一节
+if _ENV_PRIORITY != "none":
+    load_dotenv(_ENV_PATH, override=(_ENV_PRIORITY == "dotenv"))
 
 
 def _mask(value: str) -> str:
@@ -68,17 +84,30 @@ def log_config_sources() -> None:
     from obs import get_logger
 
     logger = get_logger("config")
-    if _ENV_PATH.exists():
-        logger.info("配置来源：%s（%d 项，其中 %d 项覆盖了同名进程环境变量）",
-                    _ENV_PATH, len(_FILE_VALUES), len(CONFLICTS))
+    if _ENV_PRIORITY == "none":
+        logger.warning(
+            "DEEP_AGENT_ENV_PRIORITY=none：已忽略 %s，全部配置取自进程环境变量与代码默认值",
+            _ENV_PATH,
+        )
+    elif _ENV_PATH.exists():
+        logger.info(
+            "配置来源：%s（%d 项）| 优先级=%s | 冲突 %d 项",
+            _ENV_PATH, len(_FILE_VALUES), _ENV_PRIORITY, len(CONFLICTS),
+        )
     else:
         logger.warning("未找到 %s，全部配置取自进程环境变量与代码默认值", _ENV_PATH)
 
     for key, reason in CONFLICTS.items():
         # 只暴露键名与脱敏后的值，不打印完整密钥
+        effective = os.getenv(key, "")
         logger.warning(
-            "配置冲突 %s：%s（生效值 %s）",
-            key, reason, _mask(_FILE_VALUES.get(key, "")),
+            "配置冲突 %s：%s（生效值 %s）%s",
+            key,
+            reason,
+            _mask(effective),
+            "；如需让进程环境优先，加 DEEP_AGENT_ENV_PRIORITY=process"
+            if _ENV_PRIORITY == "dotenv"
+            else "",
         )
 
 

db/db_conn.py

@@ -147,25 +147,54 @@ class _MySqlAdapter:
         self._local = threading.local()
         self._last_used: dict[int, float] = {}
 
-    def _connection(self):
+    def _connect(self, database: str | None):
+        """建一个 pymysql 连接。``database=None`` 表示不选库。"""
         import pymysql
         from pymysql.cursors import DictCursor
 
+        return pymysql.connect(
+            host=config.DB_HOST,
+            port=config.DB_PORT,
+            user=config.DB_USER,
+            password=config.DB_PASSWORD,
+            database=database,
+            charset="utf8mb4",
+            cursorclass=DictCursor,
+            autocommit=True,
+            connect_timeout=10,
+            read_timeout=60,
+            write_timeout=60,
+        )
+
+    def _ensure_database(self) -> None:
+        """库不存在则建库。
+
+        骨架不该要求使用者先手工跑一次 CREATE DATABASE——那是最容易被漏掉、
+        且报错信息（"Unknown database"）不提示该怎么解决的一步。
+        """
+        conn = self._connect(None)
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f"CREATE DATABASE IF NOT EXISTS `{config.DB_NAME}` "
+                    f"DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci"
+                )
+            logger.info("已确保数据库存在：%s", config.DB_NAME)
+        finally:
+            conn.close()
+
+    def _connection(self):
         conn = getattr(self._local, "conn", None)
         if conn is None:
-            conn = pymysql.connect(
-                host=config.DB_HOST,
-                port=config.DB_PORT,
-                user=config.DB_USER,
-                password=config.DB_PASSWORD,
-                database=config.DB_NAME,
-                charset="utf8mb4",
-                cursorclass=DictCursor,
-                autocommit=True,
-                connect_timeout=10,
-                read_timeout=60,
-                write_timeout=60,
-            )
+            try:
+                conn = self._connect(config.DB_NAME)
+            except Exception as exc:
+                # 1049 = Unknown database：自动建库后重试一次
+                if getattr(exc, "args", [None])[0] == 1049:
+                    self._ensure_database()
+                    conn = self._connect(config.DB_NAME)
+                else:
+                    raise
             self._local.conn = conn
         else:
             ident = id(conn)

scripts/http_e2e.py

@@ -7,13 +7,16 @@
 from __future__ import annotations
 
 import json
+import os
 import sys
 import time
 
 import httpx
 
-BASE = "http://127.0.0.1:8080"
-USER = "http_e2e"
+# 允许指向别的实例（例如另起一个 MySQL 档的服务端口）做方言专项验证：
+#   BASE=http://127.0.0.1:8081 python scripts/http_e2e.py
+BASE = os.getenv("BASE", "http://127.0.0.1:8080").rstrip("/")
+USER = os.getenv("E2E_USER", "http_e2e")
 PASSED: list[str] = []
 FAILED: list[str] = []
 

tests/conftest.py

@@ -17,12 +17,15 @@ import uuid
 from pathlib import Path
 
 # ---------- 测试环境（必须在项目导入前设置）----------
+# 用 setdefault 而非直接赋值：外部已给出的 DB_DIALECT / DB_* 优先生效，
+# 这样同一套测试可以直接跑在 MySQL 上验证方言无关性——
+#   DB_DIALECT=mysql DB_HOST=127.0.0.1 DB_PORT=3399 DB_USER=root DB_NAME=t python -m pytest
 _TMP_ROOT = Path(tempfile.mkdtemp(prefix="deep_agent_test_"))
-os.environ["DEEP_AGENT_SKIP_DOTENV"] = "1"
-os.environ["DB_DIALECT"] = "sqlite"
-os.environ["DB_SQLITE_PATH"] = str(_TMP_ROOT / "test.sqlite")
-os.environ["CHECKPOINTER"] = "memory"          # 测试不需要落盘图状态
-os.environ["CTX_COMPACT_ENABLED"] = "0"        # 不引入摘要中间件，测试更纯粹
+os.environ["DEEP_AGENT_ENV_PRIORITY"] = "none"   # 完全忽略本机 .env
+os.environ.setdefault("DB_DIALECT", "sqlite")
+os.environ.setdefault("DB_SQLITE_PATH", str(_TMP_ROOT / "test.sqlite"))
+os.environ.setdefault("CHECKPOINTER", "memory")   # 测试不需要落盘图状态
+os.environ["CTX_COMPACT_ENABLED"] = "0"           # 不引入摘要中间件，测试更纯粹
 os.environ["OPENAI_API_KEY"] = "sk-test-key"
 os.environ["OPENAI_BASE_URL"] = "https://example.invalid"
 os.environ["TFAGENTS_DAILY_TOKEN_LIMIT"] = "0"
@@ -31,6 +34,7 @@ os.environ["LOG_LEVEL"] = "WARNING"
 
 import pytest  # noqa: E402
 
+import config  # noqa: E402
 from db import db_conn  # noqa: E402
 
 
@@ -68,6 +72,12 @@ def _clean_tables():
     yield
 
 
+@pytest.fixture(scope="session")
+def dialect() -> str:
+    """当前测试跑的方言。用于需要区分方言的断言。"""
+    return config.DB_DIALECT
+
+
 @pytest.fixture
 def pricing_ready():
     """确保单价表里有当前模型的条目，供计费相关用例使用。"""