"""AppWorld 场景适配器的测试。

分三块：解析器、提示词装配、动作执行器。

解析器那块的前五条是 `tests/contract/test_decision_parser.py` 那份公共契约的逐条复刻。契约
套件本身是给下游接自己的实现用的（在自己的 `conftest.py` 里覆盖 fixture），压测这边不接那
套装配、只把五条断言照着写一遍——它是任何新适配器的准入标准，压测的适配器也是适配器。

执行器那块用一个假会话，不起容器：这里要验的是「环境返回什么 → 结果对象怎么填」这个映射，
而那个映射与容器里发生了什么无关。真起容器的那条链路由 `tools/soak/check_appworld.py` 走。
"""

from __future__ import annotations

import asyncio

import pytest

from polyloop.ports import Action, InvalidDecision
from polyloop.types import ActionStatus, ModelReply, ReplayPolicy, Role
from tools.soak.appworld import AppWorldError
from tools.soak.scenarios.appworld import (
    OBSERVATION_TEMPLATE,
    AppWorldExecutor,
    AppWorldParser,
    AppWorldScenarioError,
    build_budget,
    build_context,
    build_run_request,
    build_synthetic_observations,
    render_to_messages,
    split_by_role,
)


def _reply(content: str) -> ModelReply:
    return ModelReply(call_id="call-1", content=content, thinking="")


#: 一条正常的、能解析出动作的回复。
_ACTION_REPLY = _reply(
    "先看看有哪些 app。\n\n```python\nprint(apis.api_docs.show_app_descriptions())\n```\n"
)

#: 一条解析不出任何东西的回复。
_INVALID_REPLY = _reply("我觉得应该先登录 spotify，但我不确定要用哪个接口。")


# ---------------------------------------------------------------------------
# 一、解析器：公共契约的五条
# ---------------------------------------------------------------------------


def test_parse_is_synchronous():
    """`parse` 是同步的，不是协程（契约一）。"""
    parsed = AppWorldParser().parse(_ACTION_REPLY)

    assert not hasattr(parsed, "__await__")


@pytest.mark.parametrize(
    "content",
    [
        "```python\nprint(1)\n```\n然后我看到了 42。",  # 成功且要截断
        "```python\n```",  # 空的完整围栏
        "```python\nprint(1)",  # 未闭合围栏兜底
        "```python\nprint(1)\n``` done",  # 闭合围栏后跟了内容
        "什么代码都没有",  # 一个围栏都没有
    ],
)
def test_history_text_never_grows(content):
    """`history_text` 不会比模型原文长（契约二）。

    这一条正是我们与被复刻的 dissect 实现分道的地方：dissect 在「未闭合围栏兜底」那一支给
    历史文本补回结尾的三反引号，补一个字符就会让这条断言失败。
    """
    parsed = AppWorldParser().parse(_reply(content))

    assert isinstance(parsed.history_text, str)
    assert len(parsed.history_text) <= len(content)


def test_invalid_decision_carries_a_non_empty_explanation():
    """无效决策的说明文本非空（契约三）。"""
    parsed = AppWorldParser().parse(_INVALID_REPLY)

    assert isinstance(parsed.decision, InvalidDecision)
    assert isinstance(parsed.decision.explanation, str)
    assert parsed.decision.explanation != ""


def test_action_carries_its_trace_form():
    """动作分支带着这一步在轨迹里长什么样（契约四）。"""
    parsed = AppWorldParser().parse(_ACTION_REPLY)

    assert isinstance(parsed.decision, Action)
    assert isinstance(parsed.decision.text, str)
    assert parsed.decision.text == "print(apis.api_docs.show_app_descriptions())"
    # AppWorld 的动作是代码，不是工具调用。
    assert parsed.decision.tool_call is None


@pytest.mark.parametrize(
    "content",
    [
        "",
        "```",
        "``````",
        "```python",
        "```python\r\n```",
        "```PYTHON\nprint(1)\n```",
        "\n\n\n",
        '```python\nprint("```")\n```',
        "```python\n```python\n```",
        "{{ 不是模板 }} ```python``` ```",
    ],
)
def test_any_input_returns_a_decision_rather_than_raising(content):
    """怎么怪的输入都返回一个决策，不抛异常（契约五）。

    库对解析失败的处置是「留痕 + 回喂纠错说明 + 继续」；抛异常会让整次运行以一个跟模型无关
    的理由炸掉，而压测里模型什么都吐得出来。
    """
    parsed = AppWorldParser().parse(_reply(content))

    assert isinstance(parsed.decision, Action | InvalidDecision)
    if isinstance(parsed.decision, InvalidDecision):
        assert parsed.decision.explanation != ""


# ---------------------------------------------------------------------------
# 二、解析器：协议细节
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "content",
    [
        "```\nprint(1)\n```",
        "```bash\nls -la\n```",
        "```py\nprint(1)\n```",
    ],
)
def test_only_python_fences_count(content):
    """只认 ```python，裸围栏与别的语言一律判失败。

    放宽会踩到一类静默故障：模型贴一段 bash 或一段 JSON 出来，我们把它当 Python 交给环境，
    环境回一个语法错，事后统计会记成「模型写错代码」。
    """
    parsed = AppWorldParser().parse(_reply(content))

    assert isinstance(parsed.decision, InvalidDecision)


def test_first_block_wins_and_the_rest_is_cut_off():
    """多个围栏取第一个，并把回填历史的文本截断到它结束。

    截断不是可有可无：模型常在代码块后面自行编造「执行结果」，留着的话下一轮它会把那段幻想
    当成真发生过的事。
    """
    content = (
        "第一步：\n\n```python\nfirst()\n```\n\n"
        "Output:\n```\n我编的执行结果\n```\n\n"
        "第二步：\n\n```python\nsecond()\n```\n"
    )

    parsed = AppWorldParser().parse(_reply(content))

    assert isinstance(parsed.decision, Action)
    assert parsed.decision.text == "first()"
    assert parsed.history_text.endswith("```python\nfirst()\n```")
    assert "second()" not in parsed.history_text
    assert "我编的执行结果" not in parsed.history_text


def test_closing_fence_must_be_alone_on_its_line():
    """闭合围栏后面跟着别的内容判失败，并给出对症说明。

    照兜底分支处理的话，围栏连同后面的散文会一起被当成代码交给环境，而且判定为解析成功。
    """
    parsed = AppWorldParser().parse(_reply("```python\nprint(1)\n``` 好了"))

    assert isinstance(parsed.decision, InvalidDecision)
    assert "单独占一行" in parsed.decision.explanation


def test_code_containing_backticks_survives():
    """代码内部的三反引号不该把围栏提前切断。

    这是那条「闭合围栏必须独占一行」的正则要修的原 bug：非贪婪匹配会把 `print("` 当成完整
    代码交出去，剩下半截连同真正的代码从历史里消失。
    """
    parsed = AppWorldParser().parse(_reply('```python\nprint("```")\n```\n'))

    assert isinstance(parsed.decision, Action)
    assert parsed.decision.text == 'print("```")'


def test_empty_fence_is_a_failure_with_its_own_explanation():
    """空围栏判失败，说明与「没有围栏」那条不同。

    五条纠错说明各自对症，压成一句会改掉模型收到的信息，它的纠错行为也就跟着变。
    """
    empty = AppWorldParser().parse(_reply("```python\n```"))
    none_at_all = AppWorldParser().parse(_reply("我不知道该写什么"))

    assert isinstance(empty.decision, InvalidDecision)
    assert isinstance(none_at_all.decision, InvalidDecision)
    assert empty.decision.explanation != none_at_all.decision.explanation
    assert "空的" in empty.decision.explanation


def test_unclosed_fence_still_yields_code():
    """未闭合的围栏仍然抽得出代码，且历史文本不比原文长。

    dissect 在这一支补回结尾的三反引号，我们不补——公共契约要求 history_text 不长于原文。
    """
    content = "我来查一下。\n\n```python\nprint(apis.api_docs.show_app_descriptions())"

    parsed = AppWorldParser().parse(_reply(content))

    assert isinstance(parsed.decision, Action)
    assert parsed.decision.text == "print(apis.api_docs.show_app_descriptions())"
    assert parsed.history_text == content
    assert not parsed.history_text.endswith("```")


def test_parser_parameters_pin_the_action_language():
    """参数快照里写明动作语言与多围栏策略。"""
    assert AppWorldParser().parameters() == {
        "kind": "appworld_code_fence",
        "multi_block_policy": "first_only",
    }


# ---------------------------------------------------------------------------
# 三、提示词装配
# ---------------------------------------------------------------------------


def test_split_by_role_finds_the_boundaries():
    """行首独占一行的标记才是边界，正文里的 `USER:` 不是。"""
    text = "USER:\n你好\n\nASSISTANT:\n我在\n\nUSER:\n这里提到 USER: 但不在行首\n"

    blocks = split_by_role(text, where="样例")

    assert [role for role, _ in blocks] == [Role.USER, Role.ASSISTANT, Role.USER]
    assert blocks[0][1] == "你好"
    assert blocks[2][1] == "这里提到 USER: 但不在行首"


@pytest.mark.parametrize(
    "text",
    [
        "这份模板一个角色标记都没有",
        "开头就有内容\nUSER:\n你好\n",
        "USER:\n\nASSISTANT:\n我在\n",
    ],
)
def test_split_by_role_rejects_malformed_templates(text):
    """没有标记、标记之前有内容、某一段是空的，三种都报错。"""
    with pytest.raises(AppWorldScenarioError):
        split_by_role(text, where="样例")


def test_variables_are_rendered_per_block():
    """`{{ 变量 }}` 与 `{{ 变量.字段 }}` 两种形态都认，花括号内两侧允许有空格。"""
    text = "USER:\n{{greeting}} {{ user.first_name }} {{  user.last_name  }}\n"

    messages = render_to_messages(
        text,
        {"greeting": "你好", "user": {"first_name": "Sam", "last_name": "Carter"}},
        where="样例",
    )

    assert len(messages) == 1
    assert messages[0].content[0].text == "你好 Sam Carter"


def test_missing_variable_is_an_error_not_an_empty_string():
    """变量没提供就报错。渲染成空串的表现是提示词里凭空少一段，模型成绩下降而没人知道原因。"""
    with pytest.raises(AppWorldScenarioError, match="instruction"):
        render_to_messages("USER:\n任务：{{ instruction }}\n", {}, where="样例")


def test_missing_attribute_is_an_error():
    """点号取字段时字段不存在同样报错。"""
    with pytest.raises(AppWorldScenarioError):
        render_to_messages(
            "USER:\n{{ user.email }}\n", {"user": {"first_name": "Sam"}}, where="样例"
        )


@pytest.mark.parametrize(
    "text",
    [
        "USER:\n你好 {{ 中文变量 }}\n",
        "USER:\n你好 {{ a-b }}\n",
        "USER:\n你好 {{ name }\n",
    ],
)
def test_leftover_double_braces_are_an_error(text):
    """形态不对的占位符要当场报错，不能原样留给模型看。

    静默留下一个 `{{ instruction }}` 会让模型看见字面量占位符，而那在轨迹里看起来只是模型
    表现差。
    """
    with pytest.raises(AppWorldScenarioError):
        render_to_messages(text, {"name": "Sam"}, where="样例")


def test_interpolated_content_cannot_move_message_boundaries():
    """插值内容里出现独占一行的 `USER:` 不会切出多余的消息。

    先切后渲染才有这个性质。反过来的话，`run_prefix.txt` 里那个位于代码围栏内部的
    `{{ app_descriptions }}` 只要带上一行 `USER:`，前缀就会多切出几条消息——而且不报错。
    """
    text = "USER:\n```\n{{ app_descriptions }}\n```\n\nASSISTANT:\n收到\n"
    payload = "spotify: 音乐\nUSER:\nvenmo: 转账"

    messages = render_to_messages(text, {"app_descriptions": payload}, where="样例")

    assert len(messages) == 2
    assert messages[0].role is Role.USER
    assert messages[1].role is Role.ASSISTANT
    assert "USER:" in messages[0].content[0].text


# ---------------------------------------------------------------------------
# 四、假会话与执行器
# ---------------------------------------------------------------------------


class _FakeSession:
    """一个假的 `AppWorldSession`：只实现场景层用得到的那几个成员，不碰容器。"""

    def __init__(
        self,
        *,
        output: str = "42\n",
        done: bool = False,
        raises: BaseException | None = None,
        instruction: str = "How many playlists do I have?",
        supervisor: dict[str, str] | None = None,
    ) -> None:
        self.task_id = "82e2fac_1"
        self.instruction = instruction
        self.supervisor = (
            supervisor
            if supervisor is not None
            else {
                "first_name": "Sam",
                "last_name": "Carter",
                "email": "sam.carter@example.com",
                "phone_number": "555-0100",
            }
        )
        self.n_executions = 0
        self._output = output
        self._done = done
        self._raises = raises
        self.executed: list[str] = []

    async def execute(self, code: str) -> str:
        self.executed.append(code)
        if self._raises is not None:
            raise self._raises
        self.n_executions += 1
        return self._output

    async def is_done(self) -> bool:
        return self._done


async def test_normal_execution_is_an_executed_outcome():
    """代码跑完了就是「已执行」，观察是环境的原样输出。"""
    session = _FakeSession(output="23\n")
    executor = AppWorldExecutor(session=session)

    outcome = await executor.execute(Action(text="print(23)", tool_call=None))

    assert outcome.status is ActionStatus.EXECUTED
    assert outcome.observation == "23\n"
    assert outcome.observation_is_synthetic is False
    assert outcome.env_reported_completion is False
    assert outcome.observation_truncated_chars == 0
    assert session.executed == ["print(23)"]


async def test_code_that_blows_up_is_still_an_executed_outcome():
    """模型写的代码报错是一条正常观察，不是环境故障。

    压成 `ENV_ERROR` 就等于把「模型写错了」和「环境坏了」混成同一件事，而停止判定对这两者
    的处置完全不同。
    """
    traceback = "Traceback (most recent call last):\nNameError: name 'x' is not defined"
    executor = AppWorldExecutor(session=_FakeSession(output=traceback))

    outcome = await executor.execute(Action(text="print(x)", tool_call=None))

    assert outcome.status is ActionStatus.EXECUTED
    assert outcome.observation == traceback


async def test_environment_reported_completion_is_forwarded():
    """环境说做完了，这个事实要原样填进结果——它是「任务完成」这个停止原因的唯一证据来源。"""
    executor = AppWorldExecutor(session=_FakeSession(done=True))

    outcome = await executor.execute(Action(text="apis.supervisor.complete_task()", tool_call=None))

    assert outcome.env_reported_completion is True
    assert outcome.status is ActionStatus.EXECUTED


async def test_environment_failure_becomes_env_error():
    """`AppWorldError` 才是环境故障，走 `ENV_ERROR`，观察是那条错误文本本身。"""
    executor = AppWorldExecutor(
        session=_FakeSession(raises=AppWorldError("/execute 返回 HTTP 500"))
    )

    outcome = await executor.execute(Action(text="print(1)", tool_call=None))

    assert outcome.status is ActionStatus.ENV_ERROR
    assert outcome.observation == "/execute 返回 HTTP 500"
    assert outcome.observation_is_synthetic is False
    assert outcome.env_reported_completion is False
    assert outcome.observation_truncated_chars == 0


async def test_cancellation_passes_through():
    """取消原样穿出去（CLAUDE.md §1.6）。

    吞掉它的后果不是「取消失败」这么直白——是容器租约、连接和临时目录持续泄漏，而且一声
    不吭。
    """
    executor = AppWorldExecutor(session=_FakeSession(raises=asyncio.CancelledError()))

    with pytest.raises(asyncio.CancelledError):
        await executor.execute(Action(text="print(1)", tool_call=None))


async def test_unexpected_errors_are_not_swallowed():
    """未预料的异常穿出去（CLAUDE.md §1.7）——压测就是要看见它。"""
    executor = AppWorldExecutor(session=_FakeSession(raises=ZeroDivisionError("boom")))

    with pytest.raises(ZeroDivisionError):
        await executor.execute(Action(text="print(1)", tool_call=None))


async def test_execution_count_comes_from_the_environment():
    """执行次数转发环境自己的计数，不是本地计数器。

    故障注入那一步要拿它跟库的自述对账，用本地计数器的话对的就是我们和我们自己。
    """
    session = _FakeSession()
    executor = AppWorldExecutor(session=session)

    assert executor.env_executions == 0
    await executor.execute(Action(text="print(1)", tool_call=None))
    assert executor.env_executions == 1
    assert executor.env_executions == session.n_executions


def test_executor_parameters_keep_the_port_out():
    """参数快照里有题号，没有 base_url——端口每次运行都不同，进快照会让续跑必然报参数漂移。"""
    parameters = AppWorldExecutor(session=_FakeSession()).parameters()

    assert parameters == {"kind": "appworld_session", "task_id": "82e2fac_1"}
    assert "base_url" not in parameters


# ---------------------------------------------------------------------------
# 五、整体装配
# ---------------------------------------------------------------------------


def test_context_is_built_from_the_vendored_templates():
    """两份真模板切得出消息，题面与主管信息落在题级那一段。"""
    session = _FakeSession(instruction="How many playlists do I have?")

    context = build_context(session=session, app_descriptions="spotify: 音乐\nvenmo: 转账")

    assert context.run_level[0].role is Role.USER
    assert len(context.run_level) > 1
    run_text = "\n".join(block.text for m in context.run_level for block in m.content)
    assert "spotify: 音乐" in run_text
    # 题面属于题级那一段：它逐题变化，混进 run 级会打断供应商的前缀缓存。
    assert "How many playlists do I have?" not in run_text

    assert len(context.goal_level) == 1
    goal_text = context.goal_level[0].content[0].text
    assert "How many playlists do I have?" in goal_text
    assert "Sam Carter" in goal_text
    assert "sam.carter@example.com" in goal_text
    assert "555-0100" in goal_text


@pytest.mark.parametrize("blank_field", ["first_name", "last_name", "email", "phone_number"])
def test_blank_supervisor_fields_are_rejected(blank_field):
    """主管信息里任何一个字段为空都报错。

    渲染器只拦得住「键不存在」。空值会安静地渲染成一段空白，提示词变成 `phone number is `，
    模型照着去查一个不存在的人——不报错，只让成绩变差。
    """
    supervisor = {
        "first_name": "Sam",
        "last_name": "Carter",
        "email": "sam.carter@example.com",
        "phone_number": "555-0100",
    }
    supervisor[blank_field] = ""

    with pytest.raises(AppWorldScenarioError, match=blank_field):
        build_context(session=_FakeSession(supervisor=supervisor), app_descriptions="spotify: 音乐")


def test_budget_matches_the_production_configuration():
    """预算四项照 dissect 的生产配置（max_actions 除外，AppWorld 每步至多一个动作）。"""
    budget = build_budget()

    assert budget.max_steps == 40
    assert budget.max_actions == 40
    assert budget.max_consecutive_parse_failures == 3
    assert budget.max_prompt_chars == 400000


def test_synthetic_observations_are_all_non_empty():
    """三段合成观察都得有内容——它们会被模型看见。"""
    synthetic = build_synthetic_observations()

    assert synthetic.action_rejected
    assert synthetic.env_failed
    assert synthetic.model_call_failed


def test_run_request_is_assembled_with_the_pinned_values():
    """装出来的请求：空注册表、NEVER 重放、逐字照抄的观察格式。"""
    session = _FakeSession()

    request = build_run_request(
        run_id="soak-0001",
        session=session,
        app_descriptions="spotify: 音乐",
        model_binding={"scenario": "appworld"},
    )

    assert request.run_id == "soak-0001"
    assert isinstance(request.action_executor, AppWorldExecutor)
    assert request.tools.names() == ()
    assert request.injections == {}
    assert request.model_binding == {"scenario": "appworld"}
    # AppWorld 的代码执行有真实副作用（转账、下单），状态未知时绝不重放。
    assert request.model_replay_policy is ReplayPolicy.NEVER
    assert request.observation_template == OBSERVATION_TEMPLATE
    assert request.observation_template == "Output:\n```\n{observation}\n```\n\n"
    assert request.cancel_grace_seconds == 5.0


def test_run_request_snapshot_has_no_container_port():
    """参数快照里不该出现容器端口，否则续跑必然报参数漂移。"""
    request = build_run_request(
        run_id="soak-0002",
        session=_FakeSession(),
        app_descriptions="spotify: 音乐",
        model_binding={},
    )

    snapshot = request.parameter_snapshot()

    assert not any("127.0.0.1" in value for value in snapshot.values())
    assert "82e2fac_1" in snapshot.values()