mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:51:59 +00:00
按口径正式出具与口径对账（口径账本做法 5、6，拍板 L1、L5）
为什么：问数据的答案整题等价于已发布口径时，要能按那一版发起一次完整走过门禁的正式运行（路径一、二a）；只是相关、不等价时，要能看「差多少」（路径二b）。两条路都不能让客户端冒充来源、自己传参数，也不能在 Python 里算差值。
正式出具（新增 ledger/issue.py）：
- POST /api/ledger/answers/{run_id}/issue {workflow_id, version}：参数只从比对工件的绑定取，补入口缺省值再过 apps/forms 校验；经 launch_run 发起正式运行，来源 ledger_issue，origin 记 source_run、conversation_id、turn_id、binding_hash、口径和认法、是否自动；按 expect_hash 钉 graph_hash。
- 不能出具照实写原因：整题不等价、点名别的口径、版本退役或没发布、图里有人工节点或写入（apps/backfill._side_effect 抽成共用的 graph_side_effect）、缺入口字段、超过每日上限、权限（L5：使用者只能按他有权使用的应用固定着的版本出具，复用应用授权，origin 记 granted_by_app；别的写「请搭建者出具」）。
- 自动出具（L1）：比对完（只在运行一封存就比的那一次）整题独立等价、版本受管、设置开着、当天没超上限才自动；助手选用的只给一键。不自动的原因写进 actions 的 auto_reason。
- 同一运行、同一口径、同一组参数只发起一次（input_key 去重），进程内按（运行、工作流、版本）串行。
- 设置加 ledger 一组：自动出具开关、每日上限、比对时限（ledger/match 读它）。
口径对账（新增 ledger/reconcile.py，derive 扩 reconcile_graph / ours_block / assemble 的出口契约）：
- POST /api/ledger/answers/{run_id}/reconcile：按那一版确定性派生只读图（这次的取数 ours_q1 → 口径的取数 → 口径卡 → 对账卡，出口契约 metrics_from，不加报告、不调模型），来源 reconcile 的探索运行；对账卡算这次、口径、差值、相对差。
- 参数是启发得来的、不全、含遮罩值时先让人核对（needs_check，没核对的 409）；指标式不是取数一格、差异有兜底时要人选对照哪个指标，选择记进 origin。
- 多行结果：被引用的行按维度列经合并查询 LEFT JOIN 对齐、按引用先后排，最多 10 行；维度是遮罩列、合并键不唯一照实拒绝。使用者发起带 MASK_RESULTS / READ_ONLY，搭建者也带 READ_ONLY。
- GET /api/runs/{id}/ledger 的 actions 补上出具、对账两类，issued、reconciled 列出发起过的运行。
测试：tests/test_ledger_issue.py（自动出具、只给一键的几种、拒绝原因、使用者的应用授权、并发只发一次、设置）、tests/test_ledger_reconcile.py（核对参数、派生图确定与契约、差值和相对差、出处链、要人选列、多行对齐、遮罩维度和重复键拒绝、使用者只读遮罩）；第 2 波 test_ledger_api 的 actions 断言按这一波改。
This commit is contained in:
ThearchyHelios
@@ -3,10 +3,15 @@
|
||||
不另建存储：ActionRecord 本身带着工作流、版本、graph_hash、能力版本、payload_hash、回执；「口径 → 应用」就是固定这一版的
|
||||
应用版本。用在发布弹窗、退役确认、版本列表三处。
|
||||
|
||||
口径账本（docs/design/next-round/02-口径账本.md）在这里加两个只读接口：
|
||||
口径账本（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）。
|
||||
动作、出具过的正式运行、对过的账（app/ledger/view.py）；
|
||||
- GET /api/ledger/stats（router，搭建者）：账本自身给出的统计 E1–E4，从比对工件、运行、审计现算（app/ledger/stats.py）；
|
||||
- POST /api/ledger/answers/{run_id}/issue（answers_router，登录即可，第 3 波）：按口径正式出具（app/ledger/issue.py，拍板
|
||||
L1、L5：参数只从比对工件取，使用者只能按他有权使用的应用固定着的版本出具）；
|
||||
- POST /api/ledger/answers/{run_id}/reconcile（同上）：口径对账，点开才跑一次确定性派生的只读图（app/ledger/reconcile.py）；
|
||||
- GET /api/ledger/certify-queue、POST /api/ledger/certify-queue/dismiss（router，搭建者，第 3 波）：可沉淀的问题、「不是口径」
|
||||
（app/ledger/certify.py）。这几个新接口都不放给访问密钥（auth/keys.KEY_ROUTES 不登记）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -15,6 +20,7 @@ from typing import Any
|
||||
from datetime import datetime
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
@@ -100,3 +106,58 @@ async def ledger_stats(since: datetime | None = Query(default=None, description=
|
||||
|
||||
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())
|
||||
|
||||
|
||||
class IssueIn(BaseModel):
|
||||
"""按哪份口径出具：只能选口径，参数由服务端从比对工件里取（不收客户端传的参数）。"""
|
||||
|
||||
workflow_id: str = Field(min_length=1, max_length=32)
|
||||
version: int = Field(ge=1)
|
||||
|
||||
|
||||
@answers_router.post("/{run_id}/issue")
|
||||
async def issue_answer(run_id: str, payload: IssueIn) -> dict[str, Any]:
|
||||
"""按口径正式出具这次问数据的答案（app/ledger/issue.py）：经 launch_run 发起正式运行，来源 ledger_issue。同一运行、同一
|
||||
口径、同一组参数只发起一次，再点交回已经发起的那一次（created 为假）。不能出具时回 409（整题不等价、版本退役、有人工
|
||||
节点或写入、缺入口字段……）、403（权限不够：使用者只能按他有权使用的应用固定着的版本出具）、429（超过每日上限），
|
||||
detail 写明原因。"""
|
||||
from app.api.coded import CodedHTTPException
|
||||
from app.api.deps import current_principal
|
||||
from app.ledger import issue as ledger_issue
|
||||
|
||||
try:
|
||||
run, created = await ledger_issue.issue(run_id, payload.workflow_id, payload.version,
|
||||
principal=current_principal())
|
||||
except ledger_issue.IssueRefused as e:
|
||||
raise CodedHTTPException(e.status_code, e.message, e.code) from e
|
||||
return {"run_id": run.id, "created": created, "issued": ledger_issue.issued_view(run)}
|
||||
|
||||
|
||||
class ReconcileIn(BaseModel):
|
||||
"""对哪条查询、和哪份相关口径对账。inputs + confirm：参数要核对时（动作里 needs_check）人核对、改过的参数；pairs：认不出
|
||||
哪一列对哪个指标时人选的 [{ours: 这次的列, metric: 口径卡上的指标 id}]（动作里写明要不要选）。"""
|
||||
|
||||
query: str = Field(min_length=1, max_length=64)
|
||||
workflow_id: str = Field(min_length=1, max_length=32)
|
||||
version: int = Field(ge=1)
|
||||
inputs: dict[str, Any] | None = None
|
||||
confirm: bool = False
|
||||
pairs: list[dict[str, Any]] | None = None
|
||||
|
||||
|
||||
@answers_router.post("/{run_id}/reconcile", status_code=201)
|
||||
async def reconcile_answer(run_id: str, payload: ReconcileIn) -> dict[str, Any]:
|
||||
"""口径对账（app/ledger/reconcile.py）：按那一版口径确定性派生只读图，发起一次探索运行（来源 reconcile），差值在对账卡
|
||||
里算。参数是推断出来的、不全的要先核对（409，code ledger_reconcile_check_inputs）；认不出列怎么配的要人选（409，
|
||||
ledger_reconcile_needs_pairs）；多行时维度是遮罩列、键不唯一照实拒绝（409）。"""
|
||||
from app.api.coded import CodedHTTPException
|
||||
from app.api.deps import current_principal
|
||||
from app.ledger import reconcile as ledger_reconcile
|
||||
|
||||
try:
|
||||
run = await ledger_reconcile.reconcile(run_id, ledger_reconcile.Request(
|
||||
query=payload.query, workflow_id=payload.workflow_id, version=payload.version, inputs=payload.inputs,
|
||||
confirm=payload.confirm, pairs=payload.pairs), principal=current_principal())
|
||||
except ledger_reconcile.ReconcileRefused as e:
|
||||
raise CodedHTTPException(e.status_code, e.message, e.code) from e
|
||||
return {"run_id": run.id, "reconciled": ledger_reconcile.reconciled_view(run)}
|
||||
|
||||
@@ -718,6 +718,61 @@ async def apps_settings(session: AsyncSession) -> dict[str, Any]:
|
||||
return apps_values(row.value if row else None)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 口径账本：按口径自动正式出具的开关、每日上限，口径比对的时限（docs/design/next-round/02-口径账本.md 做法 5、2）
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
#: 设置里 ledger 这一组。auto_issue：问数据的答案整题等价于受管口径（Agent 独立写的 SQL 被证明等价、版本受管、图里没有人工
|
||||
#: 节点和写入）时自动按口径发起一次正式运行（拍板 L1；不满足的只给一键）。issue_daily_limit：每天最多按口径正式出具几次
|
||||
#: （自动的、一键的合计，0 表示不再出具）：正式运行含报告撰写的模型费用。match_timeout_s：一次运行的口径比对最多算几秒，
|
||||
#: 超时的照实记没比完（ledger/match.py）。召回的上限在第 4 波加。只有管理员能改（整个设置接口的改动都要管理员）
|
||||
LEDGER_GROUP = "ledger"
|
||||
LEDGER_DEFAULTS: dict[str, Any] = {"auto_issue": True, "issue_daily_limit": 20, "match_timeout_s": 8}
|
||||
LEDGER_SETTING_LABEL = {"auto_issue": "自动按口径正式出具", "issue_daily_limit": "按口径正式出具每日上限",
|
||||
"match_timeout_s": "口径比对时限（秒）"}
|
||||
#: 时限的范围：太短比不完一份口径，太长占着比对的线程
|
||||
LEDGER_TIMEOUT_RANGE = (1, 60)
|
||||
|
||||
|
||||
def _timeout_ok(value: Any) -> bool:
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool) \
|
||||
and LEDGER_TIMEOUT_RANGE[0] <= value <= LEDGER_TIMEOUT_RANGE[1]
|
||||
|
||||
|
||||
def ledger_problem(value: Any) -> str | None:
|
||||
"""设置页写进来的 ledger 这一组有什么问题，没问题返回 None。"""
|
||||
if not isinstance(value, dict):
|
||||
return f"「口径账本」设置格式有误，需要是一个对象，当前为「{value}」"
|
||||
for key, item in value.items():
|
||||
if key not in LEDGER_DEFAULTS:
|
||||
return (f"「口径账本」设置中不支持「{key}」，可设置的项："
|
||||
+ "、".join(f"「{LEDGER_SETTING_LABEL[k]}」" for k in LEDGER_DEFAULTS))
|
||||
if key == "auto_issue" and not isinstance(item, bool):
|
||||
return f"「{LEDGER_SETTING_LABEL[key]}」只能是开或关，当前为「{item}」"
|
||||
if key == "issue_daily_limit" and not _daily_limit_ok(item):
|
||||
return (f"「{LEDGER_SETTING_LABEL[key]}」需要是 0 到 {APPS_DAILY_LIMIT_MAX:,} 之间的整数（0 表示不再按口径出具），"
|
||||
f"当前为「{item}」")
|
||||
if key == "match_timeout_s" and not _timeout_ok(item):
|
||||
return (f"「{LEDGER_SETTING_LABEL[key]}」需要是 {LEDGER_TIMEOUT_RANGE[0]} 到 {LEDGER_TIMEOUT_RANGE[1]} 之间的数，"
|
||||
f"当前为「{item}」")
|
||||
return None
|
||||
|
||||
|
||||
def ledger_values(stored: Any) -> dict[str, Any]:
|
||||
"""库里存的（或刚写进来的）ledger 一组 → 补齐默认；存坏了的项按默认（自动出具的开关存坏了按关，宁可少出具）。"""
|
||||
stored = stored if isinstance(stored, dict) else {}
|
||||
auto, limit, timeout = stored.get("auto_issue"), stored.get("issue_daily_limit"), stored.get("match_timeout_s")
|
||||
return {"auto_issue": auto if isinstance(auto, bool) else (LEDGER_DEFAULTS["auto_issue"] if auto is None else False),
|
||||
"issue_daily_limit": limit if _daily_limit_ok(limit) else LEDGER_DEFAULTS["issue_daily_limit"],
|
||||
"match_timeout_s": timeout if _timeout_ok(timeout) else LEDGER_DEFAULTS["match_timeout_s"]}
|
||||
|
||||
|
||||
async def ledger_settings(session: AsyncSession) -> dict[str, Any]:
|
||||
"""设置里的 ledger 一组，没存过的补默认。按口径出具、比对从这里读。"""
|
||||
row = await session.get(Setting, LEDGER_GROUP)
|
||||
return ledger_values(row.value if row else None)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 用户设置
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -763,6 +818,8 @@ DEFAULT_SETTINGS: dict[str, Any] = {
|
||||
CATALOG_DRAFT_GROUP: catalog_draft_values(None),
|
||||
# 应用：按需视图的每日上限、使用者能不能建个人应用（管理员在设置页改）
|
||||
APPS_GROUP: apps_values(None),
|
||||
# 口径账本：自动按口径正式出具的开关、每日上限，口径比对的时限
|
||||
LEDGER_GROUP: ledger_values(None),
|
||||
}
|
||||
|
||||
|
||||
@@ -825,6 +882,7 @@ async def get_settings(session: AsyncSession = Depends(get_session)) -> dict[str
|
||||
merged[PUBLIC_URL_GROUP] = public_url_values(stored.get(NOTIFY_GROUP))
|
||||
merged[CATALOG_DRAFT_GROUP] = catalog_draft_values(stored.get(CATALOG_DRAFT_GROUP))
|
||||
merged[APPS_GROUP] = apps_values(stored.get(APPS_GROUP))
|
||||
merged[LEDGER_GROUP] = ledger_values(stored.get(LEDGER_GROUP))
|
||||
for key, value in stored.items():
|
||||
merged.setdefault(key, value)
|
||||
return merged
|
||||
@@ -924,6 +982,10 @@ async def put_settings(
|
||||
if problem := apps_problem(payload.values[APPS_GROUP]):
|
||||
raise HTTPException(422, problem)
|
||||
payload.values[APPS_GROUP] = apps_values(payload.values[APPS_GROUP])
|
||||
if LEDGER_GROUP in payload.values:
|
||||
if problem := ledger_problem(payload.values[LEDGER_GROUP]):
|
||||
raise HTTPException(422, problem)
|
||||
payload.values[LEDGER_GROUP] = ledger_values(payload.values[LEDGER_GROUP])
|
||||
if found := await _provider_problem(session, payload.values):
|
||||
raise CodedHTTPException(422, found[1], PROVIDER_INVALID, field=found[0])
|
||||
for key, value in payload.values.items():
|
||||
|
||||
@@ -107,19 +107,26 @@ async def _upgrade(session: AsyncSession, app: Any, version: AppVersion, view_id
|
||||
|
||||
async def _side_effect(session: AsyncSession, state: Any) -> str | None:
|
||||
"""新版本的图补算时会不会碰到人或外部系统：会的交回一句原因（接在「「周报 v3」」后面），不会的返回 None。
|
||||
作为流程固定的一律算；别的按图判（graph_side_effect）。"""
|
||||
if state.pin.get("use") == "process":
|
||||
return "是作为流程固定的"
|
||||
return await graph_side_effect(session, state.graph, level=state.level)
|
||||
|
||||
|
||||
async def graph_side_effect(session: AsyncSession, graph_json: Any, *, level: str | None = None) -> str | None:
|
||||
"""一张图替人重跑一次会不会碰到人或外部系统：会的交回一句原因（接在「「周报 v3」」后面），不会的返回 None。按新口径
|
||||
补算（这里）和按口径正式出具（ledger/issue.py：自动出具、一键出具都不代人做这些事）共用这一个判断。
|
||||
|
||||
判断沿用应用检查对流程钉住项的那一套（apps/lint._lint_process）：人工节点、写外部系统的能力
|
||||
（capabilities.registry.writes_externally，和运行时、发布检查同一个判断）、process.* 问题（子工作流先读好，
|
||||
process.human_in_subgraph 才查得到）。"""
|
||||
process.human_in_subgraph 才查得到）。level 是那一版发布时的级别（门禁按它查），没有按 published。"""
|
||||
from app.capabilities.registry import effective_effect, get_capability, writes_externally
|
||||
from app.engine.governance import publish_issues
|
||||
from app.engine.lint_process import load_subgraphs, with_subgraphs
|
||||
from app.engine.schema import GraphSpec, NodeType
|
||||
|
||||
if state.pin.get("use") == "process":
|
||||
return "是作为流程固定的"
|
||||
try:
|
||||
graph = GraphSpec.model_validate(state.graph)
|
||||
graph = GraphSpec.model_validate(graph_json)
|
||||
except ValueError:
|
||||
return "的图无法解析"
|
||||
if humans := [n.title for n in graph.nodes if n.type == NodeType.HUMAN]:
|
||||
@@ -132,7 +139,7 @@ async def _side_effect(session: AsyncSession, state: Any) -> str | None:
|
||||
if cap is not None and writes_externally(cap.name, effective_effect(cap), cap.endpoint_id):
|
||||
return f"的「{node.title}」会写外部系统（{name}）"
|
||||
with with_subgraphs(await load_subgraphs(session, graph)):
|
||||
issues = publish_issues(graph, level=state.level or "published")
|
||||
issues = publish_issues(graph, level=level or "published")
|
||||
if process := [i for i in issues if (i.code or "").startswith("process.")]:
|
||||
return f"按流程使用不安全：{process[0].message.rstrip('。')}"
|
||||
return None
|
||||
|
||||
@@ -131,10 +131,11 @@ class Pinned:
|
||||
|
||||
@dataclass
|
||||
class Block:
|
||||
"""派生图里的一段：照抄（或这次执行过）的取数节点 + 一张口径卡。rename 是拷出来的节点改了名的映射（不带前缀时为空）。"""
|
||||
"""派生图里的一段：照抄（或这次执行过）的取数节点 + 一张口径卡。rename 是拷出来的节点改了名的映射（不带前缀时为空）。
|
||||
card_id 为 None 的段没有口径卡（对账里「这次的取数」那一段只有一条查询）。"""
|
||||
|
||||
nodes: list[dict[str, Any]]
|
||||
card_id: str
|
||||
card_id: str | None
|
||||
rename: dict[str, str] = field(default_factory=dict)
|
||||
#: 这一段出自哪里（记进 DerivedGraph.origin）
|
||||
source: dict[str, Any] | None = None
|
||||
@@ -300,9 +301,11 @@ def _free_id(wanted: str, taken: set[str]) -> str:
|
||||
|
||||
|
||||
def assemble(kind: str, blocks: Sequence[Block], *, fields: Sequence[Mapping[str, Any]] = (),
|
||||
defaults: Mapping[str, Any] | None = None, tail: Sequence[dict[str, Any]] = ()) -> DerivedGraph:
|
||||
defaults: Mapping[str, Any] | None = None, tail: Sequence[dict[str, Any]] = (),
|
||||
contract: Mapping[str, Any] | None = None) -> DerivedGraph:
|
||||
"""入口 → 各段（首尾相接）→ tail（对账卡、召回卡）→ 出口。出口每张口径卡（含 tail 里的 metrics 节点）一个字段，
|
||||
取那张卡的指标清单。几段的节点 id 撞了直接报错：两份口径放进同一张图要带前缀（caliber_block 的 prefix）。"""
|
||||
取那张卡的指标清单；contract 给了就写成出口的出具契约（对账写 metrics_from 指向对账卡，不加报告、不调模型）。几段的
|
||||
节点 id 撞了直接报错：两份口径放进同一张图要带前缀（caliber_block 的 prefix）。"""
|
||||
if kind not in KINDS:
|
||||
raise ValueError(f"不认识的派生图种类：{kind}")
|
||||
body: list[dict[str, Any]] = [n for b in blocks for n in b.nodes] + [copy.deepcopy(n) for n in tail]
|
||||
@@ -313,15 +316,18 @@ def assemble(kind: str, blocks: Sequence[Block], *, fields: Sequence[Mapping[str
|
||||
taken = set(ids)
|
||||
entry = _free_id(ENTRY_ID, taken)
|
||||
exit_ = _free_id(EXIT_ID, taken | {entry})
|
||||
cards = [b.card_id for b in blocks] + [str(n["id"]) for n in tail if n.get("type") == "metrics"]
|
||||
cards = [b.card_id for b in blocks if b.card_id] + [str(n["id"]) for n in tail if n.get("type") == "metrics"]
|
||||
titles = {str(n["id"]): str((n.get("data") or {}).get("label") or n["id"]) for n in body}
|
||||
outputs, seen = [], set()
|
||||
for cid in cards:
|
||||
name = _free_id(titles.get(cid, cid), seen)
|
||||
seen.add(name)
|
||||
outputs.append({"name": name, "value": f"{{{{ nodes.{cid}.text }}}}"})
|
||||
exit_config: dict[str, Any] = {"fields": outputs}
|
||||
if contract:
|
||||
exit_config["contract"] = copy.deepcopy(dict(contract))
|
||||
chain = [_node(entry, "input", "入口", fields=[copy.deepcopy(dict(f)) for f in fields]), *body,
|
||||
_node(exit_, "output", "出口", fields=outputs)]
|
||||
_node(exit_, "output", "出口", **exit_config)]
|
||||
for i, node in enumerate(chain):
|
||||
node["position"] = {"x": i * _STEP_X, "y": 0}
|
||||
graph: dict[str, Any] = {
|
||||
@@ -335,6 +341,45 @@ def assemble(kind: str, blocks: Sequence[Block], *, fields: Sequence[Mapping[str
|
||||
entry_fields=tuple(copy.deepcopy(dict(f)) for f in fields), origin=origin)
|
||||
|
||||
|
||||
#: 对账图里「这次的取数」节点、对账卡、多行时按键对齐的合并查询的 id（细案 02 做法 6）
|
||||
OURS_ID = "ours_q1"
|
||||
RECONCILE_CARD = "reconcile"
|
||||
ALIGN_ID = "align"
|
||||
|
||||
|
||||
def reconcile_prefix(pinned: Pinned) -> str:
|
||||
"""对账图里口径那一段的前缀：口径的节点 id 和 ours_q1 / align / reconcile / 入口出口撞了才带 cal_，不撞不带。"""
|
||||
reserved = {OURS_ID, ALIGN_ID, RECONCILE_CARD, ENTRY_ID, EXIT_ID}
|
||||
clash = any(str(n.get("id")) in reserved for n in pinned.fetches) or pinned.node_id in reserved
|
||||
return "cal_" if clash else ""
|
||||
|
||||
|
||||
def ours_block(sql: str, tool: str, *, label: str = "这次的取数") -> Block:
|
||||
"""对账里「这次的取数」：这次执行过的 SQL 原文，放进一个调用工具节点（id ours_q1）。没有口径卡。"""
|
||||
return Block(nodes=[_node(OURS_ID, "tool", label, tool=tool, args={"sql": sql})], card_id=None,
|
||||
source={"kind": "executed", "node_id": OURS_ID})
|
||||
|
||||
|
||||
def reconcile_graph(pinned: Pinned, ours: Block, tail: Sequence[dict[str, Any]], *,
|
||||
inputs: Mapping[str, Any]) -> DerivedGraph:
|
||||
"""口径对账（细案 02 做法 6）：入口（口径的入口字段，缺省值是这次的参数）→ 这次的取数 → 口径的取数（模板里保留
|
||||
{{ input.x }}，运行时按入口渲染）→ 口径卡（caliber_from 固定那一版）→ tail（多行时的对齐合并查询、对账卡）→ 出口
|
||||
（契约 metrics_from 指向对账卡；不加报告、不调模型）。两边在同一次运行里先后查，差值在对账卡里算。
|
||||
|
||||
口径的节点 id 和 ours_q1 / align / reconcile 撞了的，口径那一段带前缀 cal_（改名映射照常写进 caliber_from.nodes）。
|
||||
同样的输入永远得到同一张图。"""
|
||||
block = caliber_block(pinned, prefix=reconcile_prefix(pinned))
|
||||
fields = []
|
||||
for f in pinned.entry_fields:
|
||||
field_ = copy.deepcopy(dict(f))
|
||||
if field_.get("name") in inputs:
|
||||
field_["default"] = inputs[field_["name"]]
|
||||
fields.append(field_)
|
||||
card = next((n for n in tail if n.get("type") == "metrics" and n.get("id") == RECONCILE_CARD), None)
|
||||
return assemble("reconcile", [ours, block], fields=fields, defaults=pinned.defaults, tail=tail,
|
||||
contract={"metrics_from": RECONCILE_CARD} if card is not None else None)
|
||||
|
||||
|
||||
def caliber_probe(pinned: Pinned, *, upgrade_policy: str | None = None) -> DerivedGraph:
|
||||
"""口径探针：一份口径的取数 + 口径卡（评测口径题的期望，细案 01 做法 8）。节点 id 和原图一样，入口字段照抄。
|
||||
同样的 pinned 永远得到同一张图（确定性）。"""
|
||||
@@ -342,6 +387,7 @@ def caliber_probe(pinned: Pinned, *, upgrade_policy: str | None = None) -> Deriv
|
||||
fields=pinned.entry_fields, defaults=pinned.defaults)
|
||||
|
||||
|
||||
__all__ = ["Block", "DerivedGraph", "ENTRY_ID", "EXIT_ID", "GONE", "KINDS", "NOT_CARD", "NOT_REUSABLE", "NotDerivable",
|
||||
"Pinned", "TEMPLATE", "assemble", "caliber_block", "caliber_probe", "entry_fields", "load_pinned",
|
||||
__all__ = ["ALIGN_ID", "Block", "DerivedGraph", "ENTRY_ID", "EXIT_ID", "GONE", "KINDS", "NOT_CARD", "NOT_REUSABLE",
|
||||
"NotDerivable", "OURS_ID", "Pinned", "RECONCILE_CARD", "TEMPLATE", "assemble", "caliber_block",
|
||||
"caliber_probe", "entry_fields", "load_pinned", "ours_block", "reconcile_graph", "reconcile_prefix", "render_inputs",
|
||||
"template_problem"]
|
||||
|
||||
@@ -0,0 +1,372 @@
|
||||
"""按口径正式出具：路径一、二a（docs/design/next-round/02-口径账本.md 做法 5，拍板 L1、L5）。
|
||||
|
||||
问数据的答案整题等价于一份已发布口径（ledger/labels.whole_of：主报告里每个数都来自同一口径的同一版、参数一致）时，可以
|
||||
按那份口径发起一次正式运行——同一个问题，从此有了一份按已发布版本出具的、完整走过门禁的答案。
|
||||
|
||||
**参数只从比对工件里取。** 发起用的输入 = 那一版入口字段的缺省值 + 比对时绑定出来的参数（whole.binding），再过 apps/forms
|
||||
同一套表单校验；不收客户端传的参数，前端只能选「按哪份口径」——不然谁都能拿一次问数据去替一份正式口径「签字」任意输入。
|
||||
绑定的值是这次运行查到过的遮罩值的，不代填出具。
|
||||
|
||||
**经 launch_run 发起**（正式运行的全部门禁：已发布未退役、升版处置、子工作流固定版本、工具在不在），来源 ledger_issue
|
||||
（服务端专用，engine/origin），origin 记 source_run、conversation_id、turn_id、binding_hash、口径和怎么认的、是不是自动的；
|
||||
按 expect_hash 钉住那一版的 graph_hash（比对时考虑的就是这张图）。出具档位照正式运行原样走。
|
||||
|
||||
**能不能出具**（plan_issue，一键和自动同一套，不满足的照实写原因）：
|
||||
- 这次运行是比对过的探索运行，整题等价于点名的这份口径（工作流、版本都对）；
|
||||
- 那一版已发布、未退役；
|
||||
- 图里没有人工节点、不会写外部系统（apps/backfill.graph_side_effect，和按新口径补算同一个判断）：替人重跑一次的正式
|
||||
运行不该替人去审批、往外部系统写；
|
||||
- 补了缺省值之后表单校验通过（缺入口字段的照实说缺哪个）；
|
||||
- 权限（拍板 L5）：搭建者及以上（关登录时同）都可以；使用者只能按他有权使用的应用固定着的这一版出具（复用应用授权，
|
||||
apps/access），别的写「请搭建者出具」；都要读得到这次运行；
|
||||
- 当天没超上限（设置 ledger.issue_daily_limit，自动的、一键的合计）。
|
||||
**自动出具**（auto_issue，拍板 L1）另要：设置里开着、认法是独立等价（how=independent：Agent 自己写的 SQL 被证明等价；
|
||||
「助手选用」的只给一键——选哪份口径是助手凭问法判断的，由人点一下代替证明）、版本是受管级别。只在运行一封存就比完的
|
||||
那一次触发（ledger/match）。不满足的写进 actions 的 auto_reason。
|
||||
|
||||
**同一运行、同一口径、同一组参数只发起一次**：按 input_key（apps/resolve，补算用的同一个指纹）认，已经发起过、没失败的
|
||||
直接交回那一次。进程内按（运行、工作流、版本）串行，自动和一键同时到也只发一次（单进程部署，同 apps/backfill 的锁）。
|
||||
|
||||
结果挂回答案：GET /api/runs/{id}/ledger 的 issued 列出 origin.source_run 等于这次运行的正式运行，档位文字用 TIER_TEXT。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import weakref
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import Run, Workflow
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
ORIGIN = "ledger_issue"
|
||||
#: 出具被拒的机器码（接口回 409 / 403 / 429 时带它）
|
||||
REFUSED = "ledger_issue_refused"
|
||||
FORBIDDEN = "ledger_issue_forbidden"
|
||||
DAILY_LIMIT = "ledger_issue_daily_limit"
|
||||
#: 算「同一次」时看哪几种状态（失败、取消的可以再发）
|
||||
_LIVE = ("queued", "running", "interrupted", "succeeded")
|
||||
|
||||
#: 不能出具的原因
|
||||
NOT_MATCHED = "这次运行还没有比对口径，比对完才能按口径出具"
|
||||
NOT_ELIGIBLE = "这次运行不能按口径出具：{why}"
|
||||
NOT_WHOLE = "整题没有等价于这份口径：{why}"
|
||||
OTHER_CALIBER = "整题等价的是另一份口径，不是「{name}」v{version}"
|
||||
MASKED_BINDING = "参数里有这次运行查到过的遮罩值，不代填出具"
|
||||
GONE = "口径所在的工作流不存在，可能已被删除"
|
||||
VERSION_GONE = "「{name}」没有 v{version} 这个版本"
|
||||
RETIRED = "「{name}」v{version} 已退役，不再接受新的正式运行"
|
||||
UNPUBLISHED = "「{name}」v{version} 没有发布过，正式运行只能从已发布的版本发起"
|
||||
SIDE_EFFECT = "「{name}」v{version}{reason}，按口径出具不代人做这些事"
|
||||
INPUTS = "按这次的参数补上缺省值后，入口字段还不齐：{problems}"
|
||||
NO_RUN_ACCESS = "没有权限查看这次运行"
|
||||
ASK_BUILDER = "这份口径没有被你能使用的应用固定，不能由你按口径出具，请搭建者出具"
|
||||
OVER_LIMIT = "今天的按口径出具已达到上限（{limit} 次），请明天再试，或请管理员调整上限"
|
||||
#: 不自动出具的原因（能一键出具时写进 actions 的 auto_reason）
|
||||
AUTO_OFF = "设置里关了自动按口径出具"
|
||||
AUTO_COMPOSED = "数来自助手选用的口径，只给一键出具：选用哪份口径是助手按问法判断的，要由人确认"
|
||||
AUTO_LEVEL = "这一版是已发布级别，自动出具只限受管级别的口径"
|
||||
AUTO_RETRO = "这次比对是事后按现在的口径回看的，不自动出具"
|
||||
|
||||
|
||||
class IssueRefused(Exception):
|
||||
def __init__(self, status_code: int, message: str, code: str = REFUSED) -> None:
|
||||
super().__init__(message)
|
||||
self.status_code = status_code
|
||||
self.message = message
|
||||
self.code = code
|
||||
|
||||
|
||||
@dataclass
|
||||
class Plan:
|
||||
"""按一份口径出具的打算：能不能（reason 为空就能）、发起用的输入、为什么不自动（auto_reason，能自动的为空）。"""
|
||||
|
||||
run_id: str
|
||||
workflow_id: str
|
||||
version: int
|
||||
caliber: str
|
||||
how: str | None = None
|
||||
level: str | None = None
|
||||
workflow_name: str = ""
|
||||
inputs: dict[str, Any] = field(default_factory=dict)
|
||||
binding: dict[str, Any] = field(default_factory=dict)
|
||||
expect_hash: str | None = None
|
||||
match_artifact: str | None = None
|
||||
#: 不能出具的原因、HTTP 状态码、机器码
|
||||
reason: str | None = None
|
||||
status: int = 409
|
||||
code: str = REFUSED
|
||||
auto_reason: str | None = None
|
||||
#: 使用者靠哪个应用的授权出具（apps/access）：记进 origin
|
||||
granted_by_app: str | None = None
|
||||
#: 已经发起过的同一次（没失败的）
|
||||
existing: str | None = None
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
return self.reason is None
|
||||
|
||||
def refuse(self, message: str, status: int = 409, code: str = REFUSED) -> "Plan":
|
||||
if self.reason is None:
|
||||
self.reason, self.status, self.code = message, status, code
|
||||
return self
|
||||
|
||||
def action(self) -> dict[str, Any]:
|
||||
"""交给界面的那一项动作（GET 账本的 actions）。"""
|
||||
from app.ledger.labels import LEVEL_TEXT
|
||||
|
||||
out: dict[str, Any] = {"kind": "issue", "workflow_id": self.workflow_id, "version": self.version,
|
||||
"caliber": self.caliber, "workflow_name": self.workflow_name, "how": self.how,
|
||||
"level": self.level, "level_text": LEVEL_TEXT.get(str(self.level or ""), ""),
|
||||
"available": self.available, "auto": self.available and self.auto_reason is None}
|
||||
if self.reason:
|
||||
out["reason"] = self.reason
|
||||
elif self.auto_reason:
|
||||
out["auto_reason"] = self.auto_reason
|
||||
if self.existing:
|
||||
out["issued_run"] = self.existing
|
||||
return out
|
||||
|
||||
|
||||
def _local_midnight() -> datetime:
|
||||
return datetime.now().astimezone().replace(hour=0, minute=0, second=0, microsecond=0).astimezone(timezone.utc)
|
||||
|
||||
|
||||
async def issued_today(session: Any) -> int:
|
||||
return int(await session.scalar(select(func.count(Run.id)).where(
|
||||
Run.origin_kind == ORIGIN, Run.created_at >= _local_midnight())) or 0)
|
||||
|
||||
|
||||
async def issued_runs(session: Any, run_id: str) -> list[Run]:
|
||||
"""这次运行按口径出具过的正式运行（origin.source_run 是它），新的在前。origin_kind 有索引，先按它筛再看 origin。"""
|
||||
rows = (await session.execute(select(Run).where(Run.origin_kind == ORIGIN).order_by(Run.created_at.desc()))).scalars()
|
||||
return [r for r in rows if isinstance(r.origin, dict) and r.origin.get("source_run") == run_id]
|
||||
|
||||
|
||||
def issued_view(run: Run) -> dict[str, Any]:
|
||||
"""GET 账本 issued 里的一项：哪一次正式运行、状态、出具档位（TIER_TEXT，完整出具 / 降档出具 / 不予出具）。"""
|
||||
from app.apps.resolve import TIER_TEXT
|
||||
|
||||
output = run.output if isinstance(run.output, dict) else {}
|
||||
issuance = output.get("_issuance") if isinstance(output.get("_issuance"), dict) else {}
|
||||
tier = issuance.get("tier")
|
||||
origin = run.origin if isinstance(run.origin, dict) else {}
|
||||
return {"run_id": run.id, "status": run.status, "end_reason": getattr(run, "end_reason", None),
|
||||
"workflow_id": run.workflow_id, "version": run.version,
|
||||
"tier": tier if tier in TIER_TEXT else None, "tier_text": TIER_TEXT.get(str(tier)),
|
||||
"auto": bool(origin.get("auto")),
|
||||
"created_at": run.created_at.isoformat() if run.created_at else None}
|
||||
|
||||
|
||||
async def _starter(session: Any, run: Run) -> Any:
|
||||
"""自动出具时按谁的权限判（拍板 L5）：这次运行的发起人。关登录时同管理员；账号不在了的按使用者算（宁可少出具）。"""
|
||||
from app.api.deps import Principal
|
||||
from app.auth import current_mode
|
||||
from app.auth.sessions import principal_of
|
||||
from app.db.models import User
|
||||
|
||||
mode = current_mode()
|
||||
if mode == "off":
|
||||
return Principal(username=run.started_by, role="admin", verified=False, mode="off", via="off")
|
||||
user = (await session.execute(select(User).where(User.username == run.started_by))).scalar_one_or_none() \
|
||||
if run.started_by else None
|
||||
if user is None:
|
||||
return Principal(username=run.started_by, role="user", verified=False, mode=mode)
|
||||
return await principal_of(session, user, mode)
|
||||
|
||||
|
||||
async def _granting_app(session: Any, principal: Any, workflow_id: str, version: int) -> str | None:
|
||||
"""使用者能按这一版出具，靠的是哪个应用的授权：固定着这一版、他至少能使用的应用（apps/access）。没有为 None。"""
|
||||
from app.apps.access import app_role, at_least
|
||||
from app.apps.versions import pinning_apps
|
||||
from app.db.models_apps import App
|
||||
|
||||
for item in await pinning_apps(session, workflow_id, version):
|
||||
app = await session.get(App, item["app_id"])
|
||||
if app is not None and at_least(await app_role(session, principal, app), "use"):
|
||||
return app.id
|
||||
return None
|
||||
|
||||
|
||||
async def plan_issue(session: Any, run: Run | None, content: dict[str, Any] | None, workflow_id: str, version: int, *,
|
||||
principal: Any, hide: Any = None, artifact: str | None = None, retro: bool = False) -> Plan:
|
||||
"""按 workflow_id@version 出具这次运行的打算（见模块说明）。content 是这次运行最新的比对工件内容（读不出为 None），
|
||||
hide 判一个绑定值是不是这次运行查到过的遮罩值。只读，不发起。"""
|
||||
from app.api.deps import can_read_run
|
||||
from app.api.settings import ledger_settings
|
||||
from app.apps import versions as app_versions
|
||||
from app.apps.backfill import graph_side_effect
|
||||
from app.apps.forms import input_schema, validate_form
|
||||
from app.apps.resolve import input_key
|
||||
from app.ledger import match as ledger_match
|
||||
from app.ledger.derive import entry_fields
|
||||
from app.ledger.labels import caliber_key
|
||||
|
||||
plan = Plan(run_id=run.id if run else "", workflow_id=workflow_id, version=int(version),
|
||||
caliber=caliber_key(workflow_id, int(version)), match_artifact=artifact)
|
||||
if run is None:
|
||||
return plan.refuse(NOT_ELIGIBLE.format(why=ledger_match.NOT_FOUND), 404)
|
||||
if principal.mode != "off" and not await can_read_run(principal, run):
|
||||
return plan.refuse(NO_RUN_ACCESS, 403, FORBIDDEN)
|
||||
if (why := ledger_match.ineligible(run)) is not None:
|
||||
return plan.refuse(NOT_ELIGIBLE.format(why=why))
|
||||
if content is None:
|
||||
return plan.refuse(NOT_MATCHED)
|
||||
whole = content.get("whole") or {}
|
||||
workflow = await session.get(Workflow, workflow_id)
|
||||
plan.workflow_name = workflow.name if workflow is not None else ""
|
||||
if not whole.get("equivalent"):
|
||||
return plan.refuse(NOT_WHOLE.format(why=whole.get("reason") or "没有比对结论"))
|
||||
if (whole.get("workflow_id"), whole.get("version")) != (workflow_id, int(version)):
|
||||
return plan.refuse(OTHER_CALIBER.format(name=plan.workflow_name or workflow_id, version=version))
|
||||
plan.how, plan.level = whole.get("how"), whole.get("level")
|
||||
plan.binding = dict(whole.get("binding") or {})
|
||||
if hide is not None and any(hide(v) for v in plan.binding.values()):
|
||||
return plan.refuse(MASKED_BINDING)
|
||||
if workflow is None:
|
||||
return plan.refuse(GONE, 404)
|
||||
row = await app_versions.version_row(session, workflow_id, int(version))
|
||||
if row is None:
|
||||
return plan.refuse(VERSION_GONE.format(name=workflow.name, version=version), 404)
|
||||
if row.retired_at is not None:
|
||||
return plan.refuse(RETIRED.format(name=workflow.name, version=version))
|
||||
if not app_versions.is_published(row, current=workflow.published_version):
|
||||
return plan.refuse(UNPUBLISHED.format(name=workflow.name, version=version))
|
||||
plan.level = row.level or plan.level
|
||||
plan.expect_hash = row.graph_hash
|
||||
if reason := await graph_side_effect(session, row.graph, level=row.level):
|
||||
return plan.refuse(SIDE_EFFECT.format(name=workflow.name, version=version, reason=reason))
|
||||
fields = list(entry_fields(row.graph or {}))
|
||||
values = {str(f["name"]): f.get("default") for f in fields if f.get("default") not in (None, "")}
|
||||
values.update({k: v for k, v in plan.binding.items() if v not in (None, "")})
|
||||
plan.inputs = values
|
||||
if errors := validate_form(input_schema(fields), values):
|
||||
return plan.refuse(INPUTS.format(problems="；".join(e["message"] for e in errors)))
|
||||
if principal.mode != "off" and not principal.at_least("builder"):
|
||||
plan.granted_by_app = await _granting_app(session, principal, workflow_id, int(version))
|
||||
if plan.granted_by_app is None:
|
||||
return plan.refuse(ASK_BUILDER, 403, FORBIDDEN)
|
||||
key = input_key(values)
|
||||
plan.existing = next((r.id for r in await issued_runs(session, run.id)
|
||||
if (r.workflow_id, r.version) == (workflow_id, int(version)) and r.status in _LIVE
|
||||
and input_key(r.input) == key), None)
|
||||
settings = await ledger_settings(session)
|
||||
if plan.existing is None and await issued_today(session) >= int(settings["issue_daily_limit"]):
|
||||
return plan.refuse(OVER_LIMIT.format(limit=settings["issue_daily_limit"]), 429, DAILY_LIMIT)
|
||||
plan.auto_reason = (AUTO_OFF if not settings["auto_issue"] else AUTO_COMPOSED if plan.how != "independent"
|
||||
else AUTO_LEVEL if plan.level != "governed" else AUTO_RETRO if retro else None)
|
||||
return plan
|
||||
|
||||
|
||||
async def _conversation_of(session: Any, run: Run) -> dict[str, str]:
|
||||
"""这次运行出自哪段对话的哪一轮：运行来源里封存的（origin_ref，第 2 波）；老运行退回 conversation_turns.run_id。"""
|
||||
from app.db.models import ConversationTurn
|
||||
|
||||
origin = run.origin if isinstance(run.origin, dict) else {}
|
||||
if origin.get("conversation_id") and origin.get("turn_id"):
|
||||
return {"conversation_id": str(origin["conversation_id"]), "turn_id": str(origin["turn_id"])}
|
||||
turn = (await session.execute(select(ConversationTurn).where(ConversationTurn.run_id == run.id)
|
||||
.order_by(ConversationTurn.created_at.desc()))).scalars().first()
|
||||
return {"conversation_id": turn.conversation_id, "turn_id": turn.id} if turn is not None else {}
|
||||
|
||||
|
||||
#: 同一（运行、工作流、版本）的出具串行：{事件循环: {键: 锁}}（asyncio.Lock 不能跨循环用，测试每个用例一个循环）
|
||||
_locks: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[tuple[str, str, int], asyncio.Lock]]" = \
|
||||
weakref.WeakKeyDictionary()
|
||||
|
||||
|
||||
def _lock(key: tuple[str, str, int]) -> asyncio.Lock:
|
||||
group = _locks.setdefault(asyncio.get_running_loop(), {})
|
||||
return group.setdefault(key, asyncio.Lock())
|
||||
|
||||
|
||||
async def _hider(run_id: str) -> Any:
|
||||
from app.api.evidence import _sealed
|
||||
from app.ledger.view import _hider as hider_of
|
||||
|
||||
return (await hider_of(await _sealed(run_id))).hide
|
||||
|
||||
|
||||
async def issue(run_id: str, workflow_id: str, version: int, *, principal: Any, auto: bool = False) -> tuple[Run, bool]:
|
||||
"""按 workflow_id@version 出具这次运行：(正式运行, 是不是这一次新发起的)。不能出具抛 IssueRefused；已经发起过同一次的
|
||||
交回那一次（第二项为 False）。auto：自动出具（自动的条件也要满足）。"""
|
||||
from app.engine.launch import LaunchRefused, launch_run
|
||||
from app.ledger import store
|
||||
|
||||
async with _lock((run_id, workflow_id, int(version))):
|
||||
recorded = await store.latest(run_id)
|
||||
hide = await _hider(run_id)
|
||||
async with SessionLocal() as session:
|
||||
run = await session.get(Run, run_id)
|
||||
plan = await plan_issue(session, run, recorded.content if recorded else None, workflow_id, version,
|
||||
principal=principal, hide=hide, artifact=recorded.artifact if recorded else None,
|
||||
retro=bool(recorded and recorded.retro))
|
||||
if plan.reason is not None:
|
||||
raise IssueRefused(plan.status, plan.reason, plan.code)
|
||||
if plan.existing is not None:
|
||||
return await session.get(Run, plan.existing), False
|
||||
if auto and plan.auto_reason is not None:
|
||||
raise IssueRefused(409, plan.auto_reason)
|
||||
from app.apps.resolve import input_key
|
||||
|
||||
origin = {"source_run": run_id, **await _conversation_of(session, run), "caliber": plan.caliber,
|
||||
"how": plan.how, "binding_hash": input_key(plan.binding), "match": plan.match_artifact,
|
||||
"auto": auto, **({"granted_by_app": plan.granted_by_app} if plan.granted_by_app else {})}
|
||||
actor = run.started_by if auto else principal.username
|
||||
try:
|
||||
launched = await launch_run(session, workflow_id=workflow_id, run_class="formal", version=int(version),
|
||||
input_payload=plan.inputs, actor=actor, origin_kind=ORIGIN, origin=origin,
|
||||
expect_hash=plan.expect_hash)
|
||||
except LaunchRefused as e:
|
||||
raise IssueRefused(e.status_code, e.message, e.code or REFUSED) from e
|
||||
return launched, True
|
||||
|
||||
|
||||
async def plans_for(run_id: str, content: dict[str, Any] | None, *, principal: Any, hide: Any, artifact: str | None,
|
||||
retro: bool) -> list[Plan]:
|
||||
"""GET 账本的 actions 里出具那几项：整题等价的那一份口径能不能出具、能不能自动（整题不等价的没有这一项）。"""
|
||||
whole = (content or {}).get("whole") or {}
|
||||
if not whole.get("equivalent") or not whole.get("workflow_id") or whole.get("version") is None:
|
||||
return []
|
||||
async with SessionLocal() as session:
|
||||
run = await session.get(Run, run_id)
|
||||
return [await plan_issue(session, run, content, str(whole["workflow_id"]), int(whole["version"]),
|
||||
principal=principal, hide=hide, artifact=artifact, retro=retro)]
|
||||
|
||||
|
||||
async def auto_issue(run_id: str) -> str | None:
|
||||
"""比对完、整题等价时看能不能自动出具（拍板 L1），能就发起；返回新发起的正式运行 id。不能的、出错的只记日志：自动出具
|
||||
是辅助步骤，不影响运行本身和答案（照实说为什么没自动出具在 GET 账本的 actions 里）。"""
|
||||
from app.ledger import store
|
||||
|
||||
try:
|
||||
recorded = await store.latest(run_id)
|
||||
whole = ((recorded.content if recorded else None) or {}).get("whole") or {}
|
||||
if not whole.get("equivalent"):
|
||||
return None
|
||||
async with SessionLocal() as session:
|
||||
run = await session.get(Run, run_id)
|
||||
if run is None:
|
||||
return None
|
||||
principal = await _starter(session, run)
|
||||
launched, fresh = await issue(run_id, str(whole["workflow_id"]), int(whole["version"]), principal=principal,
|
||||
auto=True)
|
||||
return launched.id if fresh else None
|
||||
except IssueRefused as e:
|
||||
logger.info("运行 %s 没有自动按口径出具：%s", run_id, e.message)
|
||||
return None
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception: # noqa: BLE001 - 辅助步骤
|
||||
logger.exception("运行 %s 自动按口径出具出错", run_id)
|
||||
return None
|
||||
|
||||
|
||||
__all__ = ["DAILY_LIMIT", "FORBIDDEN", "ORIGIN", "REFUSED", "IssueRefused", "Plan", "auto_issue", "issue",
|
||||
"issued_runs", "issued_today", "issued_view", "plan_issue", "plans_for"]
|
||||
@@ -19,7 +19,7 @@ MAX_CANDIDATES 份（按工作流、版本、节点排好的先后）。
|
||||
schema 用这条查询当时冻结的表结构（表结构快照），列的中文名取同一份快照里冻结的数据目录。只读封存范围内的快照（api/evidence
|
||||
的 _sealed 读法），读不出、被改过的快照不比，记 skipped。
|
||||
|
||||
**资源。** 每次运行限时 TIMEOUT_S（缺省 8 秒），在线程里跑，同时最多 CONCURRENCY（2）个（照 scheduler/auto_compare 的做法：
|
||||
**资源。** 每次运行限时（设置 ledger.match_timeout_s，缺省 TIMEOUT_S = 8 秒），在线程里跑，同时最多 CONCURRENCY（2）个（照 scheduler/auto_compare 的做法：
|
||||
信号量按事件循环分开，线程池两个线程）。时限按绑定逐组看：过了时限，还没比的口径记 skipped（比对超时）；整次比对卡在线程里
|
||||
过了时限再加 GRACE_S 还没回来，整次记 skipped。解析失败同样照实记 skipped 和原因。都只记日志和 skipped，不影响运行本身。
|
||||
|
||||
@@ -477,15 +477,21 @@ 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:
|
||||
async def match_run(run_id: str, *, trigger: str = "sealed", timeout: float | None = None) -> str | None:
|
||||
"""比对一次运行，存工件、追加 ledger.matched；返回工件 id。不比对（不是探索运行、没成功、没封存、没有查询）、比对期间
|
||||
运行换了封存时返回 None。比对本身出的问题都记成 skipped，不抛。"""
|
||||
运行换了封存时返回 None。比对本身出的问题都记成 skipped，不抛。timeout 不给按设置（ledger.match_timeout_s）。
|
||||
|
||||
运行一封存就比的（trigger=sealed）比完接着看能不能按口径自动正式出具（ledger/issue.auto_issue，拍板 L1）：事后补比、
|
||||
按需比的不自动出具——那是按现在的口径回看老答案，不该替人补发正式运行。"""
|
||||
from app.api.evidence import _sealed
|
||||
from app.api.settings import ledger_settings
|
||||
|
||||
async with SessionLocal() as session:
|
||||
run = await session.get(Run, run_id)
|
||||
if ineligible(run) is not None:
|
||||
return None
|
||||
if timeout is None:
|
||||
timeout = float((await ledger_settings(session))["match_timeout_s"])
|
||||
sealed = await _sealed(run_id)
|
||||
if not sealed.queries:
|
||||
return None
|
||||
@@ -500,7 +506,12 @@ async def match_run(run_id: str, *, trigger: str = "sealed", timeout: float = TI
|
||||
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)
|
||||
artifact = await store.save(run_id, content, seal=(sealed.run_manifest_seq, sealed.manifest_hash), trigger=trigger)
|
||||
if artifact is not None and trigger == "sealed" and (content.get("whole") or {}).get("equivalent"):
|
||||
from app.ledger import issue
|
||||
|
||||
await issue.auto_issue(run_id)
|
||||
return artifact
|
||||
|
||||
|
||||
def schedule(run_id: str, *, trigger: str = "sealed") -> bool:
|
||||
|
||||
@@ -0,0 +1,518 @@
|
||||
"""口径对账：路径二b，差在哪、差多少（docs/design/next-round/02-口径账本.md 做法 6，总方案 X3）。
|
||||
|
||||
**差在哪**在比对时就有（ledger/match：explain_sql_diff 的条目，这次 / 口径两边说），不跑查询。**差多少**要点开才跑：
|
||||
按那一版口径确定性派生一张只读小图（ledger/derive.reconcile_graph，全站唯一的派生图生成器），跑一次探索运行
|
||||
（来源 reconcile，服务端专用、系统派生、不进使用统计）：
|
||||
|
||||
入口（口径的入口字段，缺省值是这次的参数）→「这次的取数」ours_q1（执行过的 SQL 原文）→ 口径的取数（模板里保留
|
||||
{{ input.x }}）→ 口径卡（caliber_from 固定那一版）→（多行时）按维度对齐的合并查询 align → 对账卡（这次、口径、差值、
|
||||
相对差）→ 出口（契约 metrics_from 指向对账卡；不加报告、不调模型）
|
||||
|
||||
两边在同一次运行里查，差值在对账卡里算（叙述层无算术，也不在 Python 里算：compare.py 是同一工作流两次运行配对用的）。
|
||||
面板上每个数都点得开 /evidence/metrics 的出处链。
|
||||
|
||||
**参数。** 默认用比对时绑定出来的参数（缺的补缺省值）。参数是启发得来的（绑定在 heuristic 一级配上）、有没绑上的、补完
|
||||
缺省值表单还不齐、或者里面有遮罩值时，要人先核对：动作里写 needs_check 和原因，发起时要带 confirm 和 inputs（按口径入口
|
||||
的字段校验，ledger/compose.check_inputs 同一套），界面写「参数按这次的 SQL 推断，请核对」。
|
||||
|
||||
**列怎么配。** 只有一行结果时：口径卡上的指标式就是取数第一行某一列（cell(nodes.取数, 0, '列')）的，按那一列在口径取数
|
||||
结果里的位置配这次结果同一位置的列（等价时已实测列的先后计；不等价的两条 SQL 选择列表一样时位置照样对得上）。差异
|
||||
有兜底说明（fallback：两条 SQL 对不上结构）、列数不同、一个都配不上时不猜，要人选（pairs：{ours: 这次的列, metric: 口径卡
|
||||
上的指标 id}），这个选择记进对账运行的 origin。
|
||||
|
||||
**多行结果**（这次的结果不止一行）：被报告引用了的那几行（最多 MAX_ROWS 行），按维度列（不是数的列）经合并查询和口径
|
||||
取数的结果按键对齐（LEFT JOIN，按引用的先后排），每个被引用的行、每个被引用的列出一组对账指标。两边的列按位置配（列数
|
||||
不同时不猜，照实拒绝）。维度列是遮罩列、或者这次的结果里合并键不唯一时照实拒绝：前者对齐要用到遮住的值，后者对不准
|
||||
是哪一行。被引用的列是遮罩列的同样不对账。
|
||||
|
||||
只读：派生图只查库、不写；使用者发起时带上 MASK_RESULTS / READ_ONLY（同问数据），搭建者也带 READ_ONLY。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import Run, Workflow
|
||||
|
||||
ORIGIN = "reconcile"
|
||||
#: 被拒的机器码
|
||||
REFUSED = "ledger_reconcile_refused"
|
||||
CHECK_INPUTS = "ledger_reconcile_check_inputs"
|
||||
NEEDS_PAIRS = "ledger_reconcile_needs_pairs"
|
||||
FORBIDDEN = "ledger_reconcile_forbidden"
|
||||
#: 多行对账最多几行、几列（每行每列出四个对账指标）
|
||||
MAX_ROWS = 10
|
||||
MAX_COLUMNS = 5
|
||||
|
||||
#: 给人看的说法
|
||||
NOT_FOUND = "运行记录不存在"
|
||||
NO_ACCESS = "没有权限查看这次运行"
|
||||
NOT_MATCHED = "这次运行还没有比对口径，比对完才能对账"
|
||||
NO_QUERY = "这次运行的比对记录里没有这条查询"
|
||||
NOT_RELATED = "这条查询没有和「{name}」v{version} 比出差异（只有相关口径才对账）"
|
||||
NO_CARD = "「{name}」v{version} 里没有读这个取数节点的口径卡，对不了账"
|
||||
SNAPSHOT = "这条查询的快照读不出来（文件缺失或与哈希不一致），对不了账"
|
||||
CHECK_REASON = "参数按这次的 SQL 推断，请核对"
|
||||
CHECK_MASKED = "参数里有遮罩值，请核对参数"
|
||||
CHECK_INCOMPLETE = "按这次的 SQL 没能推断出全部参数，请补全并核对"
|
||||
CONFIRM_FIRST = "参数按这次的 SQL 推断，发起前请先核对参数"
|
||||
PAIRS_FALLBACK = "两条 SQL 的结构对不上，认不出这次的哪一列对应口径的哪个指标，请选择要对照的指标"
|
||||
PAIRS_NONE = "认不出这次的哪一列对应口径的哪个指标，请选择要对照的指标"
|
||||
PAIR_COLUMN = "这次的结果里没有列「{column}」"
|
||||
PAIR_MASKED = "列「{column}」设置了遮罩，不能对账"
|
||||
PAIR_METRIC = "口径卡「{card}」上没有指标「{metric}」"
|
||||
PAIRS_SHAPE = "要对照的指标写成 [{ours: 这次的列, metric: 口径卡上的指标 id}]"
|
||||
ROWS_NONE = "报告里没有引用这条查询的格，不知道要对哪几行"
|
||||
ROWS_NO_KEY = "这次的结果有多行，但没有可以用来对齐的维度列（不是数的列）"
|
||||
ROWS_MASKED_KEY = "用来对齐的维度列「{column}」设置了遮罩，不能对账"
|
||||
ROWS_DUPLICATE = "这次的结果里按维度列（{columns}）有重复的行，对不准是哪一行"
|
||||
ROWS_COLUMNS = "这次的结果和口径的取数列数不同（{ours} 列对 {theirs} 列），按位置对不上"
|
||||
ROWS_CALIBER_COLUMNS = "认不出口径取数的结果列，按位置对不上"
|
||||
|
||||
|
||||
class ReconcileRefused(Exception):
|
||||
def __init__(self, status_code: int, message: str, code: str = REFUSED) -> None:
|
||||
super().__init__(message)
|
||||
self.status_code = status_code
|
||||
self.message = message
|
||||
self.code = code
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 动作：GET 账本里能对的账
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
def _needs_check(cand: dict[str, Any], fields: list[dict[str, Any]], hide: Any) -> str | None:
|
||||
"""参数要不要人先核对：要的回原因（见模块说明），不要的为 None。fields 是比对工件里记的入口字段（name、required、
|
||||
has_default）。"""
|
||||
binding = cand.get("binding") or {}
|
||||
if hide is not None and any(hide(v) for v in binding.values()):
|
||||
return CHECK_MASKED
|
||||
if cand.get("unbound") or not binding and any(f.get("name") for f in fields):
|
||||
return CHECK_INCOMPLETE
|
||||
if any(f.get("required") and not f.get("has_default") and f.get("name") not in binding for f in fields):
|
||||
return CHECK_INCOMPLETE
|
||||
if cand.get("via") in (None, "heuristic"):
|
||||
return CHECK_REASON
|
||||
return None
|
||||
|
||||
|
||||
def actions_of(content: dict[str, Any], names: dict[str, str], hide: Any = None) -> list[dict[str, Any]]:
|
||||
"""比对工件里每条有相关口径的查询、每份相关口径一项「看差多少」：{kind: reconcile, query, workflow_id, version, node_id,
|
||||
needs_check, check_reason?, binding（遮过的）}。界面据此摆按钮、要不要先核对参数。"""
|
||||
from app.ledger.labels import LEVEL_TEXT, binding_view
|
||||
|
||||
hide = hide or (lambda _value: False)
|
||||
fields = {c.get("caliber"): c.get("fields") or [] for c in content.get("calibers") or []}
|
||||
out: list[dict[str, Any]] = []
|
||||
for query in content.get("queries") or []:
|
||||
if query.get("status") not in ("related",):
|
||||
continue
|
||||
for cand in query.get("candidates") or []:
|
||||
if cand.get("status") != "related" or cand.get("comparable") is False:
|
||||
continue
|
||||
why = _needs_check(cand, list(fields.get(cand.get("caliber")) or []), hide)
|
||||
item = {"kind": "reconcile", "query": query.get("artifact"), "caliber": cand.get("caliber"),
|
||||
"workflow_id": cand.get("workflow_id"), "version": cand.get("version"),
|
||||
"node_id": cand.get("node_id"), "workflow_name": names.get(str(cand.get("workflow_id") or "")),
|
||||
"level_text": LEVEL_TEXT.get(str(cand.get("level") or ""), ""),
|
||||
"needs_check": why is not None, "binding": binding_view(cand.get("binding"), hide),
|
||||
"fallback": bool((cand.get("diff") or {}).get("fallback"))}
|
||||
if why is not None:
|
||||
item["check_reason"] = why
|
||||
out.append(item)
|
||||
return out
|
||||
|
||||
|
||||
async def reconciled_runs(session: Any, run_id: str) -> list[Run]:
|
||||
"""对这次运行对过的账（origin.source_run 是它），新的在前。"""
|
||||
rows = (await session.execute(select(Run).where(Run.origin_kind == ORIGIN).order_by(Run.created_at.desc()))).scalars()
|
||||
return [r for r in rows if isinstance(r.origin, dict) and r.origin.get("source_run") == run_id]
|
||||
|
||||
|
||||
def reconciled_view(run: Run) -> dict[str, Any]:
|
||||
origin = run.origin if isinstance(run.origin, dict) else {}
|
||||
return {"run_id": run.id, "status": run.status, "query": origin.get("query"), "caliber": origin.get("caliber"),
|
||||
"workflow_id": origin.get("workflow_id"), "version": origin.get("version"),
|
||||
"inputs_from": origin.get("inputs_from"), "rows": origin.get("rows"),
|
||||
"created_at": run.created_at.isoformat() if run.created_at else None}
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 派生图
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
@dataclass
|
||||
class Pair:
|
||||
"""一组对照：这次结果的哪一列，对口径卡上的哪个指标（metric）或口径取数结果的哪一列（column）。"""
|
||||
|
||||
ours: str
|
||||
metric: str | None = None
|
||||
column: str | None = None
|
||||
#: 口径卡上那个指标的位置（metric 配对时）
|
||||
index: int | None = None
|
||||
name: str = ""
|
||||
unit: str = ""
|
||||
decimals: Any = None
|
||||
|
||||
def describe(self) -> dict[str, Any]:
|
||||
return {"ours": self.ours, **({"metric": self.metric} if self.metric else {}),
|
||||
**({"caliber": self.column} if self.column else {})}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Plan:
|
||||
"""一次对账要派生的东西（纯计算的结果，测试锁住确定性）。"""
|
||||
|
||||
pairs: list[Pair]
|
||||
rows: list[int] | None = None
|
||||
keys: list[tuple[str, str]] = field(default_factory=list)
|
||||
key_values: list[tuple[Any, ...]] = field(default_factory=list)
|
||||
|
||||
|
||||
def _lit(value: Any) -> str:
|
||||
"""一个值写进合并查询（SQLite）的字面量。"""
|
||||
if value is None:
|
||||
return "NULL"
|
||||
if isinstance(value, bool):
|
||||
return "1" if value else "0"
|
||||
if isinstance(value, (int, float)):
|
||||
return repr(value)
|
||||
return "'" + str(value).replace("'", "''") + "'"
|
||||
|
||||
|
||||
def _ident(name: str) -> str:
|
||||
return '"' + str(name).replace('"', '""') + '"'
|
||||
|
||||
|
||||
def _equals(alias: str, column: str, value: Any) -> str:
|
||||
return f"{alias}.{_ident(column)} IS NULL" if value is None else f"{alias}.{_ident(column)} = {_lit(value)}"
|
||||
|
||||
|
||||
def align_sql(keys: list[tuple[str, str]], pairs: list[Pair], values: list[tuple[Any, ...]]) -> str:
|
||||
"""多行对账的合并查询：这次（o）LEFT JOIN 口径（c）按维度列对齐，只留被引用的那几行，按引用的先后排。"""
|
||||
select_ = [f"o.{_ident(ours)} AS {_ident(ours)}" for ours, _ in keys]
|
||||
select_ += [f"o.{_ident(p.ours)} AS {_ident(f'ours_{j}')}, c.{_ident(p.column or '')} AS {_ident(f'cal_{j}')}"
|
||||
for j, p in enumerate(pairs)]
|
||||
on = " AND ".join(f"o.{_ident(ours)} = c.{_ident(theirs)}" for ours, theirs in keys)
|
||||
rows = ["(" + " AND ".join(_equals("o", ours, v) for (ours, _), v in zip(keys, row)) + ")" for row in values]
|
||||
order = " ".join(f"WHEN {cond} THEN {i}" for i, cond in enumerate(rows))
|
||||
return (f"SELECT {', '.join(select_)} FROM o LEFT JOIN c ON {on} WHERE {' OR '.join(rows)} "
|
||||
f"ORDER BY CASE {order} END")
|
||||
|
||||
|
||||
def _metric(mid: str, name: str, expression: str, *, unit: str = "", decimals: Any = None,
|
||||
fmt: str | None = None) -> dict[str, Any]:
|
||||
out: dict[str, Any] = {"id": mid, "name": name, "unit": unit, "expression": expression}
|
||||
if decimals is not None:
|
||||
out["decimals"] = decimals
|
||||
if fmt:
|
||||
out["format"] = fmt
|
||||
return out
|
||||
|
||||
|
||||
def _four(n: str, label: str, ours: str, theirs: str, pair: Pair) -> list[dict[str, Any]]:
|
||||
"""一组对照的四个对账指标：这次、口径、差值（这次 − 口径）、相对差（差值 ÷ 口径，口径是 0 时没有值）。"""
|
||||
diff = f"({ours}) - ({theirs})"
|
||||
return [
|
||||
_metric(f"ours_{n}", f"{label} · 这次", ours, unit=pair.unit, decimals=pair.decimals),
|
||||
_metric(f"cal_{n}", f"{label} · 口径", theirs, unit=pair.unit, decimals=pair.decimals),
|
||||
_metric(f"diff_{n}", f"{label} · 差值", diff, unit=pair.unit, decimals=pair.decimals),
|
||||
_metric(f"rel_{n}", f"{label} · 相对差", f"({diff}) / ({theirs}) if ({theirs}) else None", decimals=4,
|
||||
fmt="percent_of_ratio"),
|
||||
]
|
||||
|
||||
|
||||
def tail_nodes(plan: Plan, *, card_id: str, fetch_id: str) -> list[dict[str, Any]]:
|
||||
"""派生图的 tail：（多行时）对齐的合并查询 + 对账卡。card_id、fetch_id 是图里口径卡、口径取数节点的 id。"""
|
||||
from app.ledger.derive import ALIGN_ID, OURS_ID, RECONCILE_CARD, _node
|
||||
|
||||
metrics: list[dict[str, Any]] = []
|
||||
nodes: list[dict[str, Any]] = []
|
||||
if plan.rows is None:
|
||||
for j, pair in enumerate(plan.pairs):
|
||||
ours = f"cell(nodes.{OURS_ID}, 0, {pair.ours!r})"
|
||||
theirs = (f"nodes.{card_id}.metrics[{pair.index}].value" if pair.metric is not None
|
||||
else f"cell(nodes.{fetch_id}, 0, {pair.column!r})")
|
||||
metrics += _four(str(j), pair.name or pair.ours, ours, theirs, pair)
|
||||
else:
|
||||
nodes.append(_node(ALIGN_ID, "merge", "按维度对齐", inputs={"o": OURS_ID, "c": fetch_id},
|
||||
sql=align_sql(plan.keys, plan.pairs, plan.key_values)))
|
||||
for i, row in enumerate(plan.key_values):
|
||||
where = "，".join(f"{ours}={v}" for (ours, _), v in zip(plan.keys, row))
|
||||
for j, pair in enumerate(plan.pairs):
|
||||
metrics += _four(f"{i}_{j}", f"{where} · {pair.name or pair.ours}",
|
||||
f"cell(nodes.{ALIGN_ID}, {i}, 'ours_{j}')", f"cell(nodes.{ALIGN_ID}, {i}, 'cal_{j}')",
|
||||
pair)
|
||||
nodes.append(_node(RECONCILE_CARD, "metrics", "对账卡", caliber="口径对账", caliber_version="v1", on_missing="null",
|
||||
metrics=metrics))
|
||||
return nodes
|
||||
|
||||
|
||||
def _numeric(rows: list[Any], index: int) -> bool:
|
||||
values = [r[index] for r in rows if isinstance(r, (list, tuple)) and index < len(r) and r[index] is not None]
|
||||
return bool(values) and all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in values)
|
||||
|
||||
|
||||
def _plain_cell(expression: str, fetch: str) -> str | None:
|
||||
"""指标式就是取数第一行某一列（cell(nodes.<fetch>, 0, '列')）时交回列名，别的写法为 None。"""
|
||||
import ast
|
||||
|
||||
from app.engine.expressions import cell_parts
|
||||
|
||||
parts = cell_parts(expression.strip())
|
||||
if parts is None or parts[0] != f"nodes.{fetch}" or parts[1].strip() != "0":
|
||||
return None
|
||||
try:
|
||||
column = ast.literal_eval(parts[2])
|
||||
except (ValueError, SyntaxError):
|
||||
return None
|
||||
return column if isinstance(column, str) else None
|
||||
|
||||
|
||||
def plan_pairs(*, ours_columns: list[str], ours_rows: list[Any], hidden: set[str], caliber_columns: list[str] | None,
|
||||
metrics: list[dict[str, Any]], fetch: str, card: str, fallback: bool, chosen: Any,
|
||||
referenced: list[tuple[int, str]]) -> Plan:
|
||||
"""配列、（多行时）定行和键：交回 Plan，配不了的抛 ReconcileRefused（见模块说明）。纯计算。"""
|
||||
def check_column(column: str) -> None:
|
||||
if column not in ours_columns:
|
||||
raise ReconcileRefused(422, PAIR_COLUMN.format(column=column))
|
||||
if column.lower() in hidden:
|
||||
raise ReconcileRefused(409, PAIR_MASKED.format(column=column))
|
||||
|
||||
by_id = {str(m.get("id")): (i, m) for i, m in enumerate(metrics)}
|
||||
multi = len(ours_rows) > 1
|
||||
if multi:
|
||||
return _plan_rows(ours_columns, ours_rows, hidden, caliber_columns, referenced, check_column)
|
||||
if chosen is not None:
|
||||
if not isinstance(chosen, list) or not chosen or not all(isinstance(p, dict) for p in chosen):
|
||||
raise ReconcileRefused(422, PAIRS_SHAPE, NEEDS_PAIRS)
|
||||
pairs = []
|
||||
for item in chosen:
|
||||
ours, metric = str(item.get("ours") or ""), str(item.get("metric") or "")
|
||||
check_column(ours)
|
||||
if metric not in by_id:
|
||||
raise ReconcileRefused(422, PAIR_METRIC.format(card=card, metric=metric))
|
||||
index, m = by_id[metric]
|
||||
pairs.append(Pair(ours=ours, metric=metric, index=index, name=str(m.get("name") or metric),
|
||||
unit=str(m.get("unit") or ""), decimals=m.get("decimals")))
|
||||
return Plan(pairs=pairs)
|
||||
if fallback:
|
||||
raise ReconcileRefused(409, PAIRS_FALLBACK, NEEDS_PAIRS)
|
||||
pairs = []
|
||||
if caliber_columns is not None and len(caliber_columns) == len(ours_columns):
|
||||
for index, m in enumerate(metrics):
|
||||
column = _plain_cell(str(m.get("expression") or ""), fetch)
|
||||
if column is None or column not in caliber_columns:
|
||||
continue
|
||||
ours = ours_columns[caliber_columns.index(column)]
|
||||
if ours.lower() in hidden:
|
||||
continue
|
||||
pairs.append(Pair(ours=ours, metric=str(m.get("id")), index=index, name=str(m.get("name") or m.get("id")),
|
||||
unit=str(m.get("unit") or ""), decimals=m.get("decimals")))
|
||||
if not pairs:
|
||||
raise ReconcileRefused(409, PAIRS_NONE, NEEDS_PAIRS)
|
||||
return Plan(pairs=pairs)
|
||||
|
||||
|
||||
def _plan_rows(ours_columns: list[str], rows: list[Any], hidden: set[str], caliber_columns: list[str] | None,
|
||||
referenced: list[tuple[int, str]], check_column: Any) -> Plan:
|
||||
if caliber_columns is None:
|
||||
raise ReconcileRefused(409, ROWS_CALIBER_COLUMNS)
|
||||
if len(caliber_columns) != len(ours_columns):
|
||||
raise ReconcileRefused(409, ROWS_COLUMNS.format(ours=len(ours_columns), theirs=len(caliber_columns)))
|
||||
keys = [i for i in range(len(ours_columns)) if not _numeric(rows, i)]
|
||||
if not keys:
|
||||
raise ReconcileRefused(409, ROWS_NO_KEY)
|
||||
if masked := [ours_columns[i] for i in keys if ours_columns[i].lower() in hidden]:
|
||||
raise ReconcileRefused(409, ROWS_MASKED_KEY.format(column=masked[0]))
|
||||
tuples = [tuple(r[i] if isinstance(r, (list, tuple)) and i < len(r) else None for i in keys) for r in rows]
|
||||
if len(set(map(json.dumps, [list(t) for t in tuples]))) != len(tuples):
|
||||
raise ReconcileRefused(409, ROWS_DUPLICATE.format(columns="、".join(ours_columns[i] for i in keys)))
|
||||
wanted_rows = list(dict.fromkeys(r for r, _ in referenced if isinstance(r, int) and 0 <= r < len(rows)))[:MAX_ROWS]
|
||||
columns = list(dict.fromkeys(c for _, c in referenced if c in ours_columns and ours_columns.index(c) not in keys))
|
||||
if not wanted_rows or not columns:
|
||||
raise ReconcileRefused(409, ROWS_NONE)
|
||||
for column in columns:
|
||||
check_column(column)
|
||||
pairs = [Pair(ours=c, column=caliber_columns[ours_columns.index(c)], name=c) for c in columns[:MAX_COLUMNS]]
|
||||
return Plan(pairs=pairs, rows=wanted_rows, keys=[(ours_columns[i], caliber_columns[i]) for i in keys],
|
||||
key_values=[tuples[r] for r in wanted_rows])
|
||||
|
||||
|
||||
def _referenced(sealed: Any, artifact: str) -> list[tuple[int, str]]:
|
||||
"""报告里引用了这条查询的哪几格：[(行, 列)]，按正文的先后（单元格、表格里的格）。"""
|
||||
from app.api.evidence import _reports
|
||||
from app.engine.evidence import iter_units
|
||||
|
||||
out: list[tuple[int, str]] = []
|
||||
for report in _reports(sealed):
|
||||
doc = report.doc if report.hash_ok is True else None
|
||||
if not isinstance(doc, dict):
|
||||
continue
|
||||
catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {}
|
||||
for _, unit in iter_units(doc):
|
||||
for seg in unit.get("segments") or []:
|
||||
cite = seg.get("cite") if isinstance(seg, dict) and isinstance(seg.get("cite"), dict) else {}
|
||||
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 {}
|
||||
if cite.get("kind") == "cell" and entry.get("artifact") == artifact \
|
||||
and isinstance(locator.get("row"), int) and isinstance(locator.get("column"), str):
|
||||
out.append((locator["row"], locator["column"]))
|
||||
return out
|
||||
|
||||
|
||||
def _caliber_columns(template: str, inputs: dict[str, Any], fields: list[dict[str, Any]], dialect: str | None,
|
||||
schema: dict[str, Any] | None) -> list[str] | None:
|
||||
"""口径取数按这次的参数渲染之后选出来的列名（按位置）；认不出的为 None。"""
|
||||
from app.data.sql_semantics import _read, _result_columns, _schema_info, render_bound
|
||||
|
||||
try:
|
||||
rendered = render_bound(template, inputs, fields)
|
||||
return _result_columns(rendered, _read(dialect), _schema_info(schema))
|
||||
except Exception: # noqa: BLE001 - 认不出就按认不出说
|
||||
return None
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 发起
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
@dataclass
|
||||
class Request:
|
||||
query: str
|
||||
workflow_id: str
|
||||
version: int
|
||||
inputs: dict[str, Any] | None = None
|
||||
confirm: bool = False
|
||||
pairs: Any = None
|
||||
|
||||
|
||||
async def reconcile(run_id: str, req: Request, *, principal: Any) -> Run:
|
||||
"""对这次运行的一条查询和一份相关口径对一次账：派生只读图、发起探索运行（来源 reconcile）。不能对的抛 ReconcileRefused。"""
|
||||
from app.api.deps import can_read_run
|
||||
from app.api.evidence import _Masks, _sealed
|
||||
from app.core.artifact_store import load
|
||||
from app.engine.launch import LaunchRefused, launch_run
|
||||
from app.ledger import compose, derive, store
|
||||
from app.ledger.issue import _conversation_of
|
||||
from app.ledger.labels import caliber_key
|
||||
from app.ledger.match import _frozen
|
||||
from app.ledger.view import _hider
|
||||
from app.tools.datasource import MASK_RESULTS_OPTION, QUERY_PREFIX, READ_ONLY_OPTION
|
||||
|
||||
async with SessionLocal() as session:
|
||||
run = await session.get(Run, run_id)
|
||||
if run is None:
|
||||
raise ReconcileRefused(404, NOT_FOUND)
|
||||
if principal.mode != "off" and not await can_read_run(principal, run):
|
||||
raise ReconcileRefused(403, NO_ACCESS, FORBIDDEN)
|
||||
recorded = await store.latest(run_id)
|
||||
if recorded is None or recorded.content is None:
|
||||
raise ReconcileRefused(409, NOT_MATCHED)
|
||||
content = recorded.content
|
||||
key = caliber_key(req.workflow_id, int(req.version))
|
||||
query = next((q for q in content.get("queries") or [] if q.get("artifact") == req.query), None)
|
||||
if query is None:
|
||||
raise ReconcileRefused(404, NO_QUERY)
|
||||
async with SessionLocal() as session:
|
||||
workflow = await session.get(Workflow, req.workflow_id)
|
||||
name = workflow.name if workflow is not None else req.workflow_id
|
||||
cand = next((c for c in query.get("candidates") or [] if c.get("caliber") == key and c.get("status") == "related"),
|
||||
None)
|
||||
if cand is None:
|
||||
raise ReconcileRefused(409, NOT_RELATED.format(name=name, version=req.version))
|
||||
summary = next((c for c in content.get("calibers") or [] if c.get("caliber") == key), {})
|
||||
pinned = None
|
||||
async with SessionLocal() as session:
|
||||
for card_id in summary.get("cards") or []:
|
||||
try:
|
||||
candidate = await derive.load_pinned(session, req.workflow_id, int(req.version), str(card_id))
|
||||
except derive.NotDerivable:
|
||||
continue
|
||||
if any(node == cand.get("node_id") for node, _, _ in candidate.reads):
|
||||
pinned = candidate
|
||||
break
|
||||
if pinned is None:
|
||||
raise ReconcileRefused(409, NO_CARD.format(name=name, version=req.version))
|
||||
|
||||
# 参数：比对时绑定出来的（补缺省值），要核对的要人带着核对过的参数再来
|
||||
sealed = await _sealed(run_id)
|
||||
hider = await _hider(sealed)
|
||||
why = _needs_check(cand, list(summary.get("fields") or []), hider.hide)
|
||||
if req.confirm and req.inputs is not None:
|
||||
try:
|
||||
inputs = compose.check_inputs(pinned, req.inputs)
|
||||
except compose.ComposeRefused as e:
|
||||
raise ReconcileRefused(422, e.message, CHECK_INPUTS) from e
|
||||
inputs_from = "checked"
|
||||
elif why is not None:
|
||||
raise ReconcileRefused(409, CONFIRM_FIRST if why == CHECK_REASON else why, CHECK_INPUTS)
|
||||
else:
|
||||
try:
|
||||
inputs = compose.check_inputs(pinned, dict(cand.get("binding") or {}))
|
||||
except compose.ComposeRefused as e:
|
||||
raise ReconcileRefused(409, f"{CHECK_INCOMPLETE}：{e.message}", CHECK_INPUTS) from e
|
||||
inputs_from = "binding"
|
||||
|
||||
# 这次的那条查询：快照、遮罩列、被引用的格
|
||||
try:
|
||||
snap = load(req.query)
|
||||
except ValueError:
|
||||
snap = None
|
||||
if not isinstance(snap, dict) or not isinstance(snap.get("sql"), str) or req.query not in sealed.queries:
|
||||
raise ReconcileRefused(409, SNAPSHOT)
|
||||
source = str(snap.get("source") or query.get("source") or "")
|
||||
columns = [str(c) for c in snap.get("columns") or []]
|
||||
rows = snap.get("rows") if isinstance(snap.get("rows"), list) else []
|
||||
masks = _Masks(sealed)
|
||||
hidden = set((await masks.sets(req.query, source, snap)).hidden)
|
||||
kind = await masks.kind(source)
|
||||
schema, _ = _frozen(snap.get("schema_artifact"))
|
||||
fetch = next(n for n in pinned.fetches if n["id"] == cand.get("node_id"))
|
||||
template = str(((fetch.get("data") or {}).get("config") or {}).get("args", {}).get("sql") or "")
|
||||
plan = plan_pairs(ours_columns=columns, ours_rows=rows, hidden=hidden,
|
||||
caliber_columns=_caliber_columns(template, inputs, list(pinned.entry_fields), kind, schema),
|
||||
metrics=pinned.metrics, fetch=str(fetch["id"]), card=pinned.card_title,
|
||||
fallback=bool((cand.get("diff") or {}).get("fallback")), chosen=req.pairs,
|
||||
referenced=_referenced(sealed, req.query))
|
||||
graph = build(pinned, sql=snap["sql"], tool=f"{QUERY_PREFIX}{source}", plan=plan, inputs=inputs,
|
||||
fetch=str(fetch["id"]))
|
||||
|
||||
async with SessionLocal() as session:
|
||||
origin = {"source_run": run_id, **await _conversation_of(session, run), "query": req.query, "caliber": key,
|
||||
"workflow_id": req.workflow_id, "version": int(req.version), "inputs_from": inputs_from,
|
||||
"binding_via": cand.get("via"), "pairs": [p.describe() for p in plan.pairs],
|
||||
**({"rows": plan.rows} if plan.rows is not None else {}),
|
||||
**({"pairs_chosen": True} if req.pairs is not None else {}), "derive": graph.origin}
|
||||
options = {READ_ONLY_OPTION: True}
|
||||
if principal.mode != "off" and not principal.at_least("builder"):
|
||||
options[MASK_RESULTS_OPTION] = True
|
||||
try:
|
||||
return await launch_run(session, graph=graph.graph, input_payload=inputs, run_class="exploratory",
|
||||
actor=principal.username, origin_kind=ORIGIN, origin=origin, options=options,
|
||||
workflow_name=f"口径对账：{name} v{req.version}")
|
||||
except LaunchRefused as e:
|
||||
raise ReconcileRefused(e.status_code, e.message, e.code or REFUSED) from e
|
||||
|
||||
|
||||
def build(pinned: Any, *, sql: str, tool: str, plan: Plan, inputs: dict[str, Any], fetch: str) -> Any:
|
||||
"""按 Plan 派生对账图（ledger/derive.reconcile_graph）。fetch 是这条查询比的那个口径取数节点（那一版里的 id）。
|
||||
纯计算，同样的输入同一张图。"""
|
||||
from app.ledger import derive
|
||||
|
||||
prefix = derive.reconcile_prefix(pinned)
|
||||
tail = tail_nodes(plan, card_id=f"{prefix}{pinned.node_id}", fetch_id=f"{prefix}{fetch}")
|
||||
return derive.reconcile_graph(pinned, derive.ours_block(sql, tool), tail, inputs=inputs)
|
||||
|
||||
|
||||
__all__ = ["CHECK_INPUTS", "MAX_COLUMNS", "MAX_ROWS", "NEEDS_PAIRS", "ORIGIN", "REFUSED", "Pair", "Plan",
|
||||
"ReconcileRefused", "Request", "actions_of", "align_sql", "build", "plan_pairs", "reconcile",
|
||||
"reconciled_runs", "reconciled_view", "tail_nodes"]
|
||||
@@ -4,7 +4,9 @@ ledger 键、报告导出的片段面板，三处读同一份比对工件、按
|
||||
遮罩按读的那一刻重滤（同证据面板，api/evidence._Masks）：绑定的值是这次运行查到过的遮罩值时不给值（也就不能拿它代填出具），
|
||||
差异条目里的 SQL 片段走 _Shown 同一套换法。工作流名按现在的取（工件里不存名字）。
|
||||
|
||||
界面层不放逻辑：标签、整题判断、能做的动作（这一波还没有出具和对账，actions 是空的）都由这里给。
|
||||
界面层不放逻辑：标签、整题判断、能做的动作都由这里给。actions（第 3 波）：整题等价时按那份口径正式出具（能不能、能不能
|
||||
自动、不能的原因，ledger/issue.plan_issue，按看的这个人的权限判）；有相关口径的查询看差多少（口径对账，ledger/reconcile）。
|
||||
issued 是这次运行按口径出具过的正式运行，reconciled 是对过的账（各自的运行、状态）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -122,10 +124,20 @@ async def run_view(run_id: str) -> dict[str, Any]:
|
||||
return {**base, "state": "pending"}
|
||||
if recorded.content is None:
|
||||
return {**base, "state": "unreadable", "reason": recorded.problem, **_stamp(recorded)}
|
||||
from app.api.deps import current_principal
|
||||
from app.ledger import issue as ledger_issue
|
||||
from app.ledger import reconcile as ledger_reconcile
|
||||
|
||||
content = recorded.content
|
||||
sealed = await _sealed(run_id)
|
||||
hider = await _hider(sealed)
|
||||
names = await _names(_workflow_ids(content))
|
||||
principal = current_principal()
|
||||
plans = await ledger_issue.plans_for(run_id, content, principal=principal, hide=hider.hide,
|
||||
artifact=recorded.artifact, retro=recorded.retro)
|
||||
async with SessionLocal() as session:
|
||||
issued = [ledger_issue.issued_view(r) for r in await ledger_issue.issued_runs(session, run_id)]
|
||||
reconciled = [ledger_reconcile.reconciled_view(r) for r in await ledger_reconcile.reconciled_runs(session, run_id)]
|
||||
return {
|
||||
**base, "state": "matched", **_stamp(recorded),
|
||||
"calibers": [{**{k: c.get(k) for k in ("caliber", "workflow_id", "version", "level", "graph_hash")},
|
||||
@@ -136,8 +148,9 @@ async def run_view(run_id: str) -> dict[str, Any]:
|
||||
"reports": _labels_view(content, hider, names),
|
||||
"primary": content.get("primary"),
|
||||
"whole": _whole_view(content.get("whole") or {}, hider, names),
|
||||
# 能做的动作（按口径正式出具、对账）在第 3 波；这一波照实给空的
|
||||
"actions": [],
|
||||
"actions": [*(p.action() for p in plans), *ledger_reconcile.actions_of(content, names, hider.hide)],
|
||||
"issued": issued,
|
||||
"reconciled": reconciled,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -61,7 +61,8 @@ async def test_a_matched_run_gives_labels_and_the_whole_judgement(client, monkey
|
||||
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"
|
||||
# 第 3 波起整题等价时有「按口径出具」这一项（详细的在 tests/test_ledger_issue.py）
|
||||
assert [a["kind"] for a in body["actions"]] == ["issue"] 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()
|
||||
|
||||
@@ -0,0 +1,306 @@
|
||||
"""按口径正式出具（next-round 02 做法 5，拍板 L1、L5，app/ledger/issue.py）。
|
||||
|
||||
真的跑探索运行（虚构店铺库、mock 模型写报告），等后台比对做完，看自动出具、一键出具、拒绝原因、去重：
|
||||
- 整题独立等价于受管口径：比对完自动发起一次正式运行（来源 ledger_issue），origin 记来源运行、口径、绑定哈希，参数是
|
||||
绑定 + 缺省值，按 graph_hash 钉住；GET 账本的 actions 写能自动、issued 列出它；再点一次交回同一次；
|
||||
- 已发布级别、助手选用的：不自动，只给一键（写明为什么），一键照常发起；设置关了自动出具也不自动；
|
||||
- 拒绝写明原因：整题不等价、点名别的口径、版本退役、图里有人工节点、缺入口字段、超过每日上限、使用者没有应用授权；
|
||||
使用者有授权（应用固定着这一版）就能出具，origin 记靠的是哪个应用；
|
||||
- 设置 ledger 一组：缺省值、写坏了拒收。
|
||||
数据是虚构店铺库。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.api.deps import Principal
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import Run, Setting, WorkflowVersion
|
||||
from app.engine.runner import run_manager
|
||||
from app.ledger import issue as ledger_issue
|
||||
from app.ledger import match as ledger_match
|
||||
from app.main import app
|
||||
from tests import ledger_samples as S
|
||||
|
||||
ADMIN = Principal(username=None, role="admin", mode="off", via="off")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def engine_up(tmp_path):
|
||||
await run_manager.setup()
|
||||
await S.fresh_source(tmp_path / "shop.db")
|
||||
# 测试库按进程共用：每日上限数的是今天全部的按口径出具，别的用例发起的也算。每个用例从缺省设置起步、上限放宽，
|
||||
# 要测上限的用例自己改
|
||||
await reset_ledger(issue_daily_limit=100_000)
|
||||
yield
|
||||
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 reset_ledger(**values) -> None:
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(Setting, "ledger")
|
||||
if row is not None:
|
||||
await session.delete(row)
|
||||
if values:
|
||||
session.add(Setting(key="ledger", value=values))
|
||||
await session.commit()
|
||||
|
||||
|
||||
async def set_ledger(**values) -> None:
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(Setting, "ledger")
|
||||
merged = {**(row.value if row else {}), **values}
|
||||
if row:
|
||||
row.value = merged
|
||||
else:
|
||||
session.add(Setting(key="ledger", value=merged))
|
||||
await session.commit()
|
||||
|
||||
|
||||
async def issued(run_id: str) -> list[Run]:
|
||||
async with SessionLocal() as session:
|
||||
return await ledger_issue.issued_runs(session, run_id)
|
||||
|
||||
|
||||
async def settle(run_id: str) -> list[Run]:
|
||||
"""等自动出具发起的正式运行跑完。"""
|
||||
# 自动出具在比对里就发起了（ledger/match.match_run 等它），drain 之后有没有已经定了：只等发起了的跑完
|
||||
for _ in range(200):
|
||||
found = await issued(run_id)
|
||||
if all(r.status in ("succeeded", "failed", "cancelled") for r in found):
|
||||
return found
|
||||
await asyncio.sleep(0.05)
|
||||
return await issued(run_id)
|
||||
|
||||
|
||||
def two_writers(monkeypatch, answer: str = "本周销售额 [[v:Q1.r0.销售额]]。") -> None:
|
||||
"""mock 模型：口径那张图的报告（指令「写周销售」）引用口径卡的指标，问数据那张图的报告照 answer 写。"""
|
||||
from langchain_core.messages import AIMessage
|
||||
|
||||
from app.providers import mock_model
|
||||
|
||||
def decide(self, messages):
|
||||
text = "\n".join(str(getattr(m, "content", "")) for m in messages)
|
||||
return AIMessage(content="周销售 [[m:gmv]]。" if "写周销售" in text else answer)
|
||||
|
||||
monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide)
|
||||
|
||||
|
||||
async def equivalent_run(monkeypatch):
|
||||
from app.engine.runner import run_manager as manager
|
||||
|
||||
two_writers(monkeypatch)
|
||||
run = await manager.start(graph=S.explore_graph(S.EQUIVALENT), input_payload={}, origin_kind="canvas")
|
||||
row = await S.wait(run.id)
|
||||
assert row.status == "succeeded", row.error
|
||||
await ledger_match.drain()
|
||||
return row
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 自动出具
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_an_independent_equivalent_answer_to_a_governed_caliber_is_issued_automatically(client, monkeypatch):
|
||||
wf = await S.publish()
|
||||
row = await equivalent_run(monkeypatch)
|
||||
[formal] = await settle(row.id)
|
||||
assert formal.run_class == "formal" and formal.origin_kind == "ledger_issue" and formal.status == "succeeded"
|
||||
assert (formal.workflow_id, formal.version) == (wf, 1)
|
||||
assert formal.input == {"start": "2026-09-01", "end": "2026-09-08", "store": "all"}
|
||||
async with SessionLocal() as session:
|
||||
version = (await session.execute(select(WorkflowVersion).where(
|
||||
WorkflowVersion.workflow_id == wf, WorkflowVersion.version == 1))).scalar_one()
|
||||
assert formal.version_hash == version.graph_hash
|
||||
origin = formal.origin
|
||||
assert origin["source_run"] == row.id and origin["caliber"] == f"{wf}@1" and origin["how"] == "independent"
|
||||
assert origin["auto"] is True and origin["binding_hash"] and origin["match"]
|
||||
|
||||
body = (await client.get(f"/api/runs/{row.id}/ledger")).json()
|
||||
[action] = [a for a in body["actions"] if a["kind"] == "issue"]
|
||||
assert action["available"] is True and action["auto"] is True and action["issued_run"] == formal.id
|
||||
assert action["level_text"] == "受管" and action["workflow_name"] == "周销售"
|
||||
[item] = body["issued"]
|
||||
assert item["run_id"] == formal.id and item["auto"] is True and item["status"] == "succeeded"
|
||||
|
||||
# 再点一次「按口径出具」：同一运行、同一口径、同一组参数，交回已经发起的那一次
|
||||
again = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1})
|
||||
assert again.status_code == 200 and again.json()["created"] is False and again.json()["run_id"] == formal.id
|
||||
assert len(await issued(row.id)) == 1
|
||||
|
||||
|
||||
async def test_published_level_calibers_only_get_one_click(client, monkeypatch):
|
||||
wf = await S.publish(level="published")
|
||||
row = await equivalent_run(monkeypatch)
|
||||
assert await settle(row.id) == []
|
||||
body = (await client.get(f"/api/runs/{row.id}/ledger")).json()
|
||||
[action] = [a for a in body["actions"] if a["kind"] == "issue"]
|
||||
assert action["available"] is True and action["auto"] is False and action["auto_reason"] == ledger_issue.AUTO_LEVEL
|
||||
res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1})
|
||||
assert res.status_code == 200 and res.json()["created"] is True
|
||||
[formal] = await settle(row.id)
|
||||
assert formal.origin["auto"] is False and formal.status == "succeeded"
|
||||
assert res.json()["issued"]["run_id"] == formal.id
|
||||
|
||||
|
||||
async def test_switching_auto_issue_off_keeps_one_click(client, monkeypatch):
|
||||
await S.publish()
|
||||
await set_ledger(auto_issue=False)
|
||||
row = await equivalent_run(monkeypatch)
|
||||
assert await settle(row.id) == []
|
||||
[action] = (await client.get(f"/api/runs/{row.id}/ledger")).json()["actions"]
|
||||
assert action["auto"] is False and action["auto_reason"] == ledger_issue.AUTO_OFF
|
||||
|
||||
|
||||
async def test_composed_answers_are_never_issued_automatically(monkeypatch):
|
||||
from app.ledger import compose
|
||||
from app.look.calibers import CaliberKind
|
||||
|
||||
wf = await S.publish()
|
||||
async with SessionLocal() as session:
|
||||
kind = await CaliberKind.load(session, compose=True)
|
||||
made = compose.Composer(list(kind.options)).compose(
|
||||
{"op": "use_caliber", "id": "c1", "workflow_id": wf, "version": 1, "metrics": ["gmv"],
|
||||
"inputs": {"start": "2026-09-01", "end": "2026-09-08"}}, taken={"start"})
|
||||
graph = S.chain(S.node("start", "input"), *made.nodes, S.node("write", "report", instructions="写答案"),
|
||||
S.node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}]))
|
||||
two_writers(monkeypatch, "本周销售额 [[m:gmv]]。")
|
||||
row = await S.wait((await run_manager.start(graph=graph, input_payload={}, origin_kind="chat")).id)
|
||||
await ledger_match.drain()
|
||||
assert await settle(row.id) == []
|
||||
async with SessionLocal() as session:
|
||||
recorded = await ledger_match.store.latest(row.id)
|
||||
plan = await ledger_issue.plan_issue(session, await session.get(Run, row.id), recorded.content, wf, 1,
|
||||
principal=ADMIN)
|
||||
assert plan.available and plan.how == "composed" and plan.auto_reason == ledger_issue.AUTO_COMPOSED
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 拒绝
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_refusals_say_why(client, monkeypatch):
|
||||
wf = await S.publish(level="published")
|
||||
other = await S.publish(name="另一份口径", level="published")
|
||||
unrelated = await S.explore(monkeypatch, S.WITH_REFUNDS)
|
||||
res = await client.post(f"/api/ledger/answers/{unrelated.id}/issue", json={"workflow_id": wf, "version": 1})
|
||||
assert res.status_code == 409 and res.json()["code"] == ledger_issue.REFUSED
|
||||
assert res.json()["detail"].startswith("整题没有等价于这份口径")
|
||||
|
||||
row = await equivalent_run(monkeypatch)
|
||||
whole_on = (await ledger_match.store.latest(row.id)).content["whole"]["workflow_id"]
|
||||
named = other if whole_on == wf else wf
|
||||
res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": named, "version": 1})
|
||||
assert res.status_code == 409 and "整题等价的是另一份口径" in res.json()["detail"]
|
||||
|
||||
# 版本退役
|
||||
async with SessionLocal() as session:
|
||||
version = (await session.execute(select(WorkflowVersion).where(
|
||||
WorkflowVersion.workflow_id == whole_on, WorkflowVersion.version == 1))).scalar_one()
|
||||
version.retired_at = datetime.now(timezone.utc)
|
||||
await session.commit()
|
||||
res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": whole_on, "version": 1})
|
||||
assert res.status_code == 409 and "已退役" in res.json()["detail"]
|
||||
|
||||
|
||||
async def test_a_caliber_with_a_human_step_is_not_issued_on_anyones_behalf(client, monkeypatch):
|
||||
graph = S.caliber_graph()
|
||||
graph["nodes"].insert(3, S.node("ok", "human", mode="approve", prompt="确认"))
|
||||
graph["edges"] = [{"source": a["id"], "target": b["id"]} for a, b in zip(graph["nodes"], graph["nodes"][1:])]
|
||||
wf = await S.publish(graph=graph)
|
||||
row = await equivalent_run(monkeypatch)
|
||||
assert await settle(row.id) == []
|
||||
res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1})
|
||||
assert res.status_code == 409 and "人工节点" in res.json()["detail"] and "不代人做这些事" in res.json()["detail"]
|
||||
|
||||
|
||||
async def test_missing_entry_fields_are_named(client, monkeypatch):
|
||||
graph = S.caliber_graph()
|
||||
graph["nodes"][0]["data"]["config"]["fields"].append({"name": "region", "required": True})
|
||||
wf = await S.publish(graph=graph)
|
||||
row = await equivalent_run(monkeypatch)
|
||||
[action] = (await client.get(f"/api/runs/{row.id}/ledger")).json()["actions"]
|
||||
assert action["available"] is False and "请填写「region」" in action["reason"]
|
||||
res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1})
|
||||
assert res.status_code == 409 and "入口字段还不齐" in res.json()["detail"]
|
||||
|
||||
|
||||
async def test_the_daily_limit_counts_every_issue(client, monkeypatch):
|
||||
wf = await S.publish(level="published")
|
||||
await set_ledger(issue_daily_limit=0)
|
||||
row = await equivalent_run(monkeypatch)
|
||||
res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1})
|
||||
assert res.status_code == 429 and res.json()["code"] == ledger_issue.DAILY_LIMIT and "上限（0 次）" in res.json()["detail"]
|
||||
|
||||
|
||||
async def test_users_issue_only_versions_pinned_by_apps_they_can_use(monkeypatch):
|
||||
from app.db.models_apps import App, AppGrant, AppVersion
|
||||
|
||||
wf = await S.publish(level="published")
|
||||
row = await equivalent_run(monkeypatch)
|
||||
async with SessionLocal() as session:
|
||||
run = await session.get(Run, row.id)
|
||||
run.started_by = "carol"
|
||||
await session.commit()
|
||||
carol = Principal(username="carol", role="user", verified=True, mode="local")
|
||||
with pytest.raises(ledger_issue.IssueRefused) as e:
|
||||
await ledger_issue.issue(row.id, wf, 1, principal=carol)
|
||||
assert e.value.status_code == 403 and e.value.message == ledger_issue.ASK_BUILDER
|
||||
# 别人的运行读不到：先说读不到
|
||||
dave = Principal(username="dave", role="user", verified=True, mode="local")
|
||||
with pytest.raises(ledger_issue.IssueRefused) as e:
|
||||
await ledger_issue.issue(row.id, wf, 1, principal=dave)
|
||||
assert e.value.status_code == 403 and e.value.message == ledger_issue.NO_RUN_ACCESS
|
||||
|
||||
async with SessionLocal() as session:
|
||||
board = App(name="周销售看板", published_version=1, created_by="bob")
|
||||
session.add(board)
|
||||
await session.flush()
|
||||
session.add(AppVersion(app_id=board.id, version=1, spec={}, spec_hash="x",
|
||||
pins=[{"alias": "week", "workflow_id": wf, "version": 1}]))
|
||||
session.add(AppGrant(app_id=board.id, principal_kind="user", principal="carol", role="use"))
|
||||
await session.commit()
|
||||
board_id = board.id
|
||||
formal, created = await ledger_issue.issue(row.id, wf, 1, principal=carol)
|
||||
assert created and formal.started_by == "carol" and formal.origin["granted_by_app"] == board_id
|
||||
|
||||
|
||||
async def test_concurrent_requests_issue_once(monkeypatch):
|
||||
wf = await S.publish(level="published")
|
||||
row = await equivalent_run(monkeypatch)
|
||||
results = await asyncio.gather(*(ledger_issue.issue(row.id, wf, 1, principal=ADMIN) for _ in range(3)))
|
||||
assert sorted(created for _, created in results) == [False, False, True]
|
||||
assert len({r.id for r, _ in results}) == 1
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 设置
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_ledger_settings_have_defaults_and_refuse_bad_values(client):
|
||||
await reset_ledger()
|
||||
body = (await client.get("/api/settings")).json()
|
||||
assert body["ledger"] == {"auto_issue": True, "issue_daily_limit": 20, "match_timeout_s": 8}
|
||||
bad = await client.put("/api/settings", json={"values": {"ledger": {"auto_issue": "yes"}}})
|
||||
assert bad.status_code == 422 and "自动按口径正式出具" in bad.json()["detail"]
|
||||
bad = await client.put("/api/settings", json={"values": {"ledger": {"match_timeout_s": 600}}})
|
||||
assert bad.status_code == 422 and "口径比对时限" in bad.json()["detail"]
|
||||
ok = await client.put("/api/settings", json={"values": {"ledger": {"auto_issue": False, "issue_daily_limit": 5,
|
||||
"match_timeout_s": 3}}})
|
||||
assert ok.status_code == 200 and ok.json()["ledger"] == {"auto_issue": False, "issue_daily_limit": 5,
|
||||
"match_timeout_s": 3}
|
||||
@@ -0,0 +1,216 @@
|
||||
"""口径对账：差多少（next-round 02 做法 6，app/ledger/reconcile.py、ledger/derive.reconcile_graph）。
|
||||
|
||||
真的跑探索运行（虚构店铺库、mock 模型写报告）、等后台比对做完，再点开对账：
|
||||
- 动作：有相关口径的查询给「看差多少」；参数是按「同一列、同一运算符」猜的，要先核对（needs_check）；
|
||||
- 不核对直接发起回 409（要核对参数）；核对过的参数发起：派生图是确定的、过 validate、出口契约 metrics_from 指向对账卡、
|
||||
不加报告；跑出来的这次、口径、差值、相对差都对，差值的出处链点得开；GET 账本列出对过的账；统计 E2 数到它；
|
||||
- 认不出哪一列对哪个指标（口径的指标式不是取数的一格）：要人选，选了照记进 origin；
|
||||
- 多行结果：被引用的行按维度列经合并查询对齐，每行一组对账指标；维度是遮罩列、合并键不唯一时照实拒绝。
|
||||
数据是虚构店铺库。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import Run
|
||||
from app.engine.runner import run_manager
|
||||
from app.engine.schema import GraphSpec, validate_graph
|
||||
from app.ledger import match as ledger_match
|
||||
from app.ledger import reconcile as ledger_reconcile
|
||||
from app.main import app
|
||||
from tests import ledger_samples as S
|
||||
|
||||
BY_REGION = ("SELECT region AS 区域, SUM(amount) AS gmv FROM orders WHERE status = 'paid' "
|
||||
"AND paid_at >= '{{ input.start }}' AND paid_at < '{{ input.end }}' GROUP BY region ORDER BY region")
|
||||
BY_REGION_ALL = ("SELECT region AS 区域, SUM(amount) AS gmv FROM orders WHERE paid_at >= '2026-09-01' "
|
||||
"AND paid_at < '2026-09-08' GROUP BY region ORDER BY region")
|
||||
CHECKED = {"start": "2026-09-01", "end": "2026-09-08"}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def engine_up(tmp_path):
|
||||
await run_manager.setup()
|
||||
path = tmp_path / "shop.db"
|
||||
await S.fresh_source(path)
|
||||
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()
|
||||
|
||||
|
||||
async def card_of(run_id: str, node_id: str = "reconcile") -> dict:
|
||||
from app.api.evidence import _sealed, _sealed_cards
|
||||
|
||||
sealed = await _sealed(run_id)
|
||||
[(_, card)] = [(a, c) for n, a, c in _sealed_cards(sealed) if n == node_id]
|
||||
return {m["id"]: m for m in card["metrics"]}
|
||||
|
||||
|
||||
async def reconcile(client, run_id: str, body: dict) -> Run:
|
||||
res = await client.post(f"/api/ledger/answers/{run_id}/reconcile", json=body)
|
||||
assert res.status_code == 201, res.text
|
||||
return await S.wait(res.json()["run_id"])
|
||||
|
||||
|
||||
async def test_reconcile_the_refunds_answer_against_the_weekly_caliber(client, monkeypatch):
|
||||
wf = await S.publish()
|
||||
row = await S.explore(monkeypatch, S.WITH_REFUNDS)
|
||||
body = await ledger_of(client, row.id)
|
||||
[action] = [a for a in body["actions"] if a["kind"] == "reconcile"]
|
||||
assert action["workflow_id"] == wf and action["version"] == 1 and action["node_id"] == "fetch"
|
||||
assert action["needs_check"] is True and action["check_reason"] == ledger_reconcile.CHECK_REASON
|
||||
assert action["binding"] == [{"field": "end", "value": "2026-09-08"}, {"field": "start", "value": "2026-09-01"}]
|
||||
query = action["query"]
|
||||
|
||||
# 参数是猜的：不核对就发起，照实拒绝
|
||||
res = await client.post(f"/api/ledger/answers/{row.id}/reconcile", json={"query": query, "workflow_id": wf,
|
||||
"version": 1})
|
||||
assert res.status_code == 409 and res.json()["code"] == ledger_reconcile.CHECK_INPUTS
|
||||
assert res.json()["detail"] == ledger_reconcile.CONFIRM_FIRST
|
||||
|
||||
done = await reconcile(client, row.id, {"query": query, "workflow_id": wf, "version": 1, "inputs": CHECKED,
|
||||
"confirm": True})
|
||||
assert done.status == "succeeded", done.error
|
||||
assert done.origin_kind == "reconcile" and done.run_class == "exploratory" and done.workflow_id is None
|
||||
origin = done.origin
|
||||
assert origin["source_run"] == row.id and origin["query"] == query and origin["inputs_from"] == "checked"
|
||||
assert origin["pairs"] == [{"ours": "gmv", "metric": "gmv"}] and origin["caliber"] == f"{wf}@1"
|
||||
assert done.input == {"start": "2026-09-01", "end": "2026-09-08", "store": "all"}
|
||||
|
||||
graph = done.graph
|
||||
ids = [n["id"] for n in graph["nodes"]]
|
||||
assert ids == ["derived_in", "ours_q1", "fetch", "card", "reconcile", "derived_out"]
|
||||
nodes = {n["id"]: n for n in graph["nodes"]}
|
||||
assert nodes["ours_q1"]["data"]["config"]["args"]["sql"] == S.WITH_REFUNDS
|
||||
assert "{{ input.start }}" in nodes["fetch"]["data"]["config"]["args"]["sql"], "口径的取数保留模板，运行时按入口渲染"
|
||||
assert nodes["derived_out"]["data"]["config"]["contract"] == {"metrics_from": "reconcile"}
|
||||
assert not [n for n in graph["nodes"] if n["type"] in ("report", "llm", "agent")]
|
||||
assert not [i for i in validate_graph(GraphSpec.model_validate(graph)).issues if i.level == "error"]
|
||||
|
||||
metrics = await card_of(done.id)
|
||||
assert metrics["ours_0"]["value"] == 221.0 and metrics["cal_0"]["value"] == 170.5
|
||||
assert metrics["diff_0"]["value"] == 50.5 and metrics["rel_0"]["value"] == 0.2962
|
||||
assert metrics["rel_0"]["rendered"] == "29.62%" and metrics["diff_0"]["rendered"] == "50.50元"
|
||||
# 三个数各自点得开：差值的出处链追到两边的查询
|
||||
res = await client.get(f"/api/runs/{done.id}/evidence/metrics/reconcile/diff_0")
|
||||
assert res.status_code == 200, res.text
|
||||
inputs = [s for s in res.json()["chain"] if s["step"] == "input"]
|
||||
assert {i["node_id"] for i in inputs} == {"ours_q1", "card"}
|
||||
|
||||
listed = (await ledger_of(client, row.id))["reconciled"]
|
||||
assert [r["run_id"] for r in listed] == [done.id] and listed[0]["inputs_from"] == "checked"
|
||||
|
||||
# 统计 E2：对账运行数到了（它不进使用统计，E2 专门数它）
|
||||
stats = (await client.get("/api/ledger/stats")).json()["e2"]
|
||||
assert stats["launched"] >= 1 and stats["succeeded"] >= 1
|
||||
|
||||
|
||||
async def test_the_derived_graph_is_deterministic():
|
||||
from app.ledger import derive
|
||||
|
||||
wf = await S.publish()
|
||||
async with SessionLocal() as session:
|
||||
pinned = await derive.load_pinned(session, wf, 1, "card")
|
||||
plan = ledger_reconcile.Plan(pairs=[ledger_reconcile.Pair(ours="gmv", metric="gmv", index=0, name="销售额",
|
||||
unit="元", decimals=2)])
|
||||
a = ledger_reconcile.build(pinned, sql=S.WITH_REFUNDS, tool=S.tool(), plan=plan, inputs=CHECKED, fetch="fetch")
|
||||
b = ledger_reconcile.build(pinned, sql=S.WITH_REFUNDS, tool=S.tool(), plan=plan, inputs=CHECKED, fetch="fetch")
|
||||
assert a.graph == b.graph and a.kind == "reconcile"
|
||||
assert a.origin["from"][0] == {"kind": "executed", "node_id": "ours_q1"} and a.origin["from"][1]["node_id"] == "card"
|
||||
|
||||
|
||||
async def test_metrics_that_are_not_a_plain_cell_need_a_person_to_pick(client, monkeypatch):
|
||||
wf = await S.publish(graph=S.caliber_graph(expression="cell(nodes.fetch, 0, 'gmv') * 1"))
|
||||
row = await S.explore(monkeypatch, S.WITH_REFUNDS)
|
||||
[action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"]
|
||||
body = {"query": action["query"], "workflow_id": wf, "version": 1, "inputs": CHECKED, "confirm": True}
|
||||
res = await client.post(f"/api/ledger/answers/{row.id}/reconcile", json=body)
|
||||
assert res.status_code == 409 and res.json()["code"] == ledger_reconcile.NEEDS_PAIRS
|
||||
bad = await client.post(f"/api/ledger/answers/{row.id}/reconcile",
|
||||
json={**body, "pairs": [{"ours": "nope", "metric": "gmv"}]})
|
||||
assert bad.status_code == 422 and "没有列「nope」" in bad.json()["detail"]
|
||||
done = await reconcile(client, row.id, {**body, "pairs": [{"ours": "gmv", "metric": "gmv"}]})
|
||||
assert done.status == "succeeded", done.error
|
||||
assert done.origin["pairs_chosen"] is True
|
||||
assert (await card_of(done.id))["diff_0"]["value"] == 50.5
|
||||
|
||||
|
||||
async def test_multi_row_answers_are_aligned_by_their_dimensions(client, monkeypatch):
|
||||
wf = await S.publish(graph=S.caliber_graph(BY_REGION))
|
||||
row = await S.explore(monkeypatch, BY_REGION_ALL, text="东区 [[v:Q1.r0.gmv]]，西区 [[v:Q1.r1.gmv]]。")
|
||||
[action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"]
|
||||
done = await reconcile(client, row.id, {"query": action["query"], "workflow_id": wf, "version": 1,
|
||||
"inputs": CHECKED, "confirm": True})
|
||||
assert done.status == "succeeded", done.error
|
||||
assert done.origin["rows"] == [0, 1] and done.origin["pairs"] == [{"ours": "gmv", "caliber": "gmv"}]
|
||||
align = next(n for n in done.graph["nodes"] if n["id"] == "align")
|
||||
assert align["type"] == "merge" and align["data"]["config"]["inputs"] == {"o": "ours_q1", "c": "fetch"}
|
||||
metrics = await card_of(done.id)
|
||||
assert metrics["ours_0_0"]["name"] == "区域=east · gmv · 这次"
|
||||
assert (metrics["ours_0_0"]["value"], metrics["cal_0_0"]["value"], metrics["diff_0_0"]["value"]) == (163.0, 150.5, 12.5)
|
||||
assert (metrics["ours_1_0"]["value"], metrics["cal_1_0"]["value"], metrics["diff_1_0"]["value"]) == (58.0, 20.0, 38.0)
|
||||
|
||||
|
||||
async def test_masked_dimensions_and_duplicate_keys_are_refused(client, monkeypatch, engine_up):
|
||||
wf = await S.publish(graph=S.caliber_graph(BY_REGION))
|
||||
duplicated = ("SELECT region AS 区域, amount AS gmv FROM orders WHERE paid_at >= '2026-09-01' "
|
||||
"AND paid_at < '2026-09-08' ORDER BY id")
|
||||
row = await S.explore(monkeypatch, duplicated, text="第一单 [[v:Q1.r0.gmv]]，第二单 [[v:Q1.r1.gmv]]。")
|
||||
[action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"]
|
||||
res = await client.post(f"/api/ledger/answers/{row.id}/reconcile",
|
||||
json={"query": action["query"], "workflow_id": wf, "version": 1, "inputs": CHECKED,
|
||||
"confirm": True})
|
||||
assert res.status_code == 409 and "有重复的行" in res.json()["detail"]
|
||||
|
||||
await S.set_source(engine_up, options={"mask_columns": ["region"]})
|
||||
row = await S.explore(monkeypatch, BY_REGION_ALL, text="东区 [[v:Q1.r0.gmv]]，西区 [[v:Q1.r1.gmv]]。")
|
||||
[action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"]
|
||||
res = await client.post(f"/api/ledger/answers/{row.id}/reconcile",
|
||||
json={"query": action["query"], "workflow_id": wf, "version": 1, "inputs": CHECKED,
|
||||
"confirm": True})
|
||||
assert res.status_code == 409 and "设置了遮罩" in res.json()["detail"]
|
||||
|
||||
|
||||
def test_align_sql_keeps_the_referenced_rows_in_order():
|
||||
pairs = [ledger_reconcile.Pair(ours="gmv", column="销售额")]
|
||||
sql = ledger_reconcile.align_sql([("区域", "区域")], pairs, [("west",), ("o'east",)])
|
||||
assert sql == ('SELECT o."区域" AS "区域", o."gmv" AS "ours_0", c."销售额" AS "cal_0" FROM o LEFT JOIN c ON '
|
||||
'o."区域" = c."区域" WHERE (o."区域" = \'west\') OR (o."区域" = \'o\'\'east\') '
|
||||
'ORDER BY CASE WHEN (o."区域" = \'west\') THEN 0 WHEN (o."区域" = \'o\'\'east\') THEN 1 END')
|
||||
|
||||
|
||||
async def test_users_reconcile_their_own_answers_read_only_and_masked(client, monkeypatch):
|
||||
"""拍板 L5：对账（只读的探索运行）使用者都可以，带上问数据同一套 MASK_RESULTS / READ_ONLY；别人的答案读不到就不给对。"""
|
||||
from app.api.deps import Principal
|
||||
from app.tools.datasource import MASK_RESULTS_OPTION, READ_ONLY_OPTION
|
||||
|
||||
wf = await S.publish()
|
||||
row = await S.explore(monkeypatch, S.WITH_REFUNDS)
|
||||
async with SessionLocal() as session:
|
||||
(await session.get(Run, row.id)).started_by = "carol"
|
||||
await session.commit()
|
||||
[action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"]
|
||||
req = ledger_reconcile.Request(query=action["query"], workflow_id=wf, version=1, inputs=CHECKED, confirm=True)
|
||||
dave = Principal(username="dave", role="user", verified=True, mode="local")
|
||||
with pytest.raises(ledger_reconcile.ReconcileRefused) as e:
|
||||
await ledger_reconcile.reconcile(row.id, req, principal=dave)
|
||||
assert e.value.status_code == 403
|
||||
carol = Principal(username="carol", role="user", verified=True, mode="local")
|
||||
run = await ledger_reconcile.reconcile(row.id, req, principal=carol)
|
||||
assert run.started_by == "carol"
|
||||
assert run.options[MASK_RESULTS_OPTION] is True and run.options[READ_ONLY_OPTION] is True
|
||||
done = await S.wait(run.id)
|
||||
assert done.status == "succeeded", done.error
|
||||
Reference in New Issue
Block a user