"""B01 real factory contracts. Run with --noconftest in an isolated env."""

import asyncio
import importlib
import json
import os

import pytest


@pytest.fixture
def embedding(tmp_path, monkeypatch):
    assert os.environ.get("PYTHON_DOTENV_DISABLED") == "1"
    assert "unified-execution" in os.environ["EVOSCIENTIST_HOME"]
    from EvoScientist.config.settings import EvoScientistConfig
    from EvoScientist.llm.contracts import AgentModelSet, WebHostContext
    from EvoScientist.workspace_files import ScopedFilesystemBackend
    from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel

    runtime = importlib.import_module("EvoScientist.EvoScientist")
    web = importlib.import_module("EvoScientist.web_runtime")
    monkeypatch.setattr(runtime, "_load_mcp_tools_cached", lambda **kw: {})
    monkeypatch.setattr(runtime, "_load_mcp_config_once", lambda: ("b01-empty", {}))

    class Model(FakeMessagesListChatModel):
        def bind_tools(self, tools, **kwargs):
            return self

    def build(saver, tenant, responses, backend=None):
        root = tmp_path / tenant
        root.mkdir(exist_ok=True)
        memory = tmp_path / (tenant + "-memory")
        memory.mkdir(exist_ok=True)
        backend = backend or ScopedFilesystemBackend(root)
        model = Model(responses=responses)
        cfg = EvoScientistConfig(
            openai_api_key="b01-not-a-real-key-" + tenant,
            enable_async_subagents=False, enable_scheduler=False,
            memory_workers_enabled=False, auto_approve=False,
        )
        _, revision = web.web_tool_registry_manifest()
        graph = web.create_web_agent(
            snapshot=None,
            host=WebHostContext(
                workspace_dir=str(root), memory_dir=str(memory),
                workspace_backend=backend, checkpointer=saver,
                tool_registry_revision=revision, tool_selector_threshold=10000,
            ),
            model_set=AgentModelSet(model, model, model), config=cfg,
        )
        return graph, root, backend

    return build, runtime


def _run_config(tenant):
    # Host-owned identity: SQLite does not isolate by user_id automatically.
    identity = json.dumps([tenant, "same-conversation"], separators=(",", ":"))
    return {"configurable": {"thread_id": identity, "ai4sci_run_id": tenant}}


def _call(name, args, ident):
    from langchain_core.messages import AIMessage
    return AIMessage(content="", tool_calls=[
        {"name": name, "args": args, "id": ident, "type": "tool_call"}
    ])


def test_explicit_web_factory_preserves_environment_and_globals(embedding, tmp_path):
    from EvoScientist import paths
    from langchain_core.messages import AIMessage
    from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

    build, runtime = embedding
    before_env = dict(os.environ)
    names = ("_config", "_chat_model", "_chat_model_key", "_EvoScientist_agent")
    before_globals = {name: vars(runtime).get(name) for name in names}
    before_paths = (paths.WORKSPACE_ROOT, paths._active_workspace)

    async def scenario():
        async with AsyncSqliteSaver.from_conn_string(str(tmp_path / "pure.sqlite")) as saver:
            await saver.setup()
            for tenant in ("alice", "bob"):
                graph, _, _ = build(saver, tenant, [AIMessage(content="ok")])
                assert hasattr(graph, "ainvoke")
        changed = {key for key in before_env.keys() | os.environ.keys()
                   if before_env.get(key) != os.environ.get(key)}
        assert not changed, f"factory changed environment keys: {sorted(changed)}"
        assert all(vars(runtime).get(n) is value for n, value in before_globals.items())
        assert (paths.WORKSPACE_ROOT, paths._active_workspace) == before_paths

    asyncio.run(scenario())


def test_two_tenants_real_file_tools_and_sqlite_reopen(embedding, tmp_path):
    from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
    from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

    build, _ = embedding
    configs = {tenant: _run_config(tenant) for tenant in ("alice", "bob")}
    assert configs["alice"]["configurable"]["thread_id"] != configs["bob"]["configurable"]["thread_id"]

    async def scenario():
        db = str(tmp_path / "persistent.sqlite")
        async with AsyncSqliteSaver.from_conn_string(db) as saver:
            await saver.setup()
            graphs = {}
            for tenant in configs:
                responses = [
                    _call("write_file", {"file_path": "/workspace/probe.txt", "content": tenant + "_ONLY"}, "write"),
                    _call("read_file", {"file_path": "/workspace/probe.txt"}, "read"),
                    _call("read_file", {"file_path": "/workspace/../bob/probe.txt"}, "escape"),
                    AIMessage(content="done"),
                ]
                graphs[tenant], _, _ = build(saver, tenant, responses)
            results = await asyncio.gather(*[
                graphs[t].ainvoke({"messages": [HumanMessage(content="file probe")]}, configs[t])
                for t in configs
            ])
            for tenant, result in zip(configs, results):
                tools = {m.tool_call_id: m for m in result["messages"] if isinstance(m, ToolMessage)}
                assert (tmp_path / tenant / "probe.txt").read_text() == tenant + "_ONLY"
                assert tenant + "_ONLY" in str(tools["read"].content)
                assert "_ONLY" not in str(tools["escape"].content)
                assert "error" in str(tools["escape"].content).lower()
                other = "bob" if tenant == "alice" else "alice"
                assert other + "_ONLY" not in str(result["messages"])
        assert (tmp_path / "persistent.sqlite").stat().st_size > 0
        async with AsyncSqliteSaver.from_conn_string(db) as reopened:
            await reopened.setup()
            for tenant in configs:
                graph, _, backend = build(reopened, tenant, [AIMessage(content="restored")])
                state = await graph.aget_state(configs[tenant])
                assert tenant + "_ONLY" in str(state.values["messages"])
                other = "bob" if tenant == "alice" else "alice"
                assert other + "_ONLY" not in str(state.values["messages"])
                history = [item async for item in graph.aget_state_history(configs[tenant])]
                assert len(history) > 1
                assert all(other + "_ONLY" not in str(s.values) for s in history)
                assert backend.write("/workspace/../outside.txt", "forbidden").error
                assert not (tmp_path / "outside.txt").exists()

    asyncio.run(scenario())


def test_web_manual_execute_interrupts_before_backend(embedding, tmp_path):
    from deepagents.backends.protocol import SandboxBackendProtocol, ExecuteResponse
    from EvoScientist.workspace_files import ScopedFilesystemBackend
    from langchain_core.messages import AIMessage, HumanMessage
    from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
    from langgraph.types import Command

    calls = []

    class ProbeBackend(ScopedFilesystemBackend, SandboxBackendProtocol):
        @property
        def id(self):
            return "b01-no-shell"

        def execute(self, command, *, timeout=None):
            # No shell or process is ever launched by this test capability.
            assert command == "b01-probe"
            calls.append(command)
            return ExecuteResponse(output="probe-ok", exit_code=0)

    build, _ = embedding

    async def scenario():
        async with AsyncSqliteSaver.from_conn_string(str(tmp_path / "review.sqlite")) as saver:
            await saver.setup()
            graph, _, _ = build(saver, "alice", [
                _call("execute", {"command": "b01-probe"}, "execute"), AIMessage(content="done")
            ], ProbeBackend(tmp_path / "alice"))
            config = _run_config("alice")
            result = await graph.ainvoke({"messages": [HumanMessage(content="probe")]}, config)
            assert result.get("__interrupt__")
            assert calls == []
            state = await graph.aget_state(config)
            assert state.values["_verified_review_mode"]["mode"] == "manual"
            await graph.ainvoke(Command(resume={"decisions": [{"type": "approve"}]}), config)
            assert calls == ["b01-probe"]

    asyncio.run(scenario())