F3 step 1: fake LLM service + stop e2e

docs/DEVLOG.md

@@ -1238,7 +1238,7 @@ S 决定（10-14）：跳过"当前 master 先重跑一次基线"，直接改提
 
 未做 / 下一步：F2。`read_csv_with_encoding` 内部 bytes / str / DataFrame 三份并存的峰值内存（分析路径也一样，属于读取函数本身，不在 F1.1 顺手改；需要时改成解码后释放 bytes、或按块解码）。A1 自动复跑规则仍按计划在下一次闸门 PR 落地。
 
-## [34] 2026-10-16 · 阶段 5 F2：SSE、事件渲染、重放合成 done、停止、👍👎、gen_ui_types · commit <待提交>
+## [34] 2026-10-16 · 阶段 5 F2：SSE、事件渲染、重放合成 done、停止、👍👎、gen_ui_types · commit 50e9add
 
 当前阶段：**阶段 5 前端，F2 完成，下一步 F3**（ECharts、prepare、主题、开关、换图型、导出、注入实测）。后端停止的语义经 S 三轮评审。[33] 的 SHA 已回填 cccddd5。
 
@@ -1272,3 +1272,26 @@ S 决定（10-14）：跳过"当前 master 先重跑一次基线"，直接改提
 - 错误文案脱敏（THREAT_MODEL T11；原写 T10，与 §2.1 的 CSP 条目撞号，已改）：单独一个 PR。
 - 停止的端到端测试（Playwright + 能挂住的假模型）、验收第 5 条的"实测停止耗时"：随 F3 的假模型服务。
 - 切走会话时不中断流（分析在后端跑完、结果照常落库）；回来时按库里的 status 显示。
+
+## [35] 2026-10-16 · 阶段 5 F3（进行中）：假模型服务 + 停止端到端 · commit <待提交>
+
+当前阶段：**阶段 5 前端，F3 进行中**。第一步：公共的假模型服务 + 停止的端到端（验收第 5 条两个耗时都测出数）；下一步 ECharts 渲染。[34] 的 SHA 已回填 50e9add（PR #35 squash；DEVLOG 一律引用 CI run id 而不是分支中间 commit，squash 后零死引用）。
+
+假模型服务 `tests/fake_llm.py`（S 10-16 定：独立进程、无状态、按 model 名分派）：
+- OpenAI 兼容 `POST /v1/chat/completions`，标准库 `ThreadingHTTPServer`，不加依赖。产品代码不改，只把 base_url 指过去。
+- model 名 `<scenario>[@step]`：`ok`、`hang-<N>s`、`fail-<code>`、`slow-<N>s`；step ∈ plan/codegen/fix/answer（工作流）、act/final（ReAct）。不写 @step = 第一次调用（工作流 plan、ReAct act），按请求内容就能判定，不需要计数。
+- **fail loud**：未知 scenario / step、拼错（`hang-30`）、@step 不属于本次链路、认不出当前步骤（提示词变了）、`stream=true` → 一律 HTTP 400 + 原因。选 400 而不是 500：500 会被 LLMClient 当可重试错误退避重试，表现成"偶发慢"，400 立刻以 bad_request 失败。没有"认不出就走默认剧本"的分支。
+- 和 S 的提议差一处：没做 `stream-<N>tps`。产品不用流式调用（ChatOpenAI 非流式，SSE 是事件粒度不是 token 粒度），控 token 速率测不到任何东西；用 `slow-<N>s`（每次调用都慢 N 秒，把事件拉开）测增量渲染。
+- 测试 `tests/test_fake_llm.py`（39，四个 scenario 都在真链路上走过）：解析器正反例、认不出 → 400 且不先睡、跨链路 @step → 400、stream → 400、默认目标是第一次调用、同请求同回复；两种 agent 走真 HTTP + 真产品链路跑 ok / hang / 拼错——提示词一改这里先红（漂移守卫）；fail-500@codegen → server、退避重试 3 次（500 本来就是测重试路径，和认不出 → 400 不同），fail-401 → auth 1 次；slow-1s → 3 次调用总耗时 ≥ 3 s。ReAct 的 act/final 只按"最后一条是不是 tool"分，多轮工具调用分不出第几轮（代码注释写了边界）。
+
+Playwright：
+- 两个后端、两个 project：`nokey`（f0/f1/f2，没有 Key）和 `fake`（f3*，base_url → 假服务，`DATACHAT_LLM_MAX_RETRIES=0`）。场景用 `PUT /settings {model}` 按测试设；model 是后端全局设置，所以 `workers: 1`。
+- `e2e/f3-stop.spec.ts`：`hang-8s@codegen`，plan 到了点停止。beforeAll 先用 ok 走一轮 API 预热：后端进程首次分析要 import pandas/matplotlib/langchain + 建字体缓存，沙箱里 20 多秒（不预热时 plan 45 秒都没到）；被测窗口里混着冷启动，8 秒的上界根本量不准。CI 的新 runner 同样是冷的，以后任何测耗时的 e2e 都要预热。断言：落 cancelled、显示"已取消"不是失败、没有"未保存"、立刻重问不撞 409 并正常出结论、刷新后重放一致。
+- 实测（沙箱本地两次）：**感知延迟 1–2 ms**（点停止 → "正在停止…"，乐观 UI，门槛 < 500 ms）；点停止 → `POST /cancel` 返回 17–23 ms；**后端延迟 8048–8051 ms**（`POST /cancel` 200 → 界面"已取消"）。plan 一到就点，codegen 那次调用还剩约 8 秒，后端延迟就是这 8 秒：和"上界 ≈ 当前模型调用剩余时间"一致，只记录、不设门槛（断言只放 < 10 s 的健全性上界）。
+
+验证：pytest 654 passed（+39）；vitest 38；Playwright 8（nokey 7 + fake 1，本地连跑两次）；tsc、lint 通过。
+
+未做：
+- ECharts 渲染、prepare、主题、开关、换图型、导出（F3 下一步）。
+- late-cancel（answer 已发出后才取消）的端到端：answer 和 done 之间没有可挂住的模型调用，假服务造不出这个时序；单元测试已覆盖。
+

docs/roadmap/phase5-frontend.md

@@ -13,6 +13,9 @@
   3. **导出 PNG 得到非空 data URL，文件名已清洗** [v2] → 👍👎
   4. **重启后端 → 重新打开 URL → 历史会话还在，图表还在（走重放路径）** [v2]
   5. **按停后：后端在有限时间内停下（记录实测耗时），status 落为 cancelled，立刻重问不撞 409** [v2]
+     - 耗时分两个数，分开记，不混成一个 [F3 前定，S 10-16]：
+       - **感知延迟**（验收指标）：点"停止"→ 按钮变"正在停止…"（乐观 UI，同一帧）；最终态"已取消 / 已取消（未保存）"只来自 settle() 读到的库里 status。
+       - **后端延迟**（只记录，不设门槛）：`POST /cancel` 返回 200 → 流里出现 error(kind=cancelled) + done / 库里 status=cancelled。检查点在"每产出一个事件之前"，所以上界 ≈ 当前这次模型调用的剩余时间（秒级到几十秒），这是架构选定的粒度，不是 bug；"< 1s"之类的门槛架构上达不到，不写。用假模型服务挂住 N 秒实测，记进 DEVLOG。
   6. **用 v1 的库启动一次：feedback 表出现，旧表行数不变**（写成测试）[v2]
   7. **注入实测结论写进 THREAT_MODEL §2.3，DOM 断言在后端托管的 dist 上跑通** [v2]
 
@@ -71,7 +74,7 @@
 |---|---|---|
 | vitest + ECharts SSR SVG | 按 THREAT_MODEL §2.3，只测 v1 可达的字段：title.text、series[].name、xAxis.data[]、xAxis/yAxis.name | tooltip（SSR 不渲染） |
 | vitest + testing-library | 错误、警告、结论、预览单元格、feedback note 都按文本渲染 | — |
-| Playwright（单个 Linux job） | **跑在后端托管的 dist 上（带 CSP）**：用 `chat_model_factory` 起假模型服务，产出带注入串的图，真机悬停 tooltip，断言不产生 DOM 节点 | — |
+| Playwright（单个 Linux job） | **跑在后端托管的 dist 上（带 CSP）**：用假模型服务（`tests/fake_llm.py`，独立进程的 OpenAI 兼容 HTTP 服务，后端只改 base_url，不是进程内对象），产出带注入串的图，真机悬停 tooltip，断言不产生 DOM 节点 | — |
 
 v2 才可达的字段（link、formatter、aria 等）留到层 2 再测。
 
@@ -90,6 +93,12 @@ v2 才可达的字段（link、formatter、aria 等）留到层 2 再测。
 | F1 | 数据集（含预览）、会话、设置 | 2–3 天 |
 | F2 | SSE、事件渲染、重放合成 done、cancel、👍👎、gen_ui_types（**完成，DEVLOG [34]**） | 3–5 天 |
 | F3 | ECharts、prepare、主题、开关、换图型、导出、注入实测 + THREAT_MODEL | 4–5 天 |
+
+**F3 已定 [S 10-16]**
+- **假模型服务**是 F3/F4/T11 所有异步测试的公共设施：独立进程的 HTTP 服务，接口形状同真模型（OpenAI 兼容 chat/completions，产品代码只改 base_url 指过去，不 monkeypatch）。行为由请求控制、无副作用状态：按 model 名分派场景（如挂住 N 秒、可控失败、控制流式速率）。至少支持：挂住指定时长、可控失败（HTTP 5xx / 超时 / 坏响应）、控制 token 流速率。F3 停止测试、T11 错误文案测试、F4 重试测试都复用它，不另写。
+- **ECharts 原样渲染**：option 在后端已过 BANNED_KEYS（app/tools/charts.py），前端不做二次解释——不 eval、不执行 formatter 字符串、不动态拼 option，只"原样交给 ECharts"。
+- 测试：按需引入的 ECharts 组件覆盖**后端可能产出的全部图表类型**（从后端导出的类型清单比对，不是只测当前用例用到的几种）；渲染时不产生外部资源请求（Playwright 断言零外部请求，不只靠 CSP）。
+- T11 单独 PR，不跟 F3 合。
 | F4 | 层 1 面板 | 2–3 天 |
 | F5 | S 验收轮次、删 Streamlit | 3–5 天 |
 

tests/fake_llm.py

@@ -0,0 +1,222 @@
+"""假模型服务（F3 起所有异步测试的公共设施）：独立进程的 OpenAI 兼容 HTTP 服务。
+
+产品代码不改，只把 base_url 指过来（DEEPSEEK_BASE_URL=http://127.0.0.1:<port>/v1）。
+行为完全由请求决定，服务端没有状态：同样的 model 名 + 同样的提示词 → 同样的行为。
+测试重跑不用重置，多个后端 / worker 可以共享一个实例。
+
+model 名语法（解析器严格，见 parse_model）：
+    <scenario>[@<step>]
+    scenario：ok | hang-<N>s | fail-<code> | slow-<N>s
+      ok          每一步正常回复（固定剧本，见 reply_for）
+      hang-<N>s   在目标调用上先睡 N 秒再正常回复（测停止）
+      fail-<code> 目标调用返回 HTTP <code>（400/401/403/404/429/500/502/503）
+      slow-<N>s   每一次调用都先睡 N 秒（把事件拉开，测增量渲染）
+    step：plan | codegen | fix | answer（工作流）、act | final（ReAct）
+      不写 @step 时目标调用是"第一次调用"：工作流的 plan、ReAct 的 act（无状态可判定，不需要计数）。
+
+fail loud：认不出的东西一律 HTTP 400（不可重试，后端立刻以 http_400 失败，不会被当成 5xx 退避重试三次），
+body 写明原因——未知 scenario / step、拼错（hang-30 少了 s）、@step 不属于本次请求的链路
+（@codegen 用在 ReAct 上会永远匹配不上，静默不挂就是假绿）、认不出当前是哪一步（提示词可能改了）、
+请求要求 stream。没有"认不出就按默认走"这条路。
+
+运行：python -m tests.fake_llm --port 8790
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+import time
+import uuid
+from dataclasses import dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Optional
+
+WORKFLOW_STEPS = ("plan", "codegen", "fix", "answer")
+REACT_STEPS = ("act", "final")
+FIRST_STEP = {"workflow": "plan", "react": "act"}
+FAIL_CODES = (400, 401, 403, 404, 429, 500, 502, 503)
+MAX_SECONDS = 600
+
+# 剧本：单表数据集里 df 就是那张表（app/tools/dataset.py），行数对任何 CSV 都成立
+CODE = "result = int(len(df))"
+PLAN = "（fake）统计行数。"
+ANSWER = "（fake）结论：数据共有若干行。"
+FAIL_ANSWER = "（fake）没能得出结果。"
+
+# 步骤识别用的提示词片段。改了 app/nodes/* 或 app/react_agent.py 的提示词，这里会认不出 → 400，
+# tests/test_fake_llm.py 的真链路测试（两种 agent 都跑一遍）立刻红。
+MARK_REACT = "你是一个数据分析智能体"
+MARK_PLAN = "你是数据分析规划助手"
+MARK_FIX = "之前生成的 pandas 代码执行出错了，请修正它"
+MARK_CODEGEN = "根据分析计划写一段 pandas 代码"
+MARK_ANSWER_OK = "根据分析结果，用中文简洁、直接地回答用户问题"
+MARK_ANSWER_FAIL = "分析未能成功完成"
+
+
+class FakeError(Exception):
+    """认不出 / 不支持：返回 400 + 原因。"""
+
+
+@dataclass(frozen=True)
+class Scenario:
+    kind: str                    # ok | hang | fail | slow
+    value: int = 0               # 秒数或 HTTP 状态码
+    step: Optional[str] = None   # 目标步骤；None = 第一次调用
+
+    def targets(self, agent: str, step: str) -> bool:
+        if self.step is not None and self.step not in (WORKFLOW_STEPS if agent == "workflow" else REACT_STEPS):
+            raise FakeError(f"场景 @{self.step} 不属于 {agent} 链路，永远匹配不上——换成该链路的步骤名")
+        return step == (self.step or FIRST_STEP[agent])
+
+
+_MODEL_RE = re.compile(r"^(?P<kind>ok|hang|fail|slow)(?:-(?P<val>\d+)(?P<unit>s?))?(?:@(?P<step>[a-z]+))?$")
+
+
+def parse_model(name: str) -> Scenario:
+    m = _MODEL_RE.match(name or "")
+    if not m:
+        raise FakeError(f"无法解析 model 名 {name!r}：语法是 <ok|hang-Ns|fail-CODE|slow-Ns>[@step]")
+    kind, val, unit, step = m["kind"], m["val"], m["unit"], m["step"]
+    if step is not None and step not in WORKFLOW_STEPS + REACT_STEPS:
+        raise FakeError(f"未知步骤 @{step}：可选 {', '.join(WORKFLOW_STEPS + REACT_STEPS)}")
+    if kind == "ok":
+        if val is not None or step is not None:
+            raise FakeError(f"ok 不带参数：{name!r}")
+        return Scenario("ok")
+    if val is None:
+        raise FakeError(f"{kind} 需要参数：{name!r}")
+    n = int(val)
+    if kind in ("hang", "slow"):
+        if unit != "s":
+            raise FakeError(f"{kind} 的参数要带单位 s（如 {kind}-30s）：{name!r}")
+        if not 1 <= n <= MAX_SECONDS:
+            raise FakeError(f"{kind} 秒数要在 1–{MAX_SECONDS}：{name!r}")
+        if kind == "slow" and step is not None:
+            raise FakeError(f"slow 作用于每一次调用，不带 @step：{name!r}")
+        return Scenario(kind, n, step)
+    if unit:
+        raise FakeError(f"fail 的参数是 HTTP 状态码，不带单位：{name!r}")
+    if n not in FAIL_CODES:
+        raise FakeError(f"fail 只支持 {FAIL_CODES}：{name!r}")
+    return Scenario("fail", n, step)
+
+
+def _text(content) -> str:
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):   # OpenAI 的分段 content
+        return "".join(p.get("text", "") for p in content if isinstance(p, dict))
+    return ""
+
+
+def recognize(messages: list) -> tuple[str, str]:
+    """(agent, step)。认不出就抛 FakeError，绝不猜。"""
+    if not isinstance(messages, list) or not messages:
+        raise FakeError("请求里没有 messages")
+    whole = "\n".join(_text(m.get("content")) for m in messages if isinstance(m, dict))
+    if MARK_REACT in whole:
+        # 边界：只按"最后一条是不是 tool"分 act / final，适用于单轮工具调用的剧本。多轮（连调两次工具再收尾）
+        # 分不出第几轮——真要测时按 tool 消息个数分，仍然无状态，不要加计数器。
+        return "react", ("final" if messages[-1].get("role") == "tool" else "act")
+    for mark, step in ((MARK_PLAN, "plan"), (MARK_FIX, "fix"), (MARK_CODEGEN, "codegen"),
+                       (MARK_ANSWER_OK, "answer"), (MARK_ANSWER_FAIL, "answer")):
+        if mark in whole:
+            return "workflow", step
+    raise FakeError("无法识别当前步骤，提示词可能已变（更新 tests/fake_llm.py 的 MARK_*）")
+
+
+def reply_for(agent: str, step: str, messages: list) -> dict:
+    """固定剧本：返回 OpenAI 的 assistant message。"""
+    if agent == "react":
+        if step == "act":
+            return {"role": "assistant", "content": "", "tool_calls": [{
+                "id": "call_fake_1", "type": "function",
+                "function": {"name": "run_python_on_data", "arguments": json.dumps({"code": CODE})}}]}
+        return {"role": "assistant", "content": ANSWER}
+    if step == "plan":
+        return {"role": "assistant", "content": PLAN}
+    if step in ("codegen", "fix"):
+        return {"role": "assistant", "content": f"```python\n{CODE}\n```"}
+    whole = "\n".join(_text(m.get("content")) for m in messages)
+    return {"role": "assistant", "content": FAIL_ANSWER if MARK_ANSWER_FAIL in whole else ANSWER}
+
+
+def handle(body: dict, sleep=time.sleep) -> tuple[int, dict]:
+    """纯函数核心（HTTP 外壳只做收发），单元测试直接调它。"""
+    try:
+        if body.get("stream"):
+            raise FakeError("假服务不支持 stream=true（产品目前不用流式调用）")
+        sc = parse_model(body.get("model", ""))
+        agent, step = recognize(body.get("messages"))
+        hit = sc.targets(agent, step)
+    except FakeError as e:
+        return 400, {"error": {"message": f"[fake_llm] {e}", "type": "invalid_request_error", "code": "fake_llm"}}
+    if sc.kind == "slow" or (sc.kind == "hang" and hit):
+        sleep(sc.value)
+    if sc.kind == "fail" and hit:
+        return sc.value, {"error": {"message": f"[fake_llm] 按场景返回 {sc.value}", "type": "fake_failure",
+                                    "code": str(sc.value)}}
+    msg = reply_for(agent, step, body["messages"])
+    return 200, {
+        "id": f"chatcmpl-fake-{uuid.uuid4().hex[:8]}", "object": "chat.completion", "created": int(time.time()),
+        "model": body["model"],
+        "choices": [{"index": 0, "message": msg, "finish_reason": "tool_calls" if msg.get("tool_calls") else "stop"}],
+        "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+    }
+
+
+class _Handler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+
+    def _send(self, code: int, obj: dict) -> None:
+        data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        try:
+            self.wfile.write(data)
+        except (BrokenPipeError, ConnectionResetError):
+            pass   # 调用方已放弃（超时 / 取消）
+
+    def do_GET(self):  # noqa: N802
+        if self.path in ("/health", "/v1/health"):
+            return self._send(200, {"ok": True})
+        self._send(404, {"error": {"message": "not found"}})
+
+    def do_POST(self):  # noqa: N802
+        if self.path.rstrip("/") not in ("/v1/chat/completions", "/chat/completions"):
+            return self._send(404, {"error": {"message": f"[fake_llm] 未知路径 {self.path}"}})
+        try:
+            body = json.loads(self.rfile.read(int(self.headers.get("Content-Length") or 0)) or b"{}")
+        except ValueError:
+            return self._send(400, {"error": {"message": "[fake_llm] body 不是 JSON"}})
+        code, obj = handle(body)
+        if code == 400:
+            sys.stderr.write(f"{obj['error']['message']}\n")
+        self._send(code, obj)
+
+    def log_message(self, fmt, *args):  # 安静；400 已单独写 stderr
+        pass
+
+
+def make_server(host: str = "127.0.0.1", port: int = 0) -> ThreadingHTTPServer:
+    srv = ThreadingHTTPServer((host, port), _Handler)
+    srv.daemon_threads = True
+    return srv
+
+
+def main(argv=None) -> None:
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--host", default="127.0.0.1")
+    ap.add_argument("--port", type=int, default=8790)
+    a = ap.parse_args(argv)
+    srv = make_server(a.host, a.port)
+    print(f"fake_llm listening on http://{a.host}:{srv.server_address[1]}/v1", flush=True)
+    srv.serve_forever()
+
+
+if __name__ == "__main__":
+    main()