"""Static coverage for the isolated Codex app-server provider."""

import inspect
import io
import json
import os
import queue
import time

from core.llm_client import LLMClient
from core.llm_providers.codex_app_server import (
    LLMCodexAppServerMixin,
    _CodexAppServerProtocolError,
)
from services.llm_connection import LLMConnectionService


def test_only_codex_app_server_is_registered():
    assert "codex" not in LLMClient.PROVIDERS
    assert "codex" not in LLMClient.DEFAULT_MODELS
    assert "codex-app-server" in LLMClient.PROVIDERS
    assert LLMClient.DEFAULT_MODELS["codex-app-server"] == "gpt-5.5"
    assert not hasattr(LLMClient, "_stream_codex")
    assert hasattr(LLMClient, "_stream_codex_app_server")


def test_codex_app_server_dispatch_contracts_are_wired():
    src = inspect.getsource(LLMClient.send_user_message)
    assert 'self.provider == "codex-app-server"' in src
    assert "_codex_app_send_user_message" in src

    complete_src = inspect.getsource(LLMClient.complete)
    stream_src = inspect.getsource(LLMClient.complete_stream)
    assert "_stream_codex_app_server" in complete_src
    assert "_stream_codex_app_server" in stream_src
    assert "_stream_codex(" not in complete_src
    assert "_stream_codex(" not in stream_src
    assert '"codex-app-server"' in complete_src
    assert '"codex-app-server"' in stream_src


def test_codex_app_server_uses_app_server_protocol_and_local_images():
    src = inspect.getsource(LLMCodexAppServerMixin)
    assert '["app-server"]' in src
    assert '"thread/start"' in src
    assert '"thread/resume"' in src
    assert '"turn/start"' in src
    assert '"turn/steer"' in src
    assert '"effort"' in src
    assert '"summary"' in src
    assert '"localImage"' in src
    assert "_codex_app_image_item" in src


def test_codex_app_server_rejects_non_remote_image_urls():
    assert LLMCodexAppServerMixin._codex_app_valid_remote_url("https://example.com/a.png")
    assert LLMCodexAppServerMixin._codex_app_valid_remote_url("http://example.com/a.png")
    assert LLMCodexAppServerMixin._codex_app_valid_remote_url("data:image/png;base64,abc")
    assert not LLMCodexAppServerMixin._codex_app_valid_remote_url("/files/abc/image.png")
    assert not LLMCodexAppServerMixin._codex_app_valid_remote_url("fs://filestore/abc/image.png")


def test_codex_app_server_file_id_attachment_wins_over_relative_url(monkeypatch, tmp_path):
    class FakeStore:
        def get_required(self, file_id, user_id="", conversation_id=""):
            assert file_id == "fid1"
            return "image.png", b"png", "image/png"

    class FakeFileStore:
        @staticmethod
        def instance():
            return FakeStore()

    import core.file_store as file_store
    monkeypatch.setattr(file_store, "FileStore", FakeFileStore)

    client = LLMCodexAppServerMixin()
    items = client._codex_app_attachment_items(
        [{"file_id": "fid1", "url": "/files/fid1/image.png", "mime_type": "image/png"}],
        user_id="u1", conversation_id="c1", workdir=str(tmp_path), container_dir="/cc_sessions/c1")

    assert items[0]["type"] == "text"
    assert items[1]["type"] == "localImage"
    assert not any(item.get("url", "").startswith("/files/") for item in items)


def test_codex_app_server_keeps_mcp_live_event_name_and_arguments_raw():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    assert '"name": raw_name' in src
    assert '"arguments": raw_args' in src
    assert '"tool": raw_name' in src
    assert '"raw_name": raw_name' not in src
    assert '"raw_tool": raw_name' not in src


def test_codex_app_server_rollout_receipt_marks_preempt_handled():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    preempt_block = src[src.index('if getattr(self, "_codex_app_preempt_pending", 0) > 0:'):]
    preempt_block = preempt_block[:preempt_block.index('self._codex_app_turn_completed_for_callback')]

    assert 'if pstatus == "done":' in preempt_block
    assert "self._had_preempts_this_turn = True" in preempt_block


def test_codex_app_server_stdio_is_utf8_and_ascii_safe_json():
    src = inspect.getsource(LLMCodexAppServerMixin)
    assert 'encoding="utf-8"' in src
    assert 'errors="replace"' in src
    assert "json.dumps(msg, ensure_ascii=True)" in src


def test_codex_app_server_initialize_error_includes_stderr():
    class DeadProc:
        stdin = io.StringIO()
        stdout = io.StringIO("")

    stderr = queue.Queue()
    stderr.put("codex: app-server command failed")
    client = LLMCodexAppServerMixin()

    try:
        client._codex_app_initialize(DeadProc(), stderr_lines=stderr)
    except _CodexAppServerProtocolError as exc:
        text = str(exc)
    else:
        raise AssertionError("initialize should fail when stdout is missing")

    assert "exited before response to initialize" in text
    assert "codex: app-server command failed" in text


def test_codex_app_server_reasoning_is_wired():
    complete_src = inspect.getsource(LLMClient.complete)
    stream_src = inspect.getsource(LLMClient.complete_stream)
    provider_src = inspect.getsource(LLMCodexAppServerMixin)
    assert "thinking_budget=thinking_budget" in complete_src
    assert "thinking_budget=thinking_budget" in stream_src
    assert "_codex_app_effort" in provider_src
    assert "count_messages_tokens" in provider_src
    assert "prompt_tokens" in provider_src
    assert "prompt_mode = \"resume\"" in provider_src
    assert "mode=%s" in provider_src
    assert "full_context_text" not in provider_src
    assert "tokens_in=max(0, int(prompt_tokens or 0))" in provider_src
    assert "item/reasoning/summaryTextDelta" in provider_src
    assert "item/reasoning/textDelta" in provider_src
    assert 'item.get("type") == "reasoning"' in provider_src
    assert "_append_final_reasoning" in provider_src
    assert "text in existing or existing in text" in provider_src


def test_codex_app_server_ephemeral_streams_bypass_cold_bootstrap():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    prompt_block = src[
        src.index("if is_ephemeral:"):
        src.index("elif thread_id:")]
    assert 'prompt_mode = "ephemeral"' in prompt_block
    assert "initial_text = self._codex_app_resume_text(messages)" in prompt_block
    assert "_codex_app_full_initial_text" not in prompt_block


def test_codex_app_server_ephemeral_streams_delete_workdir():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    cleanup_block = src[src.index("if is_ephemeral and workdir:"):]
    assert "shutil.rmtree(workdir" in cleanup_block
    assert "os.replace(workdir, stale)" in cleanup_block



def test_codex_app_server_effort_mapping():
    assert LLMCodexAppServerMixin._codex_app_effort(0, "") == "low"
    assert LLMCodexAppServerMixin._codex_app_effort(5000, "") == "medium"
    assert LLMCodexAppServerMixin._codex_app_effort(10000, "") == "high"
    assert LLMCodexAppServerMixin._codex_app_effort(20000, "") == "xhigh"
    assert LLMCodexAppServerMixin._codex_app_effort(0, "max") == "xhigh"
    assert LLMCodexAppServerMixin._codex_app_reasoning_summary("low") == "none"
    assert LLMCodexAppServerMixin._codex_app_reasoning_summary("medium") == "auto"




def test_codex_mcp_config_uses_absolute_python_for_app_server_spawn():
    from core.llm_providers.codex_session import CodexSessionMixin

    src = inspect.getsource(CodexSessionMixin._codex_setup_mcp_config)
    assert 'python_bin = "/usr/bin/python3"' in src


def test_codex_mcp_config_sets_long_tool_timeout():
    from core.llm_providers.codex_session import CodexSessionMixin

    src = inspect.getsource(CodexSessionMixin._codex_setup_mcp_config)
    assert "tool_timeout_sec = 3600" in src
    assert "tool_timeout_sec = 120" not in src
    assert "tool_timeout_sec = 300" not in src
    assert "startup_timeout_sec = 20" in src
    assert "supports_parallel_tool_calls = true" in src
    assert "[features]" in src
    assert "enable_fanout = true" in src
    assert "suppress_unstable_features_warning = true" in src


def test_codex_preamble_names_disabled_native_tools():
    from core.llm_providers.codex_session import CodexSessionMixin

    preamble = CodexSessionMixin._CODEX_PAWFLOW_PREAMBLE
    assert "ApplyPatch" in preamble
    assert "exec_command" in preamble
    assert "PawFlow MCP tools" in preamble
    assert "Native/internal provider tools are forbidden" in preamble
    assert "There is no native fallback path" in preamble


def test_codex_app_server_surfaces_native_tool_leaks():
    cmd = {
        "type": "commandExecution",
        "id": "cmd-1",
        "command": "git status",
        "cwd": "/workspace",
        "source": "agent",
        "status": "completed",
        "exitCode": 0,
        "aggregatedOutput": "ok",
    }
    patch = {
        "type": "fileChange",
        "id": "patch-1",
        "status": "completed",
        "changes": [{"path": "a.py"}],
    }
    dyn = {
        "type": "dynamicToolCall",
        "id": "dyn-1",
        "tool": "apply_patch",
        "arguments": {"patch": "..."},
        "status": "completed",
        "success": True,
        "contentItems": [],
    }

    assert LLMCodexAppServerMixin._codex_app_native_tool_name(cmd) == "codex_native_commandExecution"
    assert LLMCodexAppServerMixin._codex_app_native_tool_args(cmd)["command"] == "git status"
    assert "exit_code=0" in LLMCodexAppServerMixin._codex_app_native_tool_result(cmd)
    assert LLMCodexAppServerMixin._codex_app_native_tool_name(patch) == "codex_native_fileChange"
    assert "a.py" in LLMCodexAppServerMixin._codex_app_native_tool_result(patch)
    assert LLMCodexAppServerMixin._codex_app_native_tool_name(dyn) == "codex_native_apply_patch"
    assert LLMCodexAppServerMixin._codex_app_native_tool_args(dyn) == {"patch": "..."}


def test_codex_app_server_marks_native_and_mcp_tool_origins():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    assert '"tool_origin": "native"' in src
    assert '"tool_origin": "mcp"' in src


def test_codex_app_server_steers_mcp_hint_after_native_tool():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    assert "native_tool_hint_sent = False" in src
    assert "PawFlow internal efficiency hint" in src
    assert '"method": "turn/steer"' in src
    assert "_send_native_tool_hint(native_name)" in src


def test_codex_app_server_routes_live_thinking_through_neutral_callback():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    client_src = inspect.getsource(LLMClient.complete_stream)
    assert "thinking_callback=None" in src
    assert "def _flush_live_thinking" in src
    assert "thinking_callback(text)" in src
    assert 'block_callback("thinking"' not in src
    assert 'method in ("item/reasoning/summaryTextDelta", "item/reasoning/textDelta")' in src
    assert "thinking_callback=thinking_callback" in client_src


def test_codex_app_server_live_thinking_preserves_chunk_boundary_spaces():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    block = src[src.index("def _flush_live_thinking"):src.index("def _append_final_reasoning")]
    assert 'text = "".join(live_thinking_parts)' in block
    assert 'if not text.strip():' in block
    assert '"".join(live_thinking_parts).strip()' not in block


def test_codex_app_server_live_thinking_coalesces_tiny_chunks():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    block = src[src.index("def _flush_live_thinking"):src.index("def _append_final_reasoning")]
    assert "if len(text) < 160:" in block
    assert "if now - last_thinking_emit < 4.0:" in block
    assert "now - last_thinking_emit < 1.0" not in block


def test_codex_app_server_container_dir_matches_pool_namespace():
    from core.llm_providers.codex_session import _get_sessions_base

    workdir = os.path.join(_get_sessions_base(), "alice", "conv123", "assistant")
    assert LLMCodexAppServerMixin._codex_app_container_dir(workdir) == "/cc_sessions/conv123/assistant"


def test_codex_app_server_recovers_from_stale_thread_rollout():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    assert "_codex_app_missing_rollout_error" in src
    assert "store.set_extra(conv_id, thread_key, \"\")" in src
    assert "initial_text = self._codex_app_full_initial_text(" in src
    assert "messages, workdir, container_dir" in src
    assert "stale thread id" in src


def test_codex_app_server_registers_live_app_server_session():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    assert "CodexLiveRegistry" in src
    assert "live_reg.register" in src
    assert "[codex-app-live] keep-alive" in src
    assert "[codex-app-live] active" in src
    assert "Surface LIVE while the app-server turn is running" in src
    assert "live_reg.touch(live_key)" in src
    assert "is_reuse" in src
    assert "not turn_failed" in src
    assert "proc_alive" in src
    assert "is_process_alive" in src
    assert "is_container_alive" in src
    assert "process dead but container alive" in src
    assert "get_compatible" in src
    assert "live_session.turn_lock.acquire()" in src
    assert "live_session.turn_lock.release()" in src
    assert "live_reg.ensure_sweeper" in src
    assert "idle_ttl_seconds=int(_idle_ttl) if _idle_ttl else None" in src
    assert "active_turn=True" in src
    assert "active_turn=False" in src

    assert LLMConnectionService({}).get_service_actions() == []

    rules = LLMConnectionService({}).get_parameter_rules()
    assert any(
        rule.get("when", {}).get("provider") == ["codex-app-server"]
        and rule.get("set", {}).get("api_key", {}).get("visible") is True
        and rule.get("set", {}).get("effort", {}).get("visible") is True
        for rule in rules
    )


def test_codex_app_server_hands_context_compaction_to_pawflow():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    assert 'item.get("type") == "contextCompaction"' in src
    assert "CCCompactDetected" in src
    assert "handing compaction to PawFlow" in src


def test_codex_app_server_uses_completed_message_as_text_source_of_truth():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    assert 'item.get("type") in ("message", "agentMessage")' in src
    assert "assistant delta/final mismatch" in src
    assert "turn_text_parts = [final_text]" in src
    assert "turn_text_is_final = True" in src
    assert "dropping non-final assistant delta text" in src
    assert "final_text_parts.append(final_text)" in src
    assert '"".join(final_text_parts).strip() or "".join(text_parts).strip()' in src


def test_codex_app_server_does_not_stop_live_turn_at_tool_boundary_for_preempt():
    src = inspect.getsource(LLMCodexAppServerMixin._stream_codex_app_server)
    assert "preempt_boundary_stop" not in src
    assert "preempt pending after tool_result" not in src
    assert 'finish_reason="stop"' in src
    assert '"preempt_boundary_stop"' not in src


def test_codex_app_server_preempt_targets_matching_active_turn(monkeypatch):
    class Proc:
        def poll(self):
            return None

    client = LLMClient(provider="codex-app-server", config={})
    target_proc = Proc()
    other_proc = Proc()
    lock = client._codex_app_ensure_lock()
    client._codex_app_active = {
        ("alice", "conv-main", "assistant", 1.0): {
            "proc": target_proc,
            "thread_id": "thread-main",
            "turn_id": "turn-main",
            "workdir": "/tmp/main",
            "container_dir": "/cc_sessions/main",
            "started_at": 1.0,
        },
        ("alice", "_memory_extract", "memory", 2.0): {
            "proc": other_proc,
            "thread_id": "thread-memory",
            "turn_id": "turn-memory",
            "workdir": "/tmp/memory",
            "container_dir": "/cc_sessions/memory",
            "started_at": 2.0,
        },
    }
    client._codex_app_lock = lock
    sent = []
    monkeypatch.setattr(
        client, "_codex_app_send", lambda proc, msg: sent.append((proc, msg)))

    assert client._codex_app_send_user_message(
        "preempt",
        user_id="alice",
        conversation_id="conv-main",
        agent_name="assistant",
    ) is True

    assert sent
    proc, msg = sent[0]
    assert proc is target_proc
    assert msg["method"] == "turn/steer"
    assert msg["params"]["threadId"] == "thread-main"
    assert msg["params"]["expectedTurnId"] == "turn-main"


def test_codex_app_server_preempt_rollout_check_ignores_initial_prompt_false_positive(tmp_path):
    rollout = tmp_path / "rollout.jsonl"
    preempt_text = "Confirmes bordel !!!"
    rows = [
        {"payload": {"type": "message", "role": "user", "content": [
            {"type": "input_text", "text": f"old compacted history: {preempt_text}"},
        ]}},
        {"payload": {"type": "message", "role": "assistant", "content": [
            {"type": "output_text", "text": "ordinary work continued"},
        ]}},
    ]
    rollout.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8")

    status = LLMCodexAppServerMixin._codex_app_check_preempt_in_rollout(
        str(rollout), [{"text": preempt_text, "after_line": 2}])

    assert status == "unread"


def test_codex_app_server_preempt_rollout_check_accepts_post_steer_user_message(tmp_path):
    rollout = tmp_path / "rollout.jsonl"
    preempt_text = "test preempt exact"
    rows = [
        {"payload": {"type": "message", "role": "user", "content": [
            {"type": "input_text", "text": f"old compacted history: {preempt_text}"},
        ]}},
        {"payload": {"type": "message", "role": "assistant", "content": [
            {"type": "output_text", "text": "before steer"},
        ]}},
        {"payload": {"type": "message", "role": "user", "content": [
            {"type": "input_text", "text": preempt_text},
        ]}},
        {"payload": {"type": "message", "role": "assistant", "content": [
            {"type": "output_text", "text": "ack after steer"},
        ]}},
    ]
    rollout.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8")

    status = LLMCodexAppServerMixin._codex_app_check_preempt_in_rollout(
        str(rollout), [{"text": preempt_text, "after_line": 2}])

    assert status == "done"


def test_codex_app_server_preempt_rollout_check_only_requires_post_steer_user_item(tmp_path):
    rollout = tmp_path / "rollout.jsonl"
    preempt_text = "test preempt without visible ack"
    rows = [
        {"payload": {"type": "message", "role": "assistant", "content": [
            {"type": "output_text", "text": "before steer"},
        ]}},
        {"payload": {"type": "message", "role": "user", "content": [
            {"type": "input_text", "text": preempt_text},
        ]}},
    ]
    rollout.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8")

    status = LLMCodexAppServerMixin._codex_app_check_preempt_in_rollout(
        str(rollout), [{"text": preempt_text, "after_line": 1}])

    assert status == "done"


def test_codex_live_sweeper_does_not_evict_active_turn():
    from core.codex_live_registry import CodexLiveRegistry

    reg = CodexLiveRegistry()
    key = ("user", "conv", "assistant", "svc", 0)
    session = reg.register(
        key, "container", "/tmp/work", service_id="svc",
        session_id="thread", active_turn=True)
    session.last_used = time.monotonic() - 9999

    assert reg.sweep_idle(ttl=1) == 0
    assert reg.get(key) is session

    reg.ensure_sweeper(idle_ttl_seconds=1800)
    assert reg._idle_ttl == 1800
    reg._sweeper_stop.set()
    with reg._lock:
        reg._containers.clear()


def test_codex_live_lookup_falls_back_when_pool_idx_extra_is_missing():
    from core.codex_live_registry import CodexLiveRegistry

    reg = CodexLiveRegistry()
    key = ("user", "conv", "assistant", "svc", 0)
    session = reg.register(
        key, "container", "/tmp/work", service_id="svc",
        session_id="thread")
    session.last_used = time.monotonic()

    assert reg.get(("user", "conv", "assistant", "svc", -1)) is None
    compatible = reg.get_compatible("user", "conv", "assistant", "svc")
    assert compatible == (key, session)


def test_codex_live_session_tracks_process_and_container_separately(monkeypatch):
    from core.codex_live_registry import CodexLiveSession
    from core.codex_pool import CodexPool

    class DeadProc:
        def poll(self):
            return 0

    class LiveProc:
        def poll(self):
            return None

    class Pool:
        def _is_container_alive(self, name):
            return name == "container"

    monkeypatch.setattr(CodexPool, "instance", classmethod(lambda cls: Pool()))

    session = CodexLiveSession(
        container_name="container", workdir="/tmp", service_id="svc",
        proc=DeadProc())
    assert not session.is_process_alive()
    assert session.is_container_alive()
    assert not session.is_alive()

    session.proc = LiveProc()
    assert session.is_process_alive()
    assert session.is_container_alive()
    assert session.is_alive()
