from __future__ import annotations

import json
import queue
import threading
import time
from collections.abc import Callable, Mapping
from concurrent.futures import ThreadPoolExecutor
from copy import deepcopy

import pytest

from cognitive_agent.application.codex_assistant import (
    ActiveCodexTurn,
    CodexAssistantService,
    _mode_profile_create_parts,
)

CONTROL_TERMS = ("空调", "座椅", "车窗", "遮阳帘", "氛围灯", "阅读灯", "出风口")


@pytest.mark.parametrize(("message", "mode_name"), [
    ("构建一个观影模式", "观影模式"),
    ("为主驾构建一个观影模式", "主驾观影模式"),
])
def test_local_first_callbacks_do_not_intercept_scoped_mode_draft_creation(message, mode_name):
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    events = []
    control_calls = []

    def execute_control(instruction, _progress, plan):
        control_calls.append((instruction, plan))
        return _preview_result(mode_name)

    result = assistant.stream(
        message, conversation_id="conv-merged-scene-router", principal="test",
        history=[], emit=events.append, execute_control=execute_control,
        preview_control=lambda *_args: pytest.fail("mode previews must not partially execute via local-first controls"),
        supplement_control=lambda *_args: pytest.fail("mode previews must not become unresolved control supplements"),
        control_terms=CONTROL_TERMS,
        capability_snapshot=_catalog_with_templates(_scene_template()),
        scene_capability_context=_scene_context(),
    )
    assert result["success"] is True
    assert result["route"] == "scene_template"
    assert [instruction for instruction, _ in control_calls] == [message]
    assert not client.structured_calls and not client.legacy_calls


def test_local_first_callbacks_preserve_schedule_stop_route_and_answer():
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    events = []
    result = assistant.stream(
        "停止动态模式", conversation_id="conv-merged-schedule-stop", principal="test",
        history=[], emit=events.append,
        execute_control=lambda instruction, _progress: {
            "request_id": "req-stop-merged", "utterance": instruction,
            "status": "succeeded", "execution_mode": "dry_run",
            "orchestration_route": "mode_schedule_stop",
            "schedule_execution": {"status": "idle"},
        },
        preview_control=lambda *_args: pytest.fail("schedule stop must retain its scheduler route"),
        supplement_control=lambda *_args: pytest.fail("schedule stop is not an unresolved control"),
        control_terms=CONTROL_TERMS,
    )
    assert result["success"] is True
    assert "已模拟停止动态模式" in result["answer"]
    assert not client.structured_calls and not client.legacy_calls


def _structured_plan(
    *,
    temperature: float = 24,
    shade_position: int = 100,
) -> dict[str, object]:
    return {
        "requests": [
            {
                "capability": "climate.set_temperature",
                "target_scope": "vehicle",
                "parameters": [{"name": "temperature_c", "value": temperature}],
            },
            {
                "capability": "roof.set_shade_position",
                "target_scope": "vehicle",
                "parameters": [{"name": "position_pct", "value": shade_position}],
            },
        ],
        "ambiguities": [],
        "confidence": 0.94,
    }


def _legacy_plan(*, temperature: float = 24) -> dict[str, object]:
    structured = _structured_plan(temperature=temperature)
    requests = []
    domains = {"climate.set_temperature": "climate", "roof.set_shade_position": "roof"}
    for item in structured["requests"]:
        requests.append(
            {
                "domain": domains[item["capability"]],
                "capability": item["capability"],
                "target_scope": item["target_scope"],
                "operation": "set",
                "duration_ms": None,
                "parameters": item["parameters"],
                "reason": "旧链路候选动作。",
            }
        )
    return {
        "detected_domains": ["climate", "roof"],
        "requests": requests,
        "ambiguities": [],
        "decision_summary": ["旧链路生成候选计划。"],
        "confidence": 0.94,
    }


def _completed_turn(
    thread_id: str,
    turn_id: str,
    answer: str,
) -> ActiveCodexTurn:
    turn = ActiveCodexTurn(thread_id=thread_id, turn_id=turn_id, events=queue.Queue())
    _finish_turn(turn, answer)
    return turn


def _finish_turn(turn: ActiveCodexTurn, answer: str) -> None:
    item_id = f"answer-{turn.turn_id}"
    turn.events.put(
        {
            "method": "item/started",
            "params": {
                "item": {
                    "id": item_id,
                    "type": "agentMessage",
                    "text": "",
                    "phase": "final_answer",
                }
            },
        }
    )
    turn.events.put(
        {
            "method": "item/agentMessage/delta",
            "params": {"itemId": item_id, "delta": answer},
        }
    )
    turn.events.put(
        {
            "method": "item/completed",
            "params": {
                "item": {
                    "id": item_id,
                    "type": "agentMessage",
                    "text": answer,
                    "phase": "final_answer",
                }
            },
        }
    )
    turn.events.put(
        {
            "method": "turn/completed",
            "params": {
                "turn": {
                    "id": turn.turn_id,
                    "status": "completed",
                    "items": [],
                    "error": None,
                }
            },
        }
    )


class StructuredSceneClient:
    def __init__(self) -> None:
        self.structured_calls: list[dict[str, object]] = []
        self.legacy_calls: list[dict[str, object]] = []
        self.thread_instructions: list[str] = []
        self.finished: list[ActiveCodexTurn] = []
        self.interrupted: list[ActiveCodexTurn] = []
        self.unsubscribed: list[str] = []
        self.structured_answers: queue.Queue[str | BaseException] = queue.Queue()
        self.on_structured_start: Callable[[], None] | None = None
        self.structured_turn_factory: Callable[[str, str], ActiveCodexTurn] | None = None
        self.on_legacy_start: Callable[[str, ActiveCodexTurn], None] | None = None
        self._lock = threading.Lock()

    @property
    def available(self) -> bool:
        return True

    @property
    def status(self) -> str:
        return "ready"

    def start(self, _callback_url: str, _callback_token: str) -> None:
        return None

    def close(self) -> None:
        return None

    def start_thread(
        self,
        *,
        developer_instructions: str,
        ephemeral: bool = True,
        replace_developer_instructions: bool = False,
        timeout: float | None = None,
    ) -> str:
        assert ephemeral is True
        del replace_developer_instructions, timeout
        with self._lock:
            self.thread_instructions.append(developer_instructions)
            return f"thread-{len(self.thread_instructions)}"

    def start_structured_turn(
        self,
        thread_id: str,
        text: str,
        *,
        application_context: Mapping[str, object] | None = None,
        output_schema: Mapping[str, object],
        timeout: float | None = None,
    ) -> ActiveCodexTurn:
        del timeout
        with self._lock:
            call_index = len(self.structured_calls) + 1
            self.structured_calls.append(
                {
                    "thread_id": thread_id,
                    "text": text,
                    "application_context": deepcopy(dict(application_context or {})),
                    "output_schema": deepcopy(dict(output_schema)),
                }
            )
        if self.on_structured_start is not None:
            self.on_structured_start()
        turn_id = f"structured-{call_index}"
        if self.structured_turn_factory is not None:
            return self.structured_turn_factory(thread_id, turn_id)
        try:
            answer = self.structured_answers.get_nowait()
        except queue.Empty:
            answer = json.dumps(_structured_plan(), ensure_ascii=False)
        if isinstance(answer, BaseException):
            raise answer
        return _completed_turn(thread_id, turn_id, answer)

    def start_turn(
        self,
        thread_id: str,
        text: str,
        request_token: str,
        *,
        application_context: Mapping[str, object] | None = None,
    ) -> ActiveCodexTurn:
        with self._lock:
            call_index = len(self.legacy_calls) + 1
            self.legacy_calls.append(
                {
                    "thread_id": thread_id,
                    "text": text,
                    "request_token": request_token,
                    "application_context": deepcopy(dict(application_context or {})),
                }
            )
        turn = _completed_turn(thread_id, f"legacy-{call_index}", "旧链路已生成模式。")
        if self.on_legacy_start is not None:
            self.on_legacy_start(request_token, turn)
        return turn

    def finish_turn(
        self,
        turn: ActiveCodexTurn,
        timeout: float | None = None,
    ) -> None:
        del timeout
        self.finished.append(turn)

    def interrupt(
        self,
        turn: ActiveCodexTurn,
        timeout: float | None = None,
    ) -> None:
        del timeout
        self.interrupted.append(turn)

    def unsubscribe_thread(
        self,
        thread_id: str,
        timeout: float | None = None,
    ) -> None:
        del timeout
        self.unsubscribed.append(thread_id)


def _scene_context(
    revision: str = "scene-contract-a",
    *,
    vehicle_id: str = "denza_n7",
) -> dict[str, object]:
    return {
        "schema_version": "n7web-scene-composition/v1",
        "composition_revision": revision,
        "vehicle": {"id": vehicle_id, "name": "腾势 N7"},
        "capabilities": [
            {
                "id": "climate.set_temperature",
                "domain": "climate",
                "target_scopes": ["vehicle", "driver", "passenger"],
                "parameters": [
                    {
                        "name": "temperature_c",
                        "type": "number",
                        "required": True,
                        "minimum": 16,
                        "maximum": 32,
                    }
                ],
            },
            {
                "id": "roof.set_shade_position",
                "domain": "roof",
                "target_scopes": ["vehicle"],
                "parameters": [
                    {
                        "name": "position_pct",
                        "type": "integer",
                        "required": True,
                        "minimum": 0,
                        "maximum": 100,
                    }
                ],
            },
        ],
    }


def _catalog(*, vehicle_id: str = "denza_n7") -> dict[str, object]:
    return {
        "catalog_revision": "full-runtime-revision",
        "vehicle": {"id": vehicle_id, "name": "腾势 N7"},
        "modes": [],
        "source_status": {"modes": "ready"},
    }


def _scene_template(
    *,
    template_id: str = "viewing",
    name: str = "观影模式",
    aliases: list[str] | None = None,
) -> dict[str, object]:
    revision = f"revision-{template_id}"
    plan = _legacy_plan()
    plan["metadata"] = {
        "source": "scene_template",
        "template_id": template_id,
        "template_revision": revision,
    }
    return {
        "id": template_id,
        "name": name,
        "aliases": aliases or ["电影模式"],
        "assistant_callable": True,
        "template_revision": revision,
        "plan": plan,
    }


def _catalog_with_templates(
    *templates: Mapping[str, object],
    modes: list[dict[str, object]] | None = None,
) -> dict[str, object]:
    catalog = _catalog()
    catalog["modes"] = deepcopy(modes or [])
    catalog["scene_templates"] = [deepcopy(dict(template)) for template in templates]
    catalog["source_status"]["scene_templates"] = "ready"
    return catalog


def _preview_result(
    mode_name: str,
    *,
    operation_count: int = 2,
    success: bool = True,
) -> dict[str, object]:
    if not success:
        return {
            "request_id": "req-failed",
            "status": "failed",
            "execution_mode": "dry_run",
            "hardware_execution_allowed": False,
            "orchestration_route": "mode_profile_create_preview",
            "validation": [{"code": "invalid_scene", "message": "invalid"}],
        }
    return {
        "request_id": f"req-{mode_name}",
        "status": "succeeded",
        "execution_mode": "dry_run",
        "execution_outcome": "simulated",
        "hardware_execution_allowed": False,
        "orchestration_route": "mode_profile_create_preview",
        "validation": [],
        "intent_ir": {"spec": {"actions": [{}, {}][:operation_count]}},
        "mode_profile_draft": {
            "name": mode_name,
            "instruction": f"进入{mode_name}",
            "source_request_id": f"req-{mode_name}",
            "vehicle_id": "denza_n7",
            "domains": ["climate", "roof"],
            "operation_count": operation_count,
            "generation_source": "codex_scene_composition",
        },
    }


def _invoke(
    assistant: CodexAssistantService,
    message: str,
    executor: Callable[..., dict[str, object]],
    *,
    conversation_id: str,
    principal: str = "alice",
    scene_context: Mapping[str, object] | None = None,
    catalog: Mapping[str, object] | None = None,
    emit_handler: Callable[[dict[str, object]], bool | None] | None = None,
) -> tuple[dict[str, object], list[dict[str, object]]]:
    events: list[dict[str, object]] = []

    def emit(event: dict[str, object]) -> bool | None:
        events.append(event)
        return emit_handler(event) if emit_handler is not None else None

    result = assistant.stream(
        message,
        conversation_id=conversation_id,
        principal=principal,
        history=[],
        emit=emit,
        execute_control=executor,
        capability_snapshot=catalog or _catalog(),
        capability_context={"general": "context"},
        scene_capability_context=scene_context or _scene_context(),
        control_terms=CONTROL_TERMS,
    )
    return result, events


def test_exact_scene_template_hit_skips_codex_and_previews_a_fresh_plan() -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    template = _scene_template()
    observed: list[dict[str, object]] = []

    def execute(
        instruction: str,
        _progress: Callable[[dict[str, object]], None],
        assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        assert instruction == "构建一个观影模式"
        observed.append(deepcopy(dict(assistant_plan)))
        # A service callback must not be able to mutate the catalog-owned plan.
        assistant_plan["requests"][0]["parameters"][0]["value"] = 31
        return _preview_result("观影模式")

    result, events = _invoke(
        assistant,
        "构建一个观影模式",
        execute,
        conversation_id="conv-scene-template-hit",
        catalog=_catalog_with_templates(template),
    )

    assert result["success"] is True
    assert result["route"] == "scene_template"
    assert client.structured_calls == []
    assert client.legacy_calls == []
    assert len(observed) == 1
    assert observed[0]["metadata"] == template["plan"]["metadata"]
    assert template["plan"]["requests"][0]["parameters"][0]["value"] == 24
    assert events[0]["engine"] == "n7web-scene-template"


def test_scene_template_miss_falls_through_to_structured_scene_composer() -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    observed: list[dict[str, object]] = []

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        observed.append(deepcopy(dict(assistant_plan)))
        return _preview_result("星夜模式")

    result, events = _invoke(
        assistant,
        "构建一个星夜模式",
        execute,
        conversation_id="conv-scene-template-miss",
        catalog=_catalog_with_templates(_scene_template()),
    )

    assert result["success"] is True
    assert len(client.structured_calls) == 1
    assert client.legacy_calls == []
    assert len(observed) == 1
    assert "metadata" not in observed[0]
    assert events[0]["engine"] == "codex-app-server-scene-composer"


def test_registered_mode_match_takes_precedence_over_same_named_template() -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    call_arity: list[int] = []

    def execute(*args: object) -> dict[str, object]:
        call_arity.append(len(args))
        return _preview_result("观影模式")

    result, events = _invoke(
        assistant,
        "构建一个观影模式",
        execute,
        conversation_id="conv-scene-template-registered-priority",
        catalog=_catalog_with_templates(
            _scene_template(),
            modes=[
                {
                    "id": "viewing-registered",
                    "name": "观影模式",
                    "aliases": ["电影模式"],
                    "assistant_callable": True,
                }
            ],
        ),
    )

    assert result["success"] is True
    assert call_arity == [2]
    assert client.structured_calls == []
    assert client.legacy_calls == []
    assert events[0]["engine"] == "n7web-mode-builder"


def test_explicit_scene_settings_take_precedence_over_template_defaults() -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    instructions: list[str] = []
    call_arity: list[int] = []

    def execute(*args: object) -> dict[str, object]:
        call_arity.append(len(args))
        instructions.append(str(args[0]))
        result = _preview_result("观影模式", operation_count=1)
        result["mode_profile_draft"]["generation_source"] = "explicit_scene_requirements"
        return result

    result, events = _invoke(
        assistant,
        "构建一个观影模式，空调温度调整到25度",
        execute,
        conversation_id="conv-scene-template-explicit-priority",
        catalog=_catalog_with_templates(_scene_template()),
    )

    assert result["success"] is True
    assert instructions == ["构建一个观影模式，空调温度调整到25度"]
    assert call_arity == [2]
    assert client.structured_calls == []
    assert client.legacy_calls == []
    assert events[0]["engine"] == "n7web-mode-builder"


@pytest.mark.parametrize(
    ("message", "requested_name"),
    (
        ("我想创建一个观影模式", "观影模式"),
        ("我希望构建观影模式呀！", "观影模式"),
        ("麻烦帮我新建一下观影模式吧", "观影模式"),
        ("请帮忙生成一套“观影模式”，先预览一下。", "观影模式"),
        ("可以帮我创建个「观影模式」给我看看", "观影模式"),
        ("能不能帮我创建一个影院模式，给我看看呀？", "影院模式"),
        ("能帮我创建一个观影模式吗？", "观影模式"),
        ("能否帮我创建一个影院模式吗", "影院模式"),
        ("可不可以帮我创建观影模式？", "观影模式"),
        ("请问能不能帮我创建影院模式吗？", "影院模式"),
        ("给我添加一个『电影模式』，预览看看！", "电影模式"),
        ("请帮我创建一个'影院模式', 给我看看!", "影院模式"),
        ("请 帮 我 创 建 一 个 《观 影 模 式》 吧 ！", "观影模式"),
    ),
)
def test_oral_template_create_variants_keep_the_zero_codex_fast_path(
    message: str,
    requested_name: str,
) -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    template = _scene_template(aliases=["电影模式", "影院模式"])
    observed: list[tuple[str, dict[str, object]]] = []

    def execute(
        instruction: str,
        _progress: Callable[[dict[str, object]], None],
        assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        observed.append((instruction, deepcopy(dict(assistant_plan))))
        return _preview_result(requested_name)

    result, events = _invoke(
        assistant,
        message,
        execute,
        conversation_id=f"conv-oral-template-{abs(hash(message))}",
        catalog=_catalog_with_templates(template),
    )

    assert _mode_profile_create_parts(message) == (requested_name, "")
    assert result["success"] is True
    assert result["route"] == "scene_template"
    assert events[0]["route"] == "scene_template"
    assert events[0]["engine"] == "n7web-scene-template"
    action = next(event for event in events if event["type"] == "action.completed")
    assert action["payload"]["mode_profile_draft"]["name"] == requested_name
    assert action["payload"]["execution_mode"] == "dry_run"
    assert action["payload"]["hardware_execution_allowed"] is False
    assert len(observed) == 1
    assert observed[0][1]["metadata"] == template["plan"]["metadata"]
    assert client.thread_instructions == []
    assert client.structured_calls == []
    assert client.legacy_calls == []


@pytest.mark.parametrize(
    "message",
    (
        "构建一个观影模式并预览",
        "请创建观影模式，先预览一下",
        "我希望创建观影模式：让我先看一下方案",
        "麻烦帮我生成观影模式，给我看看吧",
        "能不能帮我创建影院模式并让我先看看呀？",
    ),
)
def test_preview_and_look_suffixes_are_not_misclassified_as_device_settings(
    message: str,
) -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    template = _scene_template(aliases=["影院模式"])
    executor_arity: list[int] = []

    def execute(*args: object) -> dict[str, object]:
        executor_arity.append(len(args))
        return _preview_result("影院模式" if "影院" in message else "观影模式")

    result, events = _invoke(
        assistant,
        message,
        execute,
        conversation_id=f"conv-preview-suffix-{abs(hash(message))}",
        catalog=_catalog_with_templates(template),
    )

    parts = _mode_profile_create_parts(message)
    assert parts is not None
    assert parts[1] == ""
    assert result["route"] == "scene_template"
    assert executor_arity == [3]
    assert events[0]["engine"] == "n7web-scene-template"
    assert client.structured_calls == []
    assert client.legacy_calls == []


@pytest.mark.parametrize(
    ("message", "expected_settings"),
    (
        ("我想构建一个观影模式，空调温度调整到25度", "空调温度调整到25度"),
        ("请帮我创建一个影院模式：遮阳帘设置为100%", "遮阳帘设置为100%"),
        (
            "能不能帮我创建观影模式，把车窗全部关闭、氛围灯调暗",
            "把车窗全部关闭、氛围灯调暗",
        ),
    ),
)
def test_oral_create_with_explicit_device_settings_never_uses_template_defaults(
    message: str,
    expected_settings: str,
) -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    executor_calls: list[tuple[object, ...]] = []

    def execute(*args: object) -> dict[str, object]:
        executor_calls.append(args)
        result = _preview_result("观影模式", operation_count=1)
        result["mode_profile_draft"]["generation_source"] = (
            "explicit_scene_requirements"
        )
        return result

    result, events = _invoke(
        assistant,
        message,
        execute,
        conversation_id=f"conv-explicit-settings-{abs(hash(message))}",
        catalog=_catalog_with_templates(
            _scene_template(aliases=["影院模式"]),
        ),
    )

    parts = _mode_profile_create_parts(message)
    assert parts is not None
    assert parts[1] == expected_settings
    assert result["success"] is True
    assert events[0]["route"] == "vehicle_control"
    assert events[0]["engine"] == "n7web-mode-builder"
    assert len(executor_calls) == 1
    assert len(executor_calls[0]) == 2
    assert executor_calls[0][0] == " ".join(message.strip().split())
    assert client.structured_calls == []
    assert client.legacy_calls == []


@pytest.mark.parametrize(
    "message",
    (
        "我想创建一个星空漫游模式",
        "麻烦帮我构建星空漫游模式吧",
        "能不能帮我创建一个星空漫游模式，先给我看看呀？",
    ),
)
def test_unknown_oral_scene_names_still_fall_through_to_scene_composer(
    message: str,
) -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    executor_arity: list[int] = []

    def execute(*args: object) -> dict[str, object]:
        executor_arity.append(len(args))
        return _preview_result("星空漫游模式")

    result, events = _invoke(
        assistant,
        message,
        execute,
        conversation_id=f"conv-unknown-oral-scene-{abs(hash(message))}",
        catalog=_catalog_with_templates(_scene_template()),
    )

    parts = _mode_profile_create_parts(message)
    assert parts is not None
    assert parts[1] == ""
    assert result["success"] is True
    assert events[0]["route"] == "scene_composer"
    assert events[0]["engine"] == "codex-app-server-scene-composer"
    assert len(client.structured_calls) == 1
    assert client.legacy_calls == []
    assert executor_arity == [3]


@pytest.mark.parametrize(
    "message",
    (
        "不要创建观影模式",
        "请别执行创建观影模式",
        "先不要构建观影模式，给我解释一下",
        "请解释一下“创建一个观影模式”是什么意思",
        "引用‘构建一个观影模式’，不要执行",
        "复述：请创建一个观影模式",
        "如何创建一个观影模式？",
        "是否支持创建观影模式？",
        "能否创建观影模式？",
        "你可以创建哪些模式？",
        "构建一个观影模式，空调温度是25度还是26度？",
        "构建一个模式",
        "创建一个功能模式",
        "添加一个新模式",
        "预览观影模式",
    ),
)
def test_negative_quoted_explanatory_and_capability_queries_never_execute(
    message: str,
) -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    executor_calls = 0

    def execute(*_args: object) -> dict[str, object]:
        nonlocal executor_calls
        executor_calls += 1
        return _preview_result("观影模式")

    result, events = _invoke(
        assistant,
        message,
        execute,
        conversation_id=f"conv-safe-scene-language-{abs(hash(message))}",
        catalog=_catalog_with_templates(_scene_template()),
    )

    assert _mode_profile_create_parts(message) is None
    assert result["route"] == "agent"
    assert events[0]["route"] == "agent"
    assert executor_calls == 0
    assert client.structured_calls == []
    assert len(client.legacy_calls) == 1


def test_unknown_mode_uses_one_structured_turn_then_local_dry_run() -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    observed: list[tuple[str, dict[str, object]]] = []

    def execute(
        instruction: str,
        _progress: Callable[[dict[str, object]], None],
        assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        observed.append((instruction, deepcopy(dict(assistant_plan))))
        return _preview_result("星夜模式")

    result, events = _invoke(
        assistant,
        "构建一个星夜模式",
        execute,
        conversation_id="conv-scene-fast-success",
    )

    assert len(client.structured_calls) == 1
    assert client.legacy_calls == []
    assert observed[0][0] == "构建一个星夜模式"
    assert [item["capability"] for item in observed[0][1]["requests"]] == [
        "climate.set_temperature",
        "roof.set_shade_position",
    ]
    call = client.structured_calls[0]
    expected_context = _scene_context()
    expected_context["seat_scope_policy"] = {
        "selected_scopes": ["passenger"],
        "default_scope": "passenger",
        "applies_to": "seat capabilities in every phase",
    }
    expected_context["natural_language_routing"] = {
        "original_utterance": "构建一个星夜模式",
        "parsed_clauses": [],
        "resolved_clauses": [],
        "unresolved_clauses": ["构建一个星夜模式"],
        "cloud_input_utterance": "构建一个星夜模式",
    }
    assert call["application_context"] == expected_context
    assert call["output_schema"]["type"] == "object"
    assert call["output_schema"]["additionalProperties"] is False
    assert "requests" in call["output_schema"]["required"]
    assert result["success"] is True
    assert "预览" in result["answer"]
    assert "2项设备动作" in result["answer"]
    assert any(event["type"] == "action.completed" for event in events)
    assert all("真实下发" not in str(event) for event in events)


@pytest.mark.parametrize(
    "bad_answer",
    (
        "不是 JSON",
        "{}",
        '{"requests":"not-a-list"}',
    ),
)
def test_malformed_structured_output_falls_back_to_existing_agent_path(
    bad_answer: str,
) -> None:
    client = StructuredSceneClient()
    client.structured_answers.put(bad_answer)
    assistant = CodexAssistantService(client)
    fallback_plan = _legacy_plan(temperature=23)
    calls: list[dict[str, object]] = []

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        calls.append(deepcopy(dict(assistant_plan)))
        return _preview_result("阅读模式")

    def legacy(request_token: str, _turn: ActiveCodexTurn) -> None:
        assistant.execute_cockpit_request(request_token, fallback_plan)

    client.on_legacy_start = legacy
    result, _events = _invoke(
        assistant,
        "构建一个阅读模式",
        execute,
        conversation_id=f"conv-scene-fallback-{abs(hash(bad_answer))}",
    )

    assert len(client.structured_calls) == 1
    assert len(client.legacy_calls) == 1
    assert len(calls) == 1
    assert [item["capability"] for item in calls[0]["requests"]] == [
        "climate.set_temperature",
        "roof.set_shade_position",
    ]
    assert calls[0]["requests"][0]["parameters"] == [{"name": "temperature_c", "value": 23}]
    assert result["route"] == "agent"
    assert "预览" in result["answer"]


def test_same_normalized_scene_and_revision_reuses_plan_but_revalidates_every_request() -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    observed: list[dict[str, object]] = []

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        observed.append(deepcopy(dict(assistant_plan)))
        return _preview_result("静心模式")

    first, _ = _invoke(
        assistant,
        "创建一个静心模式",
        execute,
        conversation_id="conv-scene-cache-first",
    )
    second, _ = _invoke(
        assistant,
        "构建一个静心模式",
        execute,
        conversation_id="conv-scene-cache-second",
    )

    assert first["success"] is second["success"] is True
    assert len(client.structured_calls) == 1
    assert len(observed) == 2
    assert observed[0] == observed[1]


@pytest.mark.parametrize("changed_dimension", ("revision", "vehicle"))
def test_cache_is_invalidated_when_scene_contract_or_vehicle_changes(
    changed_dimension: str,
) -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        _assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        return _preview_result("接待模式")

    _invoke(
        assistant,
        "构建一个接待模式",
        execute,
        conversation_id="conv-scene-invalidate-first",
    )
    context = (
        _scene_context("scene-contract-b")
        if changed_dimension == "revision"
        else _scene_context(vehicle_id="another_vehicle")
    )
    catalog = _catalog(
        vehicle_id="another_vehicle" if changed_dimension == "vehicle" else "denza_n7"
    )
    _invoke(
        assistant,
        "构建一个接待模式",
        execute,
        conversation_id="conv-scene-invalidate-second",
        scene_context=context,
        catalog=catalog,
    )

    assert len(client.structured_calls) == 2


def test_failed_local_validation_is_never_cached() -> None:
    client = StructuredSceneClient()
    client.structured_answers.put(json.dumps(_structured_plan(temperature=18), ensure_ascii=False))
    client.structured_answers.put(json.dumps(_structured_plan(temperature=25), ensure_ascii=False))
    assistant = CodexAssistantService(client)
    attempts = 0

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        _assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        nonlocal attempts
        attempts += 1
        return _preview_result("派对模式", success=attempts > 1)

    first, _ = _invoke(
        assistant,
        "构建一个派对模式",
        execute,
        conversation_id="conv-scene-invalid-first",
    )
    second, _ = _invoke(
        assistant,
        "构建一个派对模式",
        execute,
        conversation_id="conv-scene-invalid-second",
    )

    assert first["success"] is False
    assert second["success"] is True
    assert len(client.structured_calls) == 2


def test_concurrent_identical_cold_requests_share_one_planner_turn() -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    planner_entered = threading.Event()
    release_planner = threading.Event()
    executor_calls = 0
    executor_lock = threading.Lock()

    def hold_planner() -> None:
        planner_entered.set()
        assert release_planner.wait(timeout=3)

    client.on_structured_start = hold_planner

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        _assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        nonlocal executor_calls
        with executor_lock:
            executor_calls += 1
        return _preview_result("露营模式")

    def run(index: int) -> dict[str, object]:
        result, _ = _invoke(
            assistant,
            "构建一个露营模式",
            execute,
            conversation_id=f"conv-scene-singleflight-{index}",
        )
        return result

    with ThreadPoolExecutor(max_workers=2) as pool:
        first = pool.submit(run, 1)
        assert planner_entered.wait(timeout=2)
        second = pool.submit(run, 2)
        time.sleep(0.15)
        assert len(client.structured_calls) == 1
        release_planner.set()
        results = [first.result(timeout=3), second.result(timeout=3)]

    assert all(result["success"] is True for result in results)
    assert len(client.structured_calls) == 1
    # A plan cache never authorizes execution: both requests still traverse the
    # current local validation/dry-run callback independently.
    assert executor_calls == 2


def test_cached_plan_is_copied_before_each_validation() -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    seen_temperatures: list[float] = []

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        requests = assistant_plan["requests"]
        assert isinstance(requests, list)
        temperature = requests[0]["parameters"][0]["value"]
        seen_temperatures.append(temperature)
        requests[0]["parameters"][0]["value"] = 31
        return _preview_result("专注模式")

    for suffix in ("first", "second"):
        _invoke(
            assistant,
            "构建一个专注模式",
            execute,
            conversation_id=f"conv-scene-copy-{suffix}",
        )

    assert len(client.structured_calls) == 1
    assert seen_temperatures == [24, 24]


def test_scene_preview_never_reports_or_requests_hardware_execution() -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    calls = 0

    def execute(
        instruction: str,
        _progress: Callable[[dict[str, object]], None],
        assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        nonlocal calls
        calls += 1
        assert instruction == "构建一个观景模式"
        assert assistant_plan["requests"]
        return _preview_result("观景模式")

    result, events = _invoke(
        assistant,
        "构建一个观景模式",
        execute,
        conversation_id="conv-scene-no-hardware",
    )

    assert calls == 1
    assert result["success"] is True
    action_completed = next(event for event in events if event["type"] == "action.completed")
    payload = action_completed["payload"]
    assert payload["execution_mode"] == "dry_run"
    assert payload["hardware_execution_allowed"] is False
    assert "approval_token" not in json.dumps(events, ensure_ascii=False)
    assert "command_plans" not in json.dumps(events, ensure_ascii=False)


def test_validated_scene_plan_cache_is_shared_across_principals() -> None:
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    executor_calls = 0

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        _assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        nonlocal executor_calls
        executor_calls += 1
        return _preview_result("露营模式")

    first, _ = _invoke(
        assistant,
        "构建一个露营模式",
        execute,
        conversation_id="conv-scene-global-cache-a",
        principal="alice",
    )
    second, _ = _invoke(
        assistant,
        "创建一个露营模式",
        execute,
        conversation_id="conv-scene-global-cache-b",
        principal="bob",
    )

    assert first["success"] is second["success"] is True
    assert len(client.structured_calls) == 1
    # Cross-user reuse applies only to the context-bound candidate plan. Each
    # user's request still enters the current local dry-run validation chain.
    assert executor_calls == 2


def test_structured_planner_timeout_returns_explicit_failure_without_general_fallback() -> None:
    client = StructuredSceneClient()
    hanging_turns: list[ActiveCodexTurn] = []

    def hanging_turn(thread_id: str, turn_id: str) -> ActiveCodexTurn:
        turn = ActiveCodexTurn(thread_id=thread_id, turn_id=turn_id, events=queue.Queue())
        hanging_turns.append(turn)
        return turn

    client.structured_turn_factory = hanging_turn
    assistant = CodexAssistantService(client)
    assistant._scene_turn_timeout = 0.05
    executor_calls = 0

    def execute(*_args: object) -> dict[str, object]:
        nonlocal executor_calls
        executor_calls += 1
        return _preview_result("超时模式")

    result, events = _invoke(
        assistant,
        "构建一个超时模式",
        execute,
        conversation_id="conv-scene-planner-timeout",
    )

    assert len(hanging_turns) == 1
    assert client.interrupted == hanging_turns
    assert client.legacy_calls == []
    assert executor_calls == 0
    assert result["route"] == "scene_composer"
    assert result["success"] is False
    assert "超过" in result["answer"]
    assert any(event["type"] == "work.completed" and event["success"] is False for event in events)
    assert any(
        event["type"] == "answer.completed"
        and event["route"] == "scene_composer"
        and "超过" in event["answer"]
        for event in events
    )
    assert any("超时" in str(event.get("display_text") or "") for event in events)
    assert all("切换完整理解链路" not in str(event) for event in events)


def test_scene_planner_concurrency_limit_rejects_excess_cold_scene_without_queueing(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("CODEX_ASSISTANT_SCENE_MAX_CONCURRENT_PLANS", "1")
    client = StructuredSceneClient()
    assistant = CodexAssistantService(client)
    first_planner_entered = threading.Event()
    release_planners = threading.Event()
    callback_lock = threading.Lock()
    active_planners = 0
    max_active_planners = 0

    def hold_planner() -> None:
        nonlocal active_planners, max_active_planners
        with callback_lock:
            active_planners += 1
            max_active_planners = max(max_active_planners, active_planners)
            first_planner_entered.set()
        try:
            assert release_planners.wait(timeout=3)
        finally:
            with callback_lock:
                active_planners -= 1

    client.on_structured_start = hold_planner

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        _assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        return _preview_result("并发模式")

    def run(index: int, mode_name: str) -> dict[str, object]:
        result, _ = _invoke(
            assistant,
            f"构建一个{mode_name}",
            execute,
            conversation_id=f"conv-scene-concurrency-{index}",
            principal=f"user-{index}",
        )
        return result

    with ThreadPoolExecutor(max_workers=1) as pool:
        first = pool.submit(run, 1, "森林模式")
        assert first_planner_entered.wait(timeout=2)
        second_result, second_events = _invoke(
            assistant,
            "构建一个阅读模式",
            execute,
            conversation_id="conv-scene-concurrency-2",
            principal="user-2",
        )
        calls_before_release = len(client.structured_calls)
        max_active_before_release = max_active_planners
        release_planners.set()
        first_result = first.result(timeout=3)

    assert first_result["success"] is True
    assert second_result["route"] == "scene_composer"
    assert second_result["success"] is False
    assert "任务较多" in second_result["answer"]
    assert any(
        event["type"] == "work.completed" and event["success"] is False for event in second_events
    )
    assert calls_before_release == 1
    assert max_active_before_release == 1
    assert len(client.structured_calls) == 1
    assert client.legacy_calls == []
    assert max_active_planners == 1


def test_singleflight_wait_timeout_fails_without_starting_a_second_agent_turn() -> None:
    client = StructuredSceneClient()
    active_ready = threading.Event()
    active_turns: list[ActiveCodexTurn] = []

    def hanging_turn(thread_id: str, turn_id: str) -> ActiveCodexTurn:
        turn = ActiveCodexTurn(thread_id=thread_id, turn_id=turn_id, events=queue.Queue())
        active_turns.append(turn)
        active_ready.set()
        return turn

    client.structured_turn_factory = hanging_turn
    assistant = CodexAssistantService(client)
    assistant._scene_turn_timeout = 2
    executor_calls = 0

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        _assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        nonlocal executor_calls
        executor_calls += 1
        return _preview_result("亲子模式")

    def run_leader() -> dict[str, object]:
        result, _ = _invoke(
            assistant,
            "构建一个亲子模式",
            execute,
            conversation_id="conv-scene-wait-leader",
            principal="leader",
        )
        return result

    with ThreadPoolExecutor(max_workers=1) as pool:
        leader = pool.submit(run_leader)
        assert active_ready.wait(timeout=2)
        assistant._scene_turn_timeout = 0.05
        waiter_result, waiter_events = _invoke(
            assistant,
            "创建一个亲子模式",
            execute,
            conversation_id="conv-scene-wait-timeout",
            principal="waiter",
        )
        structured_calls_before_completion = len(client.structured_calls)
        legacy_calls_before_completion = list(client.legacy_calls)
        executor_calls_before_completion = executor_calls
        assistant._scene_turn_timeout = 2
        _finish_turn(
            active_turns[0],
            json.dumps(_structured_plan(), ensure_ascii=False),
        )
        leader_result = leader.result(timeout=3)

    assert structured_calls_before_completion == 1
    assert legacy_calls_before_completion == []
    assert executor_calls_before_completion == 0
    assert waiter_result["route"] == "scene_composer"
    assert waiter_result["success"] is False
    assert "超过" in waiter_result["answer"]
    assert any(
        event["type"] == "work.completed" and event["success"] is False for event in waiter_events
    )
    assert any("超时" in str(event.get("display_text") or "") for event in waiter_events)
    assert leader_result["success"] is True
    assert executor_calls == 1
    assert client.interrupted == []


def test_cancelled_singleflight_waiter_does_not_cancel_or_execute_leader_plan() -> None:
    client = StructuredSceneClient()
    active_ready = threading.Event()
    active_turns: list[ActiveCodexTurn] = []

    def hanging_turn(thread_id: str, turn_id: str) -> ActiveCodexTurn:
        turn = ActiveCodexTurn(thread_id=thread_id, turn_id=turn_id, events=queue.Queue())
        active_turns.append(turn)
        active_ready.set()
        return turn

    client.structured_turn_factory = hanging_turn
    assistant = CodexAssistantService(client)
    assistant._scene_turn_timeout = 2
    executor_calls = 0

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        _assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        nonlocal executor_calls
        executor_calls += 1
        return _preview_result("陪伴模式")

    def run_leader() -> dict[str, object]:
        result, _ = _invoke(
            assistant,
            "构建一个陪伴模式",
            execute,
            conversation_id="conv-scene-cancel-leader",
            principal="leader",
        )
        return result

    def cancel_on_singleflight_wait(event: dict[str, object]) -> bool | None:
        if event.get("type") == "work.progress" and "相同场景" in str(
            event.get("display_text") or ""
        ):
            return False
        return None

    with ThreadPoolExecutor(max_workers=1) as pool:
        leader = pool.submit(run_leader)
        assert active_ready.wait(timeout=2)
        waiter_result, _ = _invoke(
            assistant,
            "创建一个陪伴模式",
            execute,
            conversation_id="conv-scene-cancel-waiter",
            principal="waiter",
            emit_handler=cancel_on_singleflight_wait,
        )
        assert waiter_result == {"cancelled": True}
        assert len(client.structured_calls) == 1
        assert client.legacy_calls == []
        assert executor_calls == 0
        assert client.interrupted == []

        _finish_turn(
            active_turns[0],
            json.dumps(_structured_plan(), ensure_ascii=False),
        )
        leader_result = leader.result(timeout=3)

    assert leader_result["success"] is True
    assert executor_calls == 1
    assert client.interrupted == []


def test_continuous_delta_stream_still_detects_browser_cancellation() -> None:
    client = StructuredSceneClient()
    active_turns: list[ActiveCodexTurn] = []

    def busy_turn(thread_id: str, turn_id: str) -> ActiveCodexTurn:
        turn = ActiveCodexTurn(thread_id=thread_id, turn_id=turn_id, events=queue.Queue())
        for _ in range(10_000):
            turn.events.put(
                {
                    "method": "item/agentMessage/delta",
                    "params": {"delta": "x"},
                }
            )
        active_turns.append(turn)
        return turn

    client.structured_turn_factory = busy_turn
    assistant = CodexAssistantService(client)
    assistant._scene_progress_interval = 0.001
    executor_calls = 0

    def execute(*_args: object) -> dict[str, object]:
        nonlocal executor_calls
        executor_calls += 1
        return _preview_result("持续输出模式")

    def disconnect_on_heartbeat(event: dict[str, object]) -> bool | None:
        if event.get("type") == "work.progress" and event.get("display_text") == (
            "小迪正在生成结构化模式方案……"
        ):
            return False
        return None

    result, _events = _invoke(
        assistant,
        "构建一个持续输出模式",
        execute,
        conversation_id="conv-scene-continuous-cancel",
        emit_handler=disconnect_on_heartbeat,
    )

    assert result == {"cancelled": True}
    assert client.interrupted == active_turns
    assert active_turns[0].events.qsize() > 0
    assert client.legacy_calls == []
    assert executor_calls == 0


def test_cancelled_singleflight_leader_allows_a_follower_to_replan() -> None:
    client = StructuredSceneClient()
    first_planner_entered = threading.Event()
    follower_waiting = threading.Event()
    release_first_planner = threading.Event()
    starts = 0
    starts_lock = threading.Lock()
    executor_calls = 0

    def hold_only_first_planner() -> None:
        nonlocal starts
        with starts_lock:
            starts += 1
            current = starts
        if current == 1:
            first_planner_entered.set()
            assert release_first_planner.wait(timeout=3)

    client.on_structured_start = hold_only_first_planner
    assistant = CodexAssistantService(client)

    def execute(
        _instruction: str,
        _progress: Callable[[dict[str, object]], None],
        _assistant_plan: Mapping[str, object],
    ) -> dict[str, object]:
        nonlocal executor_calls
        executor_calls += 1
        return _preview_result("接续模式")

    def cancel_leader(event: dict[str, object]) -> bool | None:
        if event.get("type") == "work.progress" and "小迪已开始" in str(
            event.get("display_text") or ""
        ):
            return False
        return None

    def observe_waiter(event: dict[str, object]) -> bool | None:
        if event.get("type") == "work.progress" and "相同场景" in str(
            event.get("display_text") or ""
        ):
            follower_waiting.set()
        return None

    with ThreadPoolExecutor(max_workers=2) as pool:
        leader = pool.submit(
            _invoke,
            assistant,
            "构建一个接续模式",
            execute,
            conversation_id="conv-scene-replan-leader",
            principal="leader",
            emit_handler=cancel_leader,
        )
        assert first_planner_entered.wait(timeout=2)
        follower = pool.submit(
            _invoke,
            assistant,
            "创建一个接续模式",
            execute,
            conversation_id="conv-scene-replan-follower",
            principal="follower",
            emit_handler=observe_waiter,
        )
        assert follower_waiting.wait(timeout=2)
        release_first_planner.set()
        leader_result, _leader_events = leader.result(timeout=3)
        follower_result, _follower_events = follower.result(timeout=3)

    assert leader_result == {"cancelled": True}
    assert follower_result["success"] is True
    assert len(client.structured_calls) == 2
    assert client.legacy_calls == []
    assert executor_calls == 1