From bbf6ebb79db867f08a8229bbf6b693d43229f03e Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 20:11:46 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=AC=E5=9B=9E=E6=94=B6=E5=B0=BE=EF=BC=9A?= =?UTF-8?q?=E5=90=AF=E5=8A=A8=E6=97=B6=E5=8F=AA=E6=8E=A5=E7=9D=80=E5=81=9A?= =?UTF-8?q?=E5=BD=93=E5=89=8D=E5=8F=91=E5=B8=83=E7=89=88=E7=9A=84=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E6=89=B9=E6=AC=A1=E3=80=81=E8=A1=A5=E6=B5=8B=E9=87=8D?= =?UTF-8?q?=E5=90=AF=E5=90=8E=E4=BB=8E=E5=B7=B2=E6=9C=89=E5=8F=AC=E5=9B=9E?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E9=87=8D=E7=AE=97=E8=AE=A1=E5=88=92=EF=BC=9B?= =?UTF-8?q?=E6=97=B6=E9=97=B4=E7=BA=BF=E8=AE=A4=E4=B8=8D=E5=87=BA=E7=9A=84?= =?UTF-8?q?=E4=BE=9D=E6=8D=AE=E7=A7=8D=E7=B1=BB=E4=B8=8D=E7=8C=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ledger/recall.resume：最近发布、之后又发布了更新版本的不再接着做（新版本的召回本来就覆盖更早的所有版本） - 测试：重启前只发起了一组、也没收尾，重启后接着发完自动那一批、补上收尾照样开待办，再重启一次不再发起 - run/decode：ledger.basis_changed 认不出种类时写「答案的依据有变」，不当成目录事实变了；batches 的说明照实写参数原样给搭建者 --- backend/app/ledger/recall.py | 14 ++++++---- backend/tests/test_ledger_recall.py | 41 +++++++++++++++++++++++++++++ frontend/src/lib/termsRecall.ts | 2 ++ frontend/src/run/decode.ts | 2 +- scripts/check-decode.mjs | 3 ++- 5 files changed, 55 insertions(+), 7 deletions(-) diff --git a/backend/app/ledger/recall.py b/backend/app/ledger/recall.py index a9e736a0..bfce4cba 100644 --- a/backend/app/ledger/recall.py +++ b/backend/app/ledger/recall.py @@ -494,9 +494,10 @@ def _group_state(group: Group) -> tuple[str, str | None]: return "pending", group.refused -async def batches(session: Any, workflow_id: str, version: int, *, principal: Any = None) -> dict[str, Any]: +async def batches(session: Any, workflow_id: str, version: int) -> dict[str, Any]: """版本历史里「这一版的召回」：每组的结论（数变了 / 没变 / 无法重算：原因 / 重算失败 / 还在算 / 未重算），「另有 M 组 - 未重算」和能不能接着算。参数是遮过的（不给使用者看遮罩值：这个接口只给搭建者，还是照规矩遮）。""" + 未重算」和能不能接着算。接口只给搭建者（api/ledger 的 router）：参数原样给，同他在召回运行的输入里读到的一样；使用者 + 看的是 recall 待办（task_payload，按遮罩换过）。""" from app.api.settings import LEDGER_DEFAULTS, ledger_settings plan = await build_plan(session, workflow_id, int(version)) @@ -1397,15 +1398,18 @@ def after_publish(workflow_id: str, version: int, actor: str | None) -> bool: async def resume(*, days: int = RESUME_DAYS) -> dict[str, int]: - """启动时：最近 days 天发布过的版本，自动那一批没发完的接着发；结束了还没收尾的召回运行补上收尾；表结构变化的游标没有 - 时从现在开始（升级之前的变化不追）。""" + """启动时：最近 days 天发布过、现在仍是当前发布版的版本，自动那一批没发完的接着发（之后又发布了更新的版本的不再接着做： + 新版本的召回本来就覆盖更早的所有版本）；结束了还没收尾的召回运行补上收尾；表结构变化的游标没有时从现在开始（升级之前的 + 变化不追）。""" from app.db.models import WorkflowVersion since = utcnow() - timedelta(days=days) async with SessionLocal() as session: rows = (await session.execute(select(WorkflowVersion.workflow_id, WorkflowVersion.version, WorkflowVersion.published_by) + .join(Workflow, Workflow.id == WorkflowVersion.workflow_id) .where(WorkflowVersion.published_at >= since, - WorkflowVersion.level.is_not(None)))).all() + WorkflowVersion.level.is_not(None), + Workflow.published_version == WorkflowVersion.version))).all() unsettled = [r for (r,) in (await session.execute(select(Run.id).where( Run.origin_kind == ORIGIN, Run.status.in_(("succeeded", "failed", "cancelled")), Run.manifest_hash.is_not(None)))).all()] diff --git a/backend/tests/test_ledger_recall.py b/backend/tests/test_ledger_recall.py index 86c74aea..867d09d5 100644 --- a/backend/tests/test_ledger_recall.py +++ b/backend/tests/test_ledger_recall.py @@ -561,3 +561,44 @@ async def test_recall_graph_is_deterministic_and_events_have_offline_labels(monk assert labels["ledger.matched"] == "比对已发布口径" assert _summary("ledger.recalled", {"role": "answer", "changed": False, "to_version": 2}) == "已按 v2 核对，不影响" assert _summary("ledger.basis_changed", {"kind": "structure", "items": [{"label": "x"}]}) == "依据的表结构有变（1 项）" + + +async def test_restart_resumes_the_auto_batch_and_settles_finished_runs(monkeypatch): + """服务重启：计划从已有召回运行重算——自动那一批没发完的接着发，结束了还没收尾的补上收尾（不建任务表）。""" + from app.db.base import utcnow + from app.db.models import WorkflowVersion + + await ledger_settings(auto_issue=False, issue_daily_limit=100_000, recall_limit=2) + wf = await S.publish() + await answer(monkeypatch) + await answer(monkeypatch, "2026-09-01", "2026-09-05") + await publish_v2(wf) + async with SessionLocal() as session: + row = (await session.execute(select(WorkflowVersion).where(WorkflowVersion.workflow_id == wf, + WorkflowVersion.version == 2))).scalar_one() + row.published_at = utcnow() + await session.commit() + # 「重启之前」：只发起了一组，收尾也没做（进程在这之后没了） + monkeypatch.setattr(recall, "schedule_settle", lambda run_id: None) + async with SessionLocal() as session: + plan = await recall.build_plan(session, wf, 2) + first = await recall._launch_group(plan, plan.todo()[0], batch="auto", actor=None) + first = await S.wait(first.id) + assert await recall_tasks(first.id) == [] and await events(first.id, EventType.LEDGER_RECALLED) == [] + monkeypatch.undo() + out = await recall.resume() + # 同一个测试库里别的用例的版本也会被接着做：只数这一份口径的 + assert out["launched"] >= 1 and out["settled"] >= 1 + async with SessionLocal() as session: + runs = await recall.recall_runs(session, wf, 2) + assert len(runs) == 2 and {r.origin["batch"] for r in runs} == {"auto"} + for run in runs: + await S.wait(run.id) + await recall.drain() + assert len(await recall_tasks(first.id)) == 1, "补上的收尾照样开了待办" + async with SessionLocal() as session: + view = await recall.batches(session, wf, 2) + assert view["counts"]["changed"] == 2 and view["remaining"] == 0 + await recall.resume() + async with SessionLocal() as session: + assert len(await recall.recall_runs(session, wf, 2)) == 2, "再重启一次：都算过了，不再发起" diff --git a/frontend/src/lib/termsRecall.ts b/frontend/src/lib/termsRecall.ts index 879e7149..82b40be9 100644 --- a/frontend/src/lib/termsRecall.ts +++ b/frontend/src/lib/termsRecall.ts @@ -145,6 +145,8 @@ export const RECALL_TEXT = { values: '依据有变：出现了触及有效记录条件的新码值', catalog_edit: '依据有变：处理数据源变化时改了数据目录', } as Record, + /** 认不出是哪一种依据变了（缺字段、以后加的种类）：只说依据有变，不猜 */ + basisGeneric: '答案的依据有变', basisItems: (labels: string[]) => labels.join('、'), newErrors: (count: number) => `按新目录检查新出现 ${n(count)} 条错误级问题`, sub: '封存后追加，不在封存范围内', diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index 7ac59c5b..c763383b 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -2452,7 +2452,7 @@ export function decodeRun(events: RunEvent[], final?: RunFinal, opts?: DecodeOpt out.push({ id: `lb-${seq}`, seq, kind: 'note', status: 'done', code: 'ledger_basis_changed', level: errors > 0 || d.kind === 'structure' ? 'warn' : 'info', - title: R.basis[String(d.kind)] ?? R.basis.catalog, + title: R.basis[String(d.kind)] ?? R.basisGeneric, sub: R.sub, ...(where ? { meta: where } : {}), ...(labels.length || errors ? { detail: [labels.length ? R.basisItems(labels) : '', errors ? R.newErrors(errors) : ''] diff --git a/scripts/check-decode.mjs b/scripts/check-decode.mjs index 2d0ac338..48483cf3 100644 --- a/scripts/check-decode.mjs +++ b/scripts/check-decode.mjs @@ -1927,7 +1927,8 @@ await section('口径召回：封存后的核对结论与依据有变（ledger.r .find((s) => s.code === 'ledger_basis_changed') check('表结构变了：依据的表结构已变', structure?.title === '依据的表结构已变' && structure?.level === 'warn', structure?.title) const odd = run(E(5, 'ledger.recalled', null, {}), E(6, 'ledger.basis_changed', null, { kind: 'x', items: 'y' })) - check('缺字段：不出 undefined、不瞎填数', !odd.some((s) => /undefined|NaN/.test(`${s.title}${s.sub ?? ''}${s.detail ?? ''}${s.meta ?? ''}`)), + check('缺字段：不出 undefined、不瞎填数，认不出的结论和种类不猜', !odd.some((s) => /undefined|NaN/.test(`${s.title}${s.sub ?? ''}${s.detail ?? ''}${s.meta ?? ''}`)) + && odd.some((s) => s.title === '口径召回核对') && odd.some((s) => s.title === '答案的依据有变'), odd.map((s) => s.title).join(' | ')) const phase = [E(4, 'run.finished', null, {}), E(5, 'ledger.recalled', null, {}), E(6, 'ledger.basis_changed', null, {})] .reduce((s, ev) => mod.nextPhase(s, ev), { phase: 'running', awaiting: false })