diff --git a/.gitignore b/.gitignore index c245b2e49..27f4642ac 100644 --- a/.gitignore +++ b/.gitignore @@ -31,6 +31,7 @@ bard.json !.pre-commit-config.yaml !components.yaml !/docker-compose.yaml +docker/data data/labels/instance_id.json .DS_Store /data diff --git a/dev-docs/agent-runner-pluginization/EVENT_DEBUG_FULL_QA_2026-09-05.md b/dev-docs/agent-runner-pluginization/EVENT_DEBUG_FULL_QA_2026-09-05.md index 1f85e05ea..83c75120c 100644 --- a/dev-docs/agent-runner-pluginization/EVENT_DEBUG_FULL_QA_2026-09-05.md +++ b/dev-docs/agent-runner-pluginization/EVENT_DEBUG_FULL_QA_2026-09-05.md @@ -96,4 +96,4 @@ The full regression also found Windows portability problems in the skills toolin 2. **Bounded stream pressure can fail long responses.** Two recovery runs emitted thinking and then `Streaming action consumer is too slow; response buffer full`; later recovery and six-tool runs succeeded. The error originates in the SDK's existing 128-frame response queue and is preserved in the UI. It is not a missing-tool-result or 30-second HTTP timeout. Full end-to-end flow control under sustained overload remains open; queue limits were not removed to conceal the failure. 3. Mock results verify debug behavior, not real platform credentials, permissions, delivery, or every adapter/event Cartesian combination. Already launched external tool termination timing was not independently measured. A development-branch push does not mean these remaining production release gates passed. -For reproduction, use the retained test Agent, the Mock examples in [RUNNER_QA_GUIDE.md](./RUNNER_QA_GUIDE.md), and the original welcome Agent. Local diagnostic traces and fixture artifacts are excluded from Git; they contain model responses and local runtime state. This report contains outcomes rather than raw provider reasoning or secrets. +For reproduction, use the retained test Agent, the Mock examples in [AGENT_RUNNER_QA_GUIDE.md](./AGENT_RUNNER_QA_GUIDE.md), and the original welcome Agent. Local diagnostic traces and fixture artifacts are excluded from Git; they contain model responses and local runtime state. This report contains outcomes rather than raw provider reasoning or secrets. diff --git a/dev-docs/agent-runner-pluginization/HOST_SDK_INFRASTRUCTURE.md b/dev-docs/agent-runner-pluginization/HOST_SDK_INFRASTRUCTURE.md index daa303932..c9cca99ab 100644 --- a/dev-docs/agent-runner-pluginization/HOST_SDK_INFRASTRUCTURE.md +++ b/dev-docs/agent-runner-pluginization/HOST_SDK_INFRASTRUCTURE.md @@ -3,7 +3,7 @@ 本文档描述 LangBot 作为 agent host 的内部能力与分层架构，以及 Host 内部模型。 - SDK ↔ Host 的协议数据结构（`RunnerContext`、`RunnerManifest`、`RunnerResult`、`RunnerAPIProxy` 等）的**唯一定义在** [PROTOCOL_V1.md](./PROTOCOL_V1.md)；本文只引用，不重抄。 -- 测试执行入口和 smoke 记录见 [RUNNER_QA_GUIDE.md](./RUNNER_QA_GUIDE.md)；安全发布门槛见 [SECURITY_HARDENING.md](./SECURITY_HARDENING.md)。 +- 测试执行入口和 smoke 记录见 [AGENT_RUNNER_QA_GUIDE.md](./AGENT_RUNNER_QA_GUIDE.md)；安全发布门槛见 [SECURITY_HARDENING.md](./SECURITY_HARDENING.md)。 - 本文定义的 Host 内部模型（`AgentEventEnvelope`、`AgentBinding`、`RunnerDescriptor`）不属于 SDK 协议字段。 ## 1. 目标 diff --git a/dev-docs/agent-runner-pluginization/OFFICIAL_RUNNER_PLUGINS.md b/dev-docs/agent-runner-pluginization/OFFICIAL_RUNNER_PLUGINS.md index 9f632af9a..a4a140d15 100644 --- a/dev-docs/agent-runner-pluginization/OFFICIAL_RUNNER_PLUGINS.md +++ b/dev-docs/agent-runner-pluginization/OFFICIAL_RUNNER_PLUGINS.md @@ -1,6 +1,6 @@ # 官方 Runner 插件迁移计划 -本文档描述内置 `RequestRunner` 迁出 LangBot 后，官方 runner 插件如何组织、迁移和验收。它是 [HOST_SDK_INFRASTRUCTURE.md](./HOST_SDK_INFRASTRUCTURE.md) 和 [AGENT_CONTEXT_PROTOCOL.md](./AGENT_CONTEXT_PROTOCOL.md) 的下游落地计划，不是 LangBot 宿主协议的设计前提。QA 入口和 smoke 记录见 [RUNNER_QA_GUIDE.md](./RUNNER_QA_GUIDE.md)。 +本文档描述内置 `RequestRunner` 迁出 LangBot 后，官方 runner 插件如何组织、迁移和验收。它是 [HOST_SDK_INFRASTRUCTURE.md](./HOST_SDK_INFRASTRUCTURE.md) 和 [AGENT_CONTEXT_PROTOCOL.md](./AGENT_CONTEXT_PROTOCOL.md) 的下游落地计划，不是 LangBot 宿主协议的设计前提。QA 入口和 smoke 记录见 [AGENT_RUNNER_QA_GUIDE.md](./AGENT_RUNNER_QA_GUIDE.md)。 官方 `local-agent` 可以外移，也可以重写。设计重点不是保留旧内置 runner 的内部结构，而是验证一个依附 LangBot host 基础设施的官方 agent 能否完整工作。同时，LangBot host 协议必须服务 Claude Code SDK、Codex、Pi Agent SDK、外部 Agent 平台等自管 context/runtime 的 runner，不能被官方插件的实现细节绑死。 @@ -110,7 +110,7 @@ Claude Code、Codex、Kimi Code 这类 runner 不一定通过 LangBot 的模型/ ## 7. Code-agent harness runner 当前形态 -外部 code-agent harness 由直接 runner 插件承接，例如 `acp-agent-runner`、`claude-code-agent`、`codex-agent`，每个 runner 负责把目标 harness 的 native session、workspace、MCP bridge 和输出事件转换为统一 `RunnerResult`。本地 smoke 验收入口与记录见 [RUNNER_QA_GUIDE.md](./RUNNER_QA_GUIDE.md)。 +外部 code-agent harness 由直接 runner 插件承接，例如 `acp-agent-runner`、`claude-code-agent`、`codex-agent`，每个 runner 负责把目标 harness 的 native session、workspace、MCP bridge 和输出事件转换为统一 `RunnerResult`。本地 smoke 验收入口与记录见 [AGENT_RUNNER_QA_GUIDE.md](./AGENT_RUNNER_QA_GUIDE.md)。 当前形态： diff --git a/dev-docs/agent-runner-pluginization/PROTOCOL_V1.md b/dev-docs/agent-runner-pluginization/PROTOCOL_V1.md index cc6973a8c..2e8af53d0 100644 --- a/dev-docs/agent-runner-pluginization/PROTOCOL_V1.md +++ b/dev-docs/agent-runner-pluginization/PROTOCOL_V1.md @@ -2,7 +2,7 @@ 本文档是 LangBot Host 与插件 SDK / Runtime / Runner 之间协议合同的**唯一规范来源（single source of truth）**。 -- 本文件描述当前 Protocol v1 稳定合同，不混入验收流水。当前实现状态见 [STATUS.md](./STATUS.md)，测试执行入口见 [RUNNER_QA_GUIDE.md](./RUNNER_QA_GUIDE.md)，安全发布门槛见 [SECURITY_HARDENING.md](./SECURITY_HARDENING.md)。 +- 本文件描述当前 Protocol v1 稳定合同，不混入验收流水。当前实现状态见 [STATUS.md](./STATUS.md)，测试执行入口见 [AGENT_RUNNER_QA_GUIDE.md](./AGENT_RUNNER_QA_GUIDE.md)，安全发布门槛见 [SECURITY_HARDENING.md](./SECURITY_HARDENING.md)。 - 本文件之外的任何文档**不得重新定义这里的数据结构**，只能引用，例如"见 PROTOCOL_V1 §4.2"。 - Host 内部模型（`AgentEventEnvelope`、`AgentBinding`、Descriptor、各 Store）不属于 SDK 协议，定义在 [HOST_SDK_INFRASTRUCTURE.md](./HOST_SDK_INFRASTRUCTURE.md)。 diff --git a/dev-docs/agent-runner-pluginization/README.md b/dev-docs/agent-runner-pluginization/README.md index 8d6708211..48ba2c738 100644 --- a/dev-docs/agent-runner-pluginization/README.md +++ b/dev-docs/agent-runner-pluginization/README.md @@ -9,7 +9,7 @@ ## 文档维护原则（单一事实源） - **协议数据结构（schema）唯一定义在 [PROTOCOL_V1.md](./PROTOCOL_V1.md)。** 其他文档不得重抄 schema，只能引用，例如"见 PROTOCOL_V1 §4.2"。 -- 当前实现状态、spec 差距与 runner 验收状态归 [STATUS.md](./STATUS.md)；测试执行入口归 [RUNNER_QA_GUIDE.md](./RUNNER_QA_GUIDE.md)，安全发布门槛归 [SECURITY_HARDENING.md](./SECURITY_HARDENING.md)。 +- 当前实现状态、spec 差距与 runner 验收状态归 [STATUS.md](./STATUS.md)；测试执行入口归 [AGENT_RUNNER_QA_GUIDE.md](./AGENT_RUNNER_QA_GUIDE.md)，安全发布门槛归 [SECURITY_HARDENING.md](./SECURITY_HARDENING.md)。 - Host 内部模型（`AgentEventEnvelope`、`AgentBinding`、Descriptor、各 Store）定义在 [HOST_SDK_INFRASTRUCTURE.md](./HOST_SDK_INFRASTRUCTURE.md)，不属于 SDK 协议。 - 其余专题文档只讲"为什么/边界/怎么用"，避免重复叙述。 @@ -54,7 +54,7 @@ EBA 先根据 `target_type` 选择 Pipeline 或 Agent。Pipeline 目标执行完整 Stage 链；当 Pipeline 的 AI Stage 调用 runner 时，`run_from_query()` 经 `QueryEntryAdapter` 把 `Query` 转换为 `AgentEventEnvelope` + `AgentBinding`，再委托到统一的 `run(event, binding, ...)`。Agent 目标则直接从自己的持久配置构造 binding。两条路径可以复用同一套 Runner Host capabilities，但 Pipeline 本身不会被投影或持久化为 Agent。 -下一轮测试路径、状态定义和 smoke 记录见 [RUNNER_QA_GUIDE.md](./RUNNER_QA_GUIDE.md)。 +下一轮测试路径、状态定义和 smoke 记录见 [AGENT_RUNNER_QA_GUIDE.md](./AGENT_RUNNER_QA_GUIDE.md)。 ## 术语表 @@ -87,7 +87,7 @@ EBA 先根据 `target_type` 选择 Pipeline 或 Agent。Pipeline 目标执行完 | [RUN_STEERING_AND_CHECKPOINT.md](./RUN_STEERING_AND_CHECKPOINT.md) | 运行中消息注入（steering / follow-up）与压缩摘要持久化（compaction checkpoint）的设计与落地状态记录；schema 仍以 PROTOCOL_V1 为准。 | | [STATUS.md](./STATUS.md) | 当前实现状态、spec 与实现已知差距、runner 验收状态和历史高价值记录。 | | [PLATFORM_ACTION_TOOLS.md](./PLATFORM_ACTION_TOOLS.md) | 当前事件工具、平台动作和普通工具的配置、投射及执行授权。 | -| [RUNNER_QA_GUIDE.md](./RUNNER_QA_GUIDE.md) | Runner QA 指南：保留最高价值测试路径，指导 agent 开展下一轮 WebUI / runner smoke 验证。 | +| [AGENT_RUNNER_QA_GUIDE.md](./AGENT_RUNNER_QA_GUIDE.md) | Runner QA 指南：保留最高价值测试路径，指导 agent 开展下一轮 WebUI / runner smoke 验证。 | | [SECURITY_HARDENING.md](./SECURITY_HARDENING.md) | 安全发布级 hardening 的后续发布门槛：路径隔离、权限边界、secret、资源配额、MCP / skill 投影和审计。 | ## 工作拆分 diff --git a/dev-docs/agent-runner-pluginization/RUNTIME_CONTROL_PLANE_V2.md b/dev-docs/agent-runner-pluginization/RUNTIME_CONTROL_PLANE_V2.md index 6f43ec877..a5c285d4e 100644 --- a/dev-docs/agent-runner-pluginization/RUNTIME_CONTROL_PLANE_V2.md +++ b/dev-docs/agent-runner-pluginization/RUNTIME_CONTROL_PLANE_V2.md @@ -2,7 +2,7 @@ 本文档记录 Runner 插件化之后，LangBot 如何继续演进成 Agent Platform 基础设施层。这里讨论的是 Host capability layer，不是 `Runner Protocol v2`，也不是把某个具体 Agent Platform 产品写进 LangBot core。 -> 本文是当前决策版。协议数据结构仍以 [PROTOCOL_V1.md](./PROTOCOL_V1.md) 为准；测试执行入口见 [RUNNER_QA_GUIDE.md](./RUNNER_QA_GUIDE.md)；扩展边界见 [EXTENSION_SCOPE_MATRIX.md](./EXTENSION_SCOPE_MATRIX.md)。 +> 本文是当前决策版。协议数据结构仍以 [PROTOCOL_V1.md](./PROTOCOL_V1.md) 为准；测试执行入口见 [AGENT_RUNNER_QA_GUIDE.md](./AGENT_RUNNER_QA_GUIDE.md)；扩展边界见 [EXTENSION_SCOPE_MATRIX.md](./EXTENSION_SCOPE_MATRIX.md)。 > > 实现状态说明：本文描述的是 Runtime Control Plane v2 的目标能力和分阶段落地建议。当前 Runner 插件化主线已经具备 event-first context、run-scoped authorization、EventLog / Transcript / State / sandbox 文件等 Host capability，并已落地持久 `AgentRun` / `AgentRunEvent` ledger、run control actions、最小 runtime heartbeat/claim lease 和 admin reconcile 原语。完整 Agent Platform 产品形态、daemon supervisor、runtime wakeup channel 和分布式 runtime 管控仍未完成。当前实现状态以 [STATUS.md](./STATUS.md) 为准。 diff --git a/dev-docs/agent-runner-pluginization/STATUS.md b/dev-docs/agent-runner-pluginization/STATUS.md index 70acf22ae..a58eaaa50 100644 --- a/dev-docs/agent-runner-pluginization/STATUS.md +++ b/dev-docs/agent-runner-pluginization/STATUS.md @@ -1,6 +1,6 @@ # Runner Pluginization Status -本文档是 `dev-docs/agent-runner-pluginization/` 的状态事实源。协议 schema 仍以 [PROTOCOL_V1.md](./PROTOCOL_V1.md) 为准；测试步骤以 [RUNNER_QA_GUIDE.md](./RUNNER_QA_GUIDE.md) 为准；安全发布门槛以 [SECURITY_HARDENING.md](./SECURITY_HARDENING.md) 为准。 +本文档是 `dev-docs/agent-runner-pluginization/` 的状态事实源。协议 schema 仍以 [PROTOCOL_V1.md](./PROTOCOL_V1.md) 为准；测试步骤以 [AGENT_RUNNER_QA_GUIDE.md](./AGENT_RUNNER_QA_GUIDE.md) 为准；安全发布门槛以 [SECURITY_HARDENING.md](./SECURITY_HARDENING.md) 为准。 状态快照日期：2026-09-05。代码基线为 LangBot `dev/4.11.x` / `a4d36aa2d` 和 SDK `dev/4.11.x` / `f1da058`；本文是本地检视快照，不代表远端最新状态或正式发布批准。 diff --git a/dev-docs/event-based-agents/adapters/line.md b/dev-docs/event-based-agents/adapters/line.md new file mode 100644 index 000000000..1b55e65e4 --- /dev/null +++ b/dev-docs/event-based-agents/adapters/line.md @@ -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. diff --git a/dev-docs/review/minimax-protocol.md b/dev-docs/review/minimax-protocol.md new file mode 100644 index 000000000..32d60344b --- /dev/null +++ b/dev-docs/review/minimax-protocol.md @@ -0,0 +1,50 @@ +# MiniMax requester compatibility + +Implementation reference: `https://platform.minimax.io/docs/api-reference/text-openai-api.md` +(checked 2026-10-30). + +## Discovery and saved configurations + +New MiniMax configurations use `https://api.minimax.io/v1`. Saved endpoints and +manually selected model identifiers are not migrated or rewritten. + +Direct discovery recommends MiniMax-M3.1-Flash followed by MiniMax-M3, including +when the direct service has no model-list endpoint (404). Authentication, +rate-limit, and server errors still fail the scan. Documented capabilities and +the 1,000,000-token context are defaults for these direct recommendations; +advertised context lengths are retained. + +OpenRouter discovery never invents routes or replaces advertised metadata. +All advertised Flash variants precede M3. Only known legacy MiniMax text +families (M1/M2, MiniMax-01, abab) are removed from discovery; non-text entries, +other vendors, custom identifiers, and future model families remain available. +This filtering does not prevent saved/manual legacy models from being invoked. + +## Thinking and conversation history + +- Flash supports provider defaults, adaptive thinking (`enabled`), and + `low`, `medium`, `high`, `xhigh`, `max` effort; disabling is rejected locally. +- M3 supports provider defaults, adaptive thinking, and disabled thinking. +- The default policy sends no additional thinking, admission-tier, or + output-format parameters. Gateway requests keep the existing gateway protocol. +- `remove_think` changes display text, not the assistant history sent back to + MiniMax. Both `reasoning_content` and the original inline `` content are + retained, including when the next M3 request disables thinking. + +`minimax_raw_content` is internal provider metadata. Non-streaming responses +store their original content before display filtering. Streaming responses +buffer raw content and emit one complete snapshot at the finish event, or at +normal iterator exhaustion when the finish event is absent. This matches Local +Agent's opaque-field replacement semantics: only `reasoning_content` is a +concatenated delta there. Sending raw-content deltas would silently retain only +the last fragment. The internal field is removed during request conversion and +is not replayed as content when the user switches to another provider. + +## Regression coverage + +`tests/unit_tests/provider/test_minimax_protocol.py` covers real locked SDK +serialization against mocked HTTP/SSE, saved endpoints and identifiers, +thinking validation, image input, discovery failures and metadata, streaming +termination, JSON history round trips, tool follow-ups, and provider switching. +It makes no paid inference calls. The provider suite remains the shared-provider +regression check; Python packaging is independent of frontend/runtime deployment. diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index fe1aecba0..cf74c6711 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -1,7 +1,5 @@ # Docker Compose configuration for LangBot # For Kubernetes deployment, see kubernetes.yaml and the deployment guide at https://langbot.app/docs -version: "3" - services: langbot_plugin_runtime: @@ -16,6 +14,10 @@ services: - ./data/plugin-runtime:/app/data/plugin-runtime ports: - 5401:5401 + # ACP Agent Runner daemon hub. With `daemon-enabled = true` in the plugin + # config, a user-side runner daemon behind NAT connects outward to + # ws://:8766. Harmless when the hub is disabled. + - 8766:8766 restart: on-failure environment: - TZ=Asia/Shanghai diff --git a/src/langbot/pkg/platform/adapter_names.py b/src/langbot/pkg/platform/adapter_names.py index be5c1f56b..debd12e5e 100644 --- a/src/langbot/pkg/platform/adapter_names.py +++ b/src/langbot/pkg/platform/adapter_names.py @@ -13,6 +13,7 @@ OMNI_ADAPTER_NAMES = frozenset( 'wecombot', 'wecomcs', 'officialaccount', + 'line', } ) diff --git a/src/langbot/pkg/platform/adapters/line/adapter.py b/src/langbot/pkg/platform/adapters/line/adapter.py new file mode 100644 index 000000000..9f2a6d52c --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/adapter.py @@ -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 '' diff --git a/src/langbot/pkg/platform/adapters/line/api_impl.py b/src/langbot/pkg/platform/adapters/line/api_impl.py new file mode 100644 index 000000000..263bb06f2 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/api_impl.py @@ -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') diff --git a/src/langbot/pkg/platform/adapters/line/errors.py b/src/langbot/pkg/platform/adapters/line/errors.py new file mode 100644 index 000000000..b56f459ae --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/errors.py @@ -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 diff --git a/src/langbot/pkg/platform/adapters/line/event_converter.py b/src/langbot/pkg/platform/adapters/line/event_converter.py new file mode 100644 index 000000000..605c58910 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/event_converter.py @@ -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) diff --git a/src/langbot/pkg/platform/adapters/line/line.png b/src/langbot/pkg/platform/adapters/line/line.png new file mode 100644 index 000000000..8c925c6c0 Binary files /dev/null and b/src/langbot/pkg/platform/adapters/line/line.png differ diff --git a/src/langbot/pkg/platform/adapters/line/manifest.yaml b/src/langbot/pkg/platform/adapters/line/manifest.yaml new file mode 100644 index 000000000..79c47b42d --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/manifest.yaml @@ -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 diff --git a/src/langbot/pkg/platform/adapters/line/message_converter.py b/src/langbot/pkg/platform/adapters/line/message_converter.py new file mode 100644 index 000000000..aaeecc748 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/message_converter.py @@ -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) diff --git a/src/langbot/pkg/platform/adapters/line/platform_api.py b/src/langbot/pkg/platform/adapters/line/platform_api.py new file mode 100644 index 000000000..a331b5758 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/platform_api.py @@ -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, +} diff --git a/src/langbot/pkg/platform/adapters/line/types.py b/src/langbot/pkg/platform/adapters/line/types.py new file mode 100644 index 000000000..7a62600b5 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/types.py @@ -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 diff --git a/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py b/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py index c8bb662cd..16eb776c4 100644 --- a/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py +++ b/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py @@ -24,6 +24,7 @@ from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler from litellm.secret_managers.main import get_secret, get_secret_bool from .. import errors, reasoning, requester +from . import minimax from ....utils import httpclient, image import langbot_plugin.api.entities.builtin.resource.tool as resource_tool import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query @@ -213,6 +214,7 @@ class LiteLLMRequester(requester.ProviderAPIRequester): 'bailian-chat-completions': 'qwen', 'doubao-chat-completions': 'doubao', 'mimo-chat-completions': 'mimo', + 'minimax-chat-completions': 'minimax', } default_config: dict[str, typing.Any] = { @@ -569,6 +571,9 @@ class LiteLLMRequester(requester.ProviderAPIRequester): def _known_reasoning_levels(self, model_name: str, family: str) -> list[str] | None: normalized_name = model_name.lower().rsplit('/', 1)[-1] + if family == 'minimax': + return minimax.reasoning_levels(normalized_name) + if family == 'deepseek' and normalized_name.startswith('deepseek-'): if normalized_name.startswith('deepseek-v4-'): return ['provider_default', 'disabled', 'low', 'high', 'xhigh', 'max'] @@ -728,6 +733,8 @@ class LiteLLMRequester(requester.ProviderAPIRequester): raise errors.RequesterError(str(exc), code='invalid_policy') from exc family = self._reasoning_family(model.model_entity.name, model) + if family == 'minimax': + return minimax.reasoning_parameters(level) budget = config.get('budget_tokens') if budget is not None: if self._uses_openrouter_protocol(): @@ -837,6 +844,7 @@ class LiteLLMRequester(requester.ProviderAPIRequester): for m in messages: msg_dict = m.dict(exclude_none=True) provider_fields = msg_dict.get('provider_specific_fields') + replay_fields = minimax.take_replay_fields(provider_fields) if isinstance(provider_fields, dict) else {} if isinstance(provider_fields, dict): # Stream completion metadata belongs to the response, never to # replayed chat messages. Keep provider signatures for LiteLLM. @@ -878,6 +886,9 @@ class LiteLLMRequester(requester.ProviderAPIRequester): else: msg_dict.pop('provider_specific_fields', None) + if msg_dict.get('role') == 'assistant' and reasoning_family == 'minimax': + msg_dict.update(replay_fields) + if isinstance(content, list): converted_parts = [] for part in content: @@ -1476,6 +1487,8 @@ class LiteLLMRequester(requester.ProviderAPIRequester): content = message_data.get('content', '') reasoning_content = message_data.get('reasoning_content', None) thinking_blocks = message_data.get('thinking_blocks') + if self._reasoning_family(model.model_entity.name, model) == 'minimax': + message_data['provider_specific_fields'] = minimax.capture_content(message_data) if reasoning_content or thinking_blocks: provider_fields = dict(message_data.get('provider_specific_fields') or {}) if reasoning_content: @@ -1520,6 +1533,8 @@ class LiteLLMRequester(requester.ProviderAPIRequester): reasoning_started = False reasoning_closed = False thinking_blocks_state: list[dict[str, typing.Any]] = [] + keep_minimax_content = self._reasoning_family(model.model_entity.name, model) == 'minimax' + raw_content_parts: list[str] = [] response = None async with self._client_args(args) as args: @@ -1562,6 +1577,12 @@ class LiteLLMRequester(requester.ProviderAPIRequester): delta_content = delta.get('content') or '' reasoning_content = delta.get('reasoning_content') or '' provider_fields = dict(delta.get('provider_specific_fields') or {}) + if keep_minimax_content: + if isinstance(delta.get('content'), str): + raw_content_parts.append(delta['content']) + if finish_reason and raw_content_parts: + provider_fields[minimax.RAW_CONTENT_FIELD] = ''.join(raw_content_parts) + raw_content_parts.clear() if finish_reason: # Preserve an explicit provider stop even when the final delta is empty. provider_fields['finish_reason'] = finish_reason @@ -1637,6 +1658,13 @@ class LiteLLMRequester(requester.ProviderAPIRequester): yield provider_message.MessageChunk(**chunk_data) chunk_idx += 1 + if raw_content_parts: + yield provider_message.MessageChunk( + role=role, + is_final=True, + provider_specific_fields={minimax.RAW_CONTENT_FIELD: ''.join(raw_content_parts)}, + ) + if reasoning_started and not reasoning_closed: yield provider_message.MessageChunk( role=role, @@ -1882,8 +1910,11 @@ class LiteLLMRequester(requester.ProviderAPIRequester): # rather than /models. Providers whose configured base_url # already ends in /v1 keep their original (working) URL. response = await client.get(f'{base_url}/v1/models', headers=headers) - response.raise_for_status() - payload = await httpclient.parse_json_response(response) + if self._requester_name() == 'minimax-chat-completions' and response.status_code == 404: + payload = {} + else: + response.raise_for_status() + payload = await httpclient.parse_json_response(response) models = [] for item in payload.get('data', []): @@ -1895,6 +1926,10 @@ class LiteLLMRequester(requester.ProviderAPIRequester): models.sort(key=lambda x: (x['type'] != 'llm', x['name'].lower())) + direct_minimax = self._requester_name() == 'minimax-chat-completions' + if direct_minimax or self._uses_openrouter_protocol(): + models = minimax.arrange_catalog(models, direct=direct_minimax) + return {'models': models} except httpx.HTTPStatusError as e: diff --git a/src/langbot/pkg/provider/modelmgr/requesters/minimax.py b/src/langbot/pkg/provider/modelmgr/requesters/minimax.py new file mode 100644 index 000000000..9d056806e --- /dev/null +++ b/src/langbot/pkg/provider/modelmgr/requesters/minimax.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import re +from typing import Any + + +RECOMMENDED_MODELS = ('MiniMax-M3.1-Flash', 'MiniMax-M3') +RAW_CONTENT_FIELD = 'minimax_raw_content' +_LEGACY_TEXT_MODEL = re.compile(r'^(minimax-(m[12](?:[.\-]|$)|01(?:[.\-]|$))|abab)') + + +def reasoning_levels(model_name: str) -> list[str] | None: + name = model_name.lower() + if name == RECOMMENDED_MODELS[0].lower(): + return ['provider_default', 'enabled', 'low', 'medium', 'high', 'xhigh', 'max'] + if name == RECOMMENDED_MODELS[1].lower(): + return ['provider_default', 'enabled', 'disabled'] + return None + + +def reasoning_parameters(level: str) -> dict[str, Any]: + if level == 'provider_default': + return {} + thinking_type = {'enabled': 'adaptive', 'disabled': 'disabled'}.get(level) + if thinking_type is not None: + return {'extra_body': {'thinking': {'type': thinking_type}}} + return {'reasoning_effort': level} + + +def capture_content(payload: dict[str, Any]) -> dict[str, Any]: + fields = dict(payload.get('provider_specific_fields') or {}) + if isinstance(payload.get('content'), str): + fields[RAW_CONTENT_FIELD] = payload['content'] + return fields + + +def take_replay_fields(fields: dict[str, Any]) -> dict[str, Any]: + replay = {} + raw_content = fields.pop(RAW_CONTENT_FIELD, None) + if isinstance(raw_content, str): + replay['content'] = raw_content + if isinstance(fields.get('reasoning_content'), str): + replay['reasoning_content'] = fields['reasoning_content'] + return replay + + +def arrange_catalog(models: list[dict[str, Any]], *, direct: bool) -> list[dict[str, Any]]: + priorities = {name.lower(): position for position, name in enumerate(RECOMMENDED_MODELS)} + + def model_name(model: dict[str, Any]) -> str: + name = model['id'].lower() + if not direct: + if not name.startswith('minimax/'): + return '' + name = name.removeprefix('minimax/') + return name.partition(':')[0] + + catalog = [ + dict(model) for model in models if model['type'] != 'llm' or not _LEGACY_TEXT_MODEL.match(model_name(model)) + ] + if direct: + present = {model['id'].lower() for model in catalog} + catalog.extend( + {'id': name, 'name': name, 'type': 'llm'} for name in RECOMMENDED_MODELS if name.lower() not in present + ) + for model in catalog: + if model_name(model) in priorities: + model.setdefault('context_length', 1_000_000) + model['abilities'] = list( + dict.fromkeys([*model.get('abilities', []), 'func_call', 'vision', 'reasoning']) + ) + return sorted(catalog, key=lambda model: priorities.get(model_name(model), len(priorities))) diff --git a/src/langbot/pkg/provider/modelmgr/requesters/minimaxchatcmpl.yaml b/src/langbot/pkg/provider/modelmgr/requesters/minimaxchatcmpl.yaml index b0c246c98..0447fb6d3 100644 --- a/src/langbot/pkg/provider/modelmgr/requesters/minimaxchatcmpl.yaml +++ b/src/langbot/pkg/provider/modelmgr/requesters/minimaxchatcmpl.yaml @@ -15,7 +15,7 @@ spec: zh_Hans: 基础 URL type: string required: true - default: https://api.minimax.chat/v1 + default: https://api.minimax.io/v1 - name: timeout label: en_US: Timeout diff --git a/tests/unit_tests/platform/test_line_eba_adapter.py b/tests/unit_tests/platform/test_line_eba_adapter.py new file mode 100644 index 000000000..306a09176 --- /dev/null +++ b/tests/unit_tests/platform/test_line_eba_adapter.py @@ -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() diff --git a/tests/unit_tests/provider/test_minimax_protocol.py b/tests/unit_tests/provider/test_minimax_protocol.py new file mode 100644 index 000000000..89b21781d --- /dev/null +++ b/tests/unit_tests/provider/test_minimax_protocol.py @@ -0,0 +1,273 @@ +import json +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import httpx +import pytest +import yaml +from langbot_plugin.api.entities.builtin.provider.message import Message + +from langbot.pkg.provider.modelmgr import errors +from langbot.pkg.provider.modelmgr.requesters import litellmchat, minimax + + +@pytest.fixture +def provider(): + manifest = yaml.safe_load(Path(litellmchat.__file__).with_name('minimaxchatcmpl.yaml').read_text()) + settings = {option['name']: option['default'] for option in manifest['spec']['config']} + settings.update(custom_llm_provider='openai', requester_name=manifest['metadata']['name']) + tool = {'type': 'function', 'function': {'name': 'lookup', 'parameters': {'type': 'object'}}} + app = SimpleNamespace(tool_mgr=SimpleNamespace(generate_tools_for_openai=AsyncMock(return_value=[tool]))) + return litellmchat.LiteLLMRequester(app, settings) + + +def runtime(provider, name='MiniMax-M3.1-Flash', level='provider_default'): + return SimpleNamespace( + provider=SimpleNamespace( + provider_entity=SimpleNamespace(requester=provider.requester_cfg['requester_name']), + token_mgr=SimpleNamespace(get_token=lambda: 'synthetic-minimax-test'), + ), + model_entity=SimpleNamespace( + name=name, abilities=['reasoning'], reasoning_config={'level': level}, extra_args={} + ), + ) + + +def completion(message=None): + return { + 'id': 'protocol-test', + 'created': 1, + 'model': 'MiniMax-M3.1-Flash', + 'object': 'chat.completion', + 'choices': [ + {'index': 0, 'message': message or {'role': 'assistant', 'content': 'ok'}, 'finish_reason': 'stop'} + ], + } + + +@pytest.fixture +def wire(monkeypatch): + captured = [] + queued = [] + + async def send(client, request, **kwargs): + captured.append((request, json.loads(request.content) if request.content else None)) + response = queued.pop(0) if queued else httpx.Response(200, json=completion()) + response.request = request + return response + + monkeypatch.setattr(httpx.AsyncClient, 'send', send) + return captured, queued + + +@pytest.mark.asyncio +@pytest.mark.parametrize('name', [*minimax.RECOMMENDED_MODELS, 'MiniMax-M2.7', 'private-deployment']) +@pytest.mark.parametrize( + 'base_url', [None, 'https://api.minimax.chat/v1', 'https://api.minimaxi.com/v1', 'https://proxy.test/v1'] +) +async def test_saved_endpoint_and_model_are_not_migrated(provider, wire, name, base_url): + if base_url: + provider.requester_cfg['base_url'] = base_url + await provider.invoke_llm(None, runtime(provider, name), [Message(role='user', content='hello')]) + request, body = wire[0][-1] + assert str(request.url) == f'{base_url or "https://api.minimax.io/v1"}/chat/completions' + assert body['model'] == name + assert body['messages'] == [{'role': 'user', 'content': 'hello'}] + assert not {'thinking', 'reasoning_effort', 'reasoning_split', 'service_tier'}.intersection(body) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('level', ['enabled', 'low', 'medium', 'high', 'xhigh', 'max']) +async def test_flash_effort_is_serialized_without_changing_input(provider, wire, level): + content = [ + {'type': 'text', 'text': 'inspect'}, + {'type': 'image_url', 'image_url': {'url': 'https://image.test/a.png'}}, + ] + await provider.invoke_llm(None, runtime(provider, level=level), [Message(role='user', content=content)]) + body = wire[0][-1][1] + assert body['messages'][0]['content'] == content + if level == 'enabled': + assert body['thinking'] == {'type': 'adaptive'} + assert 'reasoning_effort' not in body + else: + assert body['reasoning_effort'] == level + assert 'thinking' not in body + + +@pytest.mark.asyncio +@pytest.mark.parametrize('level,expected', [('enabled', 'adaptive'), ('disabled', 'disabled')]) +async def test_m3_thinking_switch(provider, wire, level, expected): + await provider.invoke_llm(None, runtime(provider, 'MiniMax-M3', level), []) + assert wire[0][-1][1]['thinking'] == {'type': expected} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'name,level', [('MiniMax-M3.1-Flash', 'disabled'), ('MiniMax-M3', 'max'), ('MiniMax-M2.7', 'disabled')] +) +async def test_unsupported_thinking_rejected_before_http(provider, wire, name, level): + with pytest.raises(errors.RequesterError): + await provider.invoke_llm(None, runtime(provider, name, level), []) + assert wire[0] == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize('status', [200, 404]) +async def test_direct_discovery_fallback_preserves_custom_and_nontext_models(provider, wire, status): + names = ['MiniMax-M2.7', 'MiniMax-01', 'abab6.5-chat', 'speech-02-hd', 'embo-01', 'private-deployment'] + wire[1].append(httpx.Response(status, json={'data': [{'id': name} for name in names]})) + models = (await provider.scan_models('synthetic-minimax-test'))['models'] + assert [model['id'] for model in models[:2]] == list(minimax.RECOMMENDED_MODELS) + assert all(model['context_length'] == 1_000_000 for model in models[:2]) + assert all(set(model['abilities']) == {'func_call', 'vision', 'reasoning'} for model in models[:2]) + assert {model['id'] for model in models[2:]} == (set(names[3:]) if status == 200 else set()) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('status', [401, 403, 429, 500]) +async def test_catalog_errors_are_not_hidden_by_recommendations(provider, wire, status): + wire[1].append(httpx.Response(status, json={'error': 'fixture'})) + with pytest.raises(errors.RequesterError, match=str(status)): + await provider.scan_models() + + +@pytest.mark.asyncio +async def test_gateway_keeps_all_advertised_current_variants_and_metadata(provider, wire): + provider.requester_cfg.update(requester_name='openrouter-chat-completions', base_url='https://openrouter.ai/api/v1') + names = [ + 'other/minimax-m2', + 'minimax/minimax-m3', + 'minimax/minimax-m3.1-flash', + 'minimax/minimax-m3.1-flash:free', + 'minimax/minimax-m4', + 'minimax/minimax-m2.7', + 'minimax/speech-02-hd', + ] + wire[1].append(httpx.Response(200, json={'data': [{'id': name, 'context_length': 12345} for name in names]})) + models = (await provider.scan_models())['models'] + assert [model['id'] for model in models[:3]] == [names[2], names[3], names[1]] + assert {model['id'] for model in models} == set(names) - {names[5]} + assert all(model.get('context_length') == 12345 for model in models if model['type'] == 'llm') + wire[1].append(httpx.Response(200, json={'data': []})) + assert await provider.scan_models() == {'models': []} + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +@pytest.mark.parametrize('hide_thinking', [False, True]) +@pytest.mark.parametrize('name,level', [('MiniMax-M3.1-Flash', 'provider_default'), ('MiniMax-M3', 'disabled')]) +async def test_tool_followup_replays_original_assistant_not_rendered_text( + provider, wire, stream, hide_thinking, name, level +): + tool_call = {'id': 'lookup-1', 'type': 'function', 'function': {'name': 'lookup', 'arguments': '{}'}} + original = { + 'role': 'assistant', + 'content': 'privateanswer', + 'reasoning_content': 'plan', + 'tool_calls': [tool_call], + } + model = runtime(provider, name, level) + if stream: + deltas = [ + {'role': 'assistant', 'reasoning_content': 'pl'}, + {'reasoning_content': 'an'}, + {'content': 'private'}, + {'content': 'answer'}, + {'tool_calls': [{'index': 0, **tool_call}]}, + ] + events = [] + for index, delta in enumerate([*deltas, {}]): + event = completion() + event.update(object='chat.completion.chunk') + event['choices'] = [ + {'index': 0, 'delta': delta, 'finish_reason': 'tool_calls' if index == len(deltas) else None} + ] + events.append('data: ' + json.dumps(event) + '\n\n') + wire[1].append( + httpx.Response( + 200, content=''.join(events) + 'data: [DONE]\n\n', headers={'content-type': 'text/event-stream'} + ) + ) + chunks = [chunk async for chunk in provider.invoke_llm_stream(None, model, [], remove_think=hide_thinking)] + fields = {} + calls = [] + for chunk in chunks: + for key, value in (chunk.provider_specific_fields or {}).items(): + fields[key] = fields.get(key, '') + value if key == 'reasoning_content' else value + if chunk.tool_calls: + calls = chunk.tool_calls + response = Message( + role='assistant', + content=''.join(chunk.content or '' for chunk in chunks), + provider_specific_fields=fields, + tool_calls=calls, + ) + else: + wire[1].append(httpx.Response(200, json=completion(original))) + response, _ = await provider.invoke_llm(None, model, [], remove_think=hide_thinking) + if hide_thinking: + assert response.content == 'answer' + assert response.provider_specific_fields[minimax.RAW_CONTENT_FIELD] == original['content'] + saved = Message.model_validate_json(response.model_dump_json()) + await provider.invoke_llm( + None, model, [saved, Message(role='tool', tool_call_id='lookup-1', content='result')], funcs=[object()] + ) + followup = wire[0][-1][1] + assert {key: followup['messages'][0][key] for key in original} == original + assert minimax.RAW_CONTENT_FIELD not in json.dumps(followup) + assert followup['messages'][1]['tool_call_id'] == 'lookup-1' + assert followup['tools'][0]['function']['name'] == 'lookup' + + +def test_private_replay_field_does_not_leak_when_switching_provider(provider): + message = Message( + role='assistant', content='visible', provider_specific_fields={minimax.RAW_CONTENT_FIELD: 'private'} + ) + assert provider._convert_messages([message], reasoning_family='openai') == [ + {'role': 'assistant', 'content': 'visible'} + ] + assert message.provider_specific_fields == {minimax.RAW_CONTENT_FIELD: 'private'} + + +def test_direct_catalog_does_not_overwrite_advertised_context_or_drop_nontext(): + advertised = [ + {'id': 'MiniMax-M3', 'name': 'MiniMax-M3', 'type': 'llm', 'context_length': 54321}, + {'id': 'minimax-m2-embedding', 'name': 'embedding', 'type': 'text-embedding'}, + ] + models = minimax.arrange_catalog(advertised, direct=True) + assert len(models) == 3 + assert models[1]['context_length'] == 54321 + assert models[2] == advertised[1] + assert 'abilities' not in advertised[0] + + +@pytest.mark.asyncio +@pytest.mark.parametrize('requester_name', ['openai-chat-completions', 'openrouter-chat-completions']) +async def test_other_providers_do_not_get_minimax_404_fallback(provider, wire, requester_name): + provider.requester_cfg['requester_name'] = requester_name + wire[1].append(httpx.Response(404, json={})) + with pytest.raises(errors.RequesterError, match='404'): + await provider.scan_models() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('content', ['hidden', 'hiddenanswer']) +async def test_stream_without_finish_reason_flushes_one_complete_history_snapshot(provider, wire, content): + events = [] + for fragment in [content[:5], content[5:]]: + event = completion() + event['object'] = 'chat.completion.chunk' + event['choices'] = [{'index': 0, 'delta': {'content': fragment}, 'finish_reason': None}] + events.append('data: ' + json.dumps(event) + '\n\n') + wire[1].append( + httpx.Response(200, content=''.join(events) + 'data: [DONE]\n\n', headers={'content-type': 'text/event-stream'}) + ) + chunks = [chunk async for chunk in provider.invoke_llm_stream(None, runtime(provider), [], remove_think=True)] + snapshots = [ + chunk.provider_specific_fields[minimax.RAW_CONTENT_FIELD] + for chunk in chunks + if minimax.RAW_CONTENT_FIELD in (chunk.provider_specific_fields or {}) + ] + assert snapshots == [content]