mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 05:01:57 +00:00
ThearchyHelios
b6cb5816db
为什么：一轮的运行被审批人驳回之后点「重新运行本轮」，「之前的尝试」那一行写「已手动停止」，用户其实没有停过它。会话 列表的 last_status 把被驳回的一轮归成 cancelled，左栏不标——在等审批的那一问被驳回之后，看起来像是答完了（以前驳回 记成失败，左栏标的是「失败」）。 做了什么： - store/chat 的 attemptOf：被驳回的那次记 outcome=rejected，摘要写「被驳回：张工 驳回了「查数」：…」；老留档里记成 cancelled 的不改写。ChatPage 的尝试徽标认 rejected。 - api/conversations 的 last_status：前端落库时 meta 里记着 rejected 的、页面关掉没落库而运行是 cancelled + end_reason=rejected 的，都记 rejected；手动停止照旧 cancelled。types 的 last_status 加上 rejected。 - 左栏：打开过的看最后一轮（相位 cancelled 且带着驳回），没打开过的看 last_status，都标「被驳回」（不画异常色）。 测试：test_conversation_status 加一条（两种认法的被驳回、两种手动停止），改之前失败；check-chat 加一节被驳回（刷新后 认得出、补救区的话、左栏两种、重新运行之后的尝试记录），把前端改动退回去回跑，那一节 4 项不通过，改完通过。
349 lines
15 KiB
Python
349 lines
15 KiB
Python
"""会话列表认得每个会话的最后一轮停在哪；轮次的可信度元数据有自己的一列。
|
||
|
||
以前左栏只能给「这次打开页面以来取回过内存」的会话标状态。没打开过的会话，
|
||
哪怕最后一轮正停在审批上、或者已经失败，列表里也和正常完成的长得一样——
|
||
人得一个个点进去才知道哪里在等他。
|
||
|
||
元数据（出具档位、运行类别、未查库、结局、耗时…）以前挂在 review.meta 下，
|
||
于是没复核过的轮次 review 也不为空，「没复核过」和「复核过、没问题」分不开。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import timedelta
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
|
||
from app.db.base import SessionLocal, utcnow
|
||
from app.db.models import Approval, Conversation, ConversationTurn, Run
|
||
from app.main import app
|
||
|
||
|
||
@pytest.fixture
|
||
async def client():
|
||
transport = ASGITransport(app=app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
||
yield c
|
||
|
||
|
||
async def _conversation(client, title: str) -> str:
|
||
r = await client.post("/api/conversations", json={"kind": "chat", "title": title})
|
||
assert r.status_code == 201, r.text
|
||
return r.json()["id"]
|
||
|
||
|
||
async def _turn(client, conv_id: str, question: str = "订单数多少", **patch) -> dict:
|
||
r = await client.post(f"/api/conversations/{conv_id}/turns", json={"question": question})
|
||
assert r.status_code == 201, r.text
|
||
turn = r.json()
|
||
if patch:
|
||
r = await client.patch(f"/api/conversations/{conv_id}/turns/{turn['id']}", json=patch)
|
||
assert r.status_code == 200, r.text
|
||
turn = r.json()
|
||
return turn
|
||
|
||
|
||
async def _run(status: str, *, pending: bool = False, end_reason: str | None = None) -> str:
|
||
async with SessionLocal() as session:
|
||
run = Run(workflow_name="状态测试", status=status, end_reason=end_reason)
|
||
session.add(run)
|
||
await session.flush()
|
||
if pending:
|
||
session.add(Approval(run_id=run.id, node_id="gate", status="pending"))
|
||
await session.commit()
|
||
return run.id
|
||
|
||
|
||
async def _listed(client) -> dict[str, dict]:
|
||
r = await client.get("/api/conversations")
|
||
assert r.status_code == 200, r.text
|
||
return {c["id"]: c for c in r.json()}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# REQ-1：meta 自己一列
|
||
# --------------------------------------------------------------------------
|
||
|
||
META = {"v": 1, "runClass": "exploratory", "noQuery": True, "ms": 8200, "queries": 0}
|
||
|
||
|
||
async def test_turn_meta_is_stored_and_read_back_on_its_own(client):
|
||
conv = await _conversation(client, "元数据")
|
||
turn = await _turn(client, conv, meta=META, status="done", answer="12 家")
|
||
assert turn["meta"] == META
|
||
assert turn["review"] is None, "只写了 meta，review 得还是「没复核过」"
|
||
|
||
detail = (await client.get(f"/api/conversations/{conv}")).json()
|
||
assert detail["turns"][0]["meta"] == META
|
||
|
||
|
||
async def test_patching_review_and_meta_do_not_overwrite_each_other(client):
|
||
conv = await _conversation(client, "互不覆盖")
|
||
turn = await _turn(client, conv, meta=META)
|
||
review = {"verdict": "annotated", "note": "检索降级"}
|
||
r = await client.patch(f"/api/conversations/{conv}/turns/{turn['id']}", json={"review": review})
|
||
assert r.json()["meta"] == META
|
||
r = await client.patch(f"/api/conversations/{conv}/turns/{turn['id']}",
|
||
json={"meta": {**META, "ms": 9000}})
|
||
assert r.json()["review"] == review
|
||
assert r.json()["meta"]["ms"] == 9000
|
||
|
||
|
||
async def test_old_turns_with_meta_under_review_still_read_as_meta(client):
|
||
"""迁移之前写进去的：meta 挂在 review 下。读的时候由后端兜底，前端只认 turn.meta。"""
|
||
conv = await _conversation(client, "老数据")
|
||
legacy = {"verdict": "ok", "meta": {"v": 1, "outcome": "cancelled"}}
|
||
turn = await _turn(client, conv, review=legacy, status="error", error="已取消")
|
||
assert turn["meta"] == {"v": 1, "outcome": "cancelled"}
|
||
# 新列一旦写过就以它为准
|
||
r = await client.patch(f"/api/conversations/{conv}/turns/{turn['id']}", json={"meta": META})
|
||
assert r.json()["meta"] == META
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# REQ-2：列表里的最后一轮状态
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_the_list_says_where_each_conversation_last_stopped(client):
|
||
cases: dict[str, tuple[str, str | None]] = {}
|
||
|
||
conv = await _conversation(client, "完成")
|
||
run = await _run("succeeded")
|
||
await _turn(client, conv, run_id=run, status="done", answer="12")
|
||
cases[conv] = ("done", run)
|
||
|
||
conv = await _conversation(client, "失败")
|
||
run = await _run("failed")
|
||
await _turn(client, conv, run_id=run, status="error", error="查询超时")
|
||
cases[conv] = ("error", run)
|
||
|
||
conv = await _conversation(client, "取消")
|
||
await _turn(client, conv, status="error", error="已取消")
|
||
cases[conv] = ("cancelled", None)
|
||
|
||
conv = await _conversation(client, "重启挂起")
|
||
run = await _run("interrupted")
|
||
await _turn(client, conv, run_id=run, status="error", error="服务重启，这一轮中断了",
|
||
meta={"v": 1, "outcome": "suspended", "runId": run})
|
||
cases[conv] = ("suspended", run)
|
||
|
||
conv = await _conversation(client, "在跑")
|
||
run = await _run("running")
|
||
await _turn(client, conv, run_id=run)
|
||
cases[conv] = ("running", run)
|
||
|
||
conv = await _conversation(client, "等审批")
|
||
run = await _run("interrupted", pending=True)
|
||
await _turn(client, conv, run_id=run)
|
||
cases[conv] = ("waiting", run)
|
||
|
||
conv = await _conversation(client, "停着没审批")
|
||
run = await _run("interrupted")
|
||
await _turn(client, conv, run_id=run)
|
||
cases[conv] = ("suspended", run)
|
||
|
||
conv = await _conversation(client, "页面关了但跑完了")
|
||
run = await _run("succeeded")
|
||
await _turn(client, conv, run_id=run)
|
||
cases[conv] = ("done", run)
|
||
|
||
conv = await _conversation(client, "页面关了而且失败了")
|
||
run = await _run("failed")
|
||
await _turn(client, conv, run_id=run)
|
||
cases[conv] = ("error", run)
|
||
|
||
conv = await _conversation(client, "正在建图")
|
||
await _turn(client, conv)
|
||
cases[conv] = ("running", None)
|
||
|
||
listed = await _listed(client)
|
||
got = {cid: (listed[cid]["last_status"], listed[cid]["last_run_id"]) for cid in cases}
|
||
assert got == cases
|
||
|
||
|
||
async def test_a_rejected_turn_is_listed_as_rejected_not_cancelled(client):
|
||
"""被审批人驳回结束的一轮（运行 cancelled + end_reason=rejected，05-已知遗留 (a)）：列表说「被驳回」，不和手动停止混成
|
||
cancelled——发问的人不一定知道被驳回了，以前驳回记成失败，列表标的是「失败」。认法两条：前端落库时记下的
|
||
meta.rejected，和页面关掉、轮次还停在 running 时运行的 end_reason。手动停止照旧是 cancelled。"""
|
||
cases: dict[str, tuple[str, str | None]] = {}
|
||
|
||
conv = await _conversation(client, "驳回：前端落库了")
|
||
run = await _run("cancelled", end_reason="rejected")
|
||
await _turn(client, conv, run_id=run, status="error", error="已取消",
|
||
meta={"v": 1, "runId": run, "outcome": "cancelled", "rejected": {"sub": "张工 驳回了「付款」"}})
|
||
cases[conv] = ("rejected", run)
|
||
|
||
conv = await _conversation(client, "驳回：页面关了")
|
||
run = await _run("cancelled", end_reason="rejected")
|
||
await _turn(client, conv, run_id=run)
|
||
cases[conv] = ("rejected", run)
|
||
|
||
conv = await _conversation(client, "手动停止：页面关了")
|
||
run = await _run("cancelled", end_reason="stopped")
|
||
await _turn(client, conv, run_id=run)
|
||
cases[conv] = ("cancelled", run)
|
||
|
||
conv = await _conversation(client, "手动停止：前端落库了")
|
||
run = await _run("cancelled", end_reason="stopped")
|
||
await _turn(client, conv, run_id=run, status="error", error="已取消", meta={"v": 1, "runId": run, "outcome": "cancelled"})
|
||
cases[conv] = ("cancelled", run)
|
||
|
||
listed = await _listed(client)
|
||
assert {cid: (listed[cid]["last_status"], listed[cid]["last_run_id"]) for cid in cases} == cases
|
||
detail = (await client.get(f"/api/conversations/{next(iter(cases))}")).json()
|
||
assert detail["last_status"] == "rejected"
|
||
|
||
|
||
async def test_only_the_last_turn_counts(client):
|
||
conv = await _conversation(client, "后来好了")
|
||
await _turn(client, conv, status="error", error="查询超时")
|
||
run = await _run("succeeded")
|
||
await _turn(client, conv, run_id=run, status="done", answer="好了")
|
||
assert (await _listed(client))[conv]["last_status"] == "done"
|
||
|
||
|
||
async def test_a_build_abandoned_long_ago_is_not_shown_as_running(client):
|
||
"""建图是跟着浏览器那条流走的：页面关了就没了，轮次却一直停在 running。"""
|
||
conv = await _conversation(client, "建到一半关了")
|
||
turn = await _turn(client, conv)
|
||
async with SessionLocal() as session:
|
||
row = await session.get(ConversationTurn, turn["id"])
|
||
row.updated_at = utcnow() - timedelta(hours=2)
|
||
await session.commit()
|
||
assert (await _listed(client))[conv]["last_status"] == "error"
|
||
|
||
|
||
async def test_an_empty_conversation_has_no_status(client):
|
||
conv = await _conversation(client, "空的")
|
||
item = (await _listed(client))[conv]
|
||
assert item["last_status"] is None and item["last_run_id"] is None
|
||
|
||
|
||
async def test_the_detail_carries_the_same_status(client):
|
||
conv = await _conversation(client, "详情也带")
|
||
run = await _run("interrupted", pending=True)
|
||
await _turn(client, conv, run_id=run)
|
||
detail = (await client.get(f"/api/conversations/{conv}")).json()
|
||
assert (detail["last_status"], detail["last_run_id"]) == ("waiting", run)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 重试之后：meta.runId 才是这一轮现在的运行
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_a_retried_turn_reports_the_new_run_not_the_old_one(client):
|
||
"""重试会换一次运行，可 run_id 列清不掉（TurnPatch 的 None 表示「不改」），
|
||
还指着上一次失败的运行；前端写的 meta.runId 才是权威。列表得按它报，
|
||
不然重试正在建图时，左栏还挂着上一次的「失败」。"""
|
||
conv = await _conversation(client, "重试了")
|
||
old = await _run("failed")
|
||
turn = await _turn(client, conv, run_id=old, status="error", error="查询超时",
|
||
meta={"v": 1, "runId": old})
|
||
# 重试：重新建图，还没有运行
|
||
r = await client.patch(f"/api/conversations/{conv}/turns/{turn['id']}",
|
||
json={"status": "running", "meta": {"v": 1, "runId": None}})
|
||
assert r.status_code == 200
|
||
item = (await _listed(client))[conv]
|
||
assert (item["last_status"], item["last_run_id"]) == ("running", None)
|
||
|
||
# 重试出来的运行起来了，还在跑
|
||
new = await _run("running")
|
||
await client.patch(f"/api/conversations/{conv}/turns/{turn['id']}",
|
||
json={"meta": {"v": 1, "runId": new}})
|
||
item = (await _listed(client))[conv]
|
||
assert (item["last_status"], item["last_run_id"]) == ("running", new)
|
||
detail = (await client.get(f"/api/conversations/{conv}")).json()
|
||
assert (detail["last_status"], detail["last_run_id"]) == ("running", new)
|
||
|
||
|
||
async def test_old_turns_without_run_id_in_meta_still_use_the_column(client):
|
||
conv = await _conversation(client, "老轮次")
|
||
run = await _run("interrupted", pending=True)
|
||
await _turn(client, conv, run_id=run, meta={"v": 1, "ms": 1200})
|
||
item = (await _listed(client))[conv]
|
||
assert (item["last_status"], item["last_run_id"]) == ("waiting", run)
|
||
|
||
|
||
async def test_turns_carry_updated_at_for_the_same_staleness_clock(client):
|
||
"""前端判断「建图断了」要和后端用同一只钟：后端按 updated_at 算。"""
|
||
conv = await _conversation(client, "带时间")
|
||
turn = await _turn(client, conv)
|
||
assert turn.get("updated_at"), turn
|
||
detail = (await client.get(f"/api/conversations/{conv}")).json()
|
||
assert detail["turns"][0]["updated_at"]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 活跃时间：只是打开一个旧会话，不算在这里做了事
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def _age(conv_id: str, days: int) -> None:
|
||
async with SessionLocal() as session:
|
||
row = await session.get(Conversation, conv_id)
|
||
row.last_active_at = utcnow() - timedelta(days=days)
|
||
await session.commit()
|
||
|
||
|
||
async def _order(client, *ids: str) -> list[str]:
|
||
r = await client.get("/api/conversations")
|
||
return [c["id"] for c in r.json() if c["id"] in ids]
|
||
|
||
|
||
async def test_catching_up_on_open_does_not_bump_the_conversation(client):
|
||
"""打开会话时前端会回源核对：补全截断的答案、记下运行已删、把断掉的建图写回
|
||
error、补交付错过的结果。这些都不是人在这里做了事——以前每一次都把
|
||
last_active_at 改成现在，旧会话一打开就跳到列表最上面、写着「刚刚」。"""
|
||
old = await _conversation(client, "上周的")
|
||
turn = await _turn(client, old, run_id=await _run("succeeded"), status="done", answer="十二")
|
||
recent = await _conversation(client, "昨天的")
|
||
await _turn(client, recent)
|
||
await _age(old, 7)
|
||
await _age(recent, 1)
|
||
before = (await _listed(client))[old]["last_active_at"]
|
||
assert await _order(client, old, recent) == [recent, old]
|
||
|
||
url = f"/api/conversations/{old}/turns/{turn['id']}"
|
||
for body in (
|
||
{"meta": {"v": 1, "clipped": "lost"}, "review": {}}, # 运行删了、答案补不回来
|
||
{"answer": "十二（完整版）", "meta": {"v": 1}, "review": {}}, # 截断的答案换成完整的
|
||
{"status": "error", "error": "上次没有跑完", "meta": {"v": 1}, "review": {}},
|
||
{"status": "done", "answer": "补交付", "meta": {"v": 1}, "review": {}},
|
||
):
|
||
r = await client.patch(url, json=body)
|
||
assert r.status_code == 200, r.text
|
||
assert (await _listed(client))[old]["last_active_at"] == before, body
|
||
assert await _order(client, old, recent) == [recent, old]
|
||
|
||
|
||
@pytest.mark.parametrize("body", [
|
||
{"run_id": "RUN", "status": "running"}, # 按下运行
|
||
{"status": "running", "error": "", "answer": ""}, # 重试这一轮
|
||
{"question": "换个问法"},
|
||
])
|
||
async def test_starting_something_in_a_turn_still_bumps_it(client, body):
|
||
old = await _conversation(client, "上周的")
|
||
turn = await _turn(client, old)
|
||
recent = await _conversation(client, "昨天的")
|
||
await _turn(client, recent)
|
||
await _age(old, 7)
|
||
await _age(recent, 1)
|
||
if body.get("run_id") == "RUN":
|
||
body = {**body, "run_id": await _run("running")}
|
||
r = await client.patch(f"/api/conversations/{old}/turns/{turn['id']}", json=body)
|
||
assert r.status_code == 200, r.text
|
||
assert await _order(client, old, recent) == [old, recent]
|
||
|
||
|
||
async def test_a_new_turn_bumps_it(client):
|
||
old = await _conversation(client, "上周的")
|
||
await _turn(client, old)
|
||
recent = await _conversation(client, "昨天的")
|
||
await _turn(client, recent)
|
||
await _age(old, 7)
|
||
await _age(recent, 1)
|
||
await _turn(client, old, "接着问")
|
||
assert await _order(client, old, recent) == [old, recent]
|