"""Integration tests for the local web GUI FastAPI server.

These tests exercise the JSON endpoints against a real :class:`AgentState`
without booting uvicorn, using ``fastapi.testclient.TestClient``.  Slash
commands are dispatched locally inside :class:`LocalCodingAgent` and never
hit the network, so the chat endpoint can be exercised end-to-end against
``/help``.
"""

from __future__ import annotations

import tempfile
import unittest
import json
from pathlib import Path
from unittest.mock import patch

from fastapi.testclient import TestClient

from backend.api.server import (
    AgentState,
    create_app,
    _derive_initial_session_title,
    _ensure_session_title,
    _generate_session_title,
    _runtime_event_stage,
    _save_in_progress_session,
    _sanitize_stored_session_for_resume,
)
from src.agent_types import AgentRunResult, AssistantTurn
from src.session_store import StoredAgentSession, load_agent_session, save_agent_session


def _build_client(tmp: Path) -> tuple[TestClient, AgentState]:
    state = AgentState(
        cwd=tmp,
        model='test-model',
        base_url='http://127.0.0.1:8000/v1',
        api_key='local-token',
        timeout_seconds=120.0,
        allow_shell=False,
        allow_write=False,
        session_directory=tmp / 'sessions',
    )
    return TestClient(create_app(state)), state


class GuiServerTests(unittest.TestCase):
    def test_root_serves_html(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.get('/')
            self.assertEqual(response.status_code, 200)
            self.assertIn('Claw Code', response.text)

    def test_static_assets_served(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            self.assertEqual(client.get('/static/app.css').status_code, 200)
            self.assertEqual(client.get('/static/app.js').status_code, 200)

    def test_state_snapshot_round_trip(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.get('/api/state')
            self.assertEqual(response.status_code, 200)
            payload = response.json()
            self.assertEqual(payload['model'], 'test-model')
            self.assertFalse(payload['allow_shell'])

            updated = client.post(
                '/api/state',
                json={'allow_shell': True, 'model': 'other-model'},
            )
            self.assertEqual(updated.status_code, 200)
            data = updated.json()
            self.assertTrue(data['allow_shell'])
            self.assertEqual(data['model'], 'other-model')

    def test_state_update_accepts_provider_prefixed_model(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            updated = client.post(
                '/api/state',
                json={'model': 'azure_openai/gpt-4o-mini'},
            )
            self.assertEqual(updated.status_code, 200)
            self.assertEqual(updated.json()['model'], 'azure_openai/gpt-4o-mini')

    def test_state_snapshot_can_scope_to_account(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.get('/api/state', params={'account_id': 'alice'})
            self.assertEqual(response.status_code, 200)
            payload = response.json()
            self.assertEqual(payload['account_id'], 'alice')
            self.assertIn('/accounts/alice/sessions', payload['session_directory'])
            self.assertIn('/accounts/alice/sessions', payload['upload_directory'])
            self.assertIn('/accounts/alice/python/.venv', payload['python_env_directory'])
            self.assertTrue((Path(payload['python_env_directory']) / 'bin' / 'python').exists())

    def test_state_update_is_scoped_to_account(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            updated = client.post(
                '/api/state',
                json={'account_id': 'alice', 'allow_shell': True, 'model': 'alice-model'},
            )
            self.assertEqual(updated.status_code, 200)
            self.assertTrue(updated.json()['allow_shell'])
            self.assertEqual(updated.json()['model'], 'alice-model')

            alice = client.get('/api/state', params={'account_id': 'alice'}).json()
            bob = client.get('/api/state', params={'account_id': 'bob'}).json()
            default = client.get('/api/state').json()
            self.assertEqual(alice['model'], 'alice-model')
            self.assertTrue(alice['allow_shell'])
            self.assertEqual(bob['model'], 'test-model')
            self.assertFalse(bob['allow_shell'])
            self.assertEqual(default['model'], 'test-model')
            self.assertFalse(default['allow_shell'])

    def test_run_locks_are_scoped_to_account(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            _, state = _build_client(Path(d))
            self.assertIs(state.run_lock_for('alice'), state.run_lock_for('alice'))
            self.assertIsNot(state.run_lock_for('alice'), state.run_lock_for('bob'))
            self.assertIs(
                state.run_lock_for('alice', 's1'),
                state.run_lock_for('alice', 's1'),
            )
            self.assertIsNot(
                state.run_lock_for('alice', 's1'),
                state.run_lock_for('alice', 's2'),
            )

    def test_state_update_rejects_missing_cwd(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.post(
                '/api/state',
                json={'cwd': str(Path(d) / 'does-not-exist')},
            )
            self.assertEqual(response.status_code, 400)

    def test_slash_commands_listed(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.get('/api/slash-commands')
            self.assertEqual(response.status_code, 200)
            commands = response.json()
            self.assertTrue(commands)
            primaries = {entry['primary'] for entry in commands}
            self.assertIn('help', primaries)

    def test_skills_listed(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            skill_dir = root / 'skills' / 'project-skill'
            skill_dir.mkdir(parents=True)
            (skill_dir / 'SKILL.md').write_text(
                (
                    '---\n'
                    'name: project-skill\n'
                    'description: Project skill.\n'
                    '---\n'
                    'Project skill body.\n'
                ),
                encoding='utf-8',
            )
            client, _ = _build_client(root)
            response = client.get('/api/skills')
            self.assertEqual(response.status_code, 200)
            skills = response.json()
            self.assertTrue(skills)
            by_name = {entry['name']: entry for entry in skills}
            self.assertIn('project-skill', by_name)
            self.assertTrue(by_name['project-skill']['enabled'])
            self.assertTrue(by_name['project-skill']['configurable'])
            self.assertNotIn('simplify', by_name)

    def test_skill_preference_toggle_updates_prompt_filter(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            skill_dir = root / 'skills' / 'project-skill'
            skill_dir.mkdir(parents=True)
            (skill_dir / 'SKILL.md').write_text(
                (
                    '---\n'
                    'name: project-skill\n'
                    'description: Project skill.\n'
                    '---\n'
                    'Project skill body.\n'
                ),
                encoding='utf-8',
            )
            client, state = _build_client(root)

            response = client.patch(
                '/api/skills',
                json={'skill': 'project-skill', 'enabled': False},
            )
            self.assertEqual(response.status_code, 200)
            skill = {
                entry['name']: entry
                for entry in response.json()
            }['project-skill']
            self.assertFalse(skill['enabled'])
            self.assertNotIn('project-skill', state.agent_for().runtime_config.enabled_skill_names)
            self.assertIn('verify', state.agent_for().runtime_config.enabled_skill_names)

    def test_chat_runs_local_slash_command(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.post('/api/chat', json={'prompt': '/help'})
            self.assertEqual(response.status_code, 200)
            payload = response.json()
            self.assertEqual(payload['turns'], 0)
            self.assertEqual(payload['tool_calls'], 0)
            self.assertIn('slash commands', payload['final_output'].lower())
            self.assertIn('/help', payload['final_output'])
            self.assertIn('total_tokens', payload['usage'])

    def test_chat_uses_requested_session_directory(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            client, _ = _build_client(root)
            response = client.post(
                '/api/chat',
                json={
                    'prompt': '/help',
                    'account_id': 'alice',
                    'session_id': 'thread-1',
                },
            )
            self.assertEqual(response.status_code, 200)
            payload = response.json()
            self.assertIsNone(payload['session_id'])

    def test_chat_persists_in_progress_session_before_agent_finishes(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            client, state = _build_client(root)
            agent = state.agent_for('alice', 'pending-1')

            def fake_run(
                prompt: str,
                session_id: str | None = None,
                *,
                runtime_context: str | None = None,
                event_sink: object | None = None,
            ) -> AgentRunResult:
                stored = load_agent_session(
                    'pending-1',
                    directory=root / 'accounts' / 'alice' / 'sessions',
                )
                self.assertEqual(stored.session_id, 'pending-1')
                self.assertEqual(stored.budget_state.get('status'), 'running')
                self.assertEqual(stored.messages[-1]['role'], 'user')
                self.assertEqual(stored.messages[-1]['content'], 'hello')
                return AgentRunResult(
                    final_output='done',
                    turns=1,
                    tool_calls=0,
                    transcript=stored.messages
                    + ({'role': 'assistant', 'content': 'done'},),
                    session_id=session_id,
                )

            original_run = agent.run
            agent.run = fake_run  # type: ignore[method-assign]
            try:
                response = client.post(
                    '/api/chat',
                    json={
                        'prompt': 'hello',
                        'account_id': 'alice',
                        'session_id': ' pending-1 ',
                    },
                )
            finally:
                agent.run = original_run  # type: ignore[method-assign]

            self.assertEqual(response.status_code, 200)
            self.assertEqual(response.json()['session_id'], 'pending-1')

    def test_cancel_run_marks_latest_session_cancelled(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, state = _build_client(Path(d))
            record = state.run_manager.start('alice', 'thread-1')
            self.assertFalse(record.cancel_event.is_set())

            response = client.post(
                '/api/runs/cancel',
                json={'account_id': 'alice', 'session_id': 'thread-1'},
            )

            self.assertEqual(response.status_code, 200)
            self.assertTrue(response.json()['cancelled'])
            self.assertTrue(record.cancel_event.is_set())
            latest = client.get(
                '/api/runs/latest',
                params={'account_id': 'alice', 'session_id': 'thread-1'},
            )
            self.assertEqual(latest.status_code, 200)
            self.assertEqual(latest.json()['status'], 'cancelled')

    def test_latest_run_prefers_running_over_queued_for_session(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, state = _build_client(Path(d))
            first = state.run_manager.start('alice', 'thread-1')
            state.run_manager.update(first.run_id, status='running', stage='first stage')
            second = state.run_manager.start('alice', 'thread-1')
            self.assertEqual(second.status, 'queued')

            latest = client.get(
                '/api/runs/latest',
                params={'account_id': 'alice', 'session_id': 'thread-1'},
            )

            self.assertEqual(latest.status_code, 200)
            payload = latest.json()
            self.assertEqual(payload['run_id'], first.run_id)
            self.assertEqual(payload['status'], 'running')
            self.assertEqual(payload['current_stage'], 'first stage')

    def test_latest_run_includes_runtime_event_history(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, state = _build_client(Path(d))
            record = state.run_manager.start('alice', 'thread-1')
            state.run_manager.update(record.run_id, status='running')
            state.run_manager.record_event(
                record.run_id,
                {
                    'type': 'tool_start',
                    'tool_name': 'python_exec',
                    'tool_call_id': 'call-1',
                    'arguments': {'code': 'print(1)'},
                    'assistant_content': '我用 Python 分析一下。',
                },
            )
            state.run_manager.record_event(
                record.run_id,
                {
                    'type': 'tool_result',
                    'tool_name': 'python_exec',
                    'tool_call_id': 'call-1',
                    'ok': True,
                    'metadata': {'output_preview': 'exit_code=0'},
                },
            )

            latest = client.get(
                '/api/runs/latest',
                params={'account_id': 'alice', 'session_id': 'thread-1'},
            )

            self.assertEqual(latest.status_code, 200)
            events = latest.json()['events']
            self.assertEqual([event['type'] for event in events], ['tool_start', 'tool_result'])
            self.assertEqual(events[0]['tool_name'], 'python_exec')
            self.assertEqual(events[1]['metadata']['output_preview'], 'exit_code=0')

    def test_cancel_run_marks_streaming_session_tail_cancelled(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            client, state = _build_client(root)
            session_root = root / 'accounts' / 'alice' / 'sessions'
            save_agent_session(
                StoredAgentSession(
                    session_id='thread-1',
                    model_config={'model': 'test-model'},
                    runtime_config={'cwd': str(root)},
                    system_prompt_parts=('system prompt',),
                    user_context={},
                    system_context={},
                    messages=(
                        {'role': 'user', 'content': 'hello'},
                        {
                            'role': 'assistant',
                            'content': '我先看看文件',
                            'tool_calls': [
                                {
                                    'id': 'call-1',
                                    'function': {'name': 'list_dir', 'arguments': '{}'},
                                }
                            ],
                        },
                        {
                            'role': 'tool',
                            'content': '',
                            'tool_call_id': 'call-1',
                            'state': 'streaming',
                            'metadata': {'phase': 'starting'},
                        },
                    ),
                    turns=1,
                    tool_calls=1,
                    usage={},
                    total_cost_usd=0.0,
                    file_history=(),
                    budget_state={'status': 'running'},
                    plugin_state={},
                ),
                directory=session_root,
            )
            state.run_manager.start('alice', 'thread-1')

            response = client.post(
                '/api/runs/cancel',
                json={'account_id': 'alice', 'session_id': 'thread-1'},
            )

            self.assertEqual(response.status_code, 200)
            self.assertTrue(response.json()['cancelled'])
            stored = load_agent_session('thread-1', directory=session_root)
            self.assertEqual(stored.messages[-1]['role'], 'assistant')
            self.assertEqual(stored.messages[-1]['stop_reason'], 'cancelled')
            self.assertEqual(
                stored.messages[-1]['metadata']['status'],
                'cancelled',
            )
            self.assertNotEqual(stored.messages[-1].get('state'), 'streaming')

    def test_latest_run_reports_interrupted_for_orphaned_streaming_session(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            client, _ = _build_client(root)
            session_root = root / 'accounts' / 'alice' / 'sessions'
            save_agent_session(
                StoredAgentSession(
                    session_id='thread-1',
                    model_config={'model': 'test-model'},
                    runtime_config={'cwd': str(root)},
                    system_prompt_parts=('system prompt',),
                    user_context={},
                    system_context={},
                    messages=(
                        {'role': 'user', 'content': 'hello'},
                        {
                            'role': 'tool',
                            'content': '',
                            'tool_call_id': 'call-1',
                            'state': 'streaming',
                            'metadata': {'phase': 'starting'},
                        },
                    ),
                    turns=1,
                    tool_calls=1,
                    usage={},
                    total_cost_usd=0.0,
                    file_history=(),
                    budget_state={'status': 'running'},
                    plugin_state={},
                ),
                directory=session_root,
            )

            response = client.get(
                '/api/runs/latest',
                params={'account_id': 'alice', 'session_id': 'thread-1'},
            )

            self.assertEqual(response.status_code, 200)
            self.assertEqual(response.json()['status'], 'interrupted')

    def test_resume_sanitizer_removes_ui_run_status_messages(self) -> None:
        stored = StoredAgentSession(
            session_id='thread-1',
            model_config={'model': 'test-model'},
            runtime_config={},
            system_prompt_parts=('system prompt',),
            user_context={},
            system_context={},
            messages=(
                {'role': 'user', 'content': 'hello'},
                {
                    'role': 'assistant',
                    'content': '上一次任务已中断，后台没有正在执行的进程。',
                    'state': 'final',
                    'stop_reason': 'interrupted',
                    'metadata': {'kind': 'run_status', 'status': 'interrupted'},
                },
                {'role': 'user', 'content': 'continue'},
            ),
            turns=1,
            tool_calls=0,
            usage={},
            total_cost_usd=0.0,
            file_history=(),
            budget_state={'status': 'interrupted'},
            plugin_state={},
        )

        sanitized = _sanitize_stored_session_for_resume(stored)

        self.assertEqual(
            [message['content'] for message in sanitized.messages],
            ['hello', 'continue'],
        )

    def test_runtime_event_stage_names_long_running_work(self) -> None:
        self.assertEqual(
            _runtime_event_stage({'type': 'tool_start', 'tool_name': 'python_exec'}),
            '调用工具 python_exec',
        )
        self.assertEqual(
            _runtime_event_stage({'type': 'final_text_start'}),
            '正在整理回复',
        )
        self.assertEqual(_runtime_event_stage({'type': 'server_heartbeat'}), '')

    def test_chat_uses_account_scoped_model_config(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            client.post(
                '/api/state',
                json={'account_id': 'alice', 'model': 'alice-model'},
            )

            response = client.post(
                '/api/chat',
                json={'prompt': '/status', 'account_id': 'alice'},
            )
            self.assertEqual(response.status_code, 200)
            self.assertIn('- Model: alice-model', response.json()['final_output'])

            bob = client.post(
                '/api/chat',
                json={'prompt': '/status', 'account_id': 'bob'},
            )
            self.assertEqual(bob.status_code, 200)
            self.assertIn('- Model: test-model', bob.json()['final_output'])

    def test_agent_session_store_uses_session_subdirectory(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            stored = StoredAgentSession(
                session_id='thread-1',
                model_config={'model': 'test-model'},
                runtime_config={'cwd': str(root)},
                system_prompt_parts=(),
                user_context={},
                system_context={},
                messages=({'role': 'user', 'content': 'hello'},),
                turns=1,
                tool_calls=0,
                usage={},
                total_cost_usd=0.0,
                file_history=(),
                budget_state={},
                plugin_state={},
                scratchpad_directory=str(root / 'sessions' / 'thread-1' / 'scratchpad'),
            )
            path = save_agent_session(stored, directory=root / 'sessions')
            self.assertEqual(path, root / 'sessions' / 'thread-1' / 'session.json')
            loaded = load_agent_session('thread-1', directory=root / 'sessions')
            self.assertEqual(loaded.session_id, 'thread-1')

    def test_context_budget_reports_stored_session_prompt_size(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            client, _ = _build_client(root)
            session_root = root / 'accounts' / 'alice' / 'sessions'
            stored = StoredAgentSession(
                session_id='thread-1',
                model_config={'model': 'test-model'},
                runtime_config={'cwd': str(root)},
                system_prompt_parts=('system prompt',),
                user_context={},
                system_context={},
                messages=({'role': 'user', 'content': 'hello'},),
                turns=1,
                tool_calls=0,
                usage={},
                total_cost_usd=0.0,
                file_history=(),
                budget_state={},
                plugin_state={},
            )
            save_agent_session(stored, directory=session_root)

            response = client.get(
                '/api/context-budget',
                params={'account_id': 'alice', 'session_id': 'thread-1'},
            )
            self.assertEqual(response.status_code, 200)
            payload = response.json()
            self.assertEqual(payload['model'], 'test-model')
            self.assertEqual(payload['context_window_tokens'], 128000)
            self.assertGreater(payload['projected_input_tokens'], 0)
            self.assertGreater(payload['soft_input_limit_tokens'], 0)

    def test_chat_rejects_blank_prompt(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.post('/api/chat', json={'prompt': '   '})
            self.assertEqual(response.status_code, 400)

    def test_chat_resume_unknown_session_returns_404(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.post(
                '/api/chat',
                json={'prompt': '/help', 'resume_session_id': 'missing'},
            )
            self.assertEqual(response.status_code, 404)

    def test_sessions_list_empty_when_directory_absent(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.get('/api/sessions')
            self.assertEqual(response.status_code, 200)
            self.assertEqual(response.json(), [])

    def test_sessions_are_isolated_by_account(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            client, _ = _build_client(root)
            alice_sessions = root / 'accounts' / 'alice' / 'sessions'
            alice_sessions.mkdir(parents=True)
            (alice_sessions / 's1.json').write_text(
                json.dumps(
                    {
                        'session_id': 's1',
                        'turns': 1,
                        'tool_calls': 0,
                        'messages': [{'role': 'user', 'content': 'alice only'}],
                    }
                ),
                encoding='utf-8',
            )

            alice = client.get('/api/sessions', params={'account_id': 'alice'})
            bob = client.get('/api/sessions', params={'account_id': 'bob'})
            self.assertEqual(alice.status_code, 200)
            self.assertEqual(bob.status_code, 200)
            self.assertEqual(alice.json()[0]['session_id'], 's1')
            self.assertEqual(bob.json(), [])

    def test_session_list_deduplicates_legacy_and_nested_files(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            client, _ = _build_client(root)
            session_root = root / 'accounts' / 'alice' / 'sessions'
            session_root.mkdir(parents=True)
            legacy = {
                'session_id': 'same-id',
                'turns': 1,
                'tool_calls': 0,
                'messages': [{'role': 'user', 'content': 'legacy'}],
            }
            nested = {
                'session_id': 'same-id',
                'turns': 2,
                'tool_calls': 1,
                'title': '自动生成标题',
                'messages': [{'role': 'user', 'content': 'nested'}],
            }
            (session_root / 'same-id.json').write_text(
                json.dumps(legacy),
                encoding='utf-8',
            )
            (session_root / 'same-id').mkdir()
            (session_root / 'same-id' / 'session.json').write_text(
                json.dumps(nested),
                encoding='utf-8',
            )

            response = client.get('/api/sessions', params={'account_id': 'alice'})
            self.assertEqual(response.status_code, 200)
            payload = response.json()
            self.assertEqual(len(payload), 1)
            self.assertEqual(payload[0]['session_id'], 'same-id')
            self.assertEqual(payload[0]['preview'], '自动生成标题')

    def test_session_detail_404_when_missing(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.get('/api/sessions/nope')
            self.assertEqual(response.status_code, 404)

    def test_delete_session_removes_nested_and_legacy_files(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            client, _ = _build_client(root)
            session_root = root / 'accounts' / 'alice' / 'sessions'
            nested = session_root / 'same-id'
            nested.mkdir(parents=True)
            (nested / 'session.json').write_text('{}', encoding='utf-8')
            legacy = session_root / 'same-id.json'
            legacy.write_text('{}', encoding='utf-8')

            response = client.delete(
                '/api/sessions/same-id',
                params={'account_id': 'alice'},
            )
            self.assertEqual(response.status_code, 200)
            self.assertTrue(response.json()['deleted'])
            self.assertFalse(nested.exists())
            self.assertFalse(legacy.exists())

            missing = client.delete(
                '/api/sessions/same-id',
                params={'account_id': 'alice'},
            )
            self.assertEqual(missing.status_code, 404)

    def test_update_session_title(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            client, _ = _build_client(root)
            session_root = root / 'accounts' / 'alice' / 'sessions'
            nested = session_root / 'same-id'
            nested.mkdir(parents=True)
            session_file = nested / 'session.json'
            session_file.write_text(
                json.dumps({'session_id': 'same-id', 'messages': []}),
                encoding='utf-8',
            )

            response = client.patch(
                '/api/sessions/same-id',
                params={'account_id': 'alice'},
                json={'title': ' 手动标题 '},
            )
            self.assertEqual(response.status_code, 200)
            self.assertEqual(response.json()['title'], '手动标题')
            data = json.loads(session_file.read_text(encoding='utf-8'))
            self.assertEqual(data['title'], '手动标题')
            self.assertEqual(data['title_source'], 'manual')

    def test_in_progress_session_writes_first_message_title(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            _, state = _build_client(root)
            session_id = 'first-title'
            directory = state.account_paths('alice')['sessions']
            agent = state.agent_for('alice', session_id)

            title = _save_in_progress_session(
                directory=directory,
                agent=agent,
                session_id=session_id,
                prompt='帮我分析一下线上 router session 数据，然后统计 device 分布。',
            )

            session_file = directory / session_id / 'session.json'
            data = json.loads(session_file.read_text(encoding='utf-8'))
            self.assertTrue(title.startswith('帮我分析一下线上 router'))
            self.assertEqual(data['title'], title)
            self.assertEqual(data['title_source'], 'first_message')

    def test_session_title_refines_first_message_title_after_run(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            _, state = _build_client(root)
            session_id = 'refine-title'
            directory = state.account_paths('alice')['sessions']
            agent = state.agent_for('alice', session_id)
            fallback_title = _save_in_progress_session(
                directory=directory,
                agent=agent,
                session_id=session_id,
                prompt='帮我生成地图和餐饮服务的边界数据。',
            )
            session_file = directory / session_id / 'session.json'
            data = json.loads(session_file.read_text(encoding='utf-8'))
            data['messages'] = [
                {'role': 'user', 'content': '帮我生成地图和餐饮服务的边界数据。'}
            ]
            session_file.write_text(json.dumps(data), encoding='utf-8')

            with patch(
                'backend.api.server._generate_session_title',
                return_value='地图餐饮边界数据',
            ) as generate:
                _ensure_session_title(
                    directory,
                    session_id,
                    model='test-model',
                    base_url='http://127.0.0.1:8000/v1',
                    api_key='local-token',
                    fallback_title=fallback_title,
                )

            generate.assert_called_once()
            updated = json.loads(session_file.read_text(encoding='utf-8'))
            self.assertEqual(updated['title'], '地图餐饮边界数据')
            self.assertEqual(updated['title_source'], 'llm')

    def test_session_title_preserves_existing_manual_title(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            root = Path(d)
            _, state = _build_client(root)
            session_id = 'manual-preserved'
            directory = state.account_paths('alice')['sessions']
            session_dir = directory / session_id
            session_dir.mkdir(parents=True)
            session_file = session_dir / 'session.json'
            session_file.write_text(
                json.dumps(
                    {
                        'session_id': session_id,
                        'title': '手动标题',
                        'title_source': 'manual',
                        'messages': [
                            {'role': 'user', 'content': '帮我分析数据。'},
                        ],
                    }
                ),
                encoding='utf-8',
            )

            with patch('backend.api.server._generate_session_title') as generate:
                _ensure_session_title(
                    directory,
                    session_id,
                    model='test-model',
                    base_url='http://127.0.0.1:8000/v1',
                    api_key='local-token',
                    previous_title='手动标题',
                )

            generate.assert_not_called()
            updated = json.loads(session_file.read_text(encoding='utf-8'))
            self.assertEqual(updated['title'], '手动标题')

    def test_derive_initial_session_title_strips_session_context(self) -> None:
        title = _derive_initial_session_title(
            '请帮我生成线上挖掘数据。\n\n[当前会话目录]\n/root/session'
        )
        self.assertEqual(title, '请帮我生成线上挖掘数据')

    def test_generate_session_title_uses_model_compat_client(self) -> None:
        with patch('backend.api.server.OpenAICompatClient') as client_cls:
            client = client_cls.return_value
            client.complete.return_value = AssistantTurn(content='“模型兼容标题”')

            title = _generate_session_title(
                ['帮我分析一下线上 session 数据'],
                model='ppio/pa/claude-opus-4-7',
                base_url='http://model.mify.ai.srv/v1',
                api_key='local-token',
            )

            self.assertEqual(title, '模型兼容标题')
            config = client_cls.call_args.args[0]
            self.assertEqual(config.model, 'ppio/pa/claude-opus-4-7')
            self.assertEqual(config.base_url, 'http://model.mify.ai.srv/v1')
            client.complete.assert_called_once()

    def test_clear_runtime_state(self) -> None:
        with tempfile.TemporaryDirectory() as d:
            client, _ = _build_client(Path(d))
            response = client.post('/api/clear')
            self.assertEqual(response.status_code, 200)
            self.assertIn('model', response.json())


if __name__ == '__main__':
    unittest.main()
