"""后台挂起「分层 DAG」实测脚本：带一个并行 stage 的业务链。

POST 启动一个 dag 作业（stage1 两席位并行 → stage2 → stage3），轮询打印每帧的
dispatchChain 节点状态，直观看到：①stage1 两席位**同时 active**（并行）；②stage2/3
在前一 stage 全 done 后才 active（串行）；③orchestrationPlan 在响应里回得来（持久化生效）。

用法：PYTHONPATH=. .venv/Scripts/python scripts/probe_roundtable_dag_job.py
环境：GATEWAY_URL 默认 http://127.0.0.1:8001；AUTH_COOKIE 可选。
"""

from __future__ import annotations

import asyncio
import os
import time

import httpx

GATEWAY_URL = os.getenv("GATEWAY_URL", "http://127.0.0.1:8001").rstrip("/")
AUTH_COOKIE = os.getenv("AUTH_COOKIE", "")
AUTH_BEARER = os.getenv("AUTH_BEARER", "")

# 4 个已 seed 的圆桌席位。stage1 = 风险审查 ‖ 方案设计（并行），stage2 = 执行规划，stage3 = 总结。
AGENTS = [
    {"agentId": "roundtable-risk-review", "name": "风险审查"},
    {"agentId": "roundtable-solution-design", "name": "方案设计"},
    {"agentId": "roundtable-execution-plan", "name": "执行规划"},
    {"agentId": "roundtable-summary", "name": "方案总结"},
]
PLAN = {
    "mode": "dag",
    "stages": [
        {"id": "s0", "agentIds": ["roundtable-risk-review", "roundtable-solution-design"], "goal": "并行评估风险并产出初步方案"},
        {"id": "s1", "agentIds": ["roundtable-execution-plan"], "goal": "据风险与方案制定执行规划"},
        {"id": "s2", "agentIds": ["roundtable-summary"]},
    ],
    "finalSynthesis": True,
    "coordinatorPrompt": "先并行评估、再规划执行、最后总结，逐阶段承接。",
}
INTENT = "为一家中型制造企业设计一套数字化转型的整体方案"
TERMINAL = {"done", "error", "cancelled"}


def _headers() -> dict[str, str]:
    h = {"Content-Type": "application/json"}
    if AUTH_COOKIE:
        h["Cookie"] = AUTH_COOKIE
    if AUTH_BEARER:
        h["Authorization"] = f"Bearer {AUTH_BEARER}"
    return h


def _chain_str(nodes: list[dict]) -> str:
    sym = {"pending": ".", "active": ">RUN", "done": "DONE"}
    return "  ".join(f"{n['name']}={sym.get(n['state'], '?')}" for n in nodes)


async def main() -> None:
    async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
        print(f"[probe] POST {GATEWAY_URL}/api/roundtable-jobs  mode=dag (stage1 并行)")
        resp = await client.post(
            f"{GATEWAY_URL}/api/roundtable-jobs",
            headers=_headers(),
            json={"intent": INTENT, "agents": AGENTS, "orchestrationMode": "dag", "orchestrationPlan": PLAN},
        )
        if resp.status_code != 201:
            print(f"[probe] start failed: HTTP {resp.status_code}: {resp.text[:500]}")
            return
        job = resp.json()
        job_id = job["id"]
        # 验证 plan 在响应里回得来（持久化 + _to_response 生效）
        plan_back = job.get("orchestrationPlan")
        print(f"[probe] started {job_id}  orchestrationMode={job.get('orchestrationMode')}  "
              f"orchestrationPlan stages={len((plan_back or {}).get('stages') or [])}")

        last = ""
        t0 = time.perf_counter()
        saw_parallel = False
        while True:
            await asyncio.sleep(1.0)
            r = await client.get(f"{GATEWAY_URL}/api/roundtable-jobs/{job_id}", headers=_headers())
            if r.status_code != 200:
                print(f"[probe] get failed: HTTP {r.status_code}: {r.text[:300]}")
                return
            j = r.json()
            nodes = j.get("dispatchChain") or []
            active_seats = [n["name"] for n in nodes if n["state"] == "active"]
            if len(active_seats) >= 2:
                saw_parallel = True
            line = f"status={j['status']} phase={j['phase']} consensus={j['consensusPercentage']}%  [{_chain_str(nodes)}]"
            if line != last:
                print(f"[probe] +{time.perf_counter() - t0:5.1f}s  {line}")
                last = line
            if j["status"] in TERMINAL:
                print(f"\n[probe] terminal: {j['status']}")
                if j["status"] == "error":
                    print(f"[probe] error detail: {j.get('error')}")
                # 对话顺序证明 stage 串行
                seats = [d for d in (j.get("dialogues") or []) if d["role"] == "seat"]
                print(f"[probe] 席位交付顺序: {[d['agentId'] for d in seats]}")
                leaders = [d for d in (j.get("dialogues") or []) if d["role"] == "leader"]
                print(f"[probe] 总控发言轮数: {len(leaders)}（3 stage 派活 + 1 收口 = 4）")
                print(f"[probe] 是否观察到 stage 内并行（≥2 席位同时 active）: {'是 ✅' if saw_parallel else '否 ❌'}")
                sample = next((d["content"] for d in seats if d.get("content")), "")
                print(f"[probe] 席位交付样例: {sample[:80]}")
                return


if __name__ == "__main__":
    asyncio.run(main())