"""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 fastapi.testclient import TestClient from backend.api.server import ( AgentState, create_app, _runtime_event_stage, _sanitize_stored_session_for_resume, ) from src.agent_types import AgentRunResult 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', 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_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: client, _ = _build_client(Path(d)) response = client.get('/api/skills') self.assertEqual(response.status_code, 200) skills = response.json() self.assertTrue(skills) names = {entry['name'] for entry in skills} self.assertIn('simplify', 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_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()