mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 08:31:59 +00:00
ThearchyHelios
03c18c4d78
一、文案（前端约 1600 处，后端传到界面的约 880 处） - 术语统一：工作流层面用「运行」（继续运行、重新运行、试运行），节点和工具层面用「执行」，不再用「跑」； 总时长 / 执行时长 / 等待审批；服务端；助手；Agent；token；工作流；固定版本；模型接入；系统提示；批准 / 驳回；受管级别 - 口语改书面：取不到、没取回来、读不懂、连不上、多半、照样、兜底、揉成这类说法改成「无法 / 未能 / 失败 / 请」；去掉拟人写法 - 不再露出实现细节：界面上不写 checkpoint、同一拍、哈希（凭证处除外），不写改版历史（「以前…」「老数据」）； 报错、校验、门禁、自动修复里裸写的配置键名和枚举值换成界面上的中文标签，由新的 backend/app/engine/labels.py 统一提供； 发布弹窗里结论句的选项不再是 off / require_citation - 同一件事只留一种说法：生成阶段文案、出具档位说明、画布锁定提示各合成一份 - 去掉界面上的工业场景示例（MES、ERP、「张工」「王工」）和真实产品名，换成通用示例 二、说错的提示（都有测试证明） - 数据库测连接失败时的专用提示以前从不出现（按报错文字全等比较，永远比不上）：改为按错误类型判断，不再拼出重复的前缀 - 「设置 → 供应商」改指「设置 → 模型接入」；「在设置里建数据源」改指「数据」页 - 缺少 API Key、模型接入已停用的报错不再夹英文，写明去哪里填 - 两处原因说错（「这种引用在后续版本支持」「这一版还不支持」）改成真实原因 - 节点说明里不准确的说法改正：代码节点在 microVM 或系统沙箱中执行；报告撰写、口径卡的说明不再说过头 - 导出的证据 CSV 里两列同名，第三列改为「片段编号」 三、前后端对齐 - 前端按后端原文识别日志、报错的地方，新旧两种写法都认得（历史运行里存的是旧文字），逐条补了新旧原文的核对 - 后端交回模型的原指令放进单独的 for_model 字段，前端在解析响应时去掉，不会显示到界面上 四、检查 - 检查脚本里按原文匹配的断言改成新文案；改完会变成永远成立的反向断言逐条修正，并做了变异验证 - 新增 check-copy（静态扫描前端界面字符串和后端上界面的文案）：再出现「跑」「后端」「多半」「令牌」、 界面上的 Copilot、裸写的配置键名、半角标点等就报错；确需保留的写进 scripts/copy-allow.json 并注明理由
106 lines
4.5 KiB
Python
106 lines
4.5 KiB
Python
"""发布门禁对引用模式契约（report_from）的判断。
|
||
|
||
以前 `_lint_contract` 不管有没有 report_from，一律要求 narrative：模板⑨这种只用
|
||
report_from 核对报告文档的契约，数字回指由报告撰写节点的文档保证、根本用不上
|
||
narrative，却因此发不成受管版本。有 report_from 就不再要 narrative；report_from
|
||
指错（不是报告撰写节点、不在出口上游）照样挡住发布。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
|
||
from app.engine.governance import lint_for_publish
|
||
from app.engine.schema import GraphSpec
|
||
from app.main import app
|
||
from app.seed import TEMPLATES
|
||
|
||
|
||
@pytest.fixture
|
||
async def client():
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||
yield c
|
||
|
||
|
||
def traceable() -> dict:
|
||
"""模板⑨（可追溯周报）的图：契约只写 report_from，没有 narrative。"""
|
||
tpl = next(t for t in TEMPLATES if t["name"].startswith("⑨"))
|
||
return copy.deepcopy({"nodes": tpl["nodes"], "edges": tpl["edges"]})
|
||
|
||
|
||
def contract_of(graph: dict) -> dict:
|
||
return next(n for n in graph["nodes"] if n["id"] == "done")["data"]["config"]["contract"]
|
||
|
||
|
||
def lint(graph: dict, level: str = "governed") -> list:
|
||
return lint_for_publish(GraphSpec.model_validate(graph), level=level).issues
|
||
|
||
|
||
async def publish(client, graph: dict, level: str = "governed") -> dict:
|
||
wf = (await client.post("/api/workflows", json={"name": "可追溯周报-门禁", "graph": graph})).json()
|
||
body = (await client.post(f"/api/workflows/{wf['id']}/publish", json={"level": level})).json()
|
||
await client.delete(f"/api/workflows/{wf['id']}")
|
||
return body
|
||
|
||
|
||
def test_the_template_contract_has_no_narrative_and_needs_none():
|
||
graph = traceable()
|
||
assert "narrative" not in contract_of(graph) and contract_of(graph)["report_from"] == "write"
|
||
errors = [i for i in lint(graph) if i.level == "error"]
|
||
assert errors == [], [i.message for i in errors]
|
||
|
||
|
||
async def test_the_template_publishes_as_governed(client):
|
||
body = await publish(client, traceable())
|
||
assert body["ok"] is True, body["issues"]
|
||
assert all(i["level"] == "warning" for i in body["issues"])
|
||
|
||
|
||
@pytest.mark.parametrize("target, why", [("fetch", "不是「报告撰写」"), ("ghost", "不存在")])
|
||
async def test_a_report_from_pointing_elsewhere_still_blocks(client, target, why):
|
||
graph = traceable()
|
||
contract_of(graph)["report_from"] = target
|
||
body = await publish(client, graph)
|
||
assert body["ok"] is False
|
||
bad = [i for i in body["issues"] if i["level"] == "error"]
|
||
assert [i["code"] for i in bad] == ["contract.report_from_invalid"], bad
|
||
assert why in bad[0]["message"] and bad[0]["node_id"] == "done"
|
||
|
||
|
||
async def test_a_report_that_is_not_upstream_still_blocks(client):
|
||
graph = traceable()
|
||
# 报告撰写节点挪到出口后面：出口核对时那份报告还没写出来
|
||
graph["edges"] = [e for e in graph["edges"] if e["target"] != "done"] + [
|
||
{"source": "caliber", "target": "done"}, {"source": "done", "target": "write"}]
|
||
body = await publish(client, graph)
|
||
assert body["ok"] is False
|
||
assert any(i["code"] == "contract.report_from_invalid" and "上游" in i["message"]
|
||
for i in body["issues"] if i["level"] == "error"), body["issues"]
|
||
|
||
|
||
@pytest.mark.parametrize("level, expected", [("governed", "error"), ("published", "warning")])
|
||
def test_without_report_from_or_narrative_the_gate_still_asks(level, expected):
|
||
graph = traceable()
|
||
del contract_of(graph)["report_from"]
|
||
[issue] = [i for i in lint(graph, level) if i.code == "contract.report_from_missing"]
|
||
assert issue.level == expected and issue.node_id == "done"
|
||
assert "「报告来自」" in issue.message and "「叙述」" in issue.message
|
||
|
||
|
||
def test_an_empty_report_from_counts_as_missing():
|
||
graph = traceable()
|
||
contract_of(graph)["report_from"] = ""
|
||
assert any(i.code == "contract.report_from_missing" and i.level == "error" for i in lint(graph))
|
||
|
||
|
||
def test_a_narrative_contract_is_unchanged():
|
||
graph = traceable()
|
||
contract = contract_of(graph)
|
||
del contract["report_from"]
|
||
contract["narrative"] = "{{ nodes.write.text }}"
|
||
# _lint_contract 对叙述模式照旧；受管级别另有门禁 G1 要求核对报告撰写节点的文档
|
||
assert [i for i in lint(graph, "published") if i.level == "error"] == []
|
||
assert [i.code for i in lint(graph) if i.level == "error"] == ["governed.report_from_required"]
|