import asyncio
from contextlib import asynccontextmanager
from types import SimpleNamespace

import pytest
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from lib.db.base import Base
from server.agent_runtime import session_manager as sm_mod
from server.agent_runtime.session_actor import SessionActor
from server.agent_runtime.session_manager import ManagedSession
from server.agent_runtime.session_store import SessionMetaStore
from tests.fakes import FakeSDKClient


class _FakeOptions:
    def __init__(self, **kwargs):
        self.kwargs = kwargs


class _FakeClaudeClient:
    """Minimal ClaudeSDKClient stand-in used by SessionActor.

    Implements the async-context-manager protocol plus the narrow surface the
    actor touches: ``query`` / ``interrupt`` / ``receive_response``. ``connect``
    is kept for the legacy get_or_connect path-check assertion.
    """

    def __init__(self, options):
        self.options = options
        self.connected = False

    async def __aenter__(self):
        self.connected = True
        return self

    async def __aexit__(self, exc_type, exc, tb):
        return False

    async def connect(self):
        self.connected = True

    async def query(self, prompt, session_id: str = "default"):
        pass

    async def interrupt(self):
        pass

    async def receive_response(self):
        if False:
            yield None


def _dummy_actor() -> SessionActor:
    """Build an un-started actor with a no-op FakeSDKClient — for tests that
    never touch actor IO but need a non-None ``actor`` field."""
    dummy_client = FakeSDKClient()

    @asynccontextmanager
    async def _factory():
        async with dummy_client as c:
            yield c

    return SessionActor(client_factory=_factory, on_message=lambda msg: None)


class _FakeAllow:
    def __init__(self, updated_input):
        self.updated_input = updated_input


class _FakeDeny:
    def __init__(self, message, interrupt=False):
        self.message = message
        self.interrupt = interrupt


class TestSessionManagerMore:
    def test_managed_session_buffer_and_queue_overflow(self):
        managed = ManagedSession(session_id="s1", actor=None, buffer_max_size=2)
        managed.message_buffer = [
            {"type": "stream_event", "id": "a"},
            {"type": "assistant", "id": "b"},
        ]
        managed.add_message({"type": "assistant", "id": "c"})
        assert len(managed.message_buffer) == 2
        assert all(msg["id"] != "a" for msg in managed.message_buffer)

        queue = asyncio.Queue(maxsize=1)
        queue.put_nowait({"type": "stream_event"})
        managed.subscribers = {queue}
        managed.add_message({"type": "result", "uuid": "r1"})
        assert queue.get_nowait()["type"] == "result"

        # queue has only critical message; next critical should overflow and drop subscriber
        stale_queue = asyncio.Queue(maxsize=1)
        stale_queue.put_nowait({"type": "result"})
        managed.subscribers = {stale_queue}
        managed.add_message({"type": "assistant"})
        assert stale_queue.get_nowait()["type"] == "_queue_overflow"
        assert stale_queue not in managed.subscribers

    @pytest.mark.asyncio
    async def test_pending_question_lifecycle(self):
        managed = ManagedSession(session_id="s1", actor=None)
        pending = managed.add_pending_question({"type": "ask_user_question", "questions": []})
        assert pending.question_id
        assert managed.resolve_pending_question(pending.question_id, {"Q": "A"})
        assert await pending.answer_future == {"Q": "A"}
        assert not managed.resolve_pending_question("missing", {})

        pending2 = managed.add_pending_question({"type": "ask_user_question"})
        managed.cancel_pending_questions("closed")
        with pytest.raises(RuntimeError):
            await pending2.answer_future
        assert managed.get_pending_question_payloads() == []

    @pytest.mark.asyncio
    async def test_build_options_and_connect_paths(self, session_manager, meta_store, tmp_path, monkeypatch):
        async def _fake_env(_self):
            return {}

        monkeypatch.setattr(sm_mod.SessionManager, "_build_provider_env_overrides", _fake_env)

        with monkeypatch.context() as m:
            m.setattr(sm_mod, "SDK_AVAILABLE", False)
            with pytest.raises(RuntimeError):
                await session_manager._build_options("demo")

        projects_demo = tmp_path / "projects" / "demo"
        projects_demo.mkdir(parents=True)
        meta = await meta_store.create("demo", "sdk-build-opts")

        created_clients: list[_FakeClaudeClient] = []

        def _track_client(*, options):
            c = _FakeClaudeClient(options=options)
            created_clients.append(c)
            return c

        with monkeypatch.context() as m:
            m.setattr(sm_mod, "SDK_AVAILABLE", True)
            m.setattr(sm_mod, "ClaudeAgentOptions", _FakeOptions)
            m.setattr(sm_mod, "ClaudeSDKClient", _track_client)
            m.setattr(sm_mod, "HookMatcher", None)
            managed = await session_manager.get_or_connect(meta.id)
            # Let the actor enter the async-context (connect).
            await asyncio.sleep(0)
            assert created_clients and created_clients[0].connected
            assert managed is await session_manager.get_or_connect(meta.id)
            # Graceful teardown so the actor task doesn't leak.
            await session_manager.close_session(meta.id)

        assert await session_manager._keep_stream_open_hook({}, None, None) == {"continue_": True}

    @pytest.mark.asyncio
    async def test_resolve_project_scope_and_status_helpers(self, session_manager, tmp_path, meta_store):
        (tmp_path / "projects").mkdir(parents=True, exist_ok=True)
        with pytest.raises(ValueError):
            session_manager._resolve_project_cwd("../evil")

        assert await session_manager.get_status("missing") is None
        meta = await meta_store.create("demo", "sdk-resolve-status")
        assert await session_manager.get_status(meta.id) == "idle"

    @pytest.mark.asyncio
    async def test_send_message_and_interrupt_branches(self, session_manager, meta_store):
        from tests.fakes import build_managed_with_actor

        meta = await meta_store.create("demo", "sdk-send-msg")
        managed_running, actor_running, _ = await build_managed_with_actor(
            session_id=meta.id,
            project_name="demo",
            status="running",
        )
        session_manager.sessions[meta.id] = managed_running
        try:
            with pytest.raises(ValueError):
                await session_manager.send_message(meta.id, "blocked")
        finally:
            await session_manager.close_session(meta.id)

        # Now build a client whose query explodes — verify send_message flips status to error.
        client = FakeSDKClient()

        async def _boom(prompt, session_id: str = "default"):
            raise RuntimeError("query failed")

        client.query = _boom  # type: ignore[method-assign]

        @asynccontextmanager
        async def _boom_factory():
            async with client as c:
                yield c

        actor = SessionActor(client_factory=_boom_factory, on_message=lambda m: None)
        managed = ManagedSession(session_id=meta.id, actor=actor, status="idle", project_name="demo")
        await actor.start()
        session_manager.sessions[meta.id] = managed
        try:
            with pytest.raises(RuntimeError):
                await session_manager.send_message(meta.id, "hello")
            assert managed.status == "error"
            assert (await meta_store.get(meta.id)).status == "error"
        finally:
            # send_query failure raised inside actor → actor task already done.
            session_manager.sessions.pop(meta.id, None)

        with pytest.raises(FileNotFoundError):
            await session_manager.interrupt_session("missing")

        meta2 = await meta_store.create("demo", "sdk-interrupt-1")
        await meta_store.update_status(meta2.id, "running")
        assert await session_manager.interrupt_session(meta2.id) == "interrupted"
        assert (await meta_store.get(meta2.id)).status == "interrupted"

        meta3 = await meta_store.create("demo", "sdk-interrupt-2")
        assert await session_manager.interrupt_session(meta3.id) == "idle"

        managed_idle, _, _ = await build_managed_with_actor(
            session_id=meta3.id,
            project_name="demo",
            status="completed",
        )
        session_manager.sessions[meta3.id] = managed_idle
        try:
            assert await session_manager.interrupt_session(meta3.id) == "completed"
        finally:
            await session_manager.close_session(meta3.id)

    @pytest.mark.asyncio
    async def test_process_inbox_cancel_marks_interrupted_when_running(self, session_manager, meta_store):
        """Cancel on a running session → _mark_session_terminal("interrupted")."""
        meta = await meta_store.create("demo", "sdk-cancel-1")
        managed = ManagedSession(
            session_id=meta.id,
            actor=_dummy_actor(),
            status="running",
            project_name="demo",
        )
        session_manager.sessions[meta.id] = managed
        await meta_store.update_status(meta.id, "running")

        task = asyncio.create_task(session_manager._process_inbox(managed))
        await asyncio.sleep(0)  # let the coroutine start and block on inbox
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        assert managed.status == "interrupted"
        assert (await meta_store.get(meta.id)).status == "interrupted"

    @pytest.mark.asyncio
    async def test_can_use_tool_callback_branches(self, session_manager, monkeypatch):
        monkeypatch.setattr(sm_mod, "PermissionResultAllow", _FakeAllow)
        monkeypatch.setattr(sm_mod, "PermissionResultDeny", _FakeDeny)

        allow_cb = await session_manager._build_can_use_tool_callback("unknown-session")
        # Non-AskUserQuestion tools should be denied (whitelist fallback)
        result = await allow_cb("Read", {"x": 1}, None)
        assert isinstance(result, _FakeDeny)
        assert "未授权" in result.message
        # AskUserQuestion still handled
        result2 = await allow_cb("AskUserQuestion", {"questions": []}, None)
        assert result2.updated_input == {"questions": []}

        managed = ManagedSession(session_id="s1", actor=_dummy_actor(), status="running", project_name="demo")
        session_manager.sessions["s1"] = managed
        ask_cb = await session_manager._build_can_use_tool_callback("s1")

        task = asyncio.create_task(ask_cb("AskUserQuestion", {"questions": [{"question": "Q"}]}, None))
        await asyncio.sleep(0)
        assert managed.pending_questions
        managed.cancel_pending_questions("user interrupted")
        deny = await task
        assert deny.interrupt is True
        assert "user interrupted" in deny.message

    def test_misc_helpers_and_serialization(self, session_manager):
        assert sm_mod.SessionManager._extract_plain_user_content({"type": "user", "content": " hi "}) == "hi"
        assert (
            sm_mod.SessionManager._extract_plain_user_content(
                {"type": "user", "content": [{"type": "text", "text": " hello "}]}
            )
            == "hello"
        )
        assert sm_mod.SessionManager._extract_plain_user_content({"type": "assistant"}) is None

        msg = {}
        raw = SimpleNamespace(session_id="sdk-1")
        assert session_manager._extract_sdk_session_id(raw, msg) == "sdk-1"
        assert session_manager._extract_sdk_session_id(raw, {"sessionId": "sdk-2"}) == "sdk-2"

        status = session_manager._build_runtime_status_message("error", "s1")
        assert status["type"] == "runtime_status"
        assert status["is_error"] is True

        managed = ManagedSession(
            session_id="s1",
            actor=None,
            message_buffer=[{"type": "stream_event"}, {"type": "assistant"}, {"type": "custom"}],
        )
        session_manager._prune_transient_buffer(managed)
        assert managed.message_buffer == [{"type": "custom"}]
        managed.clear_buffer()
        assert managed.message_buffer == []

        assert session_manager._resolve_result_status({"subtype": "error_timeout"}) == "error"
        assert (
            session_manager._resolve_result_status(
                {"subtype": "success", "is_error": False},
                interrupt_requested=True,
            )
            == "completed"
        )

    @pytest.mark.asyncio
    async def test_buffer_snapshots_subscribe_and_shutdown(self, session_manager, meta_store):
        from tests.fakes import build_managed_with_actor

        assert await session_manager.get_message_buffer_snapshot("missing") == []
        assert session_manager.get_buffered_messages("missing") == []
        assert await session_manager.get_pending_questions_snapshot("missing") == []
        with pytest.raises(ValueError):
            await session_manager.answer_user_question("missing", "q", {"a": "b"})

        meta = await meta_store.create("demo", "sdk-buffer-snap")
        managed, actor, client = await build_managed_with_actor(
            session_id=meta.id,
            project_name="demo",
            status="running",
        )
        managed.message_buffer.append({"type": "assistant", "uuid": "a1"})
        session_manager.sessions[meta.id] = managed

        queue = await session_manager.subscribe(meta.id, replay_buffer=True)
        assert queue.get_nowait()["uuid"] == "a1"
        await session_manager.unsubscribe(meta.id, queue)
        assert queue not in managed.subscribers

        await session_manager.shutdown_gracefully(timeout=0.01)
        assert client.disconnected is True
        assert session_manager.sessions == {}

    @pytest.mark.asyncio
    async def test_file_access_hook_allows_read_within_project_root(self, tmp_path):
        """Hook allows Read within cwd and cwd-external (non-projects) paths;
        cross-project read is denied per new sandbox policy."""
        own_project = tmp_path / "projects" / "alpha"
        own_project.mkdir(parents=True)
        other_project = tmp_path / "projects" / "beta"
        other_project.mkdir(parents=True)
        docs_dir = tmp_path / "docs"
        docs_dir.mkdir(parents=True)

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        factory = async_sessionmaker(engine, expire_on_commit=False)
        meta_store = SessionMetaStore(session_factory=factory)

        mgr = sm_mod.SessionManager(
            project_root=tmp_path,
            data_dir=tmp_path,
            meta_store=meta_store,
        )

        hook = mgr._build_file_access_hook(own_project)

        # Read own project file — allowed (within project_cwd)
        result = await hook(
            {"tool_name": "Read", "tool_input": {"file_path": str(own_project / "script.json")}},
            None,
            None,
        )
        assert result.get("continue_") is True

        # Read other project file — denied (跨项目隔离)
        result = await hook(
            {"tool_name": "Read", "tool_input": {"file_path": str(other_project / "script.json")}},
            None,
            None,
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"

        # Read docs dir — allowed (cwd 外且不在 projects/ 下，作为参考资料)
        result = await hook(
            {"tool_name": "Read", "tool_input": {"file_path": str(docs_dir / "guide.md")}},
            None,
            None,
        )
        assert result.get("continue_") is True

        await engine.dispose()

    @pytest.mark.asyncio
    async def test_file_access_hook_blocks_write_to_readonly_dir(self, tmp_path):
        """Hook denies Write to lib/, allows own project."""
        own_project = tmp_path / "projects" / "alpha"
        own_project.mkdir(parents=True)
        lib_dir = tmp_path / "lib"
        lib_dir.mkdir(parents=True)

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        factory = async_sessionmaker(engine, expire_on_commit=False)
        meta_store = SessionMetaStore(session_factory=factory)

        mgr = sm_mod.SessionManager(
            project_root=tmp_path,
            data_dir=tmp_path,
            meta_store=meta_store,
        )

        hook = mgr._build_file_access_hook(own_project)

        # Write own project file — allowed
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": str(own_project / "output.txt")}},
            None,
            None,
        )
        assert result.get("continue_") is True

        # Write to lib/ (readonly) — denied
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": str(lib_dir / "hack.py")}},
            None,
            None,
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"

        await engine.dispose()

    @pytest.mark.asyncio
    async def test_file_access_hook_allows_bash_without_path_check(self, tmp_path):
        """Hook skips Bash (not in _PATH_TOOLS)."""
        own_project = tmp_path / "projects" / "alpha"
        own_project.mkdir(parents=True)

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        factory = async_sessionmaker(engine, expire_on_commit=False)
        meta_store = SessionMetaStore(session_factory=factory)

        mgr = sm_mod.SessionManager(
            project_root=tmp_path,
            data_dir=tmp_path,
            meta_store=meta_store,
        )

        hook = mgr._build_file_access_hook(own_project)

        # Bash — not a path tool, hook continues
        result = await hook(
            {"tool_name": "Bash", "tool_input": {"command": "ls /etc"}},
            None,
            None,
        )
        assert result.get("continue_") is True

        await engine.dispose()

    @pytest.mark.asyncio
    async def test_file_access_hook_blocks_write_non_whitelisted_ext(self, tmp_path):
        """Hook denies Write/Edit for forbidden code extensions in project dir.

        New policy: blacklist of code extensions (.py/.js/.ts/.tsx/.sh/.yaml/.yml/.toml)
        instead of data-file whitelist; non-code files (Makefile, .html, .csv) allowed.
        """
        own_project = tmp_path / "projects" / "alpha"
        own_project.mkdir(parents=True)

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        factory = async_sessionmaker(engine, expire_on_commit=False)
        meta_store = SessionMetaStore(session_factory=factory)

        mgr = sm_mod.SessionManager(
            project_root=tmp_path,
            data_dir=tmp_path,
            meta_store=meta_store,
        )

        hook = mgr._build_file_access_hook(own_project)

        # Write .py in project dir — denied (code extension)
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": str(own_project / "helper.py")}},
            None,
            None,
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"
        assert ".py" in result["hookSpecificOutput"]["permissionDecisionReason"]

        # Edit .sh in project dir — denied (code extension)
        result = await hook(
            {"tool_name": "Edit", "tool_input": {"file_path": str(own_project / "run.sh")}},
            None,
            None,
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"

        # Write .json — allowed
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": str(own_project / "project.json")}},
            None,
            None,
        )
        assert result.get("continue_") is True

        # Write .md — allowed
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": str(own_project / "notes.md")}},
            None,
            None,
        )
        assert result.get("continue_") is True

        # Write .txt — allowed
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": str(own_project / "episode.txt")}},
            None,
            None,
        )
        assert result.get("continue_") is True

        # Read .py — allowed (only write is restricted)
        result = await hook(
            {"tool_name": "Read", "tool_input": {"file_path": str(own_project / "helper.py")}},
            None,
            None,
        )
        assert result.get("continue_") is True

        # Write file without extension (e.g. Makefile) — allowed (not in code blacklist)
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": str(own_project / "Makefile")}},
            None,
            None,
        )
        assert result.get("continue_") is True

        # Write .JSON (uppercase) — allowed (case-insensitive check)
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": str(own_project / "data.JSON")}},
            None,
            None,
        )
        assert result.get("continue_") is True

        await engine.dispose()

    @pytest.mark.asyncio
    async def test_file_access_hook_allows_read_agent_profile(self, tmp_path):
        """Hook allows Read for agent_runtime_profile/ files."""
        own_project = tmp_path / "projects" / "alpha"
        own_project.mkdir(parents=True)
        profile_md = tmp_path / "agent_runtime_profile" / "CLAUDE.md"
        # conftest 的 _profile_env autouse fixture 已预创建 profile dir，
        # 这里用 exist_ok=True 容忍。CLAUDE.md 内容会被本测试覆写为有意义文本。
        profile_md.parent.mkdir(parents=True, exist_ok=True)
        profile_md.write_text("# Agent instructions")

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        factory = async_sessionmaker(engine, expire_on_commit=False)
        meta_store = SessionMetaStore(session_factory=factory)

        mgr = sm_mod.SessionManager(
            project_root=tmp_path,
            data_dir=tmp_path,
            meta_store=meta_store,
        )

        hook = mgr._build_file_access_hook(own_project)

        # Read agent_runtime_profile/CLAUDE.md — allowed (readonly dir)
        result = await hook(
            {"tool_name": "Read", "tool_input": {"file_path": str(profile_md)}},
            None,
            None,
        )
        assert result.get("continue_") is True

        await engine.dispose()

    async def _make_sdk_hook_env(self, tmp_path, monkeypatch):
        """Create a SessionManager + hook with SDK dir outside project_root."""
        app_root = tmp_path / "app"
        own_project = app_root / "projects" / "alpha"
        own_project.mkdir(parents=True)

        claude_home = tmp_path / "claude_home" / "projects"
        claude_home.mkdir(parents=True)

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        factory = async_sessionmaker(engine, expire_on_commit=False)
        meta_store = SessionMetaStore(session_factory=factory)

        mgr = sm_mod.SessionManager(
            project_root=app_root,
            data_dir=app_root,
            meta_store=meta_store,
        )
        monkeypatch.setattr(sm_mod.SessionManager, "_CLAUDE_PROJECTS_DIR", claude_home)

        hook = mgr._build_file_access_hook(own_project)
        return hook, own_project, claude_home, engine

    @pytest.mark.asyncio
    async def test_file_access_hook_allows_read_sdk_tool_results(self, tmp_path, monkeypatch):
        """Hook allows Read for SDK tool-results of the CURRENT project.

        New policy: cwd-external non-projects paths (含 SDK 目录) 默认放行；
        Write 仍受 cwd 内限制约束。
        """
        hook, own_project, claude_home, engine = await self._make_sdk_hook_env(
            tmp_path,
            monkeypatch,
        )

        encoded = sm_mod.SessionManager._encode_sdk_project_path(own_project)
        tool_results_dir = claude_home / encoded / "abc-session" / "tool-results"
        tool_results_dir.mkdir(parents=True)
        result_file = tool_results_dir / "toolu_01Abc.txt"
        result_file.write_text("full bash output here")

        # Read own project's SDK tool-results — allowed
        result = await hook(
            {"tool_name": "Read", "tool_input": {"file_path": str(result_file)}},
            None,
            None,
        )
        assert result.get("continue_") is True

        # Write to SDK tool-results — still denied (write tools only allow project_cwd)
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": str(result_file)}},
            None,
            None,
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"

        await engine.dispose()

    @pytest.mark.asyncio
    async def test_file_access_hook_denies_read_other_project_dir(self, tmp_path, monkeypatch):
        """Hook denies Read for ANOTHER project's directory under projects/.

        New policy: 跨项目隔离基于 project_root/projects/<other>/ 的物理位置，
        而非 SDK 编码路径。
        """
        hook, _, _, engine = await self._make_sdk_hook_env(
            tmp_path,
            monkeypatch,
        )

        other_project = tmp_path / "app" / "projects" / "beta"
        other_project.mkdir(parents=True)
        other_file = other_project / "secret.json"
        other_file.write_text("{}")

        # Read OTHER project directly — denied (cross-project isolation)
        result = await hook(
            {"tool_name": "Read", "tool_input": {"file_path": str(other_file)}},
            None,
            None,
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"

        await engine.dispose()

    @pytest.mark.asyncio
    async def test_file_access_hook_denies_write_outside_cwd(self, tmp_path, monkeypatch):
        """Hook denies Write to any path outside project_cwd.

        New policy: 写工具一律拒绝 cwd 外路径；Read 已放宽至 cwd 外非 projects/ 路径。
        """
        hook, _, _, engine = await self._make_sdk_hook_env(tmp_path, monkeypatch)

        # Write outside cwd — denied
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": "/tmp/escape.json"}},
            None,
            None,
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"

        await engine.dispose()

    @pytest.mark.asyncio
    async def test_file_access_hook_allows_read_sdk_task_output(self, tmp_path, monkeypatch):
        """Hook allows Read for SDK task output files under /tmp/claude-*.

        New policy: cwd-external non-projects 路径默认放行（含 SDK 后台任务输出），
        写工具仍受 cwd 内限制约束。
        """
        hook, _, _, engine = await self._make_sdk_hook_env(tmp_path, monkeypatch)

        # SDK task output path pattern: /tmp/claude-{N}/{encoded}/tasks/{id}.output
        task_output = "/tmp/claude-0/-app-projects-alpha-abc123/tasks/bdgaof0ba.output"
        result = await hook(
            {"tool_name": "Read", "tool_input": {"file_path": task_output}},
            None,
            None,
        )
        assert result.get("continue_") is True

        # Write to task output — denied (write tools only allow project_cwd)
        result = await hook(
            {"tool_name": "Write", "tool_input": {"file_path": task_output}},
            None,
            None,
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"

        await engine.dispose()


class TestJsonValidationHook:
    """Tests for the PreToolUse JSON validation hook."""

    def _make_manager(self, tmp_path):
        """Build a SessionManager with minimal fakes (SDK not required)."""
        from server.agent_runtime.session_manager import SessionManager
        from server.agent_runtime.session_store import SessionMetaStore

        return SessionManager(
            project_root=tmp_path,
            data_dir=tmp_path / "data",
            meta_store=SessionMetaStore(),
        )

    async def _call_hook(
        self,
        manager,
        tool_input: dict,
        tool_name: str = "Edit",
        project_cwd=None,
    ):
        """Helper: invoke the JSON validation hook callback directly."""
        from pathlib import Path

        hook_fn = manager._build_json_validation_hook(
            Path(project_cwd) if project_cwd else Path("/tmp"),
        )
        input_data = {
            "hook_event_name": "PreToolUse",
            "tool_name": tool_name,
            "tool_input": tool_input,
        }
        return await hook_fn(input_data, None, None)

    # --- Edit: valid replacement keeps JSON valid → allow ---

    async def test_edit_valid_replacement_returns_empty(self, tmp_path):
        """Edit that keeps JSON valid is allowed."""
        json_file = tmp_path / "ep.json"
        json_file.write_text('{"title": "old"}')
        manager = self._make_manager(tmp_path)

        result = await self._call_hook(
            manager,
            {
                "file_path": str(json_file),
                "old_string": '"old"',
                "new_string": '"new"',
            },
        )
        assert result == {}

    # --- Edit: replacement breaks JSON → deny ---

    async def test_edit_breaking_replacement_denies(self, tmp_path):
        """Edit that would produce invalid JSON is denied."""
        json_file = tmp_path / "ep.json"
        json_file.write_text('{"title": "old value"}')
        manager = self._make_manager(tmp_path)

        result = await self._call_hook(
            manager,
            {
                "file_path": str(json_file),
                "old_string": '"old value"',
                "new_string": '"has "quotes" inside"',  # unescaped quotes
            },
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"
        assert (
            "无效 JSON" in result["hookSpecificOutput"]["permissionDecisionReason"]
            or "JSON" in result["hookSpecificOutput"]["permissionDecisionReason"]
        )

    # --- Edit: replace_all ---

    async def test_edit_replace_all_breaking_denies(self, tmp_path):
        """Edit with replace_all that breaks JSON is denied."""
        json_file = tmp_path / "ep.json"
        json_file.write_text('{"a": "x", "b": "x"}')
        manager = self._make_manager(tmp_path)

        result = await self._call_hook(
            manager,
            {
                "file_path": str(json_file),
                "old_string": '"x"',
                "new_string": '"y",',  # trailing comma on last occurrence
                "replace_all": True,
            },
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"

    # --- Write: valid content → allow ---

    async def test_write_valid_json_returns_empty(self, tmp_path):
        """Write with valid JSON content is allowed."""
        manager = self._make_manager(tmp_path)
        result = await self._call_hook(
            manager,
            {
                "file_path": str(tmp_path / "new.json"),
                "content": '{"segments": []}',
            },
            tool_name="Write",
        )
        assert result == {}

    # --- Write: invalid content → deny ---

    async def test_write_invalid_json_denies(self, tmp_path):
        """Write with invalid JSON content is denied."""
        manager = self._make_manager(tmp_path)
        result = await self._call_hook(
            manager,
            {
                "file_path": str(tmp_path / "bad.json"),
                "content": '{"a": 1,,}',
            },
            tool_name="Write",
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"

    # --- Non-.json file → skip ---

    async def test_non_json_file_returns_empty(self, tmp_path):
        """Hook ignores non-.json files."""
        manager = self._make_manager(tmp_path)
        result = await self._call_hook(
            manager,
            {
                "file_path": str(tmp_path / "notes.md"),
                "content": "not json {{{{",
            },
            tool_name="Write",
        )
        assert result == {}

    # --- Edit: file not found → skip (let Edit handle the error) ---

    async def test_edit_missing_file_returns_empty(self, tmp_path):
        """Hook skips if the target file doesn't exist yet."""
        manager = self._make_manager(tmp_path)
        result = await self._call_hook(
            manager,
            {
                "file_path": str(tmp_path / "ghost.json"),
                "old_string": "x",
                "new_string": "y",
            },
        )
        assert result == {}

    # --- Non-Write/Edit tool → skip ---

    async def test_non_write_edit_tool_returns_empty(self, tmp_path):
        """Hook ignores tools other than Write/Edit."""
        manager = self._make_manager(tmp_path)
        result = await self._call_hook(
            manager,
            {
                "file_path": "/some/file.json",
            },
            tool_name="Read",
        )
        assert result == {}

    # --- Edit: old_string not in file → skip (Edit will fail on its own) ---

    async def test_edit_old_string_not_found_returns_empty(self, tmp_path):
        """Hook skips if old_string is not in the file."""
        json_file = tmp_path / "ep.json"
        json_file.write_text('{"title": "hello"}')
        manager = self._make_manager(tmp_path)

        result = await self._call_hook(
            manager,
            {
                "file_path": str(json_file),
                "old_string": "not found",
                "new_string": "replacement",
            },
        )
        assert result == {}

    # --- Edit: curly/smart quotes in new_string → deny ---

    async def test_edit_curly_quotes_in_new_string_denies(self, tmp_path):
        """Edit whose new_string contains curly quotes is denied even when
        old_string doesn't exactly match the file (Claude Code may normalise
        quotes internally, bypassing the hook's str.replace simulation)."""
        json_file = tmp_path / "ep.json"
        json_file.write_text('{"segment_break": true, "title": "test"}')
        manager = self._make_manager(tmp_path)

        # old_string uses curly quotes (won't match file via Python str `in`),
        # but new_string also has curly quotes → must be blocked.
        result = await self._call_hook(
            manager,
            {
                "file_path": str(json_file),
                "old_string": "\u201csegment_break\u201d: true",
                "new_string": "\u201csegment_break\u201d: false",
            },
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"
        assert "弯引号" in result["hookSpecificOutput"]["permissionDecisionReason"]

    async def test_edit_curly_quotes_old_only_is_allowed(self, tmp_path):
        """If only old_string has curly quotes but new_string is clean,
        don't block (edit will likely fail on its own)."""
        json_file = tmp_path / "ep.json"
        json_file.write_text('{"segment_break": true}')
        manager = self._make_manager(tmp_path)

        result = await self._call_hook(
            manager,
            {
                "file_path": str(json_file),
                "old_string": "\u201csegment_break\u201d: true",
                "new_string": '"segment_break": false',
            },
        )
        # old_string not in file → hook skips → allowed
        assert result == {}

    async def test_edit_curly_quotes_in_new_string_straight_old_denies(self, tmp_path):
        """Edit with straight-quote old_string that matches file but
        curly-quote new_string is denied via the early curly-quote check."""
        json_file = tmp_path / "ep.json"
        json_file.write_text('{"segment_break": true, "title": "test"}')
        manager = self._make_manager(tmp_path)

        result = await self._call_hook(
            manager,
            {
                "file_path": str(json_file),
                "old_string": '"segment_break": true',
                "new_string": "\u201csegment_break\u201d: false",
            },
        )
        assert result["hookSpecificOutput"]["permissionDecision"] == "deny"
        assert "弯引号" in result["hookSpecificOutput"]["permissionDecisionReason"]


class TestJsonPostValidationHook:
    """Tests for the PostToolUse JSON validation hook (safety net)."""

    def _make_manager(self, tmp_path):
        from server.agent_runtime.session_manager import SessionManager
        from server.agent_runtime.session_store import SessionMetaStore

        return SessionManager(
            project_root=tmp_path,
            data_dir=tmp_path / "data",
            meta_store=SessionMetaStore(),
        )

    async def _call_post_hook(
        self,
        manager,
        tool_input: dict,
        tool_name: str = "Edit",
        project_cwd=None,
        json_backups=None,
    ):
        from pathlib import Path

        hook_fn = manager._build_json_post_validation_hook(
            Path(project_cwd) if project_cwd else Path("/tmp"),
            json_backups if json_backups is not None else {},
        )
        input_data = {
            "hook_event_name": "PostToolUse",
            "tool_name": tool_name,
            "tool_input": tool_input,
        }
        return await hook_fn(input_data, "test-tool-use-id", None)

    # --- Valid JSON after edit → pass ---

    async def test_valid_json_after_edit_returns_empty(self, tmp_path):
        """PostToolUse returns empty when file is valid JSON after edit."""
        json_file = tmp_path / "ep.json"
        json_file.write_text('{"title": "new"}')
        manager = self._make_manager(tmp_path)

        result = await self._call_post_hook(
            manager,
            {
                "file_path": str(json_file),
            },
            project_cwd=str(tmp_path),
        )
        assert result == {}

    # --- Invalid JSON after edit with backup → restore + additionalContext ---

    async def test_invalid_json_restores_backup(self, tmp_path):
        """PostToolUse restores backup when file is invalid JSON after edit."""
        json_file = tmp_path / "ep.json"
        original_content = '{"title": "original"}'

        # Simulate: file was corrupted by edit
        json_file.write_text('{"title": "broken",,}')
        manager = self._make_manager(tmp_path)

        # Provide backup
        backups: dict = {
            "test-tool-use-id": (json_file, original_content),
        }

        result = await self._call_post_hook(
            manager,
            {"file_path": str(json_file)},
            project_cwd=str(tmp_path),
            json_backups=backups,
        )

        # Should report the issue via additionalContext
        assert "additionalContext" in result.get("hookSpecificOutput", {})
        assert "回滚" in result["hookSpecificOutput"]["additionalContext"]

        # File should be restored
        assert json_file.read_text() == original_content

        # Backup should be consumed (popped)
        assert "test-tool-use-id" not in backups

    # --- Invalid JSON after edit without backup → report only ---

    async def test_invalid_json_no_backup_reports_error(self, tmp_path):
        """PostToolUse reports error when file is corrupt and no backup exists."""
        json_file = tmp_path / "ep.json"
        json_file.write_text('{"broken":,}')
        manager = self._make_manager(tmp_path)

        result = await self._call_post_hook(
            manager,
            {"file_path": str(json_file)},
            project_cwd=str(tmp_path),
            json_backups={},
        )

        ctx = result.get("hookSpecificOutput", {}).get("additionalContext", "")
        assert "无法恢复" in ctx
        assert "回滚" not in ctx

    # --- Non-.json file → skip ---

    async def test_non_json_file_returns_empty(self, tmp_path):
        """PostToolUse ignores non-.json files."""
        manager = self._make_manager(tmp_path)
        result = await self._call_post_hook(
            manager,
            {
                "file_path": str(tmp_path / "notes.md"),
            },
            project_cwd=str(tmp_path),
        )
        assert result == {}

    # --- Backup is consumed even on success ---

    async def test_backup_consumed_on_success(self, tmp_path):
        """PostToolUse pops backup even when validation succeeds."""
        json_file = tmp_path / "ep.json"
        json_file.write_text('{"title": "ok"}')
        manager = self._make_manager(tmp_path)

        backups: dict = {
            "test-tool-use-id": (json_file, '{"title": "old"}'),
        }

        result = await self._call_post_hook(
            manager,
            {"file_path": str(json_file)},
            project_cwd=str(tmp_path),
            json_backups=backups,
        )

        assert result == {}
        # Backup should be consumed to prevent memory leaks
        assert "test-tool-use-id" not in backups


# --- ManagedSession 状态机（Session Actor 重构）-----------------------------


def _make_managed_for_state_test():
    """构造一个 ManagedSession 用于状态机测试，actor 字段用 None 占位。"""
    from server.agent_runtime.session_manager import ManagedSession

    return ManagedSession(
        session_id="test",
        actor=None,  # 状态机测试不触及 actor
        status="running",
        project_name="demo",
    )


def test_on_actor_message_result_does_not_change_status():
    """P1 race 防护：sync 回调不再改 status；由 _finalize_turn 统一设置。"""
    for subtype in ("success", "error_during_execution", "error_max_turns"):
        managed = _make_managed_for_state_test()
        managed.status = "running"
        managed._on_actor_message({"type": "result", "subtype": subtype})
        assert managed.status == "running", f"subtype={subtype}"


def test_on_actor_message_non_result_message_preserves_status():
    managed = _make_managed_for_state_test()
    managed.status = "running"
    managed._on_actor_message({"type": "assistant", "content": "hi"})
    assert managed.status == "running"


def test_on_actor_message_appends_to_buffer():
    managed = _make_managed_for_state_test()
    managed._on_actor_message({"type": "assistant", "content": "hi"})
    # add_message 负责 buffer + broadcast；这里只验 buffer
    buffered = list(managed.message_buffer)
    assert len(buffered) == 1
    assert buffered[0]["type"] == "assistant"


# --- ManagedSession 对 actor 的代理 -----------------------------------------


@pytest.mark.asyncio
async def test_send_query_sets_running_and_awaits_done():
    from server.agent_runtime.session_actor import SessionActor
    from server.agent_runtime.session_manager import ManagedSession
    from tests.fakes import FakeSDKClient

    client = FakeSDKClient(messages=[{"type": "result", "subtype": "success"}])
    managed_ref: list = []

    def on_message(msg):
        managed_ref[0]._on_actor_message(msg)

    actor = SessionActor(client_factory=lambda: client, on_message=on_message)
    managed = ManagedSession(session_id="t", actor=actor, status="idle", project_name="p")
    managed_ref.append(managed)

    await actor.start()
    await managed.send_query("hi")
    assert client.sent_queries == ["hi"]
    # send_query 在 sent 即返回；status 转 running 但不会自己变（由 _finalize_turn 设置，
    # 此单元测试没挂 _process_inbox）。完整链路的 status 转换由
    # test_session_manager_user_input 集成测试覆盖。
    assert managed.status == "running"

    # 收尾
    await managed.send_disconnect()


@pytest.mark.asyncio
async def test_send_query_raises_on_cmd_error():
    from server.agent_runtime.session_actor import SessionActor
    from server.agent_runtime.session_manager import ManagedSession

    class _Explode:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *e):
            return False

        async def query(self, *a, **k):
            raise RuntimeError("boom")

        async def interrupt(self):
            pass

        async def receive_response(self):
            if False:
                yield {}

    actor = SessionActor(client_factory=lambda: _Explode(), on_message=lambda m: None)
    managed = ManagedSession(session_id="t", actor=actor, status="idle", project_name="p")
    await actor.start()
    with pytest.raises(RuntimeError, match="boom"):
        await managed.send_query("hi")
    assert managed.status == "error"


@pytest.mark.asyncio
async def test_send_interrupt_is_idempotent_via_flag():
    """_interrupting 标志防止重入。"""
    from server.agent_runtime.session_actor import SessionActor, SessionCommand
    from server.agent_runtime.session_manager import ManagedSession
    from tests.fakes import FakeSDKClient

    client = FakeSDKClient(
        block_forever=True,
        interrupt_message={"type": "result", "subtype": "error_during_execution"},
    )
    actor = SessionActor(client_factory=lambda: client, on_message=lambda m: None)
    managed = ManagedSession(session_id="t", actor=actor, status="running", project_name="p")
    await actor.start()

    # 发一个 query 让 receive_response 开始
    q = SessionCommand(type="query", prompt="x")
    await actor.enqueue(q)
    await asyncio.sleep(0.05)

    # 并发两次 send_interrupt；第二次应走 _interrupting fast-return
    await asyncio.gather(managed.send_interrupt(), managed.send_interrupt())
    # client.interrupt 至少被调一次（具体次数视 asyncio 调度，允许 1 或 2）
    assert client.interrupted

    await q.done.wait()
    await managed.send_disconnect()


@pytest.mark.asyncio
async def test_send_disconnect_waits_actor_task_done():
    from server.agent_runtime.session_actor import SessionActor
    from server.agent_runtime.session_manager import ManagedSession
    from tests.fakes import FakeSDKClient

    client = FakeSDKClient()
    actor = SessionActor(client_factory=lambda: client, on_message=lambda m: None)
    managed = ManagedSession(session_id="t", actor=actor, status="idle", project_name="p")
    await actor.start()
    await managed.send_disconnect()
    assert managed.status == "closed"
    assert actor._task is not None and actor._task.done()
    assert client.disconnected
