mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-21 16:05:19 +08:00
fix(wecombot): improve event routing and sandbox attachment delivery
This commit is contained in:
@@ -577,6 +577,19 @@ class BoxService:
|
||||
*,
|
||||
skip_host_mount_validation: bool = False,
|
||||
) -> dict:
|
||||
result = await self._execute_spec_payload_result(
|
||||
spec_payload, query, skip_host_mount_validation=skip_host_mount_validation
|
||||
)
|
||||
return self._serialize_result(result)
|
||||
|
||||
async def _execute_spec_payload_result(
|
||||
self,
|
||||
spec_payload: dict,
|
||||
query: pipeline_query.Query,
|
||||
*,
|
||||
skip_host_mount_validation: bool = False,
|
||||
) -> BoxExecutionResult:
|
||||
"""Execute within the bound Workspace and preserve the raw result for file transfers."""
|
||||
if not self._available:
|
||||
raise BoxError(
|
||||
'Box runtime is not available. Configure an available Box backend before using Box features.'
|
||||
@@ -635,7 +648,7 @@ class BoxService:
|
||||
f'query_id={query.query_id} '
|
||||
f'summary={json.dumps(self._summarize_result(result), ensure_ascii=False)}'
|
||||
)
|
||||
return self._serialize_result(result)
|
||||
return result
|
||||
|
||||
def resolve_box_session_id(self, query: pipeline_query.Query) -> str:
|
||||
"""Use the Box explicitly bound by the current Runner invocation."""
|
||||
@@ -1206,16 +1219,17 @@ class BoxService:
|
||||
async def _read_outbox_via_exec(self, query: pipeline_query.Query) -> list[dict]:
|
||||
"""Fallback: read the outbox over the exec channel (E2B / remote).
|
||||
|
||||
Uses ``client.execute`` directly (bypassing ``_serialize_result``)
|
||||
so stdout is NOT truncated by ``output_limit_chars`` - the raw
|
||||
base64 payload can be far larger than the 4000-char display limit.
|
||||
Use the same Workspace-scoped execution path as tools, retaining raw
|
||||
stdout because base64 payloads can exceed the display truncation limit.
|
||||
"""
|
||||
import json as _json
|
||||
|
||||
target_dir = f'{self.OUTBOX_MOUNT_DIR}/{self._attachment_query_key(query)}'
|
||||
max_file_bytes = self._EXEC_FALLBACK_MAX_BYTES
|
||||
# Outbound bytes travel in raw stdout, not argv. Use the regular
|
||||
# attachment limits rather than the smaller inbound command limit.
|
||||
max_file_bytes = self._ATTACHMENT_MAX_BYTES
|
||||
max_files = self._ATTACHMENT_MAX_FILES
|
||||
max_total_bytes = max_file_bytes * max_files
|
||||
max_total_bytes = self._ATTACHMENT_MAX_TOTAL_BYTES
|
||||
max_scan_entries = 1000
|
||||
script = (
|
||||
'import base64, json, os\n'
|
||||
@@ -1248,11 +1262,11 @@ class BoxService:
|
||||
' continue\n'
|
||||
' size = entry.stat(follow_symlinks=False).st_size\n'
|
||||
' if size > max_file_bytes or total_bytes + size > max_total_bytes:\n'
|
||||
' continue\n'
|
||||
" raise RuntimeError('Sandbox output attachment exceeds the export size limit')\n"
|
||||
" with open(entry.path, 'rb') as f:\n"
|
||||
' data = f.read(max_file_bytes + 1)\n'
|
||||
' if len(data) > max_file_bytes or total_bytes + len(data) > max_total_bytes:\n'
|
||||
' continue\n'
|
||||
" raise RuntimeError('Sandbox output attachment exceeds the export size limit')\n"
|
||||
' except OSError:\n'
|
||||
' continue\n'
|
||||
' rel = os.path.relpath(entry.path, target)\n'
|
||||
@@ -1269,17 +1283,26 @@ class BoxService:
|
||||
}
|
||||
if 'extra_mounts' not in spec_payload:
|
||||
spec_payload['extra_mounts'] = self.build_skill_extra_mounts(query)
|
||||
try:
|
||||
spec = self.build_spec(spec_payload)
|
||||
result = await self.client.execute(spec)
|
||||
except Exception:
|
||||
return []
|
||||
result = await self._execute_spec_payload_result(spec_payload, query)
|
||||
if not result.ok:
|
||||
return []
|
||||
if 'Sandbox output attachment exceeds the export size limit' in (result.stderr or ''):
|
||||
raise BoxValidationError(
|
||||
f'Sandbox output attachments exceed the export limit '
|
||||
f'({max_file_bytes} bytes per file, {max_total_bytes} bytes total); files were retained'
|
||||
)
|
||||
raise BoxValidationError(f'Failed to read sandbox outbox (exit code {result.exit_code})')
|
||||
try:
|
||||
return _json.loads(str(result.stdout or '').strip().splitlines()[-1])
|
||||
except Exception:
|
||||
return []
|
||||
entries = _json.loads(str(result.stdout or '').strip().splitlines()[-1])
|
||||
if not isinstance(entries, list) or any(
|
||||
not isinstance(entry, dict)
|
||||
or not isinstance(entry.get('name'), str)
|
||||
or not isinstance(entry.get('b64'), str)
|
||||
for entry in entries
|
||||
):
|
||||
raise ValueError('Invalid attachment list')
|
||||
return entries
|
||||
except (ValueError, IndexError) as exc:
|
||||
raise BoxValidationError('Sandbox outbox returned invalid attachment data') from exc
|
||||
|
||||
async def _clear_outbox(self, query: pipeline_query.Query, host_dir: str | None) -> None:
|
||||
"""Empty the per-query outbox after collection.
|
||||
|
||||
@@ -15,13 +15,24 @@ async def coalesce_stream_results(
|
||||
# Completion snapshots can add attachments and usage to the final
|
||||
# delta. Deliver that richer result once, before closing the card.
|
||||
content = pending.all_content if pending.all_content is not None else pending.content
|
||||
# Multi-round runners stream a cumulative transcript, but complete
|
||||
# with only the last model round. Keep the transcript on the same
|
||||
# card while taking attachments and metadata from the completion.
|
||||
final_round = (
|
||||
isinstance(content, str)
|
||||
and isinstance(result.content, str)
|
||||
and bool(result.content)
|
||||
and content.endswith(result.content)
|
||||
)
|
||||
same_message = (
|
||||
isinstance(result, Message)
|
||||
and result.role == pending.role
|
||||
and result.tool_calls == pending.tool_calls
|
||||
and (not content or result.content == content)
|
||||
and (result.tool_calls or []) == (pending.tool_calls or [])
|
||||
and (not content or result.content == content or final_round)
|
||||
)
|
||||
if not same_message:
|
||||
if same_message and final_round and content != result.content:
|
||||
result = result.model_copy(update={'content': content})
|
||||
elif not same_message:
|
||||
yield pending
|
||||
pending = None
|
||||
snapshot = ''
|
||||
|
||||
@@ -3,6 +3,7 @@ from __future__ import annotations
|
||||
from langbot.pkg.platform.sources.wecombot import WecomBotAdapter as LegacyWecomBotAdapter
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import time
|
||||
import traceback
|
||||
import typing
|
||||
@@ -47,6 +48,7 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
|
||||
_group_cache: dict[str, platform_entities.UserGroup] = {}
|
||||
_member_cache: dict[tuple[str, str], platform_entities.UserGroupMember] = {}
|
||||
_stream_to_monitoring_msg: dict[str, tuple[str, float]] = {}
|
||||
_business_event_ids: dict[tuple[str, str], float] = {}
|
||||
_STREAM_MAPPING_TTL: int = 600
|
||||
|
||||
class Config:
|
||||
@@ -103,6 +105,8 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
|
||||
return [
|
||||
'message.received',
|
||||
'feedback.received',
|
||||
'wecombot.enter_chat',
|
||||
'wecombot.template_card_event',
|
||||
'platform.specific',
|
||||
]
|
||||
|
||||
@@ -131,7 +135,29 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
|
||||
|
||||
_join_text_components = staticmethod(LegacyWecomBotAdapter._join_text_components)
|
||||
_iter_media_components = staticmethod(LegacyWecomBotAdapter._iter_media_components)
|
||||
_send_media = staticmethod(LegacyWecomBotAdapter._send_media)
|
||||
|
||||
@staticmethod
|
||||
async def _send_media(bot, req_id: str, item: dict) -> bool:
|
||||
"""Require upload and delivery acknowledgements instead of dropping attachments."""
|
||||
kind = item['type']
|
||||
payload = item.get('base64') or ''
|
||||
if payload.startswith('data:'):
|
||||
payload = payload.split(',', 1)[-1]
|
||||
try:
|
||||
data = base64.b64decode(payload, validate=True)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f'Invalid {kind} attachment data') from exc
|
||||
if not data:
|
||||
raise ValueError(f'{kind} attachment has no data to upload')
|
||||
result = await bot.upload_media(data, item.get('name') or f'attachment.{kind}', media_type=kind)
|
||||
media_id = result.get('media_id') if isinstance(result, dict) else None
|
||||
if not media_id:
|
||||
raise RuntimeError(f'WeCom {kind} upload failed: no media ID returned')
|
||||
ack = await getattr(bot, f'reply_{kind}')(req_id, media_id)
|
||||
if ack is None or ack.get('errcode', 0) != 0:
|
||||
code = ack.get('errcode') if isinstance(ack, dict) else 'no acknowledgement'
|
||||
raise RuntimeError(f'WeCom {kind} delivery failed: {code}')
|
||||
return True
|
||||
|
||||
async def send_message(
|
||||
self,
|
||||
@@ -145,6 +171,8 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
|
||||
raise NotSupportedError(f'send_message:{target_type}')
|
||||
items = await WecomBotMessageConverter.yiri2target(message)
|
||||
content = self._join_text_components(items)
|
||||
if list(self._iter_media_components(items)):
|
||||
raise NotSupportedError('send_message:media requires an originating message callback')
|
||||
raw = await self.bot.send_message(str(target_id), content)
|
||||
return platform_events.MessageResult(raw={'result': raw})
|
||||
|
||||
@@ -156,6 +184,22 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
|
||||
if context.get('event_type') == 'message.received' and message_id:
|
||||
source = await self.get_message(target_type, target_id, message_id)
|
||||
return await self.reply_message(source, message)
|
||||
if context.get('event_type') == 'wecombot.enter_chat':
|
||||
data = context.get('data') or {}
|
||||
raw_event = data.get('data') or {}
|
||||
if time.time() - float(data.get('timestamp') or 0) >= 5:
|
||||
raise ValueError('WeCom welcome replies must complete within 5 seconds of entering the chat')
|
||||
items = await WecomBotMessageConverter.yiri2target(message)
|
||||
if list(self._iter_media_components(items)):
|
||||
raise NotSupportedError('welcome_reply:media')
|
||||
content = self._join_text_components(items)
|
||||
callback_id = data.get('message_id') if self.config.get('enable-webhook') else raw_event.get('req_id')
|
||||
if not callback_id:
|
||||
raise ValueError('WeCom welcome reply is missing its callback ID')
|
||||
raw = await self.bot.reply_welcome(callback_id, content)
|
||||
if raw is None:
|
||||
raise RuntimeError('WeCom welcome reply was not acknowledged')
|
||||
return platform_events.MessageResult(raw={'result': raw})
|
||||
return await self.send_message(target_type, target_id, message)
|
||||
|
||||
async def reply_message(
|
||||
@@ -168,6 +212,8 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
|
||||
if not isinstance(event, WecomBotEvent):
|
||||
raise ValueError('WeComBot reply_message requires a WecomBotEvent source object')
|
||||
items = await WecomBotMessageConverter.yiri2target(message)
|
||||
if self.config.get('enable-webhook') and list(self._iter_media_components(items)):
|
||||
raise NotSupportedError('reply_media:webhook_mode; use WebSocket mode to deliver files')
|
||||
content = self._join_text_components(items)
|
||||
raw = None
|
||||
from .stream_text import format_stream_text
|
||||
@@ -196,6 +242,8 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
|
||||
if not isinstance(event, WecomBotEvent):
|
||||
raise ValueError('WeComBot reply_message_chunk requires a WecomBotEvent source object')
|
||||
items = await WecomBotMessageConverter.yiri2target(message)
|
||||
if is_final and self.config.get('enable-webhook') and list(self._iter_media_components(items)):
|
||||
raise NotSupportedError('reply_media:webhook_mode; use WebSocket mode to deliver files')
|
||||
from .stream_text import format_stream_text
|
||||
|
||||
content = format_stream_text(self._join_text_components(items))
|
||||
@@ -293,10 +341,13 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
|
||||
interaction_event = interaction_event_from_native(event)
|
||||
if interaction_event is not None:
|
||||
await self._dispatch_eba_event(interaction_event)
|
||||
else:
|
||||
await self._handle_native_event(event)
|
||||
except Exception:
|
||||
await self.logger.error(f'Error in WeComBot interaction callback: {traceback.format_exc()}')
|
||||
|
||||
async def _handle_native_event(self, event: WecomBotEvent):
|
||||
key = None
|
||||
try:
|
||||
if platform_events.FriendMessage in self.listeners or platform_events.GroupMessage in self.listeners:
|
||||
legacy_event = await self.event_converter.target2legacy(event)
|
||||
@@ -305,9 +356,24 @@ class WecomBotAdapter(WecomBotAPIMixin, abstract_platform_adapter.AbstractPlatfo
|
||||
|
||||
eba_event = await self.event_converter.target2yiri(event)
|
||||
if eba_event:
|
||||
if eba_event.type in {'wecombot.enter_chat', 'wecombot.template_card_event'}:
|
||||
message_id = eba_event.message_id
|
||||
now = time.monotonic()
|
||||
self._business_event_ids = {
|
||||
key: seen for key, seen in self._business_event_ids.items() if now - seen < 600
|
||||
}
|
||||
key = (eba_event.type, message_id)
|
||||
if message_id and key in self._business_event_ids:
|
||||
return
|
||||
if message_id:
|
||||
self._business_event_ids[key] = now
|
||||
if len(self._business_event_ids) > 2048:
|
||||
self._business_event_ids.pop(next(iter(self._business_event_ids)))
|
||||
self._cache_event(eba_event)
|
||||
await self._dispatch_eba_event(eba_event)
|
||||
except Exception:
|
||||
if key is not None:
|
||||
self._business_event_ids.pop(key, None)
|
||||
await self.logger.error(f'Error in wecombot native event: {traceback.format_exc()}')
|
||||
|
||||
async def _handle_feedback(self, **kwargs):
|
||||
|
||||
@@ -50,6 +50,36 @@ class WecomBotEventConverter(abstract_platform_adapter.AbstractEventConverter):
|
||||
)
|
||||
|
||||
async def target2yiri(self, event: WecomBotEvent) -> platform_events.Event:
|
||||
from langbot.libs.wecom_ai_bot_api.api import (
|
||||
extract_wecom_event_type,
|
||||
extract_template_card_event_payload,
|
||||
extract_template_card_action,
|
||||
)
|
||||
|
||||
event_type = extract_wecom_event_type(event) or event.get('eventtype')
|
||||
if event_type in {'enter_chat', 'template_card_event'}:
|
||||
group = event.get('chattype', event.type) == 'group'
|
||||
fields = dict(
|
||||
adapter_name=ADAPTER_NAME,
|
||||
user=platform_entities.User(id=event.userid, nickname=event.username),
|
||||
chat_id=str(event.chatid if group else event.userid),
|
||||
chat_type=platform_entities.ChatType.GROUP if group else platform_entities.ChatType.PRIVATE,
|
||||
message_id=event.message_id or '',
|
||||
timestamp=float(event.get('create_time') or time.time()),
|
||||
data=dict(event),
|
||||
source_platform_object=event,
|
||||
)
|
||||
if event_type == 'enter_chat':
|
||||
return platform_events.WecomBotEnterChatEvent(**fields)
|
||||
card = extract_template_card_event_payload(event)
|
||||
task_id, event_key, card_type = extract_template_card_action(card)
|
||||
return platform_events.WecomBotTemplateCardEvent(
|
||||
**fields,
|
||||
task_id=task_id,
|
||||
event_key=event_key,
|
||||
card_type=card_type,
|
||||
selected_items=card.get('selected_items') or card.get('SelectedItems') or [],
|
||||
)
|
||||
if event.type in {'single', 'group'} and event.msgtype != 'event':
|
||||
return await self.message_to_eba(event)
|
||||
return self.platform_specific(
|
||||
|
||||
@@ -174,6 +174,8 @@ spec:
|
||||
supported_events:
|
||||
- message.received
|
||||
- feedback.received
|
||||
- wecombot.enter_chat
|
||||
- wecombot.template_card_event
|
||||
- platform.specific
|
||||
|
||||
supported_apis:
|
||||
|
||||
Reference in New Issue
Block a user