feat(line): Native Event-Based Agent (EBA) LINE Platform Adapter (#2736)

* feat(line): EBA LINE adapter skeleton (WO-LINE-01-SKELETON-1018) - pkg/platform/adapters/line/ 9 files (LineAdapter hybrid inheritance + Capability + event namespace + config env load, zero hardcoded secrets) - LINE SDK calls all stub (NotImplementedError) pending WO-LINE-00/02/03 - Verified: compileall 7/7 + ruff 0 + import chain green against v4.11.0-hotfixed SDK (container-extracted, read-only) - Source: handoff box WO-LINE-01-SKELETON-1018 (sha256 9/9 matched) Co-authored-by: multica-agent <[email protected]> * feat(line): implement WO-LINE-02 message.received loop and converter Co-authored-by: multica-agent <[email protected]> * feat(line): implement WO-LINE-03 streaming and send API Co-authored-by: multica-agent <[email protected]> * feat(line): implement WO-LINE-04 group and user event extensions Co-authored-by: multica-agent <[email protected]> * feat(line): implement WO-LINE-05 group and user event extensions Co-authored-by: multica-agent <[email protected]> * docs(line): add EBA LINE adapter architecture guide and documentation Co-authored-by: multica-agent <[email protected]> * feat(line): resolve Ruff warnings and manifest spec update Co-authored-by: multica-agent <[email protected]> * feat(line): add webhook_url config and help_links to PR_DESCRIPTION.md Co-authored-by: multica-agent <[email protected]> * feat(line): add webhook_url config and help_links for PR #2736 Co-authored-by: multica-agent <[email protected]> * fix(line): ruff format, update help_links, remove nested duplicate dir Co-authored-by: multica-agent <[email protected]> * fix(line): resolve 12 test failures, align types, and move architecture docs to dev-docs Co-authored-by: multica-agent <[email protected]> * fix(line): resolve ruff lint and formatting issues for CI Co-authored-by: multica-agent <[email protected]> * docs(line): sanitize internal orchestration notes and standardize documentation Co-authored-by: multica-agent <[email protected]> * clean(line): remove internal tracking ticket references from source code Co-authored-by: multica-agent <[email protected]> * fix(line): resolve MessageReceivedEvent and Friend entity mapping for 100% test pass Co-authored-by: multica-agent <[email protected]> * fix(line): resolve E402 import order in event_converter.py Co-authored-by: multica-agent <[email protected]> * feat(line): implement the LINE Omni adapter on the legacy messaging flow Replace the placeholder skeleton with a working unified-webhook adapter: - verify X-Line-Signature with WebhookParser and acknowledge before dispatch - build real LINE messaging clients from channel_access_token and drop the adapter's own messages using the account id from get_bot_info - map message/image/voice events to message.received, unsend to message.deleted, membership changes to group/bot events, and the remaining LINE event types to platform.specific with chat identity - send and reply through MessagingApi with LINE's five-message budget, a reply-token to push fallback, and honest NotSupportedError paths - expose get_user_info, get_group_info, leave_group and the get_bot_info, get_profile, get_group_summary pass-through actions - keep the legacy FriendMessage/GroupMessage listeners working - move tests to tests/unit_tests/platform, add the adapter doc, and register line in OMNI_ADAPTER_NAMES Co-authored-by: multica-agent <[email protected]> --------- Co-authored-by: multica-agent <[email protected]> Co-authored-by: RockChinQ <[email protected]>
2026-11-03 22:32:27 +08:00 · 2026-10-30 12:58:04 +08:00
parent 292fe43731
commit d8271d9198
12 changed files with 1373 additions and 0 deletions
@@ -0,0 +1,72 @@
+# LINE EBA Adapter
+
+## Structure
+
+LINE is migrated into `src/langbot/pkg/platform/adapters/line/` with the standard EBA adapter layout:
+
+- `adapter.py` owns lifecycle, unified webhook handling, signature verification, listener dispatch, and outbound send/reply.
+- `event_converter.py` maps LINE webhook events to `message.received`, `message.deleted`, group membership events, and `platform.specific`.
+- `message_converter.py` maps common `MessageChain` components to LINE message objects and maps inbound LINE text/image/voice payloads back to EBA components.
+- `api_impl.py` implements the optional common APIs against the LINE Messaging API.
+- `platform_api.py` declares the safe LINE-specific pass-through actions.
+- `manifest.yaml` declares `line-omni`.
+
+The legacy `src/langbot/pkg/platform/sources/line.py` adapter is kept unchanged.
+
+Request handling reuses the proven behaviour of the legacy adapter: `WebhookParser` verifies `X-Line-Signature`, the webhook acknowledges immediately, and each event is converted and dispatched in a background task.
+
+## Configuration
+
+| Field | Required | Notes |
+|-------|----------|-------|
+| `webhook_url` | No | Generated by LangBot. Paste it into the LINE Developers console and enable "Use webhook". |
+| `channel_access_token` | Yes | Long-lived channel access token, used for all outbound calls. |
+| `channel_secret` | Yes | Channel secret, used to verify `X-Line-Signature`. |
+
+`bot_account_id` is resolved through `get_bot_info` on startup and is used to drop the adapter's own messages.
+
+## Events
+
+| Event | Notes |
+|-------|-------|
+| `message.received` | Text, image, voice, and other message types. Group and room chats are mapped to group chat identity. LINE mentions become `At` components. |
+| `message.deleted` | Emitted for LINE `unsend` events. |
+| `group.member_joined` | Emitted for `memberJoined`; the first listed member is the event subject. |
+| `group.member_left` | Emitted for `memberLeft`. |
+| `bot.invited_to_group` | Emitted when the Official Account joins a group or room (`join`). |
+| `bot.removed_from_group` | Emitted when the Official Account leaves (`leave`). |
+| `platform.specific` | `line.follow`, `line.unfollow`, `line.postback`, `line.beacon`, and any other LINE event type, with `chat_id` in `data`. |
+
+Legacy pipeline bindings still receive `FriendMessage` / `GroupMessage`; the adapter converts the unified event for them.
+
+## Common APIs
+
+Required:
+
+- `send_message`
+- `reply_message`
+
+Optional:
+
+- `get_user_info`
+- `get_group_info`
+- `leave_group`
+- `call_platform_api`
+
+`reply_message` uses the event's reply token and falls back to `push_message` when the token is rejected or already used, so multi-message answers still arrive.
+
+## Platform APIs
+
+| Action | Notes |
+|--------|-------|
+| `get_bot_info` | Returns the Official Account profile. |
+| `get_profile` | Calls `get_profile` for a user id. |
+| `get_group_summary` | Calls `get_group_summary` for a group id. |
+
+## Known Limits
+
+- LINE has no group member listing API, so `get_group_member_list` raises `NotSupportedError`.
+- Outbound images require a public URL: the Messaging API fetches `originalContentUrl`, so a base64-only `Image` component raises `NotSupportedError` instead of sending nothing.
+- LINE has no message edit API; `edit_message` raises `NotSupportedError`. Streaming output is not supported.
+- Requests are capped at five messages per call and 5000 characters per text message; longer chains are split into multiple requests.
+- Offline coverage lives in `tests/unit_tests/platform/test_line_eba_adapter.py`. A real channel (webhook delivery, reply tokens, media download) still needs verification against a live LINE Official Account.
@@ -13,6 +13,7 @@ OMNI_ADAPTER_NAMES = frozenset(
        'wecombot',
        'wecomcs',
        'officialaccount',
+        'line',
    }
 )

@@ -0,0 +1,258 @@
+from __future__ import annotations
+
+import asyncio
+import traceback
+import typing
+
+import pydantic
+import quart
+from linebot.v3.exceptions import InvalidSignatureError
+from linebot.v3.messaging import (
+    ApiClient,
+    Configuration,
+    MessagingApi,
+    MessagingApiBlob,
+    PushMessageRequest,
+    ReplyMessageRequest,
+)
+from linebot.v3.webhook import WebhookParser
+from linebot.v3.webhooks import Event
+
+import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
+import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
+from langbot_plugin.api.entities.builtin.platform import events as platform_events
+from langbot_plugin.api.entities.builtin.platform import message as platform_message
+
+from langbot.pkg.platform.adapters.line.api_impl import LineAPIMixin
+from langbot.pkg.platform.adapters.line.errors import NotSupportedError
+from langbot.pkg.platform.adapters.line.event_converter import LineEventConverter
+from langbot.pkg.platform.adapters.line.message_converter import LineMessageConverter
+from langbot.pkg.platform.adapters.line.platform_api import PLATFORM_API_MAP
+from langbot.pkg.platform.adapters.line.types import ADAPTER_NAME, MAX_MESSAGES_PER_REQUEST
+
+
+class LineAdapter(LineAPIMixin, abstract_platform_adapter.AbstractMessagePlatformAdapter):
+    """LINE Messaging API adapter in unified webhook mode."""
+
+    api_client: typing.Any = pydantic.Field(exclude=True)
+    bot: typing.Any = pydantic.Field(exclude=True)
+    blob_api: typing.Any = pydantic.Field(exclude=True)
+    parser: typing.Any = pydantic.Field(exclude=True)
+
+    message_converter: LineMessageConverter = LineMessageConverter()
+    event_converter: LineEventConverter = LineEventConverter()
+
+    config: dict
+    config_secret: str = ''
+    bot_uuid: str | None = None
+
+    listeners: dict[
+        typing.Type[platform_events.Event],
+        typing.Callable[[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None],
+    ] = {}
+    background_tasks: set = pydantic.Field(default_factory=set, exclude=True)
+
+    def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger):
+        missing_keys = [key for key in ('channel_access_token', 'channel_secret') if not config.get(key)]
+        if missing_keys:
+            raise ValueError(f'LINE adapter missing config: {missing_keys}')
+        configuration = Configuration(access_token=config['channel_access_token'])
+        api_client = ApiClient(configuration)
+        bot = MessagingApi(api_client)
+        blob_api = MessagingApiBlob(api_client)
+        bot_account_id = config.get('bot_account_id', '') or ''
+        super().__init__(
+            config=config,
+            logger=logger,
+            bot_account_id=bot_account_id,
+            config_secret=config['channel_secret'],
+            api_client=api_client,
+            bot=bot,
+            blob_api=blob_api,
+            parser=WebhookParser(config['channel_secret']),
+            listeners={},
+            background_tasks=set(),
+            message_converter=LineMessageConverter(),
+            event_converter=LineEventConverter(bot_account_id=bot_account_id, blob_api=blob_api),
+        )
+
+    def set_bot_uuid(self, bot_uuid: str):
+        self.bot_uuid = bot_uuid
+
+    def get_supported_events(self) -> list[str]:
+        return [
+            'message.received',
+            'message.deleted',
+            'group.member_joined',
+            'group.member_left',
+            'bot.invited_to_group',
+            'bot.removed_from_group',
+            'platform.specific',
+        ]
+
+    def get_supported_apis(self) -> list[str]:
+        return [
+            'send_message',
+            'reply_message',
+            'get_user_info',
+            'get_group_info',
+            'leave_group',
+            'call_platform_api',
+        ]
+
+    async def send_message(
+        self,
+        target_type: str,
+        target_id: str,
+        message: platform_message.MessageChain,
+    ) -> platform_events.MessageResult:
+        messages = await self.message_converter.yiri2target(message)
+        for batch in _batches(messages):
+            request = PushMessageRequest(to=str(target_id), messages=batch)
+            await asyncio.to_thread(self.bot.push_message, request)
+        return platform_events.MessageResult(raw={'sent': len(messages), 'target_id': str(target_id)})
+
+    async def reply_message(
+        self,
+        message_source: platform_events.MessageEvent,
+        message: platform_message.MessageChain,
+        quote_origin: bool = False,
+    ) -> platform_events.MessageResult:
+        source = await LineEventConverter.yiri2target(message_source)
+        messages = await self.message_converter.yiri2target(message)
+        reply_token = str(getattr(source, 'reply_token', '') or '')
+        chat_id = _source_chat_id(source) or str(getattr(message_source, 'chat_id', '') or '')
+        for batch in _batches(messages):
+            if reply_token:
+                try:
+                    await asyncio.to_thread(
+                        self.bot.reply_message, ReplyMessageRequest(reply_token=reply_token, messages=batch)
+                    )
+                    # A reply token is single-use; later batches of the same turn fall back to push.
+                    reply_token = ''
+                    continue
+                except Exception:
+                    await self.logger.warning(
+                        f'LINE reply token rejected, falling back to push: {traceback.format_exc()}'
+                    )
+                    reply_token = ''
+            if not chat_id:
+                raise NotSupportedError('reply_message:line_missing_target')
+            await asyncio.to_thread(self.bot.push_message, PushMessageRequest(to=chat_id, messages=batch))
+        return platform_events.MessageResult(raw={'sent': len(messages)})
+
+    async def call_platform_api(self, action: str, params: dict = {}) -> dict:
+        handler = PLATFORM_API_MAP.get(action)
+        if handler is None:
+            raise NotSupportedError(f'call_platform_api:{action}')
+        return await handler(self, dict(params or {}))
+
+    def register_listener(
+        self,
+        event_type: typing.Type[platform_events.Event],
+        callback: typing.Callable[
+            [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None
+        ],
+    ):
+        self.listeners[event_type] = callback
+
+    def unregister_listener(
+        self,
+        event_type: typing.Type[platform_events.Event],
+        callback: typing.Callable[
+            [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None
+        ],
+    ):
+        registered = self.listeners.get(event_type)
+        if registered is callback:
+            self.listeners.pop(event_type, None)
+
+    async def handle_unified_webhook(self, bot_uuid: str, path: str, request) -> typing.Any:
+        signature = request.headers.get('X-Line-Signature')
+        body = await request.get_data(as_text=True)
+        if not signature:
+            await self.logger.warning('LINE webhook rejected: missing X-Line-Signature header')
+            return quart.Response('Missing X-Line-Signature header', status=400)
+        try:
+            events = self.parser.parse(body, signature)
+        except InvalidSignatureError:
+            await self.logger.warning(
+                'LINE webhook rejected: invalid signature, check channel_access_token/channel_secret'
+            )
+            return quart.Response('Invalid signature', status=400)
+        for event in events or []:
+            task = asyncio.create_task(self._dispatch_native_event(event))
+            self.background_tasks.add(task)
+            task.add_done_callback(self.background_tasks.discard)
+        # Acknowledge immediately; LINE retries any non-200 answer.
+        return quart.Response('OK', status=200)
+
+    async def _dispatch_native_event(self, event: Event) -> None:
+        try:
+            if self.bot_account_id and _sender_id(event) == self.bot_account_id:
+                return
+            eba_event = await self.event_converter.target2yiri(event)
+            if eba_event is None:
+                return
+            if isinstance(eba_event, platform_events.MessageReceivedEvent):
+                legacy_event = eba_event.to_legacy_event()
+                listener = self.listeners.get(type(legacy_event))
+                if listener:
+                    await listener(legacy_event, self)
+            await self._dispatch_eba_event(eba_event)
+        except Exception:
+            await self.logger.error(f'LINE native event error: {traceback.format_exc()}')
+
+    async def _dispatch_eba_event(self, event: platform_events.EBAEvent) -> None:
+        for event_type in (type(event), platform_events.EBAEvent, platform_events.Event):
+            callback = self.listeners.get(event_type)
+            if callback:
+                await callback(event, self)
+                return
+
+    async def is_muted(self, group_id: int | None = None) -> bool:
+        return False
+
+    async def run_async(self):
+        if not self.bot_account_id:
+            try:
+                info = await asyncio.to_thread(self.bot.get_bot_info)
+                self.bot_account_id = info.user_id
+                self.event_converter.bot_account_id = info.user_id
+            except Exception:
+                await self.logger.warning(
+                    'LINE adapter could not resolve its account id; self-message filtering stays disabled'
+                )
+        await self.logger.info(f'{ADAPTER_NAME} adapter running in unified webhook mode')
+        while True:
+            await asyncio.sleep(1)
+
+    async def kill(self) -> bool:
+        for task in list(self.background_tasks):
+            task.cancel()
+        self.background_tasks.clear()
+        await asyncio.to_thread(self.api_client.close)
+        return True
+
+
+def _batches(messages: list[typing.Any]) -> list[list[typing.Any]]:
+    """Split a message array into LINE's per-request budget."""
+    return [
+        messages[index : index + MAX_MESSAGES_PER_REQUEST]
+        for index in range(0, len(messages), MAX_MESSAGES_PER_REQUEST)
+    ]
+
+
+def _sender_id(event: Event) -> str:
+    source = getattr(event, 'source', None)
+    return str(getattr(source, 'user_id', '') or '')
+
+
+def _source_chat_id(event: Event) -> str:
+    """Chat id of the conversation a LINE event belongs to."""
+    source = getattr(event, 'source', None)
+    for attribute in ('group_id', 'room_id', 'user_id'):
+        value = getattr(source, attribute, '') or ''
+        if value:
+            return str(value)
+    return ''
@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import typing
+
+from langbot.pkg.platform.adapters.line.errors import NotSupportedError
+from langbot_plugin.api.entities.builtin.platform import entities as platform_entities
+from langbot_plugin.api.entities.builtin.platform import events as platform_events
+from langbot_plugin.api.entities.builtin.platform import message as platform_message
+
+
+class LineAPIMixin:
+    """Optional universal APIs; LINE only exposes profile and group summaries."""
+
+    async def get_user_info(self, user_id: typing.Union[int, str]) -> platform_entities.User:
+        import asyncio
+
+        profile = await asyncio.to_thread(self.bot.get_profile, str(user_id))
+        return platform_entities.User(
+            id=profile.user_id,
+            nickname=profile.display_name,
+            avatar_url=profile.picture_url or '',
+        )
+
+    async def get_group_info(self, group_id: typing.Union[int, str]) -> platform_entities.UserGroup:
+        import asyncio
+
+        try:
+            summary = await asyncio.to_thread(self.bot.get_group_summary, str(group_id))
+        except Exception as exc:
+            raise NotSupportedError('get_group_info:group_not_found') from exc
+        return platform_entities.UserGroup(
+            id=summary.group_id,
+            name=summary.group_name,
+            avatar_url=summary.picture_url or '',
+        )
+
+    async def leave_group(self, group_id: typing.Union[int, str]) -> None:
+        import asyncio
+
+        await asyncio.to_thread(self.bot.leave_group, str(group_id))
+
+    async def get_group_member_list(
+        self,
+        group_id: typing.Union[int, str],
+    ) -> list[platform_entities.UserGroupMember]:
+        # LINE does not expose a group member listing API.
+        raise NotSupportedError('get_group_member_list')
+
+    async def get_group_list(self) -> list[platform_entities.UserGroup]:
+        raise NotSupportedError('get_group_list')
+
+    async def get_friend_list(self) -> list[platform_entities.User]:
+        raise NotSupportedError('get_friend_list')
+
+    async def edit_message(
+        self,
+        chat_type: str,
+        chat_id: typing.Union[int, str],
+        message_id: typing.Union[int, str],
+        new_content: platform_message.MessageChain,
+    ) -> None:
+        raise NotSupportedError('edit_message')
+
+    async def delete_message(
+        self,
+        chat_type: str,
+        chat_id: typing.Union[int, str],
+        message_id: typing.Union[int, str],
+    ) -> None:
+        raise NotSupportedError('delete_message')
+
+    async def forward_message(
+        self,
+        from_chat_type: str,
+        from_chat_id: typing.Union[int, str],
+        message_id: typing.Union[int, str],
+        to_chat_type: str,
+        to_chat_id: typing.Union[int, str],
+    ) -> platform_events.MessageResult:
+        raise NotSupportedError('forward_message')
+
+    async def upload_file(self, file_data: bytes, filename: str) -> str:
+        raise NotSupportedError('upload_file')
+
+    async def get_file_url(self, file_id: str) -> str:
+        raise NotSupportedError('get_file_url')
@@ -0,0 +1,10 @@
+from __future__ import annotations
+
+try:
+    from langbot_plugin.api.entities.builtin.platform.errors import NotSupportedError
+except ModuleNotFoundError:
+
+    class NotSupportedError(Exception):
+        def __init__(self, api_name: str, *args):
+            super().__init__(f"API '{api_name}' is not supported by this adapter", *args)
+            self.api_name = api_name
@@ -0,0 +1,169 @@
+from __future__ import annotations
+
+import typing
+
+from linebot.v3.webhooks import (
+    BeaconEvent,
+    Event,
+    FollowEvent,
+    GroupSource,
+    JoinEvent,
+    LeaveEvent,
+    MemberJoinedEvent,
+    MemberLeftEvent,
+    MessageEvent,
+    PostbackEvent,
+    RoomSource,
+    UnfollowEvent,
+    UnsendEvent,
+    UserSource,
+)
+
+import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
+from langbot_plugin.api.entities.builtin.platform import entities as platform_entities
+from langbot_plugin.api.entities.builtin.platform import events as platform_events
+
+from langbot.pkg.platform.adapters.line.message_converter import LineMessageConverter
+from langbot.pkg.platform.adapters.line.types import ADAPTER_NAME
+
+
+def _timestamp(event: Event) -> float:
+    try:
+        return float(event.timestamp) / 1000.0
+    except (AttributeError, TypeError, ValueError):
+        return 0.0
+
+
+def _chat_identity(event: Event) -> tuple[platform_entities.ChatType, str, typing.Any, str]:
+    """Return chat type, chat id, group entity and sender id for a LINE event source."""
+    source = event.source
+    if isinstance(source, GroupSource):
+        return (
+            platform_entities.ChatType.GROUP,
+            source.group_id,
+            platform_entities.UserGroup(id=source.group_id, name=source.group_id),
+            source.user_id or source.group_id,
+        )
+    if isinstance(source, RoomSource):
+        # LINE rooms are multi-user chats; treat them as group chats keyed by room id.
+        return (
+            platform_entities.ChatType.GROUP,
+            source.room_id,
+            platform_entities.UserGroup(id=source.room_id, name=source.room_id),
+            source.user_id or source.room_id,
+        )
+    user_id = source.user_id if isinstance(source, UserSource) else ''
+    return platform_entities.ChatType.PRIVATE, user_id, None, user_id
+
+
+class LineEventConverter(abstract_platform_adapter.AbstractEventConverter):
+    """Convert LINE webhook events into unified EBA events and legacy message events."""
+
+    def __init__(self, bot_account_id: str = '', blob_api: typing.Any = None):
+        self.bot_account_id = bot_account_id
+        self.blob_api = blob_api
+
+    @staticmethod
+    async def yiri2target(event: platform_events.Event) -> typing.Any:
+        return getattr(event, 'source_platform_object', None)
+
+    async def target2yiri(self, event: Event) -> typing.Optional[platform_events.EBAEvent]:
+        """Convert one LINE webhook event into a unified event; None when unsupported."""
+        common = dict(
+            adapter_name=ADAPTER_NAME,
+            timestamp=_timestamp(event),
+            source_platform_object=event,
+        )
+        chat_type, chat_id, group, sender_id = _chat_identity(event)
+        chat = dict(
+            chat_type=chat_type,
+            chat_id=chat_id,
+            group=group,
+        )
+        if isinstance(event, MessageEvent):
+            return platform_events.MessageReceivedEvent(
+                **common,
+                **chat,
+                message_id=str(event.message.id),
+                message_chain=await LineMessageConverter.target2yiri(
+                    event, blob_api=self.blob_api, bot_account_id=self.bot_account_id
+                ),
+                sender=platform_entities.User(id=sender_id, nickname=sender_id),
+            )
+        if isinstance(event, UnsendEvent):
+            return platform_events.MessageDeletedEvent(
+                **common,
+                **chat,
+                message_id=str(event.unsend.message_id),
+                operator=platform_entities.User(id=sender_id) if sender_id else None,
+            )
+        if isinstance(event, FollowEvent):
+            return platform_events.PlatformSpecificEvent(
+                **common, action='line.follow', data={**_event_payload(event), 'chat_id': chat_id}
+            )
+        if isinstance(event, UnfollowEvent):
+            return platform_events.PlatformSpecificEvent(
+                **common, action='line.unfollow', data={**_event_payload(event), 'chat_id': chat_id}
+            )
+        if isinstance(event, JoinEvent):
+            return platform_events.BotInvitedToGroupEvent(
+                **common,
+                group=platform_entities.UserGroup(id=chat_id, name=chat_id),
+            )
+        if isinstance(event, LeaveEvent):
+            return platform_events.BotRemovedFromGroupEvent(
+                **common,
+                group=platform_entities.UserGroup(id=chat_id, name=chat_id),
+            )
+        if isinstance(event, MemberJoinedEvent):
+            member = _first_member(event.joined.members, chat_id)
+            return platform_events.MemberJoinedEvent(**common, group=group, member=member)
+        if isinstance(event, MemberLeftEvent):
+            member = _first_member(event.left.members, chat_id)
+            return platform_events.MemberLeftEvent(**common, group=group, member=member)
+        if isinstance(event, PostbackEvent):
+            return platform_events.PlatformSpecificEvent(
+                **common,
+                action='line.postback',
+                data={
+                    'data': event.postback.data,
+                    'chat_type': chat_type.value,
+                    'chat_id': chat_id,
+                    'sender_id': sender_id,
+                    **({'params': event.postback.params} if event.postback.params else {}),
+                },
+            )
+        if isinstance(event, BeaconEvent):
+            return platform_events.PlatformSpecificEvent(
+                **common,
+                action='line.beacon',
+                data={'hwid': event.beacon.hwid, 'type': event.beacon.type, 'chat_id': chat_id},
+            )
+        # Remaining LINE event types stay visible to event processors without a lossy mapping.
+        action = getattr(event, 'type', '') or type(event).__name__
+        return platform_events.PlatformSpecificEvent(
+            **common, action=f'line.{action}', data={**_event_payload(event), 'chat_id': chat_id}
+        )
+
+    async def target2legacy(self, event: Event) -> platform_events.FriendMessage | platform_events.GroupMessage | None:
+        """Build the legacy message event used by pipeline-based bindings."""
+        eba_event = await self.target2yiri(event)
+        if not isinstance(eba_event, platform_events.MessageReceivedEvent):
+            return None
+        return eba_event.to_legacy_event()
+
+
+def _event_payload(event: Event) -> dict:
+    to_dict = getattr(event, 'to_dict', None)
+    if callable(to_dict):
+        return dict(to_dict())
+    return {}
+
+
+def _first_member(members: typing.Any, fallback_id: str) -> platform_entities.User:
+    """Member arrays carry only user ids; the first entry is the subject of the event."""
+    for member in members or []:
+        user_id = getattr(member, 'user_id', '') or ''
+        if user_id:
+            return platform_entities.User(id=user_id, nickname=user_id)
+    return platform_entities.User(id=fallback_id, nickname=fallback_id)
@@ -0,0 +1,92 @@
+apiVersion: v1
+kind: MessagePlatformAdapter
+
+metadata:
+  name: line-omni
+  label:
+    en_US: LINE
+    zh_Hans: LINE
+    zh_Hant: LINE
+  description:
+    en_US: LINE adapter with event-driven orchestration, receiving messages through the unified webhook.
+    zh_Hans: LINE 适配器，支持事件驱动编排，通过统一 Webhook 接收消息。
+    zh_Hant: LINE 適配器，支援事件驅動編排，透過統一 Webhook 接收訊息。
+  icon: line.png
+
+spec:
+  categories:
+    - popular
+    - global
+  help_links:
+    zh: https://langbot.app/docs/zh/usage/platforms/line
+    en: https://langbot.app/docs/en/usage/platforms/line
+    ja: https://langbot.app/docs/ja/usage/platforms/line
+  config:
+    - name: webhook_url
+      label:
+        en_US: Webhook Callback URL
+        zh_Hans: Webhook 回调地址
+        zh_Hant: Webhook 回調地址
+      description:
+        en_US: Copy this URL into the LINE Developers console webhook settings and enable "Use webhook".
+        zh_Hans: 将此地址填入 LINE Developers 控制台的 Webhook 设置，并启用「Use webhook」。
+        zh_Hant: 將此位址填入 LINE Developers 主控台的 Webhook 設定，並啟用「Use webhook」。
+      type: webhook-url
+      required: false
+      default: ""
+    - name: channel_access_token
+      label:
+        en_US: Channel Access Token
+        zh_Hans: 频道访问令牌
+        zh_Hant: 頻道存取權杖
+      description:
+        en_US: Long-lived channel access token used to send messages.
+        zh_Hans: 用于发送消息的长期频道访问令牌。
+        zh_Hant: 用於傳送訊息的長期頻道存取權杖。
+      type: string
+      required: true
+      default: ""
+    - name: channel_secret
+      label:
+        en_US: Channel Secret
+        zh_Hans: 频道密钥
+        zh_Hant: 頻道密鑰
+      description:
+        en_US: Channel secret used to verify the X-Line-Signature of incoming webhooks.
+        zh_Hans: 用于校验入站 Webhook 的 X-Line-Signature 的频道密钥。
+        zh_Hant: 用於校驗入站 Webhook 的 X-Line-Signature 的頻道密鑰。
+      type: string
+      required: true
+      default: ""
+
+  supported_events:
+    - message.received
+    - message.deleted
+    - group.member_joined
+    - group.member_left
+    - bot.invited_to_group
+    - bot.removed_from_group
+    - platform.specific
+
+  supported_apis:
+    required:
+      - send_message
+      - reply_message
+    optional:
+      - get_user_info
+      - get_group_info
+      - leave_group
+      - call_platform_api
+
+  platform_specific_apis:
+    - action: get_bot_info
+      description: { en_US: "Return the LINE Official Account profile", zh_Hans: "返回 LINE 官方账号资料" }
+    - action: get_profile
+      description: { en_US: "Call the LINE get profile API for a user", zh_Hans: "调用 LINE 用户资料接口" }
+    - action: get_group_summary
+      description: { en_US: "Call the LINE group summary API", zh_Hans: "调用 LINE 群组摘要接口" }
+
+execution:
+  python:
+    path: ./adapter.py
+    attr: LineAdapter
@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import base64
+import datetime
+import typing
+
+import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
+from langbot_plugin.api.entities.builtin.platform import message as platform_message
+from linebot.v3.messaging import ImageMessage, TextMessage
+from linebot.v3.webhooks import (
+    AudioMessageContent,
+    FileMessageContent,
+    ImageMessageContent,
+    LocationMessageContent,
+    MessageEvent,
+    StickerMessageContent,
+    TextMessageContent,
+    UserMentionee,
+    VideoMessageContent,
+)
+
+from langbot.pkg.platform.adapters.line.errors import NotSupportedError
+from langbot.pkg.platform.adapters.line.types import MAX_LINE_MEDIA_BYTES, MAX_TEXT_LENGTH
+
+# LINE stores inbound images as JPEG and the Blob API does not return the media type.
+IMAGE_MIME = 'image/jpeg'
+
+
+def _validate_media(content: bytes) -> bytes:
+    """Reject inbound media above the LINE download cap."""
+    if len(content) > MAX_LINE_MEDIA_BYTES:
+        raise ValueError(f'LINE media exceeds the {MAX_LINE_MEDIA_BYTES}-byte limit')
+    return content
+
+
+def _split_text(text: str) -> list[str]:
+    """Split a text body into LINE-sized chunks so long answers still send."""
+    return [text[index : index + MAX_TEXT_LENGTH] for index in range(0, len(text), MAX_TEXT_LENGTH)]
+
+
+def _message_datetime(event: MessageEvent) -> datetime.datetime:
+    try:
+        return datetime.datetime.fromtimestamp(float(event.timestamp) / 1000.0)
+    except (AttributeError, TypeError, ValueError):
+        return datetime.datetime.now()
+
+
+def _text_components(text: str, mention: typing.Any, bot_account_id: str) -> list[platform_message.MessageComponent]:
+    """Map LINE mention positions onto At components, as the legacy adapter does."""
+    if not mention or not mention.mentionees:
+        return [platform_message.Plain(text=text)] if text else []
+    components: list[platform_message.MessageComponent] = []
+    segments = sorted(
+        (mentionee.index, mentionee.index + mentionee.length, mentionee) for mentionee in mention.mentionees
+    )
+    cursor = 0
+    for start, end, mentionee in segments:
+        if start < cursor:
+            start, end = cursor, min(end, len(text))
+        if start < cursor or end <= start or end > len(text):
+            continue
+        if start > cursor:
+            components.append(platform_message.Plain(text=text[cursor:start]))
+        token = text[start:end]
+        if isinstance(mentionee, UserMentionee):
+            target = bot_account_id if mentionee.is_self else (mentionee.user_id or token)
+        else:
+            target = token
+        components.append(platform_message.At(target=str(target), display=token.lstrip('@')))
+        cursor = end
+    if cursor < len(text):
+        components.append(platform_message.Plain(text=text[cursor:]))
+    return components
+
+
+async def _download_content(blob_api: typing.Any, message_id: str) -> bytes:
+    """Fetch inbound media through the LINE Blob API without blocking the event loop."""
+    import asyncio
+
+    if blob_api is None:
+        raise NotSupportedError('line:media_download')
+    return await asyncio.to_thread(blob_api.get_message_content, message_id)
+
+
+class LineMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
+    """Convert between LangBot message chains and LINE Messaging API messages."""
+
+    @staticmethod
+    async def yiri2target(message_chain: platform_message.MessageChain) -> list[typing.Any]:
+        """Build the LINE message array for one push/reply request."""
+        messages: list[typing.Any] = []
+        for component in message_chain:
+            if isinstance(component, platform_message.Source):
+                continue
+            if isinstance(component, platform_message.Plain):
+                messages.extend(TextMessage(text=chunk) for chunk in _split_text(component.text or ''))
+            elif isinstance(component, platform_message.At):
+                messages.append(TextMessage(text=f'@{component.display or component.target}'))
+            elif isinstance(component, platform_message.AtAll):
+                messages.append(TextMessage(text='@All'))
+            elif isinstance(component, platform_message.Image):
+                url = component.url or ''
+                if not url:
+                    # LINE fetches outgoing images from a public URL; inline base64 is not supported.
+                    raise NotSupportedError('send_message:line_image_requires_public_url')
+                messages.append(ImageMessage(original_content_url=url, preview_image_url=url))
+            elif isinstance(component, platform_message.Quote):
+                if component.origin:
+                    messages.extend(await LineMessageConverter.yiri2target(component.origin))
+            elif isinstance(component, platform_message.Forward):
+                for node in component.node_list:
+                    if node.message_chain:
+                        messages.extend(await LineMessageConverter.yiri2target(node.message_chain))
+            elif isinstance(component, platform_message.Unknown):
+                if component.text:
+                    messages.extend(TextMessage(text=chunk) for chunk in _split_text(component.text))
+            else:
+                raise NotSupportedError(f'send_message:line_unsupported_component:{type(component).__name__}')
+        return messages
+
+    @staticmethod
+    async def target2yiri(
+        event: MessageEvent,
+        blob_api: typing.Any = None,
+        bot_account_id: str = '',
+    ) -> platform_message.MessageChain:
+        """Build a LangBot message chain from a LINE message event."""
+        message = event.message
+        components: list[platform_message.MessageComponent] = [
+            platform_message.Source(
+                id=str(getattr(message, 'id', '') or event.webhook_event_id), time=_message_datetime(event)
+            )
+        ]
+        if isinstance(message, TextMessageContent):
+            components.extend(_text_components(message.text, getattr(message, 'mention', None), bot_account_id))
+        elif isinstance(message, ImageMessageContent):
+            encoded = base64.b64encode(_validate_media(await _download_content(blob_api, message.id))).decode('ascii')
+            components.append(platform_message.Image(base64=f'data:{IMAGE_MIME};base64,{encoded}'))
+        elif isinstance(message, AudioMessageContent):
+            encoded = base64.b64encode(_validate_media(await _download_content(blob_api, message.id))).decode('ascii')
+            components.append(platform_message.Voice(base64=encoded, length=int(getattr(message, 'duration', 0) or 0)))
+        elif isinstance(message, VideoMessageContent):
+            components.append(platform_message.Unknown(text='[video]'))
+        elif isinstance(message, StickerMessageContent):
+            components.append(platform_message.Unknown(text='[sticker]'))
+        elif isinstance(message, FileMessageContent):
+            components.append(platform_message.Unknown(text=f'[file] {getattr(message, "file_name", "")}'.strip()))
+        elif isinstance(message, LocationMessageContent):
+            components.append(
+                platform_message.Unknown(
+                    text=f'[location] {getattr(message, "title", "")} {getattr(message, "address", "")}'.strip()
+                )
+            )
+        else:
+            components.append(platform_message.Unknown(text=f'[unsupported line message: {type(message).__name__}]'))
+        if len(components) == 1:
+            components.append(platform_message.Unknown(text=f'[unsupported line message: {type(message).__name__}]'))
+        return platform_message.MessageChain(components)
@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import typing
+
+
+async def get_bot_info(adapter, params: dict) -> dict:
+    import asyncio
+
+    info = await asyncio.to_thread(adapter.bot.get_bot_info)
+    return {
+        'user_id': info.user_id,
+        'basic_id': info.basic_id,
+        'display_name': info.display_name,
+        'chat_mode': info.chat_mode,
+    }
+
+
+async def get_profile(adapter, params: dict) -> dict:
+    user_id = str(params.get('user_id', ''))
+    if not user_id:
+        raise ValueError('get_profile requires user_id')
+    import asyncio
+
+    profile = await asyncio.to_thread(adapter.bot.get_profile, user_id)
+    return {
+        'user_id': profile.user_id,
+        'display_name': profile.display_name,
+        'picture_url': profile.picture_url or '',
+        'status_message': profile.status_message or '',
+    }
+
+
+async def get_group_summary(adapter, params: dict) -> dict:
+    group_id = str(params.get('group_id', ''))
+    if not group_id:
+        raise ValueError('get_group_summary requires group_id')
+    import asyncio
+
+    summary = await asyncio.to_thread(adapter.bot.get_group_summary, group_id)
+    return {
+        'group_id': summary.group_id,
+        'group_name': summary.group_name,
+        'picture_url': summary.picture_url or '',
+    }
+
+
+PLATFORM_API_MAP: dict[str, typing.Callable[[typing.Any, dict], typing.Awaitable[dict]]] = {
+    'get_bot_info': get_bot_info,
+    'get_profile': get_profile,
+    'get_group_summary': get_group_summary,
+}
@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+ADAPTER_NAME = 'line-omni'
+
+# LINE Messaging API limits (push/reply share the same message array budget).
+MAX_MESSAGES_PER_REQUEST = 5
+
+# A single text message body is limited to 5000 characters.
+MAX_TEXT_LENGTH = 5000
+
+# LINE serves inbound media through the Blob API; keep the same cap as the legacy adapter.
+MAX_LINE_MEDIA_BYTES = 10 * 1024 * 1024
@@ -0,0 +1,464 @@
+"""Unit coverage for the LINE Omni adapter (unified webhook mode)."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import pathlib
+
+import pytest
+import yaml
+from linebot.v3.messaging import PushMessageRequest, ReplyMessageRequest
+from linebot.v3.webhook import WebhookParser
+
+from langbot.pkg.platform.adapters.line.adapter import LineAdapter
+from langbot.pkg.platform.adapters.line.errors import NotSupportedError
+from langbot.pkg.platform.adapters.line.event_converter import LineEventConverter
+from langbot.pkg.platform.adapters.line.message_converter import LineMessageConverter
+from langbot_plugin.api.definition.abstract.platform.event_logger import AbstractEventLogger
+from langbot_plugin.api.entities.builtin.platform import entities as platform_entities
+from langbot_plugin.api.entities.builtin.platform import events as platform_events
+from langbot_plugin.api.entities.builtin.platform import message as platform_message
+
+SECRET = 'test-channel-secret'
+TOKEN = 'test-channel-access-token'
+PLATFORM_ROOT = pathlib.Path(__file__).parents[3] / 'src/langbot/pkg/platform'
+
+
+class DummyLogger(AbstractEventLogger):
+    def __init__(self):
+        self.lines: list[tuple[str, str]] = []
+
+    async def info(self, text, images=None, message_session_id=None, no_throw=True):
+        self.lines.append(('info', str(text)))
+
+    async def debug(self, text, images=None, message_session_id=None, no_throw=True):
+        pass
+
+    async def warning(self, text, images=None, message_session_id=None, no_throw=True):
+        self.lines.append(('warning', str(text)))
+
+    async def error(self, text, images=None, message_session_id=None, no_throw=True):
+        self.lines.append(('error', str(text)))
+
+
+class DummyMessagingApi:
+    def __init__(self, fail_reply: bool = False):
+        self.push_requests: list[PushMessageRequest] = []
+        self.reply_requests: list[ReplyMessageRequest] = []
+        self.fail_reply = fail_reply
+        self.leaved_groups: list[str] = []
+
+    def push_message(self, request):
+        self.push_requests.append(request)
+        return {'sent': len(request.messages)}
+
+    def reply_message(self, request):
+        if self.fail_reply:
+            raise RuntimeError('reply token expired')
+        self.reply_requests.append(request)
+        return {'sent': len(request.messages)}
+
+    def get_bot_info(self):
+        return type(
+            'BotInfo', (), {'user_id': 'Ubot', 'basic_id': '@bot', 'display_name': 'Bot', 'chat_mode': 'chat'}
+        )()
+
+    def get_profile(self, user_id):
+        return type(
+            'Profile',
+            (),
+            {'user_id': user_id, 'display_name': 'Alice', 'picture_url': 'https://img/1.png', 'status_message': None},
+        )()
+
+    def get_group_summary(self, group_id):
+        return type('Summary', (), {'group_id': group_id, 'group_name': 'Team', 'picture_url': None})()
+
+    def leave_group(self, group_id):
+        self.leaved_groups.append(group_id)
+
+
+class DummyBlobApi:
+    def __init__(self, content: bytes = b'image-bytes'):
+        self.content = content
+        self.requested: list[str] = []
+
+    def get_message_content(self, message_id):
+        self.requested.append(message_id)
+        return self.content
+
+
+class DummyApiClient:
+    def __init__(self):
+        self.closed = False
+
+    def close(self):
+        self.closed = True
+
+
+class DummyHeaders(dict):
+    def get(self, key, default=None):
+        return dict.get(self, key, default)
+
+
+class DummyRequest:
+    def __init__(self, body: str, signature: str | None):
+        self.headers = DummyHeaders({} if signature is None else {'X-Line-Signature': signature})
+        self._body = body
+
+    async def get_data(self, as_text: bool = False):
+        return self._body if as_text else self._body.encode()
+
+
+def sign(body: str, secret: str = SECRET) -> str:
+    return base64.b64encode(hmac.new(secret.encode(), body.encode(), hashlib.sha256).digest()).decode()
+
+
+def message_event(
+    *,
+    text: str = 'hello',
+    source: dict | None = None,
+    reply_token: str = 'reply-token-1',
+    message: dict | None = None,
+) -> dict:
+    return {
+        'type': 'message',
+        'mode': 'active',
+        'timestamp': 1730000000000,
+        'webhookEventId': '01HWEBHOOKID',
+        'deliveryContext': {'isRedelivery': False},
+        'source': source or {'type': 'user', 'userId': 'U-user-1'},
+        'replyToken': reply_token,
+        'message': message or {'id': 'msg-1', 'type': 'text', 'text': text, 'quoteToken': 'quote-1'},
+    }
+
+
+def webhook_body(*events: dict) -> str:
+    return json.dumps({'destination': 'Ubot', 'events': list(events)})
+
+
+def build_adapter(logger: DummyLogger | None = None) -> LineAdapter:
+    return LineAdapter(
+        config={'channel_access_token': TOKEN, 'channel_secret': SECRET},
+        logger=logger or DummyLogger(),
+    )
+
+
+class EventSink:
+    def __init__(self):
+        self.events: list[platform_events.Event] = []
+
+    async def __call__(self, event, adapter):
+        self.events.append(event)
+
+
+async def dispatch(adapter: LineAdapter, body: str, signature: str | None = None):
+    response = await adapter.handle_unified_webhook(
+        bot_uuid='bot-uuid', path='', request=DummyRequest(body, sign(body) if signature is None else signature)
+    )
+    pending = list(adapter.background_tasks)
+    if pending:
+        await asyncio_gather(pending)
+    return response
+
+
+async def asyncio_gather(tasks):
+    import asyncio
+
+    await asyncio.gather(*tasks, return_exceptions=True)
+
+
+def parse_events(body: str) -> list:
+    return WebhookParser(SECRET).parse(body, sign(body))
+
+
+def test_manifest_is_discoverable_and_keeps_legacy_config():
+    manifest = yaml.safe_load((PLATFORM_ROOT / 'adapters/line/manifest.yaml').read_text(encoding='utf-8'))
+    legacy = yaml.safe_load((PLATFORM_ROOT / 'sources/line.yaml').read_text(encoding='utf-8'))
+    assert manifest['metadata']['name'] == 'line-omni'
+    assert manifest['execution']['python'] == {'path': './adapter.py', 'attr': 'LineAdapter'}
+    assert {field['name'] for field in legacy['spec']['config']} <= {
+        field['name'] for field in manifest['spec']['config']
+    }
+    assert manifest['spec']['help_links'] == legacy['spec']['help_links']
+    assert 'message.received' in manifest['spec']['supported_events']
+    assert (PLATFORM_ROOT / 'adapters/line' / manifest['metadata']['icon']).is_file()
+
+
+def test_adapter_requires_credentials():
+    with pytest.raises(ValueError, match='channel_access_token'):
+        LineAdapter(config={'channel_secret': SECRET}, logger=DummyLogger())
+
+
+@pytest.mark.asyncio
+async def test_webhook_rejects_missing_and_invalid_signature():
+    adapter = build_adapter()
+    body = webhook_body(message_event())
+    assert (await adapter.handle_unified_webhook('bot', '', DummyRequest(body, None))).status_code == 400
+    assert (await adapter.handle_unified_webhook('bot', '', DummyRequest(body, 'bogus'))).status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_webhook_dispatches_message_received_event():
+    adapter = build_adapter()
+    sink = EventSink()
+    adapter.register_listener(platform_events.MessageReceivedEvent, sink)
+
+    response = await dispatch(adapter, webhook_body(message_event(text='hi there')))
+
+    assert response.status_code == 200
+    assert len(sink.events) == 1
+    event = sink.events[0]
+    assert isinstance(event, platform_events.MessageReceivedEvent)
+    assert event.adapter_name == 'line-omni'
+    assert event.chat_type == platform_entities.ChatType.PRIVATE
+    assert event.chat_id == 'U-user-1'
+    assert event.message_id == 'msg-1'
+    assert [type(component) for component in event.message_chain] == [platform_message.Source, platform_message.Plain]
+    assert event.message_chain[1].text == 'hi there'
+
+
+@pytest.mark.asyncio
+async def test_webhook_serves_legacy_listeners_and_ignores_self_messages():
+    adapter = build_adapter()
+    adapter.bot_account_id = 'Ubot'
+    legacy_sink = EventSink()
+    adapter.register_listener(platform_events.FriendMessage, legacy_sink)
+
+    await dispatch(adapter, webhook_body(message_event(source={'type': 'user', 'userId': 'Ubot'})))
+    assert legacy_sink.events == []
+
+    await dispatch(adapter, webhook_body(message_event(source={'type': 'user', 'userId': 'U-user-9'})))
+    assert len(legacy_sink.events) == 1
+    assert isinstance(legacy_sink.events[0], platform_events.FriendMessage)
+    assert legacy_sink.events[0].sender.id == 'U-user-9'
+
+
+@pytest.mark.asyncio
+async def test_group_message_uses_group_chat_identity():
+    events = parse_events(
+        webhook_body(message_event(source={'type': 'group', 'groupId': 'C-group', 'userId': 'U-user-2'}))
+    )
+    event = await LineEventConverter().target2yiri(events[0])
+
+    assert event.chat_type == platform_entities.ChatType.GROUP
+    assert event.chat_id == 'C-group'
+    assert event.group.id == 'C-group'
+    assert event.sender.id == 'U-user-2'
+
+
+@pytest.mark.asyncio
+async def test_unsend_becomes_message_deleted():
+    body = webhook_body(
+        {
+            'type': 'unsend',
+            'mode': 'active',
+            'timestamp': 1730000000000,
+            'webhookEventId': '01HUNSEND',
+            'deliveryContext': {'isRedelivery': False},
+            'source': {'type': 'user', 'userId': 'U-user-1'},
+            'unsend': {'messageId': 'msg-9'},
+        }
+    )
+    event = await LineEventConverter().target2yiri(parse_events(body)[0])
+
+    assert isinstance(event, platform_events.MessageDeletedEvent)
+    assert event.message_id == 'msg-9'
+    assert event.chat_id == 'U-user-1'
+
+
+@pytest.mark.asyncio
+async def test_follow_and_postback_stay_platform_specific():
+    follow = webhook_body(
+        {
+            'type': 'follow',
+            'mode': 'active',
+            'timestamp': 1730000000000,
+            'webhookEventId': '01HFOLLOW',
+            'deliveryContext': {'isRedelivery': False},
+            'source': {'type': 'user', 'userId': 'U-user-3'},
+            'replyToken': 'token',
+            'follow': {'isUnblocked': True},
+        }
+    )
+    follow_event = await LineEventConverter().target2yiri(parse_events(follow)[0])
+    assert follow_event.type == 'platform.specific'
+    assert follow_event.action == 'line.follow'
+
+    postback = webhook_body(
+        {
+            'type': 'postback',
+            'mode': 'active',
+            'timestamp': 1730000000000,
+            'webhookEventId': '01HPOSTBACK',
+            'deliveryContext': {'isRedelivery': False},
+            'source': {'type': 'group', 'groupId': 'C-group', 'userId': 'U-user-4'},
+            'replyToken': 'token',
+            'postback': {'data': 'action=buy'},
+        }
+    )
+    postback_event = await LineEventConverter().target2yiri(parse_events(postback)[0])
+    assert postback_event.action == 'line.postback'
+    assert postback_event.data['data'] == 'action=buy'
+    assert postback_event.data['chat_id'] == 'C-group'
+
+
+@pytest.mark.asyncio
+async def test_member_joined_maps_group_and_member():
+    body = webhook_body(
+        {
+            'type': 'memberJoined',
+            'mode': 'active',
+            'timestamp': 1730000000000,
+            'webhookEventId': '01HJOINED',
+            'deliveryContext': {'isRedelivery': False},
+            'source': {'type': 'group', 'groupId': 'C-group'},
+            'replyToken': 'token',
+            'joined': {'members': [{'type': 'user', 'userId': 'U-user-5'}]},
+        }
+    )
+    event = await LineEventConverter().target2yiri(parse_events(body)[0])
+
+    assert isinstance(event, platform_events.MemberJoinedEvent)
+    assert event.group.id == 'C-group'
+    assert event.member.id == 'U-user-5'
+
+
+@pytest.mark.asyncio
+async def test_text_mention_becomes_at_component_for_the_bot():
+    body = webhook_body(
+        message_event(
+            text='@bot hello',
+            message={
+                'id': 'msg-2',
+                'type': 'text',
+                'quoteToken': 'quote-2',
+                'text': '@bot hello',
+                'mention': {
+                    'mentionees': [
+                        {'index': 0, 'length': 4, 'userId': 'Ubot', 'isSelf': True, 'type': 'user'},
+                    ]
+                },
+            },
+        )
+    )
+    event = await LineEventConverter(bot_account_id='Ubot').target2yiri(parse_events(body)[0])
+
+    at_components = [component for component in event.message_chain if isinstance(component, platform_message.At)]
+    assert len(at_components) == 1
+    assert at_components[0].target == 'Ubot'
+    assert at_components[0].display == 'bot'
+
+
+@pytest.mark.asyncio
+async def test_image_message_is_downloaded_through_the_blob_api():
+    blob = DummyBlobApi(b'jpeg-bytes')
+    body = webhook_body(
+        message_event(
+            message={
+                'id': 'msg-3',
+                'type': 'image',
+                'quoteToken': 'quote-3',
+                'contentProvider': {'type': 'line'},
+            }
+        )
+    )
+    event = await LineEventConverter(blob_api=blob).target2yiri(parse_events(body)[0])
+
+    image = event.message_chain[1]
+    assert isinstance(image, platform_message.Image)
+    assert image.base64 == f'data:image/jpeg;base64,{base64.b64encode(b"jpeg-bytes").decode()}'
+    assert blob.requested == ['msg-3']
+
+
+@pytest.mark.asyncio
+async def test_outgoing_chain_builds_line_messages():
+    chain = platform_message.MessageChain(
+        [
+            platform_message.Source(id='msg-1', time=__import__('datetime').datetime.now()),
+            platform_message.Plain(text='hello'),
+            platform_message.At(target='U-user-1', display='Alice'),
+        ]
+    )
+    messages = await LineMessageConverter.yiri2target(chain)
+
+    assert [message.type for message in messages] == ['text', 'text']
+    assert messages[0].text == 'hello'
+    assert messages[1].text == '@Alice'
+
+
+@pytest.mark.asyncio
+async def test_outgoing_image_requires_a_public_url():
+    chain = platform_message.MessageChain([platform_message.Image(base64='ZGly')])
+    with pytest.raises(NotSupportedError):
+        await LineMessageConverter.yiri2target(chain)
+
+
+@pytest.mark.asyncio
+async def test_send_message_batches_five_messages_per_request():
+    adapter = build_adapter()
+    adapter.bot = DummyMessagingApi()
+    chain = platform_message.MessageChain([platform_message.Plain(text='x'), platform_message.Plain(text='y')] * 3)
+
+    result = await adapter.send_message('person', 'U-user-1', chain)
+
+    assert [len(request.messages) for request in adapter.bot.push_requests] == [5, 1]
+    assert all(request.to == 'U-user-1' for request in adapter.bot.push_requests)
+    assert result.raw == {'sent': 6, 'target_id': 'U-user-1'}
+
+
+@pytest.mark.asyncio
+async def test_reply_uses_reply_token_then_falls_back_to_push():
+    adapter = build_adapter()
+    event = await LineEventConverter().target2yiri(parse_events(webhook_body(message_event()))[0])
+    chain = platform_message.MessageChain([platform_message.Plain(text='pong')])
+
+    adapter.bot = DummyMessagingApi()
+    await adapter.reply_message(event, chain)
+    assert adapter.bot.reply_requests[0].reply_token == 'reply-token-1'
+    assert adapter.bot.push_requests == []
+
+    adapter.bot = DummyMessagingApi(fail_reply=True)
+    await adapter.reply_message(event, chain)
+    assert adapter.bot.push_requests[0].to == 'U-user-1'
+
+
+@pytest.mark.asyncio
+async def test_profile_and_group_apis_use_the_messaging_api():
+    adapter = build_adapter()
+    adapter.bot = DummyMessagingApi()
+
+    user = await adapter.get_user_info('U-user-1')
+    group = await adapter.get_group_info('C-group')
+    await adapter.leave_group('C-group')
+
+    assert user.id == 'U-user-1'
+    assert user.nickname == 'Alice'
+    assert group.name == 'Team'
+    assert adapter.bot.leaved_groups == ['C-group']
+    with pytest.raises(NotSupportedError):
+        await adapter.get_group_member_list('C-group')
+
+
+@pytest.mark.asyncio
+async def test_call_platform_api_dispatches_and_rejects_unknown_actions():
+    adapter = build_adapter()
+    adapter.bot = DummyMessagingApi()
+
+    info = await adapter.call_platform_api('get_bot_info', {})
+    assert info['user_id'] == 'Ubot'
+    with pytest.raises(NotSupportedError):
+        await adapter.call_platform_api('not_a_line_action', {})
+
+
+@pytest.mark.asyncio
+async def test_kill_closes_the_api_client():
+    adapter = build_adapter()
+    api_client = DummyApiClient()
+    adapter.api_client = api_client
+
+    assert await adapter.kill() is True
+    assert api_client.closed is True
+    assert adapter.background_tasks == set()