From 96bb372ef30b9f5482be01f4b0f9d77544624aff Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 13:18:23 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=8D=E6=A0=B8=20R-b2=EF=BC=88=E5=8D=95?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E4=B8=8B=E5=91=A8=E6=9C=9F=E6=95=B4=E7=90=86?= =?UTF-8?q?=E9=80=80=E5=9B=9E=E5=88=A4=E5=AE=9A=E4=B8=AD=E7=9A=84=E5=BE=85?= =?UTF-8?q?=E5=8A=9E=EF=BC=89=EF=BC=9A=E5=BD=93=E7=8F=AD=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=E9=9A=94=E4=B8=80=E4=BC=9A=E5=84=BF=E7=9A=84=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E6=8E=A5=E6=89=8B=EF=BC=88trust=5Fmine=EF=BC=89?= =?UTF-8?q?=E4=B8=8D=E8=AE=BA=E6=9C=89=E6=B2=A1=E6=9C=89=E5=88=AB=E7=9A=84?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=EF=BC=8C=E5=81=9C=E5=9C=A8=20resolving=20?= =?UTF-8?q?=E7=9A=84=E5=BE=85=E5=8A=9E=E5=8F=AA=E6=94=B6=E6=8B=BE=E5=81=9C?= =?UTF-8?q?=E4=BA=86=E4=B8=A4=E4=B8=AA=E7=A7=9F=E7=BA=A6=E6=97=B6=E9=95=BF?= =?UTF-8?q?=E4=BB=A5=E4=B8=8A=E7=9A=84=EF=BC=8C=E5=90=AF=E5=8A=A8=E6=97=B6?= =?UTF-8?q?=E5=8F=88=E6=B2=A1=E6=9C=89=E5=88=AB=E7=9A=84=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=E7=9A=84=E9=82=A3=E4=B8=80=E6=AC=A1=E7=85=A7=E6=97=A7=E5=85=A8?= =?UTF-8?q?=E6=94=B6=E6=8B=BE=EF=BC=9B=E9=A1=BA=E5=B8=A6=E6=8A=8A=E5=85=B6?= =?UTF-8?q?=E4=BD=99=E5=91=A8=E6=9C=9F=E8=B0=83=E7=94=A8=E7=9A=84=E6=95=B4?= =?UTF-8?q?=E7=90=86=E8=BF=87=E4=B8=80=E9=81=8D=E2=80=94=E2=80=94=E7=BB=93?= =?UTF-8?q?=E6=9E=9C=E6=9C=AA=E7=9F=A5=E7=9A=84=E5=86=99=E5=85=A5=E5=8F=AA?= =?UTF-8?q?=E8=A1=A5=E5=81=9C=E4=BA=86=E4=B8=A4=E4=B8=AA=E7=A7=9F=E7=BA=A6?= =?UTF-8?q?=E6=97=B6=E9=95=BF=E4=BB=A5=E4=B8=8A=E7=9A=84=EF=BC=88=E6=94=B6?= =?UTF-8?q?=E5=B0=BE=E5=92=8C=E5=AE=89=E6=8E=92=E6=A0=B8=E5=AE=9E=E4=B9=8B?= =?UTF-8?q?=E9=97=B4=E7=9A=84=E4=B8=8D=E7=A2=B0=EF=BC=89=EF=BC=8C=E9=97=AE?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=B2=A1=E6=9C=89=E4=BB=BB=E5=8A=A1=E8=A1=8C?= =?UTF-8?q?=E7=9A=84=E8=80=81=E8=BD=AE=E6=AC=A1=E5=8F=AA=E5=9C=A8=E5=90=AF?= =?UTF-8?q?=E5=8A=A8=E6=97=B6=E6=95=B4=E7=90=86=EF=BC=88=E8=80=81=E9=A1=B5?= =?UTF-8?q?=E9=9D=A2=E5=BC=80=E7=9A=84=E8=BD=AE=E6=AC=A1=E5=BB=BA=E5=9B=BE?= =?UTF-8?q?=E6=97=B6=E4=B9=9F=E6=98=AF=E8=BF=99=E4=B8=AA=E6=A0=B7=E5=AD=90?= =?UTF-8?q?=EF=BC=89=E3=80=82=E8=A1=A5=E6=B5=8B=E8=AF=95=EF=BC=9A=E5=88=A4?= =?UTF-8?q?=E5=AE=9A=E5=88=B0=E4=B8=80=E5=8D=8A=E5=BD=93=E7=8F=AD=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E6=9D=A5=E4=B8=80=E6=AC=A1=EF=BC=8C=E5=BE=85=E5=8A=9E?= =?UTF-8?q?=E7=85=A7=E6=A0=B7=E8=AE=B0=E5=B7=B2=E5=A4=84=E7=90=86=E3=80=81?= =?UTF-8?q?=E9=82=A3=E4=B8=80=E7=A5=A8=E4=B8=8D=E6=92=A4=E5=9B=9E=EF=BC=88?= =?UTF-8?q?=E6=94=B9=E5=89=8D=E7=9A=84=E5=86=99=E6=B3=95=E8=BF=99=E6=9D=A1?= =?UTF-8?q?=E4=B8=8D=E8=BF=87=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/chat/job.py | 4 +++- backend/app/process/actions.py | 13 ++++++++++- backend/app/process/tasks.py | 5 ++-- backend/tests/test_task_resume_errors.py | 29 ++++++++++++++++++++++++ 4 files changed, 47 insertions(+), 4 deletions(-) diff --git a/backend/app/chat/job.py b/backend/app/chat/job.py index b29b3a8c..438f6fba 100644 --- a/backend/app/chat/job.py +++ b/backend/app/chat/job.py @@ -1234,7 +1234,9 @@ async def startup(*, resume: bool | None = None, live: Any = None) -> dict[str, await session.commit() for turn_id in gone: turn_log.notify(turn_id) - for row in legacy: + # 老轮次（没有任务行）只在启动时整理：当班进程隔一会儿来的那一次（live.trust_mine）不碰——老页面开的轮次（不占锁的写入） + # 正在建图时也是这个样子 + for row in [] if live.trust_mine else legacy: if turn_run_id(row): continue # 有运行的老轮次：页面打开时请服务端接着收尾 async with SessionLocal() as session: diff --git a/backend/app/process/actions.py b/backend/app/process/actions.py index 2b6a6393..f8c8e2b2 100644 --- a/backend/app/process/actions.py +++ b/backend/app/process/actions.py @@ -729,7 +729,8 @@ async def startup(*, live: Any = None) -> int: """启动时：上次进程没了的时候已经发出（sent）、还没拿到结果的动作收成结果未知，安排核实；结果未知却还没安排核实的 （挂起之后、安排之前进程没了）补上。返回处理的条数。 - 多进程：所属的运行还在别的活着的进程里执行的不碰（那边正在发、正在等结果）。""" + 多进程：所属的运行还在别的活着的进程里执行的不碰（那边正在发、正在等结果）；当班进程隔一会儿来的那一次只收拾停了两个租约 + 时长以上的。""" from app.cluster import state from app.db.models import Run from app.engine.runner import run_manager @@ -741,6 +742,16 @@ async def startup(*, live: Any = None) -> int: ActionRecord.state.in_(("sent", "unknown")), ActionRecord.resolution.is_(None)))).all() ids = [action_id for action_id, run_id, status, owner in rows if not (status in ("running", "queued") and live.alive(owner, local=run_manager.is_active(run_id)))] + if live.trust_mine and ids: + # 当班进程隔一会儿来的那一次：刚挂起、执行它的进程正在安排核实的（收尾和安排核实之间）不碰，只收拾停了两个租约时长以上的 + from datetime import timedelta + + from app.cluster.workers import lease_s + + floor = utcnow() - timedelta(seconds=2 * lease_s()) + async with SessionLocal() as session: + ids = list((await session.execute(select(ActionRecord.id).where( + ActionRecord.id.in_(ids), ActionRecord.updated_at < floor))).scalars()) for action_id in ids: try: await on_unknown(action_id) diff --git a/backend/app/process/tasks.py b/backend/app/process/tasks.py index e1690f40..a8bbbb4c 100644 --- a/backend/app/process/tasks.py +++ b/backend/app/process/tasks.py @@ -1027,14 +1027,15 @@ async def startup(*, live: Any = None) -> int: """启动时：存量的待审批补建 Task（派给任何人，幂等）；上次停在 resolving 的（判定之后、恢复之前进程没了）按 Approval 的状态收拾——还在等的退回 pending、撤回那一票，已经交给引擎的记成已处理。返回补建的条数。 - 多进程：别的进程活着时，resolving 可能正是它在判定、恢复（几秒的事），只收拾停在 resolving 超过两个租约时长的。""" + 多进程：别的进程活着时，resolving 可能正是它在判定、恢复（几秒的事），只收拾停在 resolving 超过两个租约时长的；当班进程隔一会儿 + 来的那一次（live.trust_mine）不论有没有别的进程都这样——本进程自己也可能正在判定。只有启动时、又没有别的进程的那一次全收拾。""" from datetime import timedelta from app.cluster import state from app.cluster.workers import lease_s live = live if live is not None else await state.current(fresh=True) - stale_before = utcnow() - timedelta(seconds=2 * lease_s()) if live.others else None + stale_before = utcnow() - timedelta(seconds=2 * lease_s()) if (live.others or live.trust_mine) else None async with SessionLocal() as session: orphans = list((await session.execute( select(Approval).outerjoin(Task, Task.approval_id == Approval.id) diff --git a/backend/tests/test_task_resume_errors.py b/backend/tests/test_task_resume_errors.py index db9b885e..721a7fd2 100644 --- a/backend/tests/test_task_resume_errors.py +++ b/backend/tests/test_task_resume_errors.py @@ -84,3 +84,32 @@ async def test_any_resume_error_puts_the_task_back(monkeypatch): out = await tasks.respond(task.id, ANN, decision="approve") assert out["resolved"] is True assert (await _wait(run.id, ("succeeded", "failed"))).status == "succeeded" + + +async def test_the_periodic_takeover_leaves_a_task_that_is_being_resolved_alone(monkeypatch): + """复核 R-b2：当班进程隔一会儿的那一次接手（live.trust_mine）撞上一次正在判定、恢复的待办（停在 resolving 的那几秒）， + 单进程时也不能把它退回 pending、撤回那一票——只收拾停了两个租约时长以上的。启动时、又没有别的进程的那一次照旧全收拾。""" + from app.cluster.workers import Liveness + + run = await run_manager.start(graph=FLOW, input_payload={"q": "x"}) + await _wait(run.id, ("interrupted",)) + task = await _pending(run.id) + real = run_manager.resume + reaped = asyncio.Event() + + async def slow(*a, **k): + await asyncio.sleep(0.3) + await tasks.startup(live=Liveness().trusting()) # 判定到一半，当班进程来了一次 + reaped.set() + return await real(*a, **k) + + monkeypatch.setattr(run_manager, "resume", slow) + out = await tasks.respond(task.id, ANN, decision="approve") + assert reaped.is_set() and out["resolved"] is True + assert (await _wait(run.id, ("succeeded", "failed"))).status == "succeeded" + async with SessionLocal() as session: + row = await session.get(Task, task.id) + mine = (await session.execute(select(TaskAssignment).where(TaskAssignment.task_id == task.id))).scalars().all() + assert row.status == "resolved", row.status + assert mine and all(a.responded_by == "ann" and a.status not in ("pending", "withdrawn") for a in mine), \ + [(a.status, a.responded_by) for a in mine]