test(W29-d): 补上「数据落库 + HTTP 链路」的端到端验证（此前只测到进程内直调）

## 补的是什么空白

W28 / W29 的全部验证都是**进程内直调** —— 直接构造 ToolExecutor、直接调 Agent，
或者**手写** configured_tools。**从来没有走过「HTTP 受理 → Worker 执行 → 落库」这条真实链路。**

补测前的实测证据：

    conversation_message 里 tool_calls LIKE '%trend_chart%'  -> 0 条
    conversation_message 里 tool_calls LIKE '%svc_topic%'    -> 0 条（W28 事项码）
    agent_run 里 agent_type='financial_nl2sql'               -> 0 条
    最近一条真实 run                                          -> 2026-09-21 14:46（早于这两批改动）

而**前端取图完全依赖这条路**：前端读的是 `result.tool_calls.data.trend_chart`，
它由 `agent_persistence_service` 写进 `conversation_message.tool_calls` 这个 JSON 列。
**这条路不通，图就是「代码里有、页面上没有」。**

## 两个测试

1. `test_customer_service_trend_chart_persistence.py`
   走势问句走 HTTP + Worker，断言 `trend_chart` 与 `svc_topic` 同时出现在**出参与库里**；
   并逐条断言图数据中的涨跌数字**出现在答复正文里**（`INV-8` 的落库侧前提）。

2. `test_financial_nl2sql_http_wiring.py`
   接线后 `financial_nl2sql` 能否在**真实 HTTP 链路**上出结果。
   此前的「接线验证」是**手写 configured_tools** —— 等于把要验证的东西自己填好了：
   它能证明「工具逻辑对」，但**证明不了「接线生效」**（白名单是从 `config_release` 装配的）。
   断言 `data` / `sql` 出参 + SELECT-only 自证 + `data.total`（证明**真的执行了**，不是只编译）。

两个测试都按 `session_id`（带 uuid）隔离并在 finally 清理，不污染历史数据。

## 结果

- 两条单跑均 passed（4.12s / 2.84s）
- 全量 pytest **2547 passed / 3 skipped / 0 failed**（2545 + 2）
- ruff：两个新文件 0 告警

## 一句留给后人的话

写这两个测试时的第一个失败是 `TypeError: object RequestContext can't be used in
'await' expression` —— `WorkerRuntime.restore_context` 会 `await self.resolve_identity(...)`，
必须传**协程函数**，不能传 lambda。已在两处代码里注明，免得下一个人再踩。

tests/integration/test_customer_service_trend_chart_persistence.py

@@ -0,0 +1,144 @@
+"""`W29-d` · 图表/事项码数据**落库**的端到端验证（走真实 HTTP + Worker）。
+
+## 为什么这个文件必须存在
+
+`W29` 的图表与 `W28` 的事项码，此前的全部验证都是**进程内直调**
+（`_w29_e2e_check.py` 直接构造 `ToolExecutor`；金标探针直接调 Agent）——
+**从来没有走过「HTTP 受理 → Worker 执行 → 落库」这条真实链路**。
+
+后果是具体可查的：库里 `tool_calls LIKE '%trend_chart%'` 命中 **0 条**、
+`'%svc_topic%'` 也是 **0 条**（最近一条真实 run 停在 2026-09-21 14:46，早于这两批改动）。
+
+而**前端取图完全依赖这条路**：前端读的是 `result.tool_calls.data.trend_chart`，
+它由 `agent_persistence_service` 写进 `conversation_message.tool_calls` 这个 JSON 列。
+**这条路不通，图就是「代码里有、页面上没有」。**
+
+本文件补的就是这一段：
+
+1. 走真实 `POST /api/v1/agent-runs` → `WorkerRuntime` 执行 → 落库；
+2. 断言 **HTTP 出参**里 `result.tool_calls.data.trend_chart` 存在（= 前端取数路径本身）；
+3. 断言 **库里** `conversation_message.tool_calls` 同样含 `trend_chart` 与 `svc_topic`。
+
+## 为什么断言数据、而不是断言像素
+
+组件渲染已由 `_w29_chart_render_check.mjs`（jsdom）覆盖（11/11）。
+本文件管的是它的**上游**：数据有没有被写进库、有没有出到接口。
+分开的理由是**失败时的定位完全不同** ——
+「图不显示」既可能是数据没落库（本文件），也可能是渲染出错（那个文件），
+混在一个测试里只会得到一句"图表测试失败"。
+"""
+
+from __future__ import annotations
+
+import json
+from uuid import uuid4
+
+import httpx
+import pytest
+from sqlalchemy import delete, select
+
+from app.api.dependencies.auth import build_request_context
+from app.core.contracts import RequestContext
+from app.infrastructure.db import SessionFactory
+from app.main import create_app
+from app.model.conversation import ConversationMessage
+from app.model.platform import AgentRun
+from app.service.agent.bootstrap import get_agent_factory
+from app.worker.runtime import WorkerRuntime
+
+#: 能确定性命中 `E6` 出口的问句：含行情词 + 6 位产品代码 ⇒ **不调模型**、快。
+TREND_QUESTION = "159382今天净值走势"
+
+
+@pytest.mark.integration
+async def test_trend_chart_and_service_topic_reach_db_and_api() -> None:
+    """走势问句走完整链路后，`trend_chart` 与 `svc_topic` 必须同时出现在**出参与库里**。"""
+    context = RequestContext(
+        user_id="1",
+        trace_id=str(uuid4()),
+        roles=("customer",),
+        customer_ids=("1",),
+        data_scope="self",
+        portal="api",
+        # `fund:quote:read` 是 `query_fund_trend` 的 `required_permission`（`bootstrap.py`）。
+        permissions=("agent:run", "fund:quote:read"),
+    )
+    app = create_app()
+    app.dependency_overrides[build_request_context] = lambda: context
+    session_id = f"w29d-chart-{uuid4()}"
+    run_id: str | None = None
+    async def identity(value: object) -> RequestContext:
+        # ⚠️ `WorkerRuntime.restore_context` 会 `await self.resolve_identity(...)`，
+        # 所以这里**必须**是协程函数 —— 传同步 lambda 会得到
+        # `TypeError: object RequestContext can't be used in 'await' expression`。
+        del value
+        return context
+
+    runtime = WorkerRuntime(get_agent_factory(), resolve_identity=identity)
+
+    try:
+        async with httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app), base_url="http://test"
+        ) as client:
+            accepted = await client.post("/api/v1/agent-runs", json={
+                "agent_type": "customer_service",
+                "message": TREND_QUESTION,
+                "session_id": session_id,
+                "idempotency_key": str(uuid4()),
+            })
+            assert accepted.status_code == 202, accepted.text
+            run_id = accepted.json()["data"]["run_id"]
+
+            assert await runtime.dispatch_one(run_id=run_id), "run 未派发成功"
+            await runtime.execute(run_id)
+
+            snapshot = (await client.get(f"/api/v1/agent-runs/{run_id}")).json()["data"]
+
+        assert snapshot["status"] == "succeeded", snapshot
+        result = snapshot["result"] or {}
+
+        # ① **前端取数路径**：`result.tool_calls.data.trend_chart`
+        #    前端之所以读这一层而不是 `result.data`：后者只对 `financial_nl2sql` 的 run 暴露。
+        stored = result.get("tool_calls")
+        assert isinstance(stored, dict), f"tool_calls 形状异常：{type(stored).__name__}"
+        data = stored.get("data")
+        assert isinstance(data, dict), f"tool_calls.data 未落库：{list(stored)}"
+
+        assert "trend_chart" in data, f"data 里没有 trend_chart，实际键：{sorted(data)}"
+        chart = data["trend_chart"]
+        assert chart["kind"] == "interval_change"
+        assert chart["source"] == "fin_nav_history"
+        assert chart["intervals"], "区间数据为空 —— 图会画不出来"
+        # `INV-8` 的落库侧前提：这些数字必须在正文里出现过（正文侧由 jsdom 测试断言）。
+        for interval in chart["intervals"]:
+            assert str(interval["change_pct"]) in (result.get("content") or ""), (
+                f"图数据里的涨跌 {interval['change_pct']} 未出现在答复正文：{result.get('content')}"
+            )
+
+        # ② `W28` 的事项码同批落库（此前同样是「代码里有、库里没有」）
+        assert data.get("svc_topic"), "svc_topic 未落库"
+        assert data.get("svc_owner") and data.get("svc_retention")
+
+        # ③ 库里确实有（不只是出参拼出来的）
+        async with SessionFactory() as session:
+            message = (await session.execute(
+                select(ConversationMessage)
+                .where(
+                    ConversationMessage.session_id == session_id,
+                    ConversationMessage.role == "assistant",
+                )
+                .order_by(ConversationMessage.id.desc())
+                .limit(1)
+            )).scalars().first()
+        assert message is not None, "本次会话没有落库的助手消息"
+        raw = json.dumps(message.tool_calls, ensure_ascii=False)
+        assert "trend_chart" in raw, "库里的 tool_calls 没有 trend_chart"
+        assert "svc_topic" in raw, "库里的 tool_calls 没有 svc_topic"
+    finally:
+        # 只删本次 session 产生的行（`session_id` 带 uuid，不会误伤历史数据）。
+        async with SessionFactory() as session, session.begin():
+            await session.execute(
+                delete(ConversationMessage).where(ConversationMessage.session_id == session_id)
+            )
+            if run_id:
+                await session.execute(delete(AgentRun).where(AgentRun.run_id == run_id))

tests/integration/test_financial_nl2sql_http_wiring.py

@@ -0,0 +1,105 @@
+"""`W29-d` · NL2SQL 接线后的 **HTTP 链路**验证（此前只测到服务层）。
+
+## 为什么需要它
+
+`W29` 把 `agent_tools/financial_nl2sql:financial_query` 发布出去之后，验证只做到两层：
+
+- **服务层直调**：`FinancialNL2SQLService().query(...)`；
+- **工具层**：自建 `ToolExecutor` + **手写**
+  `configured_tools={"financial_query": ("query_financial_data",)}`。
+
+这两层都**绕过了**真实链路最关键的一段：`agent_tools` 白名单是**从库里装配**的
+（`AgentFactory` 读 `config_release` 的生效快照）。手写 `configured_tools` 等于
+**把要验证的东西自己填好了** —— 它能证明"工具逻辑对"，但**证明不了"接线生效"**。
+
+库里的证据也印证了这一点：`agent_run` 表里 `agent_type='financial_nl2sql'` **0 条**。
+
+本文件走完整链路：HTTP 受理 → Worker 从**生效配置**装配白名单 → 工具执行 → 落库 → 读出参。
+"""
+
+from __future__ import annotations
+
+from uuid import uuid4
+
+import httpx
+import pytest
+from sqlalchemy import delete
+
+from app.api.dependencies.auth import build_request_context
+from app.core.contracts import RequestContext
+from app.infrastructure.db import SessionFactory
+from app.main import create_app
+from app.model.conversation import ConversationMessage
+from app.model.platform import AgentRun
+from app.service.agent.bootstrap import get_agent_factory
+from app.worker.runtime import WorkerRuntime
+
+WRITE_WORDS = ("INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE", "GRANT", "REVOKE")
+
+
+@pytest.mark.integration
+async def test_financial_nl2sql_serves_over_http_after_wiring() -> None:
+    """接线后，`financial_nl2sql` 必须能在**真实 HTTP 链路**上出结果。
+
+    断言的是"端到端可用"，不是"工具逻辑正确"（后者由 `test_financial_nl2sql_*.py` 覆盖）。
+    两者会分别失败，所以分开测 —— 否则"NL2SQL 坏了"这句话定位不到是哪一层。
+    """
+    context = RequestContext(
+        user_id="1",
+        trace_id=str(uuid4()),
+        roles=("operator",),
+        data_scope="all",
+        portal="api",
+        permissions=("agent:run", "financial:nl2sql:read"),
+    )
+    app = create_app()
+    app.dependency_overrides[build_request_context] = lambda: context
+    session_id = f"w29d-nl2sql-{uuid4()}"
+    run_id: str | None = None
+
+    async def identity(value: object) -> RequestContext:
+        # 必须是协程函数：`restore_context` 会 `await` 它。
+        del value
+        return context
+
+    runtime = WorkerRuntime(get_agent_factory(), resolve_identity=identity)
+    try:
+        async with httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app), base_url="http://test"
+        ) as client:
+            accepted = await client.post("/api/v1/agent-runs", json={
+                "agent_type": "financial_nl2sql",
+                "message": "查询近30天净值",
+                "session_id": session_id,
+                "idempotency_key": str(uuid4()),
+            })
+            assert accepted.status_code == 202, accepted.text
+            run_id = accepted.json()["data"]["run_id"]
+            assert await runtime.dispatch_one(run_id=run_id), "run 未派发成功"
+            await runtime.execute(run_id)
+            snapshot = (await client.get(f"/api/v1/agent-runs/{run_id}")).json()["data"]
+
+        assert snapshot["status"] == "succeeded", snapshot
+        result = snapshot["result"] or {}
+
+        # ① 出参：`RunQueryService` **只**对 `financial_nl2sql` 暴露 `data` 与 `sql`
+        assert isinstance(result.get("data"), dict), f"出参缺 data：{sorted(result)}"
+        assert isinstance(result.get("sql"), str), f"出参缺 sql：{sorted(result)}"
+
+        # ② 只读自证（工具声明 `read_only=True`，这里验它真的没产出写语句）
+        sql = result["sql"]
+        upper = sql.upper()
+        assert upper.startswith("SELECT"), f"只读工具产出了非 SELECT：{sql}"
+        assert not any(word in upper for word in WRITE_WORDS), sql
+        assert "*" not in sql, f"禁止 SELECT *：{sql}"
+        assert "LIMIT" in upper, f"缺 LIMIT：{sql}"
+
+        # ③ 真的**执行**了，而不是只编译完就返回 —— `data.total` 只有执行后才填
+        assert "total" in result["data"], f"查询未真正执行：{result['data']}"
+    finally:
+        async with SessionFactory() as session, session.begin():
+            await session.execute(
+                delete(ConversationMessage).where(ConversationMessage.session_id == session_id)
+            )
+            if run_id:
+                await session.execute(delete(AgentRun).where(AgentRun.run_id == run_id))