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]