diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index 2c36cd90..d128ab85 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -332,7 +332,13 @@ async def stream_run(websocket: WebSocket, run_id: str) -> None: await websocket.accept() after = int(websocket.query_params.get("after", 0) or 0) + # 先占订阅位，再读历史。反过来的话，两步之间图正好跑出来的事件既不在 + # 历史里、也不在实时流里——永远丢了。实测就是这样：一次跑图的开头 + # 两三条 node.started/node.finished 消失，界面上表现为某个节点永远 + # 在转圈、它的子步骤跑到了顶层。 + subscription = bus.attach(run_id) try: + last_seq = after async with SessionLocal() as session: rows = await session.execute( select(RunEvent) @@ -340,6 +346,7 @@ async def stream_run(websocket: WebSocket, run_id: str) -> None: .order_by(RunEvent.seq) ) for event in rows.scalars(): + last_seq = max(last_seq, event.seq or 0) await websocket.send_json( { "seq": event.seq, @@ -358,8 +365,8 @@ async def stream_run(websocket: WebSocket, run_id: str) -> None: await websocket.close() return - stream = bus.subscribe(run_id) - pump = asyncio.create_task(_pump(websocket, stream)) + # 占位期间攒下的那几条历史里也有，按 seq 滤掉，否则前端同一步收两遍 + pump = asyncio.create_task(_pump(websocket, subscription.stream(after=last_seq))) try: # 客户端断开时读操作会抛异常，用它来结束这条连接 while True: @@ -375,6 +382,8 @@ async def stream_run(websocket: WebSocket, run_id: str) -> None: except Exception: # noqa: BLE001 with contextlib.suppress(Exception): await websocket.close() + finally: + subscription.close() async def _pump(websocket: WebSocket, stream: Any) -> None: diff --git a/backend/app/core/bus.py b/backend/app/core/bus.py index d46439f9..a94713a1 100644 --- a/backend/app/core/bus.py +++ b/backend/app/core/bus.py @@ -55,24 +55,70 @@ class EventBus: with contextlib.suppress(asyncio.QueueFull): sub.queue.put_nowait(None) - async def subscribe(self, run_id: str) -> AsyncIterator[RunEventModel]: + def attach(self, run_id: str) -> "Subscription": + """立刻占一个位，之后再慢慢消费。 + + 必须能在读历史**之前**占位。`subscribe()` 做不到这件事：它是异步 + 生成器，订阅者要等到第一次迭代才真正注册——而调用方通常先读完历史 + 才开始迭代。这中间发出的事件既不在历史里、也不在实时流里，就这么 + 没了。一次跑图从发起到订阅之间正好有几毫秒，丢掉的往往是开头那 + 两三条 node.started/node.finished，表现为"某个节点永远在转圈、 + 它的子步骤跑到了顶层"。 + """ sub = _Subscriber() - async with self._lock: - self._subs[run_id].add(sub) - try: - while True: - item = await sub.queue.get() - if item is None: - break - yield item - finally: - async with self._lock: - self._subs[run_id].discard(sub) - if not self._subs[run_id]: - self._subs.pop(run_id, None) + self._subs[run_id].add(sub) + return Subscription(self, run_id, sub) + + def _detach(self, run_id: str, sub: _Subscriber) -> None: + self._subs.get(run_id, set()).discard(sub) + if run_id in self._subs and not self._subs[run_id]: + self._subs.pop(run_id, None) + + async def subscribe(self, run_id: str) -> AsyncIterator[RunEventModel]: + """便捷入口：占位即消费。只适合不需要补历史的场景。""" + with self.attach(run_id) as sub: + async for event in sub.stream(): + yield event def subscriber_count(self, run_id: str) -> int: return len(self._subs.get(run_id, ())) +class Subscription: + """一个已经在收事件的订阅位。""" + + def __init__(self, bus: EventBus, run_id: str, sub: _Subscriber) -> None: + self._bus = bus + self._run_id = run_id + self._sub = sub + + @property + def dropped(self) -> int: + """因为消费太慢被丢掉的条数。不是 0 就说明这条时间线有洞。""" + return self._sub.dropped + + async def stream(self, after: int = 0) -> AsyncIterator[RunEventModel]: + """吐出实时事件。 + + after 用来去重：占位到读完历史之间发出的事件，两边都有一份。 + 按 seq 过滤掉已经补过的，前端就不会看到同一步出现两次。 + """ + while True: + item = await self._sub.queue.get() + if item is None: + break + if after and (item.seq or 0) <= after: + continue + yield item + + def close(self) -> None: + self._bus._detach(self._run_id, self._sub) + + def __enter__(self) -> "Subscription": + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + bus = EventBus() diff --git a/backend/tests/test_bus_attach.py b/backend/tests/test_bus_attach.py new file mode 100644 index 00000000..73eb8f3e --- /dev/null +++ b/backend/tests/test_bus_attach.py @@ -0,0 +1,144 @@ +"""事件总线的订阅时序。 + +这里守的是一个只在**并发**下才出现的缺陷：先读历史、再订阅，两步之间发出的 +事件既不在历史里、也不在实时流里，永远丢了。串行地测"发一条收一条"永远 +测不出来——必须在"读历史"的那一刻往总线里灌事件。 + +真实表现：一次跑图开头两三条 node.started/node.finished 消失，界面上某个节点 +永远在转圈，它的子步骤跑到了顶层。 +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from app.core.bus import EventBus +from app.core.events import RunEventModel + + +def _event(run_id: str, seq: int, type_: str = "node.started") -> RunEventModel: + return RunEventModel(run_id=run_id, seq=seq, type=type_, node_id=None, data={}) + + +@pytest.mark.asyncio +async def test_attach_before_history_loses_nothing() -> None: + bus = EventBus() + run_id = "r1" + got: list[int] = [] + + with bus.attach(run_id) as sub: + # 占位之后、开始消费之前发出的事件要留在队列里等着，不能丢 + await bus.publish(_event(run_id, 1)) + await bus.publish(_event(run_id, 2)) + + async def drain() -> None: + async for event in sub.stream(): + got.append(event.seq) + + task = asyncio.create_task(drain()) + await asyncio.sleep(0) + await bus.publish(_event(run_id, 3)) + await bus.close(run_id) + await asyncio.wait_for(task, timeout=2) + + assert got == [1, 2, 3] + + +@pytest.mark.asyncio +async def test_after_filter_dedups_overlap() -> None: + """占位到读完历史之间的事件两边都有一份，按 seq 滤掉重复的那份。""" + bus = EventBus() + run_id = "r2" + got: list[int] = [] + + with bus.attach(run_id) as sub: + for seq in (1, 2, 3): + await bus.publish(_event(run_id, seq)) + + # 假设补历史时读到了 seq<=2，那实时流里只该再吐 3 + async def drain() -> None: + async for event in sub.stream(after=2): + got.append(event.seq) + + task = asyncio.create_task(drain()) + await asyncio.sleep(0) + await bus.close(run_id) + await asyncio.wait_for(task, timeout=2) + + assert got == [3] + + +@pytest.mark.asyncio +async def test_detach_removes_subscriber() -> None: + bus = EventBus() + run_id = "r3" + sub = bus.attach(run_id) + assert bus.subscriber_count(run_id) == 1 + sub.close() + assert bus.subscriber_count(run_id) == 0 + + +@pytest.mark.asyncio +async def test_subscribe_registers_late() -> None: + """刻画 subscribe() 的行为，说明 attach 为什么必须存在。 + + subscribe() 是异步生成器：拿到它并不等于订阅上了，要到第一次迭代才真正 + 注册。调用方"先拿生成器、再读历史、最后开始迭代"是最自然的写法，而那 + 恰好让整个读历史的窗口都处于无人接收的状态。 + """ + bus = EventBus() + run_id = "r5" + gen = bus.subscribe(run_id) + assert bus.subscriber_count(run_id) == 0, "拿到生成器还不算订阅" + + # 这条发出去时没人收——旧代码丢事件就是丢在这个窗口里 + await bus.publish(_event(run_id, 1)) + + got: list[int] = [] + + async def drain() -> None: + async for event in gen: + got.append(event.seq) + + task = asyncio.create_task(drain()) + await asyncio.sleep(0.01) + assert bus.subscriber_count(run_id) == 1, "迭代之后才注册上" + await bus.publish(_event(run_id, 2)) + await bus.close(run_id) + await asyncio.wait_for(task, timeout=2) + + assert got == [2], "seq=1 确实丢了；attach 就是为了消掉这个窗口" + + +@pytest.mark.asyncio +async def test_concurrent_publish_during_history_read() -> None: + """把真实时序演一遍：占位 → 慢慢读历史（期间图还在跑）→ 开始消费。""" + bus = EventBus() + run_id = "r4" + history: list[int] = [] + live: list[int] = [] + + with bus.attach(run_id) as sub: + async def keep_running() -> None: + # 图不会等前端读完历史 + for seq in range(1, 6): + await bus.publish(_event(run_id, seq)) + await asyncio.sleep(0.005) + await bus.close(run_id) + + runner = asyncio.create_task(keep_running()) + + # "读历史"——只读到前两条，剩下的还没落库 + await asyncio.sleep(0.012) + history.extend([1, 2]) + + async def drain() -> None: + async for event in sub.stream(after=max(history)): + live.append(event.seq) + + await asyncio.wait_for(asyncio.gather(runner, drain()), timeout=3) + + # 3/4/5 一条都不能少——这正是修复前丢掉的那些 + assert live == [3, 4, 5], f"历史={history} 实时={live}" diff --git a/frontend/src/canvas/NodeCard.tsx b/frontend/src/canvas/NodeCard.tsx index e236efb2..b68b61f4 100644 --- a/frontend/src/canvas/NodeCard.tsx +++ b/frontend/src/canvas/NodeCard.tsx @@ -3,6 +3,7 @@ import { Handle, Position, type NodeProps } from '@xyflow/react' import { AlertTriangle, Check, Hand, Loader2, Sparkles, Wrench } from 'lucide-react' import clsx from 'clsx' import { NODE_DEFS, sourceHandles } from './nodeDefs' +import { formatDuration } from '../run/decode' import { useStudio, type FlowNode } from '../store/studio' import type { NodeType } from '../types' @@ -149,7 +150,7 @@ function NodeCardImpl({ id, data, selected }: NodeProps) { {(runtime?.durationMs != null || runtime?.error) && (
{runtime.durationMs != null && ( - {formatMs(runtime.durationMs)} + {formatDuration(runtime.durationMs)} )} {runtime.error && ( @@ -185,10 +186,7 @@ function NodeCardImpl({ id, data, selected }: NodeProps) { ) } -export function formatMs(ms: number): string { - if (ms < 1000) return `${ms}ms` - if (ms < 60_000) return `${(ms / 1000).toFixed(1)}s` - return `${Math.floor(ms / 60_000)}m${Math.round((ms % 60_000) / 1000)}s` -} - +// 耗时格式化原本在这里又写了一份（formatMs），和 decode.ts 的规则不一样： +// 画布节点上显示"0ms"，右边助手栏里同一步什么都不显示。两块并排放着， +// 用户看到的是两套说法。统一用 decode.ts 那份。 export const NodeCard = memo(NodeCardImpl) diff --git a/frontend/src/dev/preview.tsx b/frontend/src/dev/preview.tsx index 154baedf..71ffebaa 100644 --- a/frontend/src/dev/preview.tsx +++ b/frontend/src/dev/preview.tsx @@ -26,6 +26,7 @@ const CASES: { key: Bucket; question: string; phase: StreamTurn['phase']; status { key: 'human', question: '把这批客户的信用额度调高 10%', phase: 'waiting', status: '等待你的确认' }, { key: 'issue', question: '出一份本月经营分析', phase: 'done', status: '完成' }, { key: 'failed', question: '查一下不存在的那张表', phase: 'error', status: '' }, + { key: 'loop_approve', question: '写条上线公告，我过一眼再发', phase: 'done', status: '完成' }, ] function buildTurns(): StreamTurn[] { diff --git a/frontend/src/pages/ChatPage.tsx b/frontend/src/pages/ChatPage.tsx index 9a7d2937..6f6a51eb 100644 --- a/frontend/src/pages/ChatPage.tsx +++ b/frontend/src/pages/ChatPage.tsx @@ -1,14 +1,13 @@ -import { useEffect, useMemo, useRef, useState } from 'react' -import { - ChevronRight, Database, MessageSquare, Send, Square, Table2, Trash2, Wrench, XCircle, -} from 'lucide-react' -import clsx from 'clsx' +import { useEffect, useMemo, useState } from 'react' +import { Database, MessageSquare, Send, Square, Trash2 } from 'lucide-react' import { api } from '../api/client' import { ApprovalCard } from '../run/RunPanel' +import { AssistantStream, StreamEmpty, type StreamTurn } from '../run/AssistantStream' +import { decodeCopilot, decodeRun } from '../run/decode' import { useCatalog } from '../store/catalog' import { useChat, type ChatTurn } from '../store/chat' -import { useStudio, toFlow } from '../store/studio' -import { Empty, Spinner, useToast } from '../components/ui' +import { useStudio } from '../store/studio' +import { useToast } from '../components/ui' /** * 对话式入口。 @@ -19,27 +18,29 @@ import { Empty, Spinner, useToast } from '../components/ui' * * 所以这一页把图藏起来：你问，它自己接数据源、建流程、跑完、给结论。图仍然 * 真实存在（可以展开看，也能拿到画布里继续改），只是不再是必经之路。 + * + * 这里和画布右栏用的是同一个 AssistantStream、同一个 decode.ts——差别只有 + * dense 一个开关。两边各写一套的话，同一次运行在两个页面会讲出不同的故事。 */ export function ChatPage() { const { turns, busy, ask, stop, clear } = useChat() const [draft, setDraft] = useState('') - const bottomRef = useRef(null) const sources = useDataSources() + const setGraph = useStudio((s) => s.setGraph) + const toast = useToast() - useEffect(() => { - bottomRef.current?.scrollIntoView({ behavior: 'smooth' }) - }, [turns.length, turns[turns.length - 1]?.phase]) - - const send = () => { - const q = draft.trim() + const send = (text?: string) => { + const q = (text ?? draft).trim() if (!q || busy) return ask(q) setDraft('') } + const streamTurns = useMemo(() => turns.map(toStreamTurn), [turns]) + return (
-
+
问数据 @@ -53,220 +54,93 @@ export function ChatPage() { )}
-
- {!turns.length ? ( - { setDraft(q); }} /> - ) : ( -
- {turns.map((turn) => )} + } + approvalsFor={(t) => } + onOpenGraph={(graph) => { + setGraph(graph) + toast('已放到画布，切到「编排」继续改', 'ok') + }} + footer={ +
+
+