refactor(channels): make built-in channels self-contained (#4908)

* refactor(channels): own setup and instance contracts

* refactor(channels): isolate management contracts

* refactor(channels): normalize activation contracts

* fix(channels): enforce management contracts

* refactor(channels): finish setup ownership migration

* fix(channels): harden management contracts

* fix(channels): enforce lazy loading and runtime ownership

* fix(feishu): make multi-instance startup idempotent

* fix(webui): render channel setup contracts cleanly

* fix(feishu): stop websocket clients cleanly

* fix(channels): enforce persistence and activation gates

* fix(channels): preserve global feature action scope

* fix(channels): apply defaults for single plugins

* fix(channels): enforce management contract boundaries

* refactor(feishu): remove identity helper indirection

* fix(channels): preserve management setup contracts

* refactor(channels): generalize instance settings UI

* refactor(channels): package channel plugins with web UI metadata

* refactor(channels): make built-ins self-contained packages

* test(channels): colocate tests with channel packages

* fix(dingtalk): use official brand icon

* feat(channels): colocate webui translations

* docs(channels): clarify plugin ownership

* test(exec): remove output wait race

* refactor(channels): unify plugin descriptors

* fix(channels): enforce descriptor-owned contracts

* refactor(channels): finish package-owned plugin setup

* refactor(channels): use repository-owned packages only

* fix(channels): self-describe dependencies and runtime state

* fix(channels): warn about legacy entry points
This commit is contained in:
chengyongru
2026-07-19 23:30:49 +08:00
committed by GitHub
parent 7aaac37bca
commit 462a0dfb0f
388 changed files with 17093 additions and 5110 deletions
+1
View File
@@ -0,0 +1 @@
"""Feishu/Lark channel package."""
+36
View File
@@ -0,0 +1,36 @@
"""Dependency-free Feishu configuration model shared by management and runtime."""
from typing import Literal
from pydantic import Field
from nanobot.config.schema import Base
class FeishuConfig(Base):
"""Feishu/Lark channel configuration using WebSocket long connection."""
instance_id: str = "default"
name: str = "nanobot"
identity_key: str = ""
enabled: bool = False
app_id: str = ""
app_secret: str = ""
encrypt_key: str = ""
verification_token: str = ""
allow_from: list[str] = Field(default_factory=list)
react_emoji: str = "THUMBSUP"
done_emoji: str | None = None
tool_hint_prefix: str = "\U0001f527"
group_policy: Literal["open", "mention"] = "mention"
reply_to_message: bool = False
streaming: bool = True
domain: Literal["feishu", "lark"] = "feishu"
topic_isolation: bool = True
def feishu_default_config() -> dict[str, object]:
return FeishuConfig().model_dump(by_alias=True)
__all__ = ["FeishuConfig", "feishu_default_config"]
+216
View File
@@ -0,0 +1,216 @@
"""Short-lived WebUI channel connection sessions."""
from __future__ import annotations
import asyncio
import json
import secrets
import time
from dataclasses import dataclass
from typing import Any
import httpx
from nanobot.channels.connect import ChannelConnectError, QueryParams, query_first
from nanobot.channels.feishu import runtime as feishu
from nanobot.channels.feishu.instances import DEFAULT_INSTANCE_ID, validate_instance_id
@dataclass(slots=True)
class FeishuConnectSession:
id: str
instance_id: str
instance_name: str
device_code: str
qr_url: str
domain: str
interval: int
expire_in: int
created_wall: float
deadline: float
last_error: str | None = None
class FeishuConnectStore:
"""In-memory Feishu/Lark QR connection state.
Sessions intentionally live only in the gateway process and expire quickly.
The app secret is never returned to the browser; it is saved directly to
config when Feishu/Lark completes authorization.
"""
def __init__(self) -> None:
self._sessions: dict[str, FeishuConnectSession] = {}
async def handle(self, action: str, query: QueryParams) -> dict[str, Any]:
"""Handle one generic settings connection action."""
if action == "start":
return await asyncio.to_thread(
self.start,
domain=(query_first(query, "domain") or "feishu").strip(),
instance_id=(query_first(query, "instance_id") or "default").strip(),
mode=(query_first(query, "mode") or "replace").strip(),
)
session_id = (query_first(query, "session_id") or "").strip()
if not session_id:
raise ChannelConnectError("missing Feishu connect session")
if action == "poll":
return await asyncio.to_thread(self.poll, session_id)
if action == "cancel":
return self.cancel(session_id)
raise ChannelConnectError(f"unsupported Feishu connect action: {action}", status=404)
def start(
self,
*,
domain: str = "feishu",
instance_id: str = DEFAULT_INSTANCE_ID,
mode: str = "replace",
) -> dict[str, Any]:
domain = _normalize_domain(domain)
instance_id = _resolve_instance_id(instance_id, mode)
self._cleanup()
try:
feishu._init_registration(domain)
begin = feishu._begin_registration(domain)
except (RuntimeError, OSError, json.JSONDecodeError, httpx.HTTPError) as exc:
raise ChannelConnectError(
f"Unable to start Feishu/Lark connection: {exc}",
status=502,
) from exc
session_id = secrets.token_urlsafe(18)
now_wall = time.time()
now = time.monotonic()
expire_in = int(begin["expire_in"])
interval = max(2, int(begin["interval"]))
session = FeishuConnectSession(
id=session_id,
instance_id=instance_id,
instance_name=_default_instance_name(instance_id),
device_code=str(begin["device_code"]),
qr_url=str(begin["qr_url"]),
domain=domain,
interval=interval,
expire_in=expire_in,
created_wall=now_wall,
deadline=now + expire_in,
)
self._sessions[session_id] = session
return _start_payload(session)
def poll(self, session_id: str) -> dict[str, Any]:
self._cleanup()
session = self._sessions.get(session_id)
if session is None:
return {
"session_id": session_id,
"status": "expired",
"message": "This Feishu connection has expired. Start again.",
}
if time.monotonic() >= session.deadline:
self._sessions.pop(session_id, None)
return {
"session_id": session_id,
"status": "expired",
"message": "This Feishu connection has expired. Start again.",
}
try:
result = feishu.poll_registration_once(
device_code=session.device_code,
domain=session.domain,
)
except (RuntimeError, OSError, json.JSONDecodeError, httpx.HTTPError) as exc:
session.last_error = str(exc)
return _pending_payload(session)
session.domain = str(result.get("domain") or session.domain)
status = result.get("status")
if status == "succeeded":
session.instance_id = feishu.save_registration_result(
result,
instance_id=session.instance_id,
name=session.instance_name,
)
self._sessions.pop(session_id, None)
return {
"session_id": session_id,
"instance_id": session.instance_id,
"status": "succeeded",
"message": "Feishu is connected.",
"domain": session.domain,
"app_id": result.get("app_id"),
}
if status == "failed":
self._sessions.pop(session_id, None)
return {
"session_id": session_id,
"instance_id": session.instance_id,
"status": "failed",
"message": "Authorization was cancelled or expired.",
"domain": session.domain,
}
return _pending_payload(session)
def cancel(self, session_id: str) -> dict[str, Any]:
session = self._sessions.pop(session_id, None)
return {
"session_id": session_id,
"instance_id": session.instance_id if session else DEFAULT_INSTANCE_ID,
"status": "cancelled",
"message": "Feishu connection cancelled.",
}
def _cleanup(self) -> None:
now = time.monotonic()
expired = [session_id for session_id, session in self._sessions.items() if now >= session.deadline]
for session_id in expired:
self._sessions.pop(session_id, None)
def _normalize_domain(domain: str) -> str:
normalized = domain.strip().lower()
return normalized if normalized in {"feishu", "lark"} else "feishu"
def _resolve_instance_id(instance_id: str, mode: str) -> str:
if mode == "create":
return f"assistant-{secrets.token_hex(3)}"
try:
return validate_instance_id(instance_id or DEFAULT_INSTANCE_ID)
except ValueError as exc:
raise ChannelConnectError(str(exc), status=400) from exc
def _default_instance_name(instance_id: str) -> str:
return "nanobot" if instance_id == DEFAULT_INSTANCE_ID else f"nanobot {instance_id}"
def _start_payload(session: FeishuConnectSession) -> dict[str, Any]:
return {
"session_id": session.id,
"instance_id": session.instance_id,
"status": "pending",
"qr_url": session.qr_url,
"domain": session.domain,
"interval_ms": session.interval * 1000,
"expires_at_ms": int((session.created_wall + session.expire_in) * 1000),
"message": "Scan with Feishu or Lark to connect.",
}
def _pending_payload(session: FeishuConnectSession) -> dict[str, Any]:
return {
"session_id": session.id,
"instance_id": session.instance_id,
"status": "pending",
"domain": session.domain,
"interval_ms": session.interval * 1000,
"expires_at_ms": int((session.created_wall + session.expire_in) * 1000),
"message": "Waiting for authorization.",
}
+267
View File
@@ -0,0 +1,267 @@
"""Feishu-owned helpers for its persisted multi-instance configuration."""
from __future__ import annotations
import re
from typing import Any
from loguru import logger
from nanobot.channels.contracts import ChannelInstanceSpec, ChannelManagementSpec
from nanobot.channels.feishu.config import feishu_default_config
from nanobot.config.loader import merge_missing_defaults
DEFAULT_INSTANCE_ID = "default"
_INSTANCE_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$")
def validate_instance_id(value: str) -> str:
"""Return a normalized instance id or raise ValueError."""
instance_id = value.strip()
if not instance_id or not _INSTANCE_ID_RE.fullmatch(instance_id):
raise ValueError("instance id must match [A-Za-z0-9_-]+")
return instance_id
def runtime_channel_name(base_name: str, instance_id: str) -> str:
"""Return the channel key used for routing messages at runtime."""
return base_name if instance_id == DEFAULT_INSTANCE_ID else f"{base_name}.{instance_id}"
def managed_feishu_instance_specs(
section: Any,
*,
enabled_only: bool = True,
) -> list[ChannelInstanceSpec]:
return feishu_instance_specs(
section,
feishu_default_config(),
enabled_only=enabled_only,
)
def update_managed_feishu_instance(
section: Any,
values: dict[str, Any],
*,
instance_id: str = DEFAULT_INSTANCE_ID,
) -> dict[str, Any]:
existing = section if isinstance(section, dict) else {}
return upsert_feishu_instance(
existing,
feishu_default_config(),
instance_id,
values,
)
def _base_feishu_instance_config(defaults: dict[str, Any]) -> dict[str, Any]:
config = dict(defaults)
config["instanceId"] = DEFAULT_INSTANCE_ID
config["name"] = "nanobot"
return config
def _normalize_feishu_instance(
raw: dict[str, Any],
defaults: dict[str, Any],
*,
inherited: dict[str, Any] | None = None,
fallback_id: str = DEFAULT_INSTANCE_ID,
) -> dict[str, Any]:
config = merge_missing_defaults(inherited or {}, defaults)
config = merge_missing_defaults(raw, config)
raw_id = raw.get("id") or raw.get("instanceId") or raw.get("instance_id") or fallback_id
instance_id = validate_instance_id(str(raw_id))
config["id"] = instance_id
config["instanceId"] = instance_id
config.setdefault("name", "nanobot" if instance_id == DEFAULT_INSTANCE_ID else f"nanobot {instance_id}")
return config
def feishu_app_identity_key(app_id: Any, domain: Any = "feishu") -> str:
"""Return the stable identity shared by persisted and runtime instances."""
app_id = str(app_id or "").strip()
if not app_id:
return ""
normalized_domain = "lark" if str(domain or "feishu").strip().lower() == "lark" else "feishu"
return f"{normalized_domain}:{app_id}"
def _feishu_instance_inputs(
section: Any,
defaults: dict[str, Any],
) -> tuple[list[Any], dict[str, Any] | None]:
if hasattr(section, "model_dump"):
section = section.model_dump(mode="json", by_alias=True)
if not isinstance(section, dict):
section = {}
instances = section.get("instances")
if isinstance(instances, list):
inherited = {key: value for key, value in section.items() if key != "instances"}
return list(instances), inherited
return ([section] if section else [_base_feishu_instance_config(defaults)]), None
def feishu_instance_specs(
section: Any,
defaults: dict[str, Any],
*,
enabled_only: bool = False,
) -> list[ChannelInstanceSpec]:
"""Expand legacy or canonical Feishu config into runtime instance specs."""
raw_specs, inherited = _feishu_instance_inputs(section, defaults)
specs: list[ChannelInstanceSpec] = []
instance_ids: set[str] = set()
identity_owners: dict[str, str] = {}
for index, raw in enumerate(raw_specs):
if not isinstance(raw, dict):
logger.warning("Skipping invalid Feishu instance at index {}: expected an object", index)
continue
fallback_id = DEFAULT_INSTANCE_ID if index == 0 else f"assistant-{index + 1}"
try:
config = _normalize_feishu_instance(
raw,
defaults,
inherited=inherited,
fallback_id=fallback_id,
)
except ValueError as exc:
logger.warning("Skipping invalid Feishu instance config: {}", exc)
continue
instance_id = str(config["instanceId"])
if instance_id in instance_ids:
logger.warning("Skipping duplicate Feishu instance id '{}'", instance_id)
continue
instance_ids.add(instance_id)
enabled = bool(config.get("enabled", defaults.get("enabled", False)))
if enabled_only and not enabled:
continue
identity = feishu_app_identity_key(
config.get("appId") or config.get("app_id"),
config.get("domain"),
)
if enabled_only and identity:
if identity in identity_owners:
logger.warning(
"Skipping Feishu instance '{}' because it uses the same app as instance '{}'",
instance_id,
identity_owners[identity],
)
continue
identity_owners[identity] = instance_id
specs.append(
ChannelInstanceSpec(
instance_id=instance_id,
config=config,
)
)
return specs
def canonical_feishu_section(section: Any, defaults: dict[str, Any]) -> dict[str, Any]:
"""Return a canonical section, rejecting input that cannot be preserved safely."""
raw_specs, inherited = _feishu_instance_inputs(section, defaults)
instances: list[dict[str, Any]] = []
instance_ids: set[str] = set()
for index, raw in enumerate(raw_specs):
if not isinstance(raw, dict):
raise ValueError(f"Feishu instance at index {index} must be an object")
fallback_id = DEFAULT_INSTANCE_ID if index == 0 else f"assistant-{index + 1}"
try:
config = _normalize_feishu_instance(
raw,
defaults,
inherited=inherited,
fallback_id=fallback_id,
)
except ValueError as exc:
raise ValueError(f"Invalid Feishu instance at index {index}: {exc}") from exc
instance_id = str(config["instanceId"])
if instance_id in instance_ids:
raise ValueError(f"duplicate Feishu instance id '{instance_id}'")
instance_ids.add(instance_id)
instances.append(config)
return {"instances": instances}
def upsert_feishu_instance(
section: Any,
defaults: dict[str, Any],
instance_id: str,
values: dict[str, Any],
) -> dict[str, Any]:
"""Return canonical Feishu section with one instance created or updated."""
instance_id = validate_instance_id(instance_id)
canonical = canonical_feishu_section(section, defaults)
instances = canonical.setdefault("instances", [])
for instance in instances:
if instance.get("id") == instance_id or instance.get("instanceId") == instance_id:
instance.update(values)
instance["id"] = instance_id
instance["instanceId"] = instance_id
instance.setdefault("name", "nanobot" if instance_id == DEFAULT_INSTANCE_ID else f"nanobot {instance_id}")
return canonical
config = _normalize_feishu_instance(
{**values, "id": instance_id},
defaults,
fallback_id=instance_id,
)
instances.append(config)
return canonical
def update_feishu_instance_preserving_shape(
section: Any,
defaults: dict[str, Any],
instance_id: str,
values: dict[str, Any],
) -> dict[str, Any]:
"""Update background metadata without migrating a legacy flat section."""
instance_id = validate_instance_id(instance_id)
if hasattr(section, "model_dump"):
section = section.model_dump(mode="json", by_alias=True)
if (
instance_id == DEFAULT_INSTANCE_ID
and isinstance(section, dict)
and not isinstance(section.get("instances"), list)
):
return {**section, **values}
return upsert_feishu_instance(section, defaults, instance_id, values)
FEISHU_MANAGEMENT = ChannelManagementSpec(
multi_instance=True,
default_config=feishu_default_config,
instance_specs=managed_feishu_instance_specs,
update_instance_config=update_managed_feishu_instance,
runtime_name=runtime_channel_name,
)
__all__ = [
"DEFAULT_INSTANCE_ID",
"FEISHU_MANAGEMENT",
"canonical_feishu_section",
"feishu_app_identity_key",
"feishu_instance_specs",
"runtime_channel_name",
"update_feishu_instance_preserving_shape",
"upsert_feishu_instance",
"validate_instance_id",
]
+42
View File
@@ -0,0 +1,42 @@
"""Dependency-free Feishu/Lark management contract."""
from nanobot.channels._manifest import DIRECT_GROUP_POLICIES, field, required_fields
from nanobot.channels.contracts import ChannelSetupSpec
from nanobot.channels.feishu.instances import FEISHU_MANAGEMENT
from nanobot.channels.feishu.validation import validate
from nanobot.channels.plugin import ChannelPlugin
SETUP_SPEC = ChannelSetupSpec(
fields={
"appId": field(snapshot=False),
"appSecret": field("secret", snapshot=False),
"domain": field(
"enum",
choices={"feishu", "lark"},
default="feishu",
snapshot=False,
),
"groupPolicy": field(
"enum",
choices=DIRECT_GROUP_POLICIES,
default="mention",
snapshot=False,
),
"allowFrom": field("list", snapshot=False),
"topicIsolation": field("bool", default=True, snapshot=False),
},
required=required_fields("appId", "appSecret"),
official_url="https://open.feishu.cn/app",
validator=validate,
)
PLUGIN = ChannelPlugin(
name="feishu",
display_name="Feishu",
runtime=f"{__package__}.runtime:FeishuChannel",
connector=f"{__package__}.connect:FeishuConnectStore",
setup=SETUP_SPEC,
management=FEISHU_MANAGEMENT,
dependencies=("lark-oapi>=1.5.0,<2.0.0",),
webui="webui/index.tsx",
)
File diff suppressed because it is too large Load Diff
@@ -0,0 +1 @@
"""Tests for the Feishu channel package."""
@@ -0,0 +1,39 @@
import json
from nanobot.channels.feishu.runtime import _extract_share_card_content
def test_extract_interactive_card_reads_user_dsl_body_elements() -> None:
content = {
"user_dsl": json.dumps(
{
"schema": "2.0",
"body": {"elements": [{"tag": "markdown", "content": "**hello**"}]},
}
)
}
assert _extract_share_card_content(content, "interactive") == "**hello**"
def test_extract_interactive_card_reads_nested_text_elements() -> None:
content = {"elements": [[{"tag": "text", "text": "hello"}]]}
assert _extract_share_card_content(content, "interactive") == "hello"
def test_extract_interactive_card_reads_table_rows() -> None:
content = {
"elements": [
{
"tag": "table",
"columns": [
{"name": "c0", "display_name": "Name"},
{"name": "c1", "display_name": "Score"},
],
"rows": [{"c0": "Alice", "c1": 98}],
}
]
}
assert _extract_share_card_content(content, "interactive") == "Name | Score\nAlice | 98"
@@ -0,0 +1,46 @@
"""Tests for Feishu/Lark domain configuration."""
from unittest.mock import MagicMock
from nanobot.bus.queue import MessageBus
from nanobot.channels.feishu.runtime import FeishuChannel, FeishuConfig
def _make_channel(domain: str = "feishu") -> FeishuChannel:
config = FeishuConfig(
enabled=True,
app_id="cli_test",
app_secret="secret",
allow_from=["*"],
domain=domain,
)
ch = FeishuChannel(config, MessageBus())
ch._client = MagicMock()
ch._loop = None
return ch
class TestFeishuConfigDomain:
def test_domain_default_is_feishu(self):
config = FeishuConfig()
assert config.domain == "feishu"
def test_domain_accepts_lark(self):
config = FeishuConfig(domain="lark")
assert config.domain == "lark"
def test_domain_accepts_feishu(self):
config = FeishuConfig(domain="feishu")
assert config.domain == "feishu"
def test_default_config_includes_domain(self):
default_cfg = FeishuChannel.default_config()
assert "domain" in default_cfg
assert default_cfg["domain"] == "feishu"
def test_channel_persists_domain_from_config(self):
ch = _make_channel(domain="lark")
assert ch.config.domain == "lark"
def test_channel_persists_feishu_domain_from_config(self):
ch = _make_channel(domain="feishu")
assert ch.config.domain == "feishu"
@@ -0,0 +1,99 @@
import subprocess
import sys
def _run_import_probe(source: str) -> str:
proc = subprocess.run(
[sys.executable, "-c", source],
check=True,
capture_output=True,
text=True,
)
return proc.stdout.strip()
def test_feishu_module_import_does_not_import_lark_oapi():
out = _run_import_probe(
"import sys; import nanobot.channels.feishu; print('lark_oapi' in sys.modules)"
)
assert out == "False"
def test_feishu_channel_constructor_does_not_import_lark_oapi():
out = _run_import_probe(
"import sys; "
"from nanobot.bus.queue import MessageBus; "
"from nanobot.channels.feishu.runtime import FeishuChannel; "
"FeishuChannel({'enabled': True}, MessageBus()); "
"print('lark_oapi' in sys.modules)"
)
assert out == "False"
def test_lark_runtime_thread_import_clears_sdk_import_loop():
out = _run_import_probe(
"import asyncio\n"
"import sys\n"
"import tempfile\n"
"from pathlib import Path\n"
"from nanobot.channels.feishu.runtime import _load_lark_runtime\n"
"root = Path(tempfile.mkdtemp())\n"
"pkg = root / 'lark_oapi'\n"
"(pkg / 'ws').mkdir(parents=True)\n"
"(pkg / 'core').mkdir(parents=True)\n"
"(pkg / '__init__.py').write_text('class LogLevel:\\n INFO = 20\\n')\n"
"(pkg / 'ws' / '__init__.py').write_text('')\n"
"(pkg / 'ws' / 'client.py').write_text('import asyncio\\nloop = asyncio.new_event_loop()\\n')\n"
"(pkg / 'core' / '__init__.py').write_text('')\n"
"(pkg / 'core' / 'const.py').write_text(\"FEISHU_DOMAIN = 'feishu'\\nLARK_DOMAIN = 'lark'\\n\")\n"
"sys.path.insert(0, str(root))\n"
"async def main():\n"
" await asyncio.to_thread(_load_lark_runtime)\n"
" import lark_oapi.ws.client as ws\n"
" print(getattr(ws, 'loop', 'sentinel') is None)\n"
"asyncio.run(main())"
)
assert out == "True"
def test_lark_runtime_thread_import_is_serialized_for_multiple_instances():
out = _run_import_probe(
"import asyncio\n"
"import sys\n"
"import tempfile\n"
"from pathlib import Path\n"
"from nanobot.channels.feishu.runtime import _load_lark_runtime\n"
"root = Path(tempfile.mkdtemp())\n"
"pkg = root / 'lark_oapi'\n"
"(pkg / 'ws').mkdir(parents=True)\n"
"(pkg / 'core').mkdir(parents=True)\n"
"(pkg / '__init__.py').write_text('class LogLevel:\\n INFO = 20\\n')\n"
"(pkg / 'ws' / '__init__.py').write_text('')\n"
"(pkg / 'ws' / 'client.py').write_text(\n"
" 'import time\\n'\n"
" 'class ImportLoop:\\n'\n"
" ' closed = False\\n'\n"
" ' close_calls = 0\\n'\n"
" ' def is_running(self): return False\\n'\n"
" ' def is_closed(self): return self.closed\\n'\n"
" ' def close(self):\\n'\n"
" ' self.close_calls += 1\\n'\n"
" ' time.sleep(0.05)\\n'\n"
" ' if self.close_calls > 1: raise AttributeError(\"closed twice\")\\n'\n"
" ' self.closed = True\\n'\n"
" 'loop = ImportLoop()\\n'\n"
")\n"
"(pkg / 'core' / '__init__.py').write_text('')\n"
"(pkg / 'core' / 'const.py').write_text(\"FEISHU_DOMAIN = 'feishu'\\nLARK_DOMAIN = 'lark'\\n\")\n"
"sys.path.insert(0, str(root))\n"
"async def main():\n"
" await asyncio.gather(*[asyncio.to_thread(_load_lark_runtime) for _ in range(8)])\n"
" import lark_oapi.ws.client as ws\n"
" print(ws.loop is None)\n"
"asyncio.run(main())"
)
assert out == "True"
@@ -0,0 +1,448 @@
import json
import httpx
import pytest
from nanobot.channels.feishu import runtime as feishu_module
from nanobot.channels.feishu.runtime import FeishuChannel
from nanobot.config import loader
from nanobot.config.schema import Config
from nanobot.pairing import store as pairing_store
def _default_feishu_instance(data: dict) -> dict:
return data["channels"]["feishu"]["instances"][0]
@pytest.mark.asyncio
async def test_feishu_login_writes_credentials_to_active_config(monkeypatch, tmp_path):
config_path = tmp_path / "config.json"
config = Config()
config.channels.feishu = {"enabled": False, "domain": "feishu"}
loader.save_config(config, config_path)
monkeypatch.setattr(loader, "_current_config_path", config_path)
monkeypatch.setattr(
feishu_module,
"qr_register",
lambda initial_domain="feishu": {
"app_id": "cli_app",
"app_secret": "secret",
"domain": "lark",
},
)
monkeypatch.setattr(
feishu_module,
"fetch_feishu_app_identity",
lambda app_id, app_secret, domain: {
"displayName": "Voraflare Bot",
"avatarUrl": "https://example.com/avatar.png",
"identityFetchedAt": "2026-07-06T00:00:00Z",
},
)
channel = FeishuChannel({"enabled": False, "domain": "feishu"}, None)
assert await channel.login() is True
data = json.loads(config_path.read_text(encoding="utf-8"))
instance = _default_feishu_instance(data)
assert instance["id"] == "default"
assert instance["appId"] == "cli_app"
assert instance["appSecret"] == "secret"
assert instance["domain"] == "lark"
assert instance["identityKey"] == "lark:cli_app"
assert instance["enabled"] is True
assert instance["displayName"] == "Voraflare Bot"
assert instance["avatarUrl"] == "https://example.com/avatar.png"
assert instance["identityFetchedAt"] == "2026-07-06T00:00:00Z"
def test_begin_registration_requires_login_url(monkeypatch):
monkeypatch.setattr(
feishu_module,
"_post_registration",
lambda _base_url, _body: {"device_code": "device"},
)
with pytest.raises(RuntimeError, match="login URL"):
feishu_module._begin_registration()
def test_begin_registration_preserves_login_url(monkeypatch):
login_url = "https://accounts.feishu.cn/login?device_code=device"
monkeypatch.setattr(
feishu_module,
"_post_registration",
lambda _base_url, _body: {
"device_code": "device",
"verification_uri_complete": login_url,
},
)
assert feishu_module._begin_registration()["qr_url"] == login_url
def test_qr_register_returns_none_on_network_error(monkeypatch):
def raise_connect_error(_base_url, _body):
raise httpx.ConnectError("network down")
monkeypatch.setattr(feishu_module, "_post_registration", raise_connect_error)
assert feishu_module.qr_register() is None
def test_save_registration_result_keeps_credentials_when_identity_fetch_fails(monkeypatch, tmp_path):
config_path = tmp_path / "config.json"
loader.save_config(Config(), config_path)
monkeypatch.setattr(loader, "_current_config_path", config_path)
def fail_identity(_app_id, _app_secret, _domain):
raise RuntimeError("metadata unavailable")
monkeypatch.setattr(feishu_module, "fetch_feishu_app_identity", fail_identity)
feishu_module.save_registration_result({
"app_id": "cli_app",
"app_secret": "secret",
"domain": "feishu",
})
data = json.loads(config_path.read_text(encoding="utf-8"))
instance = _default_feishu_instance(data)
assert instance["appId"] == "cli_app"
assert instance["appSecret"] == "secret"
assert instance["identityKey"] == "feishu:cli_app"
assert "displayName" not in instance
assert "avatarUrl" not in instance
def test_save_registration_result_reuses_existing_app_instance(monkeypatch, tmp_path):
config_path = tmp_path / "config.json"
config = Config()
config.channels.feishu = {
"instances": [
{
"id": "default",
"instanceId": "default",
"name": "nanobot",
"enabled": True,
"appId": "cli_same",
"appSecret": "old-secret",
"domain": "feishu",
"identityKey": "feishu:cli_same",
"allowFrom": ["approved-user"],
}
]
}
loader.save_config(config, config_path)
monkeypatch.setattr(loader, "_current_config_path", config_path)
monkeypatch.setattr(feishu_module, "fetch_feishu_app_identity", lambda *_args: {})
effective_id = feishu_module.save_registration_result(
{
"app_id": "cli_same",
"app_secret": "rotated-secret",
"domain": "feishu",
},
instance_id="assistant-new",
name="nanobot assistant-new",
)
data = json.loads(config_path.read_text(encoding="utf-8"))
instances = data["channels"]["feishu"]["instances"]
assert effective_id == "default"
assert len(instances) == 1
assert instances[0]["id"] == "default"
assert instances[0]["name"] == "nanobot"
assert instances[0]["appSecret"] == "rotated-secret"
assert instances[0]["allowFrom"] == ["approved-user"]
def test_save_registration_result_resets_access_when_instance_app_changes(
monkeypatch,
tmp_path,
):
config_path = tmp_path / "config.json"
pairing_path = tmp_path / "pairing.json"
config = Config()
config.channels.feishu = {
"instances": [
{
"id": "assistant-test",
"instanceId": "assistant-test",
"name": "old assistant",
"enabled": True,
"appId": "cli_old",
"appSecret": "old-secret",
"identityKey": "feishu:cli_old",
"allowFrom": ["old-open-id"],
"allow_from": ["old-snake-open-id"],
}
]
}
loader.save_config(config, config_path)
monkeypatch.setattr(loader, "_current_config_path", config_path)
monkeypatch.setattr(pairing_store, "_store_path", lambda: pairing_path)
monkeypatch.setattr(feishu_module, "fetch_feishu_app_identity", lambda *_args: {})
approved_code = pairing_store.generate_code("feishu.assistant-test", "paired-user")
pairing_store.approve_code(approved_code)
pending_code = pairing_store.generate_code("feishu.assistant-test", "pending-user")
feishu_module.save_registration_result(
{
"app_id": "cli_new",
"app_secret": "new-secret",
"domain": "feishu",
},
instance_id="assistant-test",
name="new assistant",
)
data = json.loads(config_path.read_text(encoding="utf-8"))
instance = data["channels"]["feishu"]["instances"][0]
assert instance["appId"] == "cli_new"
assert instance["appSecret"] == "new-secret"
assert instance["identityKey"] == "feishu:cli_new"
assert instance["allowFrom"] == []
assert instance["allow_from"] == []
assert pairing_store.is_approved("feishu.assistant-test", "paired-user") is False
assert pairing_store.approve_code(pending_code) is None
def test_save_registration_result_keeps_access_when_only_secret_rotates(
monkeypatch,
tmp_path,
):
config_path = tmp_path / "config.json"
pairing_path = tmp_path / "pairing.json"
config = Config()
config.channels.feishu = {
"instances": [
{
"id": "assistant-test",
"instanceId": "assistant-test",
"name": "same assistant",
"enabled": True,
"appId": "cli_same",
"appSecret": "old-secret",
"domain": "feishu",
"identityKey": "feishu:cli_same",
"allowFrom": ["old-open-id"],
}
]
}
loader.save_config(config, config_path)
monkeypatch.setattr(loader, "_current_config_path", config_path)
monkeypatch.setattr(pairing_store, "_store_path", lambda: pairing_path)
monkeypatch.setattr(feishu_module, "fetch_feishu_app_identity", lambda *_args: {})
approved_code = pairing_store.generate_code("feishu.assistant-test", "paired-user")
pairing_store.approve_code(approved_code)
pending_code = pairing_store.generate_code("feishu.assistant-test", "pending-user")
feishu_module.save_registration_result(
{
"app_id": "cli_same",
"app_secret": "new-secret",
"domain": "feishu",
},
instance_id="assistant-test",
name="same assistant",
)
data = json.loads(config_path.read_text(encoding="utf-8"))
instance = data["channels"]["feishu"]["instances"][0]
assert instance["appSecret"] == "new-secret"
assert instance["identityKey"] == "feishu:cli_same"
assert instance["allowFrom"] == ["old-open-id"]
assert pairing_store.is_approved("feishu.assistant-test", "paired-user") is True
assert pairing_store.approve_code(pending_code) == (
"feishu.assistant-test",
"pending-user",
)
def test_save_registration_result_resets_access_when_domain_changes(
monkeypatch,
tmp_path,
):
config_path = tmp_path / "config.json"
pairing_path = tmp_path / "pairing.json"
config = Config()
config.channels.feishu = {
"instances": [
{
"id": "assistant-test",
"instanceId": "assistant-test",
"name": "lark assistant",
"enabled": True,
"appId": "cli_same",
"appSecret": "old-secret",
"domain": "lark",
"identityKey": "lark:cli_same",
"allowFrom": ["old-open-id"],
}
]
}
loader.save_config(config, config_path)
monkeypatch.setattr(loader, "_current_config_path", config_path)
monkeypatch.setattr(pairing_store, "_store_path", lambda: pairing_path)
monkeypatch.setattr(feishu_module, "fetch_feishu_app_identity", lambda *_args: {})
approved_code = pairing_store.generate_code("feishu.assistant-test", "paired-user")
pairing_store.approve_code(approved_code)
feishu_module.save_registration_result(
{
"app_id": "cli_same",
"app_secret": "new-secret",
"domain": "feishu",
},
instance_id="assistant-test",
name="feishu assistant",
)
data = json.loads(config_path.read_text(encoding="utf-8"))
instance = data["channels"]["feishu"]["instances"][0]
assert instance["domain"] == "feishu"
assert instance["identityKey"] == "feishu:cli_same"
assert instance["allowFrom"] == []
assert pairing_store.is_approved("feishu.assistant-test", "paired-user") is False
def test_sync_saved_identity_boundary_resets_access_after_manual_app_change(
monkeypatch,
tmp_path,
):
config_path = tmp_path / "config.json"
pairing_path = tmp_path / "pairing.json"
config = Config()
config.channels.feishu = {
"instances": [
{
"id": "assistant-test",
"instanceId": "assistant-test",
"name": "manual assistant",
"enabled": True,
"appId": "cli_new",
"appSecret": "secret",
"domain": "feishu",
"identityKey": "feishu:cli_old",
"allowFrom": ["old-open-id"],
"allow_from": ["old-snake-open-id"],
}
]
}
loader.save_config(config, config_path)
monkeypatch.setattr(loader, "_current_config_path", config_path)
monkeypatch.setattr(pairing_store, "_store_path", lambda: pairing_path)
approved_code = pairing_store.generate_code("feishu.assistant-test", "paired-user")
pairing_store.approve_code(approved_code)
pending_code = pairing_store.generate_code("feishu.assistant-test", "pending-user")
assert feishu_module.sync_saved_feishu_identity_boundary(
instance_id="assistant-test",
app_id="cli_new",
domain="feishu",
) is True
data = json.loads(config_path.read_text(encoding="utf-8"))
instance = data["channels"]["feishu"]["instances"][0]
assert instance["identityKey"] == "feishu:cli_new"
assert instance["allowFrom"] == []
assert instance["allow_from"] == []
assert pairing_store.is_approved("feishu.assistant-test", "paired-user") is False
assert pairing_store.approve_code(pending_code) is None
def test_sync_saved_identity_boundary_backfills_marker_without_resetting_access(
monkeypatch,
tmp_path,
):
config_path = tmp_path / "config.json"
pairing_path = tmp_path / "pairing.json"
config = Config()
config.channels.feishu = {
"instances": [
{
"id": "assistant-test",
"instanceId": "assistant-test",
"name": "existing assistant",
"enabled": True,
"appId": "cli_existing",
"appSecret": "secret",
"domain": "feishu",
"allowFrom": ["old-open-id"],
}
]
}
loader.save_config(config, config_path)
monkeypatch.setattr(loader, "_current_config_path", config_path)
monkeypatch.setattr(pairing_store, "_store_path", lambda: pairing_path)
approved_code = pairing_store.generate_code("feishu.assistant-test", "paired-user")
pairing_store.approve_code(approved_code)
assert feishu_module.sync_saved_feishu_identity_boundary(
instance_id="assistant-test",
app_id="cli_existing",
domain="feishu",
) is False
data = json.loads(config_path.read_text(encoding="utf-8"))
instance = data["channels"]["feishu"]["instances"][0]
assert instance["identityKey"] == "feishu:cli_existing"
assert instance["allowFrom"] == ["old-open-id"]
assert pairing_store.is_approved("feishu.assistant-test", "paired-user") is True
def test_sync_saved_identity_boundary_preserves_legacy_flat_config(monkeypatch, tmp_path):
config_path = tmp_path / "config.json"
config = Config()
config.channels.feishu = {
"enabled": True,
"appId": "cli_existing",
"appSecret": "secret",
"domain": "feishu",
"allowFrom": ["old-open-id"],
}
loader.save_config(config, config_path)
monkeypatch.setattr(loader, "_current_config_path", config_path)
assert feishu_module.sync_saved_feishu_identity_boundary(
instance_id="default",
app_id="cli_existing",
domain="feishu",
) is False
saved = json.loads(config_path.read_text(encoding="utf-8"))["channels"]["feishu"]
assert saved["appId"] == "cli_existing"
assert saved["appSecret"] == "secret"
assert saved["identityKey"] == "feishu:cli_existing"
assert saved["allowFrom"] == ["old-open-id"]
assert "instances" not in saved
@pytest.mark.asyncio
async def test_feishu_login_creates_missing_active_config(monkeypatch, tmp_path):
missing_config = tmp_path / "missing.json"
monkeypatch.setattr(loader, "_current_config_path", missing_config)
monkeypatch.setattr(
feishu_module,
"qr_register",
lambda initial_domain="feishu": {
"app_id": "cli_app",
"app_secret": "secret",
"domain": "feishu",
},
)
channel = FeishuChannel({}, None)
assert await channel.login() is True
assert missing_config.exists()
data = json.loads(missing_config.read_text(encoding="utf-8"))
instance = _default_feishu_instance(data)
assert instance["id"] == "default"
assert instance["appId"] == "cli_app"
@@ -0,0 +1,68 @@
# Check optional Feishu dependencies before running tests
try:
from nanobot.channels import feishu
FEISHU_AVAILABLE = getattr(feishu, "FEISHU_AVAILABLE", False)
except ImportError:
FEISHU_AVAILABLE = False
if not FEISHU_AVAILABLE:
import pytest
pytest.skip("Feishu dependencies not installed (lark-oapi)", allow_module_level=True)
from nanobot.channels.feishu.runtime import FeishuChannel
def test_parse_md_table_strips_markdown_formatting_in_headers_and_cells() -> None:
table = FeishuChannel._parse_md_table(
"""
| **Name** | __Status__ | *Notes* | ~~State~~ |
| --- | --- | --- | --- |
| **Alice** | __Ready__ | *Fast* | ~~Old~~ |
"""
)
assert table is not None
assert [col["display_name"] for col in table["columns"]] == [
"Name",
"Status",
"Notes",
"State",
]
assert table["rows"] == [
{"c0": "Alice", "c1": "Ready", "c2": "Fast", "c3": "Old"}
]
def test_split_headings_strips_embedded_markdown_before_bolding() -> None:
channel = FeishuChannel.__new__(FeishuChannel)
elements = channel._split_headings("# **Important** *status* ~~update~~")
assert elements == [
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": "**Important status update**",
},
}
]
def test_split_headings_keeps_markdown_body_and_code_blocks_intact() -> None:
channel = FeishuChannel.__new__(FeishuChannel)
elements = channel._split_headings(
"# **Heading**\n\nBody with **bold** text.\n\n```python\nprint('hi')\n```"
)
assert elements[0] == {
"tag": "div",
"text": {
"tag": "lark_md",
"content": "**Heading**",
},
}
assert elements[1]["tag"] == "markdown"
assert "Body with **bold** text." in elements[1]["content"]
assert "```python\nprint('hi')\n```" in elements[1]["content"]
@@ -0,0 +1,38 @@
from pathlib import Path
from types import SimpleNamespace
import pytest
from nanobot.channels.feishu import runtime as feishu_module
from nanobot.channels.feishu.runtime import FeishuChannel
@pytest.mark.asyncio
async def test_feishu_downloaded_media_filename_cannot_escape_media_dir(monkeypatch, tmp_path):
media_dir = tmp_path / "media"
media_dir.mkdir()
outside = tmp_path / "escaped.txt"
monkeypatch.setattr(feishu_module, "get_media_dir", lambda _channel: media_dir)
channel = FeishuChannel.__new__(FeishuChannel)
channel.logger = SimpleNamespace(
debug=lambda *args, **kwargs: None,
warning=lambda *args, **kwargs: None,
)
def fake_download(_message_id, _file_key, _resource_type):
return b"owned", "../escaped.txt"
channel._download_file_sync = fake_download
path_str, content = await channel._download_and_save_media(
"file", {"file_key": "fk_123"}, "msg_123"
)
saved_path = Path(path_str)
assert not outside.exists()
assert saved_path.parent == media_dir
assert saved_path.name == "escaped.txt"
assert saved_path.read_bytes() == b"owned"
assert content == f"[file: {saved_path}]"
@@ -0,0 +1,60 @@
"""Tests for Feishu _is_bot_mentioned logic."""
from types import SimpleNamespace
from nanobot.channels.feishu.runtime import FeishuChannel
def _make_channel(bot_open_id: str | None = None) -> FeishuChannel:
config = SimpleNamespace(
app_id="test_id",
app_secret="test_secret",
verification_token="",
event_encrypt_key="",
group_policy="mention",
)
ch = FeishuChannel.__new__(FeishuChannel)
ch.config = config
ch._bot_open_id = bot_open_id
return ch
def _make_message(mentions=None, content="hello"):
return SimpleNamespace(content=content, mentions=mentions)
def _make_mention(open_id: str, user_id: str | None = None):
mid = SimpleNamespace(open_id=open_id, user_id=user_id)
return SimpleNamespace(id=mid)
class TestIsBotMentioned:
def test_exact_match_with_bot_open_id(self):
ch = _make_channel(bot_open_id="ou_bot123")
msg = _make_message(mentions=[_make_mention("ou_bot123")])
assert ch._is_bot_mentioned(msg) is True
def test_no_match_different_bot(self):
ch = _make_channel(bot_open_id="ou_bot123")
msg = _make_message(mentions=[_make_mention("ou_other_bot")])
assert ch._is_bot_mentioned(msg) is False
def test_at_all_always_matches(self):
ch = _make_channel(bot_open_id="ou_bot123")
msg = _make_message(content="@_all hello")
assert ch._is_bot_mentioned(msg) is True
def test_fallback_heuristic_when_no_bot_open_id(self):
ch = _make_channel(bot_open_id=None)
msg = _make_message(mentions=[_make_mention("ou_some_bot", user_id=None)])
assert ch._is_bot_mentioned(msg) is True
def test_fallback_ignores_user_mentions(self):
ch = _make_channel(bot_open_id=None)
msg = _make_message(mentions=[_make_mention("ou_user", user_id="u_12345")])
assert ch._is_bot_mentioned(msg) is False
def test_no_mentions_returns_false(self):
ch = _make_channel(bot_open_id="ou_bot123")
msg = _make_message(mentions=None)
assert ch._is_bot_mentioned(msg) is False
@@ -0,0 +1,65 @@
"""Tests for FeishuChannel._resolve_mentions."""
from types import SimpleNamespace
from nanobot.channels.feishu.runtime import FeishuChannel
def _mention(key: str, name: str, open_id: str = "", user_id: str = ""):
"""Build a mock MentionEvent-like object."""
id_obj = SimpleNamespace(open_id=open_id, user_id=user_id) if (open_id or user_id) else None
return SimpleNamespace(key=key, name=name, id=id_obj)
class TestResolveMentions:
def test_single_mention_replaced(self):
text = "hello @_user_1 how are you"
mentions = [_mention("@_user_1", "Alice", open_id="ou_abc123")]
result = FeishuChannel._resolve_mentions(text, mentions)
assert "@Alice (ou_abc123)" in result
assert "@_user_1" not in result
def test_mention_with_both_ids(self):
text = "@_user_1 said hi"
mentions = [_mention("@_user_1", "Bob", open_id="ou_abc", user_id="uid_456")]
result = FeishuChannel._resolve_mentions(text, mentions)
assert "@Bob (ou_abc, user id: uid_456)" in result
def test_mention_no_id_skipped(self):
"""When mention has no id object, the placeholder is left unchanged."""
text = "@_user_1 said hi"
mentions = [SimpleNamespace(key="@_user_1", name="Charlie", id=None)]
result = FeishuChannel._resolve_mentions(text, mentions)
assert result == "@_user_1 said hi"
def test_multiple_mentions(self):
text = "@_user_1 and @_user_2 are here"
mentions = [
_mention("@_user_1", "Alice", open_id="ou_a"),
_mention("@_user_2", "Bob", open_id="ou_b"),
]
result = FeishuChannel._resolve_mentions(text, mentions)
assert "@Alice (ou_a)" in result
assert "@Bob (ou_b)" in result
assert "@_user_1" not in result
assert "@_user_2" not in result
def test_mention_before_punctuation_replaced(self):
text = "hello @_user_1, are you there?"
mentions = [_mention("@_user_1", "Alice", open_id="ou_a")]
result = FeishuChannel._resolve_mentions(text, mentions)
assert result == "hello @Alice (ou_a), are you there?"
def test_no_mentions_returns_text(self):
assert FeishuChannel._resolve_mentions("hello world", None) == "hello world"
assert FeishuChannel._resolve_mentions("hello world", []) == "hello world"
def test_empty_text_returns_empty(self):
mentions = [_mention("@_user_1", "Alice", open_id="ou_a")]
assert FeishuChannel._resolve_mentions("", mentions) == ""
def test_mention_key_not_in_text_skipped(self):
text = "hello world"
mentions = [_mention("@_user_99", "Ghost", open_id="ou_ghost")]
result = FeishuChannel._resolve_mentions(text, mentions)
assert result == "hello world"
@@ -0,0 +1,76 @@
# Check optional Feishu dependencies before running tests
try:
from nanobot.channels.feishu import runtime as feishu
FEISHU_AVAILABLE = getattr(feishu, "FEISHU_AVAILABLE", False)
except ImportError:
FEISHU_AVAILABLE = False
if not FEISHU_AVAILABLE:
import pytest
pytest.skip("Feishu dependencies not installed (lark-oapi)", allow_module_level=True)
from nanobot.channels.feishu.runtime import FeishuChannel, _extract_post_content
def test_extract_post_content_supports_post_wrapper_shape() -> None:
payload = {
"post": {
"zh_cn": {
"title": "日报",
"content": [
[
{"tag": "text", "text": "完成"},
{"tag": "img", "image_key": "img_1"},
]
],
}
}
}
text, image_keys = _extract_post_content(payload)
assert text == "日报 完成"
assert image_keys == ["img_1"]
def test_extract_post_content_keeps_direct_shape_behavior() -> None:
payload = {
"title": "Daily",
"content": [
[
{"tag": "text", "text": "report"},
{"tag": "img", "image_key": "img_a"},
{"tag": "img", "image_key": "img_b"},
]
],
}
text, image_keys = _extract_post_content(payload)
assert text == "Daily report"
assert image_keys == ["img_a", "img_b"]
def test_register_optional_event_keeps_builder_when_method_missing() -> None:
class Builder:
pass
builder = Builder()
same = FeishuChannel._register_optional_event(builder, "missing", object())
assert same is builder
def test_register_optional_event_calls_supported_method() -> None:
called = []
class Builder:
def register_event(self, handler):
called.append(handler)
return self
builder = Builder()
handler = object()
same = FeishuChannel._register_optional_event(builder, "register_event", handler)
assert same is builder
assert called == [handler]
@@ -0,0 +1,329 @@
# ruff: noqa: E402
"""Tests for Feishu reaction add/remove and auto-cleanup on stream end."""
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
pytest.importorskip("lark_oapi")
from nanobot.bus.queue import MessageBus
from nanobot.channels.feishu.runtime import FeishuChannel, FeishuConfig, _FeishuStreamBuf
def _make_channel() -> FeishuChannel:
config = FeishuConfig(
enabled=True,
app_id="cli_test",
app_secret="secret",
allow_from=["*"],
)
ch = FeishuChannel(config, MessageBus())
ch._client = MagicMock()
ch._loop = None
return ch
def _mock_reaction_create_response(reaction_id: str = "reaction_001", success: bool = True):
resp = MagicMock()
resp.success.return_value = success
resp.code = 0 if success else 99999
resp.msg = "ok" if success else "error"
if success:
resp.data = SimpleNamespace(reaction_id=reaction_id)
else:
resp.data = None
return resp
# ── _add_reaction_sync ──────────────────────────────────────────────────────
class TestAddReactionSync:
def test_returns_reaction_id_on_success(self):
ch = _make_channel()
ch._client.im.v1.message_reaction.create.return_value = _mock_reaction_create_response("rx_42")
result = ch._add_reaction_sync("om_001", "THUMBSUP")
assert result == "rx_42"
def test_returns_none_when_response_fails(self):
ch = _make_channel()
ch._client.im.v1.message_reaction.create.return_value = _mock_reaction_create_response(success=False)
assert ch._add_reaction_sync("om_001", "THUMBSUP") is None
def test_returns_none_when_response_data_is_none(self):
ch = _make_channel()
resp = MagicMock()
resp.success.return_value = True
resp.data = None
ch._client.im.v1.message_reaction.create.return_value = resp
assert ch._add_reaction_sync("om_001", "THUMBSUP") is None
def test_returns_none_on_exception(self):
ch = _make_channel()
ch._client.im.v1.message_reaction.create.side_effect = RuntimeError("network error")
assert ch._add_reaction_sync("om_001", "THUMBSUP") is None
# ── _add_reaction (async) ───────────────────────────────────────────────────
class TestAddReactionAsync:
@pytest.mark.asyncio
async def test_returns_reaction_id(self):
ch = _make_channel()
ch._add_reaction_sync = MagicMock(return_value="rx_99")
result = await ch._add_reaction("om_001", "EYES")
assert result == "rx_99"
@pytest.mark.asyncio
async def test_returns_none_when_no_client(self):
ch = _make_channel()
ch._client = None
result = await ch._add_reaction("om_001", "THUMBSUP")
assert result is None
# ── _remove_reaction_sync ───────────────────────────────────────────────────
class TestRemoveReactionSync:
def test_calls_delete_on_success(self):
ch = _make_channel()
resp = MagicMock()
resp.success.return_value = True
ch._client.im.v1.message_reaction.delete.return_value = resp
ch._remove_reaction_sync("om_001", "rx_42")
ch._client.im.v1.message_reaction.delete.assert_called_once()
def test_handles_failure_gracefully(self):
ch = _make_channel()
resp = MagicMock()
resp.success.return_value = False
resp.code = 99999
resp.msg = "not found"
ch._client.im.v1.message_reaction.delete.return_value = resp
# Should not raise
ch._remove_reaction_sync("om_001", "rx_42")
ch._client.im.v1.message_reaction.delete.assert_called_once()
def test_handles_exception_gracefully(self):
ch = _make_channel()
ch._client.im.v1.message_reaction.delete.side_effect = RuntimeError("network error")
# Should not raise
ch._remove_reaction_sync("om_001", "rx_42")
ch._client.im.v1.message_reaction.delete.assert_called_once()
# ── _remove_reaction (async) ────────────────────────────────────────────────
class TestRemoveReactionAsync:
@pytest.mark.asyncio
async def test_calls_sync_helper(self):
ch = _make_channel()
ch._remove_reaction_sync = MagicMock()
await ch._remove_reaction("om_001", "rx_42")
ch._remove_reaction_sync.assert_called_once_with("om_001", "rx_42")
@pytest.mark.asyncio
async def test_noop_when_no_client(self):
ch = _make_channel()
ch._client = None
ch._remove_reaction_sync = MagicMock()
await ch._remove_reaction("om_001", "rx_42")
ch._remove_reaction_sync.assert_not_called()
@pytest.mark.asyncio
async def test_noop_when_reaction_id_is_empty(self):
ch = _make_channel()
ch._remove_reaction_sync = MagicMock()
await ch._remove_reaction("om_001", "")
ch._remove_reaction_sync.assert_not_called()
@pytest.mark.asyncio
async def test_noop_when_reaction_id_is_none(self):
ch = _make_channel()
ch._remove_reaction_sync = MagicMock()
await ch._remove_reaction("om_001", None)
ch._remove_reaction_sync.assert_not_called()
# ── send_delta stream end: reaction auto-cleanup ────────────────────────────
class TestStreamEndReactionCleanup:
@pytest.mark.asyncio
async def test_stream_buffers_are_scoped_by_message_id(self):
ch = _make_channel()
ch._create_streaming_card_sync = MagicMock(return_value=None)
await ch.send_delta(
"oc_chat1", "first",
metadata={"message_id": "om_first"},
)
await ch.send_delta(
"oc_chat1", "second",
metadata={"message_id": "om_second"},
)
assert ch._stream_bufs["om_first"].text == "first"
assert ch._stream_bufs["om_second"].text == "second"
assert "oc_chat1" not in ch._stream_bufs
@pytest.mark.asyncio
async def test_removes_reaction_on_stream_end(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Done", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._reaction_ids["om_001"] = "rx_42"
ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
ch._remove_reaction = AsyncMock()
await ch.send_delta(
"oc_chat1", "",
metadata={"message_id": "om_001"},
stream_end=True,
)
ch._remove_reaction.assert_called_once_with("om_001", "rx_42")
@pytest.mark.asyncio
async def test_no_removal_when_message_id_missing(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Done", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
ch._remove_reaction = AsyncMock()
await ch.send_delta(
"oc_chat1", "",
stream_end=True,
)
ch._remove_reaction.assert_not_called()
@pytest.mark.asyncio
async def test_no_removal_when_reaction_id_missing(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Done", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
ch._remove_reaction = AsyncMock()
await ch.send_delta(
"oc_chat1", "",
metadata={"message_id": "om_001"},
stream_end=True,
)
ch._remove_reaction.assert_not_called()
@pytest.mark.asyncio
async def test_no_removal_when_both_ids_missing(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Done", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
ch._remove_reaction = AsyncMock()
await ch.send_delta("oc_chat1", "", stream_end=True)
ch._remove_reaction.assert_not_called()
@pytest.mark.asyncio
async def test_no_removal_when_not_stream_end(self):
ch = _make_channel()
ch._remove_reaction = AsyncMock()
await ch.send_delta(
"oc_chat1", "more text",
metadata={"message_id": "om_001", "reaction_id": "rx_42"},
)
ch._remove_reaction.assert_not_called()
@pytest.mark.asyncio
async def test_no_removal_when_resuming(self):
"""resuming=True means more tool-call rounds follow; reaction must persist."""
ch = _make_channel()
ch.config.done_emoji = "DONE"
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="partial", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._reaction_ids["om_001"] = "rx_42"
ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
ch._remove_reaction = AsyncMock()
ch._add_reaction = AsyncMock()
await ch.send_delta(
"oc_chat1", "",
metadata={"message_id": "om_001"},
stream_end=True,
resuming=True,
)
ch._remove_reaction.assert_not_called()
ch._add_reaction.assert_not_called()
# OnIt reaction id is still tracked for the eventual final stream end
assert ch._reaction_ids.get("om_001") == "rx_42"
@pytest.mark.asyncio
async def test_done_emoji_only_on_final_stream_end(self):
"""Across resuming rounds, done_emoji is added only on the final round."""
ch = _make_channel()
ch.config.done_emoji = "DONE"
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="t", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._reaction_ids["om_001"] = "rx_42"
ch._client.cardkit.v1.card_element.content.return_value = MagicMock(success=MagicMock(return_value=True))
ch._client.cardkit.v1.card.settings.return_value = MagicMock(success=MagicMock(return_value=True))
ch._remove_reaction = AsyncMock()
ch._add_reaction = AsyncMock()
# Intermediate stream end (more tool calls coming).
await ch.send_delta(
"oc_chat1", "",
metadata={"message_id": "om_001"},
stream_end=True,
resuming=True,
)
ch._remove_reaction.assert_not_called()
ch._add_reaction.assert_not_called()
# Re-prime the stream buffer for the final round (the previous stream end popped it).
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="t", card_id="card_1", sequence=5, last_edit=0.0,
)
# Final stream end (resuming=False): OnIt removed, done_emoji added.
await ch.send_delta(
"oc_chat1", "",
metadata={"message_id": "om_001"},
stream_end=True,
resuming=False,
)
ch._remove_reaction.assert_called_once_with("om_001", "rx_42")
ch._add_reaction.assert_called_once_with("om_001", "DONE")
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,644 @@
# ruff: noqa: E402
"""Tests for Feishu streaming (send_delta) via CardKit streaming API."""
import time
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
pytest.importorskip("lark_oapi")
from nanobot.bus.events import OutboundMessage
from nanobot.bus.outbound_events import ProgressEvent
from nanobot.bus.queue import MessageBus
from nanobot.channels.feishu.runtime import FeishuChannel, FeishuConfig, _FeishuStreamBuf
def _make_channel(streaming: bool = True, reply_to_message: bool = False) -> FeishuChannel:
config = FeishuConfig(
enabled=True,
app_id="cli_test",
app_secret="secret",
allow_from=["*"],
streaming=streaming,
reply_to_message=reply_to_message,
)
ch = FeishuChannel(config, MessageBus())
ch._client = MagicMock()
ch._loop = None
return ch
def _mock_create_card_response(card_id: str = "card_stream_001"):
resp = MagicMock()
resp.success.return_value = True
resp.data = SimpleNamespace(card_id=card_id)
return resp
def _mock_send_response(message_id: str = "om_stream_001"):
resp = MagicMock()
resp.success.return_value = True
resp.data = SimpleNamespace(message_id=message_id)
return resp
def _mock_content_response(success: bool = True):
resp = MagicMock()
resp.success.return_value = success
resp.code = 0 if success else 99999
resp.msg = "ok" if success else "error"
return resp
class TestFeishuStreamingConfig:
def test_streaming_default_true(self):
assert FeishuConfig().streaming is True
def test_supports_streaming_when_enabled(self):
ch = _make_channel(streaming=True)
assert ch.supports_streaming is True
def test_supports_streaming_disabled(self):
ch = _make_channel(streaming=False)
assert ch.supports_streaming is False
class TestCreateStreamingCard:
def test_returns_card_id_on_success(self):
ch = _make_channel()
ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_123")
ch._client.im.v1.message.create.return_value = _mock_send_response()
result = ch._create_streaming_card_sync("chat_id", "oc_chat1")
assert result == "card_123"
ch._client.cardkit.v1.card.create.assert_called_once()
ch._client.im.v1.message.create.assert_called_once()
def test_returns_none_on_failure(self):
ch = _make_channel()
resp = MagicMock()
resp.success.return_value = False
resp.code = 99999
resp.msg = "error"
ch._client.cardkit.v1.card.create.return_value = resp
assert ch._create_streaming_card_sync("chat_id", "oc_chat1") is None
def test_returns_none_on_exception(self):
ch = _make_channel()
ch._client.cardkit.v1.card.create.side_effect = RuntimeError("network")
assert ch._create_streaming_card_sync("chat_id", "oc_chat1") is None
def test_returns_none_when_card_send_fails(self):
ch = _make_channel()
ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_123")
resp = MagicMock()
resp.success.return_value = False
resp.code = 99999
resp.msg = "error"
resp.get_log_id.return_value = "log1"
ch._client.im.v1.message.create.return_value = resp
assert ch._create_streaming_card_sync("chat_id", "oc_chat1") is None
class TestCloseStreamingMode:
def test_returns_true_on_success(self):
ch = _make_channel()
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response(True)
assert ch._close_streaming_mode_sync("card_1", 10) is True
def test_returns_false_on_failure(self):
ch = _make_channel()
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response(False)
assert ch._close_streaming_mode_sync("card_1", 10) is False
def test_returns_false_on_exception(self):
ch = _make_channel()
ch._client.cardkit.v1.card.settings.side_effect = RuntimeError("err")
assert ch._close_streaming_mode_sync("card_1", 10) is False
class TestStreamUpdateWithReopen:
def test_reopens_streaming_mode_and_retries_update(self):
ch = _make_channel()
ch._client.cardkit.v1.card_element.content.side_effect = [
_mock_content_response(False),
_mock_content_response(True),
]
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response(True)
assert ch._stream_update_text_with_reopen_sync("card_1", "hello", 4) == (True, 6)
assert ch._client.cardkit.v1.card_element.content.call_count == 2
settings_call = ch._client.cardkit.v1.card.settings.call_args[0][0]
assert settings_call.body.sequence == 5
assert '"streaming_mode": true' in settings_call.body.settings
class TestStreamUpdateText:
def test_returns_true_on_success(self):
ch = _make_channel()
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response(True)
assert ch._stream_update_text_sync("card_1", "hello", 1) is True
def test_returns_false_on_failure(self):
ch = _make_channel()
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response(False)
assert ch._stream_update_text_sync("card_1", "hello", 1) is False
def test_returns_false_on_exception(self):
ch = _make_channel()
ch._client.cardkit.v1.card_element.content.side_effect = RuntimeError("err")
assert ch._stream_update_text_sync("card_1", "hello", 1) is False
class TestSendDelta:
@pytest.mark.asyncio
async def test_first_delta_creates_card_and_sends(self):
ch = _make_channel()
ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_new")
ch._client.im.v1.message.create.return_value = _mock_send_response("om_new")
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "Hello ")
assert "oc_chat1" in ch._stream_bufs
buf = ch._stream_bufs["oc_chat1"]
assert buf.text == "Hello "
assert buf.card_id == "card_new"
assert buf.sequence == 1
ch._client.cardkit.v1.card.create.assert_called_once()
ch._client.im.v1.message.create.assert_called_once()
ch._client.cardkit.v1.card_element.content.assert_called_once()
@pytest.mark.asyncio
async def test_first_delta_closes_blank_card_when_initial_update_fails(self):
ch = _make_channel()
ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_new")
ch._client.im.v1.message.create.return_value = _mock_send_response("om_new")
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response(False)
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response(True)
await ch.send_delta("oc_chat1", "Hello ")
buf = ch._stream_bufs["oc_chat1"]
assert buf.text == "Hello "
assert buf.card_id is None
assert ch._client.cardkit.v1.card_element.content.call_count == 2
assert ch._client.cardkit.v1.card.settings.call_count == 2
close_call = ch._client.cardkit.v1.card.settings.call_args_list[-1][0][0]
assert '"streaming_mode": false' in close_call.body.settings
@pytest.mark.asyncio
async def test_group_delta_uses_create_when_reply_disabled(self):
ch = _make_channel(reply_to_message=False)
ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_new")
ch._client.im.v1.message.create.return_value = _mock_send_response("om_new")
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta(
"oc_chat1",
"Hello ",
metadata={"message_id": "om_001", "chat_type": "group"},
)
ch._client.im.v1.message.create.assert_called_once()
ch._client.im.v1.message.reply.assert_not_called()
@pytest.mark.asyncio
async def test_group_delta_keeps_existing_topic_when_reply_disabled(self):
ch = _make_channel(reply_to_message=False)
ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_new")
reply_resp = MagicMock()
reply_resp.success.return_value = True
ch._client.im.v1.message.reply.return_value = reply_resp
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta(
"oc_chat1",
"Hello ",
metadata={"message_id": "om_001", "chat_type": "group", "thread_id": "ot_001"},
)
ch._client.im.v1.message.reply.assert_called_once()
ch._client.im.v1.message.create.assert_not_called()
request = ch._client.im.v1.message.reply.call_args[0][0]
assert request.request_body.reply_in_thread is not True
@pytest.mark.asyncio
async def test_group_delta_replies_in_thread_when_reply_enabled(self):
ch = _make_channel(reply_to_message=True)
ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_new")
reply_resp = MagicMock()
reply_resp.success.return_value = True
ch._client.im.v1.message.reply.return_value = reply_resp
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta(
"oc_chat1",
"Hello ",
metadata={"message_id": "om_001", "chat_type": "group"},
)
ch._client.im.v1.message.reply.assert_called_once()
ch._client.im.v1.message.create.assert_not_called()
request = ch._client.im.v1.message.reply.call_args[0][0]
assert request.request_body.reply_in_thread is True
@pytest.mark.asyncio
async def test_second_delta_within_interval_skips_update(self):
ch = _make_channel()
buf = _FeishuStreamBuf(text="Hello ", card_id="card_1", sequence=1, last_edit=time.monotonic())
ch._stream_bufs["oc_chat1"] = buf
await ch.send_delta("oc_chat1", "world")
assert buf.text == "Hello world"
ch._client.cardkit.v1.card_element.content.assert_not_called()
@pytest.mark.asyncio
async def test_delta_after_interval_updates_text(self):
ch = _make_channel()
buf = _FeishuStreamBuf(text="Hello ", card_id="card_1", sequence=1, last_edit=time.monotonic() - 1.0)
ch._stream_bufs["oc_chat1"] = buf
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "world")
assert buf.text == "Hello world"
assert buf.sequence == 2
ch._client.cardkit.v1.card_element.content.assert_called_once()
@pytest.mark.asyncio
async def test_stream_end_sends_final_update(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Final content", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "", stream_end=True)
assert "oc_chat1" not in ch._stream_bufs
ch._client.cardkit.v1.card_element.content.assert_called_once()
ch._client.cardkit.v1.card.settings.assert_called_once()
settings_call = ch._client.cardkit.v1.card.settings.call_args[0][0]
assert settings_call.body.sequence == 5 # after final content seq 4
@pytest.mark.asyncio
async def test_stream_end_fallback_when_no_card_id(self):
"""If card creation failed, stream_end falls back to a plain card message."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Fallback content", card_id=None, sequence=0, last_edit=0.0,
)
ch._client.im.v1.message.create.return_value = _mock_send_response("om_fb")
await ch.send_delta("oc_chat1", "", stream_end=True)
assert "oc_chat1" not in ch._stream_bufs
ch._client.cardkit.v1.card_element.content.assert_not_called()
ch._client.im.v1.message.create.assert_called_once()
@pytest.mark.asyncio
async def test_stream_end_fallback_group_uses_create_when_reply_disabled(self):
ch = _make_channel(reply_to_message=False)
ch._stream_bufs["om_001"] = _FeishuStreamBuf(
text="Fallback content", card_id=None, sequence=0, last_edit=0.0,
)
ch._client.im.v1.message.create.return_value = _mock_send_response("om_fb")
await ch.send_delta(
"oc_chat1",
"",
metadata={"message_id": "om_001", "chat_type": "group"},
stream_end=True,
)
ch._client.im.v1.message.create.assert_called_once()
ch._client.im.v1.message.reply.assert_not_called()
@pytest.mark.asyncio
async def test_stream_end_fallback_keeps_existing_topic_when_reply_disabled(self):
ch = _make_channel(reply_to_message=False)
ch._stream_bufs["om_001"] = _FeishuStreamBuf(
text="Fallback content", card_id=None, sequence=0, last_edit=0.0,
)
reply_resp = MagicMock()
reply_resp.success.return_value = True
ch._client.im.v1.message.reply.return_value = reply_resp
await ch.send_delta(
"oc_chat1",
"",
metadata={
"message_id": "om_001",
"chat_type": "group",
"thread_id": "ot_001",
},
stream_end=True,
)
ch._client.im.v1.message.reply.assert_called_once()
ch._client.im.v1.message.create.assert_not_called()
request = ch._client.im.v1.message.reply.call_args[0][0]
assert request.request_body.reply_in_thread is not True
@pytest.mark.asyncio
async def test_stream_end_fallback_group_replies_when_reply_enabled(self):
ch = _make_channel(reply_to_message=True)
ch._stream_bufs["om_001"] = _FeishuStreamBuf(
text="Fallback content", card_id=None, sequence=0, last_edit=0.0,
)
reply_resp = MagicMock()
reply_resp.success.return_value = True
ch._client.im.v1.message.reply.return_value = reply_resp
await ch.send_delta(
"oc_chat1",
"",
metadata={"message_id": "om_001", "chat_type": "group"},
stream_end=True,
)
ch._client.im.v1.message.reply.assert_called_once()
ch._client.im.v1.message.create.assert_not_called()
request = ch._client.im.v1.message.reply.call_args[0][0]
assert request.request_body.reply_in_thread is True
@pytest.mark.asyncio
async def test_stream_end_fallback_when_final_update_fails(self):
"""If streaming mode was closed (e.g. Feishu timeout), fall back to a regular card."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Lost content", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response(success=False)
ch._client.im.v1.message.create.return_value = _mock_send_response("om_fb")
await ch.send_delta("oc_chat1", "", stream_end=True)
assert "oc_chat1" not in ch._stream_bufs
assert ch._client.cardkit.v1.card.settings.call_count == 2
# Should fall back to sending a regular interactive card
ch._client.im.v1.message.create.assert_called_once()
@pytest.mark.asyncio
async def test_stream_end_reopens_streaming_card_before_fallback(self):
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Recovered content", card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.side_effect = [
_mock_content_response(False),
_mock_content_response(True),
]
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response(True)
await ch.send_delta("oc_chat1", "", stream_end=True)
assert "oc_chat1" not in ch._stream_bufs
assert ch._client.cardkit.v1.card_element.content.call_count == 2
assert ch._client.cardkit.v1.card.settings.call_count == 2
ch._client.im.v1.message.create.assert_not_called()
@pytest.mark.asyncio
async def test_stream_end_without_buf_is_noop(self):
ch = _make_channel()
await ch.send_delta("oc_chat1", "", stream_end=True)
ch._client.cardkit.v1.card_element.content.assert_not_called()
@pytest.mark.asyncio
async def test_empty_delta_skips_send(self):
ch = _make_channel()
await ch.send_delta("oc_chat1", " ")
assert "oc_chat1" in ch._stream_bufs
ch._client.cardkit.v1.card.create.assert_not_called()
@pytest.mark.asyncio
async def test_no_client_returns_early(self):
ch = _make_channel()
ch._client = None
await ch.send_delta("oc_chat1", "text")
assert "oc_chat1" not in ch._stream_bufs
@pytest.mark.asyncio
async def test_sequence_increments_correctly(self):
ch = _make_channel()
buf = _FeishuStreamBuf(text="a", card_id="card_1", sequence=5, last_edit=0.0)
ch._stream_bufs["oc_chat1"] = buf
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "b")
assert buf.sequence == 6
buf.last_edit = 0.0 # reset to bypass throttle
await ch.send_delta("oc_chat1", "c")
assert buf.sequence == 7
class TestToolHintInlineStreaming:
"""Tool hint messages should be inlined into active streaming cards."""
@pytest.mark.asyncio
async def test_tool_hint_inlined_when_stream_active(self):
"""With an active streaming buffer, tool hint appends to the card."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Partial answer", card_id="card_1", sequence=2, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
msg = OutboundMessage(
channel="feishu", chat_id="oc_chat1",
content='web_fetch("https://example.com")',
event=ProgressEvent(content='web_fetch("https://example.com")', tool_hint=True),
)
await ch.send(msg)
buf = ch._stream_bufs["oc_chat1"]
assert '🔧 web_fetch("https://example.com")' in buf.text
assert buf.sequence == 3
ch._client.cardkit.v1.card_element.content.assert_called_once()
ch._client.im.v1.message.create.assert_not_called()
@pytest.mark.asyncio
async def test_tool_hint_preserved_on_next_delta(self):
"""When new delta arrives, the tool hint is kept as permanent content and delta appends after it."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Partial answer\n\n🔧 web_fetch(\"url\")\n\n",
card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", " continued")
buf = ch._stream_bufs["oc_chat1"]
assert "Partial answer" in buf.text
assert "🔧 web_fetch" in buf.text
assert buf.text.endswith(" continued")
@pytest.mark.asyncio
async def test_tool_hint_fallback_when_no_stream(self):
"""Without an active buffer, tool hint falls back to a standalone card."""
ch = _make_channel()
ch._client.im.v1.message.create.return_value = _mock_send_response("om_hint")
msg = OutboundMessage(
channel="feishu", chat_id="oc_chat1",
content='read_file("path")',
event=ProgressEvent(content='read_file("path")', tool_hint=True),
)
await ch.send(msg)
assert "oc_chat1" not in ch._stream_bufs
ch._client.im.v1.message.create.assert_called_once()
@pytest.mark.asyncio
async def test_tool_hint_group_uses_create_when_reply_disabled(self):
ch = _make_channel(reply_to_message=False)
ch._client.im.v1.message.create.return_value = _mock_send_response("om_hint")
msg = OutboundMessage(
channel="feishu", chat_id="oc_chat1",
content='read_file("path")',
event=ProgressEvent(content='read_file("path")', tool_hint=True),
metadata={"message_id": "om_001", "chat_type": "group"},
)
await ch.send(msg)
ch._client.im.v1.message.create.assert_called_once()
ch._client.im.v1.message.reply.assert_not_called()
@pytest.mark.asyncio
async def test_tool_hint_keeps_existing_topic_when_reply_disabled(self):
ch = _make_channel(reply_to_message=False)
reply_resp = MagicMock()
reply_resp.success.return_value = True
ch._client.im.v1.message.reply.return_value = reply_resp
msg = OutboundMessage(
channel="feishu", chat_id="oc_chat1",
content='read_file("path")',
event=ProgressEvent(content='read_file("path")', tool_hint=True),
metadata={
"message_id": "om_001",
"chat_type": "group",
"thread_id": "ot_001",
},
)
await ch.send(msg)
ch._client.im.v1.message.reply.assert_called_once()
ch._client.im.v1.message.create.assert_not_called()
request = ch._client.im.v1.message.reply.call_args[0][0]
assert request.request_body.reply_in_thread is not True
@pytest.mark.asyncio
async def test_tool_hint_group_replies_when_reply_enabled(self):
ch = _make_channel(reply_to_message=True)
reply_resp = MagicMock()
reply_resp.success.return_value = True
ch._client.im.v1.message.reply.return_value = reply_resp
msg = OutboundMessage(
channel="feishu", chat_id="oc_chat1",
content='read_file("path")',
event=ProgressEvent(content='read_file("path")', tool_hint=True),
metadata={"message_id": "om_001", "chat_type": "group"},
)
await ch.send(msg)
ch._client.im.v1.message.reply.assert_called_once()
ch._client.im.v1.message.create.assert_not_called()
request = ch._client.im.v1.message.reply.call_args[0][0]
assert request.request_body.reply_in_thread is True
@pytest.mark.asyncio
async def test_consecutive_tool_hints_append(self):
"""When multiple tool hints arrive consecutively, each appends to the card."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Partial answer", card_id="card_1", sequence=2, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
msg1 = OutboundMessage(
channel="feishu", chat_id="oc_chat1",
content='$ cd /project',
event=ProgressEvent(content='$ cd /project', tool_hint=True),
)
await ch.send(msg1)
msg2 = OutboundMessage(
channel="feishu", chat_id="oc_chat1",
content='$ git status',
event=ProgressEvent(content='$ git status', tool_hint=True),
)
await ch.send(msg2)
buf = ch._stream_bufs["oc_chat1"]
assert "$ cd /project" in buf.text
assert "$ git status" in buf.text
assert buf.text.startswith("Partial answer")
assert "🔧 $ cd /project" in buf.text
assert "🔧 $ git status" in buf.text
@pytest.mark.asyncio
async def test_tool_hint_preserved_on_final_stream_end(self):
"""When stream end closes the card, tool hint is kept in the final text."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Final content\n\n🔧 web_fetch(\"url\")\n\n",
card_id="card_1", sequence=3, last_edit=0.0,
)
ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response()
ch._client.cardkit.v1.card.settings.return_value = _mock_content_response()
await ch.send_delta("oc_chat1", "", stream_end=True)
assert "oc_chat1" not in ch._stream_bufs
update_call = ch._client.cardkit.v1.card_element.content.call_args[0][0]
assert "🔧" in update_call.body.content
@pytest.mark.asyncio
async def test_empty_tool_hint_is_noop(self):
"""Empty or whitespace-only tool hint content is silently ignored."""
ch = _make_channel()
ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf(
text="Partial answer", card_id="card_1", sequence=2, last_edit=0.0,
)
for content in ("", " ", "\t\n"):
msg = OutboundMessage(
channel="feishu", chat_id="oc_chat1",
content=content,
event=ProgressEvent(content=content, tool_hint=True),
)
await ch.send(msg)
buf = ch._stream_bufs["oc_chat1"]
assert buf.text == "Partial answer"
assert buf.sequence == 2
ch._client.cardkit.v1.card_element.content.assert_not_called()
class TestSendMessageReturnsId:
def test_returns_message_id_on_success(self):
ch = _make_channel()
ch._client.im.v1.message.create.return_value = _mock_send_response("om_abc")
result = ch._send_message_sync("chat_id", "oc_chat1", "text", '{"text":"hi"}')
assert result == "om_abc"
def test_returns_none_on_failure(self):
ch = _make_channel()
resp = MagicMock()
resp.success.return_value = False
resp.code = 99999
resp.msg = "error"
resp.get_log_id.return_value = "log1"
ch._client.im.v1.message.create.return_value = resp
result = ch._send_message_sync("chat_id", "oc_chat1", "text", '{"text":"hi"}')
assert result is None
@@ -0,0 +1,115 @@
"""Tests for FeishuChannel._split_elements_by_table_limit.
Feishu cards reject messages that contain more than one table element
(API error 11310: card table number over limit). The helper splits a flat
list of card elements into groups so that each group contains at most one
table, allowing nanobot to send multiple cards instead of failing.
"""
# Check optional Feishu dependencies before running tests
try:
from nanobot.channels import feishu
FEISHU_AVAILABLE = getattr(feishu, "FEISHU_AVAILABLE", False)
except ImportError:
FEISHU_AVAILABLE = False
if not FEISHU_AVAILABLE:
import pytest
pytest.skip("Feishu dependencies not installed (lark-oapi)", allow_module_level=True)
from nanobot.channels.feishu.runtime import FeishuChannel
def _md(text: str) -> dict:
return {"tag": "markdown", "content": text}
def _table() -> dict:
return {
"tag": "table",
"columns": [{"tag": "column", "name": "c0", "display_name": "A", "width": "auto"}],
"rows": [{"c0": "v"}],
"page_size": 2,
}
split = FeishuChannel._split_elements_by_table_limit
def test_empty_list_returns_single_empty_group() -> None:
assert split([]) == [[]]
def test_no_tables_returns_single_group() -> None:
els = [_md("hello"), _md("world")]
result = split(els)
assert result == [els]
def test_single_table_stays_in_one_group() -> None:
els = [_md("intro"), _table(), _md("outro")]
result = split(els)
assert len(result) == 1
assert result[0] == els
def test_two_tables_split_into_two_groups() -> None:
# Use different row values so the two tables are not equal
t1 = {
"tag": "table",
"columns": [{"tag": "column", "name": "c0", "display_name": "A", "width": "auto"}],
"rows": [{"c0": "table-one"}],
"page_size": 2,
}
t2 = {
"tag": "table",
"columns": [{"tag": "column", "name": "c0", "display_name": "B", "width": "auto"}],
"rows": [{"c0": "table-two"}],
"page_size": 2,
}
els = [_md("before"), t1, _md("between"), t2, _md("after")]
result = split(els)
assert len(result) == 2
# First group: text before table-1 + table-1
assert t1 in result[0]
assert t2 not in result[0]
# Second group: text between tables + table-2 + text after
assert t2 in result[1]
assert t1 not in result[1]
def test_three_tables_split_into_three_groups() -> None:
tables = [
{"tag": "table", "columns": [], "rows": [{"c0": f"t{i}"}], "page_size": 1}
for i in range(3)
]
els = tables[:]
result = split(els)
assert len(result) == 3
for i, group in enumerate(result):
assert tables[i] in group
def test_leading_markdown_stays_with_first_table() -> None:
intro = _md("intro")
t = _table()
result = split([intro, t])
assert len(result) == 1
assert result[0] == [intro, t]
def test_trailing_markdown_after_second_table() -> None:
t1, t2 = _table(), _table()
tail = _md("end")
result = split([t1, t2, tail])
assert len(result) == 2
assert result[1] == [t2, tail]
def test_non_table_elements_before_first_table_kept_in_first_group() -> None:
head = _md("head")
t1, t2 = _table(), _table()
result = split([head, t1, t2])
# head + t1 in group 0; t2 in group 1
assert result[0] == [head, t1]
assert result[1] == [t2]
@@ -0,0 +1,214 @@
"""Tests for FeishuChannel tool hint formatting."""
import json
from unittest.mock import MagicMock, patch
import pytest
from pytest import mark
# Check optional Feishu dependencies before running tests
try:
from nanobot.channels import feishu
FEISHU_AVAILABLE = getattr(feishu, "FEISHU_AVAILABLE", False)
except ImportError:
FEISHU_AVAILABLE = False
if not FEISHU_AVAILABLE:
pytest.skip("Feishu dependencies not installed (lark-oapi)", allow_module_level=True)
from nanobot.bus.events import OutboundMessage
from nanobot.bus.outbound_events import ProgressEvent
from nanobot.channels.feishu.runtime import FeishuChannel
@pytest.fixture
def mock_feishu_channel():
"""Create a FeishuChannel with mocked client."""
config = MagicMock()
config.app_id = "test_app_id"
config.app_secret = "test_app_secret"
config.encrypt_key = None
config.verification_token = None
config.tool_hint_prefix = "\U0001f527" # 🔧
bus = MagicMock()
channel = FeishuChannel(config, bus)
channel._client = MagicMock()
return channel
def _get_tool_hint_card(mock_send):
"""Extract the interactive card from _send_message_sync calls."""
call_args = mock_send.call_args[0]
_, _, msg_type, content = call_args
assert msg_type == "interactive"
return json.loads(content)
@mark.asyncio
async def test_tool_hint_sends_interactive_card(mock_feishu_channel):
"""Tool hint without active buffer sends an interactive card with 🔧 style."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
content='web_search("test query")',
event=ProgressEvent(tool_hint=True),
)
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
assert mock_send.call_count == 1
card = _get_tool_hint_card(mock_send)
assert card["config"]["wide_screen_mode"] is True
md = card["elements"][0]["content"]
assert "\U0001f527" in md
assert "web_search" in md
@mark.asyncio
async def test_tool_hint_empty_content_does_not_send(mock_feishu_channel):
"""Empty tool hint messages should not be sent."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
content=" ", # whitespace only
event=ProgressEvent(tool_hint=True),
)
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
mock_send.assert_not_called()
@mark.asyncio
async def test_tool_hint_without_metadata_sends_as_normal(mock_feishu_channel):
"""Regular messages without _tool_hint should use normal formatting."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
content="Hello, world!",
metadata={}
)
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
assert mock_send.call_count == 1
call_args = mock_send.call_args[0]
_, _, msg_type, content = call_args
assert msg_type == "text"
assert json.loads(content) == {"text": "Hello, world!"}
@mark.asyncio
async def test_tool_hint_multiple_tools_in_one_message(mock_feishu_channel):
"""Multiple tool calls should each get the 🔧 prefix."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
content='web_search("query"), read_file("/path/to/file")',
event=ProgressEvent(tool_hint=True),
)
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert "web_search" in md
assert "read_file" in md
assert "\U0001f527" in md
@mark.asyncio
async def test_tool_hint_new_format_basic(mock_feishu_channel):
"""New format hints (read path, grep "pattern") should parse correctly."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
content='read src/main.py, grep "TODO"',
event=ProgressEvent(tool_hint=True),
)
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert "read src/main.py" in md
assert 'grep "TODO"' in md
@mark.asyncio
async def test_tool_hint_new_format_with_comma_in_quotes(mock_feishu_channel):
"""Commas inside quoted arguments must not cause incorrect line splits."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
content='grep "hello, world", $ echo test',
event=ProgressEvent(tool_hint=True),
)
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert 'grep "hello, world"' in md
assert "$ echo test" in md
@mark.asyncio
async def test_tool_hint_new_format_with_folding(mock_feishu_channel):
"""Folded calls (× N) should display correctly."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
content='read path × 3, grep "pattern"',
event=ProgressEvent(tool_hint=True),
)
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert "\u00d7 3" in md
assert 'grep "pattern"' in md
@mark.asyncio
async def test_tool_hint_new_format_mcp(mock_feishu_channel):
"""MCP tool format (server::tool) should parse correctly."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
content='4_5v::analyze_image("photo.jpg")',
event=ProgressEvent(tool_hint=True),
)
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert "4_5v::analyze_image" in md
@mark.asyncio
async def test_tool_hint_keeps_commas_inside_arguments(mock_feishu_channel):
"""Commas inside a single tool argument must not be split onto a new line."""
msg = OutboundMessage(
channel="feishu",
chat_id="oc_123456",
content='web_search("foo, bar"), read_file("/path/to/file")',
event=ProgressEvent(tool_hint=True),
)
with patch.object(mock_feishu_channel, '_send_message_sync') as mock_send:
await mock_feishu_channel.send(msg)
card = _get_tool_hint_card(mock_send)
md = card["elements"][0]["content"]
assert 'web_search("foo, bar")' in md
assert 'read_file("/path/to/file")' in md
@@ -0,0 +1,116 @@
from __future__ import annotations
import asyncio
import threading
from typing import Any
from nanobot.channels.feishu.websocket import FeishuWsRunner
class _CleanCloseError(Exception):
pass
class _SdkLikeClient:
"""Model the lark SDK's detached receive task and reconnect behavior."""
def __init__(self) -> None:
self._auto_reconnect = True
self.connected = asyncio.Event()
self.reconnected = asyncio.Event()
self.receive_errors = 0
self.reconnects = 0
self.disconnects = 0
self._receiving = False
self._receive_events: asyncio.Queue[Exception] = asyncio.Queue()
async def _connect(self) -> None:
self.connected.set()
asyncio.create_task(self._receive_message_loop())
async def _receive_message_loop(self) -> None:
try:
self._receiving = True
error = await self._receive_events.get()
self._receiving = False
raise error
except asyncio.CancelledError:
self._receiving = False
raise
except Exception:
self.receive_errors += 1
await self._disconnect()
if self._auto_reconnect:
self.reconnects += 1
await self._connect()
self.reconnected.set()
async def _disconnect(self) -> None:
self.disconnects += 1
if self._receiving:
await self._receive_events.put(_CleanCloseError("1000 OK"))
async def _ping_loop(self) -> None:
await asyncio.Event().wait()
def test_concurrent_loop_initialization_starts_one_thread(monkeypatch) -> None:
runner = FeishuWsRunner()
created_loops: list[asyncio.AbstractEventLoop] = []
release_start = threading.Event()
def fake_run_loop() -> None:
loop = asyncio.new_event_loop()
created_loops.append(loop)
assert release_start.wait(timeout=2)
runner._loop = loop
runner._ready.set()
monkeypatch.setattr(runner, "_run_loop", fake_run_loop)
loops: list[asyncio.AbstractEventLoop] = []
threads = [threading.Thread(target=lambda: loops.append(runner._ensure_loop())) for _ in range(2)]
for thread in threads:
thread.start()
release_start.set()
for thread in threads:
thread.join(timeout=2)
assert len(created_loops) == 1
assert loops == [created_loops[0], created_loops[0]]
created_loops[0].close()
async def test_stop_cancels_sdk_receive_loop_without_reconnecting() -> None:
runner = FeishuWsRunner()
client = _SdkLikeClient()
original_receive_loop: Any = client._receive_message_loop
await runner._start_client("default", client)
await asyncio.wait_for(client.connected.wait(), timeout=1)
await runner._stop_client("default")
await asyncio.sleep(0)
assert client.receive_errors == 0
assert client.reconnects == 0
assert client._auto_reconnect is True
assert client._receive_message_loop == original_receive_loop
async def test_network_failure_keeps_sdk_auto_reconnect_behavior() -> None:
runner = FeishuWsRunner()
client = _SdkLikeClient()
await runner._start_client("default", client)
await asyncio.wait_for(client.connected.wait(), timeout=1)
await client._receive_events.put(RuntimeError("network dropped"))
await asyncio.wait_for(client.reconnected.wait(), timeout=1)
assert client.receive_errors == 1
assert client.reconnects == 1
assert client._auto_reconnect is True
await runner._stop_client("default")
await asyncio.sleep(0)
assert client.receive_errors == 1
assert client.reconnects == 1
+34
View File
@@ -0,0 +1,34 @@
"""Feishu/Lark setup validation owned by the channel package."""
from typing import Any
from nanobot.channels.contracts import ChannelValidationContext
from nanobot.channels.validation import check, payload, required_checks, string_value
def validate(values: dict[str, Any], _context: ChannelValidationContext) -> dict[str, Any]:
checks, missing = required_checks("feishu", values)
display_name = string_value(values.get("displayName") or values.get("name"))
avatar_url = string_value(values.get("avatarUrl"))
app_id = string_value(values.get("appId"))
if app_id.startswith(("cli_", "oapi_")):
checks.append(check("app_id", "App ID", "pass", "A Feishu/Lark App ID is saved."))
elif app_id:
checks.append(
check(
"app_id",
"App ID",
"warn",
"App ID is saved, but it does not look like a standard Feishu App ID.",
)
)
status = "connected" if not missing else "needs_setup"
identity = {
"name": display_name or "Feishu assistant",
"avatar_url": avatar_url or None,
"account": app_id,
}
return payload("feishu", status, checks, identity=identity, missing_fields=missing)
__all__ = ["validate"]
+182
View File
@@ -0,0 +1,182 @@
"""Shared Feishu/Lark WebSocket runtime.
The official lark_oapi websocket client stores an asyncio loop in a module-level
variable. Running one blocking ``Client.start()`` per assistant would make
multiple Feishu instances fragile, so this module centralizes the loop patch and
starts each client through the SDK's async primitives on one dedicated loop.
"""
from __future__ import annotations
import asyncio
import threading
from collections.abc import Awaitable, Callable
from contextlib import suppress
from dataclasses import dataclass, field
from typing import Any, Protocol
from loguru import logger
class _LarkWsClient(Protocol):
"""Private SDK surface isolated behind the Feishu runtime adapter."""
_auto_reconnect: bool
_receive_message_loop: Callable[[], Awaitable[None]]
async def _connect(self) -> None: ...
async def _disconnect(self) -> None: ...
async def _ping_loop(self) -> None: ...
@dataclass
class _ClientRuntime:
client: _LarkWsClient
stop_event: asyncio.Event
task: asyncio.Task[Any] | None
receive_loop: Callable[[], Awaitable[None]]
auto_reconnect: bool
receive_tasks: set[asyncio.Task[Any]] = field(default_factory=set)
class FeishuWsRunner:
"""Run multiple lark_oapi websocket clients on one dedicated event loop."""
def __init__(self) -> None:
self._thread: threading.Thread | None = None
self._loop: asyncio.AbstractEventLoop | None = None
self._ready = threading.Event()
self._lock = threading.Lock()
self._clients: dict[str, _ClientRuntime] = {}
async def start_client(self, key: str, client: _LarkWsClient) -> None:
"""Start or replace one client runtime."""
loop = self._ensure_loop()
await asyncio.wrap_future(
asyncio.run_coroutine_threadsafe(self._start_client(key, client), loop)
)
async def stop_client(self, key: str) -> None:
"""Stop one client runtime if it is active."""
loop = self._loop
if loop is None or loop.is_closed():
return
await asyncio.wrap_future(asyncio.run_coroutine_threadsafe(self._stop_client(key), loop))
def _ensure_loop(self) -> asyncio.AbstractEventLoop:
with self._lock:
if self._loop is not None and not self._loop.is_closed():
return self._loop
self._ready.clear()
self._thread = threading.Thread(target=self._run_loop, name="feishu-ws", daemon=True)
self._thread.start()
if not self._ready.wait(timeout=10) or self._loop is None:
raise RuntimeError("Feishu WebSocket runner did not start")
return self._loop
def _run_loop(self) -> None:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
import lark_oapi.ws.client as lark_ws_client
lark_ws_client.loop = loop
self._loop = loop
self._ready.set()
loop.run_forever()
finally:
with suppress(Exception):
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
async def _start_client(self, key: str, client: _LarkWsClient) -> None:
await self._stop_client(key)
stop_event = asyncio.Event()
receive_loop = client._receive_message_loop
runtime = _ClientRuntime(
client=client,
stop_event=stop_event,
task=None,
receive_loop=receive_loop,
auto_reconnect=client._auto_reconnect,
)
# The SDK discards this task handle. Track it at the adapter boundary so
# an intentional stop can cancel recv() before closing the socket; otherwise
# the SDK logs close code 1000 as an error and starts an unwanted reconnect.
async def tracked_receive_loop() -> None:
if stop_event.is_set():
return
task = asyncio.current_task()
if task is not None:
runtime.receive_tasks.add(task)
try:
await receive_loop()
finally:
if task is not None:
runtime.receive_tasks.discard(task)
client._receive_message_loop = tracked_receive_loop
runtime.task = asyncio.create_task(self._client_main(key, client, stop_event))
self._clients[key] = runtime
async def _stop_client(self, key: str) -> None:
runtime = self._clients.pop(key, None)
if runtime is None:
return
runtime.stop_event.set()
runtime.client._auto_reconnect = False
try:
receive_tasks = tuple(runtime.receive_tasks)
for task in receive_tasks:
task.cancel()
if receive_tasks:
await asyncio.gather(*receive_tasks, return_exceptions=True)
if runtime.task is not None:
runtime.task.cancel()
with suppress(asyncio.CancelledError):
await runtime.task
with suppress(Exception):
await runtime.client._disconnect()
finally:
runtime.client._receive_message_loop = runtime.receive_loop
runtime.client._auto_reconnect = runtime.auto_reconnect
async def _client_main(
self, key: str, client: _LarkWsClient, stop_event: asyncio.Event
) -> None:
ping_task: asyncio.Task | None = None
while not stop_event.is_set():
try:
await client._connect()
ping_task = asyncio.create_task(client._ping_loop())
await stop_event.wait()
except asyncio.CancelledError:
raise
except Exception as exc:
logger.warning("Feishu WebSocket client '{}' failed: {}", key, exc)
with suppress(Exception):
await client._disconnect()
if not stop_event.is_set():
await asyncio.sleep(5)
finally:
if ping_task is not None:
ping_task.cancel()
with suppress(asyncio.CancelledError):
await ping_task
with suppress(Exception):
await client._disconnect()
_RUNNER: FeishuWsRunner | None = None
def get_feishu_ws_runner() -> FeishuWsRunner:
"""Return the process-wide Feishu WebSocket runner."""
global _RUNNER
if _RUNNER is None:
_RUNNER = FeishuWsRunner()
return _RUNNER
@@ -0,0 +1,182 @@
import { useState } from "react";
import { Loader2, RotateCcw } from "lucide-react";
import { useTranslation } from "react-i18next";
import {
channelTranslator,
type ChannelTranslator,
} from "@/channel-plugins/i18n";
import type { ChannelPluginPanelProps } from "@/channel-plugins/types";
import { ChannelInstancesPanel } from "@/components/settings/channels/ChannelInstancesPanel";
import { Button } from "@/components/ui/button";
import { enableNanobotFeature } from "@/lib/api";
import type {
NanobotChannelInstanceInfo,
NanobotFeatureInfo,
NanobotFeaturesPayload,
} from "@/lib/types";
import { FeishuConnectFlow } from "./FeishuConnectFlow";
export function FeishuAssistantsPanel({
token,
feature,
showBrandLogos,
chatAppsDocsUrl,
onFeaturesUpdate,
}: ChannelPluginPanelProps) {
const { t } = useTranslation();
const tx = channelTranslator(t, "feishu");
const instances = feature.instances?.length
? feature.instances
: [defaultFeishuInstance(feature)];
return (
<ChannelInstancesPanel
token={token}
feature={feature}
showBrandLogos={showBrandLogos}
chatAppsDocsUrl={chatAppsDocsUrl}
instances={instances}
onFeaturesUpdate={onFeaturesUpdate}
customization={{
countLabel: (count) => feishuAssistantCountLabel(count, tx),
toggleAriaLabel: (instance) => tx("custom.toggleAssistant", "{{name}} assistant", {
name: instanceDisplayName(instance),
}),
configuredLabel: tx("custom.configured", "Connected"),
needsSetupLabel: tx("custom.needsSetup", "Needs authorization"),
renderInstanceSummary: (instance) => (
maskFeishuAppId(instance.config_values?.["channels.feishu.appId"])
|| tx("custom.noAppId", "No App ID")
),
renderInstanceAction: (instance) => (
<FeishuInstanceAction
key={instance.id}
token={token}
instance={instance}
onFeaturesUpdate={onFeaturesUpdate}
/>
),
footer: (
<div className="mt-4 overflow-hidden rounded-[16px] border border-border/70 bg-background px-4 py-4">
<div className="text-[13px] font-semibold text-foreground">
{tx("custom.createAnother", "Create another assistant")}
</div>
<p className="mt-1 text-[12.5px] leading-5 text-muted-foreground">
{tx(
"custom.createHint",
"Create a separate Feishu bot for another team, space, or workflow.",
)}
</p>
<FeishuConnectFlow
token={token}
instanceId="default"
mode="create"
idleLabel={tx("custom.createAssistant", "Create assistant")}
onFeaturesUpdate={onFeaturesUpdate}
/>
</div>
),
}}
/>
);
}
function FeishuInstanceAction({
token,
instance,
onFeaturesUpdate,
}: {
token: string;
instance: NanobotChannelInstanceInfo;
onFeaturesUpdate: (payload: NanobotFeaturesPayload) => void;
}) {
const { t } = useTranslation();
const tx = channelTranslator(t, "feishu");
const [busy, setBusy] = useState(false);
const [error, setError] = useState<string | null>(null);
if (!instance.configured) {
return (
<FeishuConnectFlow
token={token}
instanceId={instance.id}
mode="replace"
idleLabel={t("settings.channels.connect", { defaultValue: "Connect" })}
onFeaturesUpdate={onFeaturesUpdate}
/>
);
}
const reconnect = async () => {
setBusy(true);
setError(null);
try {
onFeaturesUpdate(
await enableNanobotFeature(token, "feishu", { instanceId: instance.id }),
);
} catch (err) {
setError((err as Error).message);
} finally {
setBusy(false);
}
};
return (
<>
<div className="mt-3 flex justify-end">
<Button
type="button"
size="sm"
variant="outline"
className="h-8 rounded-full border-border/65 bg-background/80 px-3 text-[12px] font-semibold hover:bg-muted/70"
onClick={() => void reconnect()}
disabled={busy || !instance.enabled}
>
{busy ? (
<Loader2 className="mr-1.5 h-3.5 w-3.5 animate-spin" aria-hidden />
) : (
<RotateCcw className="mr-1.5 h-3.5 w-3.5" aria-hidden />
)}
{tx("custom.reconnect", "Reconnect")}
</Button>
</div>
{error ? (
<div className="mt-3 rounded-[12px] border border-destructive/20 px-3 py-2 text-[12px] leading-5 text-destructive">
{error}
</div>
) : null}
</>
);
}
function defaultFeishuInstance(feature: NanobotFeatureInfo): NanobotChannelInstanceInfo {
return {
id: "default",
name: "nanobot",
enabled: feature.enabled,
configured: Boolean(feature.configured),
config_values: feature.config_values ?? {},
configured_fields: feature.configured_fields ?? [],
};
}
function feishuAssistantCountLabel(
count: number,
tx: ChannelTranslator,
): string {
if (count === 0) return tx("custom.countNone", "No assistant connected");
if (count === 1) return tx("custom.countOne", "1 assistant connected");
return tx("custom.countMany", "{{count}} assistants connected", { count });
}
function instanceDisplayName(instance: NanobotChannelInstanceInfo): string {
return instance.display_name?.trim() || instance.name.trim() || instance.id;
}
function maskFeishuAppId(appId: string | undefined): string {
if (!appId) return "";
if (appId.length <= 10) return appId;
return `${appId.slice(0, 7)}...${appId.slice(-4)}`;
}
@@ -0,0 +1,48 @@
import { useTranslation } from "react-i18next";
import { channelTranslator } from "@/channel-plugins/i18n";
import { ChannelQrConnectFlow } from "@/components/settings/channels/ChannelQrConnectFlow";
import type { NanobotFeaturesPayload } from "@/lib/types";
export function FeishuConnectFlow({
token,
instanceId = "default",
mode = "replace",
idleLabel,
connectRequestId,
onFeaturesUpdate,
}: {
token: string;
instanceId?: string;
mode?: "replace" | "create";
idleLabel?: string;
connectRequestId?: number;
onFeaturesUpdate: (payload: NanobotFeaturesPayload) => void;
}) {
const { t } = useTranslation();
const tx = channelTranslator(t, "feishu");
return (
<ChannelQrConnectFlow
token={token}
channelName="feishu"
startOptions={{ domain: "feishu", instanceId, mode }}
idleLabel={idleLabel}
connectRequestId={connectRequestId}
onFeaturesUpdate={onFeaturesUpdate}
labels={{
qrAlt: tx("custom.qrAlt", "Feishu connection QR code"),
scanTitle: tx("custom.scanTitle", "Scan with Feishu"),
scanDescription: tx(
"custom.scanDescription",
"Use Feishu or Lark on your phone to scan this code. nanobot will finish setup automatically after authorization.",
),
waiting: tx("custom.waiting", "Waiting for authorization..."),
connected: tx("custom.connected", "Feishu is connected."),
stopped: tx("custom.stopped", "Connection stopped."),
connecting: tx("custom.connecting", "Connecting..."),
scanAgain: t("settings.channels.scanAgain", { defaultValue: "Scan again" }),
connect: t("settings.channels.connect", { defaultValue: "Connect" }),
}}
/>
);
}
+33
View File
@@ -0,0 +1,33 @@
import type { ChannelUiContribution } from "@/channel-plugins/types";
import { chatAppGuideUrl } from "@/components/settings/channels/catalog";
import { FeishuAssistantsPanel } from "./FeishuAssistantsPanel";
export default {
Panel: FeishuAssistantsPanel,
aliases: {
lark: {
displayName: "Lark",
initials: "LK",
logoUrl: "https://www.larksuite.com/favicon.ico",
},
},
presentation: {
displayName: "Feishu",
initials: "FS",
color: "#3370FF",
logoUrl: "https://www.feishu.cn/favicon.ico",
setup: {
mode: "connect",
command: "nanobot channels login feishu",
docsUrl: chatAppGuideUrl("feishu"),
manualFields: [
{ key: "channels.feishu.appId" },
{ key: "channels.feishu.appSecret" },
{ key: "channels.feishu.domain" },
{ key: "channels.feishu.groupPolicy" },
{ key: "channels.feishu.allowFrom" },
],
},
},
} satisfies ChannelUiContribution;
@@ -0,0 +1,73 @@
{
"description": "Use nanobot from Feishu chats and groups.",
"requirements": "Feishu app credentials, event subscription, gateway",
"setup": {
"primaryAction": "Connect with Feishu",
"docsLabel": "Open Feishu setup",
"officialLabel": "Open Feishu console",
"tryIt": "Send a DM or mention the Feishu assistant in a group.",
"summary": "Connect creates or links a Feishu app by QR code, then saves the app credentials for nanobot.",
"steps": [
"Click Connect and scan the QR code with Feishu or Lark on your phone.",
"Approve the app connection. nanobot saves the App ID and Secret automatically.",
"Send the bot a direct message or mention it in a Feishu group to test it."
],
"fields": {
"appId": {
"label": "App ID",
"placeholder": "cli_xxx"
},
"appSecret": {
"label": "App Secret",
"placeholder": "Leave blank to keep current secret",
"help": "Paste a new App Secret only when rotating credentials."
},
"domain": {
"label": "Region",
"choices": {
"feishu": "Feishu",
"lark": "Lark"
}
},
"groupPolicy": {
"label": "Group behavior",
"choices": {
"mention": "Mention only",
"open": "All messages",
"allowlist": "Allowlist"
}
},
"allowFrom": {
"label": "Allowed users",
"placeholder": "User IDs, comma separated"
},
"topicIsolation": {
"label": "Topic isolation",
"choices": {
"true": "Separate session for each topic",
"false": "One shared session for the group"
}
}
}
},
"custom": {
"toggleAssistant": "{{name}} assistant",
"configured": "Connected",
"needsSetup": "Needs authorization",
"noAppId": "No App ID",
"createAnother": "Create another assistant",
"createHint": "Create a separate Feishu bot for another team, space, or workflow.",
"createAssistant": "Create assistant",
"reconnect": "Reconnect",
"countNone": "No assistant connected",
"countOne": "1 assistant connected",
"countMany": "{{count}} assistants connected",
"qrAlt": "Feishu connection QR code",
"scanTitle": "Scan with Feishu",
"scanDescription": "Use Feishu or Lark on your phone to scan this code. nanobot will finish setup automatically after authorization.",
"waiting": "Waiting for authorization...",
"connected": "Feishu is connected.",
"stopped": "Connection stopped.",
"connecting": "Connecting..."
}
}
@@ -0,0 +1,73 @@
{
"description": "Usa nanobot en chats y grupos de Feishu.",
"requirements": "Credenciales de Feishu, suscripción a eventos y gateway",
"setup": {
"primaryAction": "Conectar Feishu",
"docsLabel": "Abrir guía de Feishu",
"officialLabel": "Abrir consola de Feishu",
"tryIt": "Envía un DM o menciona al asistente en un grupo.",
"summary": "La conexión crea o vincula una app de Feishu por QR y guarda sus credenciales.",
"steps": [
"Haz clic en Conectar y escanea el QR con Feishu o Lark.",
"Aprueba la conexión. nanobot guarda el App ID y el Secret automáticamente.",
"Envía un DM al bot o menciónalo en un grupo de Feishu."
],
"fields": {
"appId": {
"label": "App ID",
"placeholder": "cli_xxx"
},
"appSecret": {
"label": "App Secret",
"placeholder": "Déjalo vacío para conservar el secreto",
"help": "Pega uno nuevo solo al rotar credenciales."
},
"domain": {
"label": "Región",
"choices": {
"feishu": "Feishu",
"lark": "Lark"
}
},
"groupPolicy": {
"label": "Comportamiento en grupos",
"choices": {
"mention": "Solo menciones",
"open": "Todos los mensajes",
"allowlist": "Lista permitida"
}
},
"allowFrom": {
"label": "Usuarios permitidos",
"placeholder": "ID de usuario separados por comas"
},
"topicIsolation": {
"label": "Aislamiento por tema",
"choices": {
"true": "Una sesión separada por tema",
"false": "Una sesión compartida para el grupo"
}
}
}
},
"custom": {
"toggleAssistant": "Asistente {{name}}",
"configured": "Conectado",
"needsSetup": "Necesita autorización",
"noAppId": "Sin App ID",
"createAnother": "Crear otro asistente",
"createHint": "Crea un bot Feishu independiente para otro equipo o flujo.",
"createAssistant": "Crear asistente",
"reconnect": "Reconectar",
"countNone": "Ningún asistente conectado",
"countOne": "1 asistente conectado",
"countMany": "{{count}} asistentes conectados",
"qrAlt": "Código QR de conexión de Feishu",
"scanTitle": "Escanea con Feishu",
"scanDescription": "Escanea con Feishu o Lark en tu teléfono. nanobot completará la configuración tras la autorización.",
"waiting": "Esperando autorización...",
"connected": "Feishu está conectado.",
"stopped": "Conexión detenida.",
"connecting": "Conectando..."
}
}
@@ -0,0 +1,73 @@
{
"description": "Utilisez nanobot dans les conversations et groupes Feishu.",
"requirements": "Identifiants Feishu, abonnement aux événements et passerelle",
"setup": {
"primaryAction": "Connecter Feishu",
"docsLabel": "Ouvrir le guide Feishu",
"officialLabel": "Ouvrir la console Feishu",
"tryIt": "Envoyez un message privé ou mentionnez lassistant dans un groupe.",
"summary": "La connexion crée ou associe une application Feishu par QR code et enregistre ses identifiants.",
"steps": [
"Cliquez sur Connecter et scannez le QR code avec Feishu ou Lark.",
"Approuvez la connexion. nanobot enregistre automatiquement lApp ID et le Secret.",
"Envoyez un message privé au bot ou mentionnez-le dans un groupe Feishu."
],
"fields": {
"appId": {
"label": "App ID",
"placeholder": "cli_xxx"
},
"appSecret": {
"label": "App Secret",
"placeholder": "Laisser vide pour conserver le secret",
"help": "Collez un nouveau secret uniquement lors dune rotation."
},
"domain": {
"label": "Région",
"choices": {
"feishu": "Feishu",
"lark": "Lark"
}
},
"groupPolicy": {
"label": "Comportement en groupe",
"choices": {
"mention": "Mentions uniquement",
"open": "Tous les messages",
"allowlist": "Liste dautorisation"
}
},
"allowFrom": {
"label": "Utilisateurs autorisés",
"placeholder": "ID utilisateur séparés par des virgules"
},
"topicIsolation": {
"label": "Isolation par sujet",
"choices": {
"true": "Une session séparée par sujet",
"false": "Une session partagée pour le groupe"
}
}
}
},
"custom": {
"toggleAssistant": "Assistant {{name}}",
"configured": "Connecté",
"needsSetup": "Autorisation requise",
"noAppId": "Aucun App ID",
"createAnother": "Créer un autre assistant",
"createHint": "Créez un bot Feishu distinct pour une autre équipe ou un autre flux.",
"createAssistant": "Créer lassistant",
"reconnect": "Reconnecter",
"countNone": "Aucun assistant connecté",
"countOne": "1 assistant connecté",
"countMany": "{{count}} assistants connectés",
"qrAlt": "QR code de connexion Feishu",
"scanTitle": "Scanner avec Feishu",
"scanDescription": "Utilisez Feishu ou Lark sur votre téléphone pour scanner ce code. nanobot terminera la configuration après autorisation.",
"waiting": "En attente dautorisation...",
"connected": "Feishu est connecté.",
"stopped": "Connexion arrêtée.",
"connecting": "Connexion..."
}
}
@@ -0,0 +1,73 @@
{
"description": "Gunakan nanobot dari chat dan grup Feishu.",
"requirements": "Kredensial Feishu, langganan event, dan gateway",
"setup": {
"primaryAction": "Hubungkan Feishu",
"docsLabel": "Buka panduan Feishu",
"officialLabel": "Buka konsol Feishu",
"tryIt": "Kirim DM atau sebut asisten di grup.",
"summary": "Koneksi membuat atau menautkan aplikasi Feishu lewat QR dan menyimpan kredensialnya.",
"steps": [
"Klik Hubungkan dan pindai QR dengan Feishu atau Lark.",
"Setujui koneksi. nanobot menyimpan App ID dan Secret otomatis.",
"Kirim DM ke bot atau sebut di grup Feishu."
],
"fields": {
"appId": {
"label": "App ID",
"placeholder": "cli_xxx"
},
"appSecret": {
"label": "App Secret",
"placeholder": "Kosongkan untuk mempertahankan secret",
"help": "Tempel secret baru hanya saat rotasi kredensial."
},
"domain": {
"label": "Wilayah",
"choices": {
"feishu": "Feishu",
"lark": "Lark"
}
},
"groupPolicy": {
"label": "Perilaku grup",
"choices": {
"mention": "Hanya sebutan",
"open": "Semua pesan",
"allowlist": "Daftar izin"
}
},
"allowFrom": {
"label": "Pengguna yang diizinkan",
"placeholder": "ID pengguna, dipisahkan koma"
},
"topicIsolation": {
"label": "Isolasi topik",
"choices": {
"true": "Sesi terpisah untuk setiap topik",
"false": "Satu sesi bersama untuk grup"
}
}
}
},
"custom": {
"toggleAssistant": "Asisten {{name}}",
"configured": "Terhubung",
"needsSetup": "Perlu otorisasi",
"noAppId": "Tidak ada App ID",
"createAnother": "Buat asisten lain",
"createHint": "Buat bot Feishu terpisah untuk tim atau alur kerja lain.",
"createAssistant": "Buat asisten",
"reconnect": "Hubungkan ulang",
"countNone": "Belum ada asisten terhubung",
"countOne": "1 asisten terhubung",
"countMany": "{{count}} asisten terhubung",
"qrAlt": "Kode QR koneksi Feishu",
"scanTitle": "Pindai dengan Feishu",
"scanDescription": "Pindai dengan Feishu atau Lark di ponsel. nanobot akan menyelesaikan setup setelah otorisasi.",
"waiting": "Menunggu otorisasi...",
"connected": "Feishu sudah terhubung.",
"stopped": "Koneksi dihentikan.",
"connecting": "Menghubungkan..."
}
}
@@ -0,0 +1,73 @@
{
"description": "Feishu のチャットとグループから nanobot を利用します。",
"requirements": "Feishu アプリ認証情報、イベント購読、ゲートウェイ",
"setup": {
"primaryAction": "Feishu に接続",
"docsLabel": "Feishu 設定ガイドを開く",
"officialLabel": "Feishu コンソールを開く",
"tryIt": "DM を送るか、グループで Feishu アシスタントをメンションします。",
"summary": "QR コードで Feishu アプリを作成または連携し、認証情報を自動保存します。",
"steps": [
"接続をクリックし、スマートフォンの Feishu または Lark で QR コードを読み取ります。",
"アプリ接続を承認すると、nanobot が App ID と Secret を保存します。",
"ボットに DM を送るか Feishu グループでメンションします。"
],
"fields": {
"appId": {
"label": "App ID",
"placeholder": "cli_xxx"
},
"appSecret": {
"label": "App Secret",
"placeholder": "現在のシークレットを保持するには空欄",
"help": "認証情報を更新するときだけ新しい App Secret を貼り付けます。"
},
"domain": {
"label": "地域",
"choices": {
"feishu": "Feishu",
"lark": "Lark"
}
},
"groupPolicy": {
"label": "グループでの動作",
"choices": {
"mention": "メンションのみ",
"open": "すべてのメッセージ",
"allowlist": "許可リスト"
}
},
"allowFrom": {
"label": "許可するユーザー",
"placeholder": "ユーザー ID(カンマ区切り)"
},
"topicIsolation": {
"label": "トピック分離",
"choices": {
"true": "トピックごとにセッションを分離",
"false": "グループでセッションを共有"
}
}
}
},
"custom": {
"toggleAssistant": "{{name}} アシスタント",
"configured": "接続済み",
"needsSetup": "認可が必要",
"noAppId": "App ID なし",
"createAnother": "別のアシスタントを作成",
"createHint": "別のチームやワークフロー用に独立した Feishu ボットを作成します。",
"createAssistant": "アシスタントを作成",
"reconnect": "再接続",
"countNone": "接続済みアシスタントなし",
"countOne": "1 個のアシスタントを接続中",
"countMany": "{{count}} 個のアシスタントを接続中",
"qrAlt": "Feishu 接続 QR コード",
"scanTitle": "Feishu でスキャン",
"scanDescription": "スマートフォンの Feishu または Lark でスキャンしてください。認可後に nanobot が設定を完了します。",
"waiting": "認可を待っています...",
"connected": "Feishu に接続しました。",
"stopped": "接続を停止しました。",
"connecting": "接続中..."
}
}
@@ -0,0 +1,73 @@
{
"description": "Feishu 채팅과 그룹에서 nanobot을 사용합니다.",
"requirements": "Feishu 앱 자격 증명, 이벤트 구독 및 게이트웨이",
"setup": {
"primaryAction": "Feishu 연결",
"docsLabel": "Feishu 설정 가이드 열기",
"officialLabel": "Feishu 콘솔 열기",
"tryIt": "DM을 보내거나 그룹에서 Feishu 어시스턴트를 멘션하세요.",
"summary": "QR 코드로 Feishu 앱을 만들거나 연결하고 자격 증명을 자동 저장합니다.",
"steps": [
"연결을 클릭하고 휴대폰의 Feishu 또는 Lark로 QR 코드를 스캔하세요.",
"앱 연결을 승인하면 nanobot이 App ID와 Secret을 자동 저장합니다.",
"봇에 DM을 보내거나 Feishu 그룹에서 멘션하세요."
],
"fields": {
"appId": {
"label": "App ID",
"placeholder": "cli_xxx"
},
"appSecret": {
"label": "App Secret",
"placeholder": "현재 Secret을 유지하려면 비워 두세요",
"help": "자격 증명을 교체할 때만 새 App Secret을 붙여 넣으세요."
},
"domain": {
"label": "지역",
"choices": {
"feishu": "Feishu",
"lark": "Lark"
}
},
"groupPolicy": {
"label": "그룹 동작",
"choices": {
"mention": "멘션만",
"open": "모든 메시지",
"allowlist": "허용 목록"
}
},
"allowFrom": {
"label": "허용된 사용자",
"placeholder": "사용자 ID, 쉼표로 구분"
},
"topicIsolation": {
"label": "주제 격리",
"choices": {
"true": "주제별로 세션 분리",
"false": "그룹에서 하나의 세션 공유"
}
}
}
},
"custom": {
"toggleAssistant": "{{name}} 어시스턴트",
"configured": "연결됨",
"needsSetup": "인증 필요",
"noAppId": "App ID 없음",
"createAnother": "다른 어시스턴트 만들기",
"createHint": "다른 팀이나 워크플로를 위한 별도 Feishu 봇을 만드세요.",
"createAssistant": "어시스턴트 만들기",
"reconnect": "다시 연결",
"countNone": "연결된 어시스턴트 없음",
"countOne": "어시스턴트 1개 연결됨",
"countMany": "어시스턴트 {{count}}개 연결됨",
"qrAlt": "Feishu 연결 QR 코드",
"scanTitle": "Feishu로 스캔",
"scanDescription": "휴대폰의 Feishu 또는 Lark로 스캔하세요. 승인 후 nanobot이 설정을 완료합니다.",
"waiting": "승인을 기다리는 중...",
"connected": "Feishu가 연결되었습니다.",
"stopped": "연결이 중지되었습니다.",
"connecting": "연결 중..."
}
}
@@ -0,0 +1,73 @@
{
"description": "Use o nanobot em conversas e grupos do Feishu.",
"requirements": "Credenciais do Feishu, assinatura de eventos e gateway",
"setup": {
"primaryAction": "Conectar Feishu",
"docsLabel": "Abrir guia do Feishu",
"officialLabel": "Abrir console do Feishu",
"tryIt": "Envie uma DM ou mencione o assistente em um grupo.",
"summary": "A conexão cria ou vincula um app Feishu por QR e salva as credenciais.",
"steps": [
"Clique em Conectar e escaneie o QR com Feishu ou Lark.",
"Aprove a conexão. O nanobot salva App ID e Secret automaticamente.",
"Envie uma DM ao bot ou mencione-o em um grupo Feishu."
],
"fields": {
"appId": {
"label": "App ID",
"placeholder": "cli_xxx"
},
"appSecret": {
"label": "App Secret",
"placeholder": "Deixe vazio para manter o segredo",
"help": "Cole um novo apenas ao trocar credenciais."
},
"domain": {
"label": "Região",
"choices": {
"feishu": "Feishu",
"lark": "Lark"
}
},
"groupPolicy": {
"label": "Comportamento em grupos",
"choices": {
"mention": "Somente menções",
"open": "Todas as mensagens",
"allowlist": "Lista de permissão"
}
},
"allowFrom": {
"label": "Usuários permitidos",
"placeholder": "IDs de usuário separados por vírgulas"
},
"topicIsolation": {
"label": "Isolamento por tópico",
"choices": {
"true": "Uma sessão separada por tópico",
"false": "Uma sessão compartilhada para o grupo"
}
}
}
},
"custom": {
"toggleAssistant": "Assistente {{name}}",
"configured": "Conectado",
"needsSetup": "Precisa de autorização",
"noAppId": "Sem App ID",
"createAnother": "Criar outro assistente",
"createHint": "Crie um bot Feishu separado para outra equipe ou fluxo.",
"createAssistant": "Criar assistente",
"reconnect": "Reconectar",
"countNone": "Nenhum assistente conectado",
"countOne": "1 assistente conectado",
"countMany": "{{count}} assistentes conectados",
"qrAlt": "QR code de conexão do Feishu",
"scanTitle": "Escaneie com o Feishu",
"scanDescription": "Escaneie com Feishu ou Lark no celular. O nanobot concluirá a configuração após a autorização.",
"waiting": "Aguardando autorização...",
"connected": "Feishu está conectado.",
"stopped": "Conexão interrompida.",
"connecting": "Conectando..."
}
}
@@ -0,0 +1,73 @@
{
"description": "Sử dụng nanobot trong cuộc trò chuyện và nhóm Feishu.",
"requirements": "Thông tin xác thực Feishu, đăng ký sự kiện và gateway",
"setup": {
"primaryAction": "Kết nối Feishu",
"docsLabel": "Mở hướng dẫn Feishu",
"officialLabel": "Mở bảng điều khiển Feishu",
"tryIt": "Gửi tin nhắn riêng hoặc nhắc trợ lý trong nhóm.",
"summary": "Kết nối tạo hoặc liên kết ứng dụng Feishu bằng QR và lưu thông tin xác thực.",
"steps": [
"Nhấn Kết nối và quét QR bằng Feishu hoặc Lark.",
"Phê duyệt kết nối. nanobot tự lưu App ID và Secret.",
"Gửi tin nhắn riêng cho bot hoặc nhắc bot trong nhóm Feishu."
],
"fields": {
"appId": {
"label": "App ID",
"placeholder": "cli_xxx"
},
"appSecret": {
"label": "App Secret",
"placeholder": "Để trống để giữ secret hiện tại",
"help": "Chỉ dán secret mới khi xoay vòng thông tin xác thực."
},
"domain": {
"label": "Khu vực",
"choices": {
"feishu": "Feishu",
"lark": "Lark"
}
},
"groupPolicy": {
"label": "Hành vi trong nhóm",
"choices": {
"mention": "Chỉ khi được nhắc",
"open": "Mọi tin nhắn",
"allowlist": "Danh sách cho phép"
}
},
"allowFrom": {
"label": "Người dùng được phép",
"placeholder": "ID người dùng, phân tách bằng dấu phẩy"
},
"topicIsolation": {
"label": "Tách biệt chủ đề",
"choices": {
"true": "Phiên riêng cho từng chủ đề",
"false": "Dùng chung một phiên cho nhóm"
}
}
}
},
"custom": {
"toggleAssistant": "Trợ lý {{name}}",
"configured": "Đã kết nối",
"needsSetup": "Cần cấp quyền",
"noAppId": "Không có App ID",
"createAnother": "Tạo trợ lý khác",
"createHint": "Tạo bot Feishu riêng cho nhóm hoặc quy trình khác.",
"createAssistant": "Tạo trợ lý",
"reconnect": "Kết nối lại",
"countNone": "Chưa kết nối trợ lý",
"countOne": "Đã kết nối 1 trợ lý",
"countMany": "Đã kết nối {{count}} trợ lý",
"qrAlt": "Mã QR kết nối Feishu",
"scanTitle": "Quét bằng Feishu",
"scanDescription": "Quét bằng Feishu hoặc Lark trên điện thoại. nanobot sẽ hoàn tất cấu hình sau khi cấp quyền.",
"waiting": "Đang chờ cấp quyền...",
"connected": "Feishu đã kết nối.",
"stopped": "Kết nối đã dừng.",
"connecting": "Đang kết nối..."
}
}
@@ -0,0 +1,74 @@
{
"displayName": "飞书",
"description": "在飞书会话和群组中使用 nanobot。",
"requirements": "飞书应用凭据、事件订阅和网关",
"setup": {
"primaryAction": "连接飞书",
"docsLabel": "打开飞书配置指南",
"officialLabel": "打开飞书开发者后台",
"tryIt": "向飞书助手发送私信,或在群组中提及它。",
"summary": "连接流程会通过二维码创建或关联飞书应用,并自动为 nanobot 保存应用凭据。",
"steps": [
"点击连接,用手机飞书或 Lark 扫描二维码。",
"批准应用连接,nanobot 会自动保存 App ID 和 Secret。",
"向机器人发送私信,或在飞书群中提及它以完成测试。"
],
"fields": {
"appId": {
"label": "App ID",
"placeholder": "cli_xxx"
},
"appSecret": {
"label": "App Secret",
"placeholder": "留空以保留现有密钥",
"help": "仅在轮换凭据时粘贴新的 App Secret。"
},
"domain": {
"label": "区域",
"choices": {
"feishu": "飞书",
"lark": "Lark"
}
},
"groupPolicy": {
"label": "群组行为",
"choices": {
"mention": "仅提及时",
"open": "所有消息",
"allowlist": "白名单"
}
},
"allowFrom": {
"label": "允许的用户",
"placeholder": "用户 ID,用逗号分隔"
},
"topicIsolation": {
"label": "话题隔离",
"choices": {
"true": "每个话题使用独立会话",
"false": "群聊共用一个会话"
}
}
}
},
"custom": {
"toggleAssistant": "{{name}} 助手",
"configured": "已连接",
"needsSetup": "需要授权",
"noAppId": "没有 App ID",
"createAnother": "创建另一个助手",
"createHint": "为其他团队、空间或工作流创建独立的飞书机器人。",
"createAssistant": "创建助手",
"reconnect": "重新连接",
"countNone": "尚未连接助手",
"countOne": "已连接 1 个助手",
"countMany": "已连接 {{count}} 个助手",
"qrAlt": "飞书连接二维码",
"scanTitle": "使用飞书扫码",
"scanDescription": "用手机上的飞书或 Lark 扫描二维码。授权完成后,nanobot 会自动完成配置。",
"waiting": "正在等待授权...",
"connected": "飞书已连接。",
"stopped": "连接已停止。",
"connecting": "正在连接..."
}
}
@@ -0,0 +1,74 @@
{
"displayName": "飛書",
"description": "在飛書對話和群組中使用 nanobot。",
"requirements": "飛書應用程式憑證、事件訂閱和閘道",
"setup": {
"primaryAction": "連接飛書",
"docsLabel": "開啟飛書設定指南",
"officialLabel": "開啟飛書開發者後台",
"tryIt": "向飛書助手傳送私訊,或在群組中提及它。",
"summary": "連接流程會透過二維碼建立或關聯飛書應用程式,並自動為 nanobot 儲存應用程式憑證。",
"steps": [
"點擊連接,用手機飛書或 Lark 掃描二維碼。",
"批准應用程式連接,nanobot 會自動儲存 App ID 和 Secret。",
"向機器人傳送私訊,或在飛書群組中提及它以完成測試。"
],
"fields": {
"appId": {
"label": "App ID",
"placeholder": "cli_xxx"
},
"appSecret": {
"label": "App Secret",
"placeholder": "留空以保留現有密鑰",
"help": "僅在輪換憑證時貼上新的 App Secret。"
},
"domain": {
"label": "區域",
"choices": {
"feishu": "飛書",
"lark": "Lark"
}
},
"groupPolicy": {
"label": "群組行為",
"choices": {
"mention": "僅提及時",
"open": "所有訊息",
"allowlist": "允許清單"
}
},
"allowFrom": {
"label": "允許的使用者",
"placeholder": "使用者 ID,以逗號分隔"
},
"topicIsolation": {
"label": "主題隔離",
"choices": {
"true": "每個主題使用獨立工作階段",
"false": "群組共用一個工作階段"
}
}
}
},
"custom": {
"toggleAssistant": "{{name}} 助手",
"configured": "已連接",
"needsSetup": "需要授權",
"noAppId": "沒有 App ID",
"createAnother": "建立另一個助手",
"createHint": "為其他團隊、空間或工作流程建立獨立的飛書機器人。",
"createAssistant": "建立助手",
"reconnect": "重新連線",
"countNone": "尚未連接助手",
"countOne": "已連接 1 個助手",
"countMany": "已連接 {{count}} 個助手",
"qrAlt": "飛書連線 QR Code",
"scanTitle": "使用飛書掃描",
"scanDescription": "請使用手機上的飛書或 Lark 掃描此 QR Code。完成授權後,nanobot 會自動完成設定。",
"waiting": "正在等待授權…",
"connected": "飛書已連線。",
"stopped": "連線已停止。",
"connecting": "正在連線…"
}
}