From aa18904af04cdfbd292c8ee0dfa19a15560f5058 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 18 Sep 2026 09:39:24 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=90=E8=A1=8C=E5=92=8C=20Copilot=20?= =?UTF-8?q?=E5=88=86=E6=88=90=E4=B8=A4=E5=B1=82=EF=BC=9B=E9=A1=BA=E5=B8=A6?= =?UTF-8?q?=E4=BF=AE=E4=BA=86=E4=B8=80=E4=B8=AA=E4=BC=9A=E4=B8=A2=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E7=9A=84=E6=97=B6=E5=BA=8F=E7=BC=BA=E9=99=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户：「点击右上角运行后，会跳到流程；然后下方就又有 Copilot 对话框，把 结果和 Copilot 又放在一块儿了」。说得对，而且我上一版的理由是错的。 我当时把建图和跑图排进同一条时间轴，说它们是"一件事的两半"。它们在时间上 先后发生，但不是同一种东西——最直接的证据在我自己的代码里：运行那一轮的 question 只能填 undefined。一个没有提问的"对话轮次"，本来就不是对话。 更实际的问题是那个输入框：它贴在运行结果下面，暗示你可以回复这次运行。 但你不能——在那儿打字改的是画布上的图。这是个假的可供性。两者的生命周期 也对不上：Copilot 的轮次会累积，运行永远只有最新一次。 现在右栏是一个常驻的家（Copilot 对话）加两层临时盖上来的东西（节点属性、 运行过程）。运行那一层没有输入框，有自己的头部和停止按钮；对话那一层底部 留一条运行进度，点一下过去。 --- 顺着这条线查下去，发现一个更严重的问题：**后端在丢事件**。 浏览器侧连跑 5 次，2 次丢了中间一条。抓 WebSocket 原始帧才看清：所有帧的 replay 都是 false——历史读回来是空的，全部走实时；丢的是 run.started， 它在客户端订阅之前就广播了，而那时还没落库，所以历史里也没有。 根因是 _emit 的顺序：先广播、后落库。订阅者的完整性靠"先订阅 → 读历史 → 接实时"拼起来，它需要这条保证——提交早于我这次 SELECT 的，在历史里；提交 晚于我这次 SELECT 的，广播也晚于我订阅，在实时流里。后半句只有在"提交先于 广播"时才成立。交换顺序后，6 次 WebSocket 探测 + 5 次浏览器运行全部无缺口。 （上一轮我修过这个文件的另一个窗口——先读历史后订阅。那个修复是对的，但 只堵住了两步之间的缝，堵不住订阅建立之前的那一段。） tests/test_emit_order.py 守这条不变式：广播的那一刻，这条事件必须已经能从 另一个 session 查到。去掉修复后它确实失败。 --- 还修了三处： - 停在人工介入时，"在不在等人"原先取自审批列表，而那是 4 秒轮询一次的； 中断后会有几秒钟界面说"这次运行已结束"，而它正等着你点通过。改成从事件 推导（isAwaitingHuman），即时且和显示的内容一致。 - 新增 tests/conftest.py 把整个测试会话关进临时目录。test_emit_order 第一版 直接用 SessionLocal 往开发库里写，几行假运行排到了运行列表最前面。已清掉 那 12 行。这该是全局保障，不是只修一个文件。 - 开发期把 store 挂到 window.__studio，好让 playwright 直接读状态——"事件 收全了没有"在界面上看不出来，只能问 store。已确认不进生产包。 --- backend/app/engine/runner.py | 12 +- backend/tests/conftest.py | 43 ++++ backend/tests/test_emit_order.py | 111 +++++++++ frontend/src/canvas/InspectorSheet.tsx | 14 +- frontend/src/run/AssistantPanel.tsx | 333 ++++++++++++++++--------- frontend/src/run/decode.ts | 13 + frontend/src/store/studio.ts | 6 + scripts/check-decode.mjs | 7 + 8 files changed, 411 insertions(+), 128 deletions(-) create mode 100644 backend/tests/conftest.py create mode 100644 backend/tests/test_emit_order.py diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 106ebadd..f99c7c67 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -91,7 +91,16 @@ class RunManager: node_id=node_id, data=data or {}, ) - await bus.publish(event) + # 先落库，再广播。顺序反过来会漏事件，而且是结构性的漏： + # + # 订阅者的完整性靠"读历史 + 接实时"两段拼起来。它先订阅、再读历史， + # 于是需要这条保证——**提交早于我这次 SELECT 的，在历史里；提交晚于 + # 我这次 SELECT 的，广播也晚于我订阅，在实时流里**。这条保证只有在 + # "提交先于广播"时才成立。 + # + # 先广播的话，一条在订阅建立之前广播、却在读历史之后才提交的事件， + # 两边都拿不到。实测跑一次 25ms 的图，六条事件里的 run.started 有 + # 一半几率就这么没了——界面表现为某个节点永远在转圈。 if persist and event.type not in _EPHEMERAL: async with SessionLocal() as session: session.add( @@ -108,6 +117,7 @@ class RunManager: update(Run).where(Run.id == run_id).values(last_seq=event.seq) ) await session.commit() + await bus.publish(event) async def note( self, run_id: str, event_type: EventType | str, *, node_id: str | None = None, diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 00000000..e6dee138 --- /dev/null +++ b/backend/tests/conftest.py @@ -0,0 +1,43 @@ +"""把整个测试会话关进一个临时目录。 + +不这么做的话，任何一个直接用 SessionLocal 的测试都会往开发库 +（data/agentlab.db）里写行——写进去的 run 会出现在运行列表里，和真实运行 +混在一起，而且带着测试的假数据。这不是假想：test_emit_order 第一版就这么 +干了，几行 emit-order-* 的假运行直接排在了列表最前面。 + +必须在 app.* 被 import 之前设好环境变量：settings 是模块级单例，engine 在 +app.db.base 导入时就按 db_url 建好了。conftest 早于测试模块加载，所以这里 +是唯一还来得及的地方。 +""" + +from __future__ import annotations + +import os +import tempfile +from pathlib import Path + +_TMP = Path(tempfile.mkdtemp(prefix="agentlab-tests-")) +os.environ["AGENTLAB_DATA_DIR"] = str(_TMP) +# 别把真实的 key 带进测试环境 +os.environ.setdefault("AGENTLAB_SECRET_KEY", "test-only-not-a-real-key") + +import pytest # noqa: E402 + + +@pytest.fixture(scope="session", autouse=True) +def _prepare_schema(): + """建表。测试库是全新的临时文件，什么都没有。""" + import asyncio + + from app.db.base import Base, engine + + async def create() -> None: + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + asyncio.get_event_loop_policy().new_event_loop().run_until_complete(create()) + yield + + +def pytest_report_header(config) -> str: + return f"测试数据目录：{_TMP}" diff --git a/backend/tests/test_emit_order.py b/backend/tests/test_emit_order.py new file mode 100644 index 00000000..63879df3 --- /dev/null +++ b/backend/tests/test_emit_order.py @@ -0,0 +1,111 @@ +"""事件的落库与广播顺序。 + +订阅者的完整性靠"先订阅 → 读历史 → 接实时"三步拼起来，它需要这条保证： + + 提交早于我这次 SELECT 的，在历史里； + 提交晚于我这次 SELECT 的，广播也晚于我订阅，在实时流里。 + +后半句只有在"提交先于广播"时才成立。反过来（先广播后落库）会漏掉这样一条 +事件：在订阅建立之前广播、却在读历史之后才提交——两边都拿不到。 + +这不是理论。实测跑一次 25 毫秒的图，六条事件里的 run.started 有一半几率 +就这么没了，界面上表现为某个节点永远在转圈、它的子步骤跑到了顶层。 +""" + +from __future__ import annotations + +import uuid + +import pytest +from sqlalchemy import select + +from app.core.events import EventType +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent +from app.engine import runner as runner_mod +from app.engine.runner import run_manager + + +@pytest.mark.asyncio +async def test_event_is_in_db_before_it_is_broadcast(monkeypatch) -> None: + """广播的那一刻，这条事件必须已经能从库里查到。""" + run_id = f"emit-order-1-{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + session.add(Run(id=run_id, status="running", graph={}, input={})) + await session.commit() + + visible_at_publish: list[bool] = [] + original = runner_mod.bus.publish + + async def spy(event): + # 用一个独立 session 查，模拟另一个连接（WebSocket 的历史读就是这样） + async with SessionLocal() as s: + row = ( + await s.execute( + select(RunEvent).where( + RunEvent.run_id == event.run_id, RunEvent.seq == event.seq + ) + ) + ).scalar_one_or_none() + visible_at_publish.append(row is not None) + await original(event) + + monkeypatch.setattr(runner_mod.bus, "publish", spy) + + await run_manager._emit(run_id, EventType.RUN_STARTED, data={"nodes": 3}) + await run_manager._emit(run_id, EventType.NODE_STARTED, node_id="a", data={}) + + assert visible_at_publish == [True, True], ( + "广播时还查不到这条事件——说明是先广播后落库，" + "订阅者会漏掉在它订阅之前广播、在它读历史之后提交的那些" + ) + + +@pytest.mark.asyncio +async def test_ephemeral_events_still_broadcast(monkeypatch) -> None: + """不落库的流式增量照样要广播出去，不能因为跳过落库就一起跳过广播。""" + run_id = f"emit-order-2-{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + session.add(Run(id=run_id, status="running", graph={}, input={})) + await session.commit() + + published: list[str] = [] + bus_mod = runner_mod.bus + original = bus_mod.publish + + async def spy(event): + published.append(str(event.type)) + await original(event) + + monkeypatch.setattr(bus_mod, "publish", spy) + await run_manager._emit(run_id, EventType.LLM_TOKEN, node_id="a", data={"delta": "x"}) + + assert published == ["llm.token"] + async with SessionLocal() as s: + rows = ( + await s.execute(select(RunEvent).where(RunEvent.run_id == run_id)) + ).scalars().all() + assert rows == [], "llm.token 是流式增量，不该落库" + + +@pytest.mark.asyncio +async def test_seq_is_monotonic_across_emits() -> None: + """序号按发出顺序递增——前端排序和去重都建立在这上面。""" + run_id = f"emit-order-3-{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + session.add(Run(id=run_id, status="running", graph={}, input={})) + await session.commit() + + for i in range(5): + await run_manager._emit(run_id, EventType.LOG, data={"message": str(i)}) + + async with SessionLocal() as s: + seqs = [ + e.seq + for e in ( + await s.execute( + select(RunEvent).where(RunEvent.run_id == run_id).order_by(RunEvent.seq) + ) + ).scalars() + ] + assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs) diff --git a/frontend/src/canvas/InspectorSheet.tsx b/frontend/src/canvas/InspectorSheet.tsx index 66bd01fa..5678da35 100644 --- a/frontend/src/canvas/InspectorSheet.tsx +++ b/frontend/src/canvas/InspectorSheet.tsx @@ -3,8 +3,9 @@ import { ChevronLeft, Hand } from 'lucide-react' import clsx from 'clsx' import { useCatalog } from '../store/catalog' import { useStudio } from '../store/studio' -import { decodeRun, type Step } from '../run/decode' +import { decodeRun } from '../run/decode' import { Spinner } from '../components/ui' +import { lastMeaningful } from '../run/AssistantPanel' import { Inspector } from './Inspector' /** @@ -70,8 +71,7 @@ function LiveStrip() { if (!streaming && !waiting) return null // 只取最后一步说给人听，不在这里重演整条流——这是一条提示，不是第二个时间线 - const steps = decodeRun(events) - const last = flattenLast(steps) + const last = lastMeaningful(decodeRun(events)) return ( - {!!copilotTurns.length && ( - - )} )} {empty ? ( - // 没说过话时输入框就是这一栏的主体，竖直居中。之前它是个 38px 高的 - // 框挤在最底下，比上面任何一块都不起眼——"没地方引导写问题"说的就是它 + // 没说过话时输入框就是这一栏的主体，竖直居中
) : ( <>
- {raw ? : ( - ( - t.runId && pending.length ? ( -
- {pending.map((a) => ( - attachRun(runId)} /> - ))} -
- ) : null - )} - /> - )} +
- {run && } )} + + {hasRun && } + + ) +} + +/** + * 对话下面那条运行进度。 + * + * 运行搬去自己那一层之后，这条是它在对话视图里留下的唯一痕迹——没有它， + * 点了运行再切回对话，这次运行在眼前就彻底没了。 + */ +function RunStrip({ onClick }: { onClick: () => void }) { + const run = useStudio((s) => s.run) + const events = useStudio((s) => s.events) + const streaming = useStudio((s) => s.streaming) + if (!run) return null + + const steps = decodeRun(events) + const waiting = isAwaitingHuman(steps) + const last = lastMeaningful(steps) + const failed = events.some((e) => e.type === 'run.failed') + + return ( + + ) +} + +// ------------------------------------------------------------------------- +// 运行：过程和结果，没有输入框 +// ------------------------------------------------------------------------- + +function RunView({ onBack }: { onBack: () => void }) { + const run = useStudio((s) => s.run)! + const events = useStudio((s) => s.events) + const streaming = useStudio((s) => s.streaming) + const attachRun = useStudio((s) => s.attachRun) + const stopRun = useStudio((s) => s.stopRun) + const approvals = useCatalog((s) => s.approvals) + const pending = approvals.filter((a) => a.run_id === run.id && a.status === 'pending') + const [raw, setRaw] = useState(false) + + const turns = useMemo(() => { + const finished = [...events].reverse().find((e) => e.type === 'run.finished') + const failedEvent = [...events].reverse().find((e) => e.type === 'run.failed') + const steps = decodeRun(events) + // 从事件推，不查审批列表——那是 4 秒轮询一次的，中断后会有几秒钟 + // 这里说"完成"，而实际上它正等着你点通过 + const waiting = isAwaitingHuman(steps) + return [{ + id: run.id, + // 运行没有"提问"——它的输入是工具栏那张表单。硬塞一个问题气泡 + // 只会让它看起来像一次对话，而它不是 + question: undefined, + phase: failedEvent ? 'error' + : waiting ? 'waiting' + : streaming ? 'running' + : finished ? 'done' : 'running', + status: streaming ? '正在执行…' : waiting ? '等待你的确认' : finished ? '完成' : '', + steps, + // run 对象是启动时的快照，成果在 run.finished 事件里 + output: finished?.data?.output ?? run.output ?? null, + error: failedEvent ? String(failedEvent.data?.error ?? '运行失败') : run.error || undefined, + runId: run.id, + runClass: run.run_class, + }] + }, [run, events, streaming]) + + return ( +
+
+ + + 运行 + #{run.id.slice(0, 6)} + + {streaming && ( + + )} + {/* 原始事件是排查用的开发者视图，不该占一个常驻 tab——但也不能删， + 翻译层出问题时它是唯一能对照的东西 */} + +
+ +
+ {raw ? : ( + (pending.length ? ( +
+ {pending.map((a) => ( + attachRun(runId)} /> + ))} +
+ ) : null)} + /> + )} +
+ +
+ {run.run_class === 'formal' + ? 正式 v{run.version} + : 探索} + + {!!run.usage?.total_tokens && {run.usage.total_tokens} tok} + {run.usage?.cost_usd ? ${Number(run.usage.cost_usd).toFixed(4)} : null} + {run.usage?.duration_ms ? {formatDuration(run.usage.duration_ms)} : null} +
) } -function RunFooter() { - const run = useStudio((s) => s.run)! - return ( -
- {run.run_class === 'formal' - ? 正式 v{run.version} - : 探索} - - {!!run.usage?.total_tokens && {run.usage.total_tokens} tok} - {run.usage?.cost_usd ? ${Number(run.usage.cost_usd).toFixed(4)} : null} - {run.usage?.duration_ms ? {formatDuration(run.usage.duration_ms)} : null} -
- ) +/** 最后一条有意义的步骤：跳过生命周期行，那才是"此刻在干什么"。 */ +export function lastMeaningful(steps: Step[]): Step | undefined { + const flat: Step[] = [] + const walk = (list: Step[]) => list.forEach((s) => { + flat.push(s) + if (s.children) walk(s.children) + }) + walk(steps) + return [...flat].reverse().find((s) => s.kind !== 'lifecycle') ?? flat[flat.length - 1] } /** 原始事件。翻译层出问题时用来对照，平时收着。 */ diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index be168565..80f536f8 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -96,6 +96,19 @@ export function formatDuration(ms?: number): string { return `${Math.floor(ms / 60_000)}m${Math.round((ms % 60_000) / 1000)}s` } +/** + * 这次运行是不是正停在人工介入上。 + * + * 从步骤推，不要去查审批列表——那是 4 秒轮询一次的，而 run.interrupted + * 事件是即时到的。用列表的话，中断后会有几秒钟界面说"这次运行已结束"， + * 而实际上它正等着你点通过。 + */ +export function isAwaitingHuman(steps: Step[]): boolean { + return steps.some(function walk(s): boolean { + return s.status === 'waiting' || (s.children?.some(walk) ?? false) + }) +} + /** run 到终态时，所有还挂着"进行中"的生命周期行都要收尾。 */ function closeLifecycles(out: Step[], status: StepStatus) { out.forEach((s) => { diff --git a/frontend/src/store/studio.ts b/frontend/src/store/studio.ts index ac8be882..4924e3a6 100644 --- a/frontend/src/store/studio.ts +++ b/frontend/src/store/studio.ts @@ -675,3 +675,9 @@ export const useStudio = create((set, get) => ({ set({ events, runtime, activeEdges }) }, })) + +// 开发期把 store 挂到 window 上，好让 playwright 直接读状态做断言—— +// 界面上看不出"事件收全了没有"，只能问 store。生产构建里去掉。 +if (import.meta.env.DEV) { + ;(window as any).__studio = useStudio +} diff --git a/scripts/check-decode.mjs b/scripts/check-decode.mjs index a2d8a1d6..2b7104e5 100644 --- a/scripts/check-decode.mjs +++ b/scripts/check-decode.mjs @@ -110,6 +110,13 @@ console.log('\n=== 停在审批时的状态 ===') const after = mod.decodeRun(fixtures.human) const doneHost = flatten(after).find((s) => s.kind === 'node' && s.nodeId === 'h') check('恢复后回到正常状态', doneHost?.status === 'done', doneHost?.status) + + // "在不在等人"必须从事件推。查审批列表的话要等 4 秒轮询，中断后那几秒 + // 界面会说"这次运行已结束"，而它其实正等着你点通过 + check('从事件就能判断正在等人', mod.isAwaitingHuman(mod.decodeRun(upto)) === true) + check('跑完了就不再说在等人', mod.isAwaitingHuman(after) === false) + check('压根没有人工节点的运行也不误报', + mod.isAwaitingHuman(mod.decodeRun(fixtures.db)) === false) } console.log('\n=== 循环里的多轮审批 ===')