# ============================================================
# Mesh 人工审批恢复黄金测试
# 覆盖：批准后续跑依赖任务、驳回停止、非 Mesh 不命中、重复调用幂等。
# ============================================================
from __future__ import annotations

import pytest
from pydantic import ValidationError

from server.agent_core import mesh as mesh_module


@pytest.fixture()
def mesh_env(tmp_path, monkeypatch):
    store = mesh_module.MeshStore(str(tmp_path / "agent_mesh.json"))
    monkeypatch.setattr(mesh_module, "_MESH", store)
    monkeypatch.setattr(mesh_module, "_DISPATCH_THREADS", {})
    return store


def _approval_goal(store: mesh_module.MeshStore):
    goal = store.create_goal(
        "approval resume",
        template=None,
        tasks=[
            {
                "key": "analyze",
                "title": "analyze",
                "intent": "data.analyze",
                "role": "data-analyst",
                "dependsOn": [],
            },
            {
                "key": "readiness",
                "title": "readiness",
                "intent": "readiness.query",
                "role": "verifier",
                "dependsOn": ["analyze"],
            },
        ],
    )
    task = goal["tasks"][0]
    task["status"] = "AWAITING_APPROVAL"
    task["confirmationIds"] = ["confirm-approve"]
    goal["status"] = "ATTENTION"
    store.save()
    return goal, task


def _join_dispatch(goal_id: str) -> None:
    thread = mesh_module._DISPATCH_THREADS[goal_id]
    thread.join(timeout=3)
    assert not thread.is_alive(), "mesh dispatch thread did not finish"


def test_approved_confirmation_resumes_dependent_tasks(mesh_env, monkeypatch):
    calls: list[str] = []

    def fake_execute(session_id, intent, params, agent, **kwargs):
        calls.append(intent)
        return f"{intent} completed", []

    monkeypatch.setattr(mesh_module, "_execute_intent", fake_execute)
    goal, task = _approval_goal(mesh_env)

    first = mesh_module.resume_goal_after_confirmation(
        "confirm-approve", approve=True, actor="alice",
    )
    assert first["matched"] is True
    assert first["resumed"] is True
    assert first["started"] is True
    assert first["alreadyRunning"] is False
    assert task["status"] == "DONE"
    assert task["confirmationIds"] == []
    assert task["finishedAt"]
    assert "人工已批准 confirm-approve by alice" in task["result"]

    _join_dispatch(goal["id"])
    assert calls == ["readiness.query"]
    assert goal["tasks"][1]["status"] == "DONE"
    assert goal["status"] == "ACHIEVED"
    assert any(
        message.get("kind") == "RESPONSE"
        and "confirm:confirm-approve" in (message.get("evidenceRefs") or [])
        for message in mesh_env.data["messages"]
    )

    first_thread = mesh_module._DISPATCH_THREADS[goal["id"]]
    repeated = mesh_module.resume_goal_after_confirmation(
        "confirm-approve", approve=True, actor="alice",
    )
    assert repeated["matched"] is True
    assert repeated["resolved"] is True
    assert repeated["resumed"] is False
    assert repeated["reason"] == "ALREADY_RESOLVED"
    assert mesh_module._DISPATCH_THREADS[goal["id"]] is first_thread
    assert calls == ["readiness.query"]


def test_rejected_confirmation_stops_dispatch(mesh_env, monkeypatch):
    calls: list[str] = []

    def fake_execute(session_id, intent, params, agent, **kwargs):
        calls.append(intent)
        return f"{intent} completed", []

    monkeypatch.setattr(mesh_module, "_execute_intent", fake_execute)
    goal, task = _approval_goal(mesh_env)

    result = mesh_module.resume_goal_after_confirmation(
        "confirm-approve", approve=False, actor="bob",
    )
    assert result["matched"] is True
    assert result["resumed"] is False
    assert task["status"] == "FAILED"
    assert task["confirmationIds"] == []
    assert "人工驳回 confirm-approve by bob" in task["error"]
    assert goal["tasks"][1]["status"] == "PENDING"
    assert goal["status"] == "ATTENTION"
    assert mesh_module._DISPATCH_THREADS == {}
    assert calls == []
    assert any(
        message.get("kind") == "ALERT"
        and "confirm:confirm-approve" in (message.get("evidenceRefs") or [])
        for message in mesh_env.data["messages"]
    )


def test_unrelated_confirmation_is_ignored(mesh_env):
    result = mesh_module.resume_goal_after_confirmation(
        "normal-p2-confirm", approve=True, actor="alice",
    )
    assert result == {"matched": False, "resolved": False, "resumed": False}
    assert mesh_module._DISPATCH_THREADS == {}


def test_sop_stage_request_rejects_query_only_payload():
    from server.gateway.app import SopStageRequest

    with pytest.raises(ValidationError):
        SopStageRequest(sessionId="session-sop", query="换线")

    structured = SopStageRequest(sessionId="session-sop", assetId="sop-asset-1")
    assert structured.assetId == "sop-asset-1"
    assert structured.pack is None