diff --git a/backend/app/ledger/view.py b/backend/app/ledger/view.py index 81492c2b..5ce136a6 100644 --- a/backend/app/ledger/view.py +++ b/backend/app/ledger/view.py @@ -10,6 +10,7 @@ issued 是这次运行按口径出具过的正式运行,reconciled 是对过 """ from __future__ import annotations +from datetime import datetime, timezone from typing import Any from sqlalchemy import select @@ -23,6 +24,10 @@ from app.ledger import store #: state:matched 比对过;pending 已排队按需比对;ineligible 这次运行不比对(reason 写为什么);unreadable 比对记录读不出 STATES = ("matched", "pending", "ineligible", "unreadable") +#: 成功结束、还没封存的运行,结束后这么多秒内答 pending(runner 落库之后马上封存,正常是几秒)。过了还没封存的是封存出了错、 +#: 或者是封存之前的老数据:再答 pending,页面就一直转圈、等满两分钟(store/ledger.ts 的等待上限)再取还是 pending——照实说没封存 +SEAL_GRACE_S = 120 + class _Hide: """按这次运行查到过的遮罩值遮:hide(值) 这个值给不给,shown(文字) 文字里的遮罩值换成「已遮罩」。""" @@ -92,8 +97,6 @@ def _labels_view(content: dict[str, Any], hider: _Hide, names: dict[str, str]) - def _stamp(recorded: store.Recorded) -> dict[str, Any]: """比对的来历:哪份工件、哪一版算法、谁触发的、什么时候追加的;事后补比的写明按现在的口径回看。""" - from datetime import datetime, timezone - at = datetime.fromtimestamp(recorded.ts, timezone.utc).isoformat(timespec="seconds") \ if isinstance(recorded.ts, (int, float)) else None return {"artifact": recorded.artifact, "algo": recorded.algo, "trigger": recorded.trigger, "retro": recorded.retro, @@ -102,6 +105,14 @@ def _stamp(recorded: store.Recorded) -> dict[str, Any]: else ledger_labels.RETRO_NOTE} if recorded.retro else {})} +def _just_finished(finished_at: datetime | None) -> bool: + """运行是不是刚结束(SEAL_GRACE_S 以内)。没记结束时间的不算;库里读回 naive 的按 UTC 算(同 db/base.UTCDateTime)。""" + if finished_at is None: + return False + at = finished_at if finished_at.tzinfo is not None else finished_at.replace(tzinfo=timezone.utc) + return (datetime.now(timezone.utc) - at).total_seconds() <= SEAL_GRACE_S + + async def run_view(run_id: str) -> dict[str, Any]: """GET /api/runs/{run_id}/ledger 的答复:标签、整题判断、每条查询的比对、能做的动作。没比对过、能比的运行这时排上按需比对 (trigger=on_demand),答复 state=pending;不比的照实说为什么。""" @@ -117,9 +128,10 @@ async def run_view(run_id: str) -> dict[str, Any]: if recorded is None: why = ledger_match.ineligible(run) if why == ledger_match.NOT_SEALED and run.status == "succeeded" \ - and ledger_match.wanted(run.run_class, run.origin_kind): + and ledger_match.wanted(run.run_class, run.origin_kind) and _just_finished(run.finished_at): # runner 先发 run.finished、落库,再封存、再排上比对(engine/runner._finalize):页面收到 run.finished 立刻来取, - # 这一下还没封存。这不是「不比对」,是封存完就比——照实答 pending,页面按退避接着取;这里不另排按需比对 + # 这一下还没封存。这不是「不比对」,是封存完就比——照实答 pending,页面按退避接着取;这里不另排按需比对。 + # 只限刚结束的那一会儿(SEAL_GRACE_S),早就结束、一直没封存的照实答 ineligible return {**base, "state": "pending"} sealed = await _sealed(run_id) if why is None else None if sealed is not None and not sealed.queries: diff --git a/backend/tests/test_ledger_api.py b/backend/tests/test_ledger_api.py index e70a6561..fceeb799 100644 --- a/backend/tests/test_ledger_api.py +++ b/backend/tests/test_ledger_api.py @@ -102,11 +102,14 @@ async def test_a_run_that_just_finished_and_is_about_to_be_sealed_is_pending(cli await ledger_match.drain() from app.db.models import Run + from datetime import datetime, timedelta, timezone + + now = datetime.now(timezone.utc) async with SessionLocal() as session: fresh = Run(workflow_name="刚跑完", status="succeeded", run_class="exploratory", origin_kind="chat", graph={}, - input={}, output={}) + input={}, output={}, finished_at=now) failed = Run(workflow_name="失败了", status="failed", run_class="exploratory", origin_kind="chat", graph={}, - input={}, output={}) + input={}, output={}, finished_at=now) session.add_all([fresh, failed]) await session.commit() fresh_id, failed_id = fresh.id, failed.id @@ -117,6 +120,32 @@ async def test_a_run_that_just_finished_and_is_about_to_be_sealed_is_pending(cli assert (await ledger_of(client, failed_id))["state"] == "ineligible" +async def test_a_run_that_finished_long_ago_and_was_never_sealed_says_so(client, monkeypatch): + """「刚跑完、马上封存」只是结束后的那一会儿:成功结束了好一阵、一直没封存的(封存时出了错、封存这件事之前的老数据), + 再答 pending 就是让页面一直转圈、等两分钟、点「再取一次」还是 pending。照实说没封存(ineligible,NOT_SEALED); + 结束时间没记的同样不当成刚结束。""" + from datetime import datetime, timedelta, timezone + + from app.db.models import Run + from app.ledger import view as ledger_view + + old = datetime.now(timezone.utc) - timedelta(seconds=ledger_view.SEAL_GRACE_S + 60) + async with SessionLocal() as session: + stale = Run(workflow_name="很早以前跑完", status="succeeded", run_class="exploratory", origin_kind="chat", + graph={}, input={}, output={}, finished_at=old) + undated = Run(workflow_name="没记结束时间", status="succeeded", run_class="exploratory", origin_kind="chat", + graph={}, input={}, output={}) + session.add_all([stale, undated]) + await session.commit() + stale_id, undated_id = stale.id, undated.id + scheduled: list[str] = [] + monkeypatch.setattr(ledger_match, "schedule", lambda run_id, **kw: scheduled.append(run_id) or True) + for run_id in (stale_id, undated_id): + body = await ledger_of(client, run_id) + assert body["state"] == "ineligible" and body["reason"] == ledger_match.NOT_SEALED, body + assert scheduled == [] + + async def test_runs_that_are_not_compared_say_why(client, monkeypatch): row = await S.explore(monkeypatch, S.EQUIVALENT, origin="api", text="本周销售额 [[v:Q1.r0.销售额]]。") body = await ledger_of(client, row.id)