diff --git a/plugins/platforms/line/__init__.py b/plugins/platforms/line/__init__.py
new file mode 100644
index 000000000000..d4f1d7bf0e3f
--- /dev/null
+++ b/plugins/platforms/line/__init__.py
@@ -0,0 +1,3 @@
+from .adapter import register
+
+__all__ = ["register"]
diff --git a/plugins/platforms/line/adapter.py b/plugins/platforms/line/adapter.py
new file mode 100644
index 000000000000..660d9f185dcd
--- /dev/null
+++ b/plugins/platforms/line/adapter.py
@@ -0,0 +1,560 @@
+"""
+LINE Messaging API platform adapter for Hermes Agent.
+
+Logic ported from the OpenClaw LINE channel extension
+(``~/openclaw/extensions/line``):
+
+* aiohttp webhook server at ``/line/webhook``
+* HMAC-SHA256 + base64 signature verification of the raw request body
+  (``X-Line-Signature`` header), using a constant-time comparison
+* outbound sends via the LINE Messaging API; reply token is preferred
+  (one-shot, ~1 minute validity) with automatic fallback to push
+* group / room / 1:1 chat type detection from the source ID prefix
+* image / sticker handling (image attachments are fetched from the
+  ``api-data.line.me`` content endpoint and cached locally)
+
+Configuration in ``config.yaml``::
+
+    gateway:
+      platforms:
+        line:
+          enabled: true
+          extra:
+            channel_access_token: "..."   # or LINE_CHANNEL_ACCESS_TOKEN
+            channel_secret: "..."         # or LINE_CHANNEL_SECRET
+            port: 3979                    # or LINE_PORT
+            webhook_path: "/line/webhook" # default
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import hmac
+import json
+import logging
+import os
+import time
+from typing import Any, Dict, List, Optional, Tuple
+
+try:
+    from aiohttp import ClientSession, web
+
+    AIOHTTP_AVAILABLE = True
+except ImportError:
+    AIOHTTP_AVAILABLE = False
+    web = None  # type: ignore[assignment]
+    ClientSession = None  # type: ignore[assignment,misc]
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MessageEvent,
+    MessageType,
+    SendResult,
+    cache_image_from_bytes,
+)
+from gateway.platforms.helpers import MessageDeduplicator
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_PORT = 3979
+_DEFAULT_WEBHOOK_PATH = "/line/webhook"
+_MAX_RAW_BODY_BYTES = 64 * 1024
+_REPLY_TOKEN_TTL_SECONDS = 50  # LINE reply tokens are valid ~1 minute
+_API_BASE = "https://api.line.me/v2/bot"
+_DATA_API_BASE = "https://api-data.line.me/v2/bot"
+
+
+# ── Signature verification ────────────────────────────────────────────────────
+
+
+def _validate_line_signature(body: bytes, signature: str, channel_secret: str) -> bool:
+    """Constant-time HMAC-SHA256 (base64) signature check."""
+    if not signature or not channel_secret:
+        return False
+    digest = hmac.new(
+        channel_secret.encode("utf-8"), body, hashlib.sha256
+    ).digest()
+    expected = base64.b64encode(digest).decode("ascii")
+    return hmac.compare_digest(expected, signature)
+
+
+# ── Plugin-level helpers ──────────────────────────────────────────────────────
+
+
+def check_requirements() -> bool:
+    return AIOHTTP_AVAILABLE
+
+
+def validate_config(config) -> bool:
+    extra = getattr(config, "extra", {}) or {}
+    token = os.getenv("LINE_CHANNEL_ACCESS_TOKEN") or extra.get(
+        "channel_access_token", ""
+    )
+    secret = os.getenv("LINE_CHANNEL_SECRET") or extra.get("channel_secret", "")
+    return bool(token and secret)
+
+
+def is_connected(config) -> bool:
+    return validate_config(config)
+
+
+# ── Adapter ───────────────────────────────────────────────────────────────────
+
+
+class LineAdapter(BasePlatformAdapter):
+    """LINE Messaging API adapter (webhook + push)."""
+
+    # Per LINE docs: text messages capped at 5000 characters.
+    MAX_MESSAGE_LENGTH = 4500
+
+    def __init__(self, config: PlatformConfig):
+        super().__init__(config, Platform("line"))
+        extra = config.extra or {}
+        self._token = extra.get("channel_access_token") or os.getenv(
+            "LINE_CHANNEL_ACCESS_TOKEN", ""
+        )
+        self._secret = extra.get("channel_secret") or os.getenv(
+            "LINE_CHANNEL_SECRET", ""
+        )
+        self._port = int(extra.get("port") or os.getenv("LINE_PORT", str(_DEFAULT_PORT)))
+        self._webhook_path = extra.get("webhook_path") or _DEFAULT_WEBHOOK_PATH
+        self._runner: Optional["web.AppRunner"] = None
+        self._http: Optional["ClientSession"] = None
+        self._dedup = MessageDeduplicator(max_size=1000)
+        # chat_id → (reply_token, captured_at_unix)
+        self._reply_tokens: Dict[str, Tuple[str, float]] = {}
+        self._bot_user_id: Optional[str] = None
+
+    # -- lifecycle ----------------------------------------------------------
+
+    async def connect(self) -> bool:
+        if not AIOHTTP_AVAILABLE:
+            self._set_fatal_error(
+                "MISSING_SDK",
+                "aiohttp not installed. Run: pip install aiohttp",
+                retryable=False,
+            )
+            return False
+        if not self._token or not self._secret:
+            self._set_fatal_error(
+                "MISSING_CREDENTIALS",
+                "LINE_CHANNEL_ACCESS_TOKEN and LINE_CHANNEL_SECRET are both required",
+                retryable=False,
+            )
+            return False
+
+        try:
+            self._http = ClientSession()
+
+            # Best-effort fetch of bot's own userId for self-message filtering.
+            try:
+                async with self._http.get(
+                    f"{_API_BASE}/info",
+                    headers={"Authorization": f"Bearer {self._token}"},
+                ) as resp:
+                    if resp.status == 200:
+                        info = await resp.json()
+                        self._bot_user_id = info.get("userId")
+            except Exception as e:
+                logger.debug("[line] bot info fetch failed: %s", e)
+
+            app = web.Application()
+            app.router.add_post(self._webhook_path, self._handle_webhook)
+            app.router.add_get("/health", lambda _: web.Response(text="ok"))
+
+            self._runner = web.AppRunner(app)
+            await self._runner.setup()
+            site = web.TCPSite(self._runner, "0.0.0.0", self._port)
+            await site.start()
+
+            self._running = True
+            self._mark_connected()
+            logger.info(
+                "[line] Webhook server listening on 0.0.0.0:%d%s",
+                self._port,
+                self._webhook_path,
+            )
+            return True
+        except Exception as e:
+            self._set_fatal_error(
+                "CONNECT_FAILED",
+                f"LINE connection failed: {e}",
+                retryable=True,
+            )
+            logger.error("[line] Failed to connect: %s", e)
+            return False
+
+    async def disconnect(self) -> None:
+        self._running = False
+        if self._runner:
+            await self._runner.cleanup()
+            self._runner = None
+        if self._http:
+            await self._http.close()
+            self._http = None
+        self._mark_disconnected()
+        logger.info("[line] Disconnected")
+
+    # -- webhook ------------------------------------------------------------
+
+    async def _handle_webhook(self, request: "web.Request") -> "web.Response":
+        signature = request.headers.get("X-Line-Signature", "")
+        if not signature:
+            return web.json_response({"error": "Missing X-Line-Signature"}, status=400)
+
+        raw = await request.read()
+        if len(raw) > _MAX_RAW_BODY_BYTES:
+            return web.json_response({"error": "Payload too large"}, status=413)
+
+        if not _validate_line_signature(raw, signature, self._secret):
+            logger.warning("[line] webhook signature validation failed")
+            return web.json_response({"error": "Invalid signature"}, status=401)
+
+        try:
+            payload = json.loads(raw.decode("utf-8") or "{}")
+        except json.JSONDecodeError:
+            return web.json_response({"error": "Invalid JSON"}, status=400)
+
+        events = payload.get("events") or []
+        for event in events:
+            try:
+                await self._dispatch_event(event)
+            except Exception as e:
+                logger.error("[line] error handling event: %s", e, exc_info=True)
+
+        return web.json_response({"status": "ok"})
+
+    async def _dispatch_event(self, event: Dict[str, Any]) -> None:
+        if event.get("type") != "message":
+            return
+
+        message = event.get("message") or {}
+        msg_id = message.get("id") or event.get("webhookEventId")
+        if msg_id and self._dedup.is_duplicate(str(msg_id)):
+            return
+
+        source = event.get("source") or {}
+        chat_id, chat_type = _resolve_chat(source)
+        if not chat_id:
+            return
+
+        user_id = source.get("userId") or ""
+        if user_id and self._bot_user_id and user_id == self._bot_user_id:
+            return  # self-message echo
+
+        # Cache the reply token so the next outbound send can use it.
+        reply_token = event.get("replyToken")
+        if reply_token:
+            self._reply_tokens[chat_id] = (reply_token, time.time())
+
+        text = ""
+        media_urls: List[str] = []
+        media_types: List[str] = []
+        msg_type = MessageType.TEXT
+        kind = message.get("type")
+
+        if kind == "text":
+            text = message.get("text", "") or ""
+        elif kind == "image" and msg_id:
+            data = await self._fetch_message_content(str(msg_id))
+            if data:
+                cached = await cache_image_from_bytes(data, "image/jpeg")
+                if cached:
+                    media_urls.append(cached)
+                    media_types.append("image/jpeg")
+                    msg_type = MessageType.PHOTO
+        elif kind == "sticker":
+            text = "[sticker]"
+        else:
+            # Other message kinds (video, audio, location, file...) are
+            # forwarded as a marker so the agent can acknowledge.
+            text = f"[{kind or 'unknown'} message]"
+
+        event_obj = MessageEvent(
+            text=text,
+            source=self.build_source(
+                chat_id=chat_id,
+                chat_type=chat_type,
+                user_id=user_id or None,
+                user_name=None,
+                message_id=str(msg_id) if msg_id else None,
+            ),
+            message_type=msg_type,
+            media_urls=media_urls,
+            media_types=media_types,
+            message_id=str(msg_id) if msg_id else None,
+        )
+        await self.handle_message(event_obj)
+
+    async def _fetch_message_content(self, message_id: str) -> Optional[bytes]:
+        if not self._http:
+            return None
+        url = f"{_DATA_API_BASE}/message/{message_id}/content"
+        try:
+            async with self._http.get(
+                url, headers={"Authorization": f"Bearer {self._token}"}
+            ) as resp:
+                if resp.status != 200:
+                    logger.debug(
+                        "[line] content fetch %s returned %s", message_id, resp.status
+                    )
+                    return None
+                return await resp.read()
+        except Exception as e:
+            logger.warning("[line] content fetch failed: %s", e)
+            return None
+
+    # -- send ---------------------------------------------------------------
+
+    def _consume_reply_token(self, chat_id: str) -> Optional[str]:
+        entry = self._reply_tokens.pop(chat_id, None)
+        if not entry:
+            return None
+        token, captured = entry
+        if time.time() - captured > _REPLY_TOKEN_TTL_SECONDS:
+            return None
+        return token
+
+    async def _post(self, path: str, payload: Dict[str, Any]) -> Tuple[bool, str]:
+        if not self._http:
+            return False, "LINE adapter not initialized"
+        url = f"{_API_BASE}{path}"
+        try:
+            async with self._http.post(
+                url,
+                json=payload,
+                headers={
+                    "Authorization": f"Bearer {self._token}",
+                    "Content-Type": "application/json",
+                },
+            ) as resp:
+                if 200 <= resp.status < 300:
+                    return True, ""
+                body = await resp.text()
+                return False, f"HTTP {resp.status}: {body[:300]}"
+        except Exception as e:
+            return False, str(e)
+
+    async def _send_messages(
+        self, chat_id: str, messages: List[Dict[str, Any]]
+    ) -> SendResult:
+        # LINE accepts at most 5 messages per push/reply call.
+        for i in range(0, len(messages), 5):
+            batch = messages[i : i + 5]
+            reply_token = self._consume_reply_token(chat_id) if i == 0 else None
+            if reply_token:
+                ok, err = await self._post(
+                    "/message/reply",
+                    {"replyToken": reply_token, "messages": batch},
+                )
+                if not ok:
+                    # Reply token may have already been consumed by another
+                    # request, or expired between capture and use — fall back
+                    # to push.
+                    logger.debug("[line] reply failed (%s); falling back to push", err)
+                    ok, err = await self._post(
+                        "/message/push", {"to": chat_id, "messages": batch}
+                    )
+            else:
+                ok, err = await self._post(
+                    "/message/push", {"to": chat_id, "messages": batch}
+                )
+            if not ok:
+                return SendResult(success=False, error=err, retryable=True)
+        return SendResult(success=True)
+
+    async def send(
+        self,
+        chat_id: str,
+        content: str,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        formatted = self.format_message(content)
+        chunks = self.truncate_message(formatted, max_length=self.MAX_MESSAGE_LENGTH)
+        messages = [{"type": "text", "text": chunk} for chunk in chunks if chunk]
+        if not messages:
+            return SendResult(success=True)
+        return await self._send_messages(chat_id, messages)
+
+    async def send_typing(
+        self, chat_id: str, metadata: Optional[Dict[str, Any]] = None
+    ) -> None:
+        # LINE's "loading animation" API is the closest analog to a typing
+        # indicator.  It is only valid for 1:1 user chats — group/room IDs
+        # are rejected with HTTP 400.  ``loadingSeconds`` accepts 5..60 in
+        # multiples of 5; the animation auto-clears as soon as the bot
+        # sends a real message (or when the timer runs out).
+        if not self._http or not chat_id:
+            return None
+        if not chat_id.startswith("U"):
+            return None  # group/room — API does not support these
+        try:
+            ok, err = await self._post(
+                "/chat/loading/start",
+                {"chatId": chat_id, "loadingSeconds": 20},
+            )
+            if not ok:
+                logger.debug("[line] loading animation start failed: %s", err)
+        except Exception as e:
+            logger.debug("[line] loading animation start raised: %s", e)
+        return None
+
+    async def send_image(
+        self,
+        chat_id: str,
+        image_url: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        if not (image_url.startswith("http://") or image_url.startswith("https://")):
+            return SendResult(
+                success=False,
+                error="LINE image messages require a public HTTPS URL",
+                retryable=False,
+            )
+        messages: List[Dict[str, Any]] = [
+            {
+                "type": "image",
+                "originalContentUrl": image_url,
+                "previewImageUrl": image_url,
+            }
+        ]
+        if caption:
+            messages.append({"type": "text", "text": caption})
+        return await self._send_messages(chat_id, messages)
+
+    async def send_image_file(
+        self,
+        chat_id: str,
+        image_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        **kwargs,
+    ) -> SendResult:
+        # LINE's image message requires a public URL — local files cannot be
+        # uploaded directly.  Callers should host the file first.
+        return SendResult(
+            success=False,
+            error="LINE does not accept inline image uploads — host the file and call send_image() with a public URL",
+            retryable=False,
+        )
+
+    async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
+        chat_type = "dm"
+        if chat_id.startswith("C"):
+            chat_type = "group"
+        elif chat_id.startswith("R"):
+            chat_type = "group"  # multi-person room — closest analog
+        return {"name": chat_id, "type": chat_type, "chat_id": chat_id}
+
+
+def _resolve_chat(source: Dict[str, Any]) -> Tuple[str, str]:
+    """Return (chat_id, chat_type) for a LINE event source object."""
+    src_type = source.get("type")
+    if src_type == "group":
+        return source.get("groupId", ""), "group"
+    if src_type == "room":
+        return source.get("roomId", ""), "group"
+    if src_type == "user":
+        return source.get("userId", ""), "dm"
+    return "", "dm"
+
+
+# ── Interactive setup ─────────────────────────────────────────────────────────
+
+
+def interactive_setup() -> None:
+    from hermes_cli.config import get_env_value, save_env_value
+    from hermes_cli.cli_output import (
+        print_info,
+        print_success,
+        print_warning,
+        prompt,
+        prompt_yes_no,
+    )
+
+    if get_env_value("LINE_CHANNEL_ACCESS_TOKEN"):
+        print_info("LINE: already configured")
+        if not prompt_yes_no("Reconfigure LINE?", False):
+            return
+
+    print_info("Create a Messaging API channel at https://developers.line.biz/console/")
+    print_info("From the channel page you need:")
+    print_info("  • Channel access token (long-lived) — under the Messaging API tab")
+    print_info("  • Channel secret — under the Basic settings tab")
+    print()
+    print_info("Then expose your webhook port publicly (devtunnel / ngrok / cloudflared)")
+    print_info("and set the channel's webhook URL to:  https://<tunnel>/line/webhook")
+    print()
+
+    token = prompt(
+        "Channel access token",
+        default=get_env_value("LINE_CHANNEL_ACCESS_TOKEN") or "",
+        password=True,
+    )
+    if not token:
+        print_warning("Channel access token is required — skipping LINE setup")
+        return
+    save_env_value("LINE_CHANNEL_ACCESS_TOKEN", token.strip())
+
+    secret = prompt(
+        "Channel secret",
+        default=get_env_value("LINE_CHANNEL_SECRET") or "",
+        password=True,
+    )
+    if not secret:
+        print_warning("Channel secret is required — skipping LINE setup")
+        return
+    save_env_value("LINE_CHANNEL_SECRET", secret.strip())
+
+    if prompt_yes_no("Restrict access to specific LINE user IDs? (recommended)", True):
+        allowed = prompt(
+            "Allowed LINE user IDs (comma-separated, format Uxxxx...)",
+            default=get_env_value("LINE_ALLOWED_USERS") or "",
+        )
+        if allowed:
+            save_env_value("LINE_ALLOWED_USERS", allowed.replace(" ", ""))
+            print_success("Allowlist configured")
+        else:
+            save_env_value("LINE_ALLOWED_USERS", "")
+    else:
+        save_env_value("LINE_ALLOW_ALL_USERS", "true")
+        print_warning("⚠️  Open access — anyone who messages the bot can command it.")
+
+    print()
+    print_success("LINE configuration saved to ~/.hermes/.env")
+    print_info("Default webhook port: %d  (override with LINE_PORT)" % _DEFAULT_PORT)
+    print_info("Restart the gateway:  hermes gateway restart")
+
+
+# ── Plugin entry point ────────────────────────────────────────────────────────
+
+
+def register(ctx) -> None:
+    ctx.register_platform(
+        name="line",
+        label="LINE",
+        adapter_factory=lambda cfg: LineAdapter(cfg),
+        check_fn=check_requirements,
+        validate_config=validate_config,
+        is_connected=is_connected,
+        required_env=["LINE_CHANNEL_ACCESS_TOKEN", "LINE_CHANNEL_SECRET"],
+        install_hint="pip install aiohttp",
+        setup_fn=interactive_setup,
+        allowed_users_env="LINE_ALLOWED_USERS",
+        allow_all_env="LINE_ALLOW_ALL_USERS",
+        max_message_length=4500,
+        emoji="💬",
+        allow_update_command=True,
+        platform_hint=(
+            "You are chatting via LINE Messaging API. LINE renders plain "
+            "text only — markdown, HTML, and code fences are not rendered, "
+            "so prefer concise prose. Long replies are split into multiple "
+            "bubbles. Image sends require a publicly-reachable HTTPS URL."
+        ),
+    )
diff --git a/plugins/platforms/line/plugin.yaml b/plugins/platforms/line/plugin.yaml
new file mode 100644
index 000000000000..5059dfc3cf67
--- /dev/null
+++ b/plugins/platforms/line/plugin.yaml
@@ -0,0 +1,14 @@
+name: line-platform
+kind: platform
+version: 1.0.0
+description: >
+  LINE Messaging API gateway adapter for Hermes Agent.
+  Runs an aiohttp webhook server that receives LINE webhook events
+  (with HMAC-SHA256 signature verification) and relays messages to and
+  from the Hermes agent.  Supports text, image, sticker, group, room,
+  and 1:1 (user) chats.  Outbound replies use the LINE reply token when
+  available and fall back to the push API otherwise.
+author: Hermes Agent contributors
+requires_env:
+  - LINE_CHANNEL_ACCESS_TOKEN
+  - LINE_CHANNEL_SECRET
diff --git a/scripts/line_adapter_test.py b/scripts/line_adapter_test.py
new file mode 100644
index 000000000000..6ce9e4217441
--- /dev/null
+++ b/scripts/line_adapter_test.py
@@ -0,0 +1,94 @@
+"""
+Standalone test harness for the LINE platform plugin.
+
+Boots only the LineAdapter — no full Hermes gateway required — and prints
+every inbound event to stdout so you can verify webhook signature, body
+parsing, and chat-id resolution against the live LINE Messaging API.
+
+Usage:
+    export LINE_CHANNEL_ACCESS_TOKEN=...
+    export LINE_CHANNEL_SECRET=...
+    # optional: export LINE_PORT=3979
+    python scripts/line_adapter_test.py
+
+Then expose the port with a tunnel (cloudflared / ngrok / devtunnel) and
+set the channel's webhook URL in the LINE developers console to:
+    https://<tunnel-host>/line/webhook
+Click "Verify" in the console — you should see a successful signature
+verification logged here.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import sys
+from pathlib import Path
+
+# Make the repo importable when running as a script.
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+log = logging.getLogger("line-test")
+
+
+async def main() -> None:
+    if not os.getenv("LINE_CHANNEL_ACCESS_TOKEN") or not os.getenv("LINE_CHANNEL_SECRET"):
+        sys.exit(
+            "Set LINE_CHANNEL_ACCESS_TOKEN and LINE_CHANNEL_SECRET before running."
+        )
+
+    # Lazy-import so the missing-deps message above fires before any heavy
+    # gateway import work.
+    from gateway.config import PlatformConfig
+    from plugins.platforms.line.adapter import LineAdapter
+
+    cfg = PlatformConfig(enabled=True, extra={})
+    adapter = LineAdapter(cfg)
+
+    # Replace handle_message with a printer so events surface without the
+    # full gateway pipeline.
+    async def _print_event(event):  # type: ignore[no-untyped-def]
+        src = event.source
+        log.info(
+            "MSG  chat=%s type=%s user=%s text=%r media=%s",
+            src.chat_id,
+            src.chat_type,
+            src.user_id,
+            event.text,
+            event.media_urls,
+        )
+        # Echo the text back so you can test the outbound path too.
+        if event.text:
+            result = await adapter.send(src.chat_id, f"echo: {event.text}")
+            log.info("SEND ok=%s err=%s", result.success, result.error)
+
+    adapter.handle_message = _print_event  # type: ignore[assignment]
+
+    ok = await adapter.connect()
+    if not ok:
+        sys.exit(f"connect failed: {adapter._fatal_error}")
+
+    log.info(
+        "LINE adapter listening on 0.0.0.0:%d%s — Ctrl-C to stop",
+        adapter._port,
+        adapter._webhook_path,
+    )
+    try:
+        while True:
+            await asyncio.sleep(3600)
+    except (KeyboardInterrupt, asyncio.CancelledError):
+        pass
+    finally:
+        await adapter.disconnect()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
diff --git a/tests/gateway/test_line_adapter.py b/tests/gateway/test_line_adapter.py
new file mode 100644
index 000000000000..0cfac67b397f
--- /dev/null
+++ b/tests/gateway/test_line_adapter.py
@@ -0,0 +1,306 @@
+"""Tests for the LINE platform adapter plugin."""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import hmac
+import json
+import time
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from tests.gateway._plugin_adapter_loader import load_plugin_adapter
+
+
+# Load plugins/platforms/line/adapter.py under a unique module name
+# (plugin_adapter_line) so it cannot collide with other plugin adapters
+# loaded by sibling tests in the same xdist worker.
+_line_mod = load_plugin_adapter("line")
+
+LineAdapter = _line_mod.LineAdapter
+_validate_line_signature = _line_mod._validate_line_signature
+_resolve_chat = _line_mod._resolve_chat
+validate_config = _line_mod.validate_config
+check_requirements = _line_mod.check_requirements
+register = _line_mod.register
+
+
+# ── Signature validation ──────────────────────────────────────────────────────
+
+
+class TestSignatureValidation:
+
+    def _sign(self, body: bytes, secret: str) -> str:
+        return base64.b64encode(
+            hmac.new(secret.encode(), body, hashlib.sha256).digest()
+        ).decode()
+
+    def test_valid_signature_accepted(self):
+        body = b'{"events":[]}'
+        secret = "abc123"
+        sig = self._sign(body, secret)
+        assert _validate_line_signature(body, sig, secret) is True
+
+    def test_wrong_secret_rejected(self):
+        body = b'{"events":[]}'
+        sig = self._sign(body, "abc123")
+        assert _validate_line_signature(body, sig, "wrong-secret") is False
+
+    def test_tampered_body_rejected(self):
+        secret = "abc123"
+        sig = self._sign(b'{"events":[]}', secret)
+        assert (
+            _validate_line_signature(b'{"events":[{"foo":1}]}', sig, secret) is False
+        )
+
+    def test_missing_signature_rejected(self):
+        assert _validate_line_signature(b"x", "", "secret") is False
+
+    def test_missing_secret_rejected(self):
+        assert _validate_line_signature(b"x", "anysig", "") is False
+
+    def test_garbage_signature_rejected(self):
+        # Base64-decoded length mismatch must not raise.
+        assert (
+            _validate_line_signature(b"x", "definitely-not-base64-***", "secret")
+            is False
+        )
+
+
+# ── Source / chat-id resolution ───────────────────────────────────────────────
+
+
+class TestResolveChat:
+
+    def test_user_source(self):
+        assert _resolve_chat({"type": "user", "userId": "U123"}) == ("U123", "dm")
+
+    def test_group_source(self):
+        assert _resolve_chat({"type": "group", "groupId": "C456"}) == ("C456", "group")
+
+    def test_room_source(self):
+        # Multi-person rooms — LINE distinguishes them from groups, but for
+        # session routing they behave like groups.
+        assert _resolve_chat({"type": "room", "roomId": "R789"}) == ("R789", "group")
+
+    def test_unknown_source_returns_empty(self):
+        assert _resolve_chat({"type": "wat"}) == ("", "dm")
+
+    def test_missing_id_returns_empty(self):
+        assert _resolve_chat({"type": "group"}) == ("", "group")
+
+
+# ── validate_config ───────────────────────────────────────────────────────────
+
+
+class TestValidateConfig:
+
+    def test_valid_with_extra(self):
+        cfg = MagicMock()
+        cfg.extra = {"channel_access_token": "tok", "channel_secret": "sec"}
+        assert validate_config(cfg) is True
+
+    def test_valid_with_env(self, monkeypatch):
+        monkeypatch.setenv("LINE_CHANNEL_ACCESS_TOKEN", "tok")
+        monkeypatch.setenv("LINE_CHANNEL_SECRET", "sec")
+        cfg = MagicMock()
+        cfg.extra = {}
+        assert validate_config(cfg) is True
+
+    def test_missing_token_invalid(self, monkeypatch):
+        monkeypatch.delenv("LINE_CHANNEL_ACCESS_TOKEN", raising=False)
+        monkeypatch.delenv("LINE_CHANNEL_SECRET", raising=False)
+        cfg = MagicMock()
+        cfg.extra = {"channel_secret": "sec"}
+        assert validate_config(cfg) is False
+
+    def test_missing_secret_invalid(self, monkeypatch):
+        monkeypatch.delenv("LINE_CHANNEL_ACCESS_TOKEN", raising=False)
+        monkeypatch.delenv("LINE_CHANNEL_SECRET", raising=False)
+        cfg = MagicMock()
+        cfg.extra = {"channel_access_token": "tok"}
+        assert validate_config(cfg) is False
+
+
+# ── Reply token TTL ───────────────────────────────────────────────────────────
+
+
+def _make_adapter(monkeypatch) -> "LineAdapter":
+    monkeypatch.delenv("LINE_PORT", raising=False)
+    monkeypatch.setenv("LINE_CHANNEL_ACCESS_TOKEN", "tok")
+    monkeypatch.setenv("LINE_CHANNEL_SECRET", "sec")
+    from gateway.config import PlatformConfig
+
+    return LineAdapter(PlatformConfig(enabled=True))
+
+
+class TestReplyTokenLifecycle:
+
+    def test_unknown_chat_returns_none(self, monkeypatch):
+        a = _make_adapter(monkeypatch)
+        assert a._consume_reply_token("Uunknown") is None
+
+    def test_fresh_token_consumed_once(self, monkeypatch):
+        a = _make_adapter(monkeypatch)
+        a._reply_tokens["Uchat"] = ("rt-fresh", time.time())
+        assert a._consume_reply_token("Uchat") == "rt-fresh"
+        # Reply tokens are single-use — second call gets None.
+        assert a._consume_reply_token("Uchat") is None
+
+    def test_expired_token_dropped(self, monkeypatch):
+        a = _make_adapter(monkeypatch)
+        # Token captured 5 minutes ago — well past LINE's 60s validity
+        # window and our internal 50s cap.
+        a._reply_tokens["Uchat"] = ("rt-stale", time.time() - 300)
+        assert a._consume_reply_token("Uchat") is None
+        # Stale entry was popped, not left behind.
+        assert "Uchat" not in a._reply_tokens
+
+
+# ── Outbound: reply with push fallback ────────────────────────────────────────
+
+
+class TestSendMessages:
+
+    @pytest.mark.asyncio
+    async def test_reply_used_when_token_fresh(self, monkeypatch):
+        a = _make_adapter(monkeypatch)
+        a._reply_tokens["Uchat"] = ("rt-fresh", time.time())
+        a._post = AsyncMock(return_value=(True, ""))
+
+        result = await a._send_messages(
+            "Uchat", [{"type": "text", "text": "hi"}]
+        )
+
+        assert result.success is True
+        a._post.assert_awaited_once()
+        path, payload = a._post.call_args.args
+        assert path == "/message/reply"
+        assert payload["replyToken"] == "rt-fresh"
+        assert payload["messages"] == [{"type": "text", "text": "hi"}]
+
+    @pytest.mark.asyncio
+    async def test_push_used_when_no_token(self, monkeypatch):
+        a = _make_adapter(monkeypatch)
+        a._post = AsyncMock(return_value=(True, ""))
+
+        result = await a._send_messages(
+            "Uchat", [{"type": "text", "text": "hi"}]
+        )
+
+        assert result.success is True
+        path, payload = a._post.call_args.args
+        assert path == "/message/push"
+        assert payload["to"] == "Uchat"
+
+    @pytest.mark.asyncio
+    async def test_reply_failure_falls_back_to_push(self, monkeypatch):
+        a = _make_adapter(monkeypatch)
+        a._reply_tokens["Uchat"] = ("rt-already-used", time.time())
+        # First call (reply) fails; second call (push) succeeds.
+        a._post = AsyncMock(
+            side_effect=[(False, "HTTP 400: Invalid reply token"), (True, "")]
+        )
+
+        result = await a._send_messages(
+            "Uchat", [{"type": "text", "text": "hi"}]
+        )
+
+        assert result.success is True
+        assert a._post.await_count == 2
+        assert a._post.call_args_list[0].args[0] == "/message/reply"
+        assert a._post.call_args_list[1].args[0] == "/message/push"
+
+    @pytest.mark.asyncio
+    async def test_push_failure_returns_retryable_error(self, monkeypatch):
+        a = _make_adapter(monkeypatch)
+        a._post = AsyncMock(return_value=(False, "HTTP 500: oh no"))
+
+        result = await a._send_messages(
+            "Uchat", [{"type": "text", "text": "hi"}]
+        )
+
+        assert result.success is False
+        assert result.retryable is True
+        assert "500" in (result.error or "")
+
+    @pytest.mark.asyncio
+    async def test_messages_batched_in_groups_of_five(self, monkeypatch):
+        a = _make_adapter(monkeypatch)
+        a._post = AsyncMock(return_value=(True, ""))
+
+        msgs = [{"type": "text", "text": str(i)} for i in range(12)]
+        result = await a._send_messages("Uchat", msgs)
+
+        assert result.success is True
+        # 12 messages → 3 batches (5 + 5 + 2).
+        assert a._post.await_count == 3
+        sizes = [len(call.args[1]["messages"]) for call in a._post.call_args_list]
+        assert sizes == [5, 5, 2]
+
+
+# ── send_typing (loading animation API) ───────────────────────────────────────
+
+
+class TestSendTyping:
+
+    @pytest.mark.asyncio
+    async def test_typing_calls_loading_animation_for_user(self, monkeypatch):
+        a = _make_adapter(monkeypatch)
+        a._http = MagicMock()  # presence-only; _post is mocked below
+        a._post = AsyncMock(return_value=(True, ""))
+
+        await a.send_typing("Uchat")
+
+        a._post.assert_awaited_once()
+        path, payload = a._post.call_args.args
+        assert path == "/chat/loading/start"
+        assert payload["chatId"] == "Uchat"
+        assert payload["loadingSeconds"] in range(5, 65, 5)
+
+    @pytest.mark.asyncio
+    async def test_typing_skipped_for_group_chat(self, monkeypatch):
+        # LINE's loading animation endpoint only works for 1:1 user chats —
+        # calling it with a group/room ID would 400.  We skip it client-side.
+        a = _make_adapter(monkeypatch)
+        a._http = MagicMock()
+        a._post = AsyncMock(return_value=(True, ""))
+
+        await a.send_typing("Cgroup-id")
+        await a.send_typing("Rroom-id")
+
+        a._post.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_typing_swallows_failures(self, monkeypatch):
+        # A 400 from LINE shouldn't break the inbound dispatch path.
+        a = _make_adapter(monkeypatch)
+        a._http = MagicMock()
+        a._post = AsyncMock(return_value=(False, "HTTP 400: bad"))
+
+        await a.send_typing("Uchat")  # must not raise
+
+
+# ── register() smoke test ─────────────────────────────────────────────────────
+
+
+def test_register_emits_expected_metadata():
+    """register() should call ctx.register_platform with the right keys."""
+    captured = {}
+
+    def fake_register(**kwargs):
+        captured.update(kwargs)
+
+    ctx = MagicMock()
+    ctx.register_platform.side_effect = fake_register
+    register(ctx)
+
+    assert captured["name"] == "line"
+    assert captured["label"] == "LINE"
+    assert "LINE_CHANNEL_ACCESS_TOKEN" in captured["required_env"]
+    assert "LINE_CHANNEL_SECRET" in captured["required_env"]
+    assert captured["allowed_users_env"] == "LINE_ALLOWED_USERS"
+    assert captured["allow_all_env"] == "LINE_ALLOW_ALL_USERS"