From dbae680abfd420168078bf82e4766d7dba0090d6 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 29 Sep 2026 16:50:31 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=AD=A3=E5=AE=A1=E6=89=B9=E6=81=A2?= =?UTF-8?q?=E5=A4=8D=E7=9A=84=E8=BF=90=E8=A1=8C=E8=80=97=E6=97=B6=EF=BC=9A?= =?UTF-8?q?=E5=81=9C=E5=9C=A8=E5=AE=A1=E6=89=B9=E4=B8=8A=E4=B9=8B=E5=90=8E?= =?UTF-8?q?=E7=9A=84=E6=94=B6=E5=B0=BE=E4=B8=8D=E5=86=8D=E5=90=8C=E6=97=B6?= =?UTF-8?q?=E7=AE=97=E8=BF=9B=E6=89=A7=E8=A1=8C=E5=92=8C=E7=AD=89=E5=BE=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 每一段的执行时长以前一直计到收尾结束，等待时长又从 run.interrupted 算起。停在审批上之后 写审批记录、发事件这一截被两边都算，执行 + 等待比墙钟还长：平时只差几毫秒，机器一忙就差出几十毫秒 （全量并行测试里约每 20 遍失败一次的那条就是它） - 现在这一段的执行算到引擎发出中断的那一刻为止（LangGraph 这一轮的节点都跑完了才发出中断）， 之后的收尾不算执行；执行、等待、墙钟由构造保证对得上账，不再靠容差 - 新测试把中断后的收尾人为拖慢 0.4 秒，稳定复现以前的重复计算；全量测试并行连跑 26 遍全过 --- backend/app/engine/runner.py | 10 +++++++++- backend/tests/test_run_timing.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 34c52296..a6b2ad53 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -591,6 +591,10 @@ class RunManager: error_node: str | None = None deadline: asyncio.Timeout | None = None + # 停在审批上的那一刻：这一段的执行算到这里为止。后面写审批记录、发 run.interrupted 是收尾， + # 等待从 run.interrupted 算起——以前执行一直计到 finally，收尾那一截执行和等待两边都算， + # active + wait 比墙钟还长（机器一忙就差出几十毫秒） + paused: float | None = None async with self._semaphore: try: async with SessionLocal() as session: @@ -655,8 +659,11 @@ class RunManager: if mode == "custom": await self._handle_custom(run_id, chunk) elif mode == "updates": + # LangGraph 这一轮的节点都跑完了才发出中断：此刻引擎已经停下 + at = time.perf_counter() if await self._handle_updates(run_id, chunk): interrupted = True + paused = at if paused is None else paused snapshot = await app.aget_state(config) final_state = dict(snapshot.values or {}) @@ -710,7 +717,8 @@ class RunManager: error_node = terminal[1].get("node_id") logger.exception("run %s 失败", run_id) finally: - elapsed = int((time.perf_counter() - started) * 1000) + stop = paused if status == "interrupted" and paused is not None else time.perf_counter() + elapsed = int((stop - started) * 1000) self._finalizing.add(run_id) try: await self._finalize(run_id, status, error, final_state, elapsed, diff --git a/backend/tests/test_run_timing.py b/backend/tests/test_run_timing.py index 179d8bb1..67ea6c26 100644 --- a/backend/tests/test_run_timing.py +++ b/backend/tests/test_run_timing.py @@ -123,6 +123,37 @@ async def test_resumed_run_keeps_its_start_and_sums_every_segment(slow_transform } +async def test_bookkeeping_after_the_pause_is_not_counted_twice(slow_transforms, monkeypatch): + """停在审批上之后的收尾（写审批记录、发 run.interrupted）不算执行时长。 + + 以前这一段的执行一直计到 _drive 的 finally，等待又从 run.interrupted 算起：收尾那一截两边都算， + active + wait 比墙钟还长。平时只差几毫秒，机器一忙就超出容差（全量并行测试里偶发）。 + 这里把收尾拖慢 0.4 秒，不靠机器忙不忙碰运气。 + """ + original = run_manager._handle_updates + + async def slow(run_id, chunk): + hit = await original(run_id, chunk) + if hit: + await asyncio.sleep(0.4) # 审批卡已经发出去了，引擎也不在执行 + return hit + + monkeypatch.setattr(run_manager, "_handle_updates", slow) + run = await run_manager.start(graph=APPROVAL_GRAPH, input_payload={"question": "q"}) + paused = await _wait(run.id, ("interrupted", "failed")) + assert paused.status == "interrupted", paused.error + await asyncio.sleep(WAIT) + await run_manager.resume(run.id, {"approved": True}) + done = await _wait(run.id, ("succeeded", "failed")) + assert done.status == "succeeded", done.error + + usage = done.usage + # 执行、等待、墙钟严格对得上账：取整各丢不到 1 毫秒 + assert usage["wall_ms"] >= usage["active_ms"] + usage["wait_ms"] - 2, usage + # 0.4 秒的收尾没有算进执行（两段各 SEGMENT 秒，其余开销远小于 0.4 秒） + assert usage["active_ms"] < 2 * SEGMENT * 1000 + 400, usage + + @pytest.fixture def fails_once(monkeypatch): original = compiler.RUNNERS[NodeType.TRANSFORM]