"""Phase-5 target lease, timeout and ambiguous-result safety tests."""

from __future__ import annotations

import asyncio
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace

import pytest
import pytest_asyncio
from maf_test_helpers import make_maf_service
from sqlalchemy import update
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.gateway.app import (
    _scan_maf_approval_expiry,
    _start_maf_approval_scanner,
    _stop_maf_approval_scanner,
)
from app.maf_roundtable.action_coordinator import (
    ActionCoordinator,
    MafActionSecretKeyError,
    derive_target_key,
    encrypt_params,
)
from app.maf_roundtable.service import MafConflictError
from app.maf_roundtable.tool_executor import ToolExecutor
from app.maf_roundtable.workflow_executors import RoundtableState
from app.maf_roundtable.workflow_runner import WorkflowRunner
from deerflow.config.maf_roundtable_config import MafRoundtableConfig
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
from deerflow.persistence.maf_roundtable.actions import action_to_public_dict
from deerflow.persistence.maf_roundtable.model import (
    Base,
    MafRoundtableActionLeaseRow,
    MafRoundtableApprovalRow,
)

TEST_TOOL = "controlled_write"


def _snapshot() -> dict:
    tool = {
        "name": TEST_TOOL,
        "side_effect": "write",
        "risk_level": "R3",
        "target_key_fields": ["tenant_id", "object_id"],
        "target_key_namespace": "lease-test",
    }
    return {
        "tools": [tool],
        "agents": [{"agent_id": "seat-1", "tools": [tool]}],
    }


@pytest_asyncio.fixture
async def store(tmp_path):
    # A file-backed SQLite database gives concurrent sessions the same
    # transaction/constraint semantics used by deployed SQLite installations.
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_action_leases.db'}", echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    result = make_maf_roundtable_store(sf)
    assert result is not None
    yield result
    await engine.dispose()


@pytest.fixture
def svc(store):
    return make_maf_service(
        store,
        enabled=True,
        config=MafRoundtableConfig(
            enabled=True,
            approval_timeout_seconds=60,
            target_lease_seconds=30,
        ),
    )


async def _run(store, svc) -> tuple[dict, dict]:
    session = await svc.create_session(user_id="u1", body={"task_text": "test"})
    run = await svc.create_run(session_id=session["id"], user_id="u1")
    return session, run


async def _approved_action(
    store, session: dict, run: dict, *, key: str, target_key: str, agent_id: str = "seat-1"
) -> dict:
    return await store.actions.create_action({
        "session_id": session["id"],
        "run_id": run["id"],
        "agent_id": agent_id,
        "tool_name": TEST_TOOL,
        "idempotency_key": key,
        "target_key": target_key,
        "risk_level": "R3",
        "status": "approved",
    })


def test_target_key_is_server_derived_and_opaque(monkeypatch) -> None:
    monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "explicit-hmac-test-secret")
    snapshot = _snapshot()
    original = {
        "tenant_id": "tenant-a",
        "object_id": "task-7",
        "target_key": "browser-forged-value",
        "secret": "do-not-persist",
    }
    same_target_different_noise = {
        "tenant_id": "tenant-a",
        "object_id": "task-7",
        "target_key": "another-forged-value",
        "secret": "changed",
    }
    target = derive_target_key(TEST_TOOL, original, snapshot, "seat-1")
    assert target == derive_target_key(TEST_TOOL, same_target_different_noise, snapshot, "seat-1")
    assert target.startswith("maf-target:v1:")
    assert "tenant-a" not in target
    assert "task-7" not in target
    assert "browser-forged" not in target

    nested = derive_target_key(
        TEST_TOOL,
        {"tenant_id": "tenant-a", "object_id": {"id": "task-7"}},
        snapshot,
        "seat-1",
    )
    nested_other = derive_target_key(
        TEST_TOOL,
        {"tenant_id": "tenant-a", "object_id": ["task-7"]},
        snapshot,
        "seat-1",
    )
    assert nested == nested_other

    conservative_inputs = [
        {"tenant_id": "tenant-a"},
        {"tenant_id": "tenant-a", "object_id": None},
        {"tenant_id": "tenant-a", "object_id": ""},
        {"tenant_id": "tenant-a", "object_id": {"id": "task-7"}},
        {"tenant_id": "tenant-a", "object_id": ["task-7"]},
    ]
    conservative_keys = {
        derive_target_key(TEST_TOOL, args, snapshot, "seat-1")
        for args in conservative_inputs
    }
    assert len(conservative_keys) == 1

    monkeypatch.delenv("MAF_ACTION_SECRET_KEY", raising=False)
    with pytest.raises(MafActionSecretKeyError):
        derive_target_key(TEST_TOOL, original, snapshot, "seat-1")

    monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "another-test-domain-secret")
    assert target != derive_target_key(TEST_TOOL, original, snapshot, "seat-1")


def test_action_public_dto_is_an_explicit_allowlist() -> None:
    internal = {
        "id": "action-1",
        "session_id": "session-1",
        "run_id": "run-1",
        "agent_id": "seat-1",
        "tool_name": TEST_TOOL,
        "logical_action_id": "tool-call-1",
        "turn_index": 1,
        "risk_level": "R3",
        "status": "waiting_approval",
        "arguments": {"safe": "redacted"},
        "result": None,
        "external_object_id": None,
        "approval_id": "approval-1",
        "attempt_count": 0,
        "started_at": None,
        "finished_at": None,
        "created_at": "2026-01-01T00:00:00+00:00",
        "updated_at": "2026-01-01T00:00:00+00:00",
        "target_key": "private-target",
        "idempotency_key": "private-idempotency",
        "internal_canary": "must-not-leak",
    }
    public = action_to_public_dict(internal)
    assert set(public) == {
        "id", "session_id", "run_id", "agent_id", "tool_name",
        "logical_action_id", "turn_index", "risk_level", "status",
        "arguments", "result", "external_object_id", "approval_id",
        "attempt_count", "started_at", "finished_at", "created_at", "updated_at",
    }


@pytest.mark.asyncio
async def test_two_agents_same_target_are_serialized(store, svc):
    """The DB lease, not a process lock, blocks the second approved write."""
    session, run = await _run(store, svc)
    target_key = derive_target_key(
        TEST_TOOL, {"tenant_id": "tenant-a", "object_id": "task-7"}, _snapshot(), "seat-1"
    )
    first = await _approved_action(store, session, run, key="lease-first", target_key=target_key)
    second = await _approved_action(
        store, session, run, key="lease-second", target_key=target_key, agent_id="seat-2"
    )

    first_claim, second_claim = await asyncio.gather(
        store.claim_action_for_execution(first["id"], lease_seconds=30),
        store.claim_action_for_execution(second["id"], lease_seconds=30),
    )
    assert sorted([first_claim, second_claim]) == ["claimed", "target_busy"]
    winner = first if first_claim == "claimed" else second
    waiting = second if winner["id"] == first["id"] else first
    assert (await store.actions.get_action(waiting["id"]))["status"] == "approved"

    assert await store.record_action_terminal_result(
        winner["id"], status="succeeded", result={"safe": True}
    )
    assert await store.claim_action_for_execution(waiting["id"], lease_seconds=30) == "claimed"


@pytest.mark.asyncio
async def test_restart_recovers_expired_lease_as_unknown_without_replay(store, svc):
    session, run = await _run(store, svc)
    target_key = derive_target_key(
        TEST_TOOL, {"tenant_id": "tenant-a", "object_id": "task-7"}, _snapshot(), "seat-1"
    )
    action = await _approved_action(store, session, run, key="lease-restart", target_key=target_key)
    assert await store.claim_action_for_execution(action["id"], lease_seconds=1) == "claimed"

    recovered = await store.recover_expired_action_leases(
        run_id=run["id"], now=datetime.now(UTC) + timedelta(seconds=5)
    )
    assert [item["id"] for item in recovered] == [action["id"]]
    stored = await store.actions.get_action(action["id"])
    assert stored["status"] == "unknown"
    assert stored["result"] == {"error_code": "target_lease_expired_during_execution"}
    assert await store.claim_action_for_execution(action["id"], lease_seconds=30) == "unknown"
    blocked = await _approved_action(
        store,
        session,
        run,
        key="lease-restart-blocked",
        target_key=target_key,
        agent_id="seat-2",
    )
    assert await store.claim_action_for_execution(blocked["id"], lease_seconds=30) == "target_reconciliation_required"


@pytest.mark.asyncio
async def test_approval_timeout_rejects_action_and_writes_audit_event(monkeypatch, store, svc):
    monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "test-only-lease-secret")
    session, run = await _run(store, svc)
    coordinator = ActionCoordinator(svc, user_id="u1")
    actions = await coordinator.intercept_write_tool_calls(
        run_id=run["id"],
        session_id=session["id"],
        agent_id="seat-1",
        turn_index=1,
        tool_calls=[{
            "name": TEST_TOOL,
            "id": "timeout-call",
            "args": {"tenant_id": "tenant-a", "object_id": "task-7"},
        }],
        capability_snapshot=_snapshot(),
        write_policy="confirm_all_writes",
    )
    approval_id = actions[0]["approval_id"]

    expired = await store.expire_due_approvals(
        run_id=run["id"], now=datetime.now(UTC) + timedelta(minutes=2)
    )
    assert expired == [{"id": approval_id, "run_id": run["id"], "action_id": actions[0]["id"]}]
    assert (await store.approvals.get_approval(approval_id))["status"] == "expired"
    assert (await store.actions.get_action(actions[0]["id"]))["status"] == "rejected"
    with pytest.raises(MafConflictError):
        await svc.decide_approval(approval_id=approval_id, decision="approved", user_id="u1")
    events = await store.events.list_events(run["id"])
    assert any(
        event["event_type"] == "action.rejected"
        and event["payload"].get("reason") == "approval_expired"
        for event in events
    )


@pytest.mark.asyncio
async def test_external_success_with_local_result_failure_never_replays(monkeypatch, store, svc):
    """After external success, an unavailable DB result write becomes unknown."""
    monkeypatch.setenv("MAF_ACTION_SECRET_KEY", "test-only-lease-secret")
    session, run = await _run(store, svc)
    target_key = derive_target_key(
        TEST_TOOL, {"tenant_id": "tenant-a", "object_id": "task-7"}, _snapshot(), "seat-1"
    )
    action = await _approved_action(store, session, run, key="persist-failure", target_key=target_key)
    await store.action_secrets.create(
        action_id=action["id"], arguments_encrypted=encrypt_params({"tenant_id": "tenant-a", "object_id": "task-7"})
    )

    calls = 0

    class Client:
        async def execute_tool_directly(self, **_kwargs):
            nonlocal calls
            calls += 1
            return {"status": "succeeded", "result": {"executed": True}}

    adapter = SimpleNamespace(_client=Client())
    coordinator = ActionCoordinator(svc, user_id="u1")

    async def fail_terminal_persistence(*_args, **_kwargs):
        raise RuntimeError("simulated local persistence outage")

    monkeypatch.setattr(coordinator, "record_result", fail_terminal_persistence)
    executor = ToolExecutor(adapter=adapter, action_coordinator=coordinator, user_id="u1")
    first = await executor.execute(
        action_id=action["id"],
        session_id=session["id"],
        run_id=run["id"],
        agent_id="seat-1",
        thread_id="thread-1",
        approved_tool=TEST_TOOL,
        capability_snapshot=_snapshot(),
        write_policy="confirm_all_writes",
    )
    second = await executor.execute(
        action_id=action["id"],
        session_id=session["id"],
        run_id=run["id"],
        agent_id="seat-1",
        thread_id="thread-1",
        approved_tool=TEST_TOOL,
        capability_snapshot=_snapshot(),
        write_policy="confirm_all_writes",
    )

    assert first["status"] == "unknown"
    assert second["status"] == "in_progress"
    assert calls == 1
    recovered = await store.recover_expired_action_leases(
        run_id=run["id"], now=datetime.now(UTC) + timedelta(minutes=20)
    )
    assert [item["id"] for item in recovered] == [action["id"]]
    assert (await store.actions.get_action(action["id"]))["status"] == "unknown"


@pytest.mark.asyncio
async def test_workflow_recovery_stops_on_expired_lease_without_tool_replay(store, svc):
    """The restart entry point converts a stale running lease to waiting_input."""
    session, run = await _run(store, svc)
    target_key = derive_target_key(
        TEST_TOOL, {"tenant_id": "tenant-a", "object_id": "task-7"}, _snapshot(), "seat-1"
    )
    action = await _approved_action(store, session, run, key="runner-recovery", target_key=target_key)
    assert await store.claim_action_for_execution(action["id"], lease_seconds=30) == "claimed"
    async with store.session_factory() as db:
        await db.execute(
            update(MafRoundtableActionLeaseRow)
            .where(MafRoundtableActionLeaseRow.action_id == action["id"])
            .values(lease_expires_at=datetime.now(UTC) - timedelta(seconds=1))
        )
        await db.commit()

    state = RoundtableState(
        session_id=session["id"],
        run_id=run["id"],
        task_text="test",
        mode="structured",
        model_name=None,
        reasoning_effort=None,
        max_rounds=1,
        write_policy="confirm_all_writes",
        manager_agent_id=None,
        seat_agent_ids=[],
        thread_map={},
        capability_snapshot=_snapshot(),
    )
    state.current_round = 1
    state.status = "running"
    state.pending_approvals = [{
        "action_id": action["id"],
        "approval_id": "already-approved",
        "agent_id": "seat-1",
        "tool_name": TEST_TOOL,
    }]
    await store.runs.update_run(run["id"], {"status": "running", "maf_checkpoint": state.to_dict()})
    await store.sessions.update_session(session["id"], user_id="u1", fields={"status": "running"})

    runner = WorkflowRunner(SimpleNamespace(), service=svc, user_id="u1")
    recovered = await runner.recover(run["id"])
    assert recovered is not None and recovered["status"] == "waiting_input"
    stored = await store.actions.get_action(action["id"])
    assert stored["status"] == "unknown"
    assert (await store.sessions.get_session(session["id"], user_id="u1"))["status"] == "waiting_input"


@pytest.mark.asyncio
async def test_approval_expiry_scan_is_repeatable_and_get_is_observational(monkeypatch, store, svc):
    """A scan expires once; ordinary approval GETs never perform that mutation."""

    session, run = await _run(store, svc)
    coordinator = ActionCoordinator(svc, user_id="u1")
    action = (
        await coordinator.intercept_write_tool_calls(
            run_id=run["id"],
            session_id=session["id"],
            agent_id="seat-1",
            turn_index=1,
            tool_calls=[{"name": TEST_TOOL, "id": "expiry-scan", "args": {
                "tenant_id": "tenant-a", "object_id": "task-7", "value": "x"
            }}],
            capability_snapshot=_snapshot(),
            write_policy="confirm_all_writes",
        )
    )[0]

    async with store.session_factory() as db:
        await db.execute(
            update(MafRoundtableApprovalRow)
            .where(MafRoundtableApprovalRow.id == action["approval_id"])
            .values(expires_at=datetime.now(UTC) - timedelta(seconds=1))
        )
        await db.commit()

    # The normal read path must not mutate an expired pending row.
    listed = await svc.list_approvals(run_id=run["id"], user_id="u1")
    assert listed[0]["status"] == "pending"

    class FakeDispatcher:
        def __init__(self):
            self.calls: list[tuple[str, str]] = []

        async def resume(self, run_id: str, *, user_id: str):
            self.calls.append((run_id, user_id))
            return await store.runs.get_run(run_id)

    fake = FakeDispatcher()
    monkeypatch.setattr(
        "app.maf_roundtable.dispatcher.get_dispatcher",
        lambda _app: fake,
    )
    app = SimpleNamespace(
        state=SimpleNamespace(maf_roundtable_service=svc),
    )

    concurrent = await asyncio.gather(
        _scan_maf_approval_expiry(app),
        _scan_maf_approval_expiry(app),
    )
    repeat_after_restart = await _scan_maf_approval_expiry(app)
    assert sorted(concurrent, key=len) == [[], [run["id"]]]
    assert repeat_after_restart == []
    assert fake.calls == [(run["id"], "u1")]
    assert (await store.approvals.get_approval(action["approval_id"]))["status"] == "expired"
    assert (await store.actions.get_action(action["id"]))["status"] == "rejected"


@pytest.mark.asyncio
async def test_maf_approval_scanner_lifecycle_is_feature_gated():
    config = MafRoundtableConfig(enabled=True, approval_scan_interval_seconds=5)
    enabled_app = SimpleNamespace(
        state=SimpleNamespace(
            maf_roundtable_service=SimpleNamespace(enabled=True, config=config),
        )
    )
    task = _start_maf_approval_scanner(enabled_app)
    assert task is not None
    assert enabled_app.state.maf_approval_scanner_task is task
    await _stop_maf_approval_scanner(enabled_app)
    assert task.done()

    disabled_app = SimpleNamespace(
        state=SimpleNamespace(
            maf_roundtable_service=SimpleNamespace(enabled=False, config=config),
        )
    )
    assert _start_maf_approval_scanner(disabled_app) is None
    await _stop_maf_approval_scanner(disabled_app)