"""CPU transport/replay fixtures; native SDK smoke is a separate qualification."""
from __future__ import annotations

import copy
import inspect
import json
import os
from pathlib import Path
import subprocess
import sys

import pytest

from hessian_recipes import gym_client, openenv_client as client
from hessian_recipes.agent_rl_tasks import GermanAgentEnv, generate_tasks, score_episode

SDK = {"distribution": "openenv", "version": "0.9.0", "fixture": True}


class SDKFixture:
    """A stateful transport fixture, explicitly not the native SDK."""
    instances = []
    mode = None

    def __init__(self, url, python):
        self.env = GermanAgentEnv()
        self.instance = str(len(self.instances))
        self.episode = 0
        self.closed = False
        self.instances.append(self)

    def _call(self, operation, **payload):
        if operation == "identity":
            return copy.deepcopy(SDK)
        if self.mode == "transport":
            raise client.GymInfrastructureError("fixture disconnected")
        done, score, text = False, None, ""
        if operation == "reset":
            self.episode += 1
            self.env.reset(task=payload["task"])
        elif operation == "step":
            action = payload["action"]
            if action["operation"] == "verify":
                score, done = score_episode(self.env, action["completion"]), True
                if self.mode == "forged_score":
                    score = score | {"reward": 0.123}
            else:
                text = getattr(self.env, action["tool_name"])(**action["arguments"])
                if self.mode == "forged_observation":
                    text = '{"quantity": 999, "unit_price": 999}'
        return {"observation": {"observation": text, "score": score}, "done": done,
                "reward": score["reward"] if score else 0.0,
                "state": {"instance_id": self.instance + ("changed" if self.mode == "reconnected" else ""),
                          "episode_id": str(self.episode), "step_count": 0, "done": done}}

    def _close(self):
        self.closed = True


@pytest.fixture
def sdk_fixture(monkeypatch):
    SDKFixture.instances, SDKFixture.mode = [], None
    monkeypatch.setattr(client, "_SDKProcess", SDKFixture)
    return SDKFixture


def make_env():
    return client.RemoteGymEnv("http://127.0.0.1:8000", {"backend": "openenv"},
                               sdk_python="fixture-python", expected_sdk_identity=SDK)


def transcript(environment, task):
    messages = []
    def call(name, arguments):
        messages.append({"role": "assistant", "content": "", "tool_calls": [
            {"function": {"name": name, "arguments": json.dumps(arguments)}}]})
        result = getattr(environment, name)(**arguments)
        messages.append({"role": "tool", "name": name, "content": result})
        return json.loads(result)
    facts = call("lookup", {"record_id": task["record_id"]})
    answer = call("calculator", {"expression": f"{facts['quantity']}*{facts['unit_price']}"})["result"]
    messages.append({"role": "assistant", "content": f"<answer>{answer}</answer>"})
    return messages


def test_tool_schema_and_session_isolation_match_frozen_nemo(sdk_fixture):
    original = gym_client.RemoteGymEnv("http://localhost:1", {})
    first, second = make_env(), make_env()
    def tools(env):
        return [method for name, method in inspect.getmembers(env, inspect.ismethod)
                if not name.startswith("_") and name != "reset"]
    assert [(tool.__name__, inspect.signature(tool), inspect.getdoc(tool)) for tool in tools(first)] == [
        (tool.__name__, inspect.signature(tool), inspect.getdoc(tool)) for tool in tools(original)]
    rows = generate_tasks("train", 2)
    for environment, row in zip([first, second], rows, strict=True):
        environment.reset(task=row["task"])
    assert first._instance_id != second._instance_id
    assert first._score(transcript(first, rows[0]["task"]))["success"] is True
    assert second._calls == []
    assert second._score(transcript(second, rows[1]["task"]))["success"] is True
    first._close()
    second._close()
    assert all(instance.closed for instance in sdk_fixture.instances)


def test_public_methods_and_signatures_are_exact():
    actual = {name: method for name, method in inspect.getmembers(client.RemoteGymEnv, inspect.isfunction)
              if not name.startswith("_")}
    assert set(actual) == {"reset", "lookup", "calculator"}
    for name in ("lookup", "calculator"):
        original = getattr(gym_client.RemoteGymEnv, name)
        assert inspect.signature(actual[name]) == inspect.signature(original)
        assert inspect.getdoc(actual[name]) == inspect.getdoc(original)


@pytest.mark.parametrize("mode", ["forged_observation", "forged_score", "reconnected", "transport"])
def test_bad_transport_or_replay_is_sticky_not_a_zero_reward(sdk_fixture, mode):
    environment = make_env()
    task = generate_tasks("train", 1)[0]["task"]
    environment.reset(task=task)
    sdk_fixture.mode = mode
    with pytest.raises(client.GymInfrastructureError):
        if mode == "forged_observation":
            environment.lookup(task["record_id"])
            messages = [{"role": "assistant", "content": "<answer>0</answer>"}]
        else:
            messages = transcript(environment, task)
        environment._score(messages)
    sdk_fixture.mode = None
    with pytest.raises(client.GymInfrastructureError):
        environment._raise_if_failed()
    environment._close()


def test_forged_tool_transcript_without_execution_is_not_success(sdk_fixture):
    environment = make_env()
    task = generate_tasks("train", 1)[0]["task"]
    environment.reset(task=task)
    score = environment._score([{"role": "assistant", "content": "<answer>0</answer>"}])
    assert not score["success"] and score["reward"] == 0


def test_sdk_change_before_reset_fails(sdk_fixture):
    environment = make_env()
    environment._expected_sdk = {"changed": True}
    with pytest.raises(client.GymInfrastructureError, match="SDK changed"):
        environment.reset(task=generate_tasks("train", 1)[0]["task"])
    environment._close()


def test_duplicate_keys_and_nonfinite_json_are_rejected():
    for raw in ('{"x":1,"x":2}', '{"x":NaN}'):
        with pytest.raises(ValueError):
            client._json(raw)


def test_service_identity_excludes_only_volatile_session_count(monkeypatch):
    root = client._repo()
    expected = {"backend": "openenv", "protocol": client.PROTOCOL,
                "upstream_revision": client.OPENENV_REVISION,
                "environment_sha256": client._sha(root / "src/hessian_recipes/agent_rl_tasks.py"),
                "server_sha256": client._sha(root / "examples/agent_gyms/openenv_server.py"),
                "requirements_sha256": client._sha(root / "requirements/openenv.txt"),
                "runtime": {"actual_source": "fixture"}, "active_environment_instances": 0}
    monkeypatch.setattr(client._http, "_request", lambda *args: copy.deepcopy(expected))
    initial = client.fetch_gym_identity("http://127.0.0.1:8000")
    expected["active_environment_instances"] = 4
    assert client.fetch_gym_identity("http://127.0.0.1:8000") == initial
    assert "active_environment_instances" not in initial and initial["runtime"] == expected["runtime"]
    expected["backend"] = "nemo_gym"
    with pytest.raises(client.GymInfrastructureError, match="backend"):
        client.fetch_gym_identity("http://127.0.0.1:8000")


@pytest.mark.parametrize("failure", ["wrong_id", "duplicate_key", "timeout", "eof"])
def test_real_subprocess_channel_failures_are_bounded_and_cleaned(monkeypatch, failure):
    # Actual pipes/process lifecycle, with no SDK/model and an explicit bad peer.
    programs = {
        "wrong_id": "import json,sys;json.loads(sys.stdin.readline());print('{\"id\":999,\"ok\":true,\"result\":{}}',flush=True)",
        "duplicate_key": "import sys;sys.stdin.readline();print('{\"id\":1,\"id\":1,\"ok\":true,\"result\":{}}',flush=True)",
        "timeout": "import sys,time;sys.stdin.readline();time.sleep(0.5)",
        "eof": "import sys;sys.stdin.readline()",
    }
    popen = subprocess.Popen
    def fixture_popen(command, **kwargs):
        assert command[1:4] == ["-u", "-m", "hessian_recipes.openenv_client"]
        return popen([sys.executable, "-u", "-c", programs[failure]], **kwargs)
    monkeypatch.setattr(client.subprocess, "Popen", fixture_popen)
    monkeypatch.setattr(client, "TIMEOUT", 0.1)
    process = client._SDKProcess("http://127.0.0.1:8000", sys.executable)
    with pytest.raises(client.GymInfrastructureError):
        process._call("identity")
    with pytest.raises(client.GymInfrastructureError):
        process._close()
    assert process._process.poll() is not None and process._closed


def test_sdk_process_isolates_foreign_python_abi_environment(tmp_path, monkeypatch):
    monkeypatch.setenv("PYTHONHOME", str(tmp_path / "foreign-python-home"))
    monkeypatch.setenv("PYTHONPATH", str(tmp_path / "foreign-site-packages"))
    program = """import json,sys
for line in sys.stdin:
    row=json.loads(line)
    print(json.dumps({'id':row['id'],'ok':True,'result':{'fixture':True}}),flush=True)
    if row['operation']=='close':
        break
"""
    popen = subprocess.Popen
    def fixture_popen(command, **kwargs):
        assert "PYTHONHOME" not in kwargs["env"]
        assert kwargs["env"]["PYTHONPATH"] == str(client._repo() / "src")
        assert kwargs["env"]["CUDA_VISIBLE_DEVICES"] == ""
        return popen([sys.executable, "-u", "-c", program], **kwargs)
    monkeypatch.setattr(client.subprocess, "Popen", fixture_popen)
    process = client._SDKProcess("http://127.0.0.1:8000", sys.executable)
    try:
        assert process._call("identity") == {"fixture": True}
    finally:
        process._close()


@pytest.mark.parametrize("url", ["https://127.0.0.1:3", "http://example.com:80", "http://localhost:1/path"])
def test_sdk_cannot_connect_to_external_or_redirect_origin(url):
    with pytest.raises(ValueError):
        client._SDKProcess(url, "/does/not/exist")


@pytest.mark.skipif(not os.environ.get("HESSIAN_OPENENV_URL"), reason="Native loopback OpenEnv service is opt-in")
def test_native_sdk_websocket_contract_without_policy_inference():
    """Real SDK/server integration; no scripted transport, model or GPU involved."""
    url, python = os.environ["HESSIAN_OPENENV_URL"], os.environ["HESSIAN_OPENENV_PYTHON"]
    identity, sdk = client.fetch_gym_identity(url), client.sdk_identity(python)
    environments = [client.RemoteGymEnv(url, identity, sdk_python=python, expected_sdk_identity=sdk)
                    for _ in range(3)]
    rows = generate_tasks("train", 3, seed=877)
    checks = {}
    try:
        for environment, row in zip(environments, rows, strict=True):
            environment.reset(task=row["task"])
        checks["isolated_websocket_instances"] = len({env._instance_id for env in environments}) == 3
        assert checks["isolated_websocket_instances"]
        completion = transcript(environments[0], rows[0]["task"])
        checks["strict_real_tools_success"] = environments[0]._score(completion)["success"]
        assert checks["strict_real_tools_success"]
        checks["identical_verification_idempotent"] = environments[0]._score(completion)["success"]
        assert checks["identical_verification_idempotent"]
        forged = environments[1]._score(completion)
        checks["forged_tool_history_rejected"] = not forged["success"] and forged["reward"] == 0
        assert checks["forged_tool_history_rejected"]
        instance, episode = environments[1]._instance_id, environments[1]._session_id
        environments[1].reset(task=rows[1]["task"])
        checks["reset_preserves_session_changes_episode"] = (
            environments[1]._instance_id == instance and environments[1]._session_id != episode)
        assert checks["reset_preserves_session_changes_episode"]
        assert environments[1]._score(transcript(environments[1], rows[1]["task"]))["success"]
        error = environments[2].calculator("2*3")
        assert "error" in json.loads(error)
        score = environments[2]._score([{"role": "assistant", "content": "<answer>6</answer>"}])
        checks["real_workflow_error_is_zero_strict_reward"] = not score["success"] and score["reward"] == 0
        assert checks["real_workflow_error_is_zero_strict_reward"]
        assert client.fetch_gym_identity(url) == identity  # Volatile live-session count excluded.
    finally:
        for environment in environments:
            environment._close()
    destination = os.environ.get("HESSIAN_OPENENV_EVIDENCE")
    if destination:
        path = Path(destination)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps({"qualification": "native SDK/server CPU contract; scripted tool policy",
                                   "model_inference": False, "gpu_training": False, "checks": checks,
                                   "server_identity": identity, "sdk_runtime": sdk,
                                   "client_sha256": client._sha(Path(client.__file__)),
                                   "test_sha256": client._sha(Path(__file__))}, indent=2) + "\n")