"""P0-1 acceptance tests for the harness approved-tool executor.

These exercise ``deerflow.tools.approved_tool_executor.execute_approved_tool``
directly on the real LangGraph ToolNode / ToolRuntime / middleware path with
controlled tools (no LLM, no DeerFlow runtime).
"""

from __future__ import annotations

from typing import Any

import pytest
from langchain_core.tools import StructuredTool
from langgraph.prebuilt import ToolRuntime

from deerflow.agents.middlewares.tool_error_handling_middleware import (
    build_lead_runtime_middlewares,
)
from deerflow.agents.middlewares.write_guard_middleware import (
    WriteGuardMiddleware,
)
from deerflow.config.write_guard_config import WriteGuardConfig
from deerflow.tools import approved_tool_executor
from deerflow.tools.approved_tool_executor import execute_approved_tool

# ?? fakes ?????????????????????????????????????????????????????????????????


class _FakeCheckpointTuple:
    def __init__(self, channel_values: dict[str, Any]) -> None:
        self.checkpoint = {"channel_values": channel_values}


class _FakeCheckpointer:
    def __init__(self, channel_values: dict[str, Any]) -> None:
        self._cv = channel_values

    async def aget_tuple(self, config: dict[str, Any]) -> _FakeCheckpointTuple:
        return _FakeCheckpointTuple(self._cv)


CHANNEL_VALUES: dict[str, Any] = {
    "thread_data": {"marker": "THREAD_MARKER_42"},
    "sandbox": {"sandbox_id": "local"},
    "messages": [],
}


@pytest.fixture
def app_config():
    from deerflow.config import get_app_config

    # The production config mounts WriteGuard.  These low-level executor tests
    # that do not register a MAF service intentionally exercise the harness
    # path without the app-layer validator; the dedicated enabled tests below
    # provide the real bridge/service wiring.
    return get_app_config().model_copy(update={"write_guard": WriteGuardConfig(enabled=False)})


def _snapshot(action_tool: str) -> dict[str, Any]:
    meta = {"name": action_tool, "tool_name": action_tool, "side_effect": "write", "risk_level": "R3"}
    return {
        "version": 1,
        "agents": [
            {
                "agent_id": "a1",
                "allowed_tool_names": [action_tool],
                "excluded_tools": [],
                "tools": [meta],
            }
        ],
        "tools": [meta],
    }


def _ctx(action_tool: str = "sync_write") -> dict[str, Any]:
    """Common trusted execution context + matching action record."""
    return {
        "thread_id": "thread-1",
        "user_id": "u1",
        "session_id": "s1",
        "run_id": "r1",
        "agent_id": "a1",
        "turn_index": 2,
        "model_name": None,
        "excluded_tools": [],
        "capability_snapshot": _snapshot(action_tool),
        "write_policy": "confirm_high_risk",
        "approved_action_id": "action-1",
        "action_record": {
            "id": "action-1",
            "status": "running",
            "run_id": "r1",
            "session_id": "s1",
            "agent_id": "a1",
            "tool_name": action_tool,
            "turn_index": 2,
        },
    }


# ?? 1. sync StructuredTool.func runs once ??succeeded ?????????????????????


@pytest.mark.asyncio
async def test_sync_tool_runs_once_succeeded(app_config, monkeypatch):
    calls = {"n": 0}

    def _fn(payload: str) -> str:
        calls["n"] += 1
        return f"sync-ok:{payload}"

    tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("sync_write")
    result = await execute_approved_tool(
        app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
        tool_name="sync_write", params={"payload": "X"}, **ctx,
    )

    assert result["status"] == "succeeded", result
    assert calls["n"] == 1, "sync tool must run exactly once"


# ?? 2. async tool runs normally ???????????????????????????????????????????


@pytest.mark.asyncio
async def test_async_tool_runs(app_config, monkeypatch):
    calls = {"n": 0}

    async def _fn(payload: str) -> str:
        calls["n"] += 1
        return f"async-ok:{payload}"

    tool = StructuredTool.from_function(coroutine=_fn, name="async_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("async_write")
    result = await execute_approved_tool(
        app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
        tool_name="async_write", params={"payload": "Y"}, **ctx,
    )

    assert result["status"] == "succeeded", result
    assert calls["n"] == 1


# ?? 3. tool needing ToolRuntime gets the REAL runtime + checkpoint state ??


@pytest.mark.asyncio
async def test_tool_receives_real_tool_runtime_and_state(app_config, monkeypatch):
    seen: dict[str, Any] = {}

    def _fn(payload: str, runtime: ToolRuntime[dict[str, Any], dict[str, Any]]) -> str:
        # Typed ToolRuntime context avoids PydanticSerializationUnexpectedValue
        # (bare ToolRuntime defaults context to None).
        seen["runtime_type"] = type(runtime).__name__
        seen["is_real_toolruntime"] = isinstance(runtime, ToolRuntime)
        seen["marker"] = runtime.state.get("thread_data", {}).get("marker")
        seen["context_run_id"] = runtime.context.get("maf_run_id")
        seen["approved_tool"] = runtime.context.get("maf_approved_tool")
        seen["context_action_id"] = runtime.context.get("maf_approved_action_id")
        return "rt-ok"

    tool = StructuredTool.from_function(func=_fn, name="rt_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("rt_write")
    result = await execute_approved_tool(
        app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
        tool_name="rt_write", params={"payload": "Z"}, **ctx,
    )

    assert result["status"] == "succeeded", result
    # A real ToolRuntime, not a duck-typed SimpleNamespace.
    assert seen["is_real_toolruntime"] is True
    assert seen["runtime_type"] != "SimpleNamespace"
    # Real checkpoint state flowed through (sandbox / thread_data recovered).
    assert seen["marker"] == "THREAD_MARKER_42"
    # Trusted MAF context injected by the server (via Runtime on __pregel_runtime).
    assert seen["context_run_id"] == "r1"
    assert seen["approved_tool"] == "rt_write"
    assert seen["context_action_id"] == "action-1"


# ?? 4. unauthorized / disabled / nonexistent ??0 executions ???????????????


@pytest.mark.asyncio
async def test_nonexistent_tool_executes_zero_times(app_config, monkeypatch):
    calls = {"n": 0}

    def _fn(payload: str) -> str:
        calls["n"] += 1
        return "ok"

    tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("ghost_tool")  # action says ghost_tool; not in the allow-list
    result = await execute_approved_tool(
        app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
        tool_name="ghost_tool", params={"payload": "X"}, **ctx,
    )

    assert result["status"] == "failed"
    assert calls["n"] == 0


@pytest.mark.asyncio
async def test_disabled_tool_executes_zero_times(app_config, monkeypatch):
    calls = {"n": 0}

    def _fn(payload: str) -> str:
        calls["n"] += 1
        return "ok"

    tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("sync_write")
    ctx["capability_snapshot"]["agents"][0]["excluded_tools"] = ["sync_write"]
    result = await execute_approved_tool(
        app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
        tool_name="sync_write", params={"payload": "X"}, **ctx,
    )

    assert result["status"] == "failed"
    assert calls["n"] == 0


@pytest.mark.asyncio
async def test_context_mismatch_executes_zero_times(app_config, monkeypatch):
    calls = {"n": 0}

    def _fn(payload: str) -> str:
        calls["n"] += 1
        return "ok"

    tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("sync_write")
    # The stored action belongs to a DIFFERENT run ??must be refused.
    ctx["action_record"] = {
        "id": "action-1",
        "status": "running",
        "run_id": "OTHER_RUN",
        "session_id": "s1",
        "agent_id": "a1",
        "tool_name": "sync_write",
        "turn_index": 2,
    }
    result = await execute_approved_tool(
        app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
        tool_name="sync_write", params={"payload": "X"}, **ctx,
    )

    assert result["status"] == "failed"
    assert calls["n"] == 0


# ?? 5. LLM is never involved ??????????????????????????????????????????????


@pytest.mark.asyncio
async def test_llm_never_involved(app_config, monkeypatch):
    llm_calls = {"n": 0}

    def _boom(*args: Any, **kwargs: Any) -> Any:
        llm_calls["n"] += 1
        raise AssertionError("model factory must never be called by the approved executor")

    # Guard every plausible model-construction entry point.
    import deerflow.models.factory as factory

    monkeypatch.setattr(factory, "create_chat_model", _boom)

    calls = {"n": 0}

    def _fn(payload: str) -> str:
        calls["n"] += 1
        return "ok"

    tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("sync_write")
    result = await execute_approved_tool(
        app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
        tool_name="sync_write", params={"payload": "X"}, **ctx,
    )

    assert result["status"] == "succeeded", result
    assert calls["n"] == 1
    assert llm_calls["n"] == 0, "the LLM must never be involved"


# ?? 6. inconclusive external errors ??unknown (no auto-retry) ???????????????


@pytest.mark.asyncio
async def test_timeout_error_records_unknown(app_config, monkeypatch):
    """P1c: a timeout (external result unknowable) is recorded ``unknown`` ??    never ``failed`` and never auto-retried."""

    async def _fn(payload: str) -> str:
        raise TimeoutError("upstream timed out")

    tool = StructuredTool.from_function(coroutine=_fn, name="timeout_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("timeout_write")
    result = await execute_approved_tool(
        app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
        tool_name="timeout_write", params={"payload": "X"}, **ctx,
    )

    assert result["status"] == "unknown", result


@pytest.mark.asyncio
async def test_connection_error_records_unknown(app_config, monkeypatch):
    """P1c: a dropped/reset connection is likewise inconclusive ??``unknown``."""

    async def _fn(payload: str) -> str:
        raise ConnectionResetError("connection reset by peer")

    tool = StructuredTool.from_function(coroutine=_fn, name="conn_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("conn_write")
    result = await execute_approved_tool(
        app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
        tool_name="conn_write", params={"payload": "X"}, **ctx,
    )

    assert result["status"] == "unknown", result


@pytest.mark.asyncio
async def test_ordinary_error_records_failed(app_config, monkeypatch):
    """P1c: a deterministic validation error (external write definitely did NOT
    happen) is ``failed``, not ``unknown``."""

    async def _fn(payload: str) -> str:
        raise ValueError("invalid payload value")

    tool = StructuredTool.from_function(coroutine=_fn, name="bad_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("bad_write")
    result = await execute_approved_tool(
        app_config=app_config, checkpointer=_FakeCheckpointer(CHANNEL_VALUES), store=None,
        tool_name="bad_write", params={"payload": "X"}, **ctx,
    )

    assert result["status"] == "failed", result
    assert result.get("error") == "tool_execution_failed"


@pytest.mark.asyncio
async def test_tool_error_never_leaks_secrets(app_config, monkeypatch, caplog):
    secret = "TOP_SECRET_123"

    def _fn(payload: str) -> str:
        raise ValueError(f"token={secret}")

    tool = StructuredTool.from_function(func=_fn, name="bad_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("bad_write")
    with caplog.at_level("ERROR"):
        result = await execute_approved_tool(
            app_config=app_config,
            checkpointer=_FakeCheckpointer(CHANNEL_VALUES),
            store=None,
            tool_name="bad_write",
            params={"payload": "X"},
            **ctx,
        )

    assert result["status"] == "failed"
    assert secret not in str(result)
    assert secret not in caplog.text


@pytest.mark.asyncio
async def test_frozen_allow_list_blocks_extra_registry_tool(app_config, monkeypatch):
    calls = {"n": 0}

    def _allowed(payload: str) -> str:
        calls["n"] += 1
        return "ok"

    def _blocked(payload: str) -> str:
        calls["n"] += 1
        return "no"

    allowed = StructuredTool.from_function(func=_allowed, name="tool_a", description="d")
    blocked = StructuredTool.from_function(func=_blocked, name="tool_b", description="d")
    monkeypatch.setattr(
        approved_tool_executor, "get_available_tools", lambda **kw: [allowed, blocked]
    )

    snap = _snapshot("tool_a")
    snap["agents"][0]["allowed_tool_names"] = ["tool_a"]
    ctx = _ctx("tool_b")
    ctx["capability_snapshot"] = snap
    ctx["action_record"]["tool_name"] = "tool_b"
    ctx["approved_action_id"] = "action-1"

    result = await execute_approved_tool(
        app_config=app_config,
        checkpointer=_FakeCheckpointer(CHANNEL_VALUES),
        store=None,
        tool_name="tool_b",
        params={"payload": "X"},
        **ctx,
    )

    assert result["status"] == "failed"
    assert calls["n"] == 0


@pytest.mark.asyncio
async def test_action_not_running_executes_zero_times(app_config, monkeypatch):
    calls = {"n": 0}

    def _fn(payload: str) -> str:
        calls["n"] += 1
        return "ok"

    tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("sync_write")
    ctx["action_record"]["status"] = "approved"
    result = await execute_approved_tool(
        app_config=app_config,
        checkpointer=_FakeCheckpointer(CHANNEL_VALUES),
        store=None,
        tool_name="sync_write",
        params={"payload": "X"},
        **ctx,
    )

    assert result["status"] == "failed"
    assert calls["n"] == 0


def _enabled_write_guard_app_config(app_config):
    enabled_guard = WriteGuardConfig.model_validate(
        {
            "enabled": True,
            "provider": {
                "policy_use": "app.maf_roundtable.write_guard_bridge:make_policy",
                "interceptor_use": "app.maf_roundtable.write_guard_bridge:make_interceptor",
                "config": {},
            },
        }
    )
    return app_config.model_copy(update={"write_guard": enabled_guard})


async def _writeguard_store_fixture(tmp_path, monkeypatch):
    """Create a real MAF store + service with WriteGuard bridge registered.

    Uses monkeypatch so ``write_guard_bridge._maf_service`` is restored when the
    test ends (avoids leaking a disposed sqlite service into later tests).
    """
    from maf_test_helpers import make_maf_service
    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

    import deerflow.persistence.maf_roundtable.model  # noqa: F401
    from app.maf_roundtable import write_guard_bridge
    from app.maf_roundtable.write_guard_bridge import register_write_guard_service
    from deerflow.persistence.base import Base
    from deerflow.persistence.maf_roundtable import make_maf_roundtable_store

    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'wg.db'}")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    store = make_maf_roundtable_store(sf)
    svc = make_maf_service(store)
    # Capture the pre-test service *before* registration.  Registering first
    # would make monkeypatch restore this disposable sqlite-backed service at
    # teardown and leak it into the next test.
    monkeypatch.setattr(write_guard_bridge, "_maf_service", None)
    register_write_guard_service(svc)
    return engine, store, svc


@pytest.mark.asyncio
async def test_approved_tool_executor_with_writeguard_enabled(app_config, monkeypatch, tmp_path):
    """P1: real middleware chain with WriteGuard enabled passes approved execution."""
    from app.maf_roundtable.write_guard_bridge import MafWriteGuardInterceptor

    enabled_app_config = _enabled_write_guard_app_config(app_config)
    middlewares = build_lead_runtime_middlewares(app_config=enabled_app_config)
    assert any(isinstance(item, WriteGuardMiddleware) for item in middlewares)

    engine, store, svc = await _writeguard_store_fixture(tmp_path, monkeypatch)

    validate_spy_calls = {"n": 0}
    original_validate = MafWriteGuardInterceptor.validate_approved_action

    async def _spy_validate(self, *, tool_name: str, run_context: dict) -> bool:
        validate_spy_calls["n"] += 1
        return await original_validate(self, tool_name=tool_name, run_context=run_context)

    monkeypatch.setattr(MafWriteGuardInterceptor, "validate_approved_action", _spy_validate)

    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")
    action = await store.actions.create_action(
        {
            "session_id": session["id"],
            "run_id": run["id"],
            "agent_id": "a1",
            "tool_name": "sync_write",
            "idempotency_key": "wg-test-1",
            "risk_level": "R3",
            "status": "running",
            "turn_index": 2,
        }
    )

    calls = {"n": 0}

    def _fn(payload: str) -> str:
        calls["n"] += 1
        return "wg-ok"

    tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    ctx = _ctx("sync_write")
    ctx["session_id"] = session["id"]
    ctx["run_id"] = run["id"]
    ctx["approved_action_id"] = action["id"]
    ctx["action_record"] = {
        "id": action["id"],
        "status": "running",
        "run_id": run["id"],
        "session_id": session["id"],
        "agent_id": "a1",
        "tool_name": "sync_write",
        "turn_index": 2,
    }
    try:
        result = await execute_approved_tool(
            app_config=enabled_app_config,
            checkpointer=_FakeCheckpointer(CHANNEL_VALUES),
            store=None,
            tool_name="sync_write",
            params={"payload": "X"},
            **ctx,
        )
    finally:
        await engine.dispose()

    assert result["status"] == "succeeded", result
    assert calls["n"] == 1
    assert validate_spy_calls["n"] == 1
    assert result["result"] == {"executed": True}


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "db_mismatch",
    [
        "status",
        "run_id",
        "session_id",
        "agent_id",
        "tool_name",
    ],
)
async def test_writeguard_rejects_db_mismatched_action(
    app_config, monkeypatch, tmp_path, db_mismatch
):
    """Second-line defense: Guard re-reads the DB action and rejects mismatches.

    ``action_record`` passed to the executor stays consistent so the first-layer
    field check passes; only the DB row is wrong.
    """
    from app.maf_roundtable.write_guard_bridge import MafWriteGuardInterceptor

    enabled_app_config = _enabled_write_guard_app_config(app_config)
    middlewares = build_lead_runtime_middlewares(app_config=enabled_app_config)
    assert any(isinstance(item, WriteGuardMiddleware) for item in middlewares)

    engine, store, svc = await _writeguard_store_fixture(tmp_path, monkeypatch)

    validate_spy_calls = {"n": 0}
    original_validate = MafWriteGuardInterceptor.validate_approved_action

    async def _spy_validate(self, *, tool_name: str, run_context: dict) -> bool:
        validate_spy_calls["n"] += 1
        return await original_validate(self, tool_name=tool_name, run_context=run_context)

    monkeypatch.setattr(MafWriteGuardInterceptor, "validate_approved_action", _spy_validate)

    session = await svc.create_session(user_id="u1", body={"task_text": "t"})
    other = await svc.create_session(user_id="u1", body={"task_text": "other"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")

    db_fields = {
        "session_id": session["id"],
        "run_id": run["id"],
        "agent_id": "a1",
        "tool_name": "sync_write",
        "idempotency_key": f"wg-mismatch-{db_mismatch}",
        "risk_level": "R3",
        "status": "running",
        "turn_index": 2,
    }
    if db_mismatch == "status":
        db_fields["status"] = "approved"
    elif db_mismatch == "run_id":
        # No second create_run (session allows only one active run) ? fake id is enough.
        db_fields["run_id"] = "other-run-id"
    elif db_mismatch == "session_id":
        db_fields["session_id"] = other["id"]
    elif db_mismatch == "agent_id":
        db_fields["agent_id"] = "other-agent"
    elif db_mismatch == "tool_name":
        db_fields["tool_name"] = "other_tool"

    action = await store.actions.create_action(db_fields)

    tool_calls = {"n": 0}

    def _fn(payload: str) -> str:
        tool_calls["n"] += 1
        return "should-not-run"

    tool = StructuredTool.from_function(func=_fn, name="sync_write", description="d")
    monkeypatch.setattr(approved_tool_executor, "get_available_tools", lambda **kw: [tool])

    # Executor-facing record looks correct ? Guard must catch the DB mismatch.
    ctx = _ctx("sync_write")
    ctx["session_id"] = session["id"]
    ctx["run_id"] = run["id"]
    ctx["approved_action_id"] = action["id"]
    ctx["action_record"] = {
        "id": action["id"],
        "status": "running",
        "run_id": run["id"],
        "session_id": session["id"],
        "agent_id": "a1",
        "tool_name": "sync_write",
        "turn_index": 2,
    }
    try:
        result = await execute_approved_tool(
            app_config=enabled_app_config,
            checkpointer=_FakeCheckpointer(CHANNEL_VALUES),
            store=None,
            tool_name="sync_write",
            params={"payload": "X"},
            **ctx,
        )
    finally:
        await engine.dispose()

    assert tool_calls["n"] == 0
    assert validate_spy_calls["n"] == 1
    assert result["status"] == "failed"