口径账本内核:运行封存后在后台比对已发布口径、逐个数标口径、按运行读的账本接口、来源链和导出留痕、统计 E1–E4

为什么:问数据的答案要和已治理口径接起来(next-round 02 做法 2、3、9、10;总方案 X1、X2、X3、X12,拍板 L2、L4、L8)。

- 比对(ledger/match.py):探索运行(chat、canvas、trial、personal_app)成功结束、封存完、有查询快照时,在 runner 封存之后和
  notify_run 同一个位置放到后台;候选口径是每个工作流最新的已发布未退役版本里查库的调用工具节点(受管、已发布都算),按数据源
  和表预筛;逐个用 match_template 比,schema 用查询当时冻结的表结构、列名取同一份快照里冻结的数据目录。每次运行限时 8 秒、在
  线程里、同时最多 2 个;超时、快照被改过或读不出、SQL 解析不了照实记 skipped。启动时(run_manager.setup,和调度循环同一个开关)
  补比最近 7 天封存了却没比过的运行,只读工件和版本表。
- 记法(ledger/store.py,拍板 L8):内容寻址工件 kind=caliber_match(agentlab.caliber_match/1)+ run_append 的 sealed 方式追加
  ledger.matched(core/events.py 登记);工件里不放时间和工作流名,同一份封存再比一次哈希一样;meta 记 calibers / related / counts。
- 标签(ledger/labels.py):单元格按查询的比对状态;指标要每个输入都等价于同一份口径同一版、绑定一致,式子按「哪个取数节点、
  第几行、哪一列」换完和口径卡上的一字不差。caliber(how:independent 等价已证明;composed 助手选用留给第 3 波)/ related /
  reasoning / unknown;整题判断附还缺的入口字段。
- 读(ledger/view.py):GET /api/runs/{id}/ledger(ledger_api.runs_router,和证据同一道门),没比对过、能比的按需排上;证据
  片段接口加 ledger 键(没比对过的运行没有这个键,返回一字不差);报告导出的面板最上面多「口径」一步并写明「封存后比对,不在
  封存范围内」。遮罩按读的那一刻重滤:绑定的值是这次运行查到过的遮罩值时不给值,差异条目和逐行对照走 _Shown。
- 来源链:POST /api/runs 加 origin_ref(只在 origin=chat 时收,服务端核对对话访问权和轮次归属,写进 Run.origin 和封存的
  run.origin);WorkflowIn 加 origin(run_id、conversation_id、turn_id,核对规则同 from-run);报告导出和可验证报告下载各记一条
  审计 report.exported。可验证报告列封存后追加的事件时给 ledger.matched 一句只有条数的说明。
- 统计(ledger/stats.py):E1–E4 从工件、运行、查询快照、审计现算,GET /api/ledger/stats(搭建者);E3 用 app/chat/questions
  的聚类加按(数据源、normalize_sql 文本)的确定性重复;评测直接调 compute_stats(可点名 run_ids)。对账反馈、认证转化、按发布
  的召回在后面几波,照实写没有数据。

测试:test_ledger_match.py(真跑探索运行:等价、相关、没有相关口径、写死的码值、非探索运行、超时、快照被改、解析失败、预筛、
指标两种、级别、只认最新版、补比)、test_ledger_api.py(账本接口、按需比、遮罩、证据片段、导出、审计、统计)、
test_run_origin_ref.py(origin_ref 与画布来源的核对)。
This commit is contained in:
ThearchyHelios
2026-10-16 14:02:55 +08:00
parent 68ea3a7f47
commit 5b4bd91476
19 changed files with 2554 additions and 6 deletions
+6 -2
View File
@@ -8,8 +8,8 @@
另外要求运行已经结束(成功、失败、取消):失败后接着跑的运行留着上一次的封存,跑完会重新封存、换掉清单哈希,
早年的运行停在审批时也封存过——这时导出去的文件,封存哈希和系统里之后记的对不上,页面上「去系统里对照」
那一步就作废了。
只读:不追加任何事件,不改运行记录。路径挂在 /api/runs/{run_id}/… 下面,runs.py 里已经有 /{run_id}/verify,
这里叫 bundle.html,不撞名。
不追加运行事件、不改运行记录,只记一条导出审计(report.exported,format=bundle:召回要找得到谁下载过可验证报告,
next-round 02 做法 9)。路径挂在 /api/runs/{run_id}/… 下面,runs.py 里已经有 /{run_id}/verify,这里叫 bundle.html,不撞名。
"""
from __future__ import annotations
@@ -71,6 +71,10 @@ async def export_bundle(run_id: str) -> Response:
legacy=bool(sealed.seal.get("legacy")), graph=sealed.graph, source_masks=masks,
sure_columns=sure, sure_secrets=secrets,
read=lambda artifact: read_bytes(artifact, verify=False), exported_at=exported_at)))
from app.api.evidence import _reports, record_export
await record_export(run_id, "bundle", reports=[{"report_node": r.node_id, "doc_artifact": r.doc_artifact}
for r in _reports(sealed)])
return Response(page, media_type="text/html; charset=utf-8", headers={
"Content-Disposition": content_disposition(attachment_name(info), run_id=run_id),
# 导出的是这一刻的封存件,不该被哪一层缓存下来、换给下一个请求
+30 -3
View File
@@ -793,7 +793,13 @@ async def evidence_segment(run_id: str, segment_id: str, report: str | None = No
for c in step.get("masked") or []))
verdict = _effective_verdicts(chosen, sealed)[0].get(str(unit.get("id")))
challenged = _challenge_of(sealed, chosen, str(unit.get("id")))
# 口径账本(next-round 02 做法 3):这个数来自哪一份口径。比对在封存之后做、不在封存范围内,标签里写明;没比对过的运行
# (老运行、不是数的片段)没有这个键,返回一字不差
from app.ledger.view import segment_label
ledger = await segment_label(sealed, chosen.node_id, str(seg["id"]))
return {
**({"ledger": ledger} if ledger is not None else {}),
"report": {"node_id": chosen.node_id, "doc_artifact": chosen.doc_artifact},
"segment": {"id": seg["id"], "text": seg.get("text"), "kind": seg.get("kind"), "state": seg.get("state"),
"span": seg.get("span"), "unit": unit.get("id"),
@@ -2791,7 +2797,8 @@ async def export_report_html(run_id: str, report: str | None = None) -> Response
正文由服务端按报告文档渲染(不执行脚本也读得到);带出处链的数据嵌在页面里,执行脚本后每个数字、
无证据的数字、可疑的表名字段名都点得开——面板是网页上证据面板同一个拼法(_chain):口径卡、口径与
版本、原式、代入式、输入、复算、SQL、被引用的行。封存状态逐项标明;没封存的运行照常导出(和网页上
读报告一个道理)。只读:不追加任何事件,不改运行记录。
读报告一个道理)。不追加运行事件、不改运行记录,只记一条导出审计(report.exported,next-round 02 做法 9:召回要找得到
谁导出过这份报告)。口径账本比对过的运行,面板最上面多「口径」一步(同一份标签,写明封存后比对、不在封存范围内)。
一次运行里有好几份报告时用 ?report=<节点 id> 指定;不指定取成果里标注的那一份(和片段接口同一个选法)。
选报告、错误码和片段接口相同(run_not_found、evidence_report_not_found、evidence_doc_tampered、
@@ -2853,16 +2860,36 @@ async def export_report_html(run_id: str, report: str | None = None) -> Response
info = {"id": run.id, "workflow_id": run.workflow_id, "workflow_name": run.workflow_name,
"run_class": run.run_class, "status": run.status, "end_reason": run.end_reason,
"started_at": run.started_at, "finished_at": run.finished_at}
from app.ledger.view import report_labels
ledger = await report_labels(sealed, chosen.node_id)
data = build_report(info, report_node=chosen.node_id, doc_artifact=chosen.doc_artifact,
doc_hash_ok=chosen.hash_ok, doc=doc, seal=sealed.seal, chains=chains, refs=refs,
notes=notes, nodes=nodes, exported_at=datetime.now(timezone.utc).isoformat(timespec="seconds"))
return Response(render_html(data, doc), media_type="text/html; charset=utf-8", headers={
notes=notes, nodes=nodes, exported_at=datetime.now(timezone.utc).isoformat(timespec="seconds"),
ledger=ledger)
page = render_html(data, doc)
await record_export(run_id, "html", report_node=chosen.node_id, doc_artifact=chosen.doc_artifact)
return Response(page, media_type="text/html; charset=utf-8", headers={
"Content-Disposition": content_disposition(attachment_name(info), run_id=run_id),
# 导出的是这一刻的报告和出处,不该被哪一层缓存下来、换给下一个请求
"Cache-Control": "no-store",
})
async def record_export(run_id: str, fmt: str, **data: Any) -> None:
"""导出了一份报告(报告导出 html、可验证报告 bundle):记一条审计 report.exported(subject 是这次运行),谁导出、导出的
哪份报告文档。召回(口径升版后追回已交付的答案)据此找得到导出人。审计写不进去只记日志,不挡下载。"""
from app.auth.audit import record
try:
async with SessionLocal() as session:
await record(session, "report.exported", "run", run_id,
data={"format": fmt, **{k: v for k, v in data.items() if v is not None}})
await session.commit()
except Exception: # noqa: BLE001 - 留痕失败不挡下载,但要看得见
logger.exception("运行 %s 的报告导出审计没有写进去", run_id)
#: R14 之后(绑定快照、取引擎、编译核对、回查、行级核对)会抛的三类异常,按 _snapshot_failure 的顺序映射
_SNAPSHOT_ERRORS: tuple[type[BaseException], ...] = (SnapshotMissing, SnapshotTampered)
+36
View File
@@ -2,11 +2,18 @@
不另建存储:ActionRecord 本身带着工作流、版本、graph_hash、能力版本、payload_hash、回执;「口径 → 应用」就是固定这一版的
应用版本。用在发布弹窗、退役确认、版本列表三处。
口径账本(docs/design/next-round/02-口径账本.md)在这里加两个只读接口:
- GET /api/runs/{run_id}/ledger(runs_router,和证据同一道门):一次运行里每个数的口径标签、整题判断、每条查询的比对、能做的
动作(app/ledger/view.py);
- GET /api/ledger/stats(router,搭建者):账本自身给出的统计 E1–E4,从比对工件、运行、审计现算(app/ledger/stats.py)。
"""
from __future__ import annotations
from typing import Any
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
@@ -61,3 +68,32 @@ async def effects(workflow_id: str | None = None, version: int | None = None, ap
query = query.where(ActionRecord.capability == capability)
rows = (await session.execute(query.order_by(ActionRecord.created_at.desc()).limit(limit))).scalars()
return {"items": [effect_row(r) for r in rows]}
# --------------------------------------------------------------------------
# 口径账本(next-round 02)
# --------------------------------------------------------------------------
@runs_router.get("/{run_id}/ledger")
async def run_ledger(run_id: str) -> dict[str, Any]:
"""一次运行的口径账本:每个数来自哪一份口径(等价已证明 / 助手选用)、相关口径差在哪、这次的推理、没比对的原因;整题判断;
每条查询和哪几份口径比过。比对在封存之后做(note 写明不在封存范围内);没比对过、能比的运行这时排上按需比对,答复
state=pending。绑定的值是这次运行查到过的遮罩值时不给值,差异条目里的 SQL 片段按同一套遮罩换掉。"""
from app.ledger.view import run_view
return await run_view(run_id)
@router.get("/stats")
async def ledger_stats(since: datetime | None = Query(default=None, description="只算这之后发起的运行(ISO 时间)"),
until: datetime | None = Query(default=None, description="只算这之前发起的运行"),
origin: str | None = Query(default=None, description="逗号分隔的运行来源,缺省是探索运行的几种"),
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
"""账本自身给出的统计 E1–E4(app/ledger/stats.compute_stats):口径命中、对账、重复度、召回。全部现算,不另存;评测页
「口径命中与重复度」直接调同一个函数。"""
from app.api.deps import current_principal
from app.ledger.stats import compute_stats
origins = [o.strip() for o in (origin or "").split(",") if o.strip()] or None
return await compute_stats(session, since=since, until=until, origins=origins, principal=current_principal())
+34
View File
@@ -27,12 +27,24 @@ router = APIRouter(prefix="/api/runs", tags=["runs"])
#: POST /api/runs 声明了客户端不能声明的来源(422)
ORIGIN_NOT_ALLOWED = "origin_not_allowed"
#: origin_ref(问数据的运行属于哪段对话、哪一轮)只在来源是问数据时收;对话不存在、这一轮不属于这段对话(422)
ORIGIN_REF_NOT_ALLOWED = "origin_ref_not_allowed"
ORIGIN_REF_INVALID = "origin_ref_invalid"
#: 删除被拒:这次运行写过外部系统(有演练以外的生效记录,409)
RUN_HAS_EFFECTS = "run_has_effects"
class OriginRef(BaseModel):
"""问数据的运行属于哪段对话、哪一轮(next-round 02 做法 9)。服务端核对对话这个人用不用得了、这一轮是不是这段对话的,
写进 Run.origin 和封存的 run.origin 事件:运行和轮次之间从此有一条封存的链接(召回找答案的主人、统计按对话数重复都靠它),
不再只靠前端写的 meta.runId。"""
conversation_id: str = Field(min_length=1, max_length=32)
turn_id: str = Field(min_length=1, max_length=32)
class StartRunIn(BaseModel):
workflow_id: str | None = None
graph: dict[str, Any] | None = None # 直接跑一张未保存的图(画布上的"试运行")
@@ -46,6 +58,8 @@ class StartRunIn(BaseModel):
#: 运行来源(engine/origin.py):客户端只能声明 chat / canvas / launcher / api,不传按 api 记。
#: 应用、钩子、定时任务发起的来源只能由服务端写
origin: str | None = None
#: 来源是问数据时,这次运行属于哪段对话、哪一轮(OriginRef)。别的来源带它回 422
origin_ref: OriginRef | None = None
class RunOut(BaseModel):
@@ -105,6 +119,10 @@ async def start_run(
# 应用入口、钩子、定时任务的来源由服务端写:客户端能声明的话,谁都能把一次画布试跑说成应用发起的
raise CodedHTTPException(422, "运行来源只能是问数据、画布上的试运行、正式运行(手动)或接口,应用、外部系统和定时任务"
"发起的运行由服务端记录来源", ORIGIN_NOT_ALLOWED, field="origin")
if payload.origin_ref is not None and origin != "chat":
raise CodedHTTPException(422, "只有问数据的运行能写明属于哪段对话、哪一轮", ORIGIN_REF_NOT_ALLOWED,
field="origin_ref")
origin_data = await _origin_ref(session, payload.origin_ref) if payload.origin_ref is not None else None
options = None
actor = actor_of(x_actor)
memory_scope = payload.memory_scope
@@ -136,12 +154,28 @@ async def start_run(
version=payload.version,
actor=actor,
origin_kind=origin,
origin=origin_data,
options=options,
)
except LaunchRefused as e:
raise _refused(e) from e
async def _origin_ref(session: AsyncSession, ref: OriginRef) -> dict[str, str]:
"""核对 origin_ref:对话这个人用不用得了(不能用的 403,规则同助手接口的 ensure_conversation_access)、对话在不在、这一轮
是不是这段对话的。核对过的才写进运行来源:客户端不能把一次运行挂到别人的对话、别的轮次上。"""
from app.api.deps import ensure_conversation_access
from app.db.models import ConversationTurn
conversation = await ensure_conversation_access(session, ref.conversation_id)
if conversation is None:
raise CodedHTTPException(422, "运行要挂的对话不存在,可能已被删除", ORIGIN_REF_INVALID, field="origin_ref")
turn = await session.get(ConversationTurn, ref.turn_id)
if turn is None or turn.conversation_id != conversation.id:
raise CodedHTTPException(422, "运行要挂的这一轮不属于这段对话", ORIGIN_REF_INVALID, field="origin_ref")
return {"conversation_id": conversation.id, "turn_id": turn.id}
def _refused(e: LaunchRefused) -> HTTPException:
"""发起被拒的 HTTP 响应,和规则搬去 engine/launch.py 之前逐字相同:带机器码的回 {detail, code},
不带的回 {detail}。"""
+37
View File
@@ -39,11 +39,22 @@ PROVIDER_LEFT = ("剩余的问题是所用的模型接入不存在或已停用
PROVIDER_LEFT_FIX = ";也可在各条问题下方选择改用的模型接入"
class WorkflowOrigin(BaseModel):
"""新工作流出自哪次运行、哪段对话、哪一轮(next-round 02 做法 9):问数据的「在画布里打开」带上它,和从运行提取(/copilot/
from-run)记成同一个形状 {run_id, conversation_id?, turn_id?}。「口径 → 来源问题」、认证队列的「在画布里打开过」都读它。"""
run_id: str = Field(min_length=1, max_length=32)
conversation_id: str | None = Field(default=None, max_length=32)
turn_id: str | None = Field(default=None, max_length=32)
class WorkflowIn(BaseModel):
name: str = Field(min_length=1, max_length=200)
description: str = ""
graph: dict[str, Any] = Field(default_factory=dict)
tags: list[str] = Field(default_factory=list)
#: 来源(WorkflowOrigin):服务端核对运行读得到、对话用得了、这一轮属于这段对话,才记进 workflows.origin
origin: WorkflowOrigin | None = None
class WorkflowPatch(BaseModel):
@@ -137,11 +148,13 @@ async def list_workflows(session: AsyncSession = Depends(get_session)) -> list[W
async def create_workflow(
payload: WorkflowIn, session: AsyncSession = Depends(get_session)
) -> WorkflowOut:
origin = await _checked_origin(session, payload.origin) if payload.origin is not None else None
workflow = Workflow(
name=payload.name,
description=payload.description,
graph=payload.graph,
tags=payload.tags,
origin=origin,
)
session.add(workflow)
await session.flush()
@@ -154,6 +167,30 @@ async def create_workflow(
return await _workflow_out(session, workflow)
#: 新工作流的来源核对不过(422):运行读不到、这一轮不属于这段对话
WORKFLOW_ORIGIN_INVALID = "workflow_origin_invalid"
async def _checked_origin(session: AsyncSession, origin: WorkflowOrigin) -> dict[str, str]:
"""核对新工作流的来源,规则同从运行提取(/copilot/from-run):运行这个人读得到(读不到的当它不存在);对话这个人用得了
(不能用的 403);给了轮次的,这一轮要是这段对话的。对话不存在的照 from-run 记下 id(对话删了,来源仍是那次运行)。"""
from app.api.deps import can_read_run, current_principal, ensure_conversation_access
from app.db.models import ConversationTurn
run = await session.get(Run, origin.run_id)
if run is None or not await can_read_run(current_principal(), run):
raise CodedHTTPException(422, "来源的运行不存在,可能已被删除", WORKFLOW_ORIGIN_INVALID, field="origin")
conversation = await ensure_conversation_access(session, origin.conversation_id)
if origin.turn_id:
turn = await session.get(ConversationTurn, origin.turn_id)
if turn is None or not origin.conversation_id or turn.conversation_id != origin.conversation_id:
raise CodedHTTPException(422, "来源的这一轮不属于这段对话", WORKFLOW_ORIGIN_INVALID, field="origin")
return {"run_id": run.id,
**({"conversation_id": conversation.id if conversation else origin.conversation_id}
if origin.conversation_id else {}),
**({"turn_id": origin.turn_id} if origin.turn_id else {})}
@router.get("/{workflow_id}", response_model=WorkflowOut)
async def get_workflow(
workflow_id: str, session: AsyncSession = Depends(get_session)
+7
View File
@@ -1177,6 +1177,13 @@ def _summary(etype: str, data: Any) -> str:
if etype == EventType.APPROVAL_INQUIRY:
question = data.get("question")
return f"质询:{_clip(question)}" if isinstance(question, str) and question.strip() else "审批质询"
if etype == EventType.LEDGER_MATCHED:
# 口径账本的比对(next-round 02):只说各种状态几条,不带绑定和差异
counts = data.get("counts") if isinstance(data.get("counts"), dict) else {}
parts = [f"{counts[k]} 条{label}" for k, label in (("proven", "等价已证明"), ("related", "有相关口径"),
("none", "没有相关口径"), ("skipped", "没有比完"))
if isinstance(counts.get(k), int) and counts[k]]
return "封存后比对已发布口径" + (f":{'、'.join(parts)}" if parts else "")
if etype == EventType.LOG and isinstance(data.get("message"), str):
return _clip(data["message"])
return ""
+5
View File
@@ -83,6 +83,11 @@ class EventType(StrEnum):
# 探索运行里按需裁判的结论句判定:点开哪句才判哪句,追加在封存之后(post_seal),是批注不是证据。
# 封存核对只覆盖 manifest_seq 之前的事件,追加它不影响 verify;封存的报告文档一个字都不改
EVIDENCE_JUDGED = "evidence.judged"
# 口径账本(docs/design/next-round/02-口径账本.md 做法 2):探索运行封存之后,后台拿这次执行的 SQL 和已发布口径的取数比对,
# 结果存成内容寻址工件(kind=caliber_match,格式 agentlab.caliber_match/1),再追加这一条:{artifact, manifest_hash,
# algo, counts, trigger}。追加在封存之后(post_seal),是批注不是证据:比对依据的已发布口径在运行之外,不能混进运行
# 自身的封存事实;封存核对照旧一致
LEDGER_MATCHED = "ledger.matched"
# 应用平台(docs/design/app-platform-and-look.md §3.11)
# 运行来源:发起时写进 launch_run 的 notes,在封存范围内(engine/origin.py)
+20
View File
@@ -130,9 +130,18 @@ class RunManager:
for run_id, last_seq in queued:
bus.set_seq(run_id, last_seq or 0)
await self._emit(run_id, EventType.RUN_FAILED, data={"error": QUEUED_AT_RESTART})
# 口径账本:最近几天封存了、一次都没比过的探索运行,在后台补比一次(ledger/match.py)。只读工件和版本表,不查业务库、
# 不调模型;和调度循环同一个开关,拷来的数据目录、测试不补
from app.ledger import match as ledger_match
ledger_match.start_backfill()
async def shutdown(self) -> None:
self._closing = True
# 后台的口径比对先停:没做完的不记,下次启动补比会再比
from app.ledger import match as ledger_match
await ledger_match.shutdown()
# 只等还没结束的:结束了的没什么可等,而且它可能属于另一个事件循环,
# gather 会直接抛 "future belongs to a different loop"
pending = [t for t in self._tasks.values() if not t.done()]
@@ -991,10 +1000,14 @@ class RunManager:
# 每次都读到 running 和空的 output
event: RunEventModel | None = None
app_id: str | None = None
#: 收尾之后要不要放到后台比对口径(ledger/match.py):只看探索运行
run_class: str | None = None
origin_kind: str | None = None
async with SessionLocal() as session:
run = await session.get(Run, run_id)
if run:
app_id = run.app_id
run_class, origin_kind = run.run_class, run.origin_kind
timing = await _timing(session, run, elapsed_ms, now)
events_usage = await _usage_of_events(session, run_id)
inquiry = events_usage.pop("inquiry", None)
@@ -1064,6 +1077,13 @@ class RunManager:
from app.notify.service import notify_run, spawn
spawn(notify_run(run_id, "run.failed" if status == "failed" else "run.finished"))
if status == "succeeded":
# 口径账本(docs/design/next-round/02-口径账本.md 做法 2):探索运行成功结束、封存完,拿这次执行的 SQL 和已发布口径
# 比对。和通知同一个位置、同样放到后台:比对是 CPU 活,在线程里限时限并发,不拖着收尾;结果追加在封存之后
from app.ledger import match as ledger_match
if ledger_match.wanted(run_class, origin_kind):
ledger_match.schedule(run_id)
async def _seal(self, run_id: str) -> None:
"""对到此为止的全部事件算清单哈希,记下封到了哪一条。"""
+353
View File
@@ -0,0 +1,353 @@
"""逐个数标口径(docs/design/next-round/02-口径账本.md 做法 3):报告里每个引用了查询格、口径卡指标的数,来自哪一份口径。
标签四种(KINDS):
- caliber:来自口径「X」vN。how 是怎么认的:independent「等价已证明」(这次 Agent 自己写的 SQL 被证明和口径的取数等价)、
composed「助手选用」(助手按口径组合出来的;组合在第 3 波,这一波只留这个值)。
- related:相关口径——同一个数据源、查了同样的表的已发布口径,但不等价;附差在哪(explain_sql_diff 的条目)。
- reasoning:这次的推理——没有相关的已发布口径。
- unknown:没比对、比对没做完(超时、解析失败)、快照读不出来。
怎么认:
- 单元格([[v:]]、[[table:]] 里的格):按证据目录找到查询快照,取那条查询的比对状态。等价时按列映射写出口径那边的列名。
- 指标([[m:]]):指标集工件里这个指标的每个输入都要来自查询快照,而且这些快照都被证明等价于同一份口径的同一版、绑定一致;
指标式还要和那一版里某张口径卡上的某个指标相同——两边的式子把每个输入换成「口径的哪个取数节点、第几行、哪一列」之后
一字不差(caliber_from 转抄来的式子本来就是那一张卡的,同样按这个比)。输入有不来自查询的(运行输入、整形、代码)不算来自口径。
整题(whole_of):主报告里所有带标签的数都来自同一口径的同一版、绑定一致,就记 equivalent,并写出还缺的入口字段(口径入口
节点声明了、这次的绑定里没有的;required 且没有缺省值的另列)。
这里只是纯函数:标签在比对时算好、存进比对工件(ledger/match.py);接口读的时候按这次运行查到过的遮罩值遮一遍(view_label):
绑定的值是遮罩值的不给值,差异条目里的 SQL 片段走证据面板同一套遮罩(api/evidence._Shown)。
"""
from __future__ import annotations
from collections.abc import Callable, Mapping
from typing import Any
#: 标签的四种、认法的两种
KINDS = ("caliber", "related", "reasoning", "unknown")
HOWS = ("independent", "composed")
#: 界面上标签要写明的两句:比对在封存之后做,不在封存范围内;事后补比的是按现在的口径回看
POST_SEAL_NOTE = "封存后比对,不在封存范围内"
RETRO_NOTE = "按现在的口径回看"
#: 发布级别的叫法(同 blueprints/publish_plan.LEVEL_TEXT、前端 lib/terms.ts)
LEVEL_TEXT = {"published": "已发布", "governed": "受管"}
MASKED = "已遮罩"
#: unknown 的几种原因
NO_RECORD = "这一格出自的查询没有比对记录"
CARD_UNREADABLE = "指标集读不出来,没有比对"
METRIC_GONE = "指标集里没有这个指标,没有比对"
UNFINISHED = "比对没有做完"
#: related 的几种说明
FORMULA_DIFFERS = "取数和口径等价,但指标的算法和口径卡上的不同"
INPUTS_ELSEWHERE = "指标有的输入不来自查询(运行输入、整形或代码),不能算来自口径"
MIXED_CALIBERS = "指标的几个输入分别对上了不同的口径"
BINDING_CONFLICT = "指标的几个输入对上了同一份口径,但参数不一致"
#: 能追到查询快照里那一格的输入(同 engine/lineage.SNAPSHOT_VIAS)
_SNAPSHOT_VIAS = ("tool_cell", "agent_field")
def caliber_key(workflow_id: str, version: int) -> str:
"""口径版本的写法「工作流 id@版本」:工件 meta、召回反查、统计都按它认。"""
return f"{workflow_id}@{version}"
def _unknown(reason: str) -> dict[str, Any]:
return {"kind": "unknown", "reason": reason}
def _related(query: Mapping[str, Any], *, reason: str | None = None) -> dict[str, Any]:
"""相关口径:这条查询比过、没等价的那几份(附差在哪)。一份相关的都没有时是这次的推理。"""
related = [{k: cand[k] for k in ("caliber", "workflow_id", "version", "level", "node_id", "comparable",
"reason", "diff", "binding", "via") if k in cand}
for cand in query.get("candidates") or [] if cand.get("status") == "related"]
if not related:
return {"kind": "reasoning"}
out: dict[str, Any] = {"kind": "related", "related": related}
if reason:
out["reason"] = reason
return out
def _caliber(cand: Mapping[str, Any], **extra: Any) -> dict[str, Any]:
return {"kind": "caliber", "how": "independent", "caliber": cand.get("caliber"),
"workflow_id": cand.get("workflow_id"), "version": cand.get("version"), "level": cand.get("level"),
"binding": dict(cand.get("binding") or {}), **extra}
def _proven(query: Mapping[str, Any]) -> list[Mapping[str, Any]]:
return [c for c in query.get("candidates") or [] if c.get("status") == "proven"]
def _caliber_column(cand: Mapping[str, Any], column: Any) -> str | None:
"""这次结果里的列 → 口径那边同一位置的列名(列映射,按位置);没有映射、对不上时为 None。"""
for item in cand.get("columns") or []:
if item.get("ours") == column:
return item.get("caliber")
return None
def cell_label(artifact: str, column: Any, queries: Mapping[str, Mapping[str, Any]]) -> dict[str, Any]:
"""单元格的标签:它所在的那条查询的比对状态。"""
query = queries.get(artifact)
if query is None:
return _unknown(NO_RECORD)
status = query.get("status")
if status in ("proven", "composed"):
cand = _proven(query)[0] if _proven(query) else {}
label = _caliber(cand, query=artifact, node_id=cand.get("node_id"),
column=_caliber_column(cand, column) if column is not None else None)
if status == "composed":
label["how"] = "composed"
return label
if status == "related":
return {**_related(query), "query": artifact}
if status == "none":
return {"kind": "reasoning", "query": artifact}
return {**_unknown(str(query.get("reason") or UNFINISHED)), "query": artifact}
def _leaf_token(node_id: Any, row: Any, column: Any) -> str:
"""把指标式里的一个输入换成「口径的哪个取数节点、第几行、哪一列」:两边的式子按这个换完再逐字比。"""
return f"⟦{node_id}|{row}|{column}⟧"
def _mapped_expression(expression: str, tokens: Mapping[str, str]) -> str | None:
"""指标式里的输入按 tokens(输入路径 → 记号)换掉之后的写法;解析不了、有输入没换到的为 None。"""
from app.engine.expressions import ExpressionError, leaf_refs, parse_expression, substitute
try:
tree = parse_expression(expression)[0]
except ExpressionError:
return None
leaves = leaf_refs(tree)
if not leaves or any(leaf not in tokens for leaf in leaves):
return None
return substitute(tree, dict(tokens))
def card_expressions(cards: list[Mapping[str, Any]]) -> list[tuple[str, str, str]]:
"""口径版本里每张源头口径卡的每个指标:(卡的节点 id, 指标 id, 换成记号后的式子)。卡上的输入要是 cell(nodes.X, 行, '列')
(look/calibers.card_inputs 认的那种),别的写法的指标不收。"""
import ast
from app.engine.expressions import ExpressionError, cell_parts, leaf_refs, parse_expression
out: list[tuple[str, str, str]] = []
for card in cards:
for metric in card.get("metrics") or []:
expression = str((metric or {}).get("expression") or "")
try:
tree = parse_expression(expression)[0]
except ExpressionError:
continue
tokens: dict[str, str] = {}
for leaf in leaf_refs(tree):
parts = cell_parts(leaf)
if parts is None or not parts[0].startswith("nodes."):
break
try:
row, column = ast.literal_eval(parts[1]), ast.literal_eval(parts[2])
except (ValueError, SyntaxError):
break
tokens[leaf] = _leaf_token(parts[0][len("nodes."):], row, column)
else:
if (mapped := _mapped_expression(expression, tokens)) is not None:
out.append((str(card.get("node_id")), str(metric.get("id")), mapped))
return out
def metric_label(card: Mapping[str, Any] | None, metric_id: Any, queries: Mapping[str, Mapping[str, Any]],
calibers: Mapping[str, Mapping[str, Any]]) -> dict[str, Any]:
"""指标的标签(见模块说明)。card 是封存的指标集工件(读不出为 None);calibers 是 {「工作流@版本」: {level, cards}},
cards 是那一版里源头口径卡的指标式(match 从版本快照取)。"""
if card is None:
return _unknown(CARD_UNREADABLE)
metric = next((m for m in card.get("metrics") or [] if isinstance(m, dict) and m.get("id") == metric_id), None)
if metric is None:
return _unknown(METRIC_GONE)
inputs = [i for i in metric.get("inputs") or [] if isinstance(i, dict)]
from_queries = [i for i in inputs if i.get("artifact") and i.get("via") in _SNAPSHOT_VIAS]
if not from_queries:
return {"kind": "reasoning"}
found = [queries.get(str(i["artifact"])) for i in from_queries]
if any(q is None for q in found):
return _unknown(NO_RECORD)
if any(q.get("status") == "skipped" for q in found): # type: ignore[union-attr]
return _unknown(next(str(q.get("reason") or UNFINISHED) for q in found
if q.get("status") == "skipped")) # type: ignore[union-attr]
queries_found: list[Mapping[str, Any]] = [q for q in found if q is not None]
proven_keys = [{str(c.get("caliber")) for c in _proven(q)} for q in queries_found]
common = set.intersection(*proven_keys) if proven_keys else set()
nearby = _brief([c for q in queries_found for c in q.get("candidates") or []
if c.get("status") in ("proven", "related")])
if len(from_queries) < len(inputs):
return {"kind": "related", "related": nearby, "reason": INPUTS_ELSEWHERE} if nearby else {"kind": "reasoning"}
if not common:
if all(proven_keys):
return {"kind": "related", "related": nearby, "reason": MIXED_CALIBERS}
related = next((q for q in queries_found if q.get("status") == "related"), None)
return _related(related) if related is not None else {"kind": "reasoning"}
reason = FORMULA_DIFFERS
for key in sorted(common):
chosen = [next(c for c in _proven(q) if c.get("caliber") == key) for q in queries_found]
binding: dict[str, Any] = {}
conflict = False
for cand in chosen:
for name, value in (cand.get("binding") or {}).items():
if name in binding and binding[name] != value:
conflict = True
binding.setdefault(name, value)
if conflict:
reason = BINDING_CONFLICT
continue
tokens: dict[str, str] = {}
for item, cand in zip(from_queries, chosen):
locator = item.get("locator") if isinstance(item.get("locator"), dict) else {}
mapped = _caliber_column(cand, locator.get("column"))
if mapped is None or item.get("path") is None:
tokens = {}
break
tokens[str(item["path"])] = _leaf_token(cand.get("node_id"), locator.get("row"), mapped)
mine = _mapped_expression(str(metric.get("expression") or ""), tokens) if tokens else None
if mine is None:
continue
hit = next(((node, mid) for node, mid, theirs in (calibers.get(key) or {}).get("expressions") or []
if theirs == mine), None)
if hit is None:
continue
return {**_caliber(chosen[0], card=hit[0], metric=hit[1]), "binding": binding}
return {"kind": "related", "related": _brief([c for q in queries_found for c in _proven(q)
if c.get("caliber") in common]), "reason": reason}
def _brief(cands: list[Mapping[str, Any]]) -> list[dict[str, Any]]:
"""指标上的相关口径:只写是哪一版、哪个取数节点(差在哪在各个输入的查询上),同一版只写一次,最多 5 个。"""
out: list[dict[str, Any]] = []
for cand in cands:
if any(o.get("caliber") == cand.get("caliber") for o in out):
continue
out.append({k: cand[k] for k in ("caliber", "workflow_id", "version", "level", "node_id") if k in cand})
return out[:5]
def label_doc(doc: Mapping[str, Any], queries: Mapping[str, Mapping[str, Any]],
calibers: Mapping[str, Mapping[str, Any]],
load_card: Callable[[str], Mapping[str, Any] | None]) -> dict[str, dict[str, Any]]:
"""一份报告文档里每个数的标签:{片段 id: 标签}。只标引用解析了的数字、取值片段(单元格、指标);运行输入、表名、引文、
没有引用的数不标。"""
from app.engine.evidence import iter_units
catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {}
cards: dict[str, Mapping[str, Any] | None] = {}
out: dict[str, dict[str, Any]] = {}
for _, unit in iter_units(doc):
for seg in unit.get("segments") or []:
if not isinstance(seg, dict) or seg.get("kind") not in ("number", "value") or not seg.get("id"):
continue
cite = seg.get("cite") if isinstance(seg.get("cite"), dict) else {}
if cite.get("status") != "resolved":
continue
entry = catalog.get(cite.get("alias")) if isinstance(catalog.get(cite.get("alias")), dict) else {}
locator = cite.get("locator") if isinstance(cite.get("locator"), dict) else {}
artifact = str(entry.get("artifact") or "")
if cite.get("kind") == "cell" and artifact:
out[str(seg["id"])] = cell_label(artifact, locator.get("column"), queries)
elif cite.get("kind") == "metric" and artifact:
if artifact not in cards:
cards[artifact] = load_card(artifact)
out[str(seg["id"])] = metric_label(cards[artifact], locator.get("metric"), queries, calibers)
return out
def whole_of(labels: Mapping[str, Mapping[str, Any]], calibers: Mapping[str, Mapping[str, Any]]) -> dict[str, Any]:
"""整题判断(主报告):所有带标签的数都来自同一口径的同一版、绑定一致时 equivalent,附还缺的入口字段。"""
counts = {kind: 0 for kind in KINDS}
for label in labels.values():
counts[str(label.get("kind"))] = counts.get(str(label.get("kind")), 0) + 1
if not labels:
return {"equivalent": False, "reason": "报告里没有可标口径的数", "counts": counts}
keys = {label.get("caliber") for label in labels.values() if label.get("kind") == "caliber"}
if counts["caliber"] != len(labels):
reason = ("有的数没有比对" if counts["unknown"] else "有的数是这次的推理" if counts["caliber"]
else "没有数来自口径")
return {"equivalent": False, "reason": reason, "counts": counts}
if len(keys) != 1:
return {"equivalent": False, "reason": "数来自不同的口径", "counts": counts}
binding: dict[str, Any] = {}
for label in labels.values():
for name, value in (label.get("binding") or {}).items():
if name in binding and binding[name] != value:
return {"equivalent": False, "reason": "各个数的参数不一致", "counts": counts}
binding.setdefault(name, value)
key = str(next(iter(keys)))
info = calibers.get(key) or {}
fields = [f for f in info.get("fields") or [] if isinstance(f, dict) and f.get("name")]
first = next(iter(labels.values()))
return {
"equivalent": True, "caliber": key, "workflow_id": first.get("workflow_id"), "version": first.get("version"),
"level": first.get("level"),
"how": "composed" if any(label.get("how") == "composed" for label in labels.values()) else "independent",
"binding": binding,
"missing": [f["name"] for f in fields if f["name"] not in binding],
"missing_required": [f["name"] for f in fields
if f["name"] not in binding and f.get("required") and not f.get("has_default")],
"counts": counts,
}
# --------------------------------------------------------------------------
# 读的时候遮:绑定的值、差异条目里的 SQL 片段
# --------------------------------------------------------------------------
def binding_view(binding: Mapping[str, Any] | None, hide: Callable[[Any], bool]) -> list[dict[str, Any]]:
"""绑定 → [{field, value} 或 {field, masked: true}]:值是这次运行查到过的遮罩值的不给值(也就不能拿它代填出具)。"""
out = []
for name, value in (binding or {}).items():
out.append({"field": name, "masked": True, "value": MASKED} if hide(value) else {"field": name, "value": value})
return out
def diff_view(diff: Mapping[str, Any] | None, shown: Callable[[Any], Any]) -> dict[str, Any] | None:
"""差在哪:条目的两边写法、说明和逐行对照里的 SQL 片段按遮罩值换掉。"""
if not isinstance(diff, Mapping):
return None
items = [{**item, "a": shown(item.get("a")), "b": shown(item.get("b")), "text": shown(item.get("text"))}
for item in diff.get("items") or [] if isinstance(item, Mapping)]
return {"items": items, "fallback": bool(diff.get("fallback")),
"text_diff": [shown(line) for line in diff.get("text_diff") or []]}
def view_label(label: Mapping[str, Any], *, hide: Callable[[Any], bool], shown: Callable[[Any], Any],
names: Mapping[str, str]) -> dict[str, Any]:
"""一个标签交给界面的样子:绑定按遮罩给、差在哪按遮罩换,口径写上工作流名和级别的叫法(名字按现在的)。"""
out = {k: v for k, v in label.items() if k not in ("binding", "related")}
if label.get("kind") == "caliber":
out["binding"] = binding_view(label.get("binding"), hide)
out["workflow_name"] = names.get(str(label.get("workflow_id") or ""))
out["level_text"] = LEVEL_TEXT.get(str(label.get("level") or ""), "")
if label.get("kind") == "related":
out["related"] = [related_view(item, hide=hide, shown=shown, names=names)
for item in label.get("related") or []]
return out
def related_view(item: Mapping[str, Any], *, hide: Callable[[Any], bool], shown: Callable[[Any], Any],
names: Mapping[str, str]) -> dict[str, Any]:
out = {k: item[k] for k in ("caliber", "workflow_id", "version", "level", "node_id", "comparable", "reason", "via")
if k in item}
out["workflow_name"] = names.get(str(item.get("workflow_id") or ""))
out["level_text"] = LEVEL_TEXT.get(str(item.get("level") or ""), "")
if "binding" in item:
out["binding"] = binding_view(item.get("binding"), hide)
if "diff" in item:
out["diff"] = diff_view(item.get("diff"), shown)
return out
__all__ = ["HOWS", "KINDS", "LEVEL_TEXT", "POST_SEAL_NOTE", "RETRO_NOTE", "binding_view", "caliber_key",
"card_expressions", "cell_label", "diff_view", "label_doc", "metric_label", "related_view", "view_label",
"whole_of"]
+557
View File
@@ -0,0 +1,557 @@
"""认口径:运行封存之后,拿这次执行的 SQL 和已发布口径的取数比对(docs/design/next-round/02-口径账本.md 做法 2)。
**什么时候比。** 探索运行(问数据 chat、画布试运行 canvas、蓝图试运行 trial、个人应用刷新 personal_app)成功结束、封存完、
封存范围内有查询快照时,在 runner 封存之后放到后台比(和 notify_run 同一个位置 spawn):比对是 CPU 活,不该拖着运行收尾;
放在封存之前,外部的口径表就混进了运行自身的封存事实。正式运行不比 SQL——它本身就是某一版的承诺,账本按 Run.workflow_id /
version 记。系统派生的运行(评测、对账、召回)不比:它们是服务端为了量、为了比替人跑的(engine/origin.SYSTEM_ORIGINS)。
**和谁比。** 每个工作流最新的已发布、未退役版本(apps/versions.latest_published,和 look 的 caliber、蓝图复用口径同一种说法,
拍板 L4:受管、已发布都算,标签写明级别)里查库的调用工具节点。先按同一个数据源、SQL 里的表有交集预筛,每条查询最多比
MAX_CANDIDATES 份(按工作流、版本、节点排好的先后)。
**怎么比。** data/sql_semantics.match_template:先绑定参数(只是启发),再按绑定渲染口径的取数 SQL 交 explain_sql_diff 证明。
schema 用这条查询当时冻结的表结构(表结构快照),列的中文名取同一份快照里冻结的数据目录。只读封存范围内的快照(api/evidence
的 _sealed 读法),读不出、被改过的快照不比,记 skipped。
**资源。** 每次运行限时 TIMEOUT_S(缺省 8 秒),在线程里跑,同时最多 CONCURRENCY(2)个(照 scheduler/auto_compare 的做法:
信号量按事件循环分开,线程池两个线程)。时限按绑定逐组看:过了时限,还没比的口径记 skipped(比对超时);整次比对卡在线程里
过了时限再加 GRACE_S 还没回来,整次记 skipped。解析失败同样照实记 skipped 和原因。都只记日志和 skipped,不影响运行本身。
**结果。** ledger/store:内容寻址工件 kind=caliber_match + 封存之后追加的 ledger.matched。
**启动时补比。** 最近 BACKFILL_DAYS 天封存了、却一次都没比过的探索运行,启动后在后台按从新到旧逐个补比(最多 BACKFILL_LIMIT 个),
trigger=backfill。补比只读工件和版本表(加上数据源的类型,取方言用),不查业务库、不调模型。和调度循环同一个开关
(AGENTLAB_SCHEDULER):拷来的数据目录(检查脚本)、测试不在启动时替原库补做后台的事。打开账本时没比过的运行按需比
(trigger=on_demand,api/ledger 的按运行读接口),同样是事后按那时的口径回看。
评测(第 3 波)要逐题比 SQL 时直接调 compare_queries:同一个比对函数。
"""
from __future__ import annotations
import asyncio
import functools
import logging
import time
import weakref
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any
from sqlalchemy import select
from app.db.base import SessionLocal
from app.db.models import DataSource, Run, Workflow, WorkflowVersion
from app.ledger import labels as ledger_labels
from app.ledger import store
logger = logging.getLogger(__name__)
#: 比对哪几种来源的运行(见模块说明)
EXPLORE_ORIGINS = ("chat", "canvas", "trial", "personal_app")
#: 每次运行的时限(秒)、同时最多几个
TIMEOUT_S = 8.0
CONCURRENCY = 2
#: 线程过了时限还没回来,再等这么久就整次记 skipped
GRACE_S = 2.0
#: 每条查询最多比几份口径
MAX_CANDIDATES = 8
#: 启动时补比最近几天、最多几个
BACKFILL_DAYS = 7
BACKFILL_LIMIT = 200
#: skipped 的原因
TIMEOUT = "比对超过时限,没有做完"
SEAL_BROKEN = "封存核对不一致,不比对"
SNAPSHOT_TAMPERED = "查询快照与哈希不一致,疑似被修改,不比对"
SNAPSHOT_MISSING = "查询快照读不出来,不比对"
NO_SQL = "查询快照里没有 SQL,不比对"
PARSE_FAILED = "这次的 SQL 无法解析,没有比对"
MORE_CANDIDATES = "另有 {n} 份相关口径没有比对(每条查询最多比 {max} 份)"
#: 不比对的原因(GET 账本时照实说)
NOT_FOUND = "运行记录不存在"
FORMAL = "正式运行本身就是某一版口径的出具,不比对 SQL"
NOT_EXPLORE = "只比对问数据、画布试运行、蓝图试运行和个人应用刷新这几种探索运行"
NOT_SUCCEEDED = "运行没有成功结束,不比对"
NOT_SEALED = "运行还没有封存,不比对"
NO_QUERIES = "这次运行没有查询,没有可比对的"
def ineligible(run: Any) -> str | None:
"""这次运行比不比对:不比的回一句为什么,比的是 None(还要封存范围内有查询快照,见 match_run)。"""
if run is None:
return NOT_FOUND
if run.run_class == "formal":
return FORMAL
if (run.origin_kind or "") not in EXPLORE_ORIGINS:
return NOT_EXPLORE
if run.status != "succeeded":
return NOT_SUCCEEDED
if not run.manifest_hash:
return NOT_SEALED
return None
def wanted(run_class: str | None, origin_kind: str | None) -> bool:
"""runner 收尾时用:成功结束的这次运行要不要放到后台比(状态、封存由 match_run 再核对一次)。"""
return run_class != "formal" and (origin_kind or "") in EXPLORE_ORIGINS
# --------------------------------------------------------------------------
# 口径与查询
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class Fetch:
"""口径版本里一个查库的调用工具节点。"""
node_id: str
source: str
sql: str
tables: frozenset[str]
@dataclass(frozen=True)
class Caliber:
"""一个工作流最新的已发布、未退役版本(只收里面有查库节点的)。fields 是入口节点的字段原样(取缺省值用,不进工件),
cards 是源头口径卡的节点 id,expressions 是这些卡上每个指标换成记号后的式子(labels.card_expressions)。"""
workflow_id: str
version: int
level: str
graph_hash: str | None
fetches: tuple[Fetch, ...]
fields: tuple[dict[str, Any], ...] = ()
cards: tuple[str, ...] = ()
expressions: tuple[tuple[str, str, str], ...] = ()
@property
def key(self) -> str:
return ledger_labels.caliber_key(self.workflow_id, self.version)
def summary(self) -> dict[str, Any]:
"""工件里记的「考虑过的口径版本」:不放工作流名(会改)、不放缺省值。"""
return {"caliber": self.key, "workflow_id": self.workflow_id, "version": self.version, "level": self.level,
"graph_hash": self.graph_hash, "fetches": [f.node_id for f in self.fetches], "cards": list(self.cards),
"fields": [{"name": f.get("name"), "required": bool(f.get("required")),
"has_default": f.get("default") not in (None, "")} for f in self.fields]}
@dataclass
class Query:
"""封存范围内的一条查询:快照工件、哪个节点查的、数据源、SQL、结果的列名、SQL 里的表、方言、当时冻结的表结构和列的中文名。
problem 不为空时这条不比(快照读不出、被改过、没有 SQL)。"""
artifact: str
node_id: str | None
source: str
sql: str = ""
columns: list[str] | None = None
tables: frozenset[str] = frozenset()
dialect: str | None = None
schema: dict[str, Any] | None = None
names: dict[str, str] = field(default_factory=dict)
problem: str | None = None
def _table_keys(sql: str) -> frozenset[str]:
"""SQL 里 FROM / JOIN 的表名(去掉库名前缀、按 name_key),预筛用。"""
from app.data.names import name_key
from app.engine.evidence import sql_tables
try:
return frozenset(name_key(t.rsplit(".", 1)[-1]) for t in sql_tables(sql or ""))
except Exception: # noqa: BLE001 - 认不出表:这一条预筛不上,不比
return frozenset()
def caliber_of(workflow_id: str, row: Any) -> Caliber | None:
"""一个已发布版本 → Caliber;里面没有查库的调用工具节点(SQL 写在 args.sql 里)的不算口径的取数,返回 None。"""
from app.apps.tree import node_map
from app.tools.datasource import QUERY_PREFIX
graph = getattr(row, "graph", None) or {}
nodes = node_map(graph)
fetches: list[Fetch] = []
fields: list[dict[str, Any]] = []
cards: list[dict[str, Any]] = []
for node_id, node in sorted(nodes.items()):
cfg = node["config"] if isinstance(node["config"], dict) else {}
if node["type"] == "tool":
tool = str(cfg.get("tool") or "")
sql = (cfg.get("args") or {}).get("sql") if isinstance(cfg.get("args"), dict) else None
if tool.startswith(QUERY_PREFIX) and "{{" not in tool and isinstance(sql, str) and sql.strip():
fetches.append(Fetch(node_id, tool[len(QUERY_PREFIX):], sql, _table_keys(sql)))
elif node["type"] == "input":
fields += [dict(f) for f in cfg.get("fields") or [] if isinstance(f, dict) and f.get("name")]
elif node["type"] == "metrics" and cfg.get("caliber_from") in (None, "", {}):
cards.append({"node_id": node_id, "metrics": [m for m in cfg.get("metrics") or [] if isinstance(m, dict)]})
if not fetches:
return None
return Caliber(workflow_id=workflow_id, version=int(row.version), level=str(getattr(row, "level", None) or ""),
graph_hash=getattr(row, "graph_hash", None), fetches=tuple(fetches), fields=tuple(fields),
cards=tuple(c["node_id"] for c in cards),
expressions=tuple(ledger_labels.card_expressions(cards)))
async def load_calibers(session: Any) -> list[Caliber]:
"""每个工作流最新的已发布、未退役版本里的口径取数(apps/versions.is_published,同 latest_published 的说法)。先只读版本号
和级别挑出每个工作流的那一版,再读这几版的图。"""
from app.apps.versions import is_published
rows = (await session.execute(
select(WorkflowVersion.id, WorkflowVersion.workflow_id, WorkflowVersion.version, WorkflowVersion.level,
WorkflowVersion.retired_at, Workflow.published_version)
.join(Workflow, Workflow.id == WorkflowVersion.workflow_id))).all()
latest: dict[str, tuple[int, str]] = {}
for vid, wid, version, level, retired, current in rows:
probe = _VersionProbe(version, level, retired)
if is_published(probe, current=current) and version > latest.get(wid, (0, ""))[0]:
latest[wid] = (version, vid)
if not latest:
return []
ids = [vid for _, vid in latest.values()]
versions = (await session.execute(select(WorkflowVersion).where(WorkflowVersion.id.in_(ids)))).scalars().all()
out = [c for v in versions if (c := caliber_of(v.workflow_id, v)) is not None]
return sorted(out, key=lambda c: (c.workflow_id, c.version))
@dataclass(frozen=True)
class _VersionProbe:
"""is_published 只看这三样:不必为了判断把整张图读出来。"""
version: int
level: str | None
retired_at: Any
def _frozen(schema_artifact: Any) -> tuple[dict[str, Any] | None, dict[str, str]]:
"""查询当时冻结的表结构(explain_sql_diff 的 schema)和列的中文名(同一份快照里冻结的数据目录)。读不出时都没有。"""
from app.core.artifact_store import load
from app.data.sql_semantics import catalog_labels
if not isinstance(schema_artifact, str) or not schema_artifact:
return None, {}
try:
content = load(schema_artifact)
except ValueError:
return None, {}
if not isinstance(content, dict) or not isinstance(content.get("tables"), dict):
return None, {}
frozen = content.get("catalog") if isinstance(content.get("catalog"), dict) else {}
notes = {table: entry.get("notes") for table, entry in frozen.items()
if isinstance(entry, dict) and isinstance(entry.get("notes"), dict)}
return {"tables": content["tables"]}, catalog_labels(notes)
def read_queries(sealed: Any, kinds: dict[str, str | None]) -> list[Query]:
"""封存范围内的查询快照 → Query(读盘、复验哈希,放在线程里调)。合并查询不比:它的数都来自它的输入。"""
from app.core.artifact_store import load
from app.engine.merge_query import MERGE_SOURCE
out: list[Query] = []
for artifact, info in sealed.queries.items():
node_id = (info or {}).get("node_id")
try:
snap = load(artifact)
except ValueError:
out.append(Query(artifact, node_id, str((info or {}).get("source") or ""), problem=SNAPSHOT_TAMPERED))
continue
if not isinstance(snap, dict):
out.append(Query(artifact, node_id, str((info or {}).get("source") or ""), problem=SNAPSHOT_MISSING))
continue
source = str(snap.get("source") or (info or {}).get("source") or "")
if source == MERGE_SOURCE:
continue
sql = snap.get("sql")
if not isinstance(sql, str) or not sql.strip():
out.append(Query(artifact, node_id, source, problem=NO_SQL))
continue
schema, names = _frozen(snap.get("schema_artifact"))
columns = [str(c) for c in snap.get("columns") or []] if isinstance(snap.get("columns"), list) else None
out.append(Query(artifact, node_id, source, sql, columns, _table_keys(sql), kinds.get(source), schema, names))
return out
# --------------------------------------------------------------------------
# 比对(纯计算,在线程里跑)
# --------------------------------------------------------------------------
def _candidate(caliber: Caliber, fetch: Fetch, m: Any) -> dict[str, Any]:
"""一份口径取数的比对结果,进工件。"""
out: dict[str, Any] = {
"caliber": caliber.key, "workflow_id": caliber.workflow_id, "version": caliber.version,
"level": caliber.level, "node_id": fetch.node_id,
"status": "proven" if m.equivalent else "related", "comparable": m.comparable,
"via": m.level, "binding": dict(m.binding), "unbound": list(m.unbound),
}
if m.equivalent:
out["columns"] = [dict(c) for c in m.columns] if m.columns is not None else None
elif m.diff is not None:
out["diff"] = {"items": [d.as_dict() for d in m.diff.items], "fallback": m.diff.fallback,
"text_diff": list(m.diff.text_diff)}
if m.reason:
out["reason"] = m.reason
return out
def _skipped(caliber: Caliber, fetch: Fetch, reason: str) -> dict[str, Any]:
return {"caliber": caliber.key, "workflow_id": caliber.workflow_id, "version": caliber.version,
"level": caliber.level, "node_id": fetch.node_id, "status": "skipped", "reason": reason}
def compare_queries(queries: list[Query], calibers: list[Caliber], *, deadline: float | None = None
) -> list[dict[str, Any]]:
"""每条查询和预筛出来的口径取数逐个比(match_template),给出 [{artifact, node_id, source, tables, status, reason?,
candidates}]。status:proven(和某一份等价)/ related(有同源同表的口径,都不等价)/ none(没有相关的口径)/ skipped
(没比完:超时、解析失败、快照读不出)。纯计算,从不抛异常。评测逐题比 SQL 也调它。"""
from app.data.sql_semantics import match_template, normalize_sql
out: list[dict[str, Any]] = []
for query in queries:
row: dict[str, Any] = {"artifact": query.artifact, "node_id": query.node_id, "source": query.source,
"tables": sorted(query.tables)}
if query.problem:
out.append({**row, "status": "skipped", "reason": query.problem, "candidates": []})
continue
pool = [(c, f) for c in calibers for f in c.fetches if f.source == query.source and f.tables & query.tables]
if not pool:
out.append({**row, "status": "none", "candidates": []})
continue
if normalize_sql(query.sql, query.dialect).parse_error:
out.append({**row, "status": "skipped", "reason": PARSE_FAILED, "candidates": []})
continue
candidates: list[dict[str, Any]] = []
for caliber, fetch in pool[:MAX_CANDIDATES]:
if deadline is not None and time.monotonic() > deadline:
candidates.append(_skipped(caliber, fetch, TIMEOUT))
continue
try:
m = match_template(fetch.sql, list(caliber.fields), query.sql, query.dialect, schema=query.schema,
names=query.names, columns=query.columns, deadline=deadline)
except Exception: # noqa: BLE001 - match_template 自己不抛;万一抛了,这一份照实记没比
logger.exception("口径比对出错(运行查询 %s,口径 %s)", query.artifact[:12], caliber.key)
candidates.append(_skipped(caliber, fetch, "比对出错"))
continue
candidates.append(_candidate(caliber, fetch, m))
statuses = {c["status"] for c in candidates}
status = "proven" if "proven" in statuses else "skipped" if "skipped" in statuses else "related"
if len(pool) > MAX_CANDIDATES:
row["note"] = MORE_CANDIDATES.format(n=len(pool) - MAX_CANDIDATES, max=MAX_CANDIDATES)
if status == "skipped":
row["reason"] = next(c.get("reason") for c in candidates if c["status"] == "skipped")
out.append({**row, "status": status, "candidates": candidates})
return out
def compute(sealed: Any, calibers: list[Caliber], kinds: dict[str, str | None], *, timeout: float) -> dict[str, Any]:
"""一次运行的比对工件内容(见 ledger/store 的模块说明)。在线程里调:读快照、比 SQL、标数都是同步的活。"""
from app.api.evidence import _choose, _reports
from app.core.artifact_store import load
deadline = time.monotonic() + timeout
queries = read_queries(sealed, kinds)
if not sealed.trusted:
results = [{"artifact": q.artifact, "node_id": q.node_id, "source": q.source, "tables": sorted(q.tables),
"status": "skipped", "reason": SEAL_BROKEN, "candidates": []} for q in queries]
else:
results = compare_queries(queries, calibers, deadline=deadline)
used = {c["caliber"] for q in results for c in q["candidates"]}
considered = [c for c in calibers if c.key in used]
by_key = {c.key: {"level": c.level, "fields": c.summary()["fields"], "expressions": list(c.expressions)}
for c in considered}
by_artifact = {q["artifact"]: q for q in results}
sealed_cards = {artifact for _, artifact in sealed.ledger}
def load_card(artifact: str) -> dict[str, Any] | None:
if artifact not in sealed_cards:
return None
try:
card = load(artifact)
except ValueError:
return None
return card if isinstance(card, dict) else None
reports: dict[str, Any] = {}
primary: str | None = None
found = _reports(sealed)
for report in found:
if report.doc is not None and report.hash_ok is True:
reports[report.node_id] = {"labels": ledger_labels.label_doc(report.doc, by_artifact, by_key, load_card)}
if found:
primary = _choose(found, None, sealed).node_id
whole = ledger_labels.whole_of((reports.get(primary) or {}).get("labels") or {}, by_key) if primary in reports \
else {"equivalent": False, "reason": "这次运行没有可读的报告", "counts": {k: 0 for k in ledger_labels.KINDS}}
return {"format": store.FORMAT, "algo": store.ALGO, "run_id": sealed.run_id,
"manifest_hash": sealed.manifest_hash, "manifest_seq": sealed.run_manifest_seq,
"calibers": [c.summary() for c in considered], "queries": results, "reports": reports,
"primary": primary, "whole": whole}
def _timed_out(sealed: Any) -> dict[str, Any]:
"""线程过了时限还没回来:整次记 skipped(每条查询都写比对超时),标数都是没比对。"""
rows = [{"artifact": a, "node_id": (i or {}).get("node_id"), "source": str((i or {}).get("source") or ""),
"tables": [], "status": "skipped", "reason": TIMEOUT, "candidates": []} for a, i in sealed.queries.items()]
return {"format": store.FORMAT, "algo": store.ALGO, "run_id": sealed.run_id, "manifest_hash": sealed.manifest_hash,
"manifest_seq": sealed.run_manifest_seq, "calibers": [], "queries": rows, "reports": {}, "primary": None,
"whole": {"equivalent": False, "reason": TIMEOUT, "counts": {k: 0 for k in ledger_labels.KINDS}}}
# --------------------------------------------------------------------------
# 调度:后台、限时、限并发
# --------------------------------------------------------------------------
_POOL: ThreadPoolExecutor | None = None
#: 比对的空位,按事件循环分开(同 scheduler/auto_compare 的 _SLOTS):测试每个用例一个新循环,跨循环用同一个信号量会报错
_SLOTS: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Semaphore]" = weakref.WeakKeyDictionary()
_background: set[asyncio.Task[Any]] = set()
#: 正在比的运行:同一次运行不重复排队(封存后、补比、按需几处可能同时要)
_pending: set[str] = set()
def _pool() -> ThreadPoolExecutor:
global _POOL
if _POOL is None:
_POOL = ThreadPoolExecutor(max_workers=CONCURRENCY, thread_name_prefix="ledger-match")
return _POOL
def _slot() -> asyncio.Semaphore:
loop = asyncio.get_running_loop()
if loop not in _SLOTS:
_SLOTS[loop] = asyncio.Semaphore(CONCURRENCY)
return _SLOTS[loop]
async def _kinds(session: Any) -> dict[str, str | None]:
"""数据源名 → 类型(取方言)。只读应用库,不连业务库。"""
return {name: kind for name, kind in (await session.execute(select(DataSource.name, DataSource.kind))).all()}
async def match_run(run_id: str, *, trigger: str = "sealed", timeout: float = TIMEOUT_S) -> str | None:
"""比对一次运行,存工件、追加 ledger.matched;返回工件 id。不比对(不是探索运行、没成功、没封存、没有查询)、比对期间
运行换了封存时返回 None。比对本身出的问题都记成 skipped,不抛。"""
from app.api.evidence import _sealed
async with SessionLocal() as session:
run = await session.get(Run, run_id)
if ineligible(run) is not None:
return None
sealed = await _sealed(run_id)
if not sealed.queries:
return None
async with _slot():
async with SessionLocal() as session:
calibers = await load_calibers(session)
kinds = await _kinds(session)
loop = asyncio.get_running_loop()
work = loop.run_in_executor(_pool(), functools.partial(compute, sealed, calibers, kinds, timeout=timeout))
try:
content = await asyncio.wait_for(work, timeout=timeout + GRACE_S)
except TimeoutError:
logger.warning("运行 %s 的口径比对超过 %g 秒仍未完成,记为没比完", run_id, timeout + GRACE_S)
content = _timed_out(sealed)
return await store.save(run_id, content, seal=(sealed.run_manifest_seq, sealed.manifest_hash), trigger=trigger)
def schedule(run_id: str, *, trigger: str = "sealed") -> bool:
"""放到后台比对(runner 封存之后、按需)。已经在排队的同一次运行不再排;返回排没排上。"""
if run_id in _pending:
return False
_pending.add(run_id)
task = asyncio.create_task(_guarded(run_id, trigger))
_background.add(task)
task.add_done_callback(_background.discard)
return True
def pending(run_id: str) -> bool:
return run_id in _pending
async def _guarded(run_id: str, trigger: str) -> None:
try:
await match_run(run_id, trigger=trigger)
except asyncio.CancelledError:
raise
except Exception: # noqa: BLE001 - 比对是辅助步骤:出错只记日志,不影响运行和答案
logger.exception("运行 %s 的口径比对出错", run_id)
finally:
_pending.discard(run_id)
async def backfill(*, days: int = BACKFILL_DAYS, limit: int = BACKFILL_LIMIT) -> int:
"""最近 days 天封存了、一次都没比过的探索运行,从新到旧逐个补比(trigger=backfill),返回补了几个。"""
since = datetime.now(timezone.utc) - timedelta(days=days)
async with SessionLocal() as session:
ids = [rid for (rid,) in (await session.execute(
select(Run.id).where(Run.status == "succeeded", Run.run_class != "formal",
Run.origin_kind.in_(EXPLORE_ORIGINS), Run.manifest_hash.is_not(None),
Run.finished_at >= since)
.order_by(Run.finished_at.desc()).limit(limit))).all()]
done = await store.matched_run_ids(session, ids)
count = 0
for run_id in ids:
if run_id in done or run_id in _pending:
continue
_pending.add(run_id)
try:
if await match_run(run_id, trigger="backfill"):
count += 1
except asyncio.CancelledError:
raise
except Exception: # noqa: BLE001 - 一次补比出错不挡后面的
logger.exception("运行 %s 的口径补比出错", run_id)
finally:
_pending.discard(run_id)
if count:
logger.info("启动时补比了 %d 次运行的口径", count)
return count
def start_backfill() -> bool:
"""启动时在后台补比(engine/runner 的 setup 调)。和调度循环同一个开关(见模块说明);返回起没起。"""
from app.core.config import settings
if not settings.scheduler:
return False
task = asyncio.create_task(_backfill_guarded())
_background.add(task)
task.add_done_callback(_background.discard)
return True
async def _backfill_guarded() -> None:
try:
await backfill()
except asyncio.CancelledError:
raise
except Exception: # noqa: BLE001
logger.exception("启动时的口径补比出错,服务照常运行")
def _mine() -> list[asyncio.Task[Any]]:
loop = asyncio.get_running_loop()
return [t for t in _background if t.get_loop() is loop]
async def drain() -> None:
"""等这个事件循环里的后台比对都做完(测试用)。"""
while tasks := _mine():
await asyncio.gather(*tasks, return_exceptions=True)
async def shutdown() -> None:
"""关停:这个事件循环里还没做完的比对取消掉(没做完的不记,下次启动补比会再比)。"""
tasks = _mine()
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
__all__ = ["BACKFILL_DAYS", "CONCURRENCY", "EXPLORE_ORIGINS", "TIMEOUT_S", "Caliber", "Fetch", "Query", "backfill",
"caliber_of", "compare_queries", "compute", "drain", "ineligible", "load_calibers", "match_run", "pending",
"read_queries", "schedule", "shutdown", "start_backfill", "wanted"]
+289
View File
@@ -0,0 +1,289 @@
"""口径账本自身给出的统计 E1–E4(docs/design/next-round/02-口径账本.md 做法 10,总方案 X2:只有这一份实现)。
全部从比对工件(kind=caliber_match)、运行、查询快照和审计现算,不另存:另建统计表或埋点,数据和账本就成了两份。评测页
「口径命中与重复度」直接调 compute_stats(同一个函数、同一套口径),评测细案的 eval/signals.py 撤销。
范围:since / until 按运行的发起时间;origins 缺省是比对的那几种探索运行(ledger/match.EXPLORE_ORIGINS);只算「计入统计」的
运行(engine/origin.counted_runs:评测、对账、召回这些系统派生的不算)。run_ids 给了就只算这几次(评测传它自己挑的运行,
这时不再按来源和计入统计筛——调用方点名的就是要算的)。
- E1 等价命中:比对过的查询里 proven / composed / related / none / skipped 各几条(skipped 单独列,不混进 none),有候选口径
的查询里各自的比率;证明等价的绑定是在哪一级配上的;没证明的原因(不可比、参数绑不上、配上了但不等价);不等价的差异类型
分布(time / filter / aggregate / join / other / parse,每份口径的差异每类只计一次);逐个数的标签各几个、整题等价几次。
- E2 对账:对账运行(origin=reconcile)发起、完成、失败各几次;相关口径的差异说明里兜底说明的占比。「说明有用 / 看不懂」的
反馈要等对账面板(第 3 波)记成对账运行上的封存后事件,这之前照实写没有数据。
- E3 重复度:问数据运行里执行过的查询按(数据源、normalize_sql 文本)分组,出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS
段对话的算重复的形状,给出这样的形状几个、有重复形状的运行占几成;问题聚类(app/chat/questions:全站唯一的聚类,只取问题
原文、不含 history、排除系统派生的运行)里 ≥ REPEAT_RUNS 个问题的类占几成。认证队列的转化(固化、发布、驳回)在第 3 波。
- E4 召回:每个口径版本被用了多少次(比对里等价已证明的运行数、这一版的正式运行数)。每次发布召回了几组、其中数变了的比例要等
召回(第 4 波),这之前照实写没有数据,只给召回运行的次数。
"""
from __future__ import annotations
import asyncio
from collections import Counter
from collections.abc import Collection, Sequence
from datetime import datetime
from typing import Any
from sqlalchemy import func, select
from app.ledger import labels as ledger_labels
from app.ledger import store
#: 重复的门槛:同一个形状出现在几次运行里、来自几段对话(同认证队列的进队门槛)
REPEAT_RUNS = 3
REPEAT_CONVERSATIONS = 2
#: 差异类型怎么归到 E1 的几类
ASPECT_GROUP = {"time": "time", "time_range": "time", "filter": "filter", "aggregate": "aggregate",
"group_by": "aggregate", "join": "join", "tables": "join", "parse": "parse"}
DIFF_GROUPS = ("time", "filter", "aggregate", "join", "other", "parse")
STATUSES = ("proven", "composed", "related", "none", "skipped")
#: 没有数据时的说法(这一块的功能还没做)
NOT_YET_RECONCILE = "对账面板的反馈在第 3 波接上,目前没有数据"
NOT_YET_CERTIFY = "认证队列在第 3 波接上,目前没有数据"
NOT_YET_RECALL = "召回在第 4 波接上,目前没有数据"
#: 聚类最多取几个问题(同治理接口的上限)
QUESTION_LIMIT = 500
def _ratio(part: int, whole: int) -> float | None:
return round(part / whole, 4) if whole else None
def _scope(stmt: Any, *, since: datetime | None, until: datetime | None, origins: Sequence[str] | None,
run_ids: Collection[str] | None) -> Any:
from app.db.models import Run
from app.engine.origin import counted_runs
from app.ledger.match import EXPLORE_ORIGINS
if run_ids is not None:
return stmt.where(Run.id.in_(list(run_ids)))
stmt = stmt.where(counted_runs(), Run.origin_kind.in_(list(origins or EXPLORE_ORIGINS)))
if since is not None:
stmt = stmt.where(Run.created_at >= since)
if until is not None:
stmt = stmt.where(Run.created_at < until)
return stmt
async def _matches(session: Any, **scope: Any) -> list[dict[str, Any]]:
"""范围里每次运行最新的一份比对工件的内容(按工件引用的时间取最新;读不出的跳过)。"""
from app.core.artifact_store import load
from app.db.models import Artifact, Run
stmt = select(Artifact.id, Artifact.run_id, Artifact.created_at).join(Run, Run.id == Artifact.run_id) \
.where(Artifact.kind == store.KIND)
latest: dict[str, tuple[Any, str]] = {}
for artifact, run_id, at in (await session.execute(_scope(stmt, **scope))).all():
if run_id not in latest or (at and latest[run_id][0] and at > latest[run_id][0]):
latest[run_id] = (at, artifact)
def read() -> list[dict[str, Any]]:
out = []
for _, artifact in latest.values():
try:
content = load(artifact)
except ValueError:
continue
if isinstance(content, dict) and content.get("format") == store.FORMAT:
out.append(content)
return out
return await asyncio.to_thread(read)
def e1_of(contents: list[dict[str, Any]]) -> dict[str, Any]:
"""E1 等价命中(见模块说明)。纯函数:评测拿自己的比对结果也能算。"""
status = Counter({s: 0 for s in STATUSES})
with_candidates = Counter({s: 0 for s in STATUSES})
via: Counter[str] = Counter()
unproven: Counter[str] = Counter()
diffs = Counter({g: 0 for g in DIFF_GROUPS})
labels = Counter({k: 0 for k in ledger_labels.KINDS})
whole = 0
queries = 0
for content in contents:
for query in content.get("queries") or []:
queries += 1
s = str(query.get("status") or "")
status[s] += 1
if query.get("candidates"):
with_candidates[s] += 1
for cand in query.get("candidates") or []:
if cand.get("status") == "proven":
via[str(cand.get("via") or "parsed")] += 1
continue
if cand.get("status") == "skipped":
continue
if not cand.get("comparable", True):
unproven["不可比:" + str(cand.get("reason") or "")] += 1
elif cand.get("reason"):
unproven[str(cand["reason"])] += 1
else:
unproven["参数配上了,但和口径不等价"] += 1
seen = {ASPECT_GROUP.get(str(item.get("aspect")), "other")
for item in (cand.get("diff") or {}).get("items") or []}
for group in seen:
diffs[group] += 1
for report in (content.get("reports") or {}).values():
for label in (report.get("labels") or {}).values():
labels[str(label.get("kind"))] += 1
if (content.get("whole") or {}).get("equivalent"):
whole += 1
candidates = sum(with_candidates.values())
return {
"runs": len(contents), "queries": queries, "status": dict(status),
"with_candidates": candidates,
"rates": {s: _ratio(with_candidates[s], candidates) for s in STATUSES if s != "none"},
"proven_via": dict(via), "unproven_reasons": dict(unproven.most_common(20)), "diff_aspects": dict(diffs),
"labels": dict(labels), "whole_equivalent": whole,
}
async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dict[str, Any]:
from app.db.models import Run
stmt = select(Run.status, func.count()).where(Run.origin_kind == "reconcile")
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"])
by_status = {s: int(n) for s, n in (await session.execute(stmt.group_by(Run.status))).all()}
explained = fallback = 0
for content in contents:
for query in content.get("queries") or []:
for cand in query.get("candidates") or []:
if cand.get("status") == "related" and isinstance(cand.get("diff"), dict):
explained += 1
fallback += bool(cand["diff"].get("fallback"))
return {"launched": sum(by_status.values()), "succeeded": by_status.get("succeeded", 0),
"failed": by_status.get("failed", 0), "explained": explained, "fallback": fallback,
"fallback_rate": _ratio(fallback, explained),
"feedback": {"available": False, "note": NOT_YET_RECONCILE}}
#: 查询快照 → (数据源, 形状)。工件按内容寻址、不会变,记下来不会过期
_SHAPES: dict[str, tuple[str, str] | None] = {}
_SHAPES_MAX = 20000
def _shape_of(artifact: str, kinds: dict[str, str | None]) -> tuple[str, str] | None:
from app.core.artifact_store import load
from app.data.sql_semantics import normalize_sql
if artifact in _SHAPES:
return _SHAPES[artifact]
try:
snap = load(artifact)
except ValueError:
snap = None
found: tuple[str, str] | None = None
if isinstance(snap, dict) and isinstance(snap.get("sql"), str) and snap["sql"].strip():
source = str(snap.get("source") or "")
found = (source, normalize_sql(snap["sql"], kinds.get(source)).text)
if len(_SHAPES) < _SHAPES_MAX:
_SHAPES[artifact] = found
return found
async def _e3(session: Any, *, principal: Any, **scope: Any) -> dict[str, Any]:
from app.chat.questions import cluster_questions, embedder, recent_questions
from app.data.catalog import QUERY_SNAPSHOT_KIND
from app.db.models import Artifact, ConversationTurn, DataSource, Run
chat_scope = {**scope, "origins": ["chat"]}
runs = (await session.execute(_scope(select(Run.id, Run.origin), **chat_scope))).all()
run_ids = [rid for rid, _ in runs]
conversation: dict[str, str] = {rid: str((origin or {}).get("conversation_id")) for rid, origin in runs
if isinstance(origin, dict) and origin.get("conversation_id")}
if run_ids:
for rid, cid in (await session.execute(select(ConversationTurn.run_id, ConversationTurn.conversation_id)
.where(ConversationTurn.run_id.in_(run_ids)))).all():
conversation.setdefault(rid, cid)
snaps = (await session.execute(select(Artifact.id, Artifact.run_id).where(
Artifact.kind == QUERY_SNAPSHOT_KIND, Artifact.run_id.in_(run_ids)))).all() if run_ids else []
kinds = {name: kind for name, kind in (await session.execute(select(DataSource.name, DataSource.kind))).all()}
shaped = await asyncio.to_thread(lambda: [(rid, _shape_of(aid, kinds)) for aid, rid in snaps])
groups: dict[tuple[str, str], tuple[set[str], set[str]]] = {}
for rid, shape in shaped:
if shape is None:
continue
found_runs, convs = groups.setdefault(shape, (set(), set()))
found_runs.add(rid)
convs.add(conversation.get(rid) or f"run:{rid}")
repeated = [shape for shape, (rs, cs) in groups.items() if len(rs) >= REPEAT_RUNS and len(cs) >= REPEAT_CONVERSATIONS]
with_queries = {rid for rid, shape in shaped if shape is not None}
in_repeated = {rid for shape in repeated for rid in groups[shape][0]}
questions = await recent_questions(session, principal, limit=QUESTION_LIMIT)
if scope.get("since") is not None:
questions = [q for q in questions if q.at is None or q.at >= scope["since"]]
if scope.get("until") is not None:
questions = [q for q in questions if q.at is None or q.at < scope["until"]]
clusters = await cluster_questions(questions) if questions else []
clustered = sum(c.count for c in clusters if c.count >= REPEAT_RUNS)
return {
"chat_runs": len(run_ids), "runs_with_queries": len(with_queries), "shapes": len(groups),
"repeated_shapes": len(repeated), "runs_in_repeated_shapes": len(in_repeated),
"repeated_share": _ratio(len(in_repeated), len(with_queries)),
"threshold": {"runs": REPEAT_RUNS, "conversations": REPEAT_CONVERSATIONS},
"questions": len(questions), "clusters": len(clusters), "questions_in_repeated_clusters": clustered,
"cluster_share": _ratio(clustered, len(questions)), "embedder": embedder(),
"certify": {"available": False, "note": NOT_YET_CERTIFY},
}
async def _e4(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dict[str, Any]:
from app.db.models import Run
from app.engine.origin import counted_runs
proven: Counter[str] = Counter()
for content in contents:
keys = {c.get("caliber") for q in content.get("queries") or [] for c in q.get("candidates") or []
if c.get("status") == "proven" and c.get("caliber")}
for key in keys:
proven[str(key)] += 1
stmt = select(Run.workflow_id, Run.version, func.count()).where(
Run.run_class == "formal", Run.workflow_id.is_not(None), Run.version.is_not(None))
stmt = stmt.where(Run.id.in_(list(scope["run_ids"]))) if scope.get("run_ids") is not None \
else stmt.where(counted_runs())
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"])
formal = {ledger_labels.caliber_key(w, v): int(n)
for w, v, n in (await session.execute(stmt.group_by(Run.workflow_id, Run.version))).all()}
recalls = int(await session.scalar(select(func.count()).select_from(Run).where(Run.origin_kind == "recall")) or 0)
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}}
async def compute_stats(session: Any, *, since: datetime | None = None, until: datetime | None = None,
origins: Sequence[str] | None = None, run_ids: Collection[str] | None = None,
principal: Any = None) -> dict[str, Any]:
"""账本统计 E1–E4(见模块说明)。session 只读;principal 是看统计的人(问题聚类去掉他读不了的运行),不给按当前请求的身份。
返回 {window, algo, e1, e2, e3, e4};各块的键见 e1_of、_e2、_e3、_e4。比率在分母为 0 时为 None(不写成 0)。"""
from app.api.deps import current_principal
scope = {"since": since, "until": until, "origins": list(origins) if origins else None,
"run_ids": list(run_ids) if run_ids is not None else None}
contents = await _matches(session, **scope)
return {
"window": {"since": since.isoformat() if since else None, "until": until.isoformat() if until else None,
"origins": scope["origins"], "runs": scope["run_ids"]},
"algo": store.ALGO,
"e1": e1_of(contents),
"e2": await _e2(session, contents, **scope),
"e3": await _e3(session, principal=principal or current_principal(), **scope),
"e4": await _e4(session, contents, **scope),
}
__all__ = ["ASPECT_GROUP", "DIFF_GROUPS", "REPEAT_CONVERSATIONS", "REPEAT_RUNS", "STATUSES", "compute_stats", "e1_of"]
+144
View File
@@ -0,0 +1,144 @@
"""口径账本的记法(docs/design/next-round/02-口径账本.md 做法 2,拍板 L8):比对结果存成内容寻址工件,不建新表。
一次比对 = 一份工件 kind=caliber_match(格式 agentlab.caliber_match/1)+ 一条封存之后追加的事件 ledger.matched
{artifact, manifest_hash, algo, counts, trigger}(engine/run_append 的 sealed 方式,同按需裁判的 evidence.judged):
- 工件内容只放比对本身:运行 id、对着哪一次封存(manifest_hash / manifest_seq)、算法版本、考虑过的口径版本(工作流、版本、
级别、graph_hash)、每条查询的状态和绑定、差异条目、列映射、每个数的标签、整题判断。不放时间、耗时、工作流名(名字会改,
界面读的时候再取):同一份封存、同一个算法版本、同一批口径,再比一次得到的工件一字不差、哈希一样。
- 工件引用的 meta 记 {calibers: [等价已证明的「工作流@版本」], related: [相关的], counts}:召回按旧版本反查、统计按种类数,
都只扫 artifacts(kind 有索引),不逐份读内容。
- 事件说的是「这一次封存的这份比对」:读的时候只认 manifest_hash 等于运行现在封存的那几条,取最后一条(之后口径有变、补比
过的,以最新的为准)。工件读不出、被改过、不是这次运行这次封存的,照实说读不出,不拿它标数。
trigger 是谁触发的:sealed(运行封存后马上比的)、backfill(启动时补比)、on_demand(打开账本时按需比的)。后两种是事后按
那时的口径回看的,界面据此写「按现在的口径回看」。它只在事件里,不进工件内容(不然同一份比对因为触发方式不同哈希就不同)。
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from sqlalchemy import select
from app.core.events import EventType
from app.db.base import SessionLocal
from app.db.models import Run, RunEvent
#: 工件的种类和内容格式
KIND = "caliber_match"
FORMAT = "agentlab.caliber_match/1"
#: 比对算法的版本:绑定、证明、标签的规则改了就加一,界面和统计据此分得清是哪一版算法比的
ALGO = "1"
#: 触发方式(见模块说明)
TRIGGERS = ("sealed", "backfill", "on_demand")
@dataclass(frozen=True)
class Recorded:
"""一次运行现在这次封存的最新一份比对。content 读不出时为 None,problem 写为什么。"""
artifact: str
seq: int
ts: Any
trigger: str
algo: str | None
content: dict[str, Any] | None
problem: str | None = None
@property
def retro(self) -> bool:
"""事后按那时的口径回看的(启动时补比、打开时按需比的),不是运行一封存就比的。"""
return self.trigger != "sealed"
#: 读不出比对工件时的说法
UNREADABLE = "比对记录读不出来(文件缺失)"
TAMPERED = "比对记录与哈希不一致,疑似被修改,不能用来标口径"
FOREIGN = "比对记录不属于这次运行的这次封存,不能用来标口径"
def counts_of(queries: list[dict[str, Any]]) -> dict[str, int]:
"""每种查询状态各几条(proven / composed / related / none / skipped),0 也写。"""
out = {status: 0 for status in ("proven", "composed", "related", "none", "skipped")}
for query in queries:
status = str(query.get("status") or "")
if status in out:
out[status] += 1
return out
def meta_of(content: dict[str, Any]) -> dict[str, Any]:
"""工件引用的 meta:{calibers: [等价已证明的「工作流@版本」], related: [相关的], counts}。"""
proven: list[str] = []
related: list[str] = []
for query in content.get("queries") or []:
for cand in query.get("candidates") or []:
key = str(cand.get("caliber") or "")
if not key:
continue
bucket = proven if cand.get("status") == "proven" else related
if key not in bucket:
bucket.append(key)
related = [key for key in related if key not in proven]
return {"calibers": sorted(proven), "related": sorted(related), "counts": counts_of(content.get("queries") or [])}
async def save(run_id: str, content: dict[str, Any], *, seal: tuple[int | None, str | None],
trigger: str) -> str | None:
"""存比对工件、追加 ledger.matched。返回工件 id;运行在比对期间换了封存(没抢到)时只存了工件、没追加事件,返回 None
(不算这次运行的比对:读的时候只认事件)。"""
from app.core.artifact_store import put_json
from app.engine.run_append import append_events
if trigger not in TRIGGERS:
raise ValueError(f"不认识的比对触发方式:{trigger}")
meta = meta_of(content)
artifact = await put_json(content, kind=KIND, run_id=run_id, meta=meta)
written = await append_events(run_id, [(EventType.LEDGER_MATCHED, None, {
"artifact": artifact, "manifest_hash": content.get("manifest_hash"), "algo": content.get("algo"),
"counts": meta["counts"], "trigger": trigger,
})], require="sealed", seal=seal)
return artifact if written else None
async def latest(run_id: str) -> Recorded | None:
"""这次运行现在这次封存的最新一份比对;没比对过(或者比对的是以前的封存)为 None。"""
from app.core.artifact_store import load
async with SessionLocal() as session:
manifest = await session.scalar(select(Run.manifest_hash).where(Run.id == run_id))
if not manifest:
return None
rows = (await session.execute(
select(RunEvent.seq, RunEvent.ts, RunEvent.data).where(
RunEvent.run_id == run_id, RunEvent.type == EventType.LEDGER_MATCHED)
.order_by(RunEvent.seq.desc()))).all()
for seq, ts, data in rows:
if not isinstance(data, dict) or data.get("manifest_hash") != manifest or not data.get("artifact"):
continue
artifact = str(data["artifact"])
trigger = str(data.get("trigger") or "sealed")
try:
content = load(artifact)
except ValueError:
return Recorded(artifact, seq, ts, trigger, data.get("algo"), None, TAMPERED)
if not isinstance(content, dict):
return Recorded(artifact, seq, ts, trigger, data.get("algo"), None, UNREADABLE)
if content.get("format") != FORMAT or content.get("run_id") != run_id \
or content.get("manifest_hash") != manifest:
return Recorded(artifact, seq, ts, trigger, data.get("algo"), None, FOREIGN)
return Recorded(artifact, seq, ts, trigger, data.get("algo"), content)
return None
async def matched_run_ids(session: Any, run_ids: list[str]) -> set[str]:
"""这几次运行里有过比对事件的(不管是哪一次封存的):启动补比据此跳过比过的。"""
if not run_ids:
return set()
rows = await session.execute(select(RunEvent.run_id).where(
RunEvent.run_id.in_(run_ids), RunEvent.type == EventType.LEDGER_MATCHED).distinct())
return {r for (r,) in rows}
__all__ = ["ALGO", "FORMAT", "KIND", "TRIGGERS", "Recorded", "counts_of", "latest", "matched_run_ids", "meta_of", "save"]
+166
View File
@@ -0,0 +1,166 @@
"""口径账本交给界面的样子(docs/design/next-round/02-口径账本.md 做法 3):GET /api/runs/{run_id}/ledger、证据片段接口的
ledger 键、报告导出的片段面板,三处读同一份比对工件、按同一套遮罩给。
遮罩按读的那一刻重滤(同证据面板,api/evidence._Masks):绑定的值是这次运行查到过的遮罩值时不给值(也就不能拿它代填出具),
差异条目里的 SQL 片段走 _Shown 同一套换法。工作流名按现在的取(工件里不存名字)。
界面层不放逻辑:标签、整题判断、能做的动作(这一波还没有出具和对账,actions 是空的)都由这里给。
"""
from __future__ import annotations
from typing import Any
from sqlalchemy import select
from app.db.base import SessionLocal
from app.db.models import Run, Workflow
from app.ledger import labels as ledger_labels
from app.ledger import match as ledger_match
from app.ledger import store
#: state:matched 比对过;pending 已排队按需比对;ineligible 这次运行不比对(reason 写为什么);unreadable 比对记录读不出
STATES = ("matched", "pending", "ineligible", "unreadable")
class _Hide:
"""按这次运行查到过的遮罩值遮:hide(值) 这个值给不给,shown(文字) 文字里的遮罩值换成「已遮罩」。"""
def __init__(self, secrets: Any) -> None:
from app.api.evidence import _Shown
self._secrets = secrets
self._shown = _Shown(secrets)
def hide(self, value: Any) -> bool:
if self._secrets is None or value is None or isinstance(value, bool):
return False
return bool(self._secrets.hit(value) or self._secrets.weak_hit(str(value)))
def shown(self, value: Any) -> Any:
return self._shown.text(value)
async def _hider(sealed: Any) -> _Hide:
from app.api.evidence import _Masks
return _Hide(await _Masks(sealed).run_secrets())
async def _names(ids: set[str]) -> dict[str, str]:
if not ids:
return {}
async with SessionLocal() as session:
rows = await session.execute(select(Workflow.id, Workflow.name).where(Workflow.id.in_(sorted(ids))))
return {wid: name for wid, name in rows.all()}
def _workflow_ids(content: dict[str, Any]) -> set[str]:
return {str(c.get("workflow_id")) for c in content.get("calibers") or [] if c.get("workflow_id")}
def _whole_view(whole: dict[str, Any], hider: _Hide, names: dict[str, str]) -> dict[str, Any]:
out = {k: v for k, v in whole.items() if k != "binding"}
if whole.get("equivalent"):
out["binding"] = ledger_labels.binding_view(whole.get("binding"), hider.hide)
out["workflow_name"] = names.get(str(whole.get("workflow_id") or ""))
out["level_text"] = ledger_labels.LEVEL_TEXT.get(str(whole.get("level") or ""), "")
return out
def _query_view(query: dict[str, Any], hider: _Hide, names: dict[str, str]) -> dict[str, Any]:
out = {k: query[k] for k in ("artifact", "node_id", "source", "tables", "status", "reason", "note") if k in query}
cands = []
for cand in query.get("candidates") or []:
item = ledger_labels.related_view(cand, hide=hider.hide, shown=hider.shown, names=names)
item["status"] = cand.get("status")
if cand.get("columns") is not None:
item["columns"] = cand["columns"]
cands.append(item)
out["candidates"] = cands
return out
def _labels_view(content: dict[str, Any], hider: _Hide, names: dict[str, str]) -> dict[str, Any]:
return {node: {"labels": {sid: ledger_labels.view_label(label, hide=hider.hide, shown=hider.shown, names=names)
for sid, label in (report.get("labels") or {}).items()}}
for node, report in (content.get("reports") or {}).items()}
def _stamp(recorded: store.Recorded) -> dict[str, Any]:
"""比对的来历:哪份工件、哪一版算法、谁触发的、什么时候追加的;事后补比的写明按现在的口径回看。"""
from datetime import datetime, timezone
at = datetime.fromtimestamp(recorded.ts, timezone.utc).isoformat(timespec="seconds") \
if isinstance(recorded.ts, (int, float)) else None
return {"artifact": recorded.artifact, "algo": recorded.algo, "trigger": recorded.trigger, "retro": recorded.retro,
"matched_at": at, "post_seal": True, "note": ledger_labels.POST_SEAL_NOTE,
**({"retro_note": ledger_labels.RETRO_NOTE} if recorded.retro else {})}
async def run_view(run_id: str) -> dict[str, Any]:
"""GET /api/runs/{run_id}/ledger 的答复:标签、整题判断、每条查询的比对、能做的动作。没比对过、能比的运行这时排上按需比对
(trigger=on_demand),答复 state=pending;不比的照实说为什么。"""
from app.api.coded import CodedHTTPException
from app.api.evidence import RUN_NOT_FOUND, _sealed
async with SessionLocal() as session:
run = await session.get(Run, run_id)
if run is None:
raise CodedHTTPException(404, "运行记录不存在", RUN_NOT_FOUND)
base: dict[str, Any] = {"run_id": run_id, "note": ledger_labels.POST_SEAL_NOTE}
recorded = await store.latest(run_id)
if recorded is None:
why = ledger_match.ineligible(run)
sealed = await _sealed(run_id) if why is None else None
if sealed is not None and not sealed.queries:
why = ledger_match.NO_QUERIES
if why is not None:
return {**base, "state": "ineligible", "reason": why}
ledger_match.schedule(run_id, trigger="on_demand")
return {**base, "state": "pending"}
if recorded.content is None:
return {**base, "state": "unreadable", "reason": recorded.problem, **_stamp(recorded)}
content = recorded.content
sealed = await _sealed(run_id)
hider = await _hider(sealed)
names = await _names(_workflow_ids(content))
return {
**base, "state": "matched", **_stamp(recorded),
"calibers": [{**{k: c.get(k) for k in ("caliber", "workflow_id", "version", "level", "graph_hash")},
"workflow_name": names.get(str(c.get("workflow_id") or "")),
"level_text": ledger_labels.LEVEL_TEXT.get(str(c.get("level") or ""), "")}
for c in content.get("calibers") or []],
"queries": [_query_view(q, hider, names) for q in content.get("queries") or []],
"reports": _labels_view(content, hider, names),
"primary": content.get("primary"),
"whole": _whole_view(content.get("whole") or {}, hider, names),
# 能做的动作(按口径正式出具、对账)在第 3 波;这一波照实给空的
"actions": [],
}
async def report_labels(sealed: Any, report_node: str) -> dict[str, Any] | None:
"""一份报告里每个数的标签(遮过的),给证据片段接口和报告导出:{labels: {片段 id: 标签}, 来历}。没比对过、比对记录读不出、
这份报告没有标签的为 None(片段接口就不加 ledger 键,老运行的返回一字不差)。"""
recorded = await store.latest(sealed.run_id)
if recorded is None or recorded.content is None:
return None
report = (recorded.content.get("reports") or {}).get(report_node)
if not isinstance(report, dict):
return None
hider = await _hider(sealed)
names = await _names(_workflow_ids(recorded.content))
return {"labels": {sid: ledger_labels.view_label(label, hide=hider.hide, shown=hider.shown, names=names)
for sid, label in (report.get("labels") or {}).items()}, **_stamp(recorded)}
async def segment_label(sealed: Any, report_node: str, segment_id: str) -> dict[str, Any] | None:
"""证据片段接口的 ledger 键:{label: 这个片段的标签, 比对的来历(artifact 是比对工件)};这个片段没有标签(不是数、没比对过)
为 None。"""
found = await report_labels(sealed, report_node)
if found is None or segment_id not in found["labels"]:
return None
return {"label": found["labels"][segment_id], **{k: v for k, v in found.items() if k != "labels"}}
__all__ = ["STATES", "report_labels", "run_view", "segment_label"]
+7 -1
View File
@@ -317,6 +317,7 @@ def build_report(
notes: Mapping[str, str],
nodes: Mapping[str, Mapping[str, Any]],
exported_at: str,
ledger: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
"""页面里的数据(report-data)。
@@ -329,8 +330,10 @@ def build_report(
- refs:片段 id → 引用键(点开这个片段看哪条链)
- notes:片段 id → 为什么是现在这个状态(api/evidence._note)
- nodes:节点 id → {label, type}
- ledger:口径账本给这份报告的标签(app/ledger/view.report_labels,已按遮罩给):{labels: {片段 id: 标签}, note, retro, …}。
面板最上面多「口径」一步,写明「封存后比对,不在封存范围内」。没比对过的运行不给,数据块里就没有这个键(和以前一字不差)
"""
return {
out = {
"schema": SCHEMA,
"meta": {
"run_id": run.get("id"), "workflow_id": run.get("workflow_id"),
@@ -347,6 +350,9 @@ def build_report(
"refs": dict(refs),
"notes": dict(notes),
}
if ledger:
out["ledger"] = {k: v for k, v in ledger.items() if k in ("labels", "note", "retro", "retro_note", "algo")}
return out
def _iso(value: Any) -> str | None:
+42
View File
@@ -70,6 +70,12 @@
hashBad: '工件哈希核对失败', hashNone: '未能读取工件',
note: '说明',
close: '关闭', openAria: (text) => `查看出处:${text}`,
// 口径(口径账本,封存后比对):来自口径 / 相关口径 / 这次的推理 / 没比对
ledger: '口径', ledgerFrom: (name, version, level) => `来自口径「${name}」v${version}${level ? `(${level})` : ''}`,
ledgerHow: { independent: '等价已证明', composed: '助手选用' },
ledgerRelated: '相关口径', ledgerReasoning: '这次的推理:没有相关的已发布口径', ledgerUnknown: '没有比对',
dBinding: '参数', dDiffers: '差在哪', dHow: '怎么认的', dColumnOf: '对应口径的列',
unnamed: '已删除的工作流',
}
const MASKED = '已遮罩'
@@ -396,6 +402,39 @@
entity: entityPart, quote: quotePart,
}
/** 口径一步(口径账本给的标签,服务端已按遮罩给):来自口径的写明哪一版、级别、怎么认的、参数;相关口径写差在哪;
* 这次的推理、没比对照实说。比对在封存之后,底下写明不在封存范围内(note),事后补比的另写按现在的口径回看 */
function ledgerPart(label, ledger) {
const body = []
const nameOf = (x) => x.workflow_name || T.unnamed
const bindingText = (items) => (Array.isArray(items) && items.length
? items.map((b) => `${b.field} = ${b.masked ? MASKED : valueText(b.value)}`).join(';') : NONE)
if (label.kind === 'caliber') {
body.push(h('p', {}, T.ledgerFrom(nameOf(label), label.version, label.level_text || '')))
body.push(kv([
[T.dHow, T.ledgerHow[label.how] || String(label.how || NONE)],
[T.dBinding, bindingText(label.binding)],
label.column ? [T.dColumnOf, String(label.column)] : null,
]))
} else if (label.kind === 'related') {
body.push(h('p', {}, T.ledgerRelated))
for (const item of Array.isArray(label.related) ? label.related : []) {
const lines = (((item.diff || {}).items) || []).map((d) => h('li', {}, String(d.text || '')))
body.push(h('p', { class: 'sub' }, T.ledgerFrom(nameOf(item), item.version, item.level_text || '')))
if (item.reason) body.push(h('p', { class: 'muted' }, String(item.reason)))
if (lines.length) body.push(h('p', { class: 'sub' }, T.dDiffers), h('ul', {}, lines))
}
if (label.reason) body.push(h('p', { class: 'muted' }, String(label.reason)))
} else if (label.kind === 'reasoning') {
body.push(h('p', {}, T.ledgerReasoning))
} else {
body.push(h('p', {}, T.ledgerUnknown + (label.reason ? `:${label.reason}` : '')))
}
const notes = [ledger.note, ledger.retro ? ledger.retro_note : null].filter(Boolean).join(' · ')
if (notes) body.push(h('p', { class: 'muted' }, notes))
return h('section', { class: 'part', 'data-step': 'ledger' }, h('h3', {}, T.ledger), h('div', { class: 'part-body' }, ...body))
}
// ------------------------------------------------------------------ 面板
function detailOf(seg, chain, note) {
@@ -407,6 +446,9 @@
[T.dNode, nodeLabel((view.data.meta.report || {}).node_id)],
])))
const parts = [head]
const ledger = isObj(view.data.ledger) ? view.data.ledger : null
const label = ledger ? own(ledger.labels, seg.dataset.seg) : undefined
if (isObj(label)) parts.push(ledgerPart(label, ledger))
if (chain && chain.length) {
for (const step of chain) {
const render = PARTS[step.step]
+140
View File
@@ -0,0 +1,140 @@
"""口径账本测试共用的虚构样本:一个店铺库、一份受管口径「周销售」、跑一次探索运行。数据全是编的。
- 店铺库 shop(SQLite):orders(id, region, status, amount, paid_at, phone),九单,状态有 paid / refunded / open。
- 口径「周销售」v1(受管):入口两个字段 start / end,取数节点 fetch 的 SQL 是
SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{{ input.start }}' AND paid_at < '{{ input.end }}',
口径卡 card 上一个指标 gmv = cell(nodes.fetch, 0, 'gmv')。
- 探索运行:入口 → 调用工具 q(执行的 SQL 由测试给)→ 可选的口径卡 → 报告撰写(模型的答复由测试给,引用 [[v:Q1.r0.gmv]] /
[[m:gmv]])→ 出口。
"""
from __future__ import annotations
import asyncio
import sqlite3
from typing import Any
from langchain_core.messages import AIMessage
from sqlalchemy import select
from app.core.artifact_store import graph_hash
from app.data.engine import engines
from app.db.base import SessionLocal
from app.db.models import DataSource, Run, Workflow, WorkflowVersion
TEMPLATE = ("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' "
"AND paid_at >= '{{ input.start }}' AND paid_at < '{{ input.end }}'")
#: 和口径等价的写法(别名不同、条件换序)
EQUIVALENT = ("select sum(o.amount) as 销售额 from orders o where o.paid_at >= '2026-09-01' and o.status = 'paid' "
"and o.paid_at < '2026-09-08'")
#: 少了「只算已支付」这个条件(含退款的周销售)
WITH_REFUNDS = "SELECT SUM(amount) AS gmv FROM orders WHERE paid_at >= '2026-09-01' AND paid_at < '2026-09-08'"
#: 查的是别的表(没有相关口径)
ELSEWHERE = "SELECT COUNT(*) AS n FROM stores"
ROWS = [(1, "east", "paid", 100.0, "2026-09-01", "13800000001"), (2, "east", "paid", 50.5, "2026-09-03", "13800000002"),
(3, "west", "refunded", 30.0, "2026-09-04", "13800000003"), (4, "west", "paid", 20.0, "2026-09-07", "13800000004"),
(5, "east", "open", 12.5, "2026-09-07", "13800000005"), (6, "west", "paid", 70.0, "2026-09-09", "13800000006"),
(7, "east", "paid", 5.0, "2026-08-31", "13800000007"), (8, "west", "refunded", 8.0, "2026-09-02", "13800000008"),
(9, "east", "paid", 1.0, "2026-09-08", "13800000009")]
def node(nid: str, ntype: str, **config: Any) -> dict[str, Any]:
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
def chain(*nodes: dict[str, Any]) -> dict[str, Any]:
return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
def build_shop(path) -> None:
db = sqlite3.connect(path)
db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, status TEXT, amount REAL, "
"paid_at TEXT, phone TEXT); CREATE TABLE stores (id INTEGER PRIMARY KEY, name TEXT);")
db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?)", ROWS)
db.executemany("INSERT INTO stores VALUES (?, ?)", [(1, "s1"), (2, "s2")])
db.commit()
db.close()
async def set_source(path, *, options: dict[str, Any] | None = None) -> str:
async with SessionLocal() as session:
row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
if row is None:
row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
session.add(row)
row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
row.options = dict(options or {})
await session.commit()
source_id = row.id
await engines.invalidate(source_id)
return source_id
def caliber_graph(sql: str = TEMPLATE, *, expression: str = "cell(nodes.fetch, 0, 'gmv')") -> dict[str, Any]:
return chain(
node("start", "input", fields=[{"name": "start", "required": True}, {"name": "end", "required": True},
{"name": "store", "default": "all"}]),
node("fetch", "tool", tool="db_query__shop", args={"sql": sql}),
node("card", "metrics", caliber="周销售", caliber_version="v1", metrics=[
{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 2, "format": "thousands", "expression": expression}]),
node("write", "report", instructions="写周销售"),
node("out", "output", fields=[{"name": "报告", "value": "{{ nodes.write.text }}"}]),
)
async def publish(name: str = "周销售", *, graph: dict[str, Any] | None = None, level: str = "governed",
version: int = 1, workflow_id: str | None = None) -> str:
"""建一个工作流并把第 version 版按 level 发布(直接写库,和发布接口写下的几列一样)。返回工作流 id。"""
graph = graph or caliber_graph()
async with SessionLocal() as session:
wf = await session.get(Workflow, workflow_id) if workflow_id else None
if wf is None:
wf = Workflow(name=name, graph=graph, status="published")
session.add(wf)
await session.flush()
session.add(WorkflowVersion(workflow_id=wf.id, version=version, graph=graph, graph_hash=graph_hash(graph),
level=level))
wf.published_version, wf.version, wf.graph = version, version, graph
await session.commit()
return wf.id
def explore_graph(sql: str, *, card_expression: str | None = None) -> dict[str, Any]:
parts = [node("start", "input"), node("q", "tool", tool="db_query__shop", args={"sql": sql})]
if card_expression:
parts.append(node("mine", "metrics", caliber="我的口径", caliber_version="v1", metrics=[
{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 2, "format": "thousands",
"expression": card_expression}]))
parts += [node("write", "report", instructions="写答案"),
node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}])]
return chain(*parts)
def script(monkeypatch, text: str) -> None:
from app.providers import mock_model
monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
async def wait(run_id: str) -> Run:
for _ in range(400):
await asyncio.sleep(0.05)
async with SessionLocal() as session:
row = await session.get(Run, run_id)
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
return row
raise AssertionError(f"run {run_id} 没跑完:{row.status}")
async def explore(monkeypatch, sql: str, *, text: str = "本周销售额 [[v:Q1.r0.gmv]]。", origin: str | None = "canvas",
card_expression: str | None = None, run_class: str = "exploratory") -> Run:
"""跑一次探索运行、等它封存完、等后台的口径比对做完。"""
from app.engine.runner import run_manager
from app.ledger import match as ledger_match
script(monkeypatch, text)
run = await run_manager.start(graph=explore_graph(sql, card_expression=card_expression), input_payload={},
run_class=run_class, origin_kind=origin)
row = await wait(run.id)
assert row.status == "succeeded", row.error
await ledger_match.drain()
return row
+258
View File
@@ -0,0 +1,258 @@
"""口径账本的读接口(next-round 02 做法 3、9、10):GET /api/runs/{id}/ledger、证据片段的 ledger 键、报告导出的「口径」一步、
导出留痕、GET /api/ledger/stats。
- 比对过的运行:标签、整题判断、每条查询的比对,写明封存后比对、不在封存范围内;工作流名按现在的取;
- 没比对过的:能比的排上按需比对(pending → 比完 trigger=on_demand、写明按现在的口径回看),不比的照实说为什么;
- 遮罩:绑定的值是这次运行查到过的遮罩值时不给值,差异条目里的 SQL 片段换成「已遮罩」,整份答复里找不到那个号码;
- 证据片段:比对过的带 ledger 键,没比对过的没有这个键;
- 报告导出:数据块带同一份标签和那句说明;没比对过的没有这个键;每次导出(报告、可验证报告)各记一条审计 report.exported;
- 统计:E1–E4 的口径和夹具上手算的一致,skipped 单独计。
"""
from __future__ import annotations
import json
import re
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import select
from app.db.base import SessionLocal
from app.db.models import AuditEvent, Workflow, WorkflowVersion
from app.engine.runner import run_manager
from app.ledger import match as ledger_match
from app.main import app
from tests import ledger_samples as S
@pytest.fixture(autouse=True)
async def engine_up(tmp_path):
from app.db.base import utcnow
await run_manager.setup()
path = tmp_path / "shop.db"
S.build_shop(path)
await S.set_source(path)
async with SessionLocal() as session:
for row in (await session.execute(select(WorkflowVersion).where(WorkflowVersion.retired_at.is_(None)))).scalars():
row.retired_at = utcnow()
await session.commit()
yield path
await ledger_match.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_of(client, run_id: str) -> dict:
res = await client.get(f"/api/runs/{run_id}/ledger")
assert res.status_code == 200, res.text
return res.json()
def first_segment(row, text: str) -> str:
from app.core.artifact_store import load
from app.engine.evidence import iter_segments
doc = load(row.output["_evidence"]["doc_artifact"])
return next(s["id"] for s in iter_segments(doc) if s.get("text") == text and s.get("cite"))
async def test_a_matched_run_gives_labels_and_the_whole_judgement(client, monkeypatch):
wf = await S.publish()
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
body = await ledger_of(client, row.id)
assert body["state"] == "matched" and body["note"] == "封存后比对,不在封存范围内" and body["post_seal"] is True
assert body["trigger"] == "sealed" and body["retro"] is False and "retro_note" not in body
assert body["actions"] == [] and body["primary"] == "write"
[caliber] = body["calibers"]
assert caliber["workflow_name"] == "周销售" and caliber["level_text"] == "受管" and caliber["caliber"] == f"{wf}@1"
[label] = body["reports"]["write"]["labels"].values()
assert label["kind"] == "caliber" and label["how"] == "independent" and label["workflow_name"] == "周销售"
# 工件是规范 JSON(键排好序),绑定按字段名的先后给
assert label["binding"] == [{"field": "end", "value": "2026-09-08"}, {"field": "start", "value": "2026-09-01"}]
assert body["whole"]["equivalent"] is True and body["whole"]["missing"] == ["store"]
[query] = body["queries"]
assert query["status"] == "proven" and query["candidates"][0]["status"] == "proven"
# 工作流改了名:按现在的名字给
async with SessionLocal() as session:
(await session.get(Workflow, wf)).name = "周销售(新)"
await session.commit()
assert (await ledger_of(client, row.id))["calibers"][0]["workflow_name"] == "周销售(新)"
async def test_an_unmatched_run_is_matched_on_demand_and_says_it_looks_back(client, monkeypatch):
await S.publish()
real = ledger_match.schedule
monkeypatch.setattr(ledger_match, "schedule", lambda run_id, **kw: False)
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
monkeypatch.setattr(ledger_match, "schedule", real)
body = await ledger_of(client, row.id)
assert body["state"] == "pending"
await ledger_match.drain()
body = await ledger_of(client, row.id)
assert body["state"] == "matched" and body["trigger"] == "on_demand" and body["retro"] is True
assert body["retro_note"] == "按现在的口径回看"
async def test_runs_that_are_not_compared_say_why(client, monkeypatch):
row = await S.explore(monkeypatch, S.EQUIVALENT, origin="api", text="本周销售额 [[v:Q1.r0.销售额]]。")
body = await ledger_of(client, row.id)
assert body["state"] == "ineligible" and body["reason"] == ledger_match.NOT_EXPLORE
res = await client.get("/api/runs/nope/ledger")
assert res.status_code == 404 and res.json()["code"] == "run_not_found"
async def test_masked_binding_values_and_sql_fragments_are_withheld(client, monkeypatch, engine_up):
"""口径按手机号筛({{ input.phone }});这次运行先查过手机号(数据源设了遮罩),再按其中一个号码筛:绑定的值是遮罩值,
不给值;差异条目、逐行对照里的那个号码也换成「已遮罩」。"""
await S.set_source(engine_up, options={"mask_columns": ["phone"]})
template = "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND phone = '{{ input.phone }}'"
await S.publish(graph=S.caliber_graph(template))
graph = S.chain(
S.node("start", "input"),
S.node("p", "tool", tool="db_query__shop", args={"sql": "SELECT phone FROM orders ORDER BY id"}),
S.node("q", "tool", tool="db_query__shop",
args={"sql": "SELECT SUM(amount) AS gmv FROM orders WHERE phone = '13800000001'"}),
S.node("write", "report", instructions="写答案"),
S.node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}]))
S.script(monkeypatch, "这位会员的消费 [[v:Q2.r0.gmv]]。")
run = await run_manager.start(graph=graph, input_payload={}, origin_kind="canvas")
row = await S.wait(run.id)
assert row.status == "succeeded", row.error
await ledger_match.drain()
body = await ledger_of(client, row.id)
assert body["state"] == "matched"
[label] = body["reports"]["write"]["labels"].values()
assert label["kind"] == "related"
[related] = label["related"]
assert related["binding"] == [{"field": "phone", "masked": True, "value": "已遮罩"}]
assert "13800000001" not in json.dumps(body, ensure_ascii=False)
assert related["diff"]["items"] and "13800000001" not in json.dumps(related["diff"], ensure_ascii=False)
def test_diff_fragments_go_through_the_same_mask_as_the_evidence_panel():
"""差异条目、逐行对照里的 SQL 片段走证据面板同一套换法(api/evidence._Shown):这次运行查到过的遮罩值换成「已遮罩」。"""
from app.ledger.labels import view_label
from app.ledger.view import _Hide
class Secrets:
def hit(self, value):
return str(value) == "13800000001"
def weak_hit(self, text):
return text == "13800000001"
def scrub(self, text, *, prose=False, loose=None):
return text.replace("13800000001", "已遮罩")
hider = _Hide(Secrets())
label = {"kind": "related", "related": [{
"caliber": "wf@1", "workflow_id": "wf", "version": 1, "level": "published", "binding": {"phone": "13800000001"},
"diff": {"items": [{"aspect": "filter", "a": "phone = '13800000001'", "b": None,
"text": "这次只保留 phone = '13800000001' 的行,口径没有这个条件"}],
"fallback": False, "text_diff": ["- WHERE phone = '13800000001'", "+ WHERE TRUE"]}}]}
shown = view_label(label, hide=hider.hide, shown=hider.shown, names={"wf": "周销售"})
text = json.dumps(shown, ensure_ascii=False)
assert "13800000001" not in text and text.count("已遮罩") >= 4
assert shown["related"][0]["workflow_name"] == "周销售" and shown["related"][0]["level_text"] == "已发布"
async def test_the_evidence_segment_carries_the_label_only_when_matched(client, monkeypatch):
await S.publish()
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
seg = first_segment(row, "170.5")
res = await client.get(f"/api/runs/{row.id}/evidence/segments/{seg}")
assert res.status_code == 200, res.text
ledger = res.json()["ledger"]
assert ledger["label"]["kind"] == "caliber" and ledger["note"] == "封存后比对,不在封存范围内"
assert ledger["post_seal"] is True and ledger["artifact"]
monkeypatch.setattr(ledger_match, "schedule", lambda run_id, **kw: False)
other = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
res = await client.get(f"/api/runs/{other.id}/evidence/segments/{first_segment(other, '170.5')}")
assert res.status_code == 200 and "ledger" not in res.json()
def _data_block(page: str) -> dict:
found = re.search(r'<script type="application/json" id="report-data">(.*?)</script>', page, re.S)
assert found, "页面里没有数据块"
return json.loads(found.group(1))
async def exported(run_id: str) -> list[AuditEvent]:
async with SessionLocal() as session:
return list((await session.execute(select(AuditEvent).where(
AuditEvent.action == "report.exported", AuditEvent.subject_id == run_id).order_by(AuditEvent.at))).scalars())
async def test_report_export_shows_the_label_and_records_the_export(client, monkeypatch):
await S.publish()
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
res = await client.get(f"/api/runs/{row.id}/report.html")
assert res.status_code == 200, res.text
data = _data_block(res.text)
assert data["ledger"]["note"] == "封存后比对,不在封存范围内"
[label] = data["ledger"]["labels"].values()
assert label["kind"] == "caliber"
[audit] = await exported(row.id)
assert audit.subject_kind == "run" and audit.data["format"] == "html"
assert audit.data["report_node"] == "write" and audit.data["doc_artifact"] == row.output["_evidence"]["doc_artifact"]
res = await client.get(f"/api/runs/{row.id}/bundle.html")
assert res.status_code == 200, res.text
audits = await exported(row.id)
assert [a.data["format"] for a in audits] == ["html", "bundle"]
assert audits[1].data["reports"] == [{"report_node": "write", "doc_artifact": audit.data["doc_artifact"]}]
# 可验证报告把封存之后追加的比对列出来,只说几条,不带绑定
assert "封存后比对已发布口径:1 条等价已证明" in res.text
async def test_an_unmatched_report_export_has_no_ledger_block(client, monkeypatch):
monkeypatch.setattr(ledger_match, "schedule", lambda run_id, **kw: False)
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
res = await client.get(f"/api/runs/{row.id}/report.html")
assert res.status_code == 200 and "ledger" not in _data_block(res.text)
async def test_stats_count_what_the_fixtures_did(client, monkeypatch):
from datetime import datetime, timezone
since = datetime.now(timezone.utc)
wf = await S.publish()
await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
await S.explore(monkeypatch, S.WITH_REFUNDS, text="含退款 [[v:Q1.r0.gmv]]。")
await S.explore(monkeypatch, S.ELSEWHERE, text="一共 [[v:Q1.r0.n]] 家店。")
monkeypatch.setattr(ledger_match, "schedule", lambda run_id, **kw: False)
timed = await S.explore(monkeypatch, S.EQUIVALENT + " -- 超时用", text="本周销售额 [[v:Q1.r0.销售额]]。")
await ledger_match.match_run(timed.id, timeout=0)
res = await client.get("/api/ledger/stats", params={"since": since.isoformat()})
assert res.status_code == 200, res.text
body = res.json()
e1 = body["e1"]
assert e1["runs"] == 4 and e1["queries"] == 4
assert e1["status"] == {"proven": 1, "composed": 0, "related": 1, "none": 1, "skipped": 1}
assert e1["with_candidates"] == 3 and e1["rates"]["proven"] == round(1 / 3, 4)
assert e1["diff_aspects"]["filter"] == 1 and e1["labels"]["caliber"] == 1 and e1["labels"]["unknown"] == 1
assert e1["whole_equivalent"] == 1
assert body["e2"]["launched"] == 0 and body["e2"]["feedback"]["available"] is False
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
async def test_stats_take_the_run_ids_the_caller_names(monkeypatch):
"""评测(第 3 波)直接调 compute_stats,点名自己的运行:只算这几次。"""
from app.ledger.stats import compute_stats
await S.publish()
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
await S.explore(monkeypatch, S.WITH_REFUNDS, text="含退款 [[v:Q1.r0.gmv]]。")
async with SessionLocal() as session:
body = await compute_stats(session, run_ids=[row.id])
assert body["e1"]["runs"] == 1 and body["e1"]["status"]["proven"] == 1 and body["window"]["runs"] == [row.id]
+290
View File
@@ -0,0 +1,290 @@
"""认口径:运行封存之后在后台比对已发布口径(app/ledger/match.py、store.py、labels.py,next-round 02 做法 2、3)。
真的跑一遍探索运行(虚构店铺库、mock 模型写报告),等后台比对做完,看比对工件和 ledger.matched:
- 等价的写法:查询 proven、绑定取对、列按位置映射,报告里引用它的数标「来自口径 · 等价已证明」,整题等价、写出还缺的入口字段;
工件 meta 记 calibers,事件追加在封存之后、封存核对照旧一致;同一份封存再比一次工件哈希一样;
- 少一个条件:related,差在哪写「口径只保留……这次没有这个条件」,数标相关口径;
- 没有相关口径:none,数标这次的推理;正式运行、接口发起的运行、失败的运行不比;
- 超时、快照被改过、SQL 解析不了:skipped,照实写原因,数标没比对;
- 指标:输入等价、式子和口径卡一样才算来自口径,式子不同的是相关口径;
- 启动补比:只补没比过的,trigger=backfill。
"""
from __future__ import annotations
import json
import pytest
from sqlalchemy import select
from app.core.config import settings
from app.core.events import EventType
from app.db.base import SessionLocal
from app.db.models import Artifact, Run, RunEvent
from app.engine.runner import run_manager, verify_manifest
from app.ledger import match as ledger_match
from app.ledger import store
from tests import ledger_samples as S
@pytest.fixture(autouse=True)
async def engine_up(tmp_path):
await run_manager.setup()
path = tmp_path / "shop.db"
S.build_shop(path)
await S.set_source(path)
# 每个用例自己发布口径:先把别的用例发布的退役掉,候选口径只看这一个用例的
from app.db.models import WorkflowVersion
from app.db.base import utcnow
async with SessionLocal() as session:
for row in (await session.execute(select(WorkflowVersion).where(WorkflowVersion.retired_at.is_(None)))).scalars():
row.retired_at = utcnow()
await session.commit()
yield path
await ledger_match.drain()
await run_manager.shutdown()
async def matched(run_id: str) -> tuple[store.Recorded, list[RunEvent]]:
recorded = await store.latest(run_id)
assert recorded is not None and recorded.content is not None, recorded
async with SessionLocal() as session:
events = list((await session.execute(select(RunEvent).where(
RunEvent.run_id == run_id, RunEvent.type == EventType.LEDGER_MATCHED).order_by(RunEvent.seq))).scalars())
return recorded, events
def only_label(content: dict) -> dict:
[labels] = [r["labels"] for r in content["reports"].values()]
[label] = labels.values()
return label
async def test_an_equivalent_query_is_proven_and_its_number_comes_from_the_caliber(monkeypatch):
wf = await S.publish()
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
recorded, [event] = await matched(row.id)
content = recorded.content
assert content["format"] == store.FORMAT and content["algo"] == store.ALGO and content["run_id"] == row.id
assert content["manifest_hash"] == row.manifest_hash
[query] = content["queries"]
assert query["status"] == "proven" and query["source"] == "shop" and query["tables"] == ["orders"]
[cand] = query["candidates"]
assert cand["caliber"] == f"{wf}@1" and cand["level"] == "governed" and cand["node_id"] == "fetch"
assert cand["binding"] == {"start": "2026-09-01", "end": "2026-09-08"}
assert cand["columns"] == [{"index": 0, "ours": "销售额", "caliber": "gmv"}]
[caliber] = content["calibers"]
assert caliber["caliber"] == f"{wf}@1" and caliber["fetches"] == ["fetch"] and caliber["cards"] == ["card"]
assert "name" not in caliber, "工件里不存工作流名(名字会改)"
label = only_label(content)
assert label["kind"] == "caliber" and label["how"] == "independent" and label["caliber"] == f"{wf}@1"
assert label["column"] == "gmv" and label["binding"] == {"start": "2026-09-01", "end": "2026-09-08"}
whole = content["whole"]
assert whole["equivalent"] is True and whole["caliber"] == f"{wf}@1" and whole["how"] == "independent"
assert whole["missing"] == ["store"] and whole["missing_required"] == []
# 事件在封存之后,封存核对照旧一致
assert event.seq > row.manifest_seq and event.data["post_seal"] is True
assert event.data["artifact"] == recorded.artifact and event.data["trigger"] == "sealed"
assert event.data["counts"]["proven"] == 1
assert (await verify_manifest(row.id))["ok"] is True
async with SessionLocal() as session:
ref = await session.get(Artifact, (recorded.artifact, row.id))
assert ref.kind == store.KIND and ref.meta["calibers"] == [f"{wf}@1"] and ref.meta["related"] == []
# 同一份封存再比一次:工件一字不差
again = await ledger_match.match_run(row.id, trigger="on_demand")
assert again == recorded.artifact
latest = await store.latest(row.id)
assert latest.trigger == "on_demand" and latest.retro is True
async def test_a_missing_condition_is_related_and_says_where_it_differs(monkeypatch):
wf = await S.publish()
row = await S.explore(monkeypatch, S.WITH_REFUNDS)
recorded, _ = await matched(row.id)
[query] = recorded.content["queries"]
assert query["status"] == "related"
[cand] = query["candidates"]
assert cand["status"] == "related" and cand["caliber"] == f"{wf}@1"
assert cand["binding"] == {"start": "2026-09-01", "end": "2026-09-08"} and cand["via"] == "heuristic"
texts = [item["text"] for item in cand["diff"]["items"]]
assert any("口径只保留" in t and "这次没有这个条件" in t for t in texts), texts
label = only_label(recorded.content)
assert label["kind"] == "related" and label["related"][0]["caliber"] == f"{wf}@1"
assert recorded.content["whole"]["equivalent"] is False
async def test_without_a_related_caliber_the_number_is_this_runs_reasoning(monkeypatch):
await S.publish()
row = await S.explore(monkeypatch, S.ELSEWHERE, text="一共 [[v:Q1.r0.n]] 家店。")
recorded, _ = await matched(row.id)
[query] = recorded.content["queries"]
assert query["status"] == "none" and query["candidates"] == []
assert only_label(recorded.content)["kind"] == "reasoning"
assert recorded.content["calibers"] == []
async def test_a_value_differs_in_the_hard_coded_status_is_not_proven(monkeypatch):
"""写死的码值不同('paid' 对 'refunded'):normalize_sql 文本相同,照样不等价。"""
await S.publish()
sql = S.EQUIVALENT.replace("'paid'", "'refunded'")
row = await S.explore(monkeypatch, sql, text="退款 [[v:Q1.r0.销售额]]。")
recorded, _ = await matched(row.id)
assert recorded.content["queries"][0]["status"] == "related"
@pytest.mark.parametrize("origin, run_class", [("api", "exploratory"), (None, "exploratory")])
async def test_runs_that_are_not_exploration_are_not_compared(monkeypatch, origin, run_class):
await S.publish()
row = await S.explore(monkeypatch, S.EQUIVALENT, origin=origin, run_class=run_class,
text="本周销售额 [[v:Q1.r0.销售额]]。")
assert await store.latest(row.id) is None
async with SessionLocal() as session:
run = await session.get(Run, row.id)
assert ledger_match.ineligible(run) == ledger_match.NOT_EXPLORE
assert await ledger_match.match_run(row.id) is None
def test_ineligible_reasons():
from types import SimpleNamespace
base = dict(run_class="exploratory", origin_kind="chat", status="succeeded", manifest_hash="h")
assert ledger_match.ineligible(SimpleNamespace(**base)) is None
assert ledger_match.ineligible(None) == ledger_match.NOT_FOUND
assert ledger_match.ineligible(SimpleNamespace(**{**base, "run_class": "formal"})) == ledger_match.FORMAL
assert ledger_match.ineligible(SimpleNamespace(**{**base, "origin_kind": "eval"})) == ledger_match.NOT_EXPLORE
assert ledger_match.ineligible(SimpleNamespace(**{**base, "status": "failed"})) == ledger_match.NOT_SUCCEEDED
assert ledger_match.ineligible(SimpleNamespace(**{**base, "manifest_hash": None})) == ledger_match.NOT_SEALED
assert all(ledger_match.wanted("exploratory", o) for o in ("chat", "canvas", "trial", "personal_app"))
assert not ledger_match.wanted("formal", "chat") and not ledger_match.wanted("exploratory", "reconcile")
async def test_a_timeout_is_recorded_as_skipped(monkeypatch):
await S.publish()
monkeypatch.setattr(ledger_match, "schedule", lambda run_id, **kw: False)
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
assert await store.latest(row.id) is None
artifact = await ledger_match.match_run(row.id, timeout=0)
assert artifact
recorded, _ = await matched(row.id)
[query] = recorded.content["queries"]
assert query["status"] == "skipped" and query["reason"] == ledger_match.TIMEOUT
assert query["candidates"][0]["status"] == "skipped"
label = only_label(recorded.content)
assert label["kind"] == "unknown" and label["reason"] == ledger_match.TIMEOUT
async def test_a_tampered_snapshot_is_skipped(monkeypatch):
await S.publish()
monkeypatch.setattr(ledger_match, "schedule", lambda run_id, **kw: False)
# 工件按内容寻址、各用例共用一个工件库:换一种只在这里用的写法,改的那份快照不会被别的用例读到;改完放回去
row = await S.explore(monkeypatch, S.EQUIVALENT + " -- 改快照用", text="本周销售额 [[v:Q1.r0.销售额]]。")
async with SessionLocal() as session:
[end] = (await session.execute(select(RunEvent).where(
RunEvent.run_id == row.id, RunEvent.type == "tool.end"))).scalars()
snap_id = end.data["query_artifact"]
path = settings.data_dir / "artifacts" / snap_id[:2] / f"{snap_id}.json"
original = path.read_text("utf-8")
snap = json.loads(original)
snap["rows"] = [[999999]]
path.write_text(json.dumps(snap, ensure_ascii=False), "utf-8")
try:
await ledger_match.match_run(row.id)
finally:
path.write_text(original, "utf-8")
recorded, _ = await matched(row.id)
[query] = recorded.content["queries"]
assert query["status"] == "skipped" and query["reason"] == ledger_match.SNAPSHOT_TAMPERED
assert only_label(recorded.content)["kind"] == "unknown"
def test_unparseable_sql_is_skipped_with_the_reason():
caliber = ledger_match.Caliber("wf", 1, "governed", "h", (ledger_match.Fetch("fetch", "shop", S.TEMPLATE,
frozenset({"orders"})),))
query = ledger_match.Query("a1", "q", "shop", "SELEC amount FROM orders WHERE", tables=frozenset({"orders"}),
dialect="sqlite")
[row] = ledger_match.compare_queries([query], [caliber])
assert row["status"] == "skipped" and row["reason"] == ledger_match.PARSE_FAILED
def test_candidates_are_prefiltered_by_source_and_table():
fetches = (ledger_match.Fetch("a", "shop", S.TEMPLATE, frozenset({"orders"})),
ledger_match.Fetch("b", "crm", S.TEMPLATE, frozenset({"orders"})),
ledger_match.Fetch("c", "shop", "SELECT COUNT(*) FROM stores", frozenset({"stores"})))
caliber = ledger_match.Caliber("wf", 1, "published", "h", fetches)
query = ledger_match.Query("a1", "q", "shop", S.EQUIVALENT, tables=frozenset({"orders"}), dialect="sqlite")
[row] = ledger_match.compare_queries([query], [caliber])
assert [c["node_id"] for c in row["candidates"]] == ["a"] and row["status"] == "proven"
async def test_a_metric_from_equivalent_inputs_with_the_same_formula_comes_from_the_caliber(monkeypatch):
wf = await S.publish()
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[m:gmv]]。",
card_expression="cell(nodes.q, 0, '销售额')")
recorded, _ = await matched(row.id)
label = only_label(recorded.content)
assert label["kind"] == "caliber" and label["caliber"] == f"{wf}@1"
assert label["card"] == "card" and label["metric"] == "gmv"
assert recorded.content["whole"]["equivalent"] is True
async def test_a_metric_with_a_different_formula_is_only_related(monkeypatch):
wf = await S.publish()
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[m:gmv]]。",
card_expression="cell(nodes.q, 0, '销售额') * 2")
recorded, _ = await matched(row.id)
label = only_label(recorded.content)
assert label["kind"] == "related" and label["reason"] == "取数和口径等价,但指标的算法和口径卡上的不同"
assert label["related"][0]["caliber"] == f"{wf}@1"
async def test_published_and_governed_both_count_and_the_label_says_which(monkeypatch):
wf = await S.publish(level="published")
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
recorded, _ = await matched(row.id)
label = only_label(recorded.content)
assert label["kind"] == "caliber" and label["level"] == "published" and label["caliber"] == f"{wf}@1"
async def test_only_the_latest_published_version_is_a_candidate(monkeypatch):
wf = await S.publish()
await S.publish(workflow_id=wf, version=2, graph=S.caliber_graph(S.TEMPLATE.replace("'paid'", "'open'")))
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
recorded, _ = await matched(row.id)
[query] = recorded.content["queries"]
assert [c["caliber"] for c in query["candidates"]] == [f"{wf}@2"] and query["status"] == "related"
async def test_backfill_matches_only_runs_never_matched(monkeypatch):
wf = await S.publish()
monkeypatch.setattr(ledger_match, "schedule", lambda run_id, **kw: False)
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
assert await store.latest(row.id) is None
assert await ledger_match.backfill() >= 1
recorded, [event] = await matched(row.id)
assert event.data["trigger"] == "backfill" and recorded.retro is True
assert recorded.content["queries"][0]["candidates"][0]["caliber"] == f"{wf}@1"
# 再补一次:比过的不再比
await ledger_match.backfill()
_, events = await matched(row.id)
assert len(events) == 1
async def test_backfill_follows_the_scheduler_switch(monkeypatch):
monkeypatch.setattr(settings, "scheduler", False)
assert ledger_match.start_backfill() is False
async def test_only_sealed_query_snapshots_are_compared(monkeypatch):
"""比对只读封存范围内交回过的查询快照(api/evidence._sealed 的读法):事后往工件表里插一份属于这次运行的快照,不比它。"""
from app.core.artifact_store import put_json
await S.publish()
row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
await put_json({"sql": S.WITH_REFUNDS, "source": "shop", "columns": ["gmv"], "rows": [[1]]},
kind="query_snapshot", run_id=row.id, node_id="q")
await ledger_match.match_run(row.id, trigger="on_demand")
recorded, _ = await matched(row.id)
assert [q["status"] for q in recorded.content["queries"]] == ["proven"]
+133
View File
@@ -0,0 +1,133 @@
"""运行 ↔ 轮次的来源链(next-round 02 做法 9):POST /api/runs 的 origin_ref、「在画布里打开」的 WorkflowIn.origin。
- origin_ref 只在来源是问数据时收:核对对话用不用得了、对话在不在、这一轮是不是这段对话的,写进 Run.origin 和封存的 run.origin;
别的来源带它、伪造的(别的对话的轮次、不存在的对话)一律拒收;
- 新建工作流带 origin(run_id、conversation_id、turn_id):运行读不到、轮次不属于对话的拒收,对得上的记进 workflows.origin,
形状同从运行提取(/copilot/from-run)。
"""
from __future__ import annotations
import asyncio
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import select
from app.db.base import SessionLocal
from app.db.models import Run, RunEvent, Workflow
from app.engine.runner import run_manager
from app.main import app
GRAPH = {"nodes": [{"id": "start", "type": "input", "position": {"x": 0, "y": 0},
"data": {"label": "开始", "config": {"fields": [{"name": "question"}]}}},
{"id": "out", "type": "output", "position": {"x": 0, "y": 0},
"data": {"label": "成果", "config": {"fields": [{"name": "答案", "value": "{{ input.question }}"}]}}}],
"edges": [{"source": "start", "target": "out"}]}
@pytest.fixture(autouse=True)
async def engine_up():
await run_manager.setup()
yield
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 conversation_with_turn(client) -> tuple[str, str]:
res = await client.post("/api/conversations", json={"title": "问数据"})
assert res.status_code == 201, res.text
cid = res.json()["id"]
res = await client.post(f"/api/conversations/{cid}/turns", json={"question": "本周销售额多少"})
assert res.status_code == 201, res.text
return cid, res.json()["id"]
async def wait(run_id: str) -> Run:
for _ in range(200):
await asyncio.sleep(0.05)
async with SessionLocal() as session:
row = await session.get(Run, run_id)
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
return row
raise AssertionError("运行没跑完")
async def test_a_chat_run_records_its_conversation_and_turn(client):
cid, tid = await conversation_with_turn(client)
res = await client.post("/api/runs", json={"graph": GRAPH, "input": {"question": "本周销售额多少"}, "origin": "chat",
"origin_ref": {"conversation_id": cid, "turn_id": tid}})
assert res.status_code == 201, res.text
body = res.json()
assert body["origin_kind"] == "chat" and body["origin"] == {"conversation_id": cid, "turn_id": tid}
row = await wait(body["id"])
async with SessionLocal() as session:
[event] = (await session.execute(select(RunEvent).where(
RunEvent.run_id == row.id, RunEvent.type == "run.origin"))).scalars()
# 在封存范围内:事后改不了
assert event.seq <= row.manifest_seq
assert event.data["kind"] == "chat" and event.data["conversation_id"] == cid and event.data["turn_id"] == tid
async def test_origin_ref_is_only_for_chat_runs(client):
cid, tid = await conversation_with_turn(client)
res = await client.post("/api/runs", json={"graph": GRAPH, "origin": "canvas",
"origin_ref": {"conversation_id": cid, "turn_id": tid}})
assert res.status_code == 422 and res.json()["code"] == "origin_ref_not_allowed"
assert res.json()["field"] == "origin_ref"
async def test_forged_origin_refs_are_refused(client):
cid, tid = await conversation_with_turn(client)
other, other_turn = await conversation_with_turn(client)
for ref in ({"conversation_id": "nope", "turn_id": tid}, {"conversation_id": cid, "turn_id": other_turn},
{"conversation_id": cid, "turn_id": "nope"}):
res = await client.post("/api/runs", json={"graph": GRAPH, "origin": "chat", "origin_ref": ref})
assert res.status_code == 422, (ref, res.text)
assert res.json()["code"] == "origin_ref_invalid"
async def test_someone_elses_conversation_is_refused(client, monkeypatch):
"""对话按人隔离(开登录后的使用者):挂到别人的对话上,和读别人的对话一样回 403。"""
from app.api import conversations
cid, tid = await conversation_with_turn(client)
monkeypatch.setattr(conversations, "_isolated", lambda: "someone-else")
res = await client.post("/api/runs", json={"graph": GRAPH, "origin": "chat",
"origin_ref": {"conversation_id": cid, "turn_id": tid}})
assert res.status_code == 403
async def test_open_in_canvas_records_where_the_workflow_came_from(client):
cid, tid = await conversation_with_turn(client)
run = (await client.post("/api/runs", json={"graph": GRAPH, "origin": "chat",
"origin_ref": {"conversation_id": cid, "turn_id": tid}})).json()
await wait(run["id"])
res = await client.post("/api/workflows", json={"name": "问数据:本周销售额", "graph": GRAPH,
"origin": {"run_id": run["id"], "conversation_id": cid,
"turn_id": tid}})
assert res.status_code == 201, res.text
async with SessionLocal() as session:
wf = await session.get(Workflow, res.json()["id"])
assert wf.origin == {"run_id": run["id"], "conversation_id": cid, "turn_id": tid}
# 不带来源的照旧(手建的为空)
res = await client.post("/api/workflows", json={"name": "手建", "graph": GRAPH})
async with SessionLocal() as session:
assert (await session.get(Workflow, res.json()["id"])).origin is None
async def test_open_in_canvas_refuses_origins_that_do_not_check_out(client):
cid, tid = await conversation_with_turn(client)
other, other_turn = await conversation_with_turn(client)
run = (await client.post("/api/runs", json={"graph": GRAPH, "origin": "chat"})).json()
await wait(run["id"])
for origin in ({"run_id": "nope"}, {"run_id": run["id"], "conversation_id": cid, "turn_id": other_turn},
{"run_id": run["id"], "turn_id": tid}):
res = await client.post("/api/workflows", json={"name": "x", "graph": GRAPH, "origin": origin})
assert res.status_code == 422, (origin, res.text)
assert res.json()["code"] == "workflow_origin_invalid"