"""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 dataclasses import replace from pathlib import Path from unittest.mock import patch from fastapi.testclient import TestClient from backend.api.server import ( AgentState, create_app, _bash_bg_manager, _clean_session_title, _derive_initial_session_title, _ensure_session_title, _generate_session_title, _read_session_title, _runtime_event_stage, _save_in_progress_session, _sanitize_stored_session_for_resume, ) from src.agent_types import AgentRunResult, AssistantTurn from src.bash_bg_store import BgTaskSpec 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_cancel_run_kills_active_background_tasks_for_session(self) -> None: with tempfile.TemporaryDirectory() as d: client, state = _build_client(Path(d)) spec = BgTaskSpec( task_id='bg_cancel_me', account_key='alice', session_id='thread-1', run_id='old-run', pid=12345, task_dir='/tmp/bg_cancel_me', output_path='/tmp/bg_cancel_me/output', pid_path='/tmp/bg_cancel_me/pid', exit_code_path='/tmp/bg_cancel_me/exit_code', command='sleep 3600', started_at=123.0, wait_for_completion=True, ) state.bash_bg_store.record_start(spec) killed: list[str] = [] async def fake_kill(_state: AgentState, row: dict[str, object]) -> None: killed.append(str(row['task_id'])) original = _bash_bg_manager._kill_remote_or_local _bash_bg_manager._kill_remote_or_local = fake_kill # type: ignore[method-assign] try: response = client.post( '/api/runs/cancel', json={'account_id': 'alice', 'session_id': 'thread-1'}, ) finally: _bash_bg_manager._kill_remote_or_local = original # type: ignore[method-assign] self.assertEqual(response.status_code, 200) payload = response.json() self.assertTrue(payload['cancelled']) self.assertEqual(payload['cancelled_bg_task_ids'], ['bg_cancel_me']) self.assertEqual(killed, ['bg_cancel_me']) row = state.bash_bg_store.get('bg_cancel_me') self.assertIsNotNone(row) self.assertEqual(row['status'], 'cancelled') def test_cancel_run_abandons_held_session_lock(self) -> None: with tempfile.TemporaryDirectory() as d: client, state = _build_client(Path(d)) record = state.run_manager.start('alice', 'thread-1') old_lock = state.run_lock_for('alice', 'thread-1') self.assertTrue(old_lock.acquire(blocking=False)) old_agent = state.agent_for('alice', 'thread-1') try: 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()) new_lock = state.run_lock_for('alice', 'thread-1') self.assertIsNot(new_lock, old_lock) self.assertTrue(new_lock.acquire(blocking=False)) new_lock.release() self.assertIsNot(state.agent_for('alice', 'thread-1'), old_agent) finally: old_lock.release() 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'}, ), display_messages=( {'role': 'user', 'content': 'hello'}, { 'role': 'assistant', 'content': '上一次任务已中断，后台没有正在执行的进程。', 'state': 'final', 'stop_reason': 'interrupted', 'metadata': {'kind': 'run_status', 'status': 'interrupted'}, }, { 'role': 'user', 'content': 'pending', 'metadata': {'status': 'running', 'placeholder': True}, }, ), 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'], ) self.assertEqual( [message['content'] for message in sanitized.display_messages], ['hello'], ) 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, state = _build_client(root) session_root = root / 'accounts' / 'alice' / 'sessions' session_id = 'same-id' agent = state.agent_for('alice', session_id) _save_in_progress_session( directory=session_root, agent=agent, session_id=session_id, prompt='帮我分析数据。', ) response = client.patch( f'/api/sessions/{session_id}', params={'account_id': 'alice'}, json={'title': ' 手动标题 '}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()['title'], '手动标题') metadata = load_agent_session( session_id, directory=session_root ).session_metadata or {} self.assertEqual(metadata['title'], '手动标题') self.assertEqual(metadata['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')) metadata = data['session_metadata'] self.assertEqual(metadata['title'], title) self.assertEqual(metadata['title_source'], 'first_message') def test_first_message_title_is_not_preserved_as_previous_title(self) -> None: with tempfile.TemporaryDirectory() as d: root = Path(d) _, state = _build_client(root) session_id = 'first-title-previous' directory = state.account_paths('alice')['sessions'] agent = state.agent_for('alice', session_id) _save_in_progress_session( directory=directory, agent=agent, session_id=session_id, prompt='帮我生成一批地图导航测试数据。', ) self.assertIsNone(_read_session_title(directory, session_id)) def test_invalid_llm_title_is_not_preserved_as_previous_title(self) -> None: with tempfile.TemporaryDirectory() as d: root = Path(d) _, state = _build_client(root) session_id = 'invalid-llm-title' directory = state.account_paths('alice')['sessions'] agent = state.agent_for('alice', session_id) _save_in_progress_session( directory=directory, agent=agent, session_id=session_id, prompt='帮我生成一批地图导航测试数据。', ) stored = load_agent_session(session_id, directory=directory) save_agent_session( replace( stored, session_metadata={ **(stored.session_metadata or {}), 'title': '**Generating short title', 'title_source': 'llm', }, ), directory=directory, ) self.assertIsNone(_read_session_title(directory, session_id)) 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='帮我生成地图和餐饮服务的边界数据。', ) 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() metadata = load_agent_session( session_id, directory=directory ).session_metadata or {} self.assertEqual(metadata['title'], '地图餐饮边界数据') self.assertEqual(metadata['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'] agent = state.agent_for('alice', session_id) _save_in_progress_session( directory=directory, agent=agent, session_id=session_id, prompt='帮我分析数据。', ) stored = load_agent_session(session_id, directory=directory) save_agent_session( replace( stored, session_metadata={ **(stored.session_metadata or {}), 'title': '手动标题', 'title_source': 'manual', }, ), directory=directory, ) 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() metadata = load_agent_session( session_id, directory=directory ).session_metadata or {} self.assertEqual(metadata['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_clean_session_title_rejects_reasoning_leak(self) -> None: self.assertIsNone(_clean_session_title('**Generating short title')) self.assertIsNone( _clean_session_title( 'reasoning\nGenerating short Chinese title' ) ) def test_clean_session_title_strips_markdown_and_prefix(self) -> None: self.assertEqual( _clean_session_title('**标题：地图导航测试数据。**'), '地图导航测试数据', ) 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()