vent_agent_2.0/scripts/http_e2e.py

"""HTTP 层端到端验证脚本（临时工具，不属于 pytest 套件）。

覆盖：SSE 流式契约、响应头、落库、会话路由、中断/恢复闭环、调度手动触发。
运行：python scripts/http_e2e.py
"""

from __future__ import annotations

import json
import os
import sys
import time

import httpx

# 允许指向别的实例（例如另起一个 MySQL 档的服务端口）做方言专项验证：
#   BASE=http://127.0.0.1:8081 python scripts/http_e2e.py
BASE = os.getenv("BASE", "http://127.0.0.1:8080").rstrip("/")
USER = os.getenv("E2E_USER", "http_e2e")
PASSED: list[str] = []
FAILED: list[str] = []


def check(name: str, ok: bool, detail: str = "") -> None:
    (PASSED if ok else FAILED).append(name)
    print(f"  {'PASS' if ok else 'FAIL'}  {name}" + (f"  — {detail}" if detail else ""))


def sse_events(response) -> list[dict]:
    """从 httpx 流式响应里解析出全部事件。"""
    events: list[dict] = []
    for line in response.iter_lines():
        if line.startswith("data: "):
            try:
                events.append(json.loads(line[6:]))
            except json.JSONDecodeError:
                pass
    return events


def test_stream_basic() -> str | None:
    print("\n[1] POST /api/chat 流式契约")
    with httpx.Client(timeout=180) as client:
        with client.stream(
            "POST", f"{BASE}/api/chat",
            data={"message": "用一句话说明矿井通风的目的", "user_name": USER, "mode": "full"},
        ) as res:
            headers = res.headers
            events = sse_events(res)

    check("HTTP 200", res.status_code == 200, str(res.status_code))
    check("Content-Type 是 event-stream", "text/event-stream" in headers.get("content-type", ""))
    check("响应头 X-Session-Id", bool(headers.get("x-session-id")))
    check("响应头 X-Accel-Buffering=no", headers.get("x-accel-buffering") == "no")
    check("响应头 Cache-Control=no-cache", "no-cache" in headers.get("cache-control", ""))
    check("响应头 X-Trace-Id", bool(headers.get("x-trace-id")))

    types = [e["type"] for e in events]
    check("有 agent_start", "agent_start" in types)
    check("有 token", "token" in types)
    check("有 citations", "citations" in types)
    check("有 done", "done" in types)
    check("无 error", "error" not in types, str([e.get("message", "")[:60] for e in events if e["type"] == "error"]))
    check("done 在最后一个", types[-1] == "done", str(types[-3:]))

    # 每帧必须带 session_id
    check("每帧带 session_id", all(e.get("session_id") for e in events))
    sid = events[0].get("session_id") if events else None
    return sid


def test_persist_and_history(sid: str) -> None:
    print("\n[2] 落库与会话路由")
    with httpx.Client(timeout=60) as client:
        r = client.get(f"{BASE}/api/chat/history/{sid}", params={"user_name": USER})
        check("历史 200", r.status_code == 200, str(r.status_code))
        items = r.json().get("items", [])
        roles = [m["msg_role"] for m in items]
        check("历史有 user+assistant", roles[:2] == ["user", "assistant"], str(roles))
        check("顺序正确（user 在前）", roles[0] == "user")
        assistant = next((m for m in items if m["msg_role"] == "assistant"), {})
        check("assistant 内容非空", bool((assistant.get("content") or "").strip()))
        # 正文里不该混入工具结果 JSON
        body = assistant.get("content") or ""
        check("正文未混入工具结果 JSON", not body.strip().startswith('{"data"'), body[:80].replace("\n", " "))

        r = client.get(f"{BASE}/api/sessions", params={"user_name": USER})
        check("会话列表 200", r.status_code == 200)
        check("列表含该会话", any(s["id"] == sid for s in r.json().get("items", [])))

        r = client.get(f"{BASE}/api/sessions/{sid}/context", params={"user_name": USER})
        check("上下文用量 200", r.status_code == 200)
        latest = (r.json() or {}).get("latest") or {}
        check("有用量记录", int(latest.get("total_tokens") or 0) > 0, str(latest.get("total_tokens")))
        check("口径为 provider", latest.get("usage_source") == "provider", str(latest.get("usage_source")))

        r = client.get(f"{BASE}/api/usage/today", params={"user_name": USER})
        check("用量 200", r.status_code == 200)
        check("当日有消耗", int(r.json().get("total_tokens") or 0) > 0)

        r = client.get(f"{BASE}/api/model")
        check("模型信息 200", r.status_code == 200)
        d = r.json()
        check("工具数 = 4", d["agent"]["tools"]["count"] == 4, str(d["agent"]["tools"]["count"]))
        check("interrupt_on 含计划审批",
              "request_plan_approval" in d["agent"]["interrupt_on"], str(d["agent"]["interrupt_on"]))


def test_404_and_errors() -> None:
    print("\n[3] 错误语义")
    with httpx.Client(timeout=60) as client:
        r = client.post(f"{BASE}/api/chat", data={
            "message": "hi", "user_name": USER, "session_id": "not-exist-id",
        })
        check("不存在的会话 → 404", r.status_code == 404, str(r.status_code))

        r = client.post(f"{BASE}/api/chat", data={
            "message": "hi", "user_name": "other_user", "session_id": "not-exist-id",
        })
        check("他人会话 → 404（不泄露存在性）", r.status_code == 404, str(r.status_code))

        r = client.get(f"{BASE}/api/chat/interrupt", params={"session_id": "not-exist-id", "user_name": USER})
        check("中断查询 不存在的会话 → 404", r.status_code == 404, str(r.status_code))


def test_interrupt_loop() -> None:
    """plan 模式 → agent 调 request_plan_approval → interrupt → resume。"""
    print("\n[4] 中断/恢复闭环")
    with httpx.Client(timeout=240) as client:
        with client.stream(
            "POST", f"{BASE}/api/chat",
            data={
                "message": "请制定一个三步计划：把一句话总结成简报并写到 /workspace/brief.md。先提交计划等我批准。",
                "user_name": USER, "mode": "plan",
            },
        ) as res:
            events = sse_events(res)
        sid = None
        for e in events:
            sid = e.get("session_id") or sid
        types = [e["type"] for e in events]
        print("      事件:", types)

        if "interrupt" not in types:
            check("触发 interrupt", False, f"agent 未请求审批；事件={types}")
            return

        check("触发 interrupt", True)
        intr = next(e for e in events if e["type"] == "interrupt")
        check("interrupt 带 tool_approval", bool(intr.get("tool_approval")),
              json.dumps(intr, ensure_ascii=False)[:150])

        # 刷新场景：只读查询挂起的中断
        r = client.get(f"{BASE}/api/chat/interrupt", params={"session_id": sid, "user_name": USER})
        check("GET /chat/interrupt → 200", r.status_code == 200, str(r.status_code))
        if r.status_code == 200:
            check("interrupts 非空", len(r.json().get("interrupts", [])) > 0)

        # 批准并续跑
        r = client.post(f"{BASE}/api/chat/resume", json={
            "session_id": sid, "thread_id": sid, "user_name": USER,
            "approved": True, "status": "answered",
        })
        check("resume 200", r.status_code == 200, str(r.status_code))
        if r.status_code == 200:
            texts = []
            for line in r.iter_lines():
                if line.startswith("data: "):
                    try:
                        texts.append(json.loads(line[6:]))
                    except json.JSONDecodeError:
                        pass
            rtypes = [e["type"] for e in texts]
            print("      resume 事件:", rtypes)
            check("恢复后有输出", "token" in rtypes or "tool_call" in rtypes, str(rtypes))
            check("恢复后正常收尾", rtypes[-1] == "done" and "error" not in rtypes,
                  str([e.get("message", "")[:60] for e in texts if e["type"] == "error"]))

        # 已恢复：再查应无挂起
        r = client.get(f"{BASE}/api/chat/interrupt", params={"session_id": sid, "user_name": USER})
        check("恢复后无挂起（204）", r.status_code == 204, str(r.status_code))


def test_scheduler() -> None:
    print("\n[5] 定时任务")
    with httpx.Client(timeout=240) as client:
        r = client.post(f"{BASE}/api/schedules", json={
            "title": "e2e 任务", "cron": "0 9 * * *",
            "prompt": "用一句话说明今天要做什么（这是定时任务自检）。",
            "user_name": USER,
        })
        check("创建任务 200", r.status_code == 200, r.text[:200])
        if r.status_code != 200:
            return
        sid_task = r.json()["id"]

        r = client.post(f"{BASE}/api/schedules/{sid_task}/run")
        check("手动触发 200", r.status_code == 200, r.text[:250])
        if r.status_code == 200:
            body = r.json()
            check("触发有输出", bool((body.get("output") or "").strip()), str(body.get("output"))[:80])
            check("有独立会话", bool(body.get("session_id")))
            check("记录了 token", int(body.get("input_tokens") or 0) > 0, str(body.get("input_tokens")))

        r = client.get(f"{BASE}/api/schedules/{sid_task}/runs")
        check("执行记录 200", r.status_code == 200)
        runs = r.json().get("items", [])
        check("有成功记录", any(x["status"] == "success" for x in runs), str([x["status"] for x in runs]))

        r = client.post(f"{BASE}/api/schedules", json={
            "title": "坏 cron", "cron": "not a cron", "prompt": "x", "user_name": USER,
        })
        check("非法 cron → 400", r.status_code == 400, str(r.status_code))

        r = client.post(f"{BASE}/api/schedules/{sid_task}/pause")
        check("暂停 200", r.status_code == 200 and r.json().get("enabled") is False)
        r = client.delete(f"{BASE}/api/schedules/{sid_task}")
        check("删除 200", r.status_code == 200 and r.json().get("deleted") is True)


def main() -> int:
    started = time.time()
    print(f"目标 {BASE}")
    sid = test_stream_basic()
    if sid:
        test_persist_and_history(sid)
    test_404_and_errors()
    test_interrupt_loop()
    test_scheduler()

    print(f"\n{'=' * 56}")
    print(f"通过 {len(PASSED)} / {len(PASSED) + len(FAILED)}  耗时 {time.time() - started:.1f}s")
    if FAILED:
        print("失败项：")
        for name in FAILED:
            print("  -", name)
        return 1
    print("全部通过")
    return 0


if __name__ == "__main__":
    sys.exit(main())