mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
口径召回后端（W4-a）：发布新版本后反查用过旧版的答案、按参数分组在同一份数据上重算两版之差，数真变了才开 recall 待办
为什么：口径升版是被迫处置的事件（02 做法 7、拍板 L6），已交付的答案以前没人追；目录事实被驳回、表结构变了也没有地方标。
- ledger/uses.py（新）：反查旧版本被谁用过——比对工件里等价已证明的问数回答（含轮次、对话主人）、W 的正式运行（定时任务、
应用各取最近 4 期）、固定了它的工作流版本发起的正式运行、固定这一版的应用、导出审计（导出人也是主人）；预告、召回、
版本用量共用这一份
- ledger/recall.py（新）：计划按（旧版本、新版本、input_key）去重成组，从已有召回运行重算（幂等、不建任务表）；每组一次
召回运行（origin_kind=recall，探索运行），派生图经 derive.recall_graph（新旧两版取数和口径卡带前缀、召回卡里算两版之差）；
不能重算照实记原因（数据部分不能复用、新版没这张卡、入口参数不齐、图里有人工节点或写入）；每次发布最多 recall_limit 组，
超出的手动接着算；收尾时按渲染的旧值新值判数变没变：变了开 recall 待办（开登录按主人各一条、关登录派给任何人）、推
ledger.recall，没变在原答案上记 ledger.recalled；待办「按新版重新出具」走 issue.check_target 同一套判法、「知道了」
- 目录事实被驳回或修改（审阅、人工编辑、保存提案、批量审阅之后）：冻结目录里看得到这一项的探索运行记 ledger.basis_changed，
按新目录多出错误级 SQL 检查问题的才开待办；表结构变动消费 source_changes（游标、处理时改了目录的当场交过去），标依据的
表结构已变，受影响的口径试算一次，算不出来才通知发布人
- 发布钩子在 commit_publish 提交之后排队（自动的几样和调度循环同一个开关），GET /api/workflows/{id}/recall-preview 给发布
弹窗预告；GET/POST /api/ledger/recall/{wf}/{v}[/continue] 给版本历史看批次、接着算；版本用量加问数回答、导出、召回
- ledger.recall 打开订阅（全局、个人渠道），同一渠道只推一次，消息不放数；新事件 ledger.recalled、ledger.basis_changed
登记为封存后追加的事件，可验证报告离线核对页认得；E4 每次发布召回了几组、数变了的比例
- runner 收尾时召回运行在后台收尾，启动时接着做没发完的批次
This commit is contained in:
ThearchyHelios
@@ -11,7 +11,10 @@
|
||||
L1、L5：参数只从比对工件取，使用者只能按他有权使用的应用固定着的版本出具）；
|
||||
- POST /api/ledger/answers/{run_id}/reconcile（同上）：口径对账，点开才跑一次确定性派生的只读图（app/ledger/reconcile.py）；
|
||||
- GET /api/ledger/certify-queue、POST /api/ledger/certify-queue/dismiss（router，搭建者，第 3 波）：可沉淀的问题、「不是口径」
|
||||
（app/ledger/certify.py）。这几个新接口都不放给访问密钥（auth/keys.KEY_ROUTES 不登记）。
|
||||
（app/ledger/certify.py）。
|
||||
- GET /api/ledger/recall/{workflow_id}/{version}、POST …/continue（router，搭建者，第 4 波）：口径升到这一版之后的召回批次——
|
||||
每组的结论、「另有 M 组未重算」、手动接着算（app/ledger/recall.py）。版本用量（workflow-versions）另加问数回答、导出、召回。
|
||||
这几个新接口都不放给访问密钥（auth/keys.KEY_ROUTES 不登记）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -24,7 +27,7 @@ from pydantic import BaseModel, Field
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.db.base import get_session
|
||||
from app.db.base import SessionLocal, get_session
|
||||
from app.db.models import Workflow
|
||||
from app.db.models_process import ActionRecord
|
||||
|
||||
@@ -40,12 +43,18 @@ answers_router = APIRouter(prefix="/api/ledger/answers", tags=["ledger"])
|
||||
@router.get("/workflow-versions/{workflow_id}/{version}")
|
||||
async def workflow_version(workflow_id: str, version: int, session: AsyncSession = Depends(get_session)
|
||||
) -> dict[str, Any]:
|
||||
"""一个工作流版本被哪些应用固定、生效记录数、正式运行数。"""
|
||||
"""一个工作流版本被哪些应用固定、生效记录数、正式运行数；另加口径账本的用量（next-round 02 做法 8）：被几次问数回答用过
|
||||
（answers：比对里等价已证明的探索运行）、导出过几次（exports：这些答案和这一版的正式运行上的报告导出审计）、定时任务几个、
|
||||
召回（recall：以这一版为新版本召回过几组、其中数变了几组，作为旧版本被召回过几组）。"""
|
||||
from app.apps.versions import usage
|
||||
from app.ledger.uses import version_usage
|
||||
|
||||
if await session.get(Workflow, workflow_id) is None:
|
||||
raise HTTPException(404, "工作流不存在，可能已被删除")
|
||||
return await usage(session, workflow_id, version)
|
||||
base = await usage(session, workflow_id, version)
|
||||
extra = await version_usage(session, workflow_id, version)
|
||||
return {**base, "answers": extra["answers"], "exports": extra["exports"], "schedules": extra["schedules"],
|
||||
"recall": extra["recall"]}
|
||||
|
||||
|
||||
def effect_row(r: ActionRecord, *, receipt: bool = True) -> dict[str, Any]:
|
||||
@@ -163,6 +172,38 @@ async def reconcile_answer(run_id: str, payload: ReconcileIn) -> dict[str, Any]:
|
||||
return {"run_id": run.id, "reconciled": ledger_reconcile.reconciled_view(run)}
|
||||
|
||||
|
||||
@router.get("/recall/{workflow_id}/{version}")
|
||||
async def recall_batches(workflow_id: str, version: int, session: AsyncSession = Depends(get_session)
|
||||
) -> dict[str, Any]:
|
||||
"""口径升到这一版之后的召回（app/ledger/recall.batches）：每组（旧版本、参数）的结论——数变了 / 没变 / 无法重算：原因 /
|
||||
重算失败 / 还在算 / 未重算——和「另有 M 组未重算」、能不能接着算。计划从已有的召回运行现算，不另存。"""
|
||||
from app.ledger import recall
|
||||
|
||||
if await session.get(Workflow, workflow_id) is None:
|
||||
raise HTTPException(404, "工作流不存在，可能已被删除")
|
||||
return await recall.batches(session, workflow_id, version)
|
||||
|
||||
|
||||
@router.post("/recall/{workflow_id}/{version}/continue")
|
||||
async def recall_continue(workflow_id: str, version: int, session: AsyncSession = Depends(get_session)
|
||||
) -> dict[str, Any]:
|
||||
"""接着算：把还没重算的组再发起一批（每批 recall_limit 组，上限是 0 时按缺省的组数），交回发起了的召回运行和新的批次
|
||||
情况。都算过了的回 409。"""
|
||||
from app.api.coded import CodedHTTPException
|
||||
from app.api.deps import current_principal
|
||||
from app.ledger import recall
|
||||
|
||||
if await session.get(Workflow, workflow_id) is None:
|
||||
raise HTTPException(404, "工作流不存在，可能已被删除")
|
||||
current = await recall.batches(session, workflow_id, version)
|
||||
if not current["can_continue"]:
|
||||
raise CodedHTTPException(409, current.get("problem") or "没有还没重算的组", recall.REFUSED)
|
||||
launched = await recall.launch(workflow_id, version, batch="manual", actor=current_principal().username)
|
||||
async with SessionLocal() as fresh:
|
||||
view = await recall.batches(fresh, workflow_id, version)
|
||||
return {"launched": launched, **view}
|
||||
|
||||
|
||||
@router.get("/certify-queue")
|
||||
async def certify_queue(session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""可沉淀的问题（认证队列，app/ledger/certify.py）：确定性重复的 SQL 形状、问题聚类、被反复用的答案。每项给代表问题、
|
||||
|
||||
@@ -577,9 +577,12 @@ async def commit_publish(session: AsyncSession, workflow: Workflow, snapshot: Wo
|
||||
workflow.published_by = actor
|
||||
await session.commit()
|
||||
# 口径账本（next-round 02 做法 8）：从问数据固化出的口径发布了，来源运行在后台按它补比一次（不挡发布）
|
||||
from app.ledger import recall as ledger_recall
|
||||
from app.ledger.certify import after_publish
|
||||
|
||||
after_publish(workflow.id, workflow.origin)
|
||||
# 口径召回（next-round 02 做法 7、拍板 L6）：用过旧版本的答案在后台按计划排队重算（每次最多 recall_limit 组），失败不影响发布
|
||||
ledger_recall.after_publish(workflow.id, snapshot.version, actor)
|
||||
return {
|
||||
"ok": True,
|
||||
"level": level,
|
||||
@@ -590,6 +593,20 @@ async def commit_publish(session: AsyncSession, workflow: Workflow, snapshot: Wo
|
||||
}
|
||||
|
||||
|
||||
@router.get("/{workflow_id}/recall-preview")
|
||||
async def recall_preview(workflow_id: str, version: int | None = None,
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""发布弹窗的召回预告（next-round 02 做法 7、8）：把 version（缺省是最新保存的那一版）发布成新版本之后，旧版本被多少问数
|
||||
回答、导出、定时任务、应用用过，预计自动重算几组，超出上限的几组（「另有 M 组未重算」），不能重算的原因各几组。和发布之后
|
||||
真的召回走同一份计划（ledger/recall.build_plan），规模一致。只读，不发起。"""
|
||||
from app.ledger import recall as ledger_recall
|
||||
|
||||
workflow = await session.get(Workflow, workflow_id)
|
||||
if not workflow:
|
||||
raise HTTPException(404, _GONE)
|
||||
return await ledger_recall.preview(session, workflow_id, int(version or workflow.version))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 发布前检查与自动修复：两个接口都只读，不改库、不发布
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
@@ -1184,6 +1184,19 @@ def _summary(etype: str, data: Any) -> str:
|
||||
("none", "没有相关口径"), ("skipped", "没有比完"))
|
||||
if isinstance(counts.get(k), int) and counts[k]]
|
||||
return "封存后比对已发布口径" + (f"：{'、'.join(parts)}" if parts else "")
|
||||
if etype == EventType.LEDGER_RECALLED:
|
||||
# 口径召回（next-round 02 做法 7）：只说按哪一版核对、数变没变，不带新旧值和差值
|
||||
to = data.get("to_version")
|
||||
head = f"已按 v{to} 核对" if isinstance(to, int) else "口径召回核对"
|
||||
changed = data.get("changed")
|
||||
return head + ("：数变了" if changed is True else "，不影响" if changed is False else "")
|
||||
if etype == EventType.LEDGER_BASIS_CHANGED:
|
||||
# 答案依据的目录事实、表结构变了：只说哪一类、几项，不带表名、列名、取值
|
||||
kind = {"catalog": "目录事实", "structure": "表结构", "values": "取值", "catalog_edit": "目录"}.get(
|
||||
str(data.get("kind") or ""), "")
|
||||
items = data.get("items")
|
||||
count = len(items) if isinstance(items, list) else 0
|
||||
return f"依据的{kind or '东西'}有变" + (f"（{count} 项）" if count else "")
|
||||
if etype == EventType.LOG and isinstance(data.get("message"), str):
|
||||
return _clip(data["message"])
|
||||
return ""
|
||||
|
||||
@@ -65,6 +65,7 @@
|
||||
'human.requested': '请求人工审批', 'human.resolved': '人工审批结果', 'approval.inquiry': '审批质询',
|
||||
'issuance': '出具判定', 'caliber.upgrade': '口径卡升版处置', 'catalog.drift': '数据目录变化',
|
||||
'schema.drift': '表结构变化', 'report.checked': '报告自查', 'evidence.judged': '按需裁判', 'file.extracted': '读入附件',
|
||||
'ledger.matched': '比对已发布口径', 'ledger.recalled': '口径召回核对', 'ledger.basis_changed': '答案的依据有变',
|
||||
'log': '日志', 'usage': '用量',
|
||||
}
|
||||
const T = {
|
||||
|
||||
@@ -92,6 +92,16 @@ class EventType(StrEnum):
|
||||
# algo, counts, trigger}。追加在封存之后（post_seal），是批注不是证据：比对依据的已发布口径在运行之外，不能混进运行
|
||||
# 自身的封存事实；封存核对照旧一致
|
||||
LEDGER_MATCHED = "ledger.matched"
|
||||
# 口径召回（02 做法 7）：口径发布新版本后，在同一份数据上把新旧两版各算一次（召回运行，origin_kind=recall）。召回运行
|
||||
# 结束时在它自己身上记一条结论 {role: recall, changed, workflow_id, from_version, to_version, metrics, answers, tasks}；
|
||||
# 数没变的那几个原答案的运行上各记一条 {role: answer, changed: false, recall_run, workflow_id, from_version,
|
||||
# to_version}，界面上写「已按 vN 核对，不影响」（数变了的开 recall 待办，不在原答案上记）。同 ledger.matched：追加在
|
||||
# 封存之后（post_seal），是批注不是证据，不进封存范围
|
||||
LEDGER_RECALLED = "ledger.recalled"
|
||||
# 答案依据的东西变了（02 做法 7 二、三）：kind=catalog 目录里影响数字的一项被驳回或改了（冻结的目录里看得到这一项）；
|
||||
# structure 用到的表或列被删、列的类型变了；values 触及有效记录条件的新码值；catalog_edit 处理变化时改了目录。带数据源、
|
||||
# 表、哪几项（路径和叫法），按新目录跑 SQL 检查新出现的错误级问题条数。同上，封存之后追加，不进封存范围
|
||||
LEDGER_BASIS_CHANGED = "ledger.basis_changed"
|
||||
|
||||
# 应用平台（docs/design/app-platform-and-look.md §3.11）
|
||||
# 运行来源：发起时写进 launch_run 的 notes，在封存范围内（engine/origin.py）
|
||||
|
||||
@@ -919,7 +919,17 @@ async def review_entry(session: AsyncSession, source_id: str, table: str, path:
|
||||
if if_version != current:
|
||||
raise CatalogConflict(table, current)
|
||||
notes = review_item(entry.notes if entry else {}, path, action)
|
||||
return await write_entry(session, source_id, table, notes, if_version=current, actor=actor)
|
||||
written = await write_entry(session, source_id, table, notes, if_version=current, actor=actor)
|
||||
after_review(source_id, table, entry.notes if entry else None, written)
|
||||
return written
|
||||
|
||||
|
||||
def after_review(source_id: str, table: str, before: Mapping[str, Any] | None, written: CatalogEntry) -> None:
|
||||
"""审阅、人工编辑、保存提案写进目录之后（已提交）：影响数字的项被驳回或改了的，口径账本在后台找出依据了它的答案、标
|
||||
「依据有变」（ledger/recall.after_review，docs/design/next-round/02-口径账本.md 做法 7 二）。不挡这次写入，出错只记日志。"""
|
||||
from app.ledger import recall
|
||||
|
||||
recall.after_review(source_id, table, before, written.notes, written.version)
|
||||
|
||||
|
||||
async def save_human_edit(session: AsyncSession, source_id: str, table: str, submitted: Mapping[str, Any], *,
|
||||
@@ -939,7 +949,9 @@ async def save_human_edit(session: AsyncSession, source_id: str, table: str, sub
|
||||
schema_cache = getattr(row, "schema_cache", None)
|
||||
notes = apply_human_edit(entry.notes if entry else {}, submitted, table=table,
|
||||
inferable=structure_relation_ids(schema_cache, table))
|
||||
return await write_entry(session, source_id, table, notes, if_version=current, actor=actor)
|
||||
written = await write_entry(session, source_id, table, notes, if_version=current, actor=actor)
|
||||
after_review(source_id, table, entry.notes if entry else None, written)
|
||||
return written
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
@@ -1251,7 +1263,9 @@ async def save_patch(session: AsyncSession, source_id: str, table: str, changes:
|
||||
if if_version != current:
|
||||
raise CatalogConflict(table, current)
|
||||
notes = apply_patch(entry.notes if entry else {}, changes, table=table, tables=tables)
|
||||
return await write_entry(session, source_id, table, notes, if_version=current, actor=actor)
|
||||
written = await write_entry(session, source_id, table, notes, if_version=current, actor=actor)
|
||||
after_review(source_id, table, entry.notes if entry else None, written)
|
||||
return written
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
|
||||
@@ -22,7 +22,8 @@
|
||||
使用次数比运行次数多时两个都写（「被 2 次运行用到，使用 5 次」），和清单、运行列表上的数都对得上。
|
||||
- **批量审阅**（review_batch，POST /catalog/review/batch）：逐表按 if_version 乐观锁，同一张表的几条在一次写入里完成，
|
||||
一张表冲突、写不进去不影响别的表；路径不对、表不存在逐条报（results 和请求一一对应）。审阅的规则同单项审阅
|
||||
（catalog.review_item）。
|
||||
（catalog.review_item）。写进去之后影响数字的项被驳回了的，口径账本在后台标依据了它的答案「依据有变」
|
||||
（catalog.after_review → ledger/recall，next-round 02 做法 7 二）。
|
||||
- **被哪些运行用过**（table_runs，GET /catalog/{table}/usage）：取法同使用次数（catalog.query_snapshot_tables），按运行
|
||||
去重，新的在前；total 是运行次数。
|
||||
- **进度**（review_progress，GET /catalog/review-progress）：常用表（使用次数前 TOP_N 张，和起草对话框「使用次数最多的
|
||||
@@ -379,6 +380,8 @@ async def review_batch(session: AsyncSession, source: Any, items: Iterable[Batch
|
||||
continue
|
||||
for i in applied:
|
||||
results[i] = _result(items[i], ok=True, version=written.version)
|
||||
# 驳回了影响数字的项：口径账本在后台标依据了它的答案「依据有变」（catalog.after_review，同单项审阅）
|
||||
catalog.after_review(source.id, key, entry.notes if entry else None, written)
|
||||
return [r for r in results if r is not None]
|
||||
|
||||
|
||||
|
||||
@@ -11,7 +11,9 @@
|
||||
起草（建起草任务）、重新剖析（建剖析任务）、知道了、误报（必须写理由）。每个动作写 resolution，动了目录的记下目录版本，
|
||||
建了任务的记下任务编号。
|
||||
- **拉取**（list_changes / affecting）：给召回按 seq 游标拉；seq 自增、不复用。触发召回的只有结构类、处理时改了目录的、
|
||||
以及 rules_touched 的新码值（triggers_recall）；纯取值漂移只算数据质量提醒。
|
||||
以及 rules_touched 的新码值（triggers_recall）；纯取值漂移只算数据质量提醒。探查、剖析任务结束时（notify_after_job）
|
||||
口径账本在后台按游标接着拉（ledger/recall.consume_source_changes）；处理变化时改了目录的那一条当场交过去
|
||||
（ledger/recall.process_change：它记下来的时候游标早已经过去了）。
|
||||
- **说法**（summary）：一条变化给人看的一句（接口、日志用）；通知里不放它（不放表名、列名、取值，D9），见
|
||||
notify/service.notify_source_changed。
|
||||
|
||||
@@ -340,10 +342,17 @@ async def notify_after_job(source_id: str, job_id: str) -> int:
|
||||
from app.notify import service
|
||||
|
||||
service.spawn(service.notify_source_changed(source_id))
|
||||
return n
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("任务 %s 结束后推送数据源变化的通知时出错", job_id)
|
||||
return 0
|
||||
n = 0
|
||||
try:
|
||||
# 口径账本的召回（next-round 02 做法 7 三）：这次记下的变化里表或列被删、类型变了的，标用过它的答案「依据的表结构已变」
|
||||
from app.ledger import recall
|
||||
|
||||
recall.spawn(recall.consume_source_changes())
|
||||
except Exception: # noqa: BLE001 - 辅助步骤
|
||||
logger.exception("任务 %s 结束后排队召回时出错", job_id)
|
||||
return n
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
@@ -469,6 +478,11 @@ async def handle(session: AsyncSession, source: Any, row: SourceChange, action:
|
||||
if version is not None:
|
||||
out["catalog_version"] = version
|
||||
out["change"] = change_out(row)
|
||||
if version is not None and triggers_recall(out["change"]):
|
||||
# 处理时改了目录：用过这张表的答案「依据有变」（口径账本，后台；游标早已过了这一条，当场交过去）
|
||||
from app.ledger import recall
|
||||
|
||||
recall.spawn(recall.process_change(out["change"]))
|
||||
return out
|
||||
|
||||
|
||||
|
||||
@@ -17,9 +17,10 @@ from sqlalchemy.orm import Mapped, mapped_column
|
||||
from app.db.base import Base, TimestampMixin, UTCDateTime, new_id, utcnow
|
||||
|
||||
#: 待办的种类。人工节点三种 + 现有的几种中断 + 动作确认、核实、等外部系统 + 个人应用提交审核（review，
|
||||
#: docs/design/normal-user-apps.md N7，app/process/tasks.REVIEW_KIND）
|
||||
#: docs/design/normal-user-apps.md N7，app/process/tasks.REVIEW_KIND）+ 口径召回（recall，docs/design/next-round/02-口径账本.md
|
||||
#: 做法 7：口径升版后答案的数变了、答案依据的目录或表结构变了；挂在召回运行或答案的运行上，没有 Approval，app/ledger/recall）
|
||||
TASK_KINDS = ("approve", "input", "edit", "tool", "code", "dual_check", "action_confirm", "reconcile", "signal",
|
||||
"review")
|
||||
"review", "recall")
|
||||
#: pending 等人；resolving 已凑齐、正在恢复运行（条件 UPDATE 抢到的那个请求负责）；resolved / cancelled / expired 终态
|
||||
TASK_STATUSES = ("pending", "resolving", "resolved", "cancelled", "expired")
|
||||
#: 动作状态。simulated：探索运行里端点没有 test 环境时的演练，不发送
|
||||
|
||||
@@ -133,15 +133,20 @@ class RunManager:
|
||||
# 口径账本：最近几天封存了、一次都没比过的探索运行，在后台补比一次（ledger/match.py）。只读工件和版本表，不查业务库、
|
||||
# 不调模型；和调度循环同一个开关，拷来的数据目录、测试不补
|
||||
from app.ledger import match as ledger_match
|
||||
from app.ledger import recall as ledger_recall
|
||||
|
||||
ledger_match.start_backfill()
|
||||
# 口径召回（ledger/recall.py）：最近发布、自动那一批没发完的接着发，结束了还没收尾的召回运行补上收尾。同一个开关
|
||||
ledger_recall.start_resume()
|
||||
|
||||
async def shutdown(self) -> None:
|
||||
self._closing = True
|
||||
# 后台的口径比对先停：没做完的不记，下次启动补比会再比
|
||||
# 后台的口径比对、召回先停：没做完的不记，下次启动补比、接着召回会再做
|
||||
from app.ledger import match as ledger_match
|
||||
from app.ledger import recall as ledger_recall
|
||||
|
||||
await ledger_match.shutdown()
|
||||
await ledger_recall.shutdown()
|
||||
# 只等还没结束的：结束了的没什么可等，而且它可能属于另一个事件循环，
|
||||
# gather 会直接抛 "future belongs to a different loop"
|
||||
pending = [t for t in self._tasks.values() if not t.done()]
|
||||
@@ -1084,6 +1089,11 @@ class RunManager:
|
||||
|
||||
if ledger_match.wanted(run_class, origin_kind):
|
||||
ledger_match.schedule(run_id)
|
||||
if origin_kind == "recall" and status in ("succeeded", "failed", "cancelled"):
|
||||
# 口径召回运行结束、封存完（ledger/recall.py）：在后台收尾——数变了开 recall 待办，没变的在原答案上记账
|
||||
from app.ledger import recall as ledger_recall
|
||||
|
||||
ledger_recall.schedule_settle(run_id)
|
||||
|
||||
async def _seal(self, run_id: str) -> None:
|
||||
"""对到此为止的全部事件算清单哈希，记下封到了哪一条。"""
|
||||
|
||||
@@ -17,16 +17,16 @@ cell(nodes.X, 0, '列')、X 是查库的调用工具节点（look/calibers.card_
|
||||
拒收、给原因（NotDerivable）。工作流、版本、节点不在了也照实说。
|
||||
|
||||
**已实现的**：caliber_probe —— 一份口径的取数 + 口径卡（评测口径题的期望）；reconcile —— 口径对账（第 3 波，
|
||||
reconcile_graph）；组合式问数用的 caliber_block(card_id, inputs)（第 3 波）。
|
||||
reconcile_graph）；组合式问数用的 caliber_block(card_id, inputs)（第 3 波）；recall —— 口径召回（第 4 波，recall_graph）。
|
||||
|
||||
**扩展点**（第 3 波的账本按这里扩，不另写生成器）：
|
||||
- 口径对账（02 做法 6）：assemble("reconcile", [ours, caliber_block(pinned)], tail=[对账卡])。ours 是「这次的取数」
|
||||
（执行过的 SQL 原文，节点 id ours_q1…，build 一个 Block 即可，不经 caliber_block），对账卡是本地 metrics 节点（差值、
|
||||
相对差，式子引用 cell(nodes.ours_q1, 0, '列') 和口径卡的指标，on_missing=null），出口的契约写 metrics_from。
|
||||
- 口径召回（02 做法 7）：assemble("recall", [caliber_block(old, prefix="old_"), caliber_block(new, prefix="new_")],
|
||||
tail=[召回卡])。同一张图里放两份口径，取数节点 id 会撞：caliber_block 带 prefix 时把拷出来的节点改名，口径卡的
|
||||
caliber_from 带上改名映射 nodes={原 id: 新 id}，运行时 metrics._pinned_caliber 按语法树改写式子里的 cell(nodes.X…)
|
||||
和 nodes.X（第 3 波起）。
|
||||
- 口径召回（02 做法 7，recall_graph）：assemble("recall", [caliber_block(old, prefix="old_"), caliber_block(new,
|
||||
prefix="new_")], tail=[召回卡])。同一张图里放两份口径，取数节点 id 会撞：caliber_block 带 prefix 时把拷出来的节点改名，
|
||||
口径卡的 caliber_from 带上改名映射 nodes={原 id: 新 id}，运行时 metrics._pinned_caliber 按语法树改写式子里的
|
||||
cell(nodes.X…) 和 nodes.X（第 3 波起）。
|
||||
- 组合式问数（02 做法 4，ledger/compose.py）：不经 assemble（拷进的是问数据那张图），只用 caliber_block(prefix=…,
|
||||
card_id=…, inputs=…)：取数节点改名、模板按助手给的参数渲染成字面 SQL，口径卡带改名映射。
|
||||
- 新的派生种类登记进 KINDS，入口字段、出口、首尾相接都走 assemble，不在调用方拼节点。
|
||||
@@ -44,7 +44,7 @@ from typing import Any
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
#: 派生图的种类。caliber_probe、reconcile 已实现；recall 归第 4 波（见模块说明的扩展点）
|
||||
#: 派生图的种类（见模块说明）
|
||||
KINDS = ("caliber_probe", "reconcile", "recall")
|
||||
|
||||
#: 不能派生的原因的机器码
|
||||
@@ -380,6 +380,53 @@ def reconcile_graph(pinned: Pinned, ours: Block, tail: Sequence[dict[str, Any]],
|
||||
contract={"metrics_from": RECONCILE_CARD} if card is not None else None)
|
||||
|
||||
|
||||
#: 召回图里的召回卡（细案 02 做法 7）
|
||||
RECALL_CARD = "recall"
|
||||
|
||||
|
||||
def recall_prefixes(i: int) -> tuple[str, str]:
|
||||
"""召回图里第 i 对口径卡（旧版、新版）各自的前缀：第一对 old_ / new_，之后 old2_ / new2_……（同一版里两张卡读同一个
|
||||
取数节点时，各拷一份也不撞）。"""
|
||||
n = "" if i == 0 else str(i + 1)
|
||||
return f"old{n}_", f"new{n}_"
|
||||
|
||||
|
||||
def recall_fields(old: Pinned, new: Pinned, inputs: Mapping[str, Any]) -> list[dict[str, Any]]:
|
||||
"""召回图的入口：新版的入口字段在前（类型、必填照新版），旧版有、新版没有的接在后面；缺省值都换成这一组的参数。"""
|
||||
fields: list[dict[str, Any]] = []
|
||||
seen: set[str] = set()
|
||||
for f in [*new.entry_fields, *old.entry_fields]:
|
||||
name = str(f.get("name") or "")
|
||||
if not name or name in seen:
|
||||
continue
|
||||
seen.add(name)
|
||||
item = copy.deepcopy(dict(f))
|
||||
if name in inputs:
|
||||
item["default"] = inputs[name]
|
||||
fields.append(item)
|
||||
return fields
|
||||
|
||||
|
||||
def recall_graph(pairs: Sequence[tuple[Pinned, Pinned]], tail: Sequence[dict[str, Any]], *,
|
||||
inputs: Mapping[str, Any]) -> DerivedGraph:
|
||||
"""口径召回（细案 02 做法 7）：入口（这一组的参数）→ 每一对口径卡：旧版的取数 + 旧版口径卡（caliber_from 固定旧版）、
|
||||
新版的取数 + 新版口径卡（固定新版），两版的取数节点各带前缀、口径卡的 caliber_from 带改名映射（不撞 id）→ tail（召回卡：
|
||||
两版之差在这里算，ledger/recall.recall_card）→ 出口（契约 metrics_from 指向召回卡；不加报告、不调模型）。
|
||||
|
||||
两版在同一次运行里先后查同一份数据：差值只反映口径的变化（数据随时间的变化另拿原答案当时的值对照着说）。入口字段按
|
||||
第一对的新版、旧版合起来（recall_fields）。同样的输入永远得到同一张图。"""
|
||||
if not pairs:
|
||||
raise ValueError("召回图至少要一对口径卡")
|
||||
blocks: list[Block] = []
|
||||
for i, (old, new) in enumerate(pairs):
|
||||
old_prefix, new_prefix = recall_prefixes(i)
|
||||
blocks += [caliber_block(old, prefix=old_prefix), caliber_block(new, prefix=new_prefix)]
|
||||
old, new = pairs[0]
|
||||
card = next((n for n in tail if n.get("type") == "metrics" and n.get("id") == RECALL_CARD), None)
|
||||
return assemble("recall", blocks, fields=recall_fields(old, new, inputs), defaults=new.defaults or old.defaults,
|
||||
tail=tail, contract={"metrics_from": RECALL_CARD} if card is not None else None)
|
||||
|
||||
|
||||
def caliber_probe(pinned: Pinned, *, upgrade_policy: str | None = None) -> DerivedGraph:
|
||||
"""口径探针：一份口径的取数 + 口径卡（评测口径题的期望，细案 01 做法 8）。节点 id 和原图一样，入口字段照抄。
|
||||
同样的 pinned 永远得到同一张图（确定性）。"""
|
||||
@@ -388,6 +435,6 @@ def caliber_probe(pinned: Pinned, *, upgrade_policy: str | None = None) -> Deriv
|
||||
|
||||
|
||||
__all__ = ["ALIGN_ID", "Block", "DerivedGraph", "ENTRY_ID", "EXIT_ID", "GONE", "KINDS", "NOT_CARD", "NOT_REUSABLE",
|
||||
"NotDerivable", "OURS_ID", "Pinned", "RECONCILE_CARD", "TEMPLATE", "assemble", "caliber_block",
|
||||
"caliber_probe", "entry_fields", "load_pinned", "ours_block", "reconcile_graph", "reconcile_prefix", "render_inputs",
|
||||
"template_problem"]
|
||||
"NotDerivable", "OURS_ID", "Pinned", "RECALL_CARD", "RECONCILE_CARD", "TEMPLATE", "assemble", "caliber_block",
|
||||
"caliber_probe", "entry_fields", "load_pinned", "ours_block", "recall_fields", "recall_graph",
|
||||
"recall_prefixes", "reconcile_graph", "reconcile_prefix", "render_inputs", "template_problem"]
|
||||
|
||||
+65
-42
@@ -208,12 +208,7 @@ async def plan_issue(session: Any, run: Run | None, content: dict[str, Any] | No
|
||||
hide 判一个绑定值是不是这次运行查到过的遮罩值。只读，不发起。"""
|
||||
from app.api.deps import can_read_run
|
||||
from app.api.settings import ledger_settings
|
||||
from app.apps import versions as app_versions
|
||||
from app.apps.backfill import graph_side_effect
|
||||
from app.apps.forms import input_schema, validate_form
|
||||
from app.apps.resolve import input_key
|
||||
from app.ledger import match as ledger_match
|
||||
from app.ledger.derive import entry_fields
|
||||
from app.ledger.labels import caliber_key
|
||||
|
||||
plan = Plan(run_id=run.id if run else "", workflow_id=workflow_id, version=int(version),
|
||||
@@ -237,43 +232,9 @@ async def plan_issue(session: Any, run: Run | None, content: dict[str, Any] | No
|
||||
plan.binding = dict(whole.get("binding") or {})
|
||||
if hide is not None and any(hide(v) for v in plan.binding.values()):
|
||||
return plan.refuse(MASKED_BINDING)
|
||||
if workflow is None:
|
||||
return plan.refuse(GONE, 404)
|
||||
row = await app_versions.version_row(session, workflow_id, int(version))
|
||||
if row is None:
|
||||
return plan.refuse(VERSION_GONE.format(name=workflow.name, version=version), 404)
|
||||
if row.retired_at is not None:
|
||||
return plan.refuse(RETIRED.format(name=workflow.name, version=version))
|
||||
if not app_versions.is_published(row, current=workflow.published_version):
|
||||
return plan.refuse(UNPUBLISHED.format(name=workflow.name, version=version))
|
||||
plan.level = row.level or plan.level
|
||||
plan.expect_hash = row.graph_hash
|
||||
if reason := await graph_side_effect(session, row.graph, level=row.level):
|
||||
return plan.refuse(SIDE_EFFECT.format(name=workflow.name, version=version, reason=reason))
|
||||
fields = list(entry_fields(row.graph or {}))
|
||||
values = {str(f["name"]): f.get("default") for f in fields if f.get("default") not in (None, "")}
|
||||
values.update({k: v for k, v in plan.binding.items() if v not in (None, "")})
|
||||
plan.inputs = values
|
||||
if errors := validate_form(input_schema(fields), values):
|
||||
return plan.refuse(INPUTS.format(problems="；".join(e["message"] for e in errors)))
|
||||
# 参数运行时要渲染进口径的 SQL 模板：写在引号里的不能带引号、裸写的只能是数（ledger/compose 同一个判断）
|
||||
from app.ledger.compose import slot_contexts, value_problem
|
||||
|
||||
contexts = slot_contexts([n for n in (row.graph or {}).get("nodes") or [] if n.get("type") == "tool"],
|
||||
strict=False)
|
||||
if problem := next((p for k, v in values.items() if (p := value_problem(k, v, contexts.get(k, set())))), None):
|
||||
return plan.refuse(f"{problem}，不代填出具")
|
||||
if principal.mode != "off" and not principal.at_least("builder"):
|
||||
plan.granted_by_app = await _granting_app(session, principal, workflow_id, int(version))
|
||||
if plan.granted_by_app is None:
|
||||
return plan.refuse(ASK_BUILDER, 403, FORBIDDEN)
|
||||
key = input_key(values)
|
||||
plan.existing = next((r.id for r in await issued_runs(session, run.id)
|
||||
if (r.workflow_id, r.version) == (workflow_id, int(version)) and r.status in _LIVE
|
||||
and input_key(r.input) == key), None)
|
||||
if await check_target(session, plan, workflow, principal=principal, source_run=run.id) is None:
|
||||
return plan
|
||||
settings = await ledger_settings(session)
|
||||
if plan.existing is None and await issued_today(session) >= int(settings["issue_daily_limit"]):
|
||||
return plan.refuse(OVER_LIMIT.format(limit=settings["issue_daily_limit"]), 429, DAILY_LIMIT)
|
||||
if not settings["auto_issue"]:
|
||||
plan.auto_reason = AUTO_OFF
|
||||
elif run.origin_kind != "chat":
|
||||
@@ -287,6 +248,68 @@ async def plan_issue(session: Any, run: Run | None, content: dict[str, Any] | No
|
||||
return plan
|
||||
|
||||
|
||||
async def check_target(session: Any, plan: Plan, workflow: Any, *, principal: Any, source_run: str) -> Plan | None:
|
||||
"""按口径出具的那一版能不能发起（plan_issue 和召回待办的「按新版重新出具」共用，ledger/recall.reissue）：那一版已发布、
|
||||
未退役；图里没有人工节点、不会写外部系统；入口字段的缺省值加上 plan.binding 过表单校验、能渲染进取数模板；使用者只能按
|
||||
他有权使用的应用固定着的这一版（拍板 L5）；同一个来源（source_run）、同一版、同一组参数发起过的交回那一次；当天没超
|
||||
上限。填好 plan 的 level、expect_hash、inputs、granted_by_app、existing；不能出具的写进 plan.refuse，交回 None。"""
|
||||
from app.api.settings import ledger_settings
|
||||
from app.apps import versions as app_versions
|
||||
from app.apps.backfill import graph_side_effect
|
||||
from app.apps.forms import input_schema, validate_form
|
||||
from app.apps.resolve import input_key
|
||||
from app.ledger.derive import entry_fields
|
||||
|
||||
workflow_id, version = plan.workflow_id, plan.version
|
||||
if workflow is None:
|
||||
plan.refuse(GONE, 404)
|
||||
return None
|
||||
row = await app_versions.version_row(session, workflow_id, int(version))
|
||||
if row is None:
|
||||
plan.refuse(VERSION_GONE.format(name=workflow.name, version=version), 404)
|
||||
return None
|
||||
if row.retired_at is not None:
|
||||
plan.refuse(RETIRED.format(name=workflow.name, version=version))
|
||||
return None
|
||||
if not app_versions.is_published(row, current=workflow.published_version):
|
||||
plan.refuse(UNPUBLISHED.format(name=workflow.name, version=version))
|
||||
return None
|
||||
plan.level = row.level or plan.level
|
||||
plan.expect_hash = row.graph_hash
|
||||
if reason := await graph_side_effect(session, row.graph, level=row.level):
|
||||
plan.refuse(SIDE_EFFECT.format(name=workflow.name, version=version, reason=reason))
|
||||
return None
|
||||
fields = list(entry_fields(row.graph or {}))
|
||||
values = {str(f["name"]): f.get("default") for f in fields if f.get("default") not in (None, "")}
|
||||
values.update({k: v for k, v in plan.binding.items() if v not in (None, "")})
|
||||
plan.inputs = values
|
||||
if errors := validate_form(input_schema(fields), values):
|
||||
plan.refuse(INPUTS.format(problems="；".join(e["message"] for e in errors)))
|
||||
return None
|
||||
# 参数运行时要渲染进口径的 SQL 模板：写在引号里的不能带引号、裸写的只能是数（ledger/compose 同一个判断）
|
||||
from app.ledger.compose import slot_contexts, value_problem
|
||||
|
||||
contexts = slot_contexts([n for n in (row.graph or {}).get("nodes") or [] if n.get("type") == "tool"],
|
||||
strict=False)
|
||||
if problem := next((p for k, v in values.items() if (p := value_problem(k, v, contexts.get(k, set())))), None):
|
||||
plan.refuse(f"{problem}，不代填出具")
|
||||
return None
|
||||
if principal.mode != "off" and not principal.at_least("builder"):
|
||||
plan.granted_by_app = await _granting_app(session, principal, workflow_id, int(version))
|
||||
if plan.granted_by_app is None:
|
||||
plan.refuse(ASK_BUILDER, 403, FORBIDDEN)
|
||||
return None
|
||||
key = input_key(values)
|
||||
plan.existing = next((r.id for r in await issued_runs(session, source_run)
|
||||
if (r.workflow_id, r.version) == (workflow_id, int(version)) and r.status in _LIVE
|
||||
and input_key(r.input) == key), None)
|
||||
settings = await ledger_settings(session)
|
||||
if plan.existing is None and await issued_today(session) >= int(settings["issue_daily_limit"]):
|
||||
plan.refuse(OVER_LIMIT.format(limit=settings["issue_daily_limit"]), 429, DAILY_LIMIT)
|
||||
return None
|
||||
return plan
|
||||
|
||||
|
||||
async def _conversation_of(session: Any, run: Run) -> dict[str, str]:
|
||||
"""这次运行出自哪段对话的哪一轮：运行来源里封存的（origin_ref，第 2 波）；老运行退回 conversation_turns.run_id。"""
|
||||
from app.db.models import ConversationTurn
|
||||
@@ -392,4 +415,4 @@ async def auto_issue(run_id: str) -> str | None:
|
||||
|
||||
|
||||
__all__ = ["AUTO_NOT_CHAT", "DAILY_LIMIT", "FORBIDDEN", "ORIGIN", "REFUSED", "IssueRefused", "Plan", "auto_issue",
|
||||
"issue", "issued_runs", "issued_today", "issued_view", "plan_issue", "plans_for"]
|
||||
"check_target", "issue", "issued_runs", "issued_today", "issued_view", "plan_issue", "plans_for"]
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -17,8 +17,9 @@
|
||||
段对话的算重复的形状，给出这样的形状几个、有重复形状的运行占几成；问题聚类（app/chat/questions：全站唯一的聚类，只取问题
|
||||
原文、不含 history、排除系统派生的运行）里 ≥ REPEAT_RUNS 个问题的类占几成。认证队列的转化（第 3 波，ledger/certify）：固化成
|
||||
口径草稿几份（origin.freeze 的工作流）、其中发布了几份、点了「不是口径」几项。
|
||||
- E4 召回：每个口径版本被用了多少次（比对里等价已证明的运行数、这一版的正式运行数）。每次发布召回了几组、其中数变了的比例要等
|
||||
召回（第 4 波），这之前照实写没有数据，只给召回运行的次数。
|
||||
- E4 召回：每个口径版本被用了多少次（比对里等价已证明的运行数、这一版的正式运行数）；每次发布（口径升到某一版）召回了几组、
|
||||
收过尾的几组、其中数变了的几组和比例、重算失败的几组（ledger/recall：召回运行和它收尾时记的 ledger.recalled，第 4 波）。
|
||||
按召回运行的发起时间筛窗口；表结构变了之后的口径试算不算召回。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -43,7 +44,6 @@ DIFF_GROUPS = ("time", "filter", "aggregate", "join", "other", "parse")
|
||||
STATUSES = ("proven", "composed", "related", "none", "skipped")
|
||||
#: 没有数据时的说法（这一块的功能还没做）
|
||||
NOT_YET_RECONCILE = "对账面板的反馈在第 4 波接上，目前没有数据"
|
||||
NOT_YET_RECALL = "召回在第 4 波接上，目前没有数据"
|
||||
#: 聚类最多取几个问题（同治理接口的上限）
|
||||
QUESTION_LIMIT = 500
|
||||
|
||||
@@ -294,7 +294,50 @@ async def _e4(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic
|
||||
usage = [{"caliber": key, "workflow_id": key.rsplit("@", 1)[0], "version": int(key.rsplit("@", 1)[1]),
|
||||
"proven_runs": proven.get(key, 0), "formal_runs": formal.get(key, 0)}
|
||||
for key in sorted(set(proven) | set(formal))]
|
||||
return {"usage": usage, "recall_runs": recalls, "per_publish": {"available": False, "note": NOT_YET_RECALL}}
|
||||
return {"usage": usage, "recall_runs": recalls, "per_publish": await _per_publish(session, **scope)}
|
||||
|
||||
|
||||
async def _per_publish(session: Any, **scope: Any) -> dict[str, Any]:
|
||||
"""每次发布召回了几组、其中数变了的比例（E4 下半）：按（工作流、新版本）分，读召回运行收尾时记的结论。"""
|
||||
from app.core.events import EventType
|
||||
from app.db.models import Run, RunEvent
|
||||
|
||||
stmt = select(Run.id, Run.origin).where(Run.origin_kind == "recall")
|
||||
if scope.get("run_ids") is not None:
|
||||
stmt = stmt.where(Run.id.in_(list(scope["run_ids"])))
|
||||
else:
|
||||
if scope.get("since") is not None:
|
||||
stmt = stmt.where(Run.created_at >= scope["since"])
|
||||
if scope.get("until") is not None:
|
||||
stmt = stmt.where(Run.created_at < scope["until"])
|
||||
runs = [(rid, o) for rid, o in (await session.execute(stmt)).all()
|
||||
if isinstance(o, dict) and not o.get("probe") and o.get("workflow_id")]
|
||||
settled: dict[str, dict[str, Any]] = {}
|
||||
if runs:
|
||||
rows = (await session.execute(select(RunEvent.run_id, RunEvent.data).where(
|
||||
RunEvent.run_id.in_([r for r, _ in runs]), RunEvent.type == EventType.LEDGER_RECALLED)
|
||||
.order_by(RunEvent.seq))).all()
|
||||
for rid, data in rows:
|
||||
if isinstance(data, dict) and data.get("role") == "recall":
|
||||
settled[rid] = data
|
||||
by: dict[tuple[str, int], dict[str, int]] = {}
|
||||
for rid, origin in runs:
|
||||
key = (str(origin["workflow_id"]), int(origin.get("to_version") or 0))
|
||||
item = by.setdefault(key, {"groups": 0, "settled": 0, "changed": 0, "failed": 0})
|
||||
item["groups"] += 1
|
||||
data = settled.get(rid)
|
||||
if data is None:
|
||||
continue
|
||||
item["settled"] += 1
|
||||
if data.get("outcome") == "failed":
|
||||
item["failed"] += 1
|
||||
elif data.get("changed"):
|
||||
item["changed"] += 1
|
||||
items = [{"caliber": ledger_labels.caliber_key(w, v), "workflow_id": w, "version": v, **c,
|
||||
"changed_rate": _ratio(c["changed"], c["settled"] - c["failed"])}
|
||||
for (w, v), c in sorted(by.items())]
|
||||
return {"available": True, "publishes": items, "groups": sum(i["groups"] for i in items),
|
||||
"changed": sum(i["changed"] for i in items)}
|
||||
|
||||
|
||||
async def compute_stats(session: Any, *, since: datetime | None = None, until: datetime | None = None,
|
||||
|
||||
@@ -0,0 +1,382 @@
|
||||
"""一份口径的旧版本被谁用过（docs/design/next-round/02-口径账本.md 做法 7 一）：口径发布新版本之后，召回要追的那些答案。
|
||||
|
||||
**为什么单独一个模块。** 召回（ledger/recall）要按参数去重成组、重算，发布弹窗要预告规模，版本历史要写「这一版被多少问数
|
||||
回答、导出、召回用过」——三处要的是同一份「用过」的清单。各数各的，预告的规模和实际召回的就对不上（验收要求两者一致）。
|
||||
|
||||
反查的对象（工作流 W 的旧版本 V，只看 V < 新版本）：
|
||||
- **问数回答**（answer）：caliber_match 工件的 meta 里记着 W@V 等价已证明（含助手选用）的探索运行，取这次运行现在这次封存
|
||||
的最新一份比对（ledger/store.latest：之后补比过、按新口径不再等价的不算），每条等价的查询一份参数（绑定）；
|
||||
- **W@V 的正式运行**：定时任务发起的（schedule，每个定时任务最近 PERIODS 期）、应用发起的（app，每个应用最近 PERIODS 次）、
|
||||
别处发起的（formal：手动、接口、按口径出具……），参数是那次运行的输入；
|
||||
- **固定了 W@V 的工作流版本发起的正式运行**（pinned：子工作流或口径卡 caliber_from 固定在 W@V，apps/versions.pinned_upstream
|
||||
同一个认法），参数是那次运行的输入里 W 的入口有的字段；
|
||||
- **固定这一版的应用**（apps/versions.pinning_apps）：应用发起过的正式运行在上面；还没有运行的只列出来，没有可重算的参数；
|
||||
- **导出记录**：上面这些运行上的 report.exported 审计（第 2 波），导出人也是答案的主人；
|
||||
- **问数轮次**：运行 origin 里封存的 conversation_id / turn_id（origin_ref，第 2 波），老运行退回 conversation_turns.run_id。
|
||||
|
||||
**答案的主人**（owners，召回待办派给谁）：对话的 created_by（没有的退回运行的发起人）、导出人、定时任务的 created_by、应用的
|
||||
管理者（授权里的 manage，没有就是建应用的人）；别处发起的正式运行是它的发起人。只记 (user|group, 名字)，组由开待办时展开。
|
||||
|
||||
只读应用库：不查业务库、不调模型。系统派生的运行（评测、对账、召回，engine/origin.counted_runs）不算「用过」。数量有上限
|
||||
（MAX_ANSWERS、MAX_FORMAL），按新的在前取，超出的照实算进 truncated。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable, Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import and_, func, or_, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.db.models import Artifact, Run, Workflow, WorkflowVersion
|
||||
|
||||
#: 用法的种类。answer 问数回答；schedule 定时任务发起的正式运行；app 应用发起的；formal 别处发起的；pinned 固定了这一版的
|
||||
#: 别的工作流版本发起的
|
||||
KINDS = ("answer", "schedule", "app", "formal", "pinned")
|
||||
#: 每个定时任务、每个应用最近几期（细案「定时任务最近 K 期」）
|
||||
PERIODS = 4
|
||||
#: 最多看多少次问数回答、多少次正式运行（各按新的在前）
|
||||
MAX_ANSWERS = 2000
|
||||
MAX_FORMAL = 500
|
||||
#: 种类的叫法（版本用量、召回批次、待办里写「来自问数据」……）
|
||||
KIND_TEXT = {"answer": "问数回答", "schedule": "定时任务", "app": "应用", "formal": "正式运行",
|
||||
"pinned": "固定了它的工作流"}
|
||||
#: 导出审计的动作（api/evidence.py、api/bundle.py 写的）
|
||||
EXPORTED = "report.exported"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Use:
|
||||
"""一次「用过」：哪次运行、旧版本几、参数（原样，没补缺省值）、什么时候、谁是答案的主人。"""
|
||||
|
||||
kind: str
|
||||
run_id: str
|
||||
version: int
|
||||
binding: dict[str, Any]
|
||||
at: datetime | None = None
|
||||
owners: list[tuple[str, str]] = field(default_factory=list)
|
||||
conversation_id: str | None = None
|
||||
turn_id: str | None = None
|
||||
schedule_id: str | None = None
|
||||
app_id: str | None = None
|
||||
#: 问数回答：等价的是哪条查询（查询快照）、口径的哪个取数节点
|
||||
query: str | None = None
|
||||
node_id: str | None = None
|
||||
#: 固定了这一版的是哪个工作流的哪一版（pinned）
|
||||
via_workflow: str | None = None
|
||||
via_version: int | None = None
|
||||
exporters: list[str] = field(default_factory=list)
|
||||
|
||||
def add_owner(self, kind: str, name: str | None) -> None:
|
||||
if name and (kind, name) not in self.owners:
|
||||
self.owners.append((kind, name))
|
||||
|
||||
def describe(self) -> dict[str, Any]:
|
||||
"""记进召回运行 origin 的那一份（召回结束时据此开待办、在原答案上记账）：只放编号、参数的来处和主人。"""
|
||||
out: dict[str, Any] = {"kind": self.kind, "run_id": self.run_id, "version": self.version,
|
||||
"owners": [list(o) for o in self.owners]}
|
||||
for key in ("conversation_id", "turn_id", "schedule_id", "app_id", "query", "node_id", "via_workflow",
|
||||
"via_version"):
|
||||
if (value := getattr(self, key)) is not None:
|
||||
out[key] = value
|
||||
if self.exporters:
|
||||
out["exporters"] = list(self.exporters)
|
||||
if self.at is not None:
|
||||
out["at"] = self.at.isoformat()
|
||||
return out
|
||||
|
||||
|
||||
@dataclass
|
||||
class Found:
|
||||
"""反查的结果：用法、固定着旧版本的应用（{版本: [应用]}）、有没有因为上限少看了。"""
|
||||
|
||||
uses: list[Use]
|
||||
apps: dict[int, list[dict[str, Any]]] = field(default_factory=dict)
|
||||
truncated: bool = False
|
||||
|
||||
def versions(self) -> list[int]:
|
||||
return sorted({u.version for u in self.uses} | set(self.apps))
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 反查
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
async def _answers(session: AsyncSession, workflow_id: str, below: int | None, only: int | None
|
||||
) -> tuple[list[Use], bool]:
|
||||
"""问数回答：比对工件 meta 里等价已证明的那几份口径里有 W@V 的探索运行（见模块说明）。"""
|
||||
from app.engine.origin import counted_runs
|
||||
from app.ledger import store
|
||||
from app.ledger.labels import caliber_key
|
||||
|
||||
prefix = f"{workflow_id}@"
|
||||
rows = (await session.execute(
|
||||
select(Artifact.run_id, Artifact.meta, Run.finished_at).join(Run, Run.id == Artifact.run_id)
|
||||
.where(Artifact.kind == store.KIND, Run.status == "succeeded", counted_runs())
|
||||
.order_by(Run.finished_at.desc()))).all()
|
||||
wanted: dict[str, datetime | None] = {}
|
||||
for run_id, meta, at in rows:
|
||||
keys = [k for k in ((meta or {}).get("calibers") or []) if isinstance(k, str) and k.startswith(prefix)]
|
||||
if not keys or run_id in wanted:
|
||||
continue
|
||||
versions = {int(k.rsplit("@", 1)[1]) for k in keys if k.rsplit("@", 1)[1].isdigit()}
|
||||
if any((only is None or v == only) and (below is None or v < below) for v in versions):
|
||||
wanted[run_id] = at
|
||||
truncated = len(wanted) > MAX_ANSWERS
|
||||
out: list[Use] = []
|
||||
for run_id, at in list(wanted.items())[:MAX_ANSWERS]:
|
||||
recorded = await store.latest(run_id)
|
||||
content = recorded.content if recorded is not None else None
|
||||
if not isinstance(content, dict):
|
||||
continue
|
||||
seen: set[tuple[int, str]] = set()
|
||||
for query in content.get("queries") or []:
|
||||
for cand in query.get("candidates") or []:
|
||||
if cand.get("status") != "proven" or cand.get("workflow_id") != workflow_id:
|
||||
continue
|
||||
version = cand.get("version")
|
||||
if not isinstance(version, int) or (only is not None and version != only) \
|
||||
or (below is not None and version >= below):
|
||||
continue
|
||||
if cand.get("caliber") != caliber_key(workflow_id, version):
|
||||
continue
|
||||
binding = dict(cand.get("binding") or {})
|
||||
key = (version, _key(binding))
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
out.append(Use(kind="answer", run_id=run_id, version=version, binding=binding, at=at,
|
||||
query=query.get("artifact"), node_id=cand.get("node_id")))
|
||||
await _chat_owners(session, [u for u in out])
|
||||
return out, truncated
|
||||
|
||||
|
||||
def _key(binding: Mapping[str, Any]) -> str:
|
||||
from app.apps.resolve import input_key
|
||||
|
||||
return input_key(dict(binding))
|
||||
|
||||
|
||||
async def _chat_owners(session: AsyncSession, found: list[Use]) -> None:
|
||||
"""问数回答的轮次和主人：运行 origin 里封存的对话、轮次（origin_ref），老运行退回 conversation_turns.run_id；主人是
|
||||
对话的 created_by，没有的是运行的发起人。"""
|
||||
from app.db.models import Conversation, ConversationTurn
|
||||
|
||||
ids = sorted({u.run_id for u in found})
|
||||
if not ids:
|
||||
return
|
||||
runs = {rid: (origin, started) for rid, origin, started in (await session.execute(
|
||||
select(Run.id, Run.origin, Run.started_by).where(Run.id.in_(ids)))).all()}
|
||||
turns: dict[str, tuple[str, str]] = {}
|
||||
for rid, (origin, _) in runs.items():
|
||||
if isinstance(origin, dict) and origin.get("conversation_id") and origin.get("turn_id"):
|
||||
turns[rid] = (str(origin["conversation_id"]), str(origin["turn_id"]))
|
||||
missing = [rid for rid in ids if rid not in turns]
|
||||
if missing:
|
||||
for turn_id, conv, rid in (await session.execute(
|
||||
select(ConversationTurn.id, ConversationTurn.conversation_id, ConversationTurn.run_id)
|
||||
.where(ConversationTurn.run_id.in_(missing)).order_by(ConversationTurn.created_at))).all():
|
||||
turns.setdefault(rid, (conv, turn_id))
|
||||
convs = sorted({c for c, _ in turns.values()})
|
||||
owners = dict((await session.execute(select(Conversation.id, Conversation.created_by)
|
||||
.where(Conversation.id.in_(convs)))).all()) if convs else {}
|
||||
for use in found:
|
||||
if use.run_id in turns:
|
||||
use.conversation_id, use.turn_id = turns[use.run_id]
|
||||
owner = owners.get(use.conversation_id) if use.conversation_id else None
|
||||
use.add_owner("user", owner or runs.get(use.run_id, (None, None))[1])
|
||||
|
||||
|
||||
async def _app_managers(session: AsyncSession, app_ids: Iterable[str]) -> dict[str, list[tuple[str, str]]]:
|
||||
"""应用的管理者：授权里的 manage（人或组）；没有授权管理者的是建应用的人。"""
|
||||
from app.db.models_apps import App, AppGrant
|
||||
|
||||
ids = sorted({a for a in app_ids if a})
|
||||
if not ids:
|
||||
return {}
|
||||
out: dict[str, list[tuple[str, str]]] = {a: [] for a in ids}
|
||||
for app_id, kind, name in (await session.execute(
|
||||
select(AppGrant.app_id, AppGrant.principal_kind, AppGrant.principal)
|
||||
.where(AppGrant.app_id.in_(ids), AppGrant.role == "manage"))).all():
|
||||
if kind in ("user", "group") and name and (kind, name) not in out[app_id]:
|
||||
out[app_id].append((kind, name))
|
||||
for app_id, created_by in (await session.execute(select(App.id, App.created_by).where(App.id.in_(ids)))).all():
|
||||
if not out[app_id] and created_by:
|
||||
out[app_id].append(("user", created_by))
|
||||
return out
|
||||
|
||||
|
||||
async def _formal(session: AsyncSession, clause: Any, *, kind_of: Any) -> tuple[list[Use], bool]:
|
||||
"""满足 clause 的已封存、成功的正式运行（新的在前，最多 MAX_FORMAL 次）→ 用法。定时任务、应用各只留最近 PERIODS 期。
|
||||
kind_of(run) 给出这次运行算哪一种（schedule / app / formal / pinned）和旧版本。"""
|
||||
from app.db.models import Schedule
|
||||
from app.engine.origin import counted_runs
|
||||
|
||||
rows = list((await session.execute(
|
||||
select(Run).where(clause, Run.run_class == "formal", Run.status == "succeeded",
|
||||
Run.manifest_hash.is_not(None), counted_runs())
|
||||
.order_by(Run.finished_at.desc()).limit(MAX_FORMAL + 1))).scalars())
|
||||
truncated = len(rows) > MAX_FORMAL
|
||||
per: dict[tuple[str, str], int] = {}
|
||||
out: list[Use] = []
|
||||
for run in rows[:MAX_FORMAL]:
|
||||
kind, version, via = kind_of(run)
|
||||
group = ("schedule", run.schedule_id) if run.schedule_id else ("app", run.app_id) if run.app_id else None
|
||||
if group is not None:
|
||||
per[group] = per.get(group, 0) + 1
|
||||
if per[group] > PERIODS:
|
||||
continue
|
||||
use = Use(kind=kind, run_id=run.id, version=version, binding=dict(run.input or {}), at=run.finished_at,
|
||||
schedule_id=run.schedule_id, app_id=run.app_id)
|
||||
if via is not None:
|
||||
use.via_workflow, use.via_version = via
|
||||
out.append(use)
|
||||
schedules = sorted({u.schedule_id for u in out if u.schedule_id})
|
||||
makers = dict((await session.execute(select(Schedule.id, Schedule.created_by)
|
||||
.where(Schedule.id.in_(schedules)))).all()) if schedules else {}
|
||||
managers = await _app_managers(session, [u.app_id for u in out if u.app_id])
|
||||
starters = dict((await session.execute(select(Run.id, Run.started_by).where(
|
||||
Run.id.in_([u.run_id for u in out])))).all()) if out else {}
|
||||
for use in out:
|
||||
if use.schedule_id:
|
||||
use.add_owner("user", makers.get(use.schedule_id) or starters.get(use.run_id))
|
||||
elif use.app_id:
|
||||
for owner in managers.get(use.app_id) or []:
|
||||
use.add_owner(*owner)
|
||||
else:
|
||||
use.add_owner("user", starters.get(use.run_id))
|
||||
return out, truncated
|
||||
|
||||
|
||||
def _pins(graph: Any, workflow_id: str) -> set[int]:
|
||||
"""一张图里固定了 workflow_id 的哪几版（子工作流、口径卡的 caliber_from；同 apps/versions.pinned_upstream 的认法，
|
||||
直接读 JSON，不为了这个把整张图解析成 GraphSpec）。"""
|
||||
out: set[int] = set()
|
||||
for node in (graph or {}).get("nodes") or [] if isinstance(graph, dict) else []:
|
||||
if not isinstance(node, dict):
|
||||
continue
|
||||
cfg = (node.get("data") or {}).get("config") if isinstance(node.get("data"), dict) else None
|
||||
if not isinstance(cfg, dict):
|
||||
continue
|
||||
if node.get("type") == "subgraph":
|
||||
wid, version = cfg.get("workflow_id"), cfg.get("workflow_version")
|
||||
elif node.get("type") == "metrics" and isinstance(cfg.get("caliber_from"), dict):
|
||||
wid, version = cfg["caliber_from"].get("workflow_id"), cfg["caliber_from"].get("workflow_version")
|
||||
else:
|
||||
continue
|
||||
if wid == workflow_id and str(version or "").isdigit():
|
||||
out.add(int(version))
|
||||
return out
|
||||
|
||||
|
||||
async def pinned_versions(session: AsyncSession, workflow_id: str) -> dict[tuple[str, int], set[int]]:
|
||||
"""别的工作流的已发布版本里固定了 workflow_id 的：{(工作流, 版本): 固定的 W 的版本}。"""
|
||||
rows = (await session.execute(
|
||||
select(WorkflowVersion.workflow_id, WorkflowVersion.version, WorkflowVersion.graph)
|
||||
.where(WorkflowVersion.workflow_id != workflow_id, WorkflowVersion.level.is_not(None)))).all()
|
||||
return {(wid, int(v)): pins for wid, v, graph in rows if (pins := _pins(graph, workflow_id))}
|
||||
|
||||
|
||||
async def find_uses(session: AsyncSession, workflow_id: str, *, below: int | None = None,
|
||||
only: int | None = None) -> Found:
|
||||
"""W 的旧版本被谁用过（见模块说明）。below：只看版本号小于它的（召回：新版本）；only：只看这一版（版本用量）。"""
|
||||
from app.apps.versions import pinning_apps
|
||||
|
||||
def version_ok(v: Any) -> bool:
|
||||
return isinstance(v, int) and (only is None or v == only) and (below is None or v < below)
|
||||
|
||||
answers, cut_a = await _answers(session, workflow_id, below, only)
|
||||
own = Run.workflow_id == workflow_id
|
||||
if only is not None:
|
||||
own = and_(own, Run.version == only)
|
||||
elif below is not None:
|
||||
own = and_(own, Run.version < below)
|
||||
formal, cut_f = await _formal(session, own, kind_of=lambda r: (
|
||||
"schedule" if r.schedule_id else "app" if r.app_id else "formal", int(r.version), None))
|
||||
pins = await pinned_versions(session, workflow_id)
|
||||
pinned: list[Use] = []
|
||||
cut_p = False
|
||||
targets = {key: sorted(v for v in versions if version_ok(v)) for key, versions in pins.items()}
|
||||
targets = {key: vs for key, vs in targets.items() if vs}
|
||||
if targets:
|
||||
clause = or_(*(and_(Run.workflow_id == wid, Run.version == v) for (wid, v) in targets))
|
||||
pinned, cut_p = await _formal(session, clause, kind_of=lambda r: (
|
||||
"pinned", targets[(r.workflow_id, int(r.version))][-1], (r.workflow_id, int(r.version))))
|
||||
uses = [*answers, *formal, *pinned]
|
||||
await _exporters(session, uses)
|
||||
apps: dict[int, list[dict[str, Any]]] = {}
|
||||
for item in await pinning_apps(session, workflow_id):
|
||||
if version_ok(item.get("version")):
|
||||
apps.setdefault(int(item["version"]), []).append(item)
|
||||
return Found(uses=uses, apps=apps, truncated=cut_a or cut_f or cut_p)
|
||||
|
||||
|
||||
async def _exporters(session: AsyncSession, uses: list[Use]) -> None:
|
||||
"""这些运行上的报告导出审计：导出人进答案的主人。"""
|
||||
from app.db.models import AuditEvent
|
||||
|
||||
ids = sorted({u.run_id for u in uses})
|
||||
if not ids:
|
||||
return
|
||||
by: dict[str, list[str]] = {}
|
||||
for rid, actor in (await session.execute(select(AuditEvent.subject_id, AuditEvent.actor).where(
|
||||
AuditEvent.action == EXPORTED, AuditEvent.subject_kind == "run",
|
||||
AuditEvent.subject_id.in_(ids)).order_by(AuditEvent.at))).all():
|
||||
if actor and actor not in by.setdefault(rid, []):
|
||||
by[rid].append(actor)
|
||||
for use in uses:
|
||||
use.exporters = list(by.get(use.run_id) or [])
|
||||
for actor in use.exporters:
|
||||
use.add_owner("user", actor)
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 版本用量（版本历史、发布弹窗）
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
async def exports_count(session: AsyncSession, run_ids: Iterable[str]) -> int:
|
||||
from app.db.models import AuditEvent
|
||||
|
||||
ids = sorted(set(run_ids))
|
||||
if not ids:
|
||||
return 0
|
||||
return int(await session.scalar(select(func.count()).select_from(AuditEvent).where(
|
||||
AuditEvent.action == EXPORTED, AuditEvent.subject_kind == "run", AuditEvent.subject_id.in_(ids))) or 0)
|
||||
|
||||
|
||||
def summarize(found: Found) -> dict[str, Any]:
|
||||
"""用法的计数（不含参数和主人）：问数回答几次（运行数）、导出几次、定时任务几个、应用几个、各种正式运行几次。"""
|
||||
answers = {u.run_id for u in found.uses if u.kind == "answer"}
|
||||
schedules = {u.schedule_id for u in found.uses if u.schedule_id}
|
||||
apps = {u.app_id for u in found.uses if u.app_id} | {a["app_id"] for items in found.apps.values() for a in items}
|
||||
return {"answers": len(answers), "schedules": len(schedules), "apps": len(apps),
|
||||
"formal": len({u.run_id for u in found.uses if u.kind in ("formal", "schedule", "app")}),
|
||||
"pinned": len({u.run_id for u in found.uses if u.kind == "pinned"}),
|
||||
"exports": sum(len(u.exporters) for u in {u.run_id: u for u in found.uses}.values()),
|
||||
"truncated": found.truncated}
|
||||
|
||||
|
||||
async def version_usage(session: AsyncSession, workflow_id: str, version: int) -> dict[str, Any]:
|
||||
"""一个版本的用法计数（账本接口 GET /api/ledger/workflow-versions/{wf}/{v} 加的那几项）：问数回答、导出、召回。
|
||||
|
||||
召回分两头：这一版作为新版本召回过几组（recall_runs：以它为新版本发起过的召回运行；changed：其中数变了的）、作为旧版本
|
||||
被召回过几组（recalled_from）。"""
|
||||
from app.ledger import recall
|
||||
|
||||
found = await find_uses(session, workflow_id, only=version)
|
||||
out = summarize(found)
|
||||
out["exports"] = await exports_count(session, [u.run_id for u in found.uses])
|
||||
out["recall"] = await recall.version_recall_counts(session, workflow_id, version)
|
||||
return out
|
||||
|
||||
|
||||
async def workflow_name(session: AsyncSession, workflow_id: str) -> str:
|
||||
return str(await session.scalar(select(Workflow.name).where(Workflow.id == workflow_id)) or "")
|
||||
|
||||
|
||||
__all__ = ["EXPORTED", "Found", "KINDS", "KIND_TEXT", "MAX_ANSWERS", "MAX_FORMAL", "PERIODS", "Use", "exports_count",
|
||||
"find_uses", "pinned_versions", "summarize", "version_usage", "workflow_name"]
|
||||
@@ -35,7 +35,9 @@ from typing import Any
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.db.base import SessionLocal, utcnow
|
||||
from app.db.models import App, AppVersion, NotifyChannel, NotifyDelivery, Run, Setting, Task, TaskAssignment, User
|
||||
from app.db.models import (
|
||||
App, AppVersion, NotifyChannel, NotifyDelivery, Run, Setting, Task, TaskAssignment, User, Workflow,
|
||||
)
|
||||
from app.notify.core import Message, Outcome, host_of
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -69,8 +71,9 @@ EVENTS: dict[str, Event] = {
|
||||
# ---- 本轮登记的（总方案 X6）：scopes 为空的还没有推送；实现推送的那一块只改自己这一行的 scopes，消息照模块说明的规矩 ----
|
||||
# 数据源有要处理的结构 / 取值变化（03 B3）：只推 severity 为 action 的，按源汇总；源名、各级别条数、链接
|
||||
"source.changed": Event("数据源有变化要处理", ()),
|
||||
# 口径升版后召回重算，数真变了（02 第 7 项）：答案的主人有 recall 待办；口径名、版本、链接，不放新旧值和差值
|
||||
"ledger.recall": Event("口径升版，答案的数变了", ()),
|
||||
# 口径升版后召回重算，数真变了（02 第 7 项，notify_recall）：答案的主人有 recall 待办；推全局渠道和答案主人自己的个人
|
||||
# 渠道（召回运行不归哪个应用）。口径名、版本、几个答案、链接，不放新旧值和差值
|
||||
"ledger.recall": Event("口径升版，答案的数变了", ("global", "user")),
|
||||
# 定时评测有题从通过变成不通过 / 评测没跑完（01 第 15 项）：题集名、通过数、退步题数、链接；不放期望值和答案里的数。
|
||||
# 只推全局渠道：评测是搭建者们共用的一把尺，不归哪个人（notify_eval）。只推定时发起的（手动发起的那次人自己在看），
|
||||
# 叫法就照实带「定时」；没评完包括失败、到了上限、止损停下，所以不叫「失败」
|
||||
@@ -103,7 +106,7 @@ _TASK_HEAD = {"task.opened": "新的待办", "task.reassigned": "待办已改派
|
||||
"task.reminded": "待办催办"}
|
||||
_KIND_TEXT = {"approve": "审批", "input": "补充输入", "edit": "编辑稿件", "tool": "工具调用审批", "code": "代码执行审批",
|
||||
"dual_check": "双盲复算裁决", "action_confirm": "确认写入外部系统", "reconcile": "核实写入结果",
|
||||
"signal": "等外部系统回调", "review": "审核个人应用"}
|
||||
"signal": "等外部系统回调", "review": "审核个人应用", "recall": "口径召回"}
|
||||
_ENV_TEXT = {"prod": "生产环境", "test": "测试环境", "simulated": "演练"}
|
||||
|
||||
_background: set[asyncio.Task[Any]] = set()
|
||||
@@ -302,8 +305,9 @@ async def ensure_assignee_mail(users: list[str]) -> int:
|
||||
return made
|
||||
|
||||
|
||||
async def notify_task(task_id: str, event: str, *, users: list[str]) -> int:
|
||||
async def notify_task(task_id: str, event: str, *, users: list[str], exclude: frozenset[str] = frozenset()) -> int:
|
||||
"""待办的通知（task.opened / reassigned / escalated / reminded）。users 是这次要通知到的人（派给的用户名）。返回推了几个渠道。
|
||||
exclude 是这一次已经推过别的事件的渠道（口径召回先推了 ledger.recall 的，notify_recall），同一个渠道只推一次。
|
||||
|
||||
管理员开了「默认发邮件给待办处理人」时，先给还没有个人邮件渠道的处理人建一个（ensure_assignee_mail），这一条就推得到。"""
|
||||
try:
|
||||
@@ -315,7 +319,7 @@ async def notify_task(task_id: str, event: str, *, users: list[str]) -> int:
|
||||
if task is None:
|
||||
return 0
|
||||
run = await session.get(Run, task.run_id)
|
||||
channels = await _channels(session, event, run, users)
|
||||
channels = [c for c in await _channels(session, event, run, users) if c.id not in exclude]
|
||||
if not channels:
|
||||
return 0
|
||||
base = await _base_url(session)
|
||||
@@ -338,6 +342,47 @@ async def notify_task(task_id: str, event: str, *, users: list[str]) -> int:
|
||||
task_id=task_id)
|
||||
|
||||
|
||||
async def notify_recall(task_id: str, users: list[str]) -> int:
|
||||
"""口径召回的待办开出来了（app/ledger/recall）：口径升版、数真变了的那一种推 ledger.recall——订阅了它的全局渠道、答案主人
|
||||
自己的个人渠道；然后照别的待办推 task.opened（已经推过 ledger.recall 的渠道不再推，同一个渠道只推一次）。目录、表结构变了
|
||||
的那两种只推 task.opened。
|
||||
|
||||
消息照模块说明的规矩（X6）：口径名（搭建时起的工作流名）、新旧版本号、这条待办列了几个答案、召回编号、去处理的链接；不放
|
||||
旧口径的值、新口径的值、差值和原答案的值。返回推了几个渠道。"""
|
||||
from app.ledger import recall
|
||||
|
||||
try:
|
||||
await ensure_assignee_mail(users)
|
||||
except Exception: # noqa: BLE001 - 自动建渠道出错不挡这条待办的通知
|
||||
logger.exception("给待办 %s 的处理人自动建邮件渠道失败", task_id)
|
||||
event = "ledger.recall"
|
||||
async with SessionLocal() as session:
|
||||
task = await session.get(Task, task_id)
|
||||
if task is None:
|
||||
return 0
|
||||
run = await session.get(Run, task.run_id)
|
||||
reason = (task.correlation_key or "").split(":", 1)[0]
|
||||
channels = await _channels(session, event, run, users) if reason == "caliber" else []
|
||||
origin = run.origin if run is not None and isinstance(run.origin, dict) else {}
|
||||
name = str(await session.scalar(select(Workflow.name).where(Workflow.id == origin.get("workflow_id")))
|
||||
or "") if origin.get("workflow_id") else ""
|
||||
base = await _base_url(session) if channels else None
|
||||
pushed = 0
|
||||
if channels:
|
||||
answers = len(recall._uses_of(origin, recall._owner_of_key(task.correlation_key or "")))
|
||||
lines = [("状态", "待处理"), ("口径", f"「{name[:80]}」"),
|
||||
("版本", f"v{origin.get('from_version')} → v{origin.get('to_version')}"), ("答案", f"{answers} 个"),
|
||||
("召回", f"#{task.run_id[:6]}")]
|
||||
message = {"event": event, "status": "pending", "ok": False,
|
||||
"title": recall.CHANGED_TITLE.format(name=name[:80] or "口径", version=origin.get("to_version")),
|
||||
"lines": lines, "links": [("去处理", f"{base}/apps/inbox/{task_id}")] if base else [],
|
||||
"subject": {"task_id": task_id, "run_id": task.run_id, "workflow_id": origin.get("workflow_id")}}
|
||||
pushed = await _push(channels, message, subject_kind="task", subject_id=task_id, event=event,
|
||||
run_id=task.run_id, task_id=task_id)
|
||||
return pushed + await notify_task(task_id, "task.opened", users=users,
|
||||
exclude=frozenset(c.id for c in channels))
|
||||
|
||||
|
||||
async def notify_run(run_id: str, event: str) -> int:
|
||||
"""应用实例结束、失败（run.finished / run.failed）。只推应用发起的运行。
|
||||
|
||||
|
||||
@@ -15,6 +15,9 @@
|
||||
时限，但调度不对它们做提醒、升级、超时驳回——试运行不该把待办推到真实的处理人、上级那里。
|
||||
- 个人应用提交审核（kind=review，N7）：没有 Approval、不恢复运行，挂在个人应用最近一次成功的刷新上；派给选定的搭建者
|
||||
或组。在待办里只能退回，同意要在共享副本上走发布单，发布完自动记为已发布（blueprints/personal.py）。
|
||||
- 口径召回（kind=recall，docs/design/next-round/02-口径账本.md 做法 7）：没有 Approval、不恢复运行。口径升版后数真变了的
|
||||
挂在召回运行上，派给答案的主人（开登录时每人一条，关登录时任何人）；动作「按新版重新出具」（走按口径出具同一套判法）
|
||||
和「知道了」。答案依据的目录、表结构变了的另两种只有「知道了」。开、收尾、内容都在 app/ledger/recall.py。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -163,6 +166,29 @@ async def open_review(session: AsyncSession, *, run: Run, app_id: str, blueprint
|
||||
return task
|
||||
|
||||
|
||||
#: 口径召回的待办（app/ledger/recall.py）。没有 Approval，按 correlation_key（来由:运行[:附加]:派给的种类:名字）认同一条
|
||||
RECALL_KIND = "recall"
|
||||
#: recall 待办的两种决定：按新版重新出具、知道了（目录、表结构那两种只有「知道了」）
|
||||
RECALL_DECISIONS = ("reissue", "ack")
|
||||
|
||||
|
||||
async def open_recall(session: AsyncSession, *, run: Run, title: str, owner: tuple[str, str],
|
||||
correlation_key: str) -> Task:
|
||||
"""开一条 recall 待办（挂在 run 上），派给 owner：人、组（在这一刻展开成成员，同改派）、任何人；来源记 recall。调用方
|
||||
提交，提交后 announce。"""
|
||||
task = Task(run_id=run.id, node_id=None, app_id=None, app_version=None, kind=RECALL_KIND, title=title[:300],
|
||||
policy={"kind": "any"}, correlation_key=correlation_key[:200])
|
||||
session.add(task)
|
||||
await session.flush()
|
||||
kind, name = owner
|
||||
targets = await _expand(session, task, kind, name, RECALL_KIND) if kind in ("user", "group") and name \
|
||||
else [("anyone", "", RECALL_KIND)]
|
||||
for k, n, source in dict.fromkeys(targets):
|
||||
session.add(TaskAssignment(task_id=task.id, principal_kind=k, principal=n, source=source))
|
||||
await session.flush()
|
||||
return task
|
||||
|
||||
|
||||
async def review_handler(session: AsyncSession, task: Task, principal: Any) -> bool:
|
||||
"""这个人是不是这条审核待办的处理人（派给他或他的组、没撤回，同 handles）：只有他们能取个人蓝图的共享副本。"""
|
||||
return handles((await _assignments(session, [task.id])).get(task.id, []), principal)
|
||||
@@ -327,8 +353,15 @@ async def announce(task_id: str, event: EventType = EventType.TASK_OPENED) -> No
|
||||
return
|
||||
from app.notify.service import spawn
|
||||
|
||||
users = [a.principal for a in rows.get(task_id, []) if a.principal_kind == "user"]
|
||||
if task.kind == RECALL_KIND:
|
||||
# 口径召回：口径升版、数变了的推 ledger.recall，订阅了 task.opened 的渠道另推一条（同一个渠道只推一次）
|
||||
from app.notify.service import notify_recall
|
||||
|
||||
spawn(notify_recall(task_id, users))
|
||||
return
|
||||
# 推送放到后台：接收端挂住时超时加重试要几十秒，不能让运行停在「执行中」等它
|
||||
spawn(_notify(task_id, "task.opened", [a.principal for a in rows.get(task_id, []) if a.principal_kind == "user"]))
|
||||
spawn(_notify(task_id, "task.opened", users))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -542,6 +575,8 @@ async def respond(task_id: str, principal: Any, *, decision: Any, values: dict[s
|
||||
return await _respond_reconcile(session, task, principal, decision, values, note)
|
||||
if task.kind == REVIEW_KIND:
|
||||
return await _respond_review(session, task, principal, decision, note)
|
||||
if task.kind == RECALL_KIND:
|
||||
return await _respond_recall(session, task, principal, decision, note)
|
||||
if task.status != "pending":
|
||||
raise TaskRefused(_closed(task.status))
|
||||
if inquiry_busy(task.run_id):
|
||||
@@ -669,6 +704,54 @@ async def _respond_review(session: AsyncSession, task: Task, principal: Any, dec
|
||||
return {"task": await detail(task_id, principal), "resolved": True, "run": await _run(run_id)}
|
||||
|
||||
|
||||
async def _respond_recall(session: AsyncSession, task: Task, principal: Any, decision: Any,
|
||||
note: str | None) -> dict[str, Any]:
|
||||
"""recall 待办：「按新版重新出具」（app/ledger/recall.reissue，走按口径出具同一套判法；不能出具照实拒绝，待办留着）或
|
||||
「知道了」。处理完记成已处理（条件 UPDATE，同 close_review），结论和出具的那次正式运行写进 outcome，另记一条审计：运行
|
||||
已经封存，事件记不进去。"""
|
||||
if task.status != "pending":
|
||||
raise TaskRefused(_closed(task.status))
|
||||
reason = (task.correlation_key or "").split(":", 1)[0]
|
||||
allowed = RECALL_DECISIONS if reason == "caliber" else ("ack",)
|
||||
if decision not in allowed:
|
||||
raise TaskRefused("这条待办只能「按新版重新出具」或「知道了」" if reason == "caliber"
|
||||
else "这条待办只能「知道了」", status_code=422, code="task_decision_invalid")
|
||||
rows = (await _assignments(session, [task.id])).get(task.id, [])
|
||||
mine_id = _pick(task, rows, principal).id
|
||||
task_id, run_id = task.id, task.run_id
|
||||
await session.rollback()
|
||||
outcome: dict[str, Any] = {"decision": decision, "note": (note or "").strip()[:2000]}
|
||||
issued: Any = None
|
||||
if decision == "reissue":
|
||||
from app.ledger import recall
|
||||
|
||||
try:
|
||||
issued = await recall.reissue(task_id, principal)
|
||||
except recall.RecallRefused as e:
|
||||
raise TaskRefused(e.message, status_code=e.status_code, code=e.code) from e
|
||||
outcome["issued_run"] = issued.id
|
||||
async with SessionLocal() as session:
|
||||
now = utcnow()
|
||||
claimed = await session.execute(
|
||||
update(Task).where(Task.id == task_id, Task.status == "pending")
|
||||
.values(status="resolved", resolved_by=principal.username, resolved_at=now, outcome=outcome))
|
||||
if claimed.rowcount != 1:
|
||||
await session.rollback()
|
||||
raise TaskRefused(_closed("resolved"))
|
||||
await session.execute(update(TaskAssignment).where(TaskAssignment.id == mine_id).values(
|
||||
status="responded", responded_by=principal.username, responded_at=now, verified=bool(principal.verified),
|
||||
response={"decision": decision, "note": outcome["note"]}))
|
||||
await session.execute(update(TaskAssignment).where(TaskAssignment.task_id == task_id,
|
||||
TaskAssignment.status == "pending")
|
||||
.values(status="withdrawn"))
|
||||
session.add(AuditEvent(actor=principal.username, verified=bool(principal.verified),
|
||||
action=f"ledger.recall.{decision}", subject_kind="task", subject_id=task_id,
|
||||
data={"run_id": run_id, **({"issued_run": issued.id} if issued is not None else {}),
|
||||
**key_trace(principal)}))
|
||||
await session.commit()
|
||||
return {"task": await detail(task_id, principal), "resolved": True, "run": await _run(run_id)}
|
||||
|
||||
|
||||
async def close_reconcile(session: AsyncSession, action_id: str, principal: Any, outcome: str,
|
||||
note: str | None) -> str | None:
|
||||
"""核实之后把这个动作的核实待办收掉（条件 UPDATE，和 tick、别人的回复抢同一行）。返回待办 id，没有返回 None。"""
|
||||
@@ -1102,6 +1185,10 @@ async def _describe(session: AsyncSession, tasks: list[Task], principal: Any) ->
|
||||
from app.blueprints.personal import review_payload
|
||||
|
||||
payload = await review_payload(session, task)
|
||||
elif task.kind == RECALL_KIND:
|
||||
from app.ledger.recall import task_payload
|
||||
|
||||
payload = await task_payload(session, task, principal)
|
||||
run = runs.get(task.run_id)
|
||||
out.append({
|
||||
"id": task.id, "approval_id": task.approval_id, "run_id": task.run_id, "node_id": task.node_id,
|
||||
|
||||
@@ -304,7 +304,9 @@ async def test_version_detail_has_form_and_usage(client):
|
||||
assert detail["usage"]["effects"] == 1 and detail["usage"]["formal_runs"] == 1
|
||||
assert [a["name"] for a in detail["usage"]["apps"]] == ["周报看板"]
|
||||
ledger = (await client.get(f"/api/ledger/workflow-versions/{wf}/1")).json()
|
||||
assert ledger == detail["usage"]
|
||||
# 账本接口在同一份之上另加口径账本的用量（问数回答、导出、定时任务、召回，next-round 02 做法 8）
|
||||
assert {k: ledger[k] for k in detail["usage"]} == detail["usage"]
|
||||
assert ledger["answers"] == 0 and ledger["recall"] == {"groups": 0, "changed": 0, "recalled_from": 0}
|
||||
effects = (await client.get("/api/ledger/effects", params={"workflow_id": wf})).json()["items"]
|
||||
assert [(e["capability"], e["state"], e["workflow_version"]) for e in effects] == [("erp_pay", "confirmed", 1)]
|
||||
|
||||
|
||||
@@ -237,7 +237,8 @@ async def test_stats_count_what_the_fixtures_did(client, monkeypatch):
|
||||
assert body["e3"]["chat_runs"] == 0 and body["e3"]["threshold"] == {"runs": 3, "conversations": 2}
|
||||
usage = {u["caliber"]: u for u in body["e4"]["usage"]}
|
||||
assert usage[f"{wf}@1"]["proven_runs"] == 1
|
||||
assert body["e4"]["per_publish"]["available"] is False
|
||||
# 召回（第 4 波）：这一段时间里没有发布新版本，没有召回
|
||||
assert body["e4"]["per_publish"]["available"] is True and body["e4"]["per_publish"]["groups"] == 0
|
||||
|
||||
|
||||
async def test_stats_take_the_run_ids_the_caller_names(monkeypatch):
|
||||
|
||||
@@ -0,0 +1,563 @@
|
||||
"""口径召回（next-round 02 做法 7，拍板 L6、L7；app/ledger/uses.py、recall.py）。
|
||||
|
||||
真的跑：虚构店铺库、mock 模型写报告。受管口径「周销售」v1 只算已支付的，问数据问过几周、正式运行跑过；把它发布成 v2
|
||||
（改一个筛选条件）之后：
|
||||
- 计划：按（旧版本、新版本、参数）分组去重——问数回答绑定出的参数和正式运行的输入补齐缺省值之后是同一组；上限 recall_limit
|
||||
之内自动发起，超出的写「另有 M 组未重算」，手动接着算；再排一次、服务重启之后重算计划，算过的不再发起（幂等）；
|
||||
- 召回运行：派生图里新旧两版的取数和口径卡各一份（前缀不撞），召回卡里算两版之差；同一份数据上两版各算一次；
|
||||
- 收尾：数变了才开 recall 待办（挂在召回运行上，旧值、新值、差值、原答案当时的值都给），推 ledger.recall，消息里没有数；
|
||||
数没变的在原答案上记 ledger.recalled（已按 vN 核对，不影响），不开待办；
|
||||
- 不能重算的照实记：新版入口要的参数这一组里没有、图里有人工节点（有副作用）；
|
||||
- 待办「按新版重新出具」走按口径出具同一套判法、「知道了」；
|
||||
- 目录里影响数字的一项被改了：冻结目录里看得到它的答案标「依据有变」，按新目录多出错误级问题的才开待办；
|
||||
- 表结构变了：用过这张表的答案标「依据的表结构已变」，口径试算算不出来才开待办；
|
||||
- E4：每次发布召回了几组、其中数变了的比例。
|
||||
数据都是编的。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.core.events import EventType
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import Run, RunEvent, Setting, Task, TaskAssignment
|
||||
from app.engine.runner import run_manager
|
||||
from app.ledger import match as ledger_match
|
||||
from app.ledger import recall
|
||||
from app.main import app
|
||||
from tests import ledger_samples as S
|
||||
|
||||
#: v2：已支付和已退款的都算（召回要算出差值的那种改法）
|
||||
WITH_REFUNDED = ("SELECT SUM(amount) AS gmv FROM orders WHERE status IN ('paid', 'refunded') "
|
||||
"AND paid_at >= '{{ input.start }}' AND paid_at < '{{ input.end }}'")
|
||||
WEEK1 = {"start": "2026-09-01", "end": "2026-09-08"}
|
||||
|
||||
|
||||
def week_sql(start: str, end: str) -> str:
|
||||
return (f"select sum(o.amount) as 销售额 from orders o where o.paid_at >= '{start}' and o.status = 'paid' "
|
||||
f"and o.paid_at < '{end}'")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def engine_up(tmp_path):
|
||||
await run_manager.setup()
|
||||
path = tmp_path / "shop.db"
|
||||
await S.fresh_source(path)
|
||||
await ledger_settings(auto_issue=False, issue_daily_limit=100_000)
|
||||
yield path
|
||||
await ledger_match.drain()
|
||||
await recall.drain()
|
||||
await run_manager.shutdown()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
async def ledger_settings(**values) -> None:
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(Setting, "ledger")
|
||||
if row is not None:
|
||||
await session.delete(row)
|
||||
session.add(Setting(key="ledger", value=values))
|
||||
await session.commit()
|
||||
|
||||
|
||||
async def answer(monkeypatch, start: str = "2026-09-01", end: str = "2026-09-08") -> Run:
|
||||
"""问数据问一周的销售额（独立写出、和口径等价的 SQL）。"""
|
||||
return await S.explore(monkeypatch, week_sql(start, end), text="本周销售额 [[v:Q1.r0.销售额]]。", origin="chat")
|
||||
|
||||
|
||||
async def publish_v2(wf: str, sql: str = WITH_REFUNDED, *, graph=None) -> None:
|
||||
await S.publish(graph=graph or S.caliber_graph(sql), version=2, workflow_id=wf)
|
||||
|
||||
|
||||
async def run_recall(wf: str, version: int = 2, *, batch: str = "auto") -> list[Run]:
|
||||
"""按计划发起一批、等召回运行跑完、等收尾做完。"""
|
||||
ids = await recall.launch(wf, version, batch=batch)
|
||||
done = [await S.wait(rid) for rid in ids]
|
||||
await recall.drain()
|
||||
return done
|
||||
|
||||
|
||||
async def events(run_id: str, kind: EventType) -> list[dict]:
|
||||
async with SessionLocal() as session:
|
||||
return [d for d in (await session.execute(select(RunEvent.data).where(
|
||||
RunEvent.run_id == run_id, RunEvent.type == kind).order_by(RunEvent.seq))).scalars()]
|
||||
|
||||
|
||||
async def recall_tasks(run_id: str) -> list[Task]:
|
||||
async with SessionLocal() as session:
|
||||
return list((await session.execute(select(Task).where(Task.run_id == run_id, Task.kind == "recall"))).scalars())
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 计划：分组、去重、上限、幂等
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_plan_groups_answers_and_formal_runs_by_their_binding(monkeypatch):
|
||||
wf = await S.publish()
|
||||
a1 = await answer(monkeypatch)
|
||||
a2 = await answer(monkeypatch) # 同一周再问一次：同一组
|
||||
a3 = await answer(monkeypatch, "2026-09-08", "2026-09-15")
|
||||
await publish_v2(wf)
|
||||
async with SessionLocal() as session:
|
||||
plan = await recall.build_plan(session, wf, 2)
|
||||
assert plan.problem is None and [p.old.node_id for p in plan.pairs[1]] == ["card"]
|
||||
groups = {tuple(sorted(g.binding.items())): g for g in plan.groups}
|
||||
week1 = groups[(("end", "2026-09-08"), ("start", "2026-09-01"), ("store", "all"))]
|
||||
assert sorted(u.run_id for u in week1.uses) == sorted([a1.id, a2.id]), "同一组参数去重成一组（缺省值补齐后比）"
|
||||
assert week1.old_version == 1 and week1.new_version == 2 and week1.problem is None
|
||||
assert week1.inputs == {"start": "2026-09-01", "end": "2026-09-08", "store": "all"}
|
||||
assert len(plan.groups) == 2 and len(plan.todo()) == 2
|
||||
assert {u.run_id for g in plan.groups for u in g.uses} == {a1.id, a2.id, a3.id}
|
||||
assert all(u.kind == "answer" and u.turn_id is None for g in plan.groups for u in g.uses)
|
||||
|
||||
|
||||
async def test_limit_continue_and_idempotent_replanning(monkeypatch):
|
||||
await ledger_settings(auto_issue=False, issue_daily_limit=100_000, recall_limit=1)
|
||||
wf = await S.publish()
|
||||
await answer(monkeypatch)
|
||||
await answer(monkeypatch, "2026-09-08", "2026-09-15")
|
||||
await publish_v2(wf)
|
||||
async with SessionLocal() as session:
|
||||
before = await recall.preview(session, wf, 2)
|
||||
assert before["groups"] == 2 and before["will_recompute"] == 1 and before["over_limit"] == 1
|
||||
assert before["versions"][0]["answers"] == 2 and before["caliber"] is True
|
||||
first = await run_recall(wf)
|
||||
assert len(first) == 1 and first[0].origin["batch"] == "auto"
|
||||
# 再排一次（重复的发布钩子、服务重启后重算计划）：自动那一批已经发满，不再发起
|
||||
assert await recall.launch(wf, 2, batch="auto") == []
|
||||
async with SessionLocal() as session:
|
||||
view = await recall.batches(session, wf, 2)
|
||||
assert view["counts"]["pending"] == 1 and view["remaining"] == 1 and view["can_continue"] is True
|
||||
assert sum(1 for g in view["groups"] if g.get("run_id")) == 1
|
||||
# 手动接着算：发起剩下那一组；再接着算就没有了
|
||||
second = await run_recall(wf, batch="manual")
|
||||
assert len(second) == 1 and second[0].origin["batch"] == "manual"
|
||||
async with SessionLocal() as session:
|
||||
view = await recall.batches(session, wf, 2)
|
||||
assert view["remaining"] == 0 and view["can_continue"] is False
|
||||
assert await recall.launch(wf, 2, batch="manual") == []
|
||||
|
||||
|
||||
async def test_zero_limit_lists_but_does_not_recompute(monkeypatch):
|
||||
await ledger_settings(auto_issue=False, issue_daily_limit=100_000, recall_limit=0)
|
||||
wf = await S.publish()
|
||||
await answer(monkeypatch)
|
||||
await publish_v2(wf)
|
||||
assert await recall.launch(wf, 2, batch="auto") == []
|
||||
async with SessionLocal() as session:
|
||||
view = await recall.batches(session, wf, 2)
|
||||
assert view["counts"]["pending"] == 1 and view["batch_size"] == 50, "上限是 0 时手动接着算按缺省的组数"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 召回运行与收尾
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_changed_numbers_open_a_recall_task_with_all_three_values(client, monkeypatch):
|
||||
wf = await S.publish()
|
||||
a1 = await answer(monkeypatch)
|
||||
await publish_v2(wf)
|
||||
[done] = await run_recall(wf)
|
||||
assert done.status == "succeeded", done.error
|
||||
assert done.origin_kind == "recall" and done.run_class == "exploratory" and done.workflow_id is None
|
||||
ids = [n["id"] for n in done.graph["nodes"]]
|
||||
assert ids == ["derived_in", "old_fetch", "old_card", "new_fetch", "new_card", "recall", "derived_out"]
|
||||
nodes = {n["id"]: n for n in done.graph["nodes"]}
|
||||
assert nodes["old_card"]["data"]["config"]["caliber_from"] == {
|
||||
"workflow_id": wf, "workflow_version": 1, "node_id": "card", "nodes": {"fetch": "old_fetch"}}
|
||||
assert nodes["new_card"]["data"]["config"]["caliber_from"]["workflow_version"] == 2
|
||||
assert nodes["derived_out"]["data"]["config"]["contract"] == {"metrics_from": "recall"}
|
||||
assert not [n for n in done.graph["nodes"] if n["type"] in ("report", "llm", "agent")]
|
||||
|
||||
values = await recall.card_values(done.id)
|
||||
# v1 只算已支付：100 + 50.5 + 20；v2 加上这一周退款的 30 + 8
|
||||
assert values["old_0"]["value"] == 170.5 and values["new_0"]["value"] == 208.5
|
||||
assert values["diff_0"]["value"] == 38.0 and values["diff_0"]["rendered"] == "38.00元"
|
||||
|
||||
[settled] = [d for d in await events(done.id, EventType.LEDGER_RECALLED) if d.get("role") == "recall"]
|
||||
assert settled["changed"] is True and settled["answers"] == 1 and len(settled["tasks"]) == 1
|
||||
assert settled["post_seal"] is True
|
||||
assert await events(a1.id, EventType.LEDGER_RECALLED) == [], "数变了的不在原答案上记「不影响」"
|
||||
[task] = await recall_tasks(done.id)
|
||||
assert task.title == "口径「周销售」升到 v2，答案的数变了"
|
||||
res = await client.get(f"/api/tasks/{task.id}")
|
||||
assert res.status_code == 200, res.text
|
||||
payload = res.json()["payload"]
|
||||
assert payload["kind"] == "recall" and payload["reason"] == "caliber" and payload["to_version"] == 2
|
||||
[metric] = payload["metrics"]
|
||||
assert (metric["old"]["value"], metric["new"]["value"], metric["diff"]["value"]) == (170.5, 208.5, 38.0)
|
||||
assert metric["changed"] is True and metric["old"]["metric"] == "old_0" and metric["diff"]["metric"] == "diff_0"
|
||||
[ans] = payload["answers"]
|
||||
assert ans["run_id"] == a1.id and ans["kind"] == "answer"
|
||||
assert ans["original"][0]["value"] == 170.5 and ans["original"][0]["column"] == "销售额"
|
||||
assert [a["decision"] for a in payload["actions"]] == ["reissue", "ack"]
|
||||
assert payload["actions"][0]["available"] is True
|
||||
# 三个数各自点得开：召回卡上的差值的出处链追到两版的口径卡
|
||||
chain = await client.get(f"/api/runs/{done.id}/evidence/metrics/recall/diff_0")
|
||||
assert chain.status_code == 200, chain.text
|
||||
assert {s["node_id"] for s in chain.json()["chain"] if s["step"] == "input"} == {"old_card", "new_card"}
|
||||
|
||||
|
||||
async def test_unchanged_numbers_are_only_booked_on_the_answer(monkeypatch):
|
||||
wf = await S.publish()
|
||||
a1 = await answer(monkeypatch)
|
||||
# v2 只换了报告怎么写，口径的数据部分没变
|
||||
graph = S.caliber_graph()
|
||||
graph["nodes"][3]["data"]["config"]["instructions"] = "写周销售（新的写法）"
|
||||
await publish_v2(wf, graph=graph)
|
||||
[done] = await run_recall(wf)
|
||||
assert done.status == "succeeded", done.error
|
||||
assert await recall_tasks(done.id) == []
|
||||
[booked] = await events(a1.id, EventType.LEDGER_RECALLED)
|
||||
assert booked == {"role": "answer", "changed": False, "recall_run": done.id, "workflow_id": wf,
|
||||
"from_version": 1, "to_version": 2, "post_seal": True}
|
||||
# 再收一次尾不会重复记
|
||||
assert await recall.settle(done.id) is None
|
||||
assert len(await events(a1.id, EventType.LEDGER_RECALLED)) == 1
|
||||
|
||||
|
||||
async def test_incompatible_entry_and_side_effects_are_not_recomputed(monkeypatch):
|
||||
wf = await S.publish()
|
||||
await answer(monkeypatch)
|
||||
graph = S.caliber_graph(WITH_REFUNDED)
|
||||
graph["nodes"][0]["data"]["config"]["fields"].append({"name": "region", "required": True})
|
||||
await publish_v2(wf, graph=graph)
|
||||
async with SessionLocal() as session:
|
||||
[group] = (await recall.build_plan(session, wf, 2)).groups
|
||||
assert group.problem is not None and group.problem.startswith("新版 v2 的入口要的参数这一组里没有或不合规")
|
||||
assert await recall.launch(wf, 2) == []
|
||||
async with SessionLocal() as session:
|
||||
view = await recall.batches(session, wf, 2)
|
||||
assert view["counts"]["not_recomputable"] == 1 and view["groups"][0]["reason"] == group.problem
|
||||
assert view["can_continue"] is False
|
||||
|
||||
wf2 = await S.publish(name="周销售（要审批）")
|
||||
await answer(monkeypatch)
|
||||
human = S.caliber_graph(WITH_REFUNDED)
|
||||
human["nodes"].insert(3, S.node("ok", "human", mode="approve", title="确认"))
|
||||
await S.publish(graph=human, version=2, workflow_id=wf2)
|
||||
async with SessionLocal() as session:
|
||||
plan = await recall.build_plan(session, wf2, 2)
|
||||
assert plan.groups and all("人工节点" in (g.problem or "") for g in plan.groups)
|
||||
|
||||
|
||||
async def test_plan_from_formal_runs_and_exporters_as_owners(monkeypatch):
|
||||
"""正式运行也是用过旧版的答案；导出过报告的人也是答案的主人（关登录时待办派给任何人，主人照记在召回运行里）。"""
|
||||
from app.apps.audit import record
|
||||
from app.engine.launch import launch_run
|
||||
|
||||
wf = await S.publish()
|
||||
S.script(monkeypatch, "本周销售额 [[m:gmv]]。")
|
||||
async with SessionLocal() as session:
|
||||
formal = await launch_run(session, workflow_id=wf, run_class="formal", version=1, input_payload=WEEK1,
|
||||
actor="analyst")
|
||||
formal = await S.wait(formal.id)
|
||||
assert formal.status == "succeeded", formal.error
|
||||
async with SessionLocal() as session:
|
||||
await record(session, action="report.exported", subject_kind="run", subject_id=formal.id, actor="reader",
|
||||
data={})
|
||||
await session.commit()
|
||||
await publish_v2(wf)
|
||||
async with SessionLocal() as session:
|
||||
[group] = (await recall.build_plan(session, wf, 2)).groups
|
||||
usage = await recall.preview(session, wf, 2)
|
||||
[use] = group.uses
|
||||
assert use.kind == "formal" and use.exporters == ["reader"]
|
||||
assert ("user", "analyst") in use.owners and ("user", "reader") in use.owners
|
||||
assert usage["versions"][0]["exports"] == 1 and usage["versions"][0]["formal"] == 1
|
||||
[done] = await run_recall(wf)
|
||||
[task] = await recall_tasks(done.id)
|
||||
async with SessionLocal() as session:
|
||||
[assignment] = (await session.execute(select(TaskAssignment).where(TaskAssignment.task_id == task.id))).scalars()
|
||||
assert assignment.principal_kind == "anyone", "关登录时派给任何人"
|
||||
from app.api.deps import Principal
|
||||
from app.ledger.recall import task_payload
|
||||
|
||||
async with SessionLocal() as session:
|
||||
payload = await task_payload(session, await session.get(Task, task.id),
|
||||
Principal(username=None, role="admin", mode="off", via="off"))
|
||||
[ans] = payload["answers"]
|
||||
assert ans["kind"] == "formal" and ans["original"][0]["value"] == 170.5
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 待办的两个动作、通知
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_reissue_and_acknowledge(client, monkeypatch):
|
||||
wf = await S.publish()
|
||||
await answer(monkeypatch)
|
||||
await answer(monkeypatch, "2026-09-01", "2026-09-05") # 这几天也有退款：两组的数都变了
|
||||
await publish_v2(wf)
|
||||
runs = await run_recall(wf)
|
||||
tasks = [t for r in runs for t in await recall_tasks(r.id)]
|
||||
assert len(tasks) == 2
|
||||
S.script(monkeypatch, "本周销售额 [[m:gmv]]。")
|
||||
res = await client.post(f"/api/tasks/{tasks[0].id}/respond", json={"decision": "reissue"})
|
||||
assert res.status_code == 200, res.text
|
||||
body = res.json()["task"]
|
||||
assert body["status"] == "resolved" and body["outcome"]["decision"] == "reissue"
|
||||
async with SessionLocal() as session:
|
||||
issued = await session.get(Run, body["outcome"]["issued_run"])
|
||||
assert issued.origin_kind == "ledger_issue" and issued.run_class == "formal" and issued.version == 2
|
||||
assert issued.origin["recall_run"] == tasks[0].run_id and issued.origin["how"] == "recall"
|
||||
assert issued.input["start"] == "2026-09-01" and issued.input["end"] in ("2026-09-08", "2026-09-05")
|
||||
await S.wait(issued.id)
|
||||
bad = await client.post(f"/api/tasks/{tasks[1].id}/respond", json={"decision": "approve"})
|
||||
assert bad.status_code == 422
|
||||
ok = await client.post(f"/api/tasks/{tasks[1].id}/respond", json={"decision": "ack", "note": "知道了"})
|
||||
assert ok.status_code == 200 and ok.json()["task"]["outcome"]["decision"] == "ack"
|
||||
again = await client.post(f"/api/tasks/{tasks[1].id}/respond", json={"decision": "ack"})
|
||||
assert again.status_code == 409
|
||||
|
||||
|
||||
async def test_recall_notification_has_no_numbers(monkeypatch):
|
||||
from app.core.crypto import encrypt
|
||||
from app.db.models import NotifyChannel
|
||||
from app.notify import service
|
||||
|
||||
sent: list[dict] = []
|
||||
|
||||
async def capture(channel, message):
|
||||
sent.append({"channel": channel.name, **message})
|
||||
from app.notify.core import Outcome
|
||||
|
||||
return Outcome(True, 200, 1, None)
|
||||
|
||||
monkeypatch.setattr("app.notify.send", capture)
|
||||
async with SessionLocal() as session:
|
||||
row = NotifyChannel(name="召回-全局", scope="global", url=encrypt("https://hooks.example.com/recall-7d"),
|
||||
events=["ledger.recall", "task.opened"])
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
channel_id = row.id
|
||||
try:
|
||||
wf = await S.publish()
|
||||
await answer(monkeypatch)
|
||||
await publish_v2(wf)
|
||||
await run_recall(wf)
|
||||
await service.drain()
|
||||
finally:
|
||||
async with SessionLocal() as session:
|
||||
await session.delete(await session.get(NotifyChannel, channel_id))
|
||||
await session.commit()
|
||||
mine = [m for m in sent if m["channel"] == "召回-全局"]
|
||||
assert [m["event"] for m in mine] == ["ledger.recall"], "订阅了两样的渠道只推一次"
|
||||
# 看给人读的部分（标题、各行）：编号是十六进制的 id，不拿来比
|
||||
text = mine[0]["title"] + "".join(f"{k}{v}" for k, v in mine[0]["lines"])
|
||||
for number in ("170.5", "208.5", "38.0", "38.00"):
|
||||
assert number not in text, f"消息里不放数：{number}"
|
||||
assert dict(mine[0]["lines"]) == {"状态": "待处理", "口径": "「周销售」", "版本": "v1 → v2", "答案": "1 个",
|
||||
"召回": dict(mine[0]["lines"])["召回"]}
|
||||
assert "周销售" in mine[0]["title"] and "v2" in mine[0]["title"]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 目录事实被改、表结构变了
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _catalog(source_id: str, notes: dict) -> None:
|
||||
"""探查一次表结构（查询快照才冻结得进目录）、写这张表的目录。"""
|
||||
from app.data import catalog
|
||||
from app.data.engine import engines
|
||||
from app.data.introspect import introspect
|
||||
from app.db.models import DataSource
|
||||
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(DataSource, source_id)
|
||||
row.schema_cache = await introspect(row)
|
||||
await session.commit()
|
||||
await engines.invalidate(source_id)
|
||||
async with SessionLocal() as session:
|
||||
entry = await catalog.read_entry(session, source_id, "orders")
|
||||
await catalog.write_entry(session, source_id, "orders", notes, if_version=entry.version if entry else 0,
|
||||
actor="tester")
|
||||
|
||||
|
||||
def _item(value, status="confirmed"):
|
||||
return {"value": value, "source": "human", "status": status}
|
||||
|
||||
|
||||
async def test_catalog_fact_change_marks_answers_and_notifies_only_on_new_errors(engine_up, monkeypatch):
|
||||
from app.data import catalog
|
||||
|
||||
source_id = await S.set_source(engine_up)
|
||||
await _catalog(source_id, {"business_date": _item({"column": "paid_at"}),
|
||||
"columns": {"amount": {"measure": _item("flow")}}})
|
||||
row = await answer(monkeypatch)
|
||||
unrelated = await S.explore(monkeypatch, S.ELSEWHERE, text="门店 [[v:Q1.r0.n]] 家。", origin="chat")
|
||||
async with SessionLocal() as session:
|
||||
entry = await catalog.read_entry(session, source_id, "orders")
|
||||
# 改成存量：按新目录，对存量列求和、分组里没有业务日期是错误级问题（跨期加总存量）
|
||||
await catalog.save_human_edit(session, source_id, "orders",
|
||||
{**entry.notes, "columns": {"amount": {"measure": _item("stock")}}},
|
||||
if_version=entry.version, actor="tester")
|
||||
await recall.drain()
|
||||
[marked] = await events(row.id, EventType.LEDGER_BASIS_CHANGED)
|
||||
assert marked["kind"] == "catalog" and marked["table"] == "orders"
|
||||
assert marked["items"] == [{"path": "columns.amount.measure", "label": "列 amount 的度量类型", "change": "changed"}]
|
||||
assert marked["new_errors"] == 1 and marked["checks"][0]["code"] == "stock_summed"
|
||||
assert await events(unrelated.id, EventType.LEDGER_BASIS_CHANGED) == [], "没查过这张表的不标"
|
||||
[task] = await recall_tasks(row.id)
|
||||
assert task.correlation_key.startswith("catalog:")
|
||||
async with SessionLocal() as session:
|
||||
from app.api.deps import Principal
|
||||
|
||||
payload = await recall.task_payload(session, task, Principal(username=None, role="admin", mode="off", via="off"))
|
||||
assert payload["reason"] == "catalog" and payload["basis"]["new_errors"] == 1
|
||||
assert [a["decision"] for a in payload["actions"]] == ["ack"]
|
||||
|
||||
# 只改名称说明（不影响数字的项）：不标
|
||||
async with SessionLocal() as session:
|
||||
entry = await catalog.read_entry(session, source_id, "orders")
|
||||
await catalog.save_human_edit(session, source_id, "orders", {**entry.notes, "label": _item("订单")},
|
||||
if_version=entry.version, actor="tester")
|
||||
await recall.drain()
|
||||
assert len(await events(row.id, EventType.LEDGER_BASIS_CHANGED)) == 1
|
||||
|
||||
|
||||
def test_fact_changes_only_counts_items_that_affect_numbers():
|
||||
before = {"valid_filter": _item("status <> 'open'"), "label": _item("订单"),
|
||||
"columns": {"amount": {"measure": _item("flow"), "label": _item("金额")}}}
|
||||
after = {"valid_filter": {**_item("status <> 'open'"), "status": "rejected"}, "label": _item("订单表"),
|
||||
"columns": {"amount": {"label": _item("金额")}}}
|
||||
changes = recall.fact_changes(before, after)
|
||||
assert [(c["path"], c["change"]) for c in changes] == [("valid_filter", "rejected"),
|
||||
("columns.amount.measure", "removed")]
|
||||
assert recall.fact_changes(after, after) == []
|
||||
|
||||
|
||||
async def test_structure_change_marks_answers_and_probe_failure_notifies(engine_up, monkeypatch):
|
||||
import sqlite3
|
||||
|
||||
from app.data import source_changes as sc
|
||||
from app.db.models import DataSource, WorkflowVersion
|
||||
|
||||
wf = await S.publish()
|
||||
async with SessionLocal() as session:
|
||||
row = (await session.execute(select(WorkflowVersion).where(WorkflowVersion.workflow_id == wf))).scalar_one()
|
||||
row.published_by = "publisher"
|
||||
await session.commit()
|
||||
a1 = await answer(monkeypatch)
|
||||
# 列真的删掉了：口径的取数按这一版再算就算不出来
|
||||
db = sqlite3.connect(engine_up)
|
||||
db.executescript("ALTER TABLE orders DROP COLUMN amount;")
|
||||
db.close()
|
||||
async with SessionLocal() as session:
|
||||
source = (await session.execute(select(DataSource).where(DataSource.name == S.CURRENT["source"]))).scalar_one()
|
||||
change, _ = await sc.record(session, source, sc.NewChange(kind="column_removed", table="orders", column="amount",
|
||||
severity="action", key="v2"),
|
||||
sc.Detected(job_kind="introspect"))
|
||||
await session.commit()
|
||||
out = sc.change_out(change)
|
||||
assert recall.basis_kind(out) == "structure"
|
||||
assert await recall.process_change(out) == 1
|
||||
[marked] = await events(a1.id, EventType.LEDGER_BASIS_CHANGED)
|
||||
assert marked["kind"] == "structure" and marked["change"] == out["id"] and marked["column"] == "amount"
|
||||
async with SessionLocal() as session:
|
||||
[probe] = [r for r in (await session.execute(select(Run).where(Run.origin_kind == "recall"))).scalars()
|
||||
if (r.origin or {}).get("change") == out["id"]]
|
||||
assert probe.origin["probe"] == 1 and probe.origin["workflow_id"] == wf and probe.origin["inputs_from"] == "answer"
|
||||
probe = await S.wait(probe.id)
|
||||
assert probe.status == "failed"
|
||||
await recall.drain()
|
||||
[task] = await recall_tasks(probe.id)
|
||||
assert task.title == "口径「周销售」v1 依据的表结构变了，试算没有算出来"
|
||||
# 同一条变化再处理一次：不再标、不再试算
|
||||
assert await recall.process_change(out) == 0
|
||||
async with SessionLocal() as session:
|
||||
probes = (await session.execute(select(Run).where(Run.origin_kind == "recall"))).scalars().all()
|
||||
assert len([p for p in probes if (p.origin or {}).get("change") == out["id"]]) == 1
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 版本用量、预告接口、E4
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_usage_preview_batches_and_stats_endpoints(client, monkeypatch):
|
||||
wf = await S.publish()
|
||||
await answer(monkeypatch)
|
||||
usage = (await client.get(f"/api/ledger/workflow-versions/{wf}/1")).json()
|
||||
assert usage["answers"] == 1 and usage["exports"] == 0 and usage["recall"] == {"groups": 0, "changed": 0,
|
||||
"recalled_from": 0}
|
||||
await publish_v2(wf)
|
||||
preview = (await client.get(f"/api/workflows/{wf}/recall-preview", params={"version": 2})).json()
|
||||
assert preview["groups"] == 1 and preview["will_recompute"] == 1 and preview["over_limit"] == 0
|
||||
await run_recall(wf)
|
||||
view = (await client.get(f"/api/ledger/recall/{wf}/2")).json()
|
||||
assert view["counts"]["changed"] == 1 and view["groups"][0]["state"] == "changed"
|
||||
assert view["groups"][0]["changed_metrics"] == ["销售额"]
|
||||
again = await client.post(f"/api/ledger/recall/{wf}/2/continue")
|
||||
assert again.status_code == 409
|
||||
usage = (await client.get(f"/api/ledger/workflow-versions/{wf}/2")).json()
|
||||
assert usage["recall"]["groups"] == 1 and usage["recall"]["changed"] == 1
|
||||
old = (await client.get(f"/api/ledger/workflow-versions/{wf}/1")).json()
|
||||
assert old["recall"]["recalled_from"] == 1
|
||||
stats = (await client.get("/api/ledger/stats")).json()["e4"]["per_publish"]
|
||||
mine = next(p for p in stats["publishes"] if p["workflow_id"] == wf)
|
||||
assert mine == {"caliber": f"{wf}@2", "workflow_id": wf, "version": 2, "groups": 1, "settled": 1, "changed": 1,
|
||||
"failed": 0, "changed_rate": 1.0}
|
||||
|
||||
|
||||
async def test_publish_hook_queues_recall_when_automatic(client, monkeypatch):
|
||||
from app.core.config import settings
|
||||
|
||||
monkeypatch.setattr(settings, "scheduler", True)
|
||||
wf = await S.publish()
|
||||
await answer(monkeypatch)
|
||||
await S.publish(graph=S.caliber_graph(WITH_REFUNDED), version=2, workflow_id=wf, level="governed")
|
||||
async with SessionLocal() as session:
|
||||
from app.db.models import Workflow
|
||||
|
||||
(await session.get(Workflow, wf)).published_version = 1
|
||||
await session.commit()
|
||||
res = await client.post(f"/api/workflows/{wf}/publish", json={"level": "published", "version": 2})
|
||||
assert res.status_code == 200 and res.json()["ok"], res.text
|
||||
await recall.drain()
|
||||
async with SessionLocal() as session:
|
||||
runs = await recall.recall_runs(session, wf, 2)
|
||||
assert len(runs) == 1 and runs[0].origin["batch"] == "auto"
|
||||
await S.wait(runs[0].id)
|
||||
await recall.drain()
|
||||
|
||||
|
||||
async def test_recall_graph_is_deterministic_and_events_have_offline_labels(monkeypatch):
|
||||
"""同样的组永远得到同一张召回图；两个新事件在可验证报告的离线核对页、封存后事件摘要里有说法（不带数）。"""
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from app.bundle.build import _summary
|
||||
|
||||
wf = await S.publish()
|
||||
await answer(monkeypatch)
|
||||
await publish_v2(wf)
|
||||
async with SessionLocal() as session:
|
||||
plan = await recall.build_plan(session, wf, 2)
|
||||
[group] = plan.groups
|
||||
a, meta = recall.build_graph(group, plan.pairs[1])
|
||||
b, _ = recall.build_graph(group, plan.pairs[1])
|
||||
assert a.graph == b.graph and a.kind == "recall"
|
||||
assert meta == [{"k": 0, "card": "card", "metric": "gmv", "name": "销售额", "unit": "元", "decimals": 2,
|
||||
"format": "thousands", "expression": "cell(nodes.fetch, 0, 'gmv')"}]
|
||||
card = next(n for n in a.graph["nodes"] if n["id"] == "recall")["data"]["config"]
|
||||
assert [m["id"] for m in card["metrics"]] == ["old_0", "new_0", "diff_0"] and card["on_missing"] == "null"
|
||||
|
||||
script = (Path(__file__).resolve().parents[1] / "app" / "bundle" / "verifier.js").read_text(encoding="utf-8")
|
||||
block = re.search(r"const EVENT_LABEL = \{(.*?)\n \}", script, re.S)
|
||||
labels = dict(re.findall(r"'([a-z_.]+)': '([^']+)'", block.group(1)))
|
||||
assert labels["ledger.recalled"] == "口径召回核对" and labels["ledger.basis_changed"] == "答案的依据有变"
|
||||
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 项）"
|
||||
@@ -41,7 +41,8 @@ PUSHED_THIS_ROUND = {
|
||||
"key.expiring": {"global", "user"},
|
||||
# 第 4 波：W4-d 打开 source.changed 时在这里加它推给的几条路
|
||||
# （两行之间隔一行不动的，两个包各改各的、合并不撞）
|
||||
# 第 4 波：W4-a 打开 ledger.recall 时在这里加它推给的几条路
|
||||
# 第 4 波 W4-a：口径召回数真变了的待办推全局渠道和答案主人自己的个人渠道（notify_recall）
|
||||
"ledger.recall": {"global", "user"},
|
||||
}
|
||||
|
||||
|
||||
@@ -120,19 +121,24 @@ async def test_events_endpoint_says_who_receives_each_event(client):
|
||||
assert [r["event"] for r in rows] == list(service.EVENTS)
|
||||
by = {r["event"]: r for r in rows}
|
||||
assert by["eval.regressed"] == {"event": "eval.regressed", "label": "定时评测有退步", "scopes": ["global"]}
|
||||
assert by["ledger.recall"]["scopes"] == []
|
||||
assert by["ledger.recall"]["scopes"] == ["global", "user"]
|
||||
assert by["review.done"]["scopes"] == ["global", "user"]
|
||||
|
||||
|
||||
async def test_events_not_pushed_yet_cannot_be_subscribed(client):
|
||||
#: 登记了、还没有推送的演示事件（第 4 波之后本轮的六个都推了，用它守「不收订阅、不列」这条规矩）
|
||||
UNPUSHED = "demo.unpushed"
|
||||
|
||||
|
||||
async def test_events_not_pushed_yet_cannot_be_subscribed(client, monkeypatch):
|
||||
monkeypatch.setitem(service.EVENTS, UNPUSHED, service.Event("还没推送的演示事件", ()))
|
||||
resp = await client.post("/api/notify/channels", json={
|
||||
"name": "事件登记-评测", "url": URL, "events": ["task.opened", "ledger.recall"]})
|
||||
"name": "事件登记-评测", "url": URL, "events": ["task.opened", UNPUSHED]})
|
||||
assert resp.status_code == 422, resp.text
|
||||
detail = resp.json()["detail"]
|
||||
assert "口径升版，答案的数变了" in detail and "还没有推送" in detail
|
||||
assert "还没推送的演示事件" in detail and "还没有推送" in detail
|
||||
unknown = await client.post("/api/notify/channels", json={"name": "事件登记-未知", "url": URL, "events": ["nope"]})
|
||||
assert unknown.status_code == 422
|
||||
assert "口径升版，答案的数变了" not in unknown.json()["detail"], "可选的里面不列还没有推送的"
|
||||
assert "还没推送的演示事件" not in unknown.json()["detail"], "可选的里面不列还没有推送的"
|
||||
assert "待办打开" in unknown.json()["detail"]
|
||||
|
||||
|
||||
@@ -163,16 +169,17 @@ async def test_scope_mismatch_is_refused_only_for_newly_ticked_events(client):
|
||||
|
||||
async def test_channels_only_walk_the_paths_the_event_is_registered_for(monkeypatch):
|
||||
monkeypatch.setitem(service.EVENTS, "demo.personal", service.Event("只推个人的演示事件", ("user",)))
|
||||
monkeypatch.setitem(service.EVENTS, UNPUSHED, service.Event("还没推送的演示事件", ()))
|
||||
async with SessionLocal() as session:
|
||||
row = NotifyChannel(name="事件登记-全局", scope="global", url=encrypt(URL),
|
||||
events=["demo.personal", "task.opened", "ledger.recall"])
|
||||
events=["demo.personal", "task.opened", UNPUSHED])
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
try:
|
||||
ids = lambda found: [c.id for c in found] # noqa: E731
|
||||
assert row.id not in ids(await service._channels(session, "demo.personal", None, [])), "全局渠道这条路没登记"
|
||||
assert row.id in ids(await service._channels(session, "task.opened", None, []))
|
||||
assert await service._channels(session, "ledger.recall", None, []) == [], "还没有推送的：谁都不推"
|
||||
assert await service._channels(session, UNPUSHED, None, []) == [], "还没有推送的：谁都不推"
|
||||
with pytest.raises(ValueError, match="没有登记"):
|
||||
await service._channels(session, "demo.unregistered", None, [])
|
||||
finally:
|
||||
|
||||
Reference in New Issue
Block a user