diff --git a/pyproject.toml b/pyproject.toml index 5319a8aaf..78b88cd67 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,7 +71,7 @@ dependencies = [ "langchain-text-splitters>=1.1.2", "chromadb>=1.0.0,<2.0.0", "qdrant-client (>=1.15.1,<2.0.0)", - "langbot-plugin==0.7.11rc1", + "langbot-plugin @ git+https://github.com/langbot-app/langbot-plugin-sdk.git@8c29bc0787be3b514fb379c5154dfc5eaa561cbd", "asyncpg>=0.30.0", "line-bot-sdk>=3.19.0", "matrix-nio>=0.25.2", diff --git a/src/langbot/libs/dingtalk_api/api.py b/src/langbot/libs/dingtalk_api/api.py index 91aa55a58..48032b72a 100644 --- a/src/langbot/libs/dingtalk_api/api.py +++ b/src/langbot/libs/dingtalk_api/api.py @@ -80,7 +80,9 @@ def is_stream_card_configured(config: dict) -> bool: return False template_id = config.get('card_template_id') # Older manifests stored instructional placeholder text as the default. - return isinstance(template_id, str) and bool(template_id.strip()) and template_id.strip() != '填写你的卡片template_id' + return ( + isinstance(template_id, str) and bool(template_id.strip()) and template_id.strip() != '填写你的卡片template_id' + ) class DingTalkClient: @@ -283,9 +285,9 @@ class DingTalkClient: data = response.json() if response.status_code != 200 or data.get('errcode', 0) != 0 or data.get('code'): raise RuntimeError( - f"DingTalk delivery failed (HTTP {response.status_code}, " - f"code={data.get('errcode', data.get('code', 'unknown'))}): " - f"{data.get('errmsg', data.get('message', 'Unknown error'))}" + f'DingTalk delivery failed (HTTP {response.status_code}, ' + f'code={data.get("errcode", data.get("code", "unknown"))}): ' + f'{data.get("errmsg", data.get("message", "Unknown error"))}' ) return data @@ -333,7 +335,8 @@ class DingTalkClient: response = await client.post( 'https://oapi.dingtalk.com/media/upload', params={'access_token': token, 'type': 'image' if is_image else 'file'}, - files={'media': (name, body)}, timeout=30.0, + files={'media': (name, body)}, + timeout=30.0, ) uploaded = self._check_send_result(response) media_id = uploaded.get('media_id') @@ -341,9 +344,11 @@ class DingTalkClient: raise RuntimeError('DingTalk media upload returned no media ID') if not await self.check_access_token(): await self.get_access_token() - params = {'photoURL': media_id} if is_image else { - 'mediaId': media_id, 'fileName': name, 'fileType': os.path.splitext(name)[1].lstrip('.') - } + params = ( + {'photoURL': media_id} + if is_image + else {'mediaId': media_id, 'fileName': name, 'fileType': os.path.splitext(name)[1].lstrip('.')} + ) payload = { 'robotCode': self.robot_code or self.key, 'msgKey': 'sampleImageMsg' if is_image else 'sampleFile', @@ -359,7 +364,8 @@ class DingTalkClient: response = await client.post( f'{DINGTALK_OPENAPI_BASE}/v1.0/robot/{endpoint}', headers={'x-acs-dingtalk-access-token': self.access_token}, - json=payload, timeout=30.0, + json=payload, + timeout=30.0, ) return self._check_send_result(response) @@ -724,7 +730,14 @@ class DingTalkClient: else: open_space_id = f'dtv1.card//IM_ROBOT.{incoming_message.sender_staff_id}' - card_param_map = {'content': '', 'answer': '', 'reasoning': '', 'flowStatus': '1', 'brand': '', 'hasReasoning': ''} + card_param_map = { + 'content': '', + 'answer': '', + 'reasoning': '', + 'flowStatus': '1', + 'brand': '', + 'hasReasoning': '', + } if incoming_message.message_type == 'text': card_param_map['query'] = incoming_message.get_text_list()[0] else: diff --git a/src/langbot/libs/dingtalk_api/thinking.py b/src/langbot/libs/dingtalk_api/thinking.py index a58d51f75..d4a8b54ee 100644 --- a/src/langbot/libs/dingtalk_api/thinking.py +++ b/src/langbot/libs/dingtalk_api/thinking.py @@ -2,17 +2,17 @@ import re -_MARKER = re.compile(r"", re.IGNORECASE) +_MARKER = re.compile(r'', re.IGNORECASE) def visible_think_markers(content: str, markdown: bool = True) -> str: def replace(match): tag = match.group(1).lower() - separator = "\n\n" if markdown else "\n" - return separator + (f"`{tag}`" if markdown else tag) + separator + separator = '\n\n' if markdown else '\n' + return separator + (f'`{tag}`' if markdown else tag) + separator # Normalize only marker boundaries; preserve paragraph spacing in the body. - return re.sub(r"[ \t\r\n]*()[ \t\r\n]*", replace, content, flags=re.IGNORECASE).strip() + return re.sub(r'[ \t\r\n]*()[ \t\r\n]*', replace, content, flags=re.IGNORECASE).strip() def card_text_fields(content: str) -> dict[str, str]: @@ -20,20 +20,20 @@ def card_text_fields(content: str) -> dict[str, str]: inside = False offset = 0 for match in _MARKER.finditer(content): - (thinking if inside else answer).append(content[offset:match.start()]) - inside = match.group(0).lower() == "" + (thinking if inside else answer).append(content[offset : match.start()]) + inside = match.group(0).lower() == '' offset = match.end() tail = content[offset:] # A stream can stop in the middle of an opening or closing marker. for size in range(min(len(tail), 7), 0, -1): - if any(marker.startswith(tail[-size:].lower()) for marker in ("", "")): + if any(marker.startswith(tail[-size:].lower()) for marker in ('', '')): tail = tail[:-size] break (thinking if inside else answer).append(tail) - reasoning = "\n".join(part.strip() for part in thinking if part.strip()) - response = "\n".join(part.strip() for part in answer if part.strip()) + reasoning = '\n'.join(part.strip() for part in thinking if part.strip()) + response = '\n'.join(part.strip() for part in answer if part.strip()) return { - "reasoning": "> " + reasoning.replace("\n", "\n> ") if reasoning else "", - "answer": response, - "hasReasoning": "true" if reasoning else "", + 'reasoning': '> ' + reasoning.replace('\n', '\n> ') if reasoning else '', + 'answer': response, + 'hasReasoning': 'true' if reasoning else '', } diff --git a/src/langbot/libs/official_account_api/api.py b/src/langbot/libs/official_account_api/api.py index 881ce99ae..5daf36a6c 100644 --- a/src/langbot/libs/official_account_api/api.py +++ b/src/langbot/libs/official_account_api/api.py @@ -33,15 +33,26 @@ async def _read_callback_xml(client, req): if root.find('Encrypt') is not None: crypt = WXBizMsgCrypt(client.token, client.aes, client.appid) code, xml = await asyncio.to_thread( - crypt.DecryptMsg, body, req.args.get('msg_signature', ''), - req.args.get('timestamp', ''), req.args.get('nonce', ''), + crypt.DecryptMsg, + body, + req.args.get('msg_signature', ''), + req.args.get('timestamp', ''), + req.args.get('nonce', ''), ) if code != 0: raise ValueError('Official Account callback decryption failed') return xml.decode('utf-8') if isinstance(xml, bytes) else xml - digest = hashlib.sha1(''.join(sorted([ - client.token, req.args.get('timestamp', ''), req.args.get('nonce', ''), - ])).encode('utf-8')).hexdigest() + digest = hashlib.sha1( + ''.join( + sorted( + [ + client.token, + req.args.get('timestamp', ''), + req.args.get('nonce', ''), + ] + ) + ).encode('utf-8') + ).hexdigest() if not hmac.compare_digest(digest, req.args.get('signature', '')): raise ValueError('Official Account callback signature verification failed') return body.decode('utf-8') @@ -232,9 +243,15 @@ class OAClient: if message_data.get('MsgId'): message_data['MsgId'] = int(message_data['MsgId']) else: - message_data['MsgId'] = ':'.join(str(message_data.get(key) or '') for key in ( - 'FromUserName', 'CreateTime', 'Event', 'EventKey', - )) + message_data['MsgId'] = ':'.join( + str(message_data.get(key) or '') + for key in ( + 'FromUserName', + 'CreateTime', + 'Event', + 'EventKey', + ) + ) return message_data @@ -423,9 +440,11 @@ class OAClientForLongerResponse: root = await asyncio.to_thread(ET.fromstring, xml_msg) from_user = root.find('FromUserName').text to_user = root.find('ToUserName').text - message_key = (from_user, root.findtext('MsgId') or ( - root.findtext('CreateTime'), root.findtext('Event'), root.findtext('EventKey') - )) + message_key = ( + from_user, + root.findtext('MsgId') + or (root.findtext('CreateTime'), root.findtext('Event'), root.findtext('EventKey')), + ) cached = self._callback_responses.get(message_key) if cached and time.monotonic() - cached[0] < 600: return cached[1] @@ -502,9 +521,15 @@ class OAClientForLongerResponse: if message_data.get('MsgId'): message_data['MsgId'] = int(message_data['MsgId']) else: - message_data['MsgId'] = ':'.join(str(message_data.get(key) or '') for key in ( - 'FromUserName', 'CreateTime', 'Event', 'EventKey', - )) + message_data['MsgId'] = ':'.join( + str(message_data.get(key) or '') + for key in ( + 'FromUserName', + 'CreateTime', + 'Event', + 'EventKey', + ) + ) return message_data diff --git a/src/langbot/libs/wecom_customer_service_api/api.py b/src/langbot/libs/wecom_customer_service_api/api.py index c681be505..916cd3f39 100644 --- a/src/langbot/libs/wecom_customer_service_api/api.py +++ b/src/langbot/libs/wecom_customer_service_api/api.py @@ -198,7 +198,9 @@ class WecomCSClient: params['cursor'] = cursor async with self._http_client_context() as client: response = await client.post( - self.base_url + '/kf/sync_msg', params={'access_token': self.access_token}, json=params, + self.base_url + '/kf/sync_msg', + params={'access_token': self.access_token}, + json=params, ) data = await httpclient.parse_json_response(response) if data.get('errcode') in (40014, 42001) and not refreshed: @@ -206,7 +208,7 @@ class WecomCSClient: refreshed = True continue if data.get('errcode') != 0: - raise RuntimeError(f"WeCom sync_msg failed (errcode={data.get('errcode')})") + raise RuntimeError(f'WeCom sync_msg failed (errcode={data.get("errcode")})') messages.extend(data.get('msg_list') or []) next_cursor = data.get('next_cursor') if data.get('has_more'): @@ -452,9 +454,12 @@ class WecomCSClient: self._received_messages = { key: seen for key, seen in self._received_messages.items() if now - seen < 4 * 86400 } - message_id = message_data.get('msgid') or hashlib.sha256( - json.dumps(message_data, sort_keys=True, ensure_ascii=False, separators=(',', ':')).encode() - ).hexdigest() + message_id = ( + message_data.get('msgid') + or hashlib.sha256( + json.dumps(message_data, sort_keys=True, ensure_ascii=False, separators=(',', ':')).encode() + ).hexdigest() + ) if not message_id or message_id not in self._received_messages: if message_id: if len(self._received_messages) >= 4096: diff --git a/src/langbot/pkg/agent/runner/persistent_state_store.py b/src/langbot/pkg/agent/runner/persistent_state_store.py index e0a58af1a..4665f127a 100644 --- a/src/langbot/pkg/agent/runner/persistent_state_store.py +++ b/src/langbot/pkg/agent/runner/persistent_state_store.py @@ -192,7 +192,10 @@ class PersistentStateStore: if self._check_scope_enabled('conversation', binding) and event.conversation_id: if 'external.conversation_id' not in snapshot['conversation']: from .context_reset import get_reset_generation - generation = await get_reset_generation(self._db_engine, event.workspace_id, event.bot_id, event.conversation_id) + + generation = await get_reset_generation( + self._db_engine, event.workspace_id, event.bot_id, event.conversation_id + ) snapshot['conversation']['external.conversation_id'] = generation or event.conversation_id return snapshot diff --git a/src/langbot/pkg/agent/runner/platform_tools.py b/src/langbot/pkg/agent/runner/platform_tools.py index 3f0b4d198..5b7dc6509 100644 --- a/src/langbot/pkg/agent/runner/platform_tools.py +++ b/src/langbot/pkg/agent/runner/platform_tools.py @@ -742,6 +742,7 @@ async def execute_platform_tool( raise ValueError(f'Platform API {definition.api} is no longer supported by bot {bot_id}') api_func = getattr(bot.adapter, definition.api, None) if reply_via_source: + async def api_func(target_type, target_id, message): source = await bot.adapter.get_message( chat_type=target_type, @@ -749,14 +750,17 @@ async def execute_platform_tool( message_id=delivery['reply_target']['message_id'], ) return await bot.adapter.reply_message(source, message) + # Some platforms require a one-time event credential instead of an # ordinary conversation send. Only pass the Host-frozen event context. event_reply_handler = getattr(bot.adapter, 'reply_to_event', None) if definition.name == 'event_reply' and callable(event_reply_handler): + async def api_func(target_type, target_id, message): return await event_reply_handler( context=context, target_type=target_type, target_id=target_id, message=message ) + if not callable(api_func): raise ValueError(f'Platform API {definition.api} is declared but not implemented') if definition.api == 'send_message': diff --git a/src/langbot/pkg/agent/runner/transcript_store.py b/src/langbot/pkg/agent/runner/transcript_store.py index 6f1a7757d..426740746 100644 --- a/src/langbot/pkg/agent/runner/transcript_store.py +++ b/src/langbot/pkg/agent/runner/transcript_store.py @@ -336,6 +336,7 @@ class TranscriptStore: strict_thread: bool, ) -> typing.Any: from .context_reset import active_history_condition + query = query.where(active_history_condition(Transcript)) if bot_id is not None: query = query.where(Transcript.bot_id == bot_id) diff --git a/src/langbot/pkg/api/http/controller/groups/monitoring.py b/src/langbot/pkg/api/http/controller/groups/monitoring.py index 929c061e1..aae227e25 100644 --- a/src/langbot/pkg/api/http/controller/groups/monitoring.py +++ b/src/langbot/pkg/api/http/controller/groups/monitoring.py @@ -32,6 +32,7 @@ def validate_monitoring_window(start_time: datetime.datetime | None, end_time: d if start_time >= end or end - start_time > datetime.timedelta(days=365): quart.abort(400, description='Monitoring time range must be positive and no longer than 365 days') + @group.group_class('monitoring', '/api/v1/monitoring') class MonitoringRouterGroup(group.RouterGroup): async def initialize(self) -> None: @@ -545,15 +546,21 @@ class MonitoringRouterGroup(group.RouterGroup): } ) - @self.route('/sessions/reset-context', methods=['POST'], - auth_type=group.AuthType.USER_TOKEN_OR_API_KEY, permission=Permission.RESOURCE_MANAGE) + @self.route( + '/sessions/reset-context', + methods=['POST'], + auth_type=group.AuthType.USER_TOKEN_OR_API_KEY, + permission=Permission.RESOURCE_MANAGE, + ) async def reset_session_context(request_context: RequestContext): body = await quart.request.get_json() if not isinstance(body, dict): return self.http_status(400, 'invalid_request', 'Expected a JSON object') try: result = await self.ap.monitoring_service.reset_session_context( - request_context, body.get('bot_id'), body.get('session_id'), + request_context, + body.get('bot_id'), + body.get('session_id'), ) except ValueError as exc: return self.http_status(400, 'invalid_request', str(exc)) diff --git a/src/langbot/pkg/api/http/controller/groups/platform/adapters.py b/src/langbot/pkg/api/http/controller/groups/platform/adapters.py index 92a3e2d26..4715ceb95 100644 --- a/src/langbot/pkg/api/http/controller/groups/platform/adapters.py +++ b/src/langbot/pkg/api/http/controller/groups/platform/adapters.py @@ -162,23 +162,43 @@ class AdaptersRouterGroup(group.RouterGroup): slack_setup = get_slack_setup(self.ap) - @self.route('/slack/setup', methods=['POST'], permission=Permission.RESOURCE_MANAGE, - auth_type=group.AuthType.USER_TOKEN_OR_API_KEY) + @self.route( + '/slack/setup', + methods=['POST'], + permission=Permission.RESOURCE_MANAGE, + auth_type=group.AuthType.USER_TOKEN_OR_API_KEY, + ) async def slack_start(request_context: RequestContext): data = await quart.request.get_json() if not isinstance(data, dict): return self.http_status(400, -1, 'invalid_input') try: - result = await slack_setup.start(request_context, **{ - key: data[key] for key in ('bot_uuid', 'access_token', 'refresh_token', 'webhook_url', - 'name', 'socket_mode', 'redirect_url') if key in data - }) + result = await slack_setup.start( + request_context, + **{ + key: data[key] + for key in ( + 'bot_uuid', + 'access_token', + 'refresh_token', + 'webhook_url', + 'name', + 'socket_mode', + 'redirect_url', + ) + if key in data + }, + ) return self.success(data=result) except (SlackSetupError, TypeError) as exc: return self.http_status(400, -1, str(exc) if isinstance(exc, SlackSetupError) else 'invalid_input') - @self.route('/slack/setup/', methods=['GET', 'DELETE'], - permission=Permission.RESOURCE_MANAGE, auth_type=group.AuthType.USER_TOKEN_OR_API_KEY) + @self.route( + '/slack/setup/', + methods=['GET', 'DELETE'], + permission=Permission.RESOURCE_MANAGE, + auth_type=group.AuthType.USER_TOKEN_OR_API_KEY, + ) async def slack_status(session_id: str, request_context: RequestContext): try: slack_setup.owned(request_context, session_id) @@ -194,15 +214,22 @@ class AdaptersRouterGroup(group.RouterGroup): @self.route('/slack/setup/callback', methods=['GET'], auth_type=group.AuthType.NONE) async def slack_callback(): try: - await slack_setup.finish(quart.request.args.get('state', ''), quart.request.args.get('code', ''), - quart.request.args.get('error')) + await slack_setup.finish( + quart.request.args.get('state', ''), + quart.request.args.get('code', ''), + quart.request.args.get('error'), + ) message = 'Slack authorization completed. Return to LangBot to finish setup. / Slack 授权完成，请返回 LangBot 完成配置。' status = 200 except SlackSetupError: message = 'Slack authorization failed or expired. Return to LangBot for details. / Slack 授权失败或已过期，请返回 LangBot 查看详情。' status = 400 - return quart.Response(message, status=status, content_type='text/plain; charset=utf-8', - headers={'Cache-Control': 'no-store', 'Referrer-Policy': 'no-referrer'}) + return quart.Response( + message, + status=status, + content_type='text/plain; charset=utf-8', + headers={'Cache-Control': 'no-store', 'Referrer-Policy': 'no-referrer'}, + ) @self.route('', methods=['GET']) async def _() -> str: diff --git a/src/langbot/pkg/api/http/service/agent.py b/src/langbot/pkg/api/http/service/agent.py index 20298dd12..394c3b3dd 100644 --- a/src/langbot/pkg/api/http/service/agent.py +++ b/src/langbot/pkg/api/http/service/agent.py @@ -452,7 +452,9 @@ class AgentService: if runner_id: descriptor = await self.ap.runner_registry.get(context, runner_id) config['runner_config'][runner_id] = await self.ap.llm_model_service.apply_default_model( - context, descriptor.config_schema, config['runner_config'][runner_id], + context, + descriptor.config_schema, + config['runner_config'][runner_id], ) patterns = agent_data.get('supported_event_patterns', AGENT_DEFAULT_EVENT_PATTERNS) new_uuid = str(uuid.uuid4()) @@ -566,7 +568,9 @@ class AgentService: if parameters is None: parameters = self.ap.pipeline_service._get_default_values_from_schema(descriptor.config_schema) if draft: - parameters = await self.ap.llm_model_service.apply_default_model(context, descriptor.config_schema, parameters) + parameters = await self.ap.llm_model_service.apply_default_model( + context, descriptor.config_schema, parameters + ) if not isinstance(parameters, dict): raise ValueError('Processor parameters must be an object') for field in descriptor.config_schema: diff --git a/src/langbot/pkg/api/http/service/model.py b/src/langbot/pkg/api/http/service/model.py index bbf992974..601d79d30 100644 --- a/src/langbot/pkg/api/http/service/model.py +++ b/src/langbot/pkg/api/http/service/model.py @@ -198,6 +198,7 @@ class LLMModelsService: async def get_starred_model(self, context: TenantContext) -> str | None: from ....entity.persistence.metadata import WorkspaceMetadata + result = await self.ap.persistence_mgr.execute_async( sqlalchemy.select(WorkspaceMetadata.value).where( WorkspaceMetadata.workspace_uuid == require_workspace_uuid(context), @@ -212,6 +213,7 @@ class LLMModelsService: async def set_starred_model(self, context: TenantContext, model_uuid: str | None) -> None: from ....entity.persistence.metadata import WorkspaceMetadata + workspace_uuid = require_workspace_uuid(context) if model_uuid and await self.get_llm_model(context, model_uuid) is None: raise ValueError('Model not found in this Workspace') @@ -222,11 +224,16 @@ class LLMModelsService: else: from sqlalchemy.dialects.sqlite import insert statement = insert(WorkspaceMetadata).values( - workspace_uuid=workspace_uuid, key='starred_llm_model', value=model_uuid or '', + workspace_uuid=workspace_uuid, + key='starred_llm_model', + value=model_uuid or '', + ) + await self.ap.persistence_mgr.execute_async( + statement.on_conflict_do_update( + index_elements=['workspace_uuid', 'key'], + set_={'value': model_uuid or ''}, + ) ) - await self.ap.persistence_mgr.execute_async(statement.on_conflict_do_update( - index_elements=['workspace_uuid', 'key'], set_={'value': model_uuid or ''}, - )) async def get_default_model(self, context: TenantContext) -> str | None: starred = await self.get_starred_model(context) @@ -245,9 +252,18 @@ class LLMModelsService: async def apply_default_model(self, context: TenantContext, schema: list[dict], parameters: dict) -> dict: import copy - fields = [field for field in schema if field.get('type') in ( - 'llm-model-selector', 'select-llm-model', 'model-fallback-selector', - ) and field.get('name')] + + fields = [ + field + for field in schema + if field.get('type') + in ( + 'llm-model-selector', + 'select-llm-model', + 'model-fallback-selector', + ) + and field.get('name') + ] if not fields: return parameters model_uuid = await self.get_default_model(context) @@ -264,6 +280,7 @@ class LLMModelsService: elif not result.get(name): result[name] = model_uuid return result + async def get_llm_models(self, context: TenantContext, include_secret: bool = False) -> list[dict]: """Get all LLM models with provider info""" result = await self.ap.persistence_mgr.execute_async( diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py index fc3e30813..1cb70f0b8 100644 --- a/src/langbot/pkg/api/http/service/monitoring.py +++ b/src/langbot/pkg/api/http/service/monitoring.py @@ -132,6 +132,7 @@ def _workspace_transaction(method): result = await method(self, context, *args, **kwargs) if method.__name__ in {'record_message', 'update_message_status'}: from ....utils.inflight import inflight_hub + inflight_hub.notify(workspace_uuid) return result @@ -1430,10 +1431,13 @@ class MonitoringService(ExecutionDetailsMixin): pipeline_timing = await self.ap.persistence_mgr.execute_async( sqlalchemy.select(AgentRun.started_at, AgentRun.finished_at) .select_from(Message) - .join(AgentRun, sqlalchemy.and_( - AgentRun.workspace_id == workspace_uuid, - AgentRun.run_id == Message.run_id, - )) + .join( + AgentRun, + sqlalchemy.and_( + AgentRun.workspace_id == workspace_uuid, + AgentRun.run_id == Message.run_id, + ), + ) .where( *pipeline_conditions, Message.status == 'success', @@ -1443,10 +1447,7 @@ class MonitoringService(ExecutionDetailsMixin): .order_by(Message.timestamp.desc()) .limit(self._detail_limit()) ) - durations.extend( - (row[1] - row[0]).total_seconds() * 1000 - for row in pipeline_timing.all() if row[0] and row[1] - ) + durations.extend((row[1] - row[0]).total_seconds() * 1000 for row in pipeline_timing.all() if row[0] and row[1]) durations.sort() pipeline_status_result = await self.ap.persistence_mgr.execute_async( sqlalchemy.select( @@ -1679,21 +1680,39 @@ class MonitoringService(ExecutionDetailsMixin): workspace = require_workspace_uuid(context) active = await self.get_executions(context, statuses=['running', 'queued'], limit=100, include_summary=False) recent = await self.get_executions( - context, start_time=datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds=30), - limit=100, include_summary=False, + context, + start_time=datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds=30), + limit=100, + include_summary=False, ) items = {(r['source'], r['id']): r for r in recent['items'] + active['items'] if r['source'] != 'event'} # Keep the actual terminal state of older runs that just left the active set. for kind, model, column, scope in [ - ('agent', persistence_agent_run.AgentRun, persistence_agent_run.AgentRun.run_id, persistence_agent_run.AgentRun.workspace_id), - ('pipeline', persistence_monitoring.MonitoringMessage, persistence_monitoring.MonitoringMessage.id, persistence_monitoring.MonitoringMessage.workspace_uuid), + ( + 'agent', + persistence_agent_run.AgentRun, + persistence_agent_run.AgentRun.run_id, + persistence_agent_run.AgentRun.workspace_id, + ), + ( + 'pipeline', + persistence_monitoring.MonitoringMessage, + persistence_monitoring.MonitoringMessage.id, + persistence_monitoring.MonitoringMessage.workspace_uuid, + ), ]: missing = [r['id'] for r in (previous or [])[:100] if r['source'] == kind and (kind, r['id']) not in items] if missing: - records = await self._execution_rows(workspace, sqlalchemy.select(model).where(scope == workspace, column.in_(missing))) + records = await self._execution_rows( + workspace, sqlalchemy.select(model).where(scope == workspace, column.in_(missing)) + ) names = await self._agent_names(workspace, records) if kind == 'agent' else {} for record in records: - row = self._serialize_agent_execution(record, names) if kind == 'agent' else self._serialize_pipeline_execution(record) + row = ( + self._serialize_agent_execution(record, names) + if kind == 'agent' + else self._serialize_pipeline_execution(record) + ) items[(kind, row['id'])] = row rows = list(items.values()) Run = persistence_agent_run.AgentRun @@ -1701,18 +1720,28 @@ class MonitoringService(ExecutionDetailsMixin): agent_ids = [r['id'] for r in rows if r['source'] == 'agent'] pipeline_ids = [r['id'] for r in rows if r['source'] == 'pipeline'] linked = await self.ap.persistence_mgr.execute_async( - sqlalchemy.select(Run.run_id, Message.id.label('message_id')).outerjoin( - Message, sqlalchemy.and_(Message.workspace_uuid == workspace, Message.run_id == Run.run_id) - ).where(Run.workspace_id == workspace, sqlalchemy.or_(Run.run_id.in_(agent_ids), Message.id.in_(pipeline_ids))) + sqlalchemy.select(Run.run_id, Message.id.label('message_id')) + .outerjoin(Message, sqlalchemy.and_(Message.workspace_uuid == workspace, Message.run_id == Run.run_id)) + .where( + Run.workspace_id == workspace, sqlalchemy.or_(Run.run_id.in_(agent_ids), Message.id.in_(pipeline_ids)) + ) ) - identities = {r.run_id: ('pipeline', r.message_id) if r.message_id in pipeline_ids else ('agent', r.run_id) for r in linked.all()} + identities = { + r.run_id: ('pipeline', r.message_id) if r.message_id in pipeline_ids else ('agent', r.run_id) + for r in linked.all() + } inflight_hub.track(workspace, identities) Event = persistence_agent_run.AgentRunEvent - latest = sqlalchemy.select(Event.run_id, sqlalchemy.func.max(Event.sequence).label('seq')).where( - Event.run_id.in_(identities) - ).group_by(Event.run_id).subquery() + latest = ( + sqlalchemy.select(Event.run_id, sqlalchemy.func.max(Event.sequence).label('seq')) + .where(Event.run_id.in_(identities)) + .group_by(Event.run_id) + .subquery() + ) phases = await self.ap.persistence_mgr.execute_async( - sqlalchemy.select(Event.run_id, Event.type).join(latest, sqlalchemy.and_(Event.run_id == latest.c.run_id, Event.sequence == latest.c.seq)) + sqlalchemy.select(Event.run_id, Event.type).join( + latest, sqlalchemy.and_(Event.run_id == latest.c.run_id, Event.sequence == latest.c.seq) + ) ) for run_id, event_type in phases.all(): row = items.get(identities[run_id]) @@ -1951,7 +1980,10 @@ class MonitoringService(ExecutionDetailsMixin): if end_time: query = query.where(source.c.timestamp <= end_time) return await self._session_projection_page( - workspace_uuid, query.order_by(source.c.timestamp.desc(), source.c.id.desc()), limit, offset, + workspace_uuid, + query.order_by(source.c.timestamp.desc(), source.c.id.desc()), + limit, + offset, ) conditions = [persistence_monitoring.MonitoringMessage.workspace_uuid == workspace_uuid] @@ -2204,47 +2236,99 @@ class MonitoringService(ExecutionDetailsMixin): workspace = require_workspace_uuid(context) execute = self.ap.persistence_mgr.execute_async source = self._session_message_source(workspace, [bot_id]) - exists = (await execute(sqlalchemy.select(source.c.session_id).where( - source.c.session_id == session_id, - ).limit(1))).first() + exists = ( + await execute( + sqlalchemy.select(source.c.session_id) + .where( + source.c.session_id == session_id, + ) + .limit(1) + ) + ).first() if exists is None: session = persistence_monitoring.MonitoringSession - exists = (await execute(sqlalchemy.select(session.session_id).where( - session.workspace_uuid == workspace, session.bot_id == bot_id, session.session_id == session_id, - ).limit(1))).first() + exists = ( + await execute( + sqlalchemy.select(session.session_id) + .where( + session.workspace_uuid == workspace, + session.bot_id == bot_id, + session.session_id == session_id, + ) + .limit(1) + ) + ).first() if exists is None: raise LookupError('Session not found') # Legacy sessions use uppercase PERSON/GROUP; envelope IDs use lowercase. conversation = re.sub(r'^(PERSON|GROUP)_', lambda match: match[1].lower() + '_', session_id) run = persistence_agent_run.AgentRun - busy = (await execute(sqlalchemy.select(run.run_id).where( - run.workspace_id == workspace, run.bot_id == bot_id, run.conversation_id == conversation, - run.status.in_(['created', 'queued', 'claimed', 'running']), - ).limit(1))).first() + busy = ( + await execute( + sqlalchemy.select(run.run_id) + .where( + run.workspace_id == workspace, + run.bot_id == bot_id, + run.conversation_id == conversation, + run.status.in_(['created', 'queued', 'claimed', 'running']), + ) + .limit(1) + ) + ).first() message = persistence_monitoring.MonitoringMessage - pending = (await execute(sqlalchemy.select(message.id).where( - message.workspace_uuid == workspace, message.bot_id == bot_id, - message.session_id.in_([session_id, conversation]), message.status == 'pending', - ).limit(1))).first() + pending = ( + await execute( + sqlalchemy.select(message.id) + .where( + message.workspace_uuid == workspace, + message.bot_id == bot_id, + message.session_id.in_([session_id, conversation]), + message.status == 'pending', + ) + .limit(1) + ) + ).first() interaction = persistence_agent_interaction.AgentInteraction - waiting = (await execute(sqlalchemy.select(interaction.id).where( - interaction.workspace_id == workspace, interaction.bot_id == bot_id, - interaction.conversation_id == conversation, interaction.status.in_(['pending', 'submitted']), - sqlalchemy.or_(interaction.expires_at.is_(None), interaction.expires_at > datetime.datetime.utcnow()), - ).limit(1))).first() + waiting = ( + await execute( + sqlalchemy.select(interaction.id) + .where( + interaction.workspace_id == workspace, + interaction.bot_id == bot_id, + interaction.conversation_id == conversation, + interaction.status.in_(['pending', 'submitted']), + sqlalchemy.or_( + interaction.expires_at.is_(None), interaction.expires_at > datetime.datetime.utcnow() + ), + ) + .limit(1) + ) + ).first() if busy or pending or waiting: raise RuntimeError('Session is processing a task; wait until it finishes') - await execute(sqlalchemy.delete(RunnerState).where( - RunnerState.workspace_id == workspace, RunnerState.bot_id == bot_id, - RunnerState.conversation_id == conversation, RunnerState.scope == 'conversation', - )) + await execute( + sqlalchemy.delete(RunnerState).where( + RunnerState.workspace_id == workspace, + RunnerState.bot_id == bot_id, + RunnerState.conversation_id == conversation, + RunnerState.scope == 'conversation', + ) + ) generation = str(uuid.uuid4()) - await execute(sqlalchemy.insert(RunnerState).values( - runner_id=HOST_RESET_RUNNER, binding_identity=HOST_RESET_RUNNER, scope='conversation', - scope_key=reset_scope(workspace, bot_id, conversation), state_key='generation', - value_json=json.dumps(generation), workspace_id=workspace, bot_id=bot_id, - conversation_id=conversation, created_at=datetime.datetime.utcnow(), - )) + await execute( + sqlalchemy.insert(RunnerState).values( + runner_id=HOST_RESET_RUNNER, + binding_identity=HOST_RESET_RUNNER, + scope='conversation', + scope_key=reset_scope(workspace, bot_id, conversation), + state_key='generation', + value_json=json.dumps(generation), + workspace_id=workspace, + bot_id=bot_id, + conversation_id=conversation, + created_at=datetime.datetime.utcnow(), + ) + ) return {'reset': True, 'session_id': session_id} async def get_sessions( @@ -2266,7 +2350,14 @@ class MonitoringService(ExecutionDetailsMixin): workspace_uuid = require_workspace_uuid(context) if bot_ids and not pipeline_ids and not execution_statuses and mode == 'all': return await self.get_bot_conversation_sessions( - workspace_uuid, bot_ids, start_time, end_time, user_query, is_active, limit, offset, + workspace_uuid, + bot_ids, + start_time, + end_time, + user_query, + is_active, + limit, + offset, ) conditions = [persistence_monitoring.MonitoringSession.workspace_uuid == workspace_uuid] diff --git a/src/langbot/pkg/api/http/service/monitoring_execution_details.py b/src/langbot/pkg/api/http/service/monitoring_execution_details.py index 8bfce101d..467a8026f 100644 --- a/src/langbot/pkg/api/http/service/monitoring_execution_details.py +++ b/src/langbot/pkg/api/http/service/monitoring_execution_details.py @@ -104,16 +104,24 @@ class ExecutionDetailsMixin: message = models.MonitoringMessage conversation = sa.func.coalesce( EventLog.conversation_id, - sa.select(AgentRun.conversation_id).where( + sa.select(AgentRun.conversation_id) + .where( AgentRun.workspace_id == workspace, AgentRun.bot_id == EventLog.bot_id, AgentRun.event_id == EventLog.event_id, - ).order_by(AgentRun.id).limit(1).scalar_subquery(), + ) + .order_by(AgentRun.id) + .limit(1) + .scalar_subquery(), sa.case( - (self._execution_json_text(EventLog.input_json, 'chat_type') == 'group', - sa.literal('group_') + self._execution_json_text(EventLog.input_json, 'chat_id')), - (self._execution_json_text(EventLog.input_json, 'chat_type') == 'private', - sa.literal('person_') + self._execution_json_text(EventLog.input_json, ('sender', 'id'))), + ( + self._execution_json_text(EventLog.input_json, 'chat_type') == 'group', + sa.literal('group_') + self._execution_json_text(EventLog.input_json, 'chat_id'), + ), + ( + self._execution_json_text(EventLog.input_json, 'chat_type') == 'private', + sa.literal('person_') + self._execution_json_text(EventLog.input_json, ('sender', 'id')), + ), ), ) values = { @@ -122,8 +130,12 @@ class ExecutionDetailsMixin: 'timestamp': sa.func.coalesce(EventLog.event_time, EventLog.created_at), 'bot_id': EventLog.bot_id, 'bot_name': sa.func.coalesce(self._execution_json_text(EventLog.metadata_json, 'bot_name'), ''), - 'pipeline_id': sa.func.coalesce(self._execution_json_text(EventLog.metadata_json, ('routes', 0, 'target_uuid')), ''), - 'pipeline_name': sa.func.coalesce(self._execution_json_text(EventLog.metadata_json, ('routes', 0, 'target_type')), ''), + 'pipeline_id': sa.func.coalesce( + self._execution_json_text(EventLog.metadata_json, ('routes', 0, 'target_uuid')), '' + ), + 'pipeline_name': sa.func.coalesce( + self._execution_json_text(EventLog.metadata_json, ('routes', 0, 'target_type')), '' + ), 'message_content': sa.func.coalesce(EventLog.input_json, EventLog.input_summary, ''), 'session_id': conversation, 'status': sa.func.coalesce(self._execution_json_text(EventLog.metadata_json, 'status'), 'success'), @@ -137,45 +149,62 @@ class ExecutionDetailsMixin: } columns = list(message.__table__.columns) stored = sa.select(*columns).where(message.workspace_uuid == workspace, message.bot_id.in_(bot_ids)) - incoming = sa.select(*[ - values.get(column.name, sa.literal(None)).label(column.name) for column in columns - ]).where( + incoming = sa.select( + *[values.get(column.name, sa.literal(None)).label(column.name) for column in columns] + ).where( EventLog.workspace_id == workspace, EventLog.bot_id.in_(bot_ids), EventLog.source == 'platform', EventLog.event_type == 'message.received', conversation.is_not(None), conversation != '', - ~sa.exists(sa.select(message.id).where( - message.workspace_uuid == workspace, - message.bot_id == EventLog.bot_id, - sa.func.coalesce(message.role, 'user') != 'assistant', - sa.or_(message.event_id == EventLog.event_id, message.id == EventLog.event_id), - )), + ~sa.exists( + sa.select(message.id).where( + message.workspace_uuid == workspace, + message.bot_id == EventLog.bot_id, + sa.func.coalesce(message.role, 'user') != 'assistant', + sa.or_(message.event_id == EventLog.event_id, message.id == EventLog.event_id), + ) + ), ) return sa.union_all(stored, incoming).subquery() async def _session_projection_page(self, workspace, statement, limit, offset): - count = (await self._execution_query(workspace, sa.select(sa.func.count()).select_from(statement.subquery()))).scalar_one() + count = ( + await self._execution_query(workspace, sa.select(sa.func.count()).select_from(statement.subquery())) + ).scalar_one() rows = (await self._execution_query(workspace, statement.limit(limit).offset(offset))).mappings().all() return [ {key: value.isoformat() if isinstance(value, datetime.datetime) else value for key, value in row.items()} for row in rows ], count - async def get_bot_conversation_sessions(self, workspace, bot_ids, start_time, end_time, user_query, is_active, limit, offset): + async def get_bot_conversation_sessions( + self, workspace, bot_ids, start_time, end_time, user_query, is_active, limit, offset + ): """Read session summaries across processor types without mutating history.""" messages = self._session_message_source(workspace, bot_ids) key = [messages.c.workspace_uuid, messages.c.bot_id, messages.c.session_id] - ranked = sa.select( - *messages.c, - sa.func.row_number().over(partition_by=key, order_by=[ - sa.case((messages.c.role == 'user', 0), else_=1), messages.c.timestamp.desc(), messages.c.id.desc(), - ]).label('position'), - sa.func.min(messages.c.timestamp).over(partition_by=key).label('first_seen'), - sa.func.max(messages.c.timestamp).over(partition_by=key).label('last_seen'), - sa.func.count().over(partition_by=key).label('count'), - ).where(messages.c.session_id != '').subquery() + ranked = ( + sa.select( + *messages.c, + sa.func.row_number() + .over( + partition_by=key, + order_by=[ + sa.case((messages.c.role == 'user', 0), else_=1), + messages.c.timestamp.desc(), + messages.c.id.desc(), + ], + ) + .label('position'), + sa.func.min(messages.c.timestamp).over(partition_by=key).label('first_seen'), + sa.func.max(messages.c.timestamp).over(partition_by=key).label('last_seen'), + sa.func.count().over(partition_by=key).label('count'), + ) + .where(messages.c.session_id != '') + .subquery() + ) session = models.MonitoringSession fields = list(session.__table__.columns) projected = { @@ -184,15 +213,21 @@ class ExecutionDetailsMixin: 'message_count': ranked.c['count'], 'is_active': sa.literal(True), } - derived = sa.select(*[ - (projected[field.name] if field.name in projected else ranked.c[field.name]).label(field.name) - for field in fields - ]).where(ranked.c.position == 1) + derived = sa.select( + *[ + (projected[field.name] if field.name in projected else ranked.c[field.name]).label(field.name) + for field in fields + ] + ).where(ranked.c.position == 1) existing = sa.select(*fields).where(session.workspace_uuid == workspace, session.bot_id.in_(bot_ids)) combined = sa.union_all(existing, derived).subquery() group = [combined.c.workspace_uuid, combined.c.bot_id, combined.c.session_id] merged = sa.select( - *[combined.c[field.name] for field in fields if field.name not in {'message_count', 'start_time', 'last_activity'}], + *[ + combined.c[field.name] + for field in fields + if field.name not in {'message_count', 'start_time', 'last_activity'} + ], sa.func.max(combined.c.message_count).over(partition_by=group).label('message_count'), sa.func.min(combined.c.start_time).over(partition_by=group).label('start_time'), sa.func.max(combined.c.last_activity).over(partition_by=group).label('last_activity'), @@ -209,7 +244,10 @@ class ExecutionDetailsMixin: if is_active is not None: query = query.where(merged.c.is_active == is_active) return await self._session_projection_page( - workspace, query.order_by(merged.c.last_activity.desc(), merged.c.session_id), limit, offset, + workspace, + query.order_by(merged.c.last_activity.desc(), merged.c.session_id), + limit, + offset, ) def _execution_json_text(self, column, key): @@ -335,12 +373,10 @@ class ExecutionDetailsMixin: # Delivered pipeline ingress is not an unhandled execution, including # older queued/aggregated messages whose trace identity was lost. ~sa.and_( - sa.func.coalesce( - self._execution_json_text(EventLog.metadata_json, ('routes', 0, 'target_type')), '' - ) == 'pipeline', - sa.func.coalesce( - self._execution_json_text(EventLog.metadata_json, ('routes', 0, 'status')), '' - ) == 'delivered', + sa.func.coalesce(self._execution_json_text(EventLog.metadata_json, ('routes', 0, 'target_type')), '') + == 'pipeline', + sa.func.coalesce(self._execution_json_text(EventLog.metadata_json, ('routes', 0, 'status')), '') + == 'delivered', ), ~sa.exists( sa.select(AgentRun.id).where( @@ -400,11 +436,13 @@ class ExecutionDetailsMixin: 'runner_id': None, 'debug': row.source == 'debug', 'has_error': status == 'failed', - 'status_reason': metadata.get('reason') or '; '.join( + 'status_reason': metadata.get('reason') + or '; '.join( route.get('reason') or route.get('failure_code') or 'failure_reason_not_recorded' for route in metadata.get('routes', []) if isinstance(route, dict) and route.get('status') == 'failed' - ) or None, + ) + or None, } async def enrich_execution_rows(self, workspace, items): @@ -415,10 +453,13 @@ class ExecutionDetailsMixin: linked = await self._execution_query( workspace, sa.select(models.MonitoringMessage.id, AgentRun) - .join(AgentRun, sa.and_( - AgentRun.workspace_id == workspace, - AgentRun.run_id == models.MonitoringMessage.run_id, - )) + .join( + AgentRun, + sa.and_( + AgentRun.workspace_id == workspace, + AgentRun.run_id == models.MonitoringMessage.run_id, + ), + ) .where( models.MonitoringMessage.workspace_uuid == workspace, models.MonitoringMessage.id.in_(pipeline_items), @@ -426,10 +467,9 @@ class ExecutionDetailsMixin: ) for message_id, run in linked.all(): metrics = self._serialize_agent_execution(run) - pipeline_items[message_id].update({ - key: metrics[key] - for key in ('started_at_ms', 'finished_at_ms', 'duration_ms', 'usage', 'cost') - }) + pipeline_items[message_id].update( + {key: metrics[key] for key in ('started_at_ms', 'finished_at_ms', 'duration_ms', 'usage', 'cost')} + ) calls = models.MonitoringLLMCall usage_rows = await self._execution_query( @@ -439,10 +479,12 @@ class ExecutionDetailsMixin: sa.func.sum(calls.input_tokens), sa.func.sum(calls.output_tokens), sa.func.sum(calls.total_tokens), - ).where( + ) + .where( calls.workspace_uuid == workspace, calls.message_id.in_(pipeline_items), - ).group_by(calls.message_id), + ) + .group_by(calls.message_id), ) for message_id, input_tokens, output_tokens, total_tokens in usage_rows.all(): if not pipeline_items[message_id].get('usage'): diff --git a/src/langbot/pkg/api/http/service/pipeline.py b/src/langbot/pkg/api/http/service/pipeline.py index 15ef3408b..bdd24f3d7 100644 --- a/src/langbot/pkg/api/http/service/pipeline.py +++ b/src/langbot/pkg/api/http/service/pipeline.py @@ -69,7 +69,9 @@ class PipelineService: runner_config.setdefault('expire-time', 0) ai_config['runner_config'] = {selected.id: self._get_default_values_from_schema(selected.config_schema)} ai_config['runner_config'][selected.id] = await self.ap.llm_model_service.apply_default_model( - context, selected.config_schema, ai_config['runner_config'][selected.id], + context, + selected.config_schema, + ai_config['runner_config'][selected.id], ) return config diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index 6a97430e1..77fa0ade2 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/src/langbot/pkg/api/mcp/server.py @@ -207,35 +207,65 @@ class LangBotMCPServer: return _dump(await ap.monitoring_service.get_message_details(context, identifier)) return _dump(await ap.monitoring_service.get_session_analysis(context, identifier)) - @mcp.tool(description='Read a bounded snapshot of active and recently started Workspace executions, including the latest progress event. No percentage is estimated.') + @mcp.tool( + description='Read a bounded snapshot of active and recently started Workspace executions, including the latest progress event. No percentage is estimated.' + ) async def get_inflight_executions() -> str: return _dump(await ap.monitoring_service.get_inflight_snapshot(_authorized(Permission.RESOURCE_VIEW))) - @mcp.tool(description='Reset a bot session model context, preserving monitoring history. Refuses busy sessions.') + @mcp.tool( + description='Reset a bot session model context, preserving monitoring history. Refuses busy sessions.' + ) async def reset_session_context(bot_id: str, session_id: str) -> str: - return _dump(await ap.monitoring_service.reset_session_context( - _authorized(Permission.RESOURCE_MANAGE), bot_id, session_id, - )) + return _dump( + await ap.monitoring_service.reset_session_context( + _authorized(Permission.RESOURCE_MANAGE), + bot_id, + session_id, + ) + ) - @mcp.tool(description='Create/configure a Slack app for an existing Workspace bot draft. Returns a setup session; poll it for the installation authorization URL. The user must approve Slack installation. Configuration tokens stay in memory only.') - async def start_slack_setup(bot_uuid: str, access_token: str, refresh_token: str, - webhook_url: str = '', name: str = 'LangBot', - socket_mode: bool = False, redirect_url: str = '') -> str: + @mcp.tool( + description='Create/configure a Slack app for an existing Workspace bot draft. Returns a setup session; poll it for the installation authorization URL. The user must approve Slack installation. Configuration tokens stay in memory only.' + ) + async def start_slack_setup( + bot_uuid: str, + access_token: str, + refresh_token: str, + webhook_url: str = '', + name: str = 'LangBot', + socket_mode: bool = False, + redirect_url: str = '', + ) -> str: from langbot.pkg.api.http.service.slack_setup import get_slack_setup - return _dump(await get_slack_setup(ap).start( - _authorized(Permission.RESOURCE_MANAGE), bot_uuid=bot_uuid, access_token=access_token, - refresh_token=refresh_token, webhook_url=webhook_url, name=name, - socket_mode=socket_mode, redirect_url=redirect_url, - )) - @mcp.tool(description='Read an owned Slack setup session. Success includes adapter credentials; save them with the existing bot update tool. Socket Mode additionally needs an App-Level Token with connections:write.') + return _dump( + await get_slack_setup(ap).start( + _authorized(Permission.RESOURCE_MANAGE), + bot_uuid=bot_uuid, + access_token=access_token, + refresh_token=refresh_token, + webhook_url=webhook_url, + name=name, + socket_mode=socket_mode, + redirect_url=redirect_url, + ) + ) + + @mcp.tool( + description='Read an owned Slack setup session. Success includes adapter credentials; save them with the existing bot update tool. Socket Mode additionally needs an App-Level Token with connections:write.' + ) async def get_slack_setup_status(session_id: str) -> str: from langbot.pkg.api.http.service.slack_setup import get_slack_setup + return _dump(get_slack_setup(ap).status(_authorized(Permission.RESOURCE_MANAGE), session_id)) - @mcp.tool(description='Discard an owned Slack setup session and its temporary credentials. Does not delete the Slack app.') + @mcp.tool( + description='Discard an owned Slack setup session and its temporary credentials. Does not delete the Slack app.' + ) async def cancel_slack_setup(session_id: str) -> str: from langbot.pkg.api.http.service.slack_setup import get_slack_setup + service = get_slack_setup(ap) service.owned(_authorized(Permission.RESOURCE_MANAGE), session_id) service.discard(session_id) @@ -463,13 +493,17 @@ class LangBotMCPServer: context = _authorized(Permission.RESOURCE_VIEW) return _dump({'uuid': await ap.llm_model_service.get_starred_model(context)}) - @mcp.tool(description='Star one LLM model for this Workspace, replacing the previous choice. Pass null to clear.') + @mcp.tool( + description='Star one LLM model for this Workspace, replacing the previous choice. Pass null to clear.' + ) async def set_starred_model(model_uuid: str | None = None) -> str: context = _authorized(Permission.PROVIDER_SECRET_MANAGE) await ap.llm_model_service.set_starred_model(context, model_uuid) return _dump({'uuid': model_uuid}) - @mcp.tool(description='Resolve the default LLM model: starred model, then the available Space wizard recommendation.') + @mcp.tool( + description='Resolve the default LLM model: starred model, then the available Space wizard recommendation.' + ) async def get_default_model() -> str: context = _authorized(Permission.RESOURCE_MANAGE) return _dump({'uuid': await ap.llm_model_service.get_default_model(context)}) diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index 3a066714f..41aefe6f5 100644 --- a/src/langbot/pkg/core/app.py +++ b/src/langbot/pkg/core/app.py @@ -359,6 +359,7 @@ class Application: async def run(self): self.event_loop_monitor.start() from ..utils.execution_lease import maintain_execution_leases + self.task_mgr.create_task( maintain_execution_leases(self), name='execution-lease-maintenance', diff --git a/src/langbot/pkg/persistence/alembic/versions/0036_execution_host_leases.py b/src/langbot/pkg/persistence/alembic/versions/0036_execution_host_leases.py index 35052f4ab..120d77275 100644 --- a/src/langbot/pkg/persistence/alembic/versions/0036_execution_host_leases.py +++ b/src/langbot/pkg/persistence/alembic/versions/0036_execution_host_leases.py @@ -10,15 +10,32 @@ depends_on = None def upgrade(): + inspector = sa.inspect(op.get_bind()) for table in ('agent_run', 'monitoring_messages'): - op.add_column(table, sa.Column('execution_owner_id', sa.String(36), nullable=True)) - op.add_column(table, sa.Column('execution_lease_expires_at', sa.DateTime(), nullable=True)) - for column in ('execution_owner_id', 'execution_lease_expires_at'): - op.create_index(f'ix_{table}_{column}', table, [column]) + if not inspector.has_table(table): + continue + columns = {c['name'] for c in inspector.get_columns(table)} + indexes = {i['name'] for i in inspector.get_indexes(table)} + for column, column_type in ( + ('execution_owner_id', sa.String(36)), + ('execution_lease_expires_at', sa.DateTime()), + ): + if column not in columns: + op.add_column(table, sa.Column(column, column_type, nullable=True)) + index = f'ix_{table}_{column}' + if index not in indexes: + op.create_index(index, table, [column]) def downgrade(): + inspector = sa.inspect(op.get_bind()) for table in ('agent_run', 'monitoring_messages'): + if not inspector.has_table(table): + continue + columns = {c['name'] for c in inspector.get_columns(table)} + indexes = {i['name'] for i in inspector.get_indexes(table)} for column in ('execution_owner_id', 'execution_lease_expires_at'): - op.drop_index(f'ix_{table}_{column}', table_name=table) - op.drop_column(table, column) + if f'ix_{table}_{column}' in indexes: + op.drop_index(f'ix_{table}_{column}', table_name=table) + if column in columns: + op.drop_column(table, column) diff --git a/src/langbot/pkg/persistence/tenant_uow.py b/src/langbot/pkg/persistence/tenant_uow.py index 1671b83d5..396b20103 100644 --- a/src/langbot/pkg/persistence/tenant_uow.py +++ b/src/langbot/pkg/persistence/tenant_uow.py @@ -489,8 +489,7 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s and type(source.type) is sqlalchemy.Text and getattr(getattr(source, 'table', None), 'name', None) in {'event_log', 'agent_run'} and ( - source.name == 'metadata_json' - or (source.table.name == 'event_log' and source.name == 'input_json') + source.name == 'metadata_json' or (source.table.name == 'event_log' and source.name == 'input_json') ) ) if not (pipeline_json_cas or monitoring_metadata): diff --git a/src/langbot/pkg/pipeline/pipelinemgr.py b/src/langbot/pkg/pipeline/pipelinemgr.py index 7ff0d2b4f..e15be688f 100644 --- a/src/langbot/pkg/pipeline/pipelinemgr.py +++ b/src/langbot/pkg/pipeline/pipelinemgr.py @@ -449,7 +449,9 @@ class RuntimePipeline: try: from ..platform.processing_indicator import processing_indicator - async with processing_indicator(query.adapter, query.launcher_type.value.lower(), query.launcher_id): + async with processing_indicator( + query.adapter, query.launcher_type.value.lower(), query.launcher_id + ): return await self._process_query(query) except asyncio.CancelledError: lane_outcome = 'cancelled' diff --git a/src/langbot/pkg/platform/adapters/dingtalk/adapter.py b/src/langbot/pkg/platform/adapters/dingtalk/adapter.py index bb175f59e..60a14087d 100644 --- a/src/langbot/pkg/platform/adapters/dingtalk/adapter.py +++ b/src/langbot/pkg/platform/adapters/dingtalk/adapter.py @@ -166,9 +166,9 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo assert isinstance(message_source.source_platform_object, DingTalkEvent) incoming_message = message_source.source_platform_object.incoming_message markdown_enabled = self.config.get('markdown_card', False) - text_message = platform_message.MessageChain([ - part for part in message if not isinstance(part, (platform_message.Image, platform_message.File)) - ]) + text_message = platform_message.MessageChain( + [part for part in message if not isinstance(part, (platform_message.Image, platform_message.File))] + ) content, at = await DingTalkMessageConverter.yiri2target(text_message, markdown_enabled) raw = await self.bot.send_message(content, incoming_message, at) await self._send_attachments(message, incoming_message) @@ -206,9 +206,9 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo last_updates[message_id] = now markdown_enabled = self.config.get('markdown_card', False) - text_message = platform_message.MessageChain([ - part for part in message if not isinstance(part, (platform_message.Image, platform_message.File)) - ]) + text_message = platform_message.MessageChain( + [part for part in message if not isinstance(part, (platform_message.Image, platform_message.File))] + ) content, _ = await DingTalkMessageConverter.yiri2target(text_message, markdown_enabled) card_instance, card_instance_id = self.card_instance_id_dict[message_id] if not content and bot_message.content: diff --git a/src/langbot/pkg/platform/adapters/dingtalk/subscriptions.py b/src/langbot/pkg/platform/adapters/dingtalk/subscriptions.py index fb6aad958..3ebabed8d 100644 --- a/src/langbot/pkg/platform/adapters/dingtalk/subscriptions.py +++ b/src/langbot/pkg/platform/adapters/dingtalk/subscriptions.py @@ -14,14 +14,23 @@ def subscription_events(message): headers = message.headers action = headers.event_type or data.get('EventType') or data.get('eventType') or '' raw = {'headers': vars(headers).copy(), 'data': data} - stamp = headers.event_born_time or data.get('TimeStamp') or data.get('EventTime') or data.get('timestamp') or headers.time or 0 + stamp = ( + headers.event_born_time + or data.get('TimeStamp') + or data.get('EventTime') + or data.get('timestamp') + or headers.time + or 0 + ) stamp = float(stamp) common = dict( adapter_name=ADAPTER_NAME, timestamp=stamp / 1000 if stamp > 10_000_000_000 else stamp, source_platform_object=raw, ) - group_id = data.get('OpenConversationId') or data.get('openConversationId') or data.get('chatId') or data.get('ChatId') + group_id = ( + data.get('OpenConversationId') or data.get('openConversationId') or data.get('chatId') or data.get('ChatId') + ) group = entities.UserGroup(id=str(group_id or ''), name=str(data.get('Title') or data.get('title') or '')) owner_id = data.get('Owner') or data.get('owner') if owner_id: @@ -38,13 +47,18 @@ def subscription_events(message): yield events.MemberJoinedEvent(group=group, member=member, inviter=operator, **common) else: yield events.MemberLeftEvent( - group=group, member=member, operator=operator, - is_kicked=action == 'chat_remove_member', **common, + group=group, + member=member, + operator=operator, + is_kicked=action == 'chat_remove_member', + **common, ) elif group_id and action in {'chat_update_title', 'chat_update_owner'}: yield events.GroupInfoUpdatedEvent( - group=group, operator=operator, - changed_fields=['name' if action == 'chat_update_title' else 'owner_id'], **common, + group=group, + operator=operator, + changed_fields=['name' if action == 'chat_update_title' else 'owner_id'], + **common, ) else: yield events.PlatformSpecificEvent(action=action or 'unknown', data=raw, **common) diff --git a/src/langbot/pkg/platform/adapters/lark/adapter.py b/src/langbot/pkg/platform/adapters/lark/adapter.py index c6b677595..df820ae74 100644 --- a/src/langbot/pkg/platform/adapters/lark/adapter.py +++ b/src/langbot/pkg/platform/adapters/lark/adapter.py @@ -570,7 +570,9 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte ContentCardElementRequest.builder() .card_id(card_id) .element_id('streaming_txt') - .request_body(ContentCardElementRequestBody.builder().content(streaming_text(content)).sequence(sequence).build()) + .request_body( + ContentCardElementRequestBody.builder().content(streaming_text(content)).sequence(sequence).build() + ) .build() ) response: ContentCardElementResponse = await self.api_client.cardkit.v1.card_element.acontent( @@ -578,7 +580,9 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte ) if not response.success(): if self._streaming_mode_closed(response): - await self._replace_streaming_card(card_id, content, finished=is_final and not bot_message.tool_calls) + await self._replace_streaming_card( + card_id, content, finished=is_final and not bot_message.tool_calls + ) else: raise RuntimeError(f'Lark card_element update failed: {response.code} {response.msg}') self.card_last_update_dict[card_id] = now diff --git a/src/langbot/pkg/platform/adapters/lark/event_converter.py b/src/langbot/pkg/platform/adapters/lark/event_converter.py index cdea19549..b2736009d 100644 --- a/src/langbot/pkg/platform/adapters/lark/event_converter.py +++ b/src/langbot/pkg/platform/adapters/lark/event_converter.py @@ -46,9 +46,7 @@ class LarkEventConverter(abstract_platform_adapter.AbstractEventConverter): operator = cls._event_user(data.get('operator_id')) if event_type == 'im.chat.member.bot.deleted_v1': return [platform_events.BotRemovedFromGroupEvent(group=group, operator=operator, **common)] - if event_type in { - 'im.chat.member.user.added_v1', 'im.chat.member.user.deleted_v1' - }: + if event_type in {'im.chat.member.user.added_v1', 'im.chat.member.user.deleted_v1'}: events = [] for user in data.get('users') or []: member = cls._event_user(user.get('user_id'), user.get('name') or '') @@ -56,15 +54,25 @@ class LarkEventConverter(abstract_platform_adapter.AbstractEventConverter): continue if event_type == 'im.chat.member.user.added_v1': inviter = operator if operator and operator.id != member.id else None - events.append(platform_events.MemberJoinedEvent( - group=group, member=member, inviter=inviter, - join_type='invite' if inviter else None, **common, - )) + events.append( + platform_events.MemberJoinedEvent( + group=group, + member=member, + inviter=inviter, + join_type='invite' if inviter else None, + **common, + ) + ) else: - events.append(platform_events.MemberLeftEvent( - group=group, member=member, operator=operator, - is_kicked=operator is not None and operator.id != member.id, **common, - )) + events.append( + platform_events.MemberLeftEvent( + group=group, + member=member, + operator=operator, + is_kicked=operator is not None and operator.id != member.id, + **common, + ) + ) return events if event_type == 'im.chat.updated_v1': after = data.get('after_change') or {} @@ -77,9 +85,14 @@ class LarkEventConverter(abstract_platform_adapter.AbstractEventConverter): group.avatar_url = after.get('avatar') owner = cls._event_user(after.get('owner_id')) group.owner_id = owner.id if owner else None - return [platform_events.GroupInfoUpdatedEvent( - group=group, operator=operator, changed_fields=sorted(changed), **common, - )] + return [ + platform_events.GroupInfoUpdatedEvent( + group=group, + operator=operator, + changed_fields=sorted(changed), + **common, + ) + ] context = message_context or {} if event_type == 'im.message.recalled_v1': return [platform_events.MessageDeletedEvent(message_id=data['message_id'], **context, **common)] @@ -87,11 +100,16 @@ class LarkEventConverter(abstract_platform_adapter.AbstractEventConverter): user = cls._event_user(data.get('user_id')) if user is None: user = platform_entities.User(id=data.get('app_id') or '', is_bot=data.get('operator_type') == 'app') - return [platform_events.MessageReactionEvent( - message_id=data['message_id'], user=user, - reaction=(data.get('reaction_type') or {}).get('emoji_type') or '', - is_add=event_type == 'im.message.reaction.created_v1', **context, **common, - )] + return [ + platform_events.MessageReactionEvent( + message_id=data['message_id'], + user=user, + reaction=(data.get('reaction_type') or {}).get('emoji_type') or '', + is_add=event_type == 'im.message.reaction.created_v1', + **context, + **common, + ) + ] return [] _processed_thread_quote_cache: typing.ClassVar[dict[str, float]] = {} diff --git a/src/langbot/pkg/platform/adapters/lark/stream_card.py b/src/langbot/pkg/platform/adapters/lark/stream_card.py index 88b165eef..332400a82 100644 --- a/src/langbot/pkg/platform/adapters/lark/stream_card.py +++ b/src/langbot/pkg/platform/adapters/lark/stream_card.py @@ -7,7 +7,7 @@ def split_content(content: str) -> tuple[str, str]: reasoning, answer = [], [] inside, offset = False, 0 for marker in re.finditer(r'', content, re.IGNORECASE): - (reasoning if inside else answer).append(content[offset:marker.start()]) + (reasoning if inside else answer).append(content[offset : marker.start()]) inside = marker.group().lower() == '' offset = marker.end() tail = content[offset:] @@ -38,32 +38,62 @@ def build_card(content: str = '', *, streaming: bool = False, finished: bool = T elements.append({'tag': 'markdown', 'content': streaming_text(content), 'element_id': 'streaming_txt'}) else: if reasoning: - elements.append({ - 'tag': 'collapsible_panel', 'expanded': not finished, - 'header': {'title': {'tag': 'plain_text', 'content': '思考过程 / Thinking'}}, - 'elements': [{'tag': 'markdown', 'content': reasoning}], - }) + elements.append( + { + 'tag': 'collapsible_panel', + 'expanded': not finished, + 'header': {'title': {'tag': 'plain_text', 'content': '思考过程 / Thinking'}}, + 'elements': [{'tag': 'markdown', 'content': reasoning}], + } + ) if answer: elements.append({'tag': 'markdown', 'content': answer}) - elements.append({ - 'tag': 'column_set', 'horizontal_spacing': '8px', - 'columns': [ - {'tag': 'column', 'width': 'weighted', 'weight': 1, 'vertical_align': 'center', - 'elements': [{'tag': 'markdown', 'content': '内容由 AI 生成 / AI-generated', - 'text_size': 'notation'}]}, - {'tag': 'column', 'width': 'auto', 'vertical_align': 'center', - 'elements': [{'tag': 'markdown', 'content': '[LangBot](https://langbot.app?utm_source=feishu&utm_medium=bot_card&utm_campaign=langbot)', 'text_size': 'notation', - 'text_align': 'right', - 'icon': {'tag': 'custom_icon', - 'img_key': 'img_v3_02p3_05c65d5d-9bad-440a-a2fb-c89571bfd5bg'}}]}, - ], - }) + elements.append( + { + 'tag': 'column_set', + 'horizontal_spacing': '8px', + 'columns': [ + { + 'tag': 'column', + 'width': 'weighted', + 'weight': 1, + 'vertical_align': 'center', + 'elements': [ + {'tag': 'markdown', 'content': '内容由 AI 生成 / AI-generated', 'text_size': 'notation'} + ], + }, + { + 'tag': 'column', + 'width': 'auto', + 'vertical_align': 'center', + 'elements': [ + { + 'tag': 'markdown', + 'content': '[LangBot](https://langbot.app?utm_source=feishu&utm_medium=bot_card&utm_campaign=langbot)', + 'text_size': 'notation', + 'text_align': 'right', + 'icon': { + 'tag': 'custom_icon', + 'img_key': 'img_v3_02p3_05c65d5d-9bad-440a-a2fb-c89571bfd5bg', + }, + } + ], + }, + ], + } + ) config = {'update_multi': True} if streaming: - config.update(streaming_mode=True, streaming_config={ - 'print_step': {'default': 3}, 'print_frequency_ms': {'default': 40}, 'print_strategy': 'fast', - }) + config.update( + streaming_mode=True, + streaming_config={ + 'print_step': {'default': 3}, + 'print_frequency_ms': {'default': 40}, + 'print_strategy': 'fast', + }, + ) return { - 'schema': '2.0', 'config': config, + 'schema': '2.0', + 'config': config, 'body': {'direction': 'vertical', 'padding': '12px 12px 12px 12px', 'elements': elements}, } diff --git a/src/langbot/pkg/platform/adapters/wecomcs/event_converter.py b/src/langbot/pkg/platform/adapters/wecomcs/event_converter.py index 7bb717495..4c3d76eb2 100644 --- a/src/langbot/pkg/platform/adapters/wecomcs/event_converter.py +++ b/src/langbot/pkg/platform/adapters/wecomcs/event_converter.py @@ -55,7 +55,9 @@ class WecomCSEventConverter(abstract_platform_adapter.AbstractEventConverter): } event_class = event_classes.get(event_type) if event_class: - fields = {key: value for key, value in data.items() if key in event_class.model_fields and value is not None} + fields = { + key: value for key, value in data.items() if key in event_class.model_fields and value is not None + } fields.update(common) fields['data'] = dict(data) actor_id = data.get('servicer_userid') or customer_id diff --git a/src/langbot/pkg/platform/sources/dingtalk.py b/src/langbot/pkg/platform/sources/dingtalk.py index 1401fe036..a8fd2349d 100644 --- a/src/langbot/pkg/platform/sources/dingtalk.py +++ b/src/langbot/pkg/platform/sources/dingtalk.py @@ -613,9 +613,9 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): incoming_message = event.incoming_message markdown_enabled = self.config.get('markdown_card', False) - text_message = platform_message.MessageChain([ - part for part in message if not isinstance(part, (platform_message.Image, platform_message.File)) - ]) + text_message = platform_message.MessageChain( + [part for part in message if not isinstance(part, (platform_message.Image, platform_message.File))] + ) content, at = await DingTalkMessageConverter.yiri2target(text_message, markdown_enabled) await self.bot.send_message(content, incoming_message, at) for part in message: @@ -647,9 +647,9 @@ class DingTalkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): if (msg_seq - 1) % 8 == 0 or is_final: markdown_enabled = self.config.get('markdown_card', False) - text_message = platform_message.MessageChain([ - part for part in message if not isinstance(part, (platform_message.Image, platform_message.File)) - ]) + text_message = platform_message.MessageChain( + [part for part in message if not isinstance(part, (platform_message.Image, platform_message.File))] + ) content, at = await DingTalkMessageConverter.yiri2target(text_message, markdown_enabled) if not content and bot_message.content: content = bot_message.content # 兼容直接传入content的情况 diff --git a/src/langbot/pkg/platform/sources/slack.py b/src/langbot/pkg/platform/sources/slack.py index cbfd50b7d..86481f9a1 100644 --- a/src/langbot/pkg/platform/sources/slack.py +++ b/src/langbot/pkg/platform/sources/slack.py @@ -66,7 +66,6 @@ class SlackEventConverter(abstract_platform_adapter.AbstractEventConverter): yiri_chain = await OmniMessageConverter.target2yiri(event, bot.bot_token) if event.type == 'channel': - sender = platform_entities.GroupMember( id=event.user_id, member_name=str(event.sender_name), @@ -108,7 +107,10 @@ class SlackAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): raise command_errors.ParamNotEnoughError('Slack机器人缺少相关配置项，请查看文档或联系管理员') bot = SlackClient( - bot_token=config['bot_token'], signing_secret=config.get('signing_secret', ''), logger=logger, unified_mode=True + bot_token=config['bot_token'], + signing_secret=config.get('signing_secret', ''), + logger=logger, + unified_mode=True, ) super().__init__( @@ -174,6 +176,7 @@ class SlackAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): if self.config.get('socket_mode', False): await self.bot.run_socket(self.config['app_token']) return + # 统一 webhook 模式下，不启动独立的 Quart 应用 # 保持运行但不启动独立端口 async def keep_alive(): diff --git a/src/langbot/pkg/provider/session/sessionmgr.py b/src/langbot/pkg/provider/session/sessionmgr.py index a43f3c271..4e4f1076b 100644 --- a/src/langbot/pkg/provider/session/sessionmgr.py +++ b/src/langbot/pkg/provider/session/sessionmgr.py @@ -272,11 +272,14 @@ class SessionManager: """获取会话""" session_key, execution_context = _query_session_key(query) from ...agent.runner.context_reset import get_reset_generation + generation = None persistence = getattr(self.ap, 'persistence_mgr', None) if persistence is not None: generation = await get_reset_generation( - persistence.get_db_engine(), execution_context.workspace_uuid, query.bot_uuid, + persistence.get_db_engine(), + execution_context.workspace_uuid, + query.bot_uuid, f'{str(query.launcher_type.value).lower()}_{query.launcher_id}', ) now = time.monotonic() diff --git a/src/langbot/pkg/utils/execution_lease.py b/src/langbot/pkg/utils/execution_lease.py index d25e8f001..037441de7 100644 --- a/src/langbot/pkg/utils/execution_lease.py +++ b/src/langbot/pkg/utils/execution_lease.py @@ -41,44 +41,87 @@ async def reconcile_execution_leases(engine, workspace_uuid, *, now=None, owner_ conditions.extend([AgentRun.queue_name.is_(None), AgentRun.claimed_by_runtime_id.is_(None)]) async with engine.begin() as connection: if engine.dialect.name == 'postgresql': - await connection.execute(sa.text("SELECT set_config('langbot.workspace_uuid', :workspace, true)"), - {'workspace': workspace_uuid}) - await connection.execute(sa.update(model).where( - *conditions, model.execution_owner_id == owner_id, - model.execution_lease_expires_at > now, - ).values(execution_lease_expires_at=now + datetime.timedelta(seconds=LEASE_SECONDS))) + await connection.execute( + sa.text("SELECT set_config('langbot.workspace_uuid', :workspace, true)"), + {'workspace': workspace_uuid}, + ) + await connection.execute( + sa.update(model) + .where( + *conditions, + model.execution_owner_id == owner_id, + model.execution_lease_expires_at > now, + ) + .values(execution_lease_expires_at=now + datetime.timedelta(seconds=LEASE_SECONDS)) + ) expired = [*conditions, model.execution_lease_expires_at <= now] - rows = (await connection.execute(sa.select(identity, scope).where(*expired) - .order_by(model.execution_lease_expires_at).limit(500))).all() + rows = ( + await connection.execute( + sa.select(identity, scope).where(*expired).order_by(model.execution_lease_expires_at).limit(500) + ) + ).all() if not rows: continue - values = {'status': 'failed', 'status_reason': INTERRUPTED_REASON, - 'finished_at': now, 'updated_at': now} if model is AgentRun else { - 'status': 'error', 'level': 'error', - } + values = ( + {'status': 'failed', 'status_reason': INTERRUPTED_REASON, 'finished_at': now, 'updated_at': now} + if model is AgentRun + else { + 'status': 'error', + 'level': 'error', + } + ) if model is MonitoringMessage: - messages = (await connection.execute(sa.select(model).where( - model.id.in_([row[0] for row in rows]), - ))).mappings().all() + messages = ( + ( + await connection.execute( + sa.select(model).where( + model.id.in_([row[0] for row in rows]), + ) + ) + ) + .mappings() + .all() + ) for message in messages: - result = await connection.execute(sa.update(model).where( - *expired, model.id == message['id'], - ).values(**values)) + result = await connection.execute( + sa.update(model) + .where( + *expired, + model.id == message['id'], + ) + .values(**values) + ) changed += result.rowcount if result.rowcount: - await connection.execute(sa.insert(MonitoringError).values( - id=str(uuid.uuid4()), timestamp=now, - error_type='ExecutionInterrupted', error_message=INTERRUPTED_REASON, - message_id=message['id'], - **{key: message[key] for key in ( - 'workspace_uuid', 'bot_id', 'bot_name', 'pipeline_id', - 'pipeline_name', 'session_id', - )}, - )) + await connection.execute( + sa.insert(MonitoringError).values( + id=str(uuid.uuid4()), + timestamp=now, + error_type='ExecutionInterrupted', + error_message=INTERRUPTED_REASON, + message_id=message['id'], + **{ + key: message[key] + for key in ( + 'workspace_uuid', + 'bot_id', + 'bot_name', + 'pipeline_id', + 'pipeline_name', + 'session_id', + ) + }, + ) + ) else: - result = await connection.execute(sa.update(model).where( - *expired, identity.in_([row[0] for row in rows]), - ).values(**values)) + result = await connection.execute( + sa.update(model) + .where( + *expired, + identity.in_([row[0] for row in rows]), + ) + .values(**values) + ) changed += result.rowcount workspaces.update(row[1] for row in rows) for workspace in workspaces: @@ -92,7 +135,8 @@ async def maintain_execution_leases(ap): count = 0 for binding in await ap.workspace_service.list_active_execution_bindings(): count += await reconcile_execution_leases( - ap.persistence_mgr.get_db_engine(), binding.workspace_uuid, + ap.persistence_mgr.get_db_engine(), + binding.workspace_uuid, ) if count: ap.logger.info(f'Marked {count} interrupted executions as failed after host lease expiry') diff --git a/uv.lock b/uv.lock index fcc862510..ed290ba41 100644 --- a/uv.lock +++ b/uv.lock @@ -2185,7 +2185,7 @@ requires-dist = [ { name = "gewechat-client", specifier = ">=0.1.5" }, { name = "html2text", specifier = ">=2024.2.26" }, { name = "httpx", extras = ["socks"], specifier = ">=0.28.1" }, - { name = "langbot-plugin", specifier = "==0.7.11rc1" }, + { name = "langbot-plugin", git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=8c29bc0787be3b514fb379c5154dfc5eaa561cbd" }, { name = "langchain", specifier = ">=1.3.9" }, { name = "langchain-core", specifier = ">=1.3.3" }, { name = "langchain-text-splitters", specifier = ">=1.1.2" }, @@ -2256,7 +2256,7 @@ dev = [ [[package]] name = "langbot-plugin" version = "0.7.11rc1" -source = { registry = "https://pypi.org/simple" } +source = { git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=8c29bc0787be3b514fb379c5154dfc5eaa561cbd#8c29bc0787be3b514fb379c5154dfc5eaa561cbd" } dependencies = [ { name = "aiofiles" }, { name = "aiohttp" }, @@ -2276,10 +2276,6 @@ dependencies = [ { name = "watchdog" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/77/e4/882ed1fd15067e3ca778da2850833ad61716741a6f63d3328028496536b8/langbot_plugin-0.7.11rc1.tar.gz", hash = "sha256:2ef8ba6f1998da365c7ade53573b8cd65f1a660f772ed97385ca445f2b0faa58", size = 690128, upload-time = "2026-10-08T07:49:33.789Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/57/84/21f04d8441c47c910a345a90d5ca5ecc1222269bb951f015f404996b7a36/langbot_plugin-0.7.11rc1-py3-none-any.whl", hash = "sha256:73227613ef827cfcccb04c2535034139ce5315c54dcd32ac01fcbca2150b7d13", size = 442833, upload-time = "2026-10-08T07:49:32.305Z" }, -] [[package]] name = "langchain" diff --git a/web/src/app/home/agents/AgentDetailContent.tsx b/web/src/app/home/agents/AgentDetailContent.tsx index 6d8a65ee2..a371b3ef5 100644 --- a/web/src/app/home/agents/AgentDetailContent.tsx +++ b/web/src/app/home/agents/AgentDetailContent.tsx @@ -63,31 +63,42 @@ export default function AgentDetailContent({ const [deleteConfirmOpen, setDeleteConfirmOpen] = useState(false); const [deleting, setDeleting] = useState(false); const runnerStatusKey = JSON.stringify([ - userInfo?.account_uuid, currentWorkspace?.workspace.uuid, id, i18n.language, + userInfo?.account_uuid, + currentWorkspace?.workspace.uuid, + id, + i18n.language, ]); const [runnerStatusState, setRunnerStatusState] = useState<{ - key: string; status: RunnerStatus; + key: string; + status: RunnerStatus; } | null>(null); - const runnerStatus = runnerStatusState?.key === runnerStatusKey - ? runnerStatusState.status : runnerStatusCache.get(runnerStatusKey) ?? null; - const updateRunnerStatus = useCallback((status: RunnerStatus) => { - setRunnerStatusState((previous) => { - if (status.tone === 'neutral') { - const cached = previous?.key === runnerStatusKey - ? previous.status : runnerStatusCache.get(runnerStatusKey); - if (cached) return { key: runnerStatusKey, status: cached }; + const runnerStatus = + runnerStatusState?.key === runnerStatusKey + ? runnerStatusState.status + : (runnerStatusCache.get(runnerStatusKey) ?? null); + const updateRunnerStatus = useCallback( + (status: RunnerStatus) => { + setRunnerStatusState((previous) => { + if (status.tone === 'neutral') { + const cached = + previous?.key === runnerStatusKey + ? previous.status + : runnerStatusCache.get(runnerStatusKey); + if (cached) return { key: runnerStatusKey, status: cached }; + } + return { key: runnerStatusKey, status }; + }); + if (status.tone !== 'neutral' && !formDirty) { + runnerStatusCache.delete(runnerStatusKey); + runnerStatusCache.set(runnerStatusKey, status); + if (runnerStatusCache.size > 100) { + const oldest = runnerStatusCache.keys().next().value; + if (oldest !== undefined) runnerStatusCache.delete(oldest); + } } - return { key: runnerStatusKey, status }; - }); - if (status.tone !== 'neutral' && !formDirty) { - runnerStatusCache.delete(runnerStatusKey); - runnerStatusCache.set(runnerStatusKey, status); - if (runnerStatusCache.size > 100) { - const oldest = runnerStatusCache.keys().next().value; - if (oldest !== undefined) runnerStatusCache.delete(oldest); - } - } - }, [runnerStatusKey, formDirty]); + }, + [runnerStatusKey, formDirty], + ); const [availableEventTypes, setAvailableEventTypes] = useState([ 'message.received', ]); diff --git a/web/src/app/home/agents/PluginProcessorDetailContent.tsx b/web/src/app/home/agents/PluginProcessorDetailContent.tsx index 10c54910d..938809088 100644 --- a/web/src/app/home/agents/PluginProcessorDetailContent.tsx +++ b/web/src/app/home/agents/PluginProcessorDetailContent.tsx @@ -30,7 +30,10 @@ import PluginProcessorTrace, { ProcessorPayload, } from './components/PluginProcessorTrace'; import ProcessorRunList from './components/ProcessorRunList'; -import { processorRunDuration, formatRunDuration } from './components/processor-run-timing'; +import { + processorRunDuration, + formatRunDuration, +} from './components/processor-run-timing'; import PluginProcessorSettings from './components/PluginProcessorSettings'; import DynamicFormComponent from '@/app/home/components/dynamic-form/DynamicFormComponent'; @@ -55,13 +58,19 @@ export default function PluginProcessorDetailContent({ }) { const { t } = useTranslation(); const [, setSearchParams] = useSearchParams(); - const setActiveTab = useCallback((tab: string) => { - setSearchParams((current) => { - const next = new URLSearchParams(current); - next.set('tab', tab); - return next; - }, { preventScrollReset: true }); - }, [setSearchParams]); + const setActiveTab = useCallback( + (tab: string) => { + setSearchParams( + (current) => { + const next = new URLSearchParams(current); + next.set('tab', tab); + return next; + }, + { preventScrollReset: true }, + ); + }, + [setSearchParams], + ); const [platformTools, setPlatformTools] = useState([]); const toolLabels = Object.fromEntries( platformTools.map((tool) => [tool.name, extractI18nObject(tool.label)]), @@ -83,7 +92,8 @@ export default function PluginProcessorDetailContent({ const [needsInitialSave, setNeedsInitialSave] = useState( Boolean(agent.component_ref) && !agent.supported_event_patterns?.length, ); - const dirty = needsInitialSave || + const dirty = + needsInitialSave || componentRef !== savedConfig.componentRef || !isEqual(parameters, savedConfig.parameters); const [runs, setRuns] = useState([]); @@ -161,22 +171,25 @@ export default function PluginProcessorDetailContent({ [t, available, hasParameters, canOperate, setActiveTab], ); - const applyInstalledProcessor = useCallback((component: RunnerDescriptor) => { - setComponents((current) => [ - ...current.filter((item) => item.id !== component.id), - component, - ]); - setComponentRef(component.id); - setParameters( - Object.fromEntries( - (component.config_schema ?? []) - .filter((field) => field.default !== undefined) - .map((field) => [field.name, field.default]), - ), - ); - setActiveTab('config'); - validate.current = null; - }, [setActiveTab]); + const applyInstalledProcessor = useCallback( + (component: RunnerDescriptor) => { + setComponents((current) => [ + ...current.filter((item) => item.id !== component.id), + component, + ]); + setComponentRef(component.id); + setParameters( + Object.fromEntries( + (component.config_schema ?? []) + .filter((field) => field.default !== undefined) + .map((field) => [field.name, field.default]), + ), + ); + setActiveTab('config'); + validate.current = null; + }, + [setActiveTab], + ); const load = useCallback(async () => { setLoading(true); @@ -364,7 +377,11 @@ export default function PluginProcessorDetailContent({ const logsContent = (
{failed && ( - + )}
@@ -382,104 +399,112 @@ export default function PluginProcessorDetailContent({
- {runs.length > 0 && ( - void openRun(run)} - footer={ - cursor !== null ? ( - - ) : undefined - } - /> - )} - -
- {!selected ? ( - - - {loading - ? t('common.loading') - : t('agents.eventProcessor.noRuns')} - - - - ) : ( - <> -
-
-

- {eventPatternLabel(selected.metadata.event_type ?? '', t)} -

-

- {new Date(selected.created_at * 1000).toLocaleString()} -

-
-
- {processorRunDuration(selected) !== null && ( - - - {formatRunDuration(processorRunDuration(selected)!)} - - )} - - {t(`agents.eventProcessor.status_${selected.status}`, { defaultValue: selected.status })} - -
-
-
-

{t('agents.eventProcessor.input')}

- - {selected.metadata.delivery != null && ( - - )} -
-
-

- - {t('agents.eventProcessor.type')} · {t('agents.eventProcessor.logsTab')} -

- -
- {selected.status === 'failed' && selected.status_reason && ( - - - {selected.status_reason} - - - )} - {eventCursor !== null && ( + {runs.length > 0 && ( + void openRun(run)} + footer={ + cursor !== null ? ( - )} - - )} -
- + ) : undefined + } + /> + )} + +
+ {!selected ? ( + + + {loading + ? t('common.loading') + : t('agents.eventProcessor.noRuns')} + + + + ) : ( + <> +
+
+

+ {eventPatternLabel(selected.metadata.event_type ?? '', t)} +

+

+ {new Date(selected.created_at * 1000).toLocaleString()} +

+
+
+ {processorRunDuration(selected) !== null && ( + + + {formatRunDuration(processorRunDuration(selected)!)} + + )} + + {t(`agents.eventProcessor.status_${selected.status}`, { + defaultValue: selected.status, + })} + +
+
+
+

+ {t('agents.eventProcessor.input')} +

+ + {selected.metadata.delivery != null && ( + + )} +
+
+

+ + {t('agents.eventProcessor.type')} ·{' '} + {t('agents.eventProcessor.logsTab')} +

+ +
+ {selected.status === 'failed' && selected.status_reason && ( + + + {selected.status_reason} + + + )} + {eventCursor !== null && ( + + )} + + )} +
+
); @@ -489,10 +514,13 @@ export default function PluginProcessorDetailContent({ return ( <> -
{ - event.preventDefault(); - void save(); - }} /> + { + event.preventDefault(); + void save(); + }} + /> + {t('agents.eventProcessor.type')} @@ -551,38 +582,40 @@ export default function PluginProcessorDetailContent({ ) : undefined } configTitle={t('agents.eventProcessor.type')} - configIcon={} + configIcon={ + + } monitoring={{ label: t('pipelines.monitoring.title'), workbenchLabel: t('pipelines.monitoring.workbench'), content: logsContent, }} configContent={ -
- {!component ? ( -

- {t('agents.eventProcessor.selectComponent')} -

- ) : component.config_schema.length === 0 ? ( -

- {t('agents.eventProcessor.noSettings')} -

- ) : ( -
- - setParameters(values as Record) - } - onValidate={(fn) => { - validate.current = fn; - }} - /> -
- )} -
+
+ {!component ? ( +

+ {t('agents.eventProcessor.selectComponent')} +

+ ) : component.config_schema.length === 0 ? ( +

+ {t('agents.eventProcessor.noSettings')} +

+ ) : ( +
+ + setParameters(values as Record) + } + onValidate={(fn) => { + validate.current = fn; + }} + /> +
+ )} +
} debugTitle={canOperate ? t('agents.debugTab') : undefined} debugDescription={t('agents.eventProcessor.debugNotice')} diff --git a/web/src/app/home/agents/components/AgentApiToolPicker.tsx b/web/src/app/home/agents/components/AgentApiToolPicker.tsx index 74e0a6f72..812e739ca 100644 --- a/web/src/app/home/agents/components/AgentApiToolPicker.tsx +++ b/web/src/app/home/agents/components/AgentApiToolPicker.tsx @@ -217,7 +217,8 @@ export default function AgentApiToolPicker({ if (checked) selection.add(tool.name); else selection.delete(tool.name); } - if (tools.some((tool) => tool.kind === 'platform')) onPlatformChange([...platform]); + if (tools.some((tool) => tool.kind === 'platform')) + onPlatformChange([...platform]); if (tools.some((tool) => tool.kind === 'host')) onHostChange([...host]); }; @@ -294,8 +295,16 @@ export default function AgentApiToolPicker({ {groupLabel(group)}