From f96f663a80a98b6e26df8c497fd2e4d15f5db4a8 Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Thu, 8 Oct 2026 13:24:37 +0800 Subject: [PATCH] feat(assistant): add streaming background sessions and refresh workspace UI --- docs/API_KEY_AUTH.md | 39 + .../api/http/controller/groups/assistant.py | 71 +- src/langbot/pkg/api/http/service/assistant.py | 237 +++- src/langbot/pkg/api/http/service/space.py | 18 +- .../api/service/test_space_service.py | 42 + .../api/test_management_assistant.py | 249 +++- web/package.json | 1 + web/pnpm-lock.yaml | 11 + .../home/components/AssistantToolResult.tsx | 61 +- .../home/components/WorkspaceAssistant.tsx | 1178 ++++++++++------- web/src/app/home/components/assistant-dock.ts | 89 -- .../dynamic-form/DynamicFormItemComponent.tsx | 10 +- .../components/workspace-assistant.module.css | 238 ++++ web/src/app/infra/http/BackendClient.ts | 52 + web/src/i18n/locales/en-US.ts | 21 +- web/src/i18n/locales/es-ES.ts | 22 +- web/src/i18n/locales/ja-JP.ts | 22 +- web/src/i18n/locales/ru-RU.ts | 22 +- web/src/i18n/locales/th-TH.ts | 20 +- web/src/i18n/locales/vi-VN.ts | 21 +- web/src/i18n/locales/zh-Hans.ts | 19 +- web/src/i18n/locales/zh-Hant.ts | 19 +- web/tests/e2e/workspace-assistant.spec.ts | 540 ++++++++ web/tests/unit/assistant-dock-rail.test.mjs | 175 --- 24 files changed, 2351 insertions(+), 826 deletions(-) delete mode 100644 web/src/app/home/components/assistant-dock.ts create mode 100644 web/src/app/home/components/workspace-assistant.module.css create mode 100644 web/tests/e2e/workspace-assistant.spec.ts delete mode 100644 web/tests/unit/assistant-dock-rail.test.mjs diff --git a/docs/API_KEY_AUTH.md b/docs/API_KEY_AUTH.md index 3eac9eb10..256da5f45 100644 --- a/docs/API_KEY_AUTH.md +++ b/docs/API_KEY_AUTH.md @@ -360,3 +360,42 @@ curl -X POST \ - API-key-enabled endpoints use the same resource shapes as the web UI - No need to learn different API paths - use the existing API documentation with API key authentication - API keys never select a Workspace from a request header; their persisted binding is authoritative + +## Workspace assistant browser session + +The Workspace assistant endpoints under `/api/v1/assistant` are account-session +features, not API-key/MCP automation tools. They require a Workspace-scoped user +token and `runtime.operate` for recommendation and turns. Conversation reads +also enforce account ownership. + +- `GET /recommended-model` resolves and validates the recommended tool-capable + model when the panel opens. An explicit `model_uuid` in a turn always takes + precedence; it never silently falls back if unavailable. +- `POST /conversations/{id}/turn/stream` accepts the same revision, text/model, + or approval decision as `/turn`, and returns NDJSON. `snapshot` frames contain + the public conversation; `text` frames contain the cumulative text for the + current model invocation; `phase` frames identify the model round or running + tool. A snapshot replaces transient text/tool progress. `completed` contains + the final conversation (including approval or failed states); `error` is a + transport/admission failure. Blank lines are heartbeats. +- `GET /conversations` lists the latest 100 conversations owned by the current + account in this Workspace. `GET /conversations/{id}` includes transient + `progress` (cumulative text, round and running tool) while execution is active. +- Disconnecting a streaming response only detaches the browser. The backend owns + the task until completion, approval, timeout or explicit stop. Slow or absent + consumers never block execution; reconnect using the conversation snapshot. +- `POST /conversations/{id}/stop` accepts `{ "revision": N }` for the active turn. + It enforces ownership and revision, cancels execution and records `stopped`. + Stopping cannot roll back tool effects and clients must never replay a write + automatically. Writes still require confirmation of saved arguments. +- Background execution lives in the current backend process, with at most four + simultaneous turns and the existing 120-second turn limit. This is not a durable + job queue: backend restarts do not resume interrupted tasks. + +The original JSON `/turn` endpoint remains compatible for existing callers. + +The assistant recommendation uses Space `/api/v1/models/assistant/recommendation`, +which selects only from the assistant pool using fresh successful health checks. +An unavailable recommendation returns 503 without falling back to general chat rankings. +The browser checks Workspace model providers before mounting the assistant; an empty +provider list shows Account sign-in and model settings actions instead of a composer. diff --git a/src/langbot/pkg/api/http/controller/groups/assistant.py b/src/langbot/pkg/api/http/controller/groups/assistant.py index f902c6f44..7684db2e1 100644 --- a/src/langbot/pkg/api/http/controller/groups/assistant.py +++ b/src/langbot/pkg/api/http/controller/groups/assistant.py @@ -1,5 +1,8 @@ """Web-session-only assistant endpoints; resource tools use the existing service layer.""" +import asyncio +import json + import quart from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, ValidationError @@ -10,6 +13,11 @@ from ...context import RequestContext from ...service.assistant import AssistantError, AssistantService +class StopInput(BaseModel): + model_config = ConfigDict(extra='forbid') + revision: int = Field(ge=0, strict=True) + + class TurnInput(BaseModel): model_config = ConfigDict(extra='forbid') revision: int = Field(ge=0, strict=True) @@ -23,6 +31,27 @@ class AssistantRouterGroup(group.RouterGroup): async def initialize(self): service = AssistantService(self.ap) + @self.route('/conversations', methods=['GET'], permission=Permission.RESOURCE_VIEW) + async def list_conversations(request_context: RequestContext): + return self.success(data={'conversations': await service.list_conversations(request_context)}) + + @self.route('/conversations//stop', methods=['POST'], permission=Permission.RUNTIME_OPERATE) + async def stop(conversation_id: str, request_context: RequestContext): + try: + body = StopInput.model_validate(await quart.request.get_json()) + return self.success(data=await service.stop(request_context, conversation_id, body.revision)) + except ValidationError: + return self.http_status(400, 'invalid_input', 'Invalid stop request') + except AssistantError as exc: + return self.http_status(exc.status, exc.code, exc.code) + + @self.route('/recommended-model', methods=['GET'], permission=Permission.RUNTIME_OPERATE) + async def recommended_model(request_context: RequestContext): + try: + return self.success(data=await service.recommended_model(request_context)) + except AssistantError as exc: + return self.http_status(exc.status, exc.code, exc.code) + @self.route('/conversations', methods=['POST'], permission=Permission.RUNTIME_OPERATE) async def create(request_context: RequestContext): try: @@ -33,7 +62,7 @@ class AssistantRouterGroup(group.RouterGroup): @self.route('/conversations/', methods=['GET'], permission=Permission.RESOURCE_VIEW) async def get(conversation_id: str, request_context: RequestContext): try: - return self.success(data=service.public_view(await service.get(request_context, conversation_id))) + return self.success(data=await service.progress(request_context, conversation_id)) except AssistantError as exc: return self.http_status(exc.status, exc.code, exc.code) @@ -56,3 +85,43 @@ class AssistantRouterGroup(group.RouterGroup): return self.http_status(400, 'invalid_input', 'Invalid assistant request') except AssistantError as exc: return self.http_status(exc.status, exc.code, exc.code) + + @self.route( + '/conversations//turn/stream', methods=['POST'], permission=Permission.RUNTIME_OPERATE + ) + async def stream_turn(conversation_id: str, request_context: RequestContext): + try: + body = TurnInput.model_validate(await quart.request.get_json()) + if (body.text is None) == (body.approved is None) or (body.text is not None and not body.text.strip()): + return self.http_status(400, 'invalid_input', 'Provide text or an approval decision') + await service.get(request_context, conversation_id) + except ValidationError: + return self.http_status(400, 'invalid_input', 'Invalid assistant request') + except AssistantError as exc: + return self.http_status(exc.status, exc.code, exc.code) + try: + return assistant_stream_response(service, request_context, conversation_id, body) + except AssistantError as exc: + return self.http_status(exc.status, exc.code, exc.code) + + +def assistant_stream_response(service, context, conversation_id, body): + """Detach execution from the subscriber; only an explicit stop cancels a run.""" + queue = service.start_run(context, conversation_id, body) + + async def stream(): + while True: + try: + frame = await asyncio.wait_for(queue.get(), timeout=15) + except TimeoutError: + yield '\n' + continue + yield json.dumps(frame, ensure_ascii=False) + '\n' + if frame['kind'] in {'completed', 'error'}: + break + + response = quart.Response(stream(), content_type='application/x-ndjson; charset=utf-8') + response.timeout = None + response.headers['Cache-Control'] = 'no-store' + response.headers['X-Accel-Buffering'] = 'no' + return response diff --git a/src/langbot/pkg/api/http/service/assistant.py b/src/langbot/pkg/api/http/service/assistant.py index 74e7073ae..033b53750 100644 --- a/src/langbot/pkg/api/http/service/assistant.py +++ b/src/langbot/pkg/api/http/service/assistant.py @@ -1,6 +1,7 @@ """In-process management assistant using the existing model and resource services.""" import asyncio +import contextlib import json import uuid @@ -44,6 +45,187 @@ class AssistantService: self.ap = ap # ponytail: per-process admission; shared quotas if multiple workers need a global ceiling. self._slots = asyncio.Semaphore(4) + self._runs = {} + + async def list_conversations(self, context): + self._scope(context, '') + result = await self.ap.persistence_mgr.execute_async( + sa.select(Conversation) + .where( + Conversation.workspace_uuid == context.workspace_uuid, + Conversation.account_uuid == context.account_uuid, + ) + .order_by(Conversation.updated_at.desc(), Conversation.uuid) + .limit(100) + ) + return [ + dict( + uuid=row['uuid'], + status=row['status'], + revision=row['revision'], + title=next( + ( + m.get('content', '')[:80] + for m in row['messages'] + if m['role'] == 'user' and isinstance(m.get('content'), str) + ), + '', + ), + ) + for row in result.mappings() + ] + + async def progress(self, context, conversation_id): + conversation = await self.get(context, conversation_id) + run = self._runs.get(conversation_id) + if run: + return {**(run['snapshot'] or self.public_view(conversation)), 'progress': run['progress']} + return self.public_view(conversation) + + def start_run(self, context, conversation_id, body): + """Own execution independently of any HTTP subscriber; never block on a slow browser.""" + if conversation_id in self._runs: + raise AssistantError('stale_turn') + if len(self._runs) >= 4: + raise AssistantError('busy', 429) + queue = asyncio.Queue(maxsize=32) + run = { + 'queue': queue, + 'snapshot': None, + 'progress': {}, + 'stopping': False, + 'revision': body.revision + 1, + 'started': asyncio.Event(), + } + self._runs[conversation_id] = run + + async def emit(frame): + if frame['kind'] == 'snapshot': + run['snapshot'] = frame['data'] + run['progress'] = {} + elif frame['kind'] == 'text': + run['progress']['text'] = frame['data']['text'] + elif frame['kind'] == 'phase': + run['progress'].update(frame['data']) + if queue.full(): + queue.get_nowait() + queue.put_nowait(frame) + + async def execute(): + run['started'].set() + tenant_scope = getattr(self.ap.persistence_mgr, 'tenant_scope', None) + scope = tenant_scope(context.workspace_uuid) if callable(tenant_scope) else contextlib.nullcontext() + try: + async with scope: + try: + conversation = await self.turn( + context, + conversation_id, + body.revision, + body.text, + body.approved, + str(body.model_uuid) if body.model_uuid else None, + on_event=emit, + ) + except asyncio.CancelledError: + conversation = await self.get(context, conversation_id) + # Preserve interrupted tool outcomes; stopping cannot undo an external write. + if run['stopping'] and ( + conversation['error'] == 'result_unknown' + or (conversation['status'] == 'running' and conversation['revision'] == run['revision']) + ): + await self._save(context, conversation, 'failed', 'stopped') + await emit({'kind': 'completed', 'data': self.public_view(conversation)}) + except AssistantError as exc: + await emit({'kind': 'error', 'code': exc.code}) + except Exception: + self.ap.logger.exception('Assistant background run failed') + await emit({'kind': 'error', 'code': 'turn_failed'}) + finally: + self._runs.pop(conversation_id, None) + + run['task'] = asyncio.create_task(execute()) + return queue + + async def stop(self, context, conversation_id, revision): + require_permission(context, Permission.RUNTIME_OPERATE) + conversation = await self.get(context, conversation_id) + run = self._runs.get(conversation_id) + if run and run['revision'] != revision: + raise AssistantError('stale_turn') + if not run: + if conversation['revision'] != revision: + raise AssistantError('stale_turn') + return self.public_view(conversation) + await run['started'].wait() + if not run['stopping']: + run['stopping'] = True + run['task'].cancel() + await asyncio.shield(run['task']) + return self.public_view(await self.get(context, conversation_id)) + + async def recommended_model(self, context): + require_permission(context, Permission.RUNTIME_OPERATE) + try: + recommended = await self.ap.space_service.get_recommended_assistant_model(context) + model = await self.ap.model_mgr.get_model_by_uuid( + ExecutionContext.from_request(context), recommended['uuid'] + ) + if 'func_call' not in (model.model_entity.abilities or []): + raise ValueError('Recommended model does not support tools') + return {'uuid': recommended['uuid'], 'name': model.model_entity.name} + except Exception as exc: + raise AssistantError('model_unavailable', 503) from exc + + async def _invoke(self, model, kwargs, on_event): + if on_event is None: + return await model.provider.invoke_llm(**kwargs) + message = {'role': 'assistant', 'content': ''} + calls = {} + finished = False + async for chunk in model.provider.invoke_llm_stream(**kwargs): + finished = finished or chunk.is_final + content = chunk.content or '' + if isinstance(content, list): + content = ''.join(item.text or '' for item in content if item.type == 'text') + if chunk.all_content is not None: + message['content'] = chunk.all_content + else: + message['content'] += content + for field in ('resp_message_id', 'provider_specific_fields'): + value = getattr(chunk, field, None) + if value is not None: + if field == 'provider_specific_fields': + message[field] = {**message.get(field, {}), **value} + else: + message[field] = value + for call in chunk.tool_calls or []: + value = call.model_dump(mode='json') + if call.id in calls: + saved = calls[call.id] + saved['function']['arguments'] += call.function.arguments or '' + if call.function.name: + saved['function']['name'] = call.function.name + if value.get('provider_specific_fields'): + saved['provider_specific_fields'] = { + **(saved.get('provider_specific_fields') or {}), + **value['provider_specific_fields'], + } + else: + calls[call.id] = value + if len(calls) > 8 or len(json.dumps(message, ensure_ascii=False)) + len(json.dumps(calls)) > 64000: + raise AssistantError('response_too_large') + if content or chunk.all_content is not None: + await on_event({'kind': 'text', 'data': {'text': message['content']}}) + if not finished: + raise AssistantError('stream_interrupted') + if calls: + message['tool_calls'] = list(calls.values()) + return Message.model_validate(message) + + async def _snapshot(self, conversation, on_event): + if on_event is not None: + await on_event({'kind': 'snapshot', 'data': self.public_view(conversation)}) @staticmethod def _scope(context, conversation_id): @@ -150,7 +332,7 @@ class AssistantService: raise AssistantError('stale_turn') conversation.update(status=status, error=error) - async def turn(self, context, conversation_id, revision, text=None, approved=None, model_uuid=None): + async def turn(self, context, conversation_id, revision, text=None, approved=None, model_uuid=None, on_event=None): require_permission(context, Permission.RUNTIME_OPERATE) if model_uuid is not None and text is None: raise AssistantError('invalid_input', 400) @@ -201,7 +383,7 @@ class AssistantService: await self._save(context, conversation, 'running') try: if not conversation['model_uuid']: - recommended = await self.ap.space_service.get_recommended_chat_model(context) + recommended = await self.recommended_model(context) conversation['model_uuid'] = recommended['uuid'] execution = ExecutionContext.from_request(context) model = selected_model or await self.ap.model_mgr.get_model_by_uuid( @@ -215,22 +397,30 @@ class AssistantService: return conversation if approved is not None: calls = self._calls(conversation['messages'][-1]) - await self._execute(context, conversation, calls, approved) - for _ in range(8): - response = await model.provider.invoke_llm( - query=None, - model=model, - messages=[Message(role='system', content=SYSTEM_PROMPT)] - + [Message.model_validate(message) for message in conversation['messages']], - funcs=tool_definitions(context), - extra_args=model.model_entity.extra_args or {}, - remove_think=True, - execution_context=execution, + await self._execute(context, conversation, calls, approved, on_event) + for round_index in range(8): + await self._snapshot(conversation, on_event) + if on_event is not None: + await on_event({'kind': 'phase', 'data': {'phase': 'thinking', 'round': round_index + 1}}) + response = await self._invoke( + model, + dict( + query=None, + model=model, + messages=[Message(role='system', content=SYSTEM_PROMPT)] + + [Message.model_validate(message) for message in conversation['messages']], + funcs=tool_definitions(context), + extra_args=model.model_entity.extra_args or {}, + remove_think=True, + execution_context=execution, + ), + on_event, ) message = response.model_dump(mode='json') if len(json.dumps(message, ensure_ascii=False)) > 64000: raise AssistantError('response_too_large') conversation['messages'].append(message) + await self._snapshot(conversation, on_event) calls = self._calls(message) if not calls: await self._save(context, conversation, 'ready') @@ -249,9 +439,12 @@ class AssistantService: if any(TOOLS[call['function']['name']][2] for call in calls): await self._save(context, conversation, 'approval') return conversation - await self._execute(context, conversation, calls, True) + await self._execute(context, conversation, calls, True, on_event) await self._save(context, conversation, 'failed', 'round_limit') except asyncio.CancelledError: + partial = self._runs.get(conversation_id, {}).get('progress', {}).get('text') + if partial: + conversation['messages'].append(Message(role='assistant', content=partial).model_dump(mode='json')) await asyncio.shield(self._save(context, conversation, 'failed', 'result_unknown')) raise except TimeoutError: @@ -280,7 +473,7 @@ class AssistantService: ).model_dump(mode='json') ) - async def _execute(self, context, conversation, calls, approved): + async def _execute(self, context, conversation, calls, approved, on_event=None): # Validate the complete saved batch before any write, including after approval. if approved: for call in calls: @@ -293,6 +486,19 @@ class AssistantService: else: # Persist before effects. A crash leaves a running/unknown operation, never a replayable approval. await self._save(context, conversation, 'running') + if on_event is not None: + await on_event( + { + 'kind': 'phase', + 'data': { + 'phase': 'tool', + 'tool': { + 'name': func['name'], + 'arguments': redact_secrets(json.loads(func['arguments'] or '{}')), + }, + }, + } + ) try: result = await execute_tool(self.ap, context, func['name'], json.loads(func['arguments'] or '{}')) except Exception as exc: @@ -308,3 +514,4 @@ class AssistantService: raise AssistantError('tool_failed') self._append_result(conversation, call, result) await self._save(context, conversation, 'running') + await self._snapshot(conversation, on_event) diff --git a/src/langbot/pkg/api/http/service/space.py b/src/langbot/pkg/api/http/service/space.py index b07f4f1ed..c95713f56 100644 --- a/src/langbot/pkg/api/http/service/space.py +++ b/src/langbot/pkg/api/http/service/space.py @@ -290,7 +290,23 @@ class SpaceService: selection = await self.get_model_selection('chat') if not selection: raise ValueError('No recommended chat model is available') - recommended = selection[0] + return await self._resolve_recommended_model(context, selection[0]) + + async def get_recommended_assistant_model(self, context: typing.Any) -> dict: + """Resolve the dedicated, health-filtered assistant recommendation without fallback.""" + space_url = self._get_space_config()['url'] + session = httpclient.get_session() + async with session.get(f'{space_url}/api/v1/models/assistant/recommendation') as response: + if response.status != 200: + raise ValueError('No recommended assistant model is available') + payload = await httpclient.read_json_limited(response) + if payload.get('code') != 0: + raise ValueError('No recommended assistant model is available') + recommended = SpaceModel.model_validate(payload['data']['model']) + return await self._resolve_recommended_model(context, recommended) + + async def _resolve_recommended_model(self, context: typing.Any, recommended: SpaceModel) -> dict: + """Map a public catalog model to this Workspace, refreshing the catalog once.""" async def find_local_model(): result = await self.ap.persistence_mgr.execute_async( diff --git a/tests/unit_tests/api/service/test_space_service.py b/tests/unit_tests/api/service/test_space_service.py index a05754af5..8dd1543e7 100644 --- a/tests/unit_tests/api/service/test_space_service.py +++ b/tests/unit_tests/api/service/test_space_service.py @@ -1025,3 +1025,45 @@ class TestSpaceServiceCreditsCache: assert result == 500 assert 'test@example.com' in service._credits_cache assert service._credits_cache['test@example.com'][0] == 500 + + +@pytest.mark.parametrize('status', [200, 503]) +async def test_assistant_recommendation_uses_dedicated_endpoint_without_fallback(status): + service = SpaceService(SimpleNamespace(instance_config=SimpleNamespace(data={}))) + service.get_model_selection = AsyncMock() + service._resolve_recommended_model = AsyncMock(return_value={'uuid': 'local', 'name': 'deepseek-v4-flash'}) + response = MagicMock(status=status) + _set_response_body( + response, + { + 'code': 0, + 'data': { + 'model': { + 'uuid': 'upstream', + 'model_id': 'deepseek-v4-flash', + 'category': 'chat', + 'provider': 'deepseek', + 'status': 'active', + } + }, + }, + ) + session = MagicMock() + session.get.return_value.__aenter__ = AsyncMock(return_value=response) + session.get.return_value.__aexit__ = AsyncMock(return_value=None) + context = object() + with patch('langbot.pkg.api.http.service.space.httpclient.get_session', return_value=session): + if status == 200: + assert await service.get_recommended_assistant_model(context) == { + 'uuid': 'local', + 'name': 'deepseek-v4-flash', + } + args = service._resolve_recommended_model.await_args.args + assert args[0] is context + assert args[1].model_id == 'deepseek-v4-flash' + else: + with pytest.raises(ValueError, match='No recommended assistant model'): + await service.get_recommended_assistant_model(context) + service._resolve_recommended_model.assert_not_awaited() + service.get_model_selection.assert_not_awaited() + session.get.assert_called_once_with('https://space.langbot.app/api/v1/models/assistant/recommendation') diff --git a/tests/unit_tests/api/test_management_assistant.py b/tests/unit_tests/api/test_management_assistant.py index a3f993e6b..8e6fc78b7 100644 --- a/tests/unit_tests/api/test_management_assistant.py +++ b/tests/unit_tests/api/test_management_assistant.py @@ -74,7 +74,7 @@ async def assistant(tmp_path): ap = SimpleNamespace( persistence_mgr=SimpleNamespace(execute_async=execute), logger=logging.getLogger('assistant-test'), - space_service=SimpleNamespace(get_recommended_chat_model=AsyncMock(return_value={'uuid': 'model'})), + space_service=SimpleNamespace(get_recommended_assistant_model=AsyncMock(return_value={'uuid': 'model'})), model_mgr=SimpleNamespace(get_model_by_uuid=AsyncMock(return_value=model)), pipeline_service=SimpleNamespace(create_pipeline=AsyncMock(return_value='created-pipeline')), ) @@ -234,3 +234,250 @@ async def test_model_switch_preserves_history_and_rejects_invalid_selection(assi ap.model_mgr.get_model_by_uuid.assert_awaited_with( next_provider.invoke_llm.call_args.kwargs['execution_context'], 'second' ) + + +@pytest.mark.asyncio +async def test_recommendation_resolved_before_any_turn(assistant): + service, ap, provider = assistant + assert await service.recommended_model(context()) == {'uuid': 'model', 'name': 'test-model'} + provider.invoke_llm.assert_not_awaited() + model = ap.model_mgr.get_model_by_uuid.return_value + model.model_entity.abilities = [] + with pytest.raises(AssistantError, match='model_unavailable'): + await service.recommended_model(context()) + + +@pytest.mark.asyncio +async def test_stream_selected_model_deltas_tools_and_approval(assistant): + from langbot_plugin.api.entities.builtin.provider.message import MessageChunk + import copy + + service, ap, old_provider = assistant + events = [] + inputs = [] + + async def stream(**kwargs): + inputs.append(kwargs) + if len(inputs) == 1: + yield MessageChunk(role='assistant', content='Create ') + yield MessageChunk(role='assistant', content='this?') + call = proposal().tool_calls[0].model_dump() + call['function']['arguments'] = '{"name":"Demo",' + yield MessageChunk(role='assistant', tool_calls=[call]) + call['function']['arguments'] = '"description":"Test draft"}' + yield MessageChunk( + role='assistant', tool_calls=[call], is_final=True, provider_specific_fields={'signature': 'kept'} + ) + else: + yield MessageChunk(role='assistant', content='Done') + yield MessageChunk(role='assistant', content='', all_content='Done', is_final=True) + + chosen = SimpleNamespace( + provider=SimpleNamespace(invoke_llm_stream=stream), + model_entity=SimpleNamespace(name='manual-model', abilities=['func_call'], extra_args={}), + ) + ap.model_mgr.get_model_by_uuid.return_value = chosen + + async def emit(event): + events.append(copy.deepcopy(event)) + + ctx = context() + cid = (await service.create(ctx))['uuid'] + pending = await service.turn(ctx, cid, 0, text='Create', model_uuid='manual', on_event=emit) + assert pending['status'] == 'approval' + assert pending['model_uuid'] == 'manual' + assert inputs[0]['model'] is chosen + ap.space_service.get_recommended_assistant_model.assert_not_awaited() + old_provider.invoke_llm.assert_not_awaited() + assert [e['data']['text'] for e in events if e['kind'] == 'text'] == ['Create ', 'Create this?'] + ap.pipeline_service.create_pipeline.assert_not_awaited() + events.clear() + complete = await service.turn(ctx, cid, 1, approved=True, on_event=emit) + assert complete['status'] == 'ready' + assert complete['messages'][-1]['content'] == 'Done' + ap.pipeline_service.create_pipeline.assert_awaited_once() + assert inputs[1]['messages'][2].provider_specific_fields == {'signature': 'kept'} + tool_start = next(i for i, e in enumerate(events) if e['kind'] == 'phase' and e['data']['phase'] == 'tool') + result = next( + i + for i, e in enumerate(events) + if e['kind'] == 'snapshot' and any(m['role'] == 'tool' for m in e['data']['messages']) + ) + assert tool_start < result + with pytest.raises(AssistantError, match='stale_turn'): + await service.turn(ctx, cid, 1, approved=True, on_event=emit) + + +@pytest.mark.asyncio +async def test_truncated_stream_never_executes_proposed_write(assistant): + from langbot_plugin.api.entities.builtin.provider.message import MessageChunk + + service, ap, provider = assistant + + async def stream(**kwargs): + yield MessageChunk(role='assistant', tool_calls=proposal().tool_calls) + + provider.invoke_llm_stream = stream + cid = (await service.create(context()))['uuid'] + result = await service.turn(context(), cid, 0, text='Create', on_event=AsyncMock()) + assert result['status'] == 'failed' + assert result['error'] == 'stream_interrupted' + ap.pipeline_service.create_pipeline.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_cancelled_stream_persists_non_replayable_state(assistant): + from langbot_plugin.api.entities.builtin.provider.message import MessageChunk + + service, ap, provider = assistant + started = asyncio.Event() + + async def stream(**kwargs): + yield MessageChunk(role='assistant', content='Hello') + started.set() + await asyncio.Event().wait() + + provider.invoke_llm_stream = stream + ctx = context() + cid = (await service.create(ctx))['uuid'] + task = asyncio.create_task(service.turn(ctx, cid, 0, text='Hello', on_event=AsyncMock())) + await asyncio.wait_for(started.wait(), 5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + saved = await service.get(ctx, cid) + assert saved['status'] == 'failed' + assert saved['error'] == 'result_unknown' + with pytest.raises(AssistantError, match='stale_turn'): + await service.turn(ctx, cid, 0, text='Hello') + + +@pytest.mark.asyncio +async def test_http_stream_emits_before_provider_completion_and_restores_tenant(assistant): + import contextlib + import json + import quart + from langbot_plugin.api.entities.builtin.provider.message import MessageChunk + from langbot.pkg.api.http.controller.groups.assistant import assistant_stream_response, TurnInput + + service, ap, provider = assistant + released = asyncio.Event() + scoped = [] + + @contextlib.asynccontextmanager + async def tenant_scope(workspace): + scoped.append(workspace) + try: + yield + finally: + scoped.pop() + + async def stream(**kwargs): + assert scoped == ['workspace-a'] + yield MessageChunk(role='assistant', content='First') + await released.wait() + yield MessageChunk(role='assistant', content=' second', is_final=True) + + ap.persistence_mgr.tenant_scope = tenant_scope + provider.invoke_llm_stream = stream + ctx = context() + cid = (await service.create(ctx))['uuid'] + app = quart.Quart(__name__) + + @app.post('/stream') + async def route(): + return assistant_stream_response(service, ctx, cid, TurnInput(revision=0, text='Hi')) + + frames = [] + async with app.test_client().request('/stream', method='POST') as connection: + while not any(frame['kind'] == 'text' for frame in frames): + chunk = await asyncio.wait_for(connection.receive(), 5) + frames.extend(json.loads(line) for line in chunk.decode().splitlines() if line.strip()) + assert frames[-1]['data']['text'] == 'First' + assert not released.is_set() + released.set() + while not any(frame['kind'] == 'completed' for frame in frames): + chunk = await asyncio.wait_for(connection.receive(), 5) + frames.extend(json.loads(line) for line in chunk.decode().splitlines() if line.strip()) + assert frames[-1]['data']['messages'][-1]['content'] == 'First second' + assert scoped == [] + + +@pytest.mark.asyncio +async def test_background_run_survives_http_disconnect_and_finishes(assistant): + import json + import quart + from langbot_plugin.api.entities.builtin.provider.message import MessageChunk + from langbot.pkg.api.http.controller.groups.assistant import assistant_stream_response, TurnInput + + service, ap, provider = assistant + released = asyncio.Event() + + async def stream(**kwargs): + yield MessageChunk(role='assistant', content='Background') + await released.wait() + yield MessageChunk(role='assistant', content=' done', is_final=True) + + provider.invoke_llm_stream = stream + ctx = context() + cid = (await service.create(ctx))['uuid'] + app = quart.Quart(__name__) + + @app.post('/stream') + async def route(): + return assistant_stream_response(service, ctx, cid, TurnInput(revision=0, text='Hi')) + + async with app.test_client().request('/stream', method='POST') as connection: + while True: + chunk = await asyncio.wait_for(connection.receive(), 5) + if any(json.loads(line)['kind'] == 'text' for line in chunk.decode().splitlines() if line.strip()): + break + task = service._runs[cid]['task'] + await connection.disconnect() + assert not task.done() + assert (await service.progress(ctx, cid))['progress']['text'] == 'Background' + released.set() + await asyncio.wait_for(task, 5) + completed = await service.get(ctx, cid) + assert completed['status'] == 'ready' + assert completed['messages'][-1]['content'] == 'Background done' + + +@pytest.mark.asyncio +async def test_stop_is_private_revision_scoped_and_does_not_stop_other_chat(assistant): + from langbot_plugin.api.entities.builtin.provider.message import MessageChunk + from langbot.pkg.api.http.controller.groups.assistant import TurnInput + + service, ap, provider = assistant + released = asyncio.Event() + + async def stream(**kwargs): + yield MessageChunk(role='assistant', content='Partial') + await released.wait() + yield MessageChunk(role='assistant', content=' done', is_final=True) + + provider.invoke_llm_stream = stream + ctx = context() + ids = [(await service.create(ctx))['uuid'] for _ in range(2)] + tasks = [] + for cid in ids: + queue = service.start_run(ctx, cid, TurnInput(revision=0, text='Background task')) + tasks.append(service._runs[cid]['task']) + while (await asyncio.wait_for(queue.get(), 5))['kind'] != 'text': + pass + for other in (context(account='bob'), context(workspace='workspace-b')): + assert await service.list_conversations(other) == [] + with pytest.raises(AssistantError, match='conversation_not_found'): + await service.stop(other, ids[0], 1) + with pytest.raises(AssistantError, match='stale_turn'): + await service.stop(ctx, ids[0], 0) + result = await service.stop(ctx, ids[0], 1) + assert result['status'] == 'failed' + assert result['error'] == 'stopped' + assert result['messages'][-1]['content'] == 'Partial' + assert not tasks[1].done() + assert len(await service.list_conversations(ctx)) == 2 + assert (await service.stop(ctx, ids[0], 1))['error'] == 'stopped' + released.set() + await asyncio.wait_for(tasks[1], 5) + assert (await service.get(ctx, ids[1]))['status'] == 'ready' diff --git a/web/package.json b/web/package.json index 554594214..358f146c6 100644 --- a/web/package.json +++ b/web/package.json @@ -55,6 +55,7 @@ "@radix-ui/react-toggle": "^1.1.8", "@radix-ui/react-toggle-group": "^1.1.9", "@radix-ui/react-tooltip": "^1.2.7", + "@remixicon/react": "^4.9.0", "@simplewebauthn/browser": "^14.0.0", "@tailwindcss/postcss": "^4.1.5", "@tanstack/react-table": "^8.21.3", diff --git a/web/pnpm-lock.yaml b/web/pnpm-lock.yaml index fd0c6c5f6..a861a5328 100644 --- a/web/pnpm-lock.yaml +++ b/web/pnpm-lock.yaml @@ -93,6 +93,9 @@ dependencies: '@radix-ui/react-tooltip': specifier: ^1.2.7 version: 1.2.8(@types/react-dom@19.2.3)(@types/react@19.2.10)(react-dom@19.2.1)(react@19.2.1) + '@remixicon/react': + specifier: ^4.9.0 + version: 4.9.0(react@19.2.1) '@simplewebauthn/browser': specifier: ^14.0.0 version: 14.0.0 @@ -2921,6 +2924,14 @@ packages: resolution: {integrity: sha512-JtyZR+mqgBibTo8xea3B6ZRmzZiM/YeVBtUkas6zMuXjAlfIFIW2FgqeM9eLyvEaYX66vr6DJMK+4U6LV0KhNw==} dev: false + /@remixicon/react@4.9.0(react@19.2.1): + resolution: {integrity: sha512-5/jLDD4DtKxH2B4QVXTobvV1C2uL8ab9D5yAYNtFt+w80O0Ys1xFOrspqROL3fjrZi+7ElFUWE37hBfaAl6U+Q==} + peerDependencies: + react: '>=18.2.0' + dependencies: + react: 19.2.1 + dev: false + /@rolldown/binding-android-arm64@1.0.3: resolution: {integrity: sha512-454rs7jHngixp/NMxd5srYD57OnzSlZ/eFTETjORQHLwJG1lRtmNOJcBerZlfu4GjKqeq8aCCIQrMdHyhI51Hw==} engines: {node: ^20.19.0 || >=22.12.0} diff --git a/web/src/app/home/components/AssistantToolResult.tsx b/web/src/app/home/components/AssistantToolResult.tsx index 4eabe60c1..c5ee065bf 100644 --- a/web/src/app/home/components/AssistantToolResult.tsx +++ b/web/src/app/home/components/AssistantToolResult.tsx @@ -4,6 +4,7 @@ import { ChevronDown, CircleAlert, MinusCircle, + LoaderCircle, } from 'lucide-react'; import { useTranslation } from 'react-i18next'; @@ -16,11 +17,11 @@ export type AssistantTool = { export default function AssistantToolResult({ tool, content, - defaultCollapsed = false, + defaultCollapsed = true, }: { tool?: AssistantTool; content: string; - /** Turn finished: collapse to one line until the user expands it again. */ + /** Keep tool steps quiet until the user requests details. */ defaultCollapsed?: boolean; }) { const { t } = useTranslation(); @@ -36,11 +37,17 @@ export default function AssistantToolResult({ result && typeof result === 'object' && !Array.isArray(result) ? (result as Record) : {}; + const running = data.status === 'running'; const failed = !!data.error; const denied = data.status === 'denied'; const partial = !!data.truncated; - const Icon = - failed || partial ? CircleAlert : denied ? MinusCircle : CheckCircle2; + const Icon = running + ? LoaderCircle + : failed || partial + ? CircleAlert + : denied + ? MinusCircle + : CheckCircle2; const items = Array.isArray(result) ? result : Array.isArray(data.items) @@ -56,13 +63,15 @@ export default function AssistantToolResult({ : t(`assistant.operations.${tool?.name}`, { defaultValue: t('assistant.toolResult'), }); - const status = failed - ? 'failed' - : denied - ? 'denied' - : partial - ? 'partial' - : 'completed'; + const status = running + ? 'toolRunning' + : failed + ? 'failed' + : denied + ? 'denied' + : partial + ? 'partial' + : 'completed'; const url = typeof data.url === 'string' && /^\/home\/(pipelines|knowledge)\?id=[\w-]+$/.test(data.url) @@ -79,7 +88,7 @@ export default function AssistantToolResult({ return ( ); } return ( -
-
+
+
- {label} - + {label} + {tool && t(`assistant.${status}`)}
- {failed ? ( + {running ? ( +

{t('assistant.toolRunning')}

+ ) : failed ? (

{t('assistant.operationFailed')}

) : denied ? (

@@ -136,7 +147,7 @@ export default function AssistantToolResult({ )} {name &&

{name}

} {items && ( -
    +
      {items.slice(0, 6).map((item: unknown, index: number) => { const entry = item && typeof item === 'object' @@ -168,9 +179,9 @@ export default function AssistantToolResult({ )} )} -
      +
      {t('assistant.details')} - 
      +        
                 {tool ? JSON.stringify(tool.result, null, 2) : content}
      diff --git a/web/src/app/home/components/WorkspaceAssistant.tsx b/web/src/app/home/components/WorkspaceAssistant.tsx index 58834a7b0..3bc9ee401 100644 --- a/web/src/app/home/components/WorkspaceAssistant.tsx +++ b/web/src/app/home/components/WorkspaceAssistant.tsx @@ -1,43 +1,36 @@ +import { useEffect, useRef, useState, useId } from 'react'; +import { createPortal } from 'react-dom'; +import { Plus, Send, ChevronDown, LoaderCircle, Square } from 'lucide-react'; import { - useEffect, - useRef, - useState, - type MouseEvent as ReactMouseEvent, - type PointerEvent as ReactPointerEvent, -} from 'react'; -import { MessageCircle, Plus, Send, X, LoaderCircle } from 'lucide-react'; + RiSparkling2Line, + RiArrowUpSLine, + RiCompass3Line, + RiMagicLine, + RiArrowRightUpLine, +} from '@remixicon/react'; +import styles from './workspace-assistant.module.css'; import { useTranslation } from 'react-i18next'; import ReactMarkdown from 'react-markdown'; import remarkGfm from 'remark-gfm'; +import rehypeHighlight from 'rehype-highlight'; +import '@/styles/github-markdown.css'; import { backendClient, useCurrentWorkspace, userInfo } from '@/app/infra/http'; +import { toast } from 'sonner'; +import { httpClient } from '@/app/infra/http/HttpClient'; +import SettingsDialog, { + SettingsSection, +} from './settings-dialog/SettingsDialog'; import { Button } from '@/components/ui/button'; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from '@/components/ui/select'; import DynamicFormItemComponent from './dynamic-form/DynamicFormItemComponent'; import { DynamicFormItemType } from '@/app/infra/entities/form/dynamic'; import AssistantToolResult, { AssistantTool } from './AssistantToolResult'; -import { - Popover, - PopoverContent, - PopoverTrigger, -} from '@/components/ui/popover'; -import { - ASSISTANT_RAIL_WIDTH, - clampAssistantPosition as clampInViewport, - resolveAssistantEdge, - shouldCollapseRail, - shouldExpandRail, - type AssistantDragPosition, - type AssistantEdge, -} from './assistant-dock'; - -const ASSISTANT_LONG_PRESS_MS = 260; - -function viewport(): { width: number; height: number } { - return { width: window.innerWidth, height: window.innerHeight }; -} - -function clampAssistantPosition(x: number, y: number): AssistantDragPosition { - return clampInViewport(x, y, viewport()); -} type Conversation = { uuid: string; @@ -48,6 +41,12 @@ type Conversation = { error: string | null; model_name: string | null; model_uuid: string | null; + progress?: { + text?: string; + phase?: 'thinking' | 'tool'; + round?: number; + tool?: Omit; + }; }; export default function WorkspaceAssistant() { @@ -59,348 +58,333 @@ export default function WorkspaceAssistant() { return null; const identity = `${workspace.workspace.uuid}:${userInfo.account_uuid}`; return ( - ); } -function AssistantPanel({ storageKey }: { storageKey: string }) { +function AssistantEntry({ storageKey }: { storageKey: string }) { const { t } = useTranslation(); - const [open, setOpen] = useState(false); + const [available, setAvailable] = useState(null); + const [failed, setFailed] = useState(false); + const [settingsOpen, setSettingsOpen] = useState(false); + const [section, setSection] = useState('models'); + const [loginBusy, setLoginBusy] = useState(false); + const [retry, setRetry] = useState(0); + + useEffect(() => { + let active = true; + async function check() { + try { + const { providers } = await backendClient.getModelProviders(); + if (active) { + setAvailable(providers.length > 0); + setFailed(false); + } + } catch { + if (active) setFailed(true); + } + } + if (!settingsOpen) void check(); + window.addEventListener('focus', check); + return () => { + active = false; + window.removeEventListener('focus', check); + }; + }, [settingsOpen, retry]); + + async function login() { + setLoginBusy(true); + try { + const response = await httpClient.getSpaceAuthorizeUrl( + `${window.location.origin}/auth/space/callback`, + ); + window.location.href = response.authorize_url; + } catch { + toast.error(t('common.spaceLoginFailed')); + setLoginBusy(false); + } + } + + return ( + <> + {available ? ( + + ) : available === false || failed ? ( + createPortal( + , + document.body, + ) + ) : null} + + + ); +} + +function AssistantSessions({ storageKey }: { storageKey: string }) { + const [version, setVersion] = useState(0); + function select(id: string | null) { + if (id) localStorage.setItem(storageKey, id); + else localStorage.removeItem(storageKey); + setVersion((value) => value + 1); + } + return ( + + ); +} + +function AssistantPanel({ + storageKey, + onSelect, +}: { + storageKey: string; + onSelect: (id: string | null) => void; +}) { + const { t } = useTranslation(); + const [open, setOpen] = useState(true); + const [closing, setClosing] = useState(false); + const closeTimer = useRef(null); const [conversation, setConversation] = useState(null); const [text, setText] = useState(''); const [sending, setBusy] = useState(false); const [loading, setLoading] = useState(false); - const busy = sending || loading; + const busy = sending || loading || conversation?.status === 'running'; + const [stopping, setStopping] = useState(false); + const [sessions, setSessions] = useState< + { uuid: string; title: string; status: Conversation['status'] }[] + >([]); const [error, setError] = useState(false); const [modelUuid, setModelUuid] = useState(''); + const [modelName, setModelName] = useState(); + const manualModel = useRef(false); + const [modelError, setModelError] = useState(false); + const [liveText, setLiveText] = useState(''); + const [liveTool, setLiveTool] = useState(null); + const [phase, setPhase] = useState<'thinking' | 'tool' | null>(null); + const [round, setRound] = useState(1); const [pendingText, setPendingText] = useState(null); const controller = useRef(new AbortController()); - const end = useRef(null); + const messageList = useRef(null); + const followLatest = useRef(true); + const scrollPosition = useRef(0); - const [dragPosition, setDragPosition] = - useState(null); - const [dragging, setDragging] = useState(false); - const [dockedEdge, setDockedEdge] = useState(null); - // The rail collapses only while the pointer is away. Hovering any part of the - // control restores the full button, and the two states never race because - // every transition is derived from the same `dockedEdge` snapshot. - const [railExpanded, setRailExpanded] = useState(false); - const pointerOver = useRef(false); - const containerRef = useRef(null); - const hoverLocked = useRef(false); - const dragState = useRef<{ - pointerId: number; - startX: number; - startY: number; - originX: number; - originY: number; - active: boolean; - } | null>(null); - const longPressTimer = useRef(null); - const suppressClick = useRef(false); - const buttonRef = useRef(null); + const panelId = useId(); + const expandButton = useRef(null); + const input = useRef(null); - function clearLongPressTimer() { - if (longPressTimer.current !== null) { - window.clearTimeout(longPressTimer.current); - longPressTimer.current = null; - } - } + useEffect( + () => () => { + if (closeTimer.current !== null) window.clearTimeout(closeTimer.current); + }, + [], + ); - useEffect(() => clearLongPressTimer, []); - - /* - * Seed the resting position from the rendered default (bottom-right) so the - * very first visit already docks and collapses. Without this the button would - * only ever dock after a manual drag, which reads as "collapse is broken". - */ - useEffect(() => { - if (window.localStorage.getItem(`${storageKey}:button-position`)) return; - const rect = buttonRef.current?.getBoundingClientRect(); - if (!rect) return; - const position = clampAssistantPosition(rect.left, rect.top); - setDragPosition(position); - const edge = resolveAssistantEdge(position.x, viewport()); - setDockedEdge(edge); - if (edge) { - try { - window.localStorage.setItem(`${storageKey}:button-docked-edge`, edge); - } catch { - // Persisting the dock is best-effort only. - } - } - }, [storageKey]); - - useEffect(() => { - const stored = window.localStorage.getItem(`${storageKey}:button-position`); - if (stored) { - try { - const parsed = JSON.parse(stored) as AssistantDragPosition; - if (typeof parsed?.x === 'number' && typeof parsed?.y === 'number') { - const position = clampAssistantPosition(parsed.x, parsed.y); - setDragPosition(position); - setDockedEdge(resolveAssistantEdge(position.x, viewport())); - } - } catch { - window.localStorage.removeItem(`${storageKey}:button-position`); - } - } - const storedEdge = window.localStorage.getItem( - `${storageKey}:button-docked-edge`, - ); - if (storedEdge === 'left' || storedEdge === 'right') { - setDockedEdge((current) => current ?? storedEdge); - } - }, [storageKey]); - - useEffect(() => { - const onResize = () => { - setDragPosition((prev) => { - if (!prev) return prev; - const next = clampAssistantPosition(prev.x, prev.y); - setDockedEdge(resolveAssistantEdge(next.x, viewport())); - return next; - }); + function collapse() { + if (closeTimer.current !== null) return; + const finish = () => { + closeTimer.current = null; + setOpen(false); + setClosing(false); + window.requestAnimationFrame(() => expandButton.current?.focus()); }; - window.addEventListener('resize', onResize); - return () => window.removeEventListener('resize', onResize); - }, []); - - function persistDragPosition(position: AssistantDragPosition) { - try { - window.localStorage.setItem( - `${storageKey}:button-position`, - JSON.stringify(position), - ); - } catch { - // Ignore storage failures (private mode or quota); dragging still works. + if (window.matchMedia('(prefers-reduced-motion: reduce)').matches) { + finish(); + return; } + setClosing(true); + closeTimer.current = window.setTimeout(finish, 180); } - function applyRestingPosition(position: AssistantDragPosition) { - const edge = resolveAssistantEdge(position.x, viewport()); - setDragPosition(position); - setDockedEdge(edge); - // A fresh dock always collapses; the rail re-expands on the next hover. - setRailExpanded(false); - // Hold the collapse until the pointer leaves, otherwise the still-hovering - // cursor would fight the new state. - if (edge) lockHoverUntilPointerExit(); - try { - if (edge) { - window.localStorage.setItem(`${storageKey}:button-docked-edge`, edge); - } else { - window.localStorage.removeItem(`${storageKey}:button-docked-edge`); - } - } catch { - // Persisting the dock is best-effort only. - } + function expand() { + setOpen(true); + window.requestAnimationFrame(() => input.current?.focus()); } - function endDrag(commit: boolean, clientX?: number, clientY?: number) { - const state = dragState.current; - clearLongPressTimer(); - dragState.current = null; - if (!state?.active) return; - setDragging(false); - const next = clampAssistantPosition( - state.originX + ((clientX ?? state.startX) - state.startX), - state.originY + ((clientY ?? state.startY) - state.startY), - ); - if (commit) { - applyRestingPosition(next); - persistDragPosition(next); - } else { - setDragPosition(next); - } - } - - /* - * Rail hover recovery. The subtle race: a drag usually ends with the pointer - * still sitting on the button, so the browser fires no new `pointerenter` - * once the button collapses. Re-expanding on `pointermove` would therefore - * undo the collapse immediately. - * - * Instead the drop "locks" hover until the pointer physically leaves the - * control. A window-level move listener watches for that exit (the element - * can shift under a stationary cursor, so `pointerleave` alone is not - * reliable) and clears the lock; only then does hovering reveal the button. - */ - function handlePointerEnter() { - pointerOver.current = true; - if (hoverLocked.current) return; - if (shouldExpandRail({ dockedEdge, railExpanded, dragging })) - setRailExpanded(true); - } - - function handlePointerLeave() { - pointerOver.current = false; - hoverLocked.current = false; - // Never collapse mid-drag; the drop handler owns the final state. - if (dragState.current || dragging) return; - if (shouldCollapseRail({ dockedEdge, railExpanded, dragging })) - setRailExpanded(false); - } - - /* - * A click or an aborted swipe never arms the long press, so no drop handler - * runs. Re-sync the rail from the real pointer position on release, otherwise - * a button expanded by hover would stay expanded with the cursor gone. - */ - function settleRailAfterRelease() { - if (pointerOver.current || dragState.current || dragging) return; - if (shouldCollapseRail({ dockedEdge, railExpanded, dragging })) - setRailExpanded(false); - } - - function lockHoverUntilPointerExit() { - hoverLocked.current = true; - const releaseOnExit = (moveEvent: PointerEvent) => { - const rect = containerRef.current?.getBoundingClientRect(); - if (!rect) return; - const outside = - moveEvent.clientX < rect.left || - moveEvent.clientX > rect.right || - moveEvent.clientY < rect.top || - moveEvent.clientY > rect.bottom; - if (!outside) return; - hoverLocked.current = false; - window.removeEventListener('pointermove', releaseOnExit); - }; - window.addEventListener('pointermove', releaseOnExit); - } - - function onButtonPointerDown(event: ReactPointerEvent) { - if (event.button !== 0) return; - suppressClick.current = false; - const rect = buttonRef.current?.getBoundingClientRect(); - if (!rect) return; - const state = { - pointerId: event.pointerId, - startX: event.clientX, - startY: event.clientY, - originX: rect.left, - originY: rect.top, - active: false, - }; - dragState.current = state; - clearLongPressTimer(); - - const onWindowMove = (moveEvent: PointerEvent) => { - if (moveEvent.pointerId !== state.pointerId) return; - const deltaX = moveEvent.clientX - state.startX; - const deltaY = moveEvent.clientY - state.startY; - - if (!state.active) { - // Cancel the long-press when the user is clearly scrolling or swiping. - if (Math.hypot(deltaX, deltaY) > 8) clearLongPressTimer(); - return; - } - - moveEvent.preventDefault(); - setDragPosition( - clampAssistantPosition(state.originX + deltaX, state.originY + deltaY), - ); - }; - - const onWindowUp = (upEvent: PointerEvent) => { - if (upEvent.pointerId !== state.pointerId) return; - window.removeEventListener('pointermove', onWindowMove); - window.removeEventListener('pointerup', onWindowUp); - window.removeEventListener('pointercancel', onWindowUp); - endDrag(true, upEvent.clientX, upEvent.clientY); - }; - - window.addEventListener('pointermove', onWindowMove, { passive: false }); - window.addEventListener('pointerup', onWindowUp); - window.addEventListener('pointercancel', onWindowUp); - - // Release outside a drag still needs to reconcile the rail. - window.addEventListener('pointerup', settleRailAfterRelease, { - once: true, - }); - - longPressTimer.current = window.setTimeout(() => { - if (dragState.current !== state) return; - state.active = true; - setDragging(true); - suppressClick.current = true; - setDragPosition(clampAssistantPosition(state.originX, state.originY)); - }, ASSISTANT_LONG_PRESS_MS); - } - - function onButtonClick(event: ReactMouseEvent) { - // After a drag the trailing click must not toggle the panel. Radix's - // trigger skips its own toggle when the event default is prevented. - if (suppressClick.current) { - suppressClick.current = false; - event.preventDefault(); - event.stopPropagation(); - } - } - - useEffect(() => { - if (conversation?.model_uuid) setModelUuid(conversation.model_uuid); - }, [conversation?.model_uuid]); - useEffect(() => { const abort = new AbortController(); controller.current = abort; + setLoading(true); + async function initialize() { + try { + const id = localStorage.getItem(storageKey); + let saved: Conversation | null = null; + if (id) { + saved = await backendClient.request({ + method: 'GET', + url: `/api/v1/assistant/conversations/${encodeURIComponent(id)}`, + signal: abort.signal, + }); + if (abort.signal.aborted) return; + setConversation(saved); + } + const model = saved?.model_uuid + ? { uuid: saved.model_uuid, name: saved.model_name ?? undefined } + : await backendClient.request<{ uuid: string; name: string }>({ + method: 'GET', + url: '/api/v1/assistant/recommended-model', + signal: abort.signal, + }); + if (!abort.signal.aborted && !manualModel.current) { + setModelUuid(model.uuid); + setModelName(model.name); + } + } catch { + if (!abort.signal.aborted) setModelError(true); + } finally { + if (!abort.signal.aborted) setLoading(false); + } + } + void initialize(); return () => abort.abort(); - }, []); + }, [storageKey]); useEffect(() => { - if (!open || busy || (conversation && conversation.status !== 'running')) - return; - const id = localStorage.getItem(storageKey); - if (!id) return; let active = true; - setLoading(true); - backendClient - .request({ - method: 'GET', - url: `/api/v1/assistant/conversations/${encodeURIComponent(id)}`, - signal: controller.current.signal, - }) - .then((value) => { - if (active) setConversation(value); - }) - .catch(() => { - if (active) { - localStorage.removeItem(storageKey); - setError(true); + let timer: ReturnType; + async function refresh() { + try { + const result = await backendClient.request<{ + conversations: typeof sessions; + }>({ + method: 'GET', + url: '/api/v1/assistant/conversations', + signal: controller.current.signal, + }); + if (active) setSessions(result.conversations ?? []); + if (!sending && conversation?.uuid) { + const latest = await backendClient.request({ + method: 'GET', + url: `/api/v1/assistant/conversations/${conversation.uuid}`, + signal: controller.current.signal, + }); + if (active) { + setConversation(latest); + setLiveText(latest.progress?.text ?? ''); + setPhase(latest.progress?.phase ?? null); + setRound(latest.progress?.round ?? 1); + setLiveTool( + latest.progress?.tool + ? { ...latest.progress.tool, result: { status: 'running' } } + : null, + ); + } } - }) - .finally(() => { - if (active) setLoading(false); - }); + } catch { + // Polling is read-only. A failed refresh never retries a turn. + } finally { + if (active) timer = setTimeout(refresh, 1500); + } + } + void refresh(); return () => { active = false; - setLoading(false); + clearTimeout(timer); }; - // Load only when opening; turn requests own subsequent state updates. - // eslint-disable-next-line react-hooks/exhaustive-deps - }, [open, storageKey]); + }, [conversation?.uuid, sending]); + + async function stop() { + if (!conversation || stopping) return; + setStopping(true); + try { + const latest = await backendClient.request({ + method: 'POST', + url: `/api/v1/assistant/conversations/${conversation.uuid}/stop`, + data: { revision: conversation.revision }, + signal: controller.current.signal, + }); + setConversation(latest); + setLiveText(''); + setLiveTool(null); + setPhase(null); + } catch { + setError(true); + } finally { + setStopping(false); + } + } + + // Restore the reading position when reopening; new conversations start at the bottom. + useEffect(() => { + if (open && messageList.current && !followLatest.current) { + messageList.current.scrollTop = scrollPosition.current; + } + }, [open]); - // `open` is a dependency so a freshly opened panel jumps to the newest turn - // instead of leaving the user at the oldest message. The rAF waits for the - // popover to lay out before measuring the sentinel. useEffect(() => { if (!open) return; const frame = window.requestAnimationFrame(() => { - end.current?.scrollIntoView({ behavior: 'auto', block: 'nearest' }); + const list = messageList.current; + if (list && followLatest.current) { + list.scrollTop = list.scrollHeight; + scrollPosition.current = list.scrollTop; + } }); return () => window.cancelAnimationFrame(frame); - }, [open, conversation, busy, pendingText]); + }, [open, conversation, busy, pendingText, liveText, liveTool]); async function submit(approved?: boolean) { if ( busy || (approved === undefined && - (!text.trim() || (conversation && conversation.status !== 'ready'))) + (!text.trim() || + !modelUuid || + (conversation && conversation.status !== 'ready'))) ) return; const sentText = approved === undefined ? text.trim() : null; @@ -411,6 +395,8 @@ function AssistantPanel({ storageKey }: { storageKey: string }) { } setBusy(true); setError(false); + setRound(1); + setPhase('thinking'); try { let current = conversation; if (!current) { @@ -422,24 +408,48 @@ function AssistantPanel({ storageKey }: { storageKey: string }) { localStorage.setItem(storageKey, current.uuid); setConversation(current); } - const updated = await backendClient.request({ - method: 'POST', - url: `/api/v1/assistant/conversations/${current.uuid}/turn`, - data: { + const updated = await backendClient.streamAssistant( + current.uuid, + { revision: current.revision, ...(approved === undefined - ? { - text: sentText, - ...(modelUuid ? { model_uuid: modelUuid } : {}), - } + ? { text: sentText!, model_uuid: modelUuid } : { approved }), }, - timeout: 130000, - signal: controller.current.signal, - }); + (event) => { + if (event.kind === 'snapshot') { + const snapshot = event.data as Conversation; + setConversation(snapshot); + setPendingText(null); + setLiveText(''); + setLiveTool(null); + if (snapshot.model_uuid) { + setModelUuid(snapshot.model_uuid); + setModelName(snapshot.model_name ?? undefined); + } + } else if (event.kind === 'text') { + setLiveText((event.data as { text: string }).text); + } else if (event.kind === 'phase') { + const state = event.data as { + phase: 'thinking' | 'tool'; + round?: number; + tool?: Omit; + }; + setPhase(state.phase); + if (state.round) setRound(state.round); + setLiveTool( + state.tool + ? { ...state.tool, result: { status: 'running' } } + : null, + ); + } + }, + controller.current.signal, + ); setConversation(updated); if (sentText) setPendingText(null); } catch { + if (controller.current.signal.aborted) return; setError(true); // A lost response may already have executed a write. Refresh, never replay. const id = localStorage.getItem(storageKey); @@ -466,117 +476,137 @@ function AssistantPanel({ storageKey }: { storageKey: string }) { } } finally { setBusy(false); + setLiveText(''); + setLiveTool(null); + setPhase(null); } } function reset() { - localStorage.removeItem(storageKey); - setConversation(null); - setText(''); - setError(false); - setPendingText(null); + onSelect(null); } - // Collapse only when docked, idle and not hovered. The container keeps its - // resting box, so the hidden button and the visible strip share one anchor - // and cannot drift apart; only the strip is painted while collapsed. - const railCollapsed = !!dockedEdge && !railExpanded && !dragging; - const inlinePosition = dragPosition - ? { left: dragPosition.x, top: dragPosition.y } - : undefined; + // A user message starts a turn; its tool steps and replies share one bubble. + const messageGroups: { + key: number; + role: 'user' | 'assistant'; + messages: Conversation['messages']; + }[] = []; + const visibleMessages: Conversation['messages'] = [ + ...(conversation?.messages ?? []), + ]; + if (liveText) visibleMessages.push({ role: 'assistant', content: liveText }); + if (liveTool) + visibleMessages.push({ role: 'tool', content: '', tool: liveTool }); + for (const [index, message] of visibleMessages.entries()) { + const role = message.role === 'user' ? 'user' : 'assistant'; + const previous = messageGroups.at(-1); + if (role === 'assistant' && previous?.role === 'assistant') { + previous.messages.push(message); + } else { + messageGroups.push({ key: index, role, messages: [message] }); + } + } - return ( -
      - - - - - {railCollapsed && ( -