diff --git a/backend/app/api/bundle.py b/backend/app/api/bundle.py index f0256724..c25ef73f 100644 --- a/backend/app/api/bundle.py +++ b/backend/app/api/bundle.py @@ -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), # 导出的是这一刻的封存件，不该被哪一层缓存下来、换给下一个请求 diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index 9ce10855..e647c40c 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -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) diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index 10bc2165..2fb98348 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -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()) diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index bc3ce553..f06d8b4a 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -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}。""" diff --git a/backend/app/api/workflows.py b/backend/app/api/workflows.py index 044a9297..91cefb94 100644 --- a/backend/app/api/workflows.py +++ b/backend/app/api/workflows.py @@ -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) diff --git a/backend/app/bundle/build.py b/backend/app/bundle/build.py index a7ca0679..caada4ab 100644 --- a/backend/app/bundle/build.py +++ b/backend/app/bundle/build.py @@ -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 "" diff --git a/backend/app/core/events.py b/backend/app/core/events.py index ed898610..bf891b8f 100644 --- a/backend/app/core/events.py +++ b/backend/app/core/events.py @@ -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） diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 25c5e5e1..714fa680 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -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: """对到此为止的全部事件算清单哈希，记下封到了哪一条。""" diff --git a/backend/app/ledger/labels.py b/backend/app/ledger/labels.py new file mode 100644 index 00000000..5511ca6d --- /dev/null +++ b/backend/app/ledger/labels.py @@ -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"] diff --git a/backend/app/ledger/match.py b/backend/app/ledger/match.py new file mode 100644 index 00000000..a05f5db8 --- /dev/null +++ b/backend/app/ledger/match.py @@ -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"] diff --git a/backend/app/ledger/stats.py b/backend/app/ledger/stats.py new file mode 100644 index 00000000..38aa108e --- /dev/null +++ b/backend/app/ledger/stats.py @@ -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"] diff --git a/backend/app/ledger/store.py b/backend/app/ledger/store.py new file mode 100644 index 00000000..3fd3aff2 --- /dev/null +++ b/backend/app/ledger/store.py @@ -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"] diff --git a/backend/app/ledger/view.py b/backend/app/ledger/view.py new file mode 100644 index 00000000..5e827e04 --- /dev/null +++ b/backend/app/ledger/view.py @@ -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"] diff --git a/backend/app/report_export/build.py b/backend/app/report_export/build.py index 1832abba..cf353fc6 100644 --- a/backend/app/report_export/build.py +++ b/backend/app/report_export/build.py @@ -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: diff --git a/backend/app/report_export/runtime.js b/backend/app/report_export/runtime.js index e0f7eeef..f88755e7 100644 --- a/backend/app/report_export/runtime.js +++ b/backend/app/report_export/runtime.js @@ -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] diff --git a/backend/tests/ledger_samples.py b/backend/tests/ledger_samples.py new file mode 100644 index 00000000..4992edf0 --- /dev/null +++ b/backend/tests/ledger_samples.py @@ -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 diff --git a/backend/tests/test_ledger_api.py b/backend/tests/test_ledger_api.py new file mode 100644 index 00000000..d3fda04a --- /dev/null +++ b/backend/tests/test_ledger_api.py @@ -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'', 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] diff --git a/backend/tests/test_ledger_match.py b/backend/tests/test_ledger_match.py new file mode 100644 index 00000000..ab25b27b --- /dev/null +++ b/backend/tests/test_ledger_match.py @@ -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"] diff --git a/backend/tests/test_run_origin_ref.py b/backend/tests/test_run_origin_ref.py new file mode 100644 index 00000000..086ccde5 --- /dev/null +++ b/backend/tests/test_run_origin_ref.py @@ -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"