diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 61707fc7..545384dd 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -2591,6 +2591,10 @@ class FromRunIn(BaseModel): #: 问数据回答上点「做成应用」时带上:这次运行出自哪段对话的哪一轮,记进新工作流的来源(workflows.origin) conversation_id: str | None = None turn_id: str | None = None + #: 固化成口径草稿(next-round 02 做法 8,app/ledger/certify.freeze):报告引用的查询变成固定 SQL 的取数节点,跨轮变化 + #: 的字面量成为参数(不保留 Agent)。peers 是同一组(同一个 SQL 形状)的别的运行,认哪些字面量位置变了 + freeze: bool = False + peers: list[str] = Field(default_factory=list, max_length=50) @router.post("/from-run") @@ -2612,6 +2616,15 @@ async def extract_template( # 对话只是记进来源:不能用的不让记(D-d),不存在的照旧记下 id await _ensure_conversation_access(session, payload.conversation_id) + if payload.freeze: + # 固化成口径草稿:取数是固定的 SQL(以后比对认得出等价),不是保留 Agent 的执行路径 + from app.ledger.certify import CertifyRefused, freeze + + try: + return await freeze(session, payload.run_id, peers=payload.peers, name=payload.name, + conversation_id=payload.conversation_id, turn_id=payload.turn_id) + except CertifyRefused as e: + raise HTTPException(e.status_code, e.message) from e try: path = await extract_run_path(session, payload.run_id) except PathRefused as e: diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index 814ee9b3..5c8baa41 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -161,3 +161,30 @@ async def reconcile_answer(run_id: str, payload: ReconcileIn) -> dict[str, Any]: 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)} + + +@router.get("/certify-queue") +async def certify_queue(session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """可沉淀的问题(认证队列,app/ledger/certify.py):确定性重复的 SQL 形状、问题聚类、被反复用的答案。每项给代表问题、 + 轮次、(形状的)参数候选和相关口径;点过「不是口径」的不再列。固化走 POST /api/copilot/from-run(freeze: true)。""" + from app.api.deps import current_principal + from app.ledger import certify + + return await certify.queue(session, principal=current_principal()) + + +class DismissIn(BaseModel): + item_id: str = Field(min_length=1, max_length=64) + kind: str | None = Field(default=None, max_length=20) + note: str | None = Field(default=None, max_length=500) + + +@router.post("/certify-queue/dismiss") +async def certify_dismiss(payload: DismissIn, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """「不是口径」:记一条审计(ledger.certify.dismiss),这一项之后不再进队。""" + from app.ledger import certify + + if payload.kind is not None and payload.kind not in certify.KINDS: + raise HTTPException(422, "可沉淀的问题只有三种:确定性重复、问题聚类、被反复用的答案") + await certify.dismiss(session, payload.item_id, kind=payload.kind, note=payload.note) + return {"ok": True, "item_id": payload.item_id} diff --git a/backend/app/api/workflows.py b/backend/app/api/workflows.py index 91cefb94..d362c884 100644 --- a/backend/app/api/workflows.py +++ b/backend/app/api/workflows.py @@ -564,6 +564,10 @@ async def commit_publish(session: AsyncSession, workflow: Workflow, snapshot: Wo workflow.published_version = snapshot.version workflow.published_by = actor await session.commit() + # 口径账本(next-round 02 做法 8):从问数据固化出的口径发布了,来源运行在后台按它补比一次(不挡发布) + from app.ledger.certify import after_publish + + after_publish(workflow.id, workflow.origin) return { "ok": True, "level": level, diff --git a/backend/app/ledger/certify.py b/backend/app/ledger/certify.py new file mode 100644 index 00000000..4fd90cb7 --- /dev/null +++ b/backend/app/ledger/certify.py @@ -0,0 +1,550 @@ +"""沉淀:可沉淀的问题(认证队列)和「固化成口径草稿」(docs/design/next-round/02-口径账本.md 做法 8,拍板 L9)。 + +同一类问题被反复问、Agent 反复写出同一个形状的 SQL,就该沉淀成一份口径:以后再问,比对器认得出「等价已证明」,还能按口径 +正式出具。以前的 from-run 提取保留 Agent,沉淀出的模板没有固定 SQL,以后永远比对不上;聚类又混着 history。 + +**认证队列**(GET /api/ledger/certify-queue,搭建者及以上)三种信号: +- 确定性重复(shape):问数据的探索运行里执行过、被报告引用(比对工件里标签记着的查询)、又没对上任何口径(none / related) + 的查询,按(数据源、normalize_sql 的规范文本)分组;出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS 段对话的进队。 + 跨运行变化的字面量位置是参数候选,不变的是口径常量。展示的是规范文本(字面量都换成了「?」,不带取值); +- 问题聚类(cluster):app/chat/questions 全站唯一的聚类(只取问题原文、不含 history、排除系统派生的运行),≥ REPEAT_RUNS + 个问题的类; +- 被反复用的答案(reused):导出过 ≥ 2 次(report.exported 审计,第 2 波)、做成过应用(蓝图的来源是这次运行)、在画布里 + 打开过或提取过(工作流的 origin.run_id 是这次运行)的轮次。 +每项给代表问题、轮次链接、(shape 的)SQL 形状和参数候选、有没有相关口径(可以对账的那份)、是不是已经固化过。点「不是口径」 +写一条审计(ledger.certify.dismiss,subject 是这一项的 id),之后不再进队。门槛同 ledger/stats(E3 的重复度)。 + +**固化成口径草稿**(freeze,from-run 加 freeze 选项后转调它):代表运行里报告引用了的查询,各变成一个调用工具节点——SQL 用 +规范化后的文本,跨轮变化的字面量换成 '{{ input.p1 }}'(同组的别的运行 peers 用来认哪些位置变了),入口字段的缺省值取 +代表运行里的值;只有一行的结果每个数值列建一项口径卡指标(cell(nodes.取数, 0, '列'));报告节点保留,出口契约写 +report_from + metrics_from(有多行结果的另加 cells: true),过 validate_graph。新工作流记 origin.sources: +[{run_id, conversation_id, turn_id, question}],这就是「口径 → 来源问题」的记录。之后由人在画布上审改、按受管级别发布—— +提取器是作者,不是发布者。 + +**发布之后补比**(after_publish):工作流发布时,origin.sources 里的来源运行在后台按现在的口径补比一次(trigger=certified, +标「按之后发布的口径回看」),E1 也就有了回溯数据。 +""" +from __future__ import annotations + +import asyncio +import hashlib +import logging +from collections.abc import Sequence +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 + +logger = logging.getLogger(__name__) + +#: 门槛:同一个形状出现在几次运行里、来自几段对话;被导出几次算反复用(同 ledger/stats 的 E3) +REPEAT_RUNS = 3 +REPEAT_CONVERSATIONS = 2 +EXPORTS = 2 +#: 看最近多少次问数据运行、每项最多列几轮 +RUN_LIMIT = 500 +TURNS_LIMIT = 10 +QUESTION_LIMIT = 500 +#: 「不是口径」的审计 +DISMISS_ACTION = "ledger.certify.dismiss" +SUBJECT_KIND = "ledger_certify" +KINDS = ("shape", "cluster", "reused") + +#: 固化被拒的说法 +NOT_FOUND = "运行记录不存在,可能已被删除" +NOT_MATCHED = "这次运行还没有比对口径,比对完才能固化(打开它的口径账本会按需比对)" +NOT_SUCCEEDED = "这次运行没有成功结束,提取出的查询不完整" +NO_QUERIES = "这次运行的报告没有引用任何查询结果,没有可固化的取数" +UNPARSED = "这次的 SQL 无法规范化({reason}),不能固化成固定的取数" +SNAPSHOT = "查询快照读不出来(文件缺失或与哈希不一致),不能固化" +#: 新工作流的说明、报告指令 +DRAFT_NAME = "{question}(固化的口径草稿)" +DRAFT_DESCRIPTION = ("从问数据的回答固化:报告引用了的 {n} 条查询变成固定的取数,跨轮变化的 {p} 个字面量成为入口参数。" + "经人工审核、按受管级别发布后,再问同样的问题就能认出等价、按口径出具") +REPORT_INSTRUCTIONS = "回答:{question}" + + +class CertifyRefused(Exception): + def __init__(self, status_code: int, message: str) -> None: + super().__init__(message) + self.status_code = status_code + self.message = message + + +def item_id(kind: str, key: str) -> str: + return f"{kind}:{hashlib.sha256(key.encode()).hexdigest()[:16]}" + + +# ========================================================================== +# SQL 的形状和参数 +# ========================================================================== + + +@dataclass(frozen=True) +class Shape: + """一条查询的形状:数据源、规范文本(字面量换成「?」)、字面量(SQL 写法,字符串带引号)、规范化失败的原因。""" + + source: str + text: str + literals: tuple[str, ...] + problem: str | None = None + + +def shape_of(snap: Any, kinds: dict[str, str | None]) -> Shape | None: + from app.data.sql_semantics import normalize_sql + + if not isinstance(snap, dict) or not isinstance(snap.get("sql"), str) or not snap["sql"].strip(): + return None + source = str(snap.get("source") or "") + norm = normalize_sql(snap["sql"], kinds.get(source)) + return Shape(source, norm.text, norm.literals, norm.parse_error) + + +def varying(literals: Sequence[Sequence[str]]) -> list[int]: + """几次运行同一个形状的字面量:哪几个位置跨运行变过(参数候选),不变的是口径常量。长度不一致的那几份不算。""" + rows = [list(row) for row in literals] + if not rows: + return [] + width = len(rows[0]) + same = [r for r in rows if len(r) == width] + return [i for i in range(width) if len({r[i] for r in same}) > 1] + + +def _unquote(literal: str) -> str | None: + """'2026-09-01' → 2026-09-01;不是普通单引号字符串(N'…'、x'…'、$$…$$)的为 None。""" + if len(literal) >= 2 and literal[0] == "'" and literal[-1] == "'": + return literal[1:-1].replace("''", "'") + return None + + +def _number(literal: str) -> int | float | None: + try: + value = float(literal) + except ValueError: + return None + return int(value) if value.is_integer() and "." not in literal and "e" not in literal.lower() else value + + +def template_of(shape: Shape, params: Sequence[int], dialect: str | None, *, + start: int = 1) -> tuple[str, list[dict[str, Any]]]: + """规范文本 + 参数位置 → (口径的取数 SQL 模板, 入口字段)。参数位置的字面量换成 '{{ input.pN }}'(字符串)或 + {{ input.pN }}(数,N 从 start 起),缺省值是这一次的值;别的位置照原样写回字面量。规范文本按方言重新切词,按「?」的 + 先后对上字面量。""" + from app.data.sql_semantics import PLACEHOLDER, _read, _tokens + + tokens = [t for t in _tokens(shape.text, _read(dialect)) if shape.text[t.start:t.end + 1] == PLACEHOLDER] + if len(tokens) != len(shape.literals): + raise CertifyRefused(409, UNPARSED.format(reason="字面量对不上")) + out: list[str] = [] + fields: list[dict[str, Any]] = [] + pos = 0 + wanted = set(params) + for i, (token, literal) in enumerate(zip(tokens, shape.literals)): + out.append(shape.text[pos:token.start]) + pos = token.end + 1 + text, number = _unquote(literal), _number(literal) + if i in wanted and (text is not None or number is not None): + name = f"p{start + len(fields)}" + if text is not None: + out.append(f"'{{{{ input.{name} }}}}'") + fields.append({"name": name, "type": "string", "required": True, "default": text}) + else: + out.append(f"{{{{ input.{name} }}}}") + fields.append({"name": name, "type": "number", "required": True, "default": number}) + else: + out.append(literal) + out.append(shape.text[pos:]) + return "".join(out), fields + + +# ========================================================================== +# 认证队列 +# ========================================================================== + + +@dataclass +class _Member: + run_id: str + artifact: str + literals: tuple[str, ...] + conversation_id: str | None + turn_id: str | None + question: str + related: list[dict[str, Any]] = field(default_factory=list) + + +def referenced_queries(content: dict[str, Any]) -> list[str]: + """比对工件里被报告引用了的查询(标签记着的:单元格的 query、指标的 queries),按先后去重。""" + out: list[str] = [] + for report in (content.get("reports") or {}).values(): + for label in (report.get("labels") or {}).values(): + for artifact in [label.get("query"), *(label.get("queries") or [])]: + if isinstance(artifact, str) and artifact and artifact not in out: + out.append(artifact) + return out + + +async def _turns(session: Any, runs: list[Run]) -> dict[str, dict[str, Any]]: + """运行 → {conversation_id, turn_id, question}:运行来源里封存的(origin_ref);老运行退回 conversation_turns.run_id。""" + from app.db.models import ConversationTurn + + out: dict[str, dict[str, Any]] = {} + ids = [r.id for r in runs] + if ids: + for turn in (await session.execute(select(ConversationTurn).where(ConversationTurn.run_id.in_(ids)))).scalars(): + out.setdefault(turn.run_id, {"conversation_id": turn.conversation_id, "turn_id": turn.id, + "question": turn.question or ""}) + for run in runs: + origin = run.origin if isinstance(run.origin, dict) else {} + found = out.setdefault(run.id, {"conversation_id": None, "turn_id": None, "question": ""}) + if origin.get("conversation_id"): + found["conversation_id"] = str(origin["conversation_id"]) + found["turn_id"] = str(origin.get("turn_id") or found.get("turn_id") or "") or None + if not found["question"] and isinstance(run.input, dict) and isinstance(run.input.get("question"), str): + found["question"] = run.input["question"] + return out + + +async def _dismissed(session: Any) -> set[str]: + from app.db.models import AuditEvent + + return {sid for (sid,) in (await session.execute(select(AuditEvent.subject_id).where( + AuditEvent.action == DISMISS_ACTION, AuditEvent.subject_kind == SUBJECT_KIND))).all() if sid} + + +async def _frozen(session: Any) -> dict[str, list[dict[str, Any]]]: + """来源运行 → 从它固化出的工作流 [{workflow_id, name, status, published_version}](origin.sources)。""" + from app.db.models import Workflow + + out: dict[str, list[dict[str, Any]]] = {} + for wf in (await session.execute(select(Workflow).where(Workflow.origin.is_not(None)))).scalars(): + for src in (wf.origin or {}).get("sources") or [] if isinstance(wf.origin, dict) else []: + if isinstance(src, dict) and src.get("run_id"): + out.setdefault(str(src["run_id"]), []).append( + {"workflow_id": wf.id, "name": wf.name, "status": wf.status, + "published_version": wf.published_version}) + return out + + +def _turn_view(member: Any) -> dict[str, Any]: + return {"run_id": member.run_id, "conversation_id": member.conversation_id, "turn_id": member.turn_id, + "question": member.question} + + +async def queue(session: Any, *, principal: Any) -> dict[str, Any]: + """认证队列(见模块说明):{items: [...], threshold, considered}。只读。""" + from app.chat.questions import cluster_questions, embedder, recent_questions + from app.core.artifact_store import load + from app.db.models import Artifact, DataSource + from app.engine.origin import counted_runs + from app.ledger import store + + dismissed = await _dismissed(session) + frozen = await _frozen(session) + kinds = {name: kind for name, kind in (await session.execute(select(DataSource.name, DataSource.kind))).all()} + runs = list((await session.execute(select(Run).where( + Run.origin_kind == "chat", Run.status == "succeeded", Run.run_class != "formal", counted_runs()) + .order_by(Run.created_at.desc()).limit(RUN_LIMIT))).scalars()) + turns = await _turns(session, runs) + by_run = {r.id: r for r in runs} + # 每次运行最新的一份比对(store.latest 按事件认这一次封存的;这里只要内容,按工件引用的时间取最新) + latest: dict[str, tuple[Any, str]] = {} + if runs: + for artifact, run_id, at in (await session.execute(select(Artifact.id, Artifact.run_id, Artifact.created_at) + .where(Artifact.kind == store.KIND, + Artifact.run_id.in_(list(by_run))))).all(): + if run_id not in latest or (at and latest[run_id][0] and at > latest[run_id][0]): + latest[run_id] = (at, artifact) + + def read() -> dict[tuple[str, str], list[_Member]]: + groups: dict[tuple[str, str], list[_Member]] = {} + for run_id, (_, artifact) in latest.items(): + try: + content = load(artifact) + except ValueError: + continue + if not isinstance(content, dict) or content.get("format") != store.FORMAT: + continue + statuses = {q.get("artifact"): q for q in content.get("queries") or []} + for query in referenced_queries(content): + row = statuses.get(query) or {} + if row.get("status") not in ("none", "related"): + continue + try: + shape = shape_of(load(query), kinds) + except ValueError: + shape = None + if shape is None or shape.problem: + continue + info = turns.get(run_id) or {} + related = [{k: c.get(k) for k in ("caliber", "workflow_id", "version", "level")} + for c in row.get("candidates") or [] if c.get("status") == "related"] + groups.setdefault((shape.source, shape.text), []).append(_Member( + run_id, query, shape.literals, info.get("conversation_id"), info.get("turn_id"), + str(info.get("question") or ""), related)) + return groups + + groups = await asyncio.to_thread(read) + items: list[dict[str, Any]] = [] + for (source, text), members in groups.items(): + once = list({m.run_id: m for m in members}.values()) + conversations = {m.conversation_id or f"run:{m.run_id}" for m in once} + if len(once) < REPEAT_RUNS or len(conversations) < REPEAT_CONVERSATIONS: + continue + iid = item_id("shape", f"{source}\n{text}") + if iid in dismissed: + continue + params = varying([m.literals for m in once]) + related = list({r["caliber"]: r for m in once for r in m.related if r.get("caliber")}.values()) + rep = once[0] + items.append({ + "id": iid, "kind": "shape", "source": source, "shape": text, "runs": len(once), + "conversations": len(conversations), "question": rep.question, + "turns": [_turn_view(m) for m in once[:TURNS_LIMIT]], + "params": [{"index": i, "distinct": len({m.literals[i] for m in once if len(m.literals) > i})} + for i in params], + "constants": len(rep.literals) - len(params), "related": related, + "representative": {"run_id": rep.run_id, "query": rep.artifact}, + "peers": [m.run_id for m in once[1:]], + "frozen_into": [w for m in once for w in frozen.get(m.run_id, [])], + }) + # 问题聚类 + questions = await recent_questions(session, principal, limit=QUESTION_LIMIT) + for cluster in await cluster_questions(questions) if questions else []: + if cluster.count < REPEAT_RUNS: + continue + seed = cluster.members[0] + iid = item_id("cluster", " ".join(seed.text.split())) + if iid in dismissed: + continue + items.append({"id": iid, "kind": "cluster", "question": seed.text, "count": cluster.count, + "turns": [{"run_id": q.run_id, "turn_id": q.turn_id, "question": q.text} + for q in cluster.members[:TURNS_LIMIT]], + "frozen_into": [w for q in cluster.members if q.run_id for w in frozen.get(q.run_id, [])]}) + items += await _reused(session, runs, turns, dismissed, frozen) + return {"items": items, "considered": {"runs": len(runs), "matched": len(latest), "questions": len(questions)}, + "threshold": {"runs": REPEAT_RUNS, "conversations": REPEAT_CONVERSATIONS, "exports": EXPORTS}, + "embedder": embedder()} + + +async def _reused(session: Any, runs: list[Run], turns: dict[str, dict[str, Any]], dismissed: set[str], + frozen: dict[str, list[dict[str, Any]]]) -> list[dict[str, Any]]: + """被反复用的答案:导出过 ≥ EXPORTS 次、做成过应用、在画布里打开过或提取过(见模块说明)。""" + from sqlalchemy import func + + from app.db.models import AuditEvent, Workflow + from app.db.models_blueprints import Blueprint + + ids = [r.id for r in runs] + if not ids: + return [] + exports = {sid: int(n) for sid, n in (await session.execute( + select(AuditEvent.subject_id, func.count()).where(AuditEvent.action == "report.exported", + AuditEvent.subject_kind == "run", + AuditEvent.subject_id.in_(ids)) + .group_by(AuditEvent.subject_id))).all()} + apps: set[str] = set() + for source in (await session.execute(select(Blueprint.source))).scalars(): + if isinstance(source, dict) and source.get("kind") == "run" and source.get("id") in ids: + apps.add(str(source["id"])) + canvas: dict[str, list[str]] = {} + for wf_id, origin, tags in (await session.execute(select(Workflow.id, Workflow.origin, Workflow.tags) + .where(Workflow.origin.is_not(None)))).all(): + if isinstance(origin, dict) and origin.get("run_id") in ids and not origin.get("sources"): + canvas.setdefault(str(origin["run_id"]), []).append("extracted" if "extracted" in (tags or []) else "canvas") + out: list[dict[str, Any]] = [] + for run in runs: + uses = {"exports": exports.get(run.id, 0), "app": run.id in apps, + "extracted": "extracted" in canvas.get(run.id, []), "canvas": "canvas" in canvas.get(run.id, [])} + if not (uses["exports"] >= EXPORTS or uses["app"] or uses["extracted"] or uses["canvas"]): + continue + iid = item_id("reused", run.id) + if iid in dismissed: + continue + info = turns.get(run.id) or {} + out.append({"id": iid, "kind": "reused", "question": info.get("question") or "", "uses": uses, + "turns": [{"run_id": run.id, "conversation_id": info.get("conversation_id"), + "turn_id": info.get("turn_id"), "question": info.get("question") or ""}], + "representative": {"run_id": run.id}, "frozen_into": frozen.get(run.id, [])}) + return out + + +async def dismiss(session: Any, iid: str, *, kind: str | None = None, note: str | None = None) -> None: + """「不是口径」:写一条审计,之后这一项不再进队。""" + from app.auth.audit import record + + await record(session, DISMISS_ACTION, SUBJECT_KIND, iid, + data={k: v for k, v in {"kind": kind, "note": note}.items() if v}) + await session.commit() + + +# ========================================================================== +# 固化成口径草稿 +# ========================================================================== + + +async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name: str = "", + conversation_id: str | None = None, turn_id: str | None = None) -> dict[str, Any]: + """把一次问数据运行固化成口径草稿(见模块说明),落库成新的草稿工作流,返回 {workflow_id, name, nodes, edges, params, + sources, ...}。peers 是同一组(同一个 SQL 形状)的别的运行:认哪些字面量位置跨轮变了。不能固化抛 CertifyRefused。""" + from app.api.deps import can_read_run, current_principal + from app.api.evidence import _reports, _sealed + from app.core.artifact_store import graph_hash, load + from app.db.models import DataSource, Workflow, WorkflowVersion + from app.engine.governance import REPORT_POLICY + from app.engine.layout import auto_layout + from app.engine.schema import GraphSpec + from app.ledger import store + from app.tools.datasource import QUERY_PREFIX + + run = await session.get(Run, run_id) + principal = current_principal() + if run is None or not await can_read_run(principal, run): + raise CertifyRefused(404, NOT_FOUND) + if run.status != "succeeded": + raise CertifyRefused(409, NOT_SUCCEEDED) + recorded = await store.latest(run_id) + if recorded is None or recorded.content is None: + raise CertifyRefused(409, NOT_MATCHED) + queries = referenced_queries(recorded.content) + if not queries: + raise CertifyRefused(409, NO_QUERIES) + kinds = {n: k for n, k in (await session.execute(select(DataSource.name, DataSource.kind))).all()} + + # 同组的别的运行:它们封存的查询里同形状的那一条的字面量 + peer_literals: dict[tuple[str, str], list[tuple[str, ...]]] = {} + peer_sources: list[dict[str, Any]] = [] + peer_runs = [r for r in [await session.get(Run, p) for p in dict.fromkeys(peers) if p != run_id] if r is not None] + readable = [r for r in peer_runs if await can_read_run(principal, r)] + for peer in readable: + sealed = await _sealed(peer.id) + for artifact in sealed.queries: + try: + shape = shape_of(load(artifact), kinds) + except ValueError: + continue + if shape is not None and not shape.problem: + peer_literals.setdefault((shape.source, shape.text), []).append(shape.literals) + sealed = await _sealed(run_id) + shapes: list[Shape] = [] + snaps: list[dict[str, Any]] = [] + for artifact in queries: + try: + snap = load(artifact) + except ValueError: + raise CertifyRefused(409, SNAPSHOT) from None + shape = shape_of(snap, kinds) + if shape is None: + raise CertifyRefused(409, SNAPSHOT) + if shape.problem: + raise CertifyRefused(409, UNPARSED.format(reason=shape.problem)) + shapes.append(shape) + snaps.append(snap) + + def nid(i: int) -> str: + return "fetch" if len(shapes) == 1 else f"fetch_{i + 1}" + + nodes: list[dict[str, Any]] = [] + fields: list[dict[str, Any]] = [] + metrics: list[dict[str, Any]] = [] + multi = False + for i, (shape, snap) in enumerate(zip(shapes, snaps)): + params = varying([shape.literals, *peer_literals.get((shape.source, shape.text), [])]) + template, new_fields = template_of(shape, params, kinds.get(shape.source), start=len(fields) + 1) + fields += new_fields + nodes.append({"id": nid(i), "type": "tool", "position": {"x": 0, "y": 0}, + "data": {"label": f"取数 {i + 1}" if len(shapes) > 1 else "取数", + "config": {"tool": f"{QUERY_PREFIX}{shape.source}", "args": {"sql": template}}}}) + rows = snap.get("rows") if isinstance(snap.get("rows"), list) else [] + columns = [str(c) for c in snap.get("columns") or []] + if len(rows) == 1 and isinstance(rows[0], (list, tuple)): + for col, value in zip(columns, rows[0]): + if isinstance(value, (int, float)) and not isinstance(value, bool): + metrics.append({"id": f"m{len(metrics) + 1}", "name": col, + "expression": f"cell(nodes.{nid(i)}, 0, {col!r})"}) + else: + multi = True + question = ((await _turns(session, [run])).get(run.id) or {}).get("question") or "" + reports = _reports(sealed) + report_cfg: dict[str, Any] = {} + if reports: + found = next((n for n in (run.graph or {}).get("nodes") or [] if n.get("id") == reports[0].node_id), None) + cfg = ((found or {}).get("data") or {}).get("config") or {} + report_cfg = {k: cfg[k] for k in ("provider", "model", "entities") if k in cfg} + report_cfg["instructions"] = REPORT_INSTRUCTIONS.format(question=question or "(问题)") + # 报告的核对规则一律按受管级别的要求写(engine/governance.REPORT_POLICY,和发布门禁同一份):草稿要能按受管发布 + report_cfg.update({key: value for key, value, _ in REPORT_POLICY}) + contract: dict[str, Any] = {"report_from": "write"} + tail: list[dict[str, Any]] = [] + if metrics: + tail.append({"id": "card", "type": "metrics", "position": {"x": 0, "y": 0}, + "data": {"label": "口径卡", "config": {"caliber": (question or "口径")[:40], "caliber_version": "v1", + "metrics": metrics}}}) + report_cfg["metrics_from"] = ["card"] + # 受管出具要写必需指标(不然「不予出具」这一档触发不了):固化出的每个指标都是这个口径要给的数 + contract.update({"metrics_from": "card", "required": [m["id"] for m in metrics]}) + if multi or not metrics: + contract["cells"] = True + chain = [{"id": "start", "type": "input", "position": {"x": 0, "y": 0}, + "data": {"label": "入口", "config": {"fields": fields}}}, + *nodes, *tail, + {"id": "write", "type": "report", "position": {"x": 0, "y": 0}, + "data": {"label": "报告", "config": report_cfg}}, + {"id": "out", "type": "output", "position": {"x": 0, "y": 0}, + "data": {"label": "出口", "config": {"fields": [{"name": "报告", "value": "{{ nodes.write.text }}"}], + "contract": contract}}}] + edges = [{"source": a["id"], "target": b["id"]} for a, b in zip(chain, chain[1:])] + graph = auto_layout(GraphSpec.model_validate({"nodes": chain, "edges": edges})).model_dump(mode="json") + + info = (await _turns(session, [run])).get(run.id) or {} + sources = [{"run_id": run.id, "conversation_id": conversation_id or info.get("conversation_id"), + "turn_id": turn_id or info.get("turn_id"), "question": question}] + for peer, peer_info in (await _turns(session, readable)).items(): + sources.append({"run_id": peer, "conversation_id": peer_info.get("conversation_id"), + "turn_id": peer_info.get("turn_id"), "question": peer_info.get("question") or ""}) + workflow = Workflow( + name=(name or DRAFT_NAME.format(question=(question or run.workflow_name or "问数据")[:30]))[:200], + description=DRAFT_DESCRIPTION.format(n=len(shapes), p=len(fields)), graph=graph, tags=["certified"], + status="draft", + origin={"run_id": run.id, **({"conversation_id": sources[0]["conversation_id"]} if sources[0]["conversation_id"] + else {}), **({"turn_id": sources[0]["turn_id"]} if sources[0]["turn_id"] else {}), + "freeze": True, "sources": sources}) + session.add(workflow) + await session.flush() + session.add(WorkflowVersion(workflow_id=workflow.id, version=1, graph=graph, graph_hash=graph_hash(graph), + note=f"从运行 {run.id} 固化")) + await session.commit() + return {"workflow_id": workflow.id, "name": workflow.name, "nodes": len(chain), "edges": len(edges), + "dropped_nodes": 0, "source_run": run.id, "frozen": True, "params": [f["name"] for f in fields], + "metrics": [m["id"] for m in metrics], "sources": sources} + + +# ========================================================================== +# 发布之后补比 +# ========================================================================== + +_background: set[asyncio.Task[Any]] = set() + + +def after_publish(workflow_id: str, origin: Any) -> int: + """工作流发布了:从问数据固化来的(origin.sources),来源运行在后台按现在的口径补比一次(trigger=certified)。返回排上了 + 几个。不挡发布,失败只记日志(ledger/match 的 schedule 自己兜着)。""" + from app.ledger import match as ledger_match + + sources = origin.get("sources") if isinstance(origin, dict) else None + count = 0 + for src in sources or []: + if isinstance(src, dict) and src.get("run_id"): + try: + if ledger_match.schedule(str(src["run_id"]), trigger="certified"): + count += 1 + except Exception: # noqa: BLE001 - 补比是辅助步骤 + logger.exception("工作流 %s 发布后补比来源运行出错", workflow_id) + return count + + +__all__ = ["DISMISS_ACTION", "KINDS", "REPEAT_CONVERSATIONS", "REPEAT_RUNS", "CertifyRefused", "Shape", "after_publish", + "dismiss", "freeze", "item_id", "queue", "referenced_queries", "shape_of", "template_of", "varying"] diff --git a/backend/app/ledger/labels.py b/backend/app/ledger/labels.py index 7b791abc..08a23cac 100644 --- a/backend/app/ledger/labels.py +++ b/backend/app/ledger/labels.py @@ -30,6 +30,8 @@ HOWS = ("independent", "composed") #: 界面上标签要写明的两句:比对在封存之后做,不在封存范围内;事后补比的是按现在的口径回看 POST_SEAL_NOTE = "封存后比对,不在封存范围内" RETRO_NOTE = "按现在的口径回看" +#: 固化出的口径发布之后对来源运行补比的(store 的 trigger=certified) +CERTIFIED_NOTE = "按之后发布的口径回看" #: 发布级别的叫法(同 blueprints/publish_plan.LEVEL_TEXT、前端 lib/terms.ts) LEVEL_TEXT = {"published": "已发布", "governed": "受管"} MASKED = "已遮罩" @@ -356,6 +358,6 @@ def related_view(item: Mapping[str, Any], *, hide: Callable[[Any], bool], shown: return out -__all__ = ["HOWS", "KINDS", "LEVEL_TEXT", "POST_SEAL_NOTE", "RETRO_NOTE", "binding_view", "caliber_key", +__all__ = ["CERTIFIED_NOTE", "HOWS", "KINDS", "LEVEL_TEXT", "POST_SEAL_NOTE", "RETRO_NOTE", "binding_view", "caliber_key", "card_expressions", "cell_label", "diff_view", "label_doc", "metric_label", "related_view", "view_label", "whole_of"] diff --git a/backend/app/ledger/reconcile.py b/backend/app/ledger/reconcile.py index 1bf35caa..4e9daf26 100644 --- a/backend/app/ledger/reconcile.py +++ b/backend/app/ledger/reconcile.py @@ -65,7 +65,7 @@ PAIRS_NONE = "认不出这次的哪一列对应口径的哪个指标,请选择 PAIR_COLUMN = "这次的结果里没有列「{column}」" PAIR_MASKED = "列「{column}」设置了遮罩,不能对账" PAIR_METRIC = "口径卡「{card}」上没有指标「{metric}」" -PAIRS_SHAPE = "要对照的指标写成 [{ours: 这次的列, metric: 口径卡上的指标 id}]" +PAIRS_SHAPE = "要对照的指标需要写成一组「这次的列 → 口径卡上的指标」的对照" ROWS_NONE = "报告里没有引用这条查询的格,不知道要对哪几行" ROWS_NO_KEY = "这次的结果有多行,但没有可以用来对齐的维度列(不是数的列)" ROWS_MASKED_KEY = "用来对齐的维度列「{column}」设置了遮罩,不能对账" diff --git a/backend/app/ledger/stats.py b/backend/app/ledger/stats.py index 38aa108e..b0f4bb8e 100644 --- a/backend/app/ledger/stats.py +++ b/backend/app/ledger/stats.py @@ -10,11 +10,13 @@ - E1 等价命中:比对过的查询里 proven / composed / related / none / skipped 各几条(skipped 单独列,不混进 none),有候选口径 的查询里各自的比率;证明等价的绑定是在哪一级配上的;没证明的原因(不可比、参数绑不上、配上了但不等价);不等价的差异类型 分布(time / filter / aggregate / join / other / parse,每份口径的差异每类只计一次);逐个数的标签各几个、整题等价几次。 -- E2 对账:对账运行(origin=reconcile)发起、完成、失败各几次;相关口径的差异说明里兜底说明的占比。「说明有用 / 看不懂」的 - 反馈要等对账面板(第 3 波)记成对账运行上的封存后事件,这之前照实写没有数据。 +- E2 对账:对账运行(origin=reconcile,第 3 波起由 ledger/reconcile 发起)发起、完成、失败各几次,对过账的答案几个,其中参数 + 经人核对过的、对照的指标由人选的各几次;相关口径的差异说明里兜底说明的占比。「说明有用 / 看不懂」的反馈要等对账面板 + (第 4 波)记成对账运行上的封存后事件,这之前照实写没有数据。 - E3 重复度:问数据运行里执行过的查询按(数据源、normalize_sql 文本)分组,出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS 段对话的算重复的形状,给出这样的形状几个、有重复形状的运行占几成;问题聚类(app/chat/questions:全站唯一的聚类,只取问题 - 原文、不含 history、排除系统派生的运行)里 ≥ REPEAT_RUNS 个问题的类占几成。认证队列的转化(固化、发布、驳回)在第 3 波。 + 原文、不含 history、排除系统派生的运行)里 ≥ REPEAT_RUNS 个问题的类占几成。认证队列的转化(第 3 波,ledger/certify):固化成 + 口径草稿几份(origin.freeze 的工作流)、其中发布了几份、点了「不是口径」几项。 - E4 召回:每个口径版本被用了多少次(比对里等价已证明的运行数、这一版的正式运行数)。每次发布召回了几组、其中数变了的比例要等 召回(第 4 波),这之前照实写没有数据,只给召回运行的次数。 """ @@ -40,8 +42,7 @@ ASPECT_GROUP = {"time": "time", "time_range": "time", "filter": "filter", "aggre DIFF_GROUPS = ("time", "filter", "aggregate", "join", "other", "parse") STATUSES = ("proven", "composed", "related", "none", "skipped") #: 没有数据时的说法(这一块的功能还没做) -NOT_YET_RECONCILE = "对账面板的反馈在第 3 波接上,目前没有数据" -NOT_YET_CERTIFY = "认证队列在第 3 波接上,目前没有数据" +NOT_YET_RECONCILE = "对账面板的反馈在第 4 波接上,目前没有数据" NOT_YET_RECALL = "召回在第 4 波接上,目前没有数据" #: 聚类最多取几个问题(同治理接口的上限) QUESTION_LIMIT = 500 @@ -154,6 +155,8 @@ async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic if scope.get("until") is not None: stmt = stmt.where(Run.created_at < scope["until"]) by_status = {s: int(n) for s, n in (await session.execute(stmt.group_by(Run.status))).all()} + origins = [o for (o,) in (await session.execute(stmt.with_only_columns(Run.origin))).all() if isinstance(o, dict)] + answers = {o.get("source_run") for o in origins if o.get("source_run")} explained = fallback = 0 for content in contents: for query in content.get("queries") or []: @@ -162,7 +165,10 @@ async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic explained += 1 fallback += bool(cand["diff"].get("fallback")) return {"launched": sum(by_status.values()), "succeeded": by_status.get("succeeded", 0), - "failed": by_status.get("failed", 0), "explained": explained, "fallback": fallback, + "failed": by_status.get("failed", 0), "answers": len(answers), + "inputs_checked": sum(1 for o in origins if o.get("inputs_from") == "checked"), + "pairs_chosen": sum(1 for o in origins if o.get("pairs_chosen")), + "explained": explained, "fallback": fallback, "fallback_rate": _ratio(fallback, explained), "feedback": {"available": False, "note": NOT_YET_RECONCILE}} @@ -233,10 +239,37 @@ async def _e3(session: Any, *, principal: Any, **scope: Any) -> dict[str, Any]: "threshold": {"runs": REPEAT_RUNS, "conversations": REPEAT_CONVERSATIONS}, "questions": len(questions), "clusters": len(clusters), "questions_in_repeated_clusters": clustered, "cluster_share": _ratio(clustered, len(questions)), "embedder": embedder(), - "certify": {"available": False, "note": NOT_YET_CERTIFY}, + "certify": await _certify(session, **scope), } +async def _certify(session: Any, **scope: Any) -> dict[str, Any]: + """认证队列的转化:固化成口径草稿几份、其中发布了几份、点了「不是口径」几项(按固化、点选的时间筛窗口)。""" + from app.db.models import AuditEvent, Workflow + from app.ledger.certify import DISMISS_ACTION + + since, until = scope.get("since"), scope.get("until") + frozen = published = 0 + for created, origin, current in (await session.execute( + select(Workflow.created_at, Workflow.origin, Workflow.published_version) + .where(Workflow.origin.is_not(None)))).all(): + if not isinstance(origin, dict) or not origin.get("freeze"): + continue + if (since is not None and created is not None and created < since) or \ + (until is not None and created is not None and created >= until): + continue + frozen += 1 + published += current is not None + stmt = select(func.count()).select_from(AuditEvent).where(AuditEvent.action == DISMISS_ACTION) + if since is not None: + stmt = stmt.where(AuditEvent.at >= since) + if until is not None: + stmt = stmt.where(AuditEvent.at < until) + dismissed = int(await session.scalar(stmt) or 0) + return {"available": True, "frozen": frozen, "published": published, "dismissed": dismissed, + "publish_rate": _ratio(published, frozen)} + + async def _e4(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dict[str, Any]: from app.db.models import Run from app.engine.origin import counted_runs diff --git a/backend/app/ledger/store.py b/backend/app/ledger/store.py index 5846eb51..6ccb7f39 100644 --- a/backend/app/ledger/store.py +++ b/backend/app/ledger/store.py @@ -32,8 +32,8 @@ FORMAT = "agentlab.caliber_match/1" #: 2(第 3 波):认出助手选用的口径(查询 composed、标签 how=composed),指标的标签记它取自哪几条查询(queries,认证队列 #: 认「被报告引用的查询」用) ALGO = "2" -#: 触发方式(见模块说明) -TRIGGERS = ("sealed", "backfill", "on_demand") +#: 触发方式(见模块说明)。certified(第 3 波):从问数据固化出的口径发布之后,对来源运行补比的(ledger/certify) +TRIGGERS = ("sealed", "backfill", "on_demand", "certified") @dataclass(frozen=True) diff --git a/backend/app/ledger/view.py b/backend/app/ledger/view.py index 33e860c6..d1c8f356 100644 --- a/backend/app/ledger/view.py +++ b/backend/app/ledger/view.py @@ -98,7 +98,8 @@ def _stamp(recorded: store.Recorded) -> dict[str, Any]: if isinstance(recorded.ts, (int, float)) else None return {"artifact": recorded.artifact, "algo": recorded.algo, "trigger": recorded.trigger, "retro": recorded.retro, "matched_at": at, "post_seal": True, "note": ledger_labels.POST_SEAL_NOTE, - **({"retro_note": ledger_labels.RETRO_NOTE} if recorded.retro else {})} + **({"retro_note": ledger_labels.CERTIFIED_NOTE if recorded.trigger == "certified" + else ledger_labels.RETRO_NOTE} if recorded.retro else {})} async def run_view(run_id: str) -> dict[str, Any]: diff --git a/backend/tests/test_ledger_certify.py b/backend/tests/test_ledger_certify.py new file mode 100644 index 00000000..e91dee59 --- /dev/null +++ b/backend/tests/test_ledger_certify.py @@ -0,0 +1,224 @@ +"""沉淀:可沉淀的问题(认证队列)和「固化成口径草稿」(next-round 02 做法 8,app/ledger/certify.py)。 + +真的跑三次问数据(两段对话、同一个 SQL 形状、日期不同,虚构店铺库),等后台比对做完: +- 队列:确定性重复进队(3 次运行、2 段对话),变了的字面量是参数候选、不变的是常量,形状里不带取值;同一个问题问了 3 次 + 进问题聚类;导出过两次的、做成过应用的答案进「被反复用的答案」;点「不是口径」写审计后不再进队; +- 固化:报告引用的查询变成固定 SQL 的取数节点,跨轮变化的日期成为入口参数(缺省值取代表运行的),常量 'paid' 照写;单行 + 结果的数值列成口径卡指标;出口契约 report_from + metrics_from;过 validate_graph 和受管级别的发布门禁;origin.sources 记 + 三个来源问题; +- 按受管级别发布之后:来源运行在后台按它补比(trigger=certified),查询变成等价已证明、出了队;再问一次同样的形状,标 + 「等价已证明」; +- 统计 E3:认证队列的转化(固化、发布、不是口径)有数了。 +数据是虚构店铺库。 +""" +from __future__ import annotations + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.db.base import SessionLocal +from app.db.models import Conversation, ConversationTurn, Run, Workflow +from app.engine.runner import run_manager +from app.engine.schema import GraphSpec, validate_graph +from app.ledger import certify +from app.ledger import match as ledger_match +from app.ledger import store +from app.main import app +from tests import ledger_samples as S + +QUESTION = "九月第一周卖了多少" +WEEKS = [("2026-09-01", "2026-09-08"), ("2026-09-02", "2026-09-09"), ("2026-09-03", "2026-09-10")] + + +def weekly(start: str, end: str) -> str: + return (f"SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{start}' " + f"AND paid_at < '{end}'") + + +@pytest.fixture(autouse=True) +async def engine_up(tmp_path): + await run_manager.setup() + await S.fresh_source(tmp_path / "shop.db") + 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 conversation() -> str: + async with SessionLocal() as session: + conv = Conversation(title="问销售", kind="chat") + session.add(conv) + await session.commit() + return conv.id + + +async def ask(monkeypatch, conversation_id: str, sql: str, question: str = QUESTION) -> Run: + """问数据的一轮:建轮次、跑一次问数据运行(来源记着对话和轮次)、挂回轮次、等比对做完。""" + async with SessionLocal() as session: + turn = ConversationTurn(conversation_id=conversation_id, question=question) + session.add(turn) + await session.commit() + turn_id = turn.id + S.script(monkeypatch, "这一周销售额 [[v:Q1.r0.gmv]]。") + run = await run_manager.start(graph=S.explore_graph(sql), input_payload={"question": question}, origin_kind="chat", + origin={"conversation_id": conversation_id, "turn_id": turn_id}) + row = await S.wait(run.id) + assert row.status == "succeeded", row.error + async with SessionLocal() as session: + (await session.get(ConversationTurn, turn_id)).run_id = row.id + await session.commit() + await ledger_match.drain() + return row + + +async def three_asks(monkeypatch) -> list[Run]: + a, b = await conversation(), await conversation() + return [await ask(monkeypatch, conv, weekly(*week)) for conv, week in zip((a, a, b), WEEKS)] + + +def shape_items(body: dict) -> list[dict]: + return [i for i in body["items"] if i["kind"] == "shape" and i["source"] == S.CURRENT["source"]] + + +# -------------------------------------------------------------------------- +# 队列 +# -------------------------------------------------------------------------- + + +def test_varying_literals_are_parameters_and_the_template_puts_them_back(): + shape = certify.Shape("shop", "SELECT SUM(amount) AS gmv FROM orders WHERE status = ? AND paid_at >= ? AND x = ?", + ("'paid'", "'2026-09-01'", "3")) + assert certify.varying([shape.literals, ("'paid'", "'2026-09-02'", "4")]) == [1, 2] + template, fields = certify.template_of(shape, [1, 2], "sqlite", start=3) + assert template == ("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{{ input.p3 }}' " + "AND x = {{ input.p4 }}") + assert fields == [{"name": "p3", "type": "string", "required": True, "default": "2026-09-01"}, + {"name": "p4", "type": "number", "required": True, "default": 3}] + + +async def test_repeated_shapes_questions_and_reused_answers_queue_up(client, monkeypatch): + runs = await three_asks(monkeypatch) + body = (await client.get("/api/ledger/certify-queue")).json() + [item] = shape_items(body) + assert item["runs"] == 3 and item["conversations"] == 2 and item["question"] == QUESTION + assert item["params"] == [{"index": 1, "distinct": 3}, {"index": 2, "distinct": 3}] and item["constants"] == 1 + assert "2026-09" not in item["shape"] and "'paid'" not in item["shape"] and "?" in item["shape"] + assert {t["run_id"] for t in item["turns"]} == {r.id for r in runs} + assert all(t["turn_id"] and t["conversation_id"] for t in item["turns"]) + assert item["representative"]["run_id"] in {r.id for r in runs} and len(item["peers"]) == 2 + assert item["related"] == [] and item["frozen_into"] == [] + clusters = [i for i in body["items"] if i["kind"] == "cluster" and i["question"] == QUESTION] + assert clusters and clusters[0]["count"] >= 3 + + # 被反复用的答案:导出过两次的、做成过应用的 + from app.db.models_blueprints import Blueprint + + async with SessionLocal() as session: + from app.auth.audit import record + + for _ in range(2): + await record(session, "report.exported", "run", runs[0].id, data={"format": "html"}) + session.add(Blueprint(name="周销售应用", source={"kind": "run", "id": runs[1].id}, spec={})) + await session.commit() + body = (await client.get("/api/ledger/certify-queue")).json() + reused = {i["representative"]["run_id"]: i for i in body["items"] if i["kind"] == "reused"} + assert reused[runs[0].id]["uses"]["exports"] == 2 and reused[runs[1].id]["uses"]["app"] is True + assert runs[2].id not in reused + + # 「不是口径」:记审计,之后不再进队 + res = await client.post("/api/ledger/certify-queue/dismiss", json={"item_id": item["id"], "kind": "shape"}) + assert res.status_code == 200 + assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == [] + stats = (await client.get("/api/ledger/stats")).json()["e3"]["certify"] + assert stats["available"] is True and stats["dismissed"] >= 1 + + +async def test_two_runs_or_one_conversation_are_not_enough(client, monkeypatch): + conv = await conversation() + for week in WEEKS: + await ask(monkeypatch, conv, weekly(*week)) + assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == [], "只来自一段对话" + + +# -------------------------------------------------------------------------- +# 固化、发布、再问 +# -------------------------------------------------------------------------- + + +async def test_freezing_makes_a_fixed_sql_draft_that_passes_the_governed_gate(client, monkeypatch): + from app.engine.governance import publish_issues + + runs = await three_asks(monkeypatch) + [item] = shape_items((await client.get("/api/ledger/certify-queue")).json()) + rep = item["representative"]["run_id"] + res = await client.post("/api/copilot/from-run", json={"run_id": rep, "freeze": True, "peers": item["peers"]}) + assert res.status_code == 200, res.text + made = res.json() + assert made["frozen"] is True and made["params"] == ["p1", "p2"] and made["metrics"] == ["m1"] + async with SessionLocal() as session: + wf = await session.get(Workflow, made["workflow_id"]) + assert wf.status == "draft" and "certified" in wf.tags + nodes = {n["id"]: n for n in wf.graph["nodes"]} + assert set(nodes) == {"start", "fetch", "card", "write", "out"} + rep_run = next(r for r in runs if r.id == rep) + start, end = WEEKS[[r.id for r in runs].index(rep)] + assert nodes["start"]["data"]["config"]["fields"] == [ + {"name": "p1", "type": "string", "required": True, "default": start}, + {"name": "p2", "type": "string", "required": True, "default": end}] + sql = nodes["fetch"]["data"]["config"]["args"]["sql"] + assert "'{{ input.p1 }}'" in sql and "'{{ input.p2 }}'" in sql and "'paid'" in sql + assert nodes["fetch"]["data"]["config"]["tool"] == S.tool() + assert nodes["card"]["data"]["config"]["metrics"] == [ + {"id": "m1", "name": "gmv", "expression": "cell(nodes.fetch, 0, 'gmv')"}] + assert nodes["out"]["data"]["config"]["contract"] == {"report_from": "write", "metrics_from": "card", + "required": ["m1"]} + assert {k: nodes["write"]["data"]["config"].get(k) for k in ("numbers", "on_violation", "claims")} == { + "numbers": "strict", "on_violation": "fail", "claims": "require_citation"} + assert nodes["write"]["data"]["config"]["instructions"] == f"回答:{QUESTION}" + spec = GraphSpec.model_validate(wf.graph) + assert not [i for i in validate_graph(spec).issues if i.level == "error"] + assert not [i for i in publish_issues(spec, level="governed") if i.level == "error"], \ + [i.message for i in publish_issues(spec, level="governed") if i.level == "error"] + assert [s["run_id"] for s in wf.origin["sources"]][0] == rep and len(wf.origin["sources"]) == 3 + assert all(s["question"] == QUESTION for s in wf.origin["sources"]) + assert wf.origin["run_id"] == rep_run.id and wf.origin["freeze"] is True + + # 队列上标着已经固化过 + [again] = shape_items((await client.get("/api/ledger/certify-queue")).json()) + assert {w["workflow_id"] for w in again["frozen_into"]} == {wf.id} + + # 按受管级别发布:来源运行在后台补比,变成等价已证明、出了队 + res = await client.post(f"/api/workflows/{wf.id}/publish", json={"level": "governed"}) + assert res.status_code == 200, res.text + await ledger_match.drain() + for run in runs: + recorded = await store.latest(run.id) + assert recorded.trigger == "certified" and recorded.retro is True + [query] = recorded.content["queries"] + assert query["status"] == "proven" and query["candidates"][0]["caliber"] == f"{wf.id}@1" + body = (await client.get(f"/api/runs/{runs[0].id}/ledger")).json() + assert body["retro_note"] == "按之后发布的口径回看" + assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == [] + + # 再问一次同样的形状:标等价已证明 + later = await ask(monkeypatch, await conversation(), weekly("2026-09-05", "2026-09-12")) + [label] = (await store.latest(later.id)).content["reports"]["write"]["labels"].values() + assert label["kind"] == "caliber" and label["how"] == "independent" and label["caliber"] == f"{wf.id}@1" + assert label["binding"] == {"p1": "2026-09-05", "p2": "2026-09-12"} + + stats = (await client.get("/api/ledger/stats")).json()["e3"]["certify"] + assert stats["frozen"] >= 1 and stats["published"] >= 1 + + +async def test_freezing_needs_a_matched_successful_run(client, monkeypatch): + res = await client.post("/api/copilot/from-run", json={"run_id": "nope", "freeze": True}) + assert res.status_code == 404 + row = await S.explore(monkeypatch, S.ELSEWHERE, text="一共几家店,没有引用。") + res = await client.post("/api/copilot/from-run", json={"run_id": row.id, "freeze": True}) + assert res.status_code == 409 and res.json()["detail"] == certify.NO_QUERIES diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index 29929850..f3a63aee 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -4522,6 +4522,8 @@ export const LEDGER_TEXT = { /** 副标题:比对在封存之后;事后补比、打开时按需比的另写一句 */ sub: '封存后比对,不在封存范围内', retro: '按现在的口径回看', + /** 从问数据固化出的口径发布之后,对来源运行补比的那一次(服务端 trigger=certified) */ + certified: '按之后发布的口径回看', /** 标签的四种、认法的两种(第 4 波的证据面板、轮次摘要用同一套) */ kind: { caliber: '来自口径', related: '相关口径', reasoning: '这次的推理', unknown: '没有比对' }, how: { independent: '等价已证明', composed: '助手选用' }, diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index 68023974..1b93e375 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -2411,7 +2411,8 @@ export function decodeRun(events: RunEvent[], final?: RunFinal, opts?: DecodeOpt id: `lm-${seq}`, seq, kind: 'note', status: 'done', code: 'ledger_matched', level: typeof counts.skipped === 'number' && counts.skipped > 0 ? 'warn' : 'info', title: LEDGER_TEXT.timeline(counts), - sub: [LEDGER_TEXT.sub, retro ? LEDGER_TEXT.retro : ''].filter(Boolean).join(' · '), + sub: [LEDGER_TEXT.sub, d.trigger === 'certified' ? LEDGER_TEXT.certified : retro ? LEDGER_TEXT.retro : ''] + .filter(Boolean).join(' · '), }) break } diff --git a/scripts/check-decode.mjs b/scripts/check-decode.mjs index dc2cd0a6..e0f1b875 100644 --- a/scripts/check-decode.mjs +++ b/scripts/check-decode.mjs @@ -1850,6 +1850,9 @@ await section('口径账本:封存后比对已发布口径(ledger.matched) const late = run({ counts: { proven: 0, related: 0, none: 1, skipped: 1 }, trigger: 'backfill' }).find((s) => s.code === 'ledger_matched') check('事后补比的写明按现在的口径回看;没比完的是提醒', late?.title === '比对已发布口径:1 条没有相关的已发布口径,1 条没有比完' && late?.sub === '封存后比对,不在封存范围内 · 按现在的口径回看' && late?.level === 'warn', `${late?.title} / ${late?.sub} / ${late?.level}`) + const certified = run({ counts: { proven: 1 }, trigger: 'certified' }).find((s) => s.code === 'ledger_matched') + check('固化出的口径发布后补比的写明按之后发布的口径回看', certified?.sub === '封存后比对,不在封存范围内 · 按之后发布的口径回看', + certified?.sub) const odd = run({ counts: 'x' }).find((s) => s.code === 'ledger_matched') check('缺字段:只说比对过,不出 undefined、不瞎填数', odd?.title === '比对已发布口径' && !/undefined|NaN/.test(`${odd?.title}${odd?.sub}`), odd?.title)