Copilot 等待期不再是黑箱，外加项目第一组测试

从点"生成"到第一个节点落到画布上，本机实测 31.7 秒（DeepSeek V4 pro，中等复杂度需求）。这段时间界面上只有"正在起草…"五个字一动不动——用户没法判断是在想还是已经卡死。生成质量本身没问题，瓶颈全在这段等待的体验。三处改动： ## 思考转发 _iter_ops 原先只取正文、丢掉其余一切，模型的 thinking 块也一并丢了。现在用 thinking_text() 抽出来单独转成 thinking 事件，前端可折叠地实时显示。刻意不用"把非 JSON 行当成思考"这种取巧写法：那会把模型的 markdown 注释、协议跑偏时的乱输出一并当作思考展示出来，反而误导。不支持 thinking 的模型就只有心跳，不硬凑假的思考。注意这条只对支持 adaptive thinking 的模型（Claude 4.6+）生效，当前默认的 DeepSeek V4 pro 不会产出——代码路径用构造的 chunk 验证过，但端到端触发需要启用 Anthropic provider。 ## 心跳与阶段模型沉默时每 3 秒补一拍，带阶段和已用时长。必须在 _with_heartbeat 这一层做：模型不吐 chunk 时 _iter_ops 里那个 async for 的循环体根本不执行，在里面判断时间差轮不到。阶段跟着实际操作走（planning → building → wiring → finalizing），阶段会变本身就是"它还活着"的信号。实测那 31.7 秒里补了 10 拍。 ## 失败后不用重敲需求生成失败只有一个 ✕，重试得重新打开 Modal 把需求再敲一遍——而失败多半跟需求本身无关（模型抽风、协议跑偏、网断了）。store 里留住 lastInstruction，错误浮条上直接给"用同一需求重试"。 ## 顺带：项目第一组测试这组守的是体验契约不是实现细节——thinking 和 heartbeat 两路反馈只要断了就是回归，而这种回归在界面上表现为"又变回干等"，很难被注意到。4 个用例覆盖： thinking 块转发、纯文本模型不伪造思考、沉默期补心跳、阶段随进度推进。 pytest 走 [dev] extra，environment.yml 一并带上。
2026-11-15 00:11:59 +00:00 · 2026-09-14 16:01:06 +08:00
parent b634c89be4
commit 1756476a30
7 changed files with 324 additions and 27 deletions
@@ -1,6 +1,8 @@
 from __future__ import annotations

+import asyncio
 import json
+import time
 from typing import Any

 from fastapi import APIRouter, Depends, HTTPException
@@ -378,11 +380,25 @@ def _apply_op(nodes: dict[str, dict[str, Any]], edges: list[dict[str, Any]], op:


 async def _iter_ops(model: Any, messages: list[Any]):
-    """流式读模型输出，按行切出操作。done 之后即停——后面就算有闲话也不要了。"""
+    """流式读模型输出，按行切出操作。done 之后即停——后面就算有闲话也不要了。
+
+    除了操作，还把模型的思考原样转出去。原先这里只取正文、丢掉其余一切，
+    于是从请求发出到第一个操作到达之间（实测 5~30 秒）前端一个字都收不到，
+    只能干挂着"正在起草…"——用户没法判断是在想、还是已经卡死了。
+
+    思考走 thinking_text()（Claude 4.6+ 的 thinking 块），不是"把非 JSON 行
+    当成思考"：后者会把模型的 markdown 注释、协议跑偏时的乱输出一并当作思考
+    展示出来，反而误导。不支持 thinking 的模型就只有心跳，不硬凑。
+    """
    from app.engine.state import message_text as _text
+    from app.engine.state import thinking_text

    buffer = ""
    async for chunk in model.astream(messages):
+        reasoning = thinking_text(chunk)
+        if reasoning:
+            yield {"op": "thinking", "delta": reasoning}
+
        buffer += _text(chunk)
        while "\n" in buffer:
            line, buffer = buffer.split("\n", 1)
@@ -396,6 +412,45 @@ async def _iter_ops(model: Any, messages: list[Any]):
        yield op


+# 多久没动静就补一拍心跳。3 秒是"用户开始怀疑是不是卡了"的量级。
+_HEARTBEAT_SECONDS = 3.0
+
+
+async def _with_heartbeat(source: Any):
+    """模型沉默时按拍补心跳，让前端能显示阶段和已用时长。
+
+    必须在这一层做而不是在 _iter_ops 里判断时间差：模型不吐 chunk 时那个
+    async for 的循环体根本不执行，压根轮不到检查。
+    """
+    started = time.monotonic()
+    phase = "planning"
+    iterator = source.__aiter__()
+    while True:
+        pending = asyncio.ensure_future(iterator.__anext__())
+        while True:
+            done, _ = await asyncio.wait({pending}, timeout=_HEARTBEAT_SECONDS)
+            if pending in done:
+                try:
+                    op = pending.result()
+                except StopAsyncIteration:
+                    return
+                kind = op.get("op")
+                # 阶段跟着实际操作走，前端据此换文案
+                if kind == "add_node":
+                    phase = "building"
+                elif kind in ("add_edge", "connect"):
+                    phase = "wiring"
+                elif kind == "done":
+                    phase = "finalizing"
+                yield op
+                break
+            yield {
+                "op": "heartbeat",
+                "phase": phase,
+                "elapsed_ms": int((time.monotonic() - started) * 1000),
+            }
+
+
@router.post("/generate-stream")
 async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(get_session)):
    """流式版生成：SSE 逐操作推送，前端边收边把节点摆上画布。
@@ -451,11 +506,16 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
        explanation = ""
        yield f"data: {json.dumps({'op': 'model', 'model': model_id}, ensure_ascii=False)}\n\n"
        try:
-            async for op in _iter_ops(model, messages):
-                if op.get("op") == "done":
+            async for op in _with_heartbeat(_iter_ops(model, messages)):
+                kind = op.get("op")
+                # 思考和心跳不是图操作，不进 _apply_op，直接转给前端
+                if kind in ("thinking", "heartbeat"):
+                    yield f"data: {json.dumps(op, ensure_ascii=False)}\n\n"
+                    continue
+                if kind == "done":
                    explanation = str(op.get("explanation", ""))
                changed = _apply_op(nodes, edges, op)
-                if changed or op.get("op") in ("plan", "done"):
+                if changed or kind in ("plan", "done"):
                    yield f"data: {json.dumps(op, ensure_ascii=False)}\n\n"
        except Exception as e:  # noqa: BLE001
            yield f"data: {json.dumps({'op': 'error', 'message': f'{type(e).__name__}: {e}'}, ensure_ascii=False)}\n\n"
@@ -31,6 +31,8 @@ dependencies = [
 # 再加一次 OCI 镜像拉取（实测约 55s），所以不进默认安装。
 # environment.yml 里已经带上了；单独装用 `pip install 'agentlab-backend[microvm]'`。
 microvm = ["microsandbox>=0.6"]
+# 测试：conda 环境默认不装，跑测试前 pip install 'agentlab-backend[dev]'
+dev = ["pytest>=8.0", "pytest-asyncio>=0.24"]

 [build-system]
 requires = ["hatchling"]
@@ -0,0 +1,5 @@
+[pytest]
+# asyncio_mode=auto 省掉每个协程测试都写一遍 @pytest.mark.asyncio
+asyncio_mode = auto
+testpaths = tests
+addopts = -q --tb=short
@@ -0,0 +1,115 @@
+"""Copilot 流式生成：等待期间必须有反馈。
+
+这组测试守的是一个体验契约，不是实现细节：从请求发出到第一个节点落到画布上，
+本机实测有 31.7 秒（DeepSeek V4 pro，中等复杂度的需求）。这段时间里前端如果
+一个事件都收不到，用户没法判断是在想还是已经卡死——所以 thinking 和 heartbeat
+这两路反馈只要断了，就是回归。
+"""
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from app.api import copilot as cp
+from app.api.copilot import _iter_ops, _with_heartbeat
+
+
+class Chunk:
+    """模拟流式 chunk。content 是块列表时对应 Claude 4.6+ 的 thinking 混排。"""
+
+    def __init__(self, content):
+        self.content = content
+
+
+class FakeModel:
+    def __init__(self, chunks, delay: float = 0.0):
+        self._chunks = chunks
+        self._delay = delay
+
+    async def astream(self, messages):
+        for chunk in self._chunks:
+            if self._delay:
+                await asyncio.sleep(self._delay)
+            yield chunk
+
+
+async def _collect(gen):
+    return [op async for op in gen]
+
+
+@pytest.mark.asyncio
+async def test_thinking_blocks_become_events():
+    """支持 thinking 的模型，思考要原样转给前端。"""
+    model = FakeModel([
+        Chunk([{"type": "thinking", "thinking": "用户要一个调研流程。"}]),
+        Chunk([{"type": "thinking", "thinking": "先搜索，再抓取。"}]),
+        Chunk([{"type": "text", "text": '{"op":"plan","summary":"三步"}\n'}]),
+        Chunk([{"type": "text", "text": '{"op":"done","explanation":"完成"}\n'}]),
+    ])
+    ops = await _collect(_iter_ops(model, []))
+    kinds = [o["op"] for o in ops]
+
+    assert kinds.count("thinking") == 2
+    assert ops[0]["delta"] == "用户要一个调研流程。"
+    assert kinds[-1] == "done", "done 之后就该停，后面的闲话不要"
+
+
+@pytest.mark.asyncio
+async def test_plain_text_is_not_passed_off_as_thinking():
+    """不支持 thinking 的模型不该被凭空造出思考。
+
+    有一种取巧写法是"把非 JSON 行当成思考"，那会把模型的 markdown 注释、
+    协议跑偏时的乱输出一并当作思考展示——宁可只有心跳，也不要假的思考。
+    """
+    model = FakeModel([
+        Chunk("模型的随口注释，不是 JSON\n"),
+        Chunk('{"op":"plan","summary":"x"}\n'),
+        Chunk('{"op":"done","explanation":"y"}\n'),
+    ])
+    kinds = [o["op"] for o in await _collect(_iter_ops(model, []))]
+
+    assert "thinking" not in kinds
+    assert kinds == ["plan", "done"]
+
+
+@pytest.mark.asyncio
+async def test_heartbeat_fires_while_model_is_silent(monkeypatch):
+    """模型长时间不吐字时要补心跳，否则前端一片死寂。
+
+    心跳必须在 _with_heartbeat 这一层做：模型不产出 chunk 时，_iter_ops 里那个
+    async for 的循环体根本不执行，在里面判断时间差是轮不到的。
+    """
+    monkeypatch.setattr(cp, "_HEARTBEAT_SECONDS", 0.2)
+
+    class SlowModel:
+        async def astream(self, messages):
+            await asyncio.sleep(0.9)  # 模拟"正在想"
+            yield Chunk('{"op":"plan","summary":"终于"}\n')
+            yield Chunk('{"op":"done","explanation":"z"}\n')
+
+    ops = await _collect(_with_heartbeat(_iter_ops(SlowModel(), [])))
+    beats = [o for o in ops if o["op"] == "heartbeat"]
+
+    assert len(beats) >= 2, "0.9 秒的沉默至少该有两拍"
+    assert beats[0]["phase"] == "planning"
+    assert beats[0]["elapsed_ms"] > 0
+    assert [o["op"] for o in ops][-1] == "done"
+
+
+@pytest.mark.asyncio
+async def test_heartbeat_phase_follows_progress(monkeypatch):
+    """阶段要跟着实际操作走——阶段会变本身就是"它还活着"的信号。"""
+    monkeypatch.setattr(cp, "_HEARTBEAT_SECONDS", 0.2)
+
+    class Staged:
+        async def astream(self, messages):
+            yield Chunk('{"op":"add_node","node":{"id":"a","type":"input"}}\n')
+            await asyncio.sleep(0.5)  # 放完节点后卡一会儿
+            yield Chunk('{"op":"done","explanation":"d"}\n')
+
+    ops = await _collect(_with_heartbeat(_iter_ops(Staged(), [])))
+    beats = [o for o in ops if o["op"] == "heartbeat"]
+
+    assert beats, "add_node 之后的停顿也该有心跳"
+    assert beats[0]["phase"] == "building", "放过节点了，阶段不该还停在 planning"
@@ -15,4 +15,4 @@ dependencies:
      # [microvm] 带上独立内核的沙箱后端（约 35MB wheel，首次运行还要下载
      # 约 50MB 运行时，外加一次 OCI 镜像拉取）。不需要就把方括号去掉，
      # 沙箱会退回 Seatbelt —— 代价是内存限额在 macOS 上限不住。
-      - -e ./backend[microvm]
+      - -e ./backend[microvm,dev]
@@ -1,6 +1,7 @@
 import { useCallback, useEffect, useRef, useState } from 'react'
 import {
-  AlertTriangle, Check, ChevronDown, Copy, LayoutGrid, Plus, Save, ShieldCheck, Sparkles, Trash2, Wand2,
+  AlertTriangle, Check, ChevronDown, ChevronRight, Copy, LayoutGrid, Plus, Save, ShieldCheck,
+  Sparkles, Trash2, Wand2,
 } from 'lucide-react'
 import clsx from 'clsx'
 import { api } from '../api/client'
@@ -475,36 +476,101 @@ function CopilotModal({ open, onClose }: { open: boolean; onClose: () => void })
 }

 /** Copilot 工作时的画布浮条：显示当前操作，可随时取消；结束后展示说明。 */
+// 阶段文案。从提交到第一个节点落地中间有 5~30 秒，这段时间界面上只有
+// "正在起草…"三个字一动不动，用户分不清是在想还是已经卡死——阶段会变的
+// 本身就是"它还活着"的信号。
+const COPILOT_PHASES: Record<string, string> = {
+  connecting: '正在连接模型…',
+  planning: '正在理解需求、规划结构…',
+  building: '正在放置节点…',
+  wiring: '正在连接数据流…',
+  finalizing: '正在排版和校验…',
+}
+
 function CopilotStatusBar() {
  const copilot = useStudio((s) => s.copilot)
  const stopCopilot = useStudio((s) => s.stopCopilot)
+  const retryCopilot = useStudio((s) => s.retryCopilot)
  const [dismissed, setDismissed] = useState(false)
+  const [showThinking, setShowThinking] = useState(true)
+  const thinkingRef = useRef<HTMLDivElement>(null)

  useEffect(() => {
    // 新一轮生成开始时重新显示
    if (copilot.active) setDismissed(false)
  }, [copilot.active])

+  // 思考是流式追加的，跟着滚到底，否则用户只能看到最早那几行
+  useEffect(() => {
+    const el = thinkingRef.current
+    if (el) el.scrollTop = el.scrollHeight
+  }, [copilot.thinking])
+
  if (dismissed) return null
  if (!copilot.active && !copilot.explanation && !copilot.error) return null

+  const seconds = copilot.elapsedMs ? Math.round(copilot.elapsedMs / 1000) : 0
+  // 有具体操作就显示操作，还没开始动就显示阶段——不再是恒定的"正在起草…"
+  const headline = copilot.lastOp || COPILOT_PHASES[copilot.phase] || '正在起草…'
+
  return (
    <div className="fade-up pointer-events-auto absolute bottom-4 left-4 z-30 max-w-md rounded-lg border bg-panel px-3 py-2 shadow-xl"
         style={{ borderColor: copilot.error ? 'var(--err)' : '#bc8cff' }}>
      {copilot.active ? (
-        <div className="flex items-center gap-2">
-          <Spinner size={13} />
-          <span className="min-w-0 flex-1 truncate text-[11.5px]">
-            <span className="mr-1.5 font-semibold" style={{ color: '#bc8cff' }}>Copilot</span>
-            {copilot.model && <span className="mr-1.5 text-faint">{copilot.model}</span>}
-            {copilot.lastOp}
-          </span>
-          <button className="btn btn-sm" onClick={stopCopilot}>取消</button>
+        <div>
+          <div className="flex items-center gap-2">
+            <Spinner size={13} />
+            <span className="min-w-0 flex-1 truncate text-[11.5px]">
+              <span className="mr-1.5 font-semibold" style={{ color: '#bc8cff' }}>Copilot</span>
+              {copilot.model && <span className="mr-1.5 text-faint">{copilot.model}</span>}
+              {headline}
+            </span>
+            {seconds > 0 && <span className="mono text-[10px] text-faint">{seconds}s</span>}
+            <button className="btn btn-sm" onClick={stopCopilot}>取消</button>
+          </div>
+
+          {/* 支持 thinking 的模型（Claude 4.6+）才有内容；没有就只显示阶段 */}
+          {copilot.thinking && (
+            <div className="mt-1.5 border-t pt-1.5">
+              <button
+                className="mb-1 flex items-center gap-1 text-[10px] text-faint hover:text-dim"
+                onClick={() => setShowThinking((v) => !v)}
+              >
+                <ChevronRight
+                  size={10}
+                  style={{ transform: showThinking ? 'rotate(90deg)' : 'none', transition: 'transform .15s' }}
+                />
+                模型正在想什么
+              </button>
+              {showThinking && (
+                <div
+                  ref={thinkingRef}
+                  className="max-h-28 overflow-y-auto whitespace-pre-wrap rounded bg-bg px-2 py-1.5 text-[10.5px] leading-relaxed text-dim"
+                >
+                  {copilot.thinking}
+                </div>
+              )}
+            </div>
+          )}
        </div>
      ) : copilot.error ? (
-        <div className="flex items-start gap-2">
-          <span className="min-w-0 flex-1 text-[11.5px] text-[var(--err)]">{copilot.error}</span>
-          <button className="btn btn-ghost btn-sm" onClick={() => setDismissed(true)}>✕</button>
+        <div>
+          <div className="flex items-start gap-2">
+            <span className="min-w-0 flex-1 text-[11.5px] text-[var(--err)]">{copilot.error}</span>
+            <button className="btn btn-ghost btn-sm" onClick={() => setDismissed(true)}>✕</button>
+          </div>
+          {/* 失败多半跟需求本身无关（模型抽风、协议跑偏、网断了），
+              不该逼用户重新打开 Modal 把需求再敲一遍 */}
+          {copilot.lastInstruction && (
+            <div className="mt-1.5 flex items-center gap-2 border-t pt-1.5">
+              <span className="min-w-0 flex-1 truncate text-[10px] text-faint" title={copilot.lastInstruction}>
+                {copilot.lastInstruction}
+              </span>
+              <button className="btn btn-sm" onClick={() => { setDismissed(false); retryCopilot() }}>
+                用同一需求重试
+              </button>
+            </div>
+          )}
        </div>
      ) : (
        <div className="flex items-start gap-2">
@@ -78,7 +78,19 @@ interface StudioState {
  streaming: boolean
  unsubscribe: (() => void) | null

-  copilot: { active: boolean; lastOp: string; explanation: string; error: string; model: string }
+  copilot: {
+    active: boolean
+    lastOp: string
+    explanation: string
+    error: string
+    model: string
+    // 从提交到第一个操作之间有 5~30 秒，得让用户看见这段在发生什么
+    phase: 'connecting' | 'planning' | 'building' | 'wiring' | 'finalizing' | ''
+    thinking: string        // 模型的思考原文（只有支持 thinking 的模型有）
+    elapsedMs: number
+    lastInstruction: string // 失败后"用同一需求重试"要用
+    lastUseBase: boolean
+  }
  copilotNew: string[]
  cancelCopilot: (() => void) | null

@@ -100,6 +112,7 @@ interface StudioState {
  startFormalRun: (input: Record<string, any>) => Promise<Run | null>
  runCopilot: (instruction: string, useBase: boolean, model?: string | null) => void
  stopCopilot: () => void
+  retryCopilot: () => void
  attachRun: (runId: string) => Promise<void>
  stopRun: () => Promise<void>
  clearRun: () => void
@@ -121,7 +134,8 @@ export const useStudio = create<StudioState>((set, get) => ({
  activeEdges: [],
  streaming: false,
  unsubscribe: null,
-  copilot: { active: false, lastOp: '', explanation: '', error: '', model: '' },
+  copilot: { active: false, lastOp: '', explanation: '', error: '', model: '',
+             phase: '', thinking: '', elapsedMs: 0, lastInstruction: '', lastUseBase: false },
  copilotNew: [],
  cancelCopilot: null,

@@ -297,7 +311,11 @@ export const useStudio = create<StudioState>((set, get) => ({
    const state = get()
    state.cancelCopilot?.()
    set({
-      copilot: { active: true, lastOp: '正在起草…', explanation: '', error: '', model: model ?? '' },
+      copilot: {
+        active: true, lastOp: '', explanation: '', error: '', model: model ?? '',
+        phase: 'connecting', thinking: '', elapsedMs: 0,
+        lastInstruction: instruction, lastUseBase: useBase,
+      },
      copilotNew: [],
    })
    if (!useBase) set({ nodes: [], edges: [], dirty: true })
@@ -327,8 +345,28 @@ export const useStudio = create<StudioState>((set, get) => ({
            // 后端首帧告知实际用的模型，浮条上直接显示，不用猜
            set({ copilot: { ...s.copilot, model: op.model ?? '' } })
            break
+          case 'thinking':
+            // 模型正在想什么，原样接上去。看着它想比盯着"正在起草…"强得多
+            set({
+              copilot: {
+                ...s.copilot,
+                phase: s.copilot.phase === 'connecting' ? 'planning' : s.copilot.phase,
+                thinking: (s.copilot.thinking + (op.delta ?? '')).slice(-4000),
+              },
+            })
+            break
+          case 'heartbeat':
+            // 模型沉默期间的补拍，用来刷新阶段和已用时长
+            set({
+              copilot: {
+                ...s.copilot,
+                phase: (op.phase as any) ?? s.copilot.phase,
+                elapsedMs: op.elapsed_ms ?? s.copilot.elapsedMs,
+              },
+            })
+            break
          case 'plan':
-            set({ copilot: { ...s.copilot, lastOp: op.summary ?? '规划中' } })
+            set({ copilot: { ...s.copilot, phase: 'planning', lastOp: op.summary ?? '规划中' } })
            break
          case 'add_node': {
            const n = op.node
@@ -393,21 +431,24 @@ export const useStudio = create<StudioState>((set, get) => ({
            })
            break
          case 'done':
-            set({ copilot: { ...s.copilot, lastOp: '排版整理中…', explanation: op.explanation ?? '' } })
+            set({ copilot: { ...s.copilot, phase: 'finalizing', lastOp: '排版整理中…',
+                             explanation: op.explanation ?? '' } })
            break
          case 'final': {
            // 后端排版+校验后的最终图整体落位；高亮集合保留几秒供辨认
            const { nodes, edges } = toFlow(op.graph)
            set({ nodes, edges, dirty: true,
-                  copilot: { active: false, lastOp: '', model: get().copilot.model,
-                             explanation: op.explanation ?? get().copilot.explanation, error: '' } })
+                  copilot: { ...get().copilot, active: false, lastOp: '', phase: '',
+                             thinking: '', elapsedMs: 0, error: '',
+                             explanation: op.explanation ?? get().copilot.explanation } })
            void get().validate()
            setTimeout(() => set({ copilotNew: [] }), 6000)
            break
          }
          case 'error':
-            set({ copilot: { active: false, lastOp: '', explanation: '',
-                             model: get().copilot.model, error: op.message ?? '生成失败' } })
+            // 保留 lastInstruction：失败浮条上的"重试"要用它，不能让用户重填
+            set({ copilot: { ...get().copilot, active: false, lastOp: '', phase: '',
+                             thinking: '', explanation: '', error: op.message ?? '生成失败' } })
            break
        }
      },
@@ -425,11 +466,19 @@ export const useStudio = create<StudioState>((set, get) => ({
  stopCopilot: () => {
    get().cancelCopilot?.()
    set({
-      copilot: { ...get().copilot, active: false },
+      copilot: { ...get().copilot, active: false, phase: '', thinking: '' },
      cancelCopilot: null,
    })
  },

+  retryCopilot: () => {
+    // 失败后用同一条需求重来。以前只能重新打开 Modal 把需求再敲一遍，
+    // 而失败往往跟需求本身无关（模型抽风、协议跑偏、网络断了）
+    const { lastInstruction, lastUseBase, model } = get().copilot
+    if (!lastInstruction) return
+    get().runCopilot(lastInstruction, lastUseBase, model || null)
+  },
+
  startFormalRun: async (input) => {
    const { workflow } = get()
    if (!workflow) return null