mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:31:58 +00:00
ThearchyHelios
5b4bd91476
为什么：问数据的答案要和已治理口径接起来（next-round 02 做法 2、3、9、10；总方案 X1、X2、X3、X12，拍板 L2、L4、L8）。
- 比对（ledger/match.py）：探索运行（chat、canvas、trial、personal_app）成功结束、封存完、有查询快照时，在 runner 封存之后和
notify_run 同一个位置放到后台；候选口径是每个工作流最新的已发布未退役版本里查库的调用工具节点（受管、已发布都算），按数据源
和表预筛；逐个用 match_template 比，schema 用查询当时冻结的表结构、列名取同一份快照里冻结的数据目录。每次运行限时 8 秒、在
线程里、同时最多 2 个；超时、快照被改过或读不出、SQL 解析不了照实记 skipped。启动时（run_manager.setup，和调度循环同一个开关）
补比最近 7 天封存了却没比过的运行，只读工件和版本表。
- 记法（ledger/store.py，拍板 L8）：内容寻址工件 kind=caliber_match（agentlab.caliber_match/1）+ run_append 的 sealed 方式追加
ledger.matched（core/events.py 登记）；工件里不放时间和工作流名，同一份封存再比一次哈希一样；meta 记 calibers / related / counts。
- 标签（ledger/labels.py）：单元格按查询的比对状态；指标要每个输入都等价于同一份口径同一版、绑定一致，式子按「哪个取数节点、
第几行、哪一列」换完和口径卡上的一字不差。caliber（how：independent 等价已证明；composed 助手选用留给第 3 波）/ related /
reasoning / unknown；整题判断附还缺的入口字段。
- 读（ledger/view.py）：GET /api/runs/{id}/ledger（ledger_api.runs_router，和证据同一道门），没比对过、能比的按需排上；证据
片段接口加 ledger 键（没比对过的运行没有这个键，返回一字不差）；报告导出的面板最上面多「口径」一步并写明「封存后比对，不在
封存范围内」。遮罩按读的那一刻重滤：绑定的值是这次运行查到过的遮罩值时不给值，差异条目和逐行对照走 _Shown。
- 来源链：POST /api/runs 加 origin_ref（只在 origin=chat 时收，服务端核对对话访问权和轮次归属，写进 Run.origin 和封存的
run.origin）；WorkflowIn 加 origin（run_id、conversation_id、turn_id，核对规则同 from-run）；报告导出和可验证报告下载各记一条
审计 report.exported。可验证报告列封存后追加的事件时给 ledger.matched 一句只有条数的说明。
- 统计（ledger/stats.py）：E1–E4 从工件、运行、查询快照、审计现算，GET /api/ledger/stats（搭建者）；E3 用 app/chat/questions
的聚类加按（数据源、normalize_sql 文本）的确定性重复；评测直接调 compute_stats（可点名 run_ids）。对账反馈、认证转化、按发布
的召回在后面几波，照实写没有数据。
测试：test_ledger_match.py（真跑探索运行：等价、相关、没有相关口径、写死的码值、非探索运行、超时、快照被改、解析失败、预筛、
指标两种、级别、只认最新版、补比）、test_ledger_api.py（账本接口、按需比、遮罩、证据片段、导出、审计、统计）、
test_run_origin_ref.py（origin_ref 与画布来源的核对）。
134 lines
6.6 KiB
Python
134 lines
6.6 KiB
Python
"""运行 ↔ 轮次的来源链（next-round 02 做法 9）：POST /api/runs 的 origin_ref、「在画布里打开」的 WorkflowIn.origin。
|
||
|
||
- origin_ref 只在来源是问数据时收：核对对话用不用得了、对话在不在、这一轮是不是这段对话的，写进 Run.origin 和封存的 run.origin；
|
||
别的来源带它、伪造的（别的对话的轮次、不存在的对话）一律拒收；
|
||
- 新建工作流带 origin（run_id、conversation_id、turn_id）：运行读不到、轮次不属于对话的拒收，对得上的记进 workflows.origin，
|
||
形状同从运行提取（/copilot/from-run）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
from sqlalchemy import select
|
||
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import Run, RunEvent, Workflow
|
||
from app.engine.runner import run_manager
|
||
from app.main import app
|
||
|
||
GRAPH = {"nodes": [{"id": "start", "type": "input", "position": {"x": 0, "y": 0},
|
||
"data": {"label": "开始", "config": {"fields": [{"name": "question"}]}}},
|
||
{"id": "out", "type": "output", "position": {"x": 0, "y": 0},
|
||
"data": {"label": "成果", "config": {"fields": [{"name": "答案", "value": "{{ input.question }}"}]}}}],
|
||
"edges": [{"source": "start", "target": "out"}]}
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def engine_up():
|
||
await run_manager.setup()
|
||
yield
|
||
await run_manager.shutdown()
|
||
|
||
|
||
@pytest.fixture
|
||
async def client():
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||
yield c
|
||
|
||
|
||
async def conversation_with_turn(client) -> tuple[str, str]:
|
||
res = await client.post("/api/conversations", json={"title": "问数据"})
|
||
assert res.status_code == 201, res.text
|
||
cid = res.json()["id"]
|
||
res = await client.post(f"/api/conversations/{cid}/turns", json={"question": "本周销售额多少"})
|
||
assert res.status_code == 201, res.text
|
||
return cid, res.json()["id"]
|
||
|
||
|
||
async def wait(run_id: str) -> Run:
|
||
for _ in range(200):
|
||
await asyncio.sleep(0.05)
|
||
async with SessionLocal() as session:
|
||
row = await session.get(Run, run_id)
|
||
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
|
||
return row
|
||
raise AssertionError("运行没跑完")
|
||
|
||
|
||
async def test_a_chat_run_records_its_conversation_and_turn(client):
|
||
cid, tid = await conversation_with_turn(client)
|
||
res = await client.post("/api/runs", json={"graph": GRAPH, "input": {"question": "本周销售额多少"}, "origin": "chat",
|
||
"origin_ref": {"conversation_id": cid, "turn_id": tid}})
|
||
assert res.status_code == 201, res.text
|
||
body = res.json()
|
||
assert body["origin_kind"] == "chat" and body["origin"] == {"conversation_id": cid, "turn_id": tid}
|
||
row = await wait(body["id"])
|
||
async with SessionLocal() as session:
|
||
[event] = (await session.execute(select(RunEvent).where(
|
||
RunEvent.run_id == row.id, RunEvent.type == "run.origin"))).scalars()
|
||
# 在封存范围内：事后改不了
|
||
assert event.seq <= row.manifest_seq
|
||
assert event.data["kind"] == "chat" and event.data["conversation_id"] == cid and event.data["turn_id"] == tid
|
||
|
||
|
||
async def test_origin_ref_is_only_for_chat_runs(client):
|
||
cid, tid = await conversation_with_turn(client)
|
||
res = await client.post("/api/runs", json={"graph": GRAPH, "origin": "canvas",
|
||
"origin_ref": {"conversation_id": cid, "turn_id": tid}})
|
||
assert res.status_code == 422 and res.json()["code"] == "origin_ref_not_allowed"
|
||
assert res.json()["field"] == "origin_ref"
|
||
|
||
|
||
async def test_forged_origin_refs_are_refused(client):
|
||
cid, tid = await conversation_with_turn(client)
|
||
other, other_turn = await conversation_with_turn(client)
|
||
for ref in ({"conversation_id": "nope", "turn_id": tid}, {"conversation_id": cid, "turn_id": other_turn},
|
||
{"conversation_id": cid, "turn_id": "nope"}):
|
||
res = await client.post("/api/runs", json={"graph": GRAPH, "origin": "chat", "origin_ref": ref})
|
||
assert res.status_code == 422, (ref, res.text)
|
||
assert res.json()["code"] == "origin_ref_invalid"
|
||
|
||
|
||
async def test_someone_elses_conversation_is_refused(client, monkeypatch):
|
||
"""对话按人隔离（开登录后的使用者）：挂到别人的对话上，和读别人的对话一样回 403。"""
|
||
from app.api import conversations
|
||
|
||
cid, tid = await conversation_with_turn(client)
|
||
monkeypatch.setattr(conversations, "_isolated", lambda: "someone-else")
|
||
res = await client.post("/api/runs", json={"graph": GRAPH, "origin": "chat",
|
||
"origin_ref": {"conversation_id": cid, "turn_id": tid}})
|
||
assert res.status_code == 403
|
||
|
||
|
||
async def test_open_in_canvas_records_where_the_workflow_came_from(client):
|
||
cid, tid = await conversation_with_turn(client)
|
||
run = (await client.post("/api/runs", json={"graph": GRAPH, "origin": "chat",
|
||
"origin_ref": {"conversation_id": cid, "turn_id": tid}})).json()
|
||
await wait(run["id"])
|
||
res = await client.post("/api/workflows", json={"name": "问数据：本周销售额", "graph": GRAPH,
|
||
"origin": {"run_id": run["id"], "conversation_id": cid,
|
||
"turn_id": tid}})
|
||
assert res.status_code == 201, res.text
|
||
async with SessionLocal() as session:
|
||
wf = await session.get(Workflow, res.json()["id"])
|
||
assert wf.origin == {"run_id": run["id"], "conversation_id": cid, "turn_id": tid}
|
||
|
||
# 不带来源的照旧（手建的为空）
|
||
res = await client.post("/api/workflows", json={"name": "手建", "graph": GRAPH})
|
||
async with SessionLocal() as session:
|
||
assert (await session.get(Workflow, res.json()["id"])).origin is None
|
||
|
||
|
||
async def test_open_in_canvas_refuses_origins_that_do_not_check_out(client):
|
||
cid, tid = await conversation_with_turn(client)
|
||
other, other_turn = await conversation_with_turn(client)
|
||
run = (await client.post("/api/runs", json={"graph": GRAPH, "origin": "chat"})).json()
|
||
await wait(run["id"])
|
||
for origin in ({"run_id": "nope"}, {"run_id": run["id"], "conversation_id": cid, "turn_id": other_turn},
|
||
{"run_id": run["id"], "turn_id": tid}):
|
||
res = await client.post("/api/workflows", json={"name": "x", "graph": GRAPH, "origin": origin})
|
||
assert res.status_code == 422, (origin, res.text)
|
||
assert res.json()["code"] == "workflow_origin_invalid"
|