refactor(gateway): migrate slack/dingtalk/whatsapp/matrix/feishu/telegram/wecom/email/sms adapters to bundled plugins

Salvage of PR #41284 onto current main. Relocates the last 9 inline messaging
adapters (+ satellites: telegram_network, feishu_comment/_rules/meeting_invite,
wecom_crypto, wecom_callback) from gateway/platforms/ into self-contained
bundled plugins under plugins/platforms/<x>/, discovered via the platform
registry. Strips the per-platform core touchpoints from gateway/run.py,
gateway/config.py, hermes_cli/gateway.py, hermes_cli/setup.py, and
tools/send_message_tool.py.

Carries forward the migration fixes (explicit enabled:false honored,
get_connected_platforms forces discovery, plugin is_connected via
gateway.get_env_value, logs --component gateway matches plugins.platforms.*,
matrix hidden on Windows).

Additionally ports config keys main added since the PR base: the matrix
plugin's _apply_yaml_config now also covers allowed_users,
ignore_user_patterns, process_notices, and session_scope (the inline
gateway/config.py matrix block gained these in the 1340 commits the PR sat
open; they would otherwise have been silently dropped on deletion).

plugins/platforms/feishu/__init__.py

@@ -0,0 +1,3 @@
+from .adapter import register
+
+__all__ = ["register"]

plugins/platforms/feishu/feishu_comment_rules.py

@@ -0,0 +1,429 @@
+"""
+Feishu document comment access-control rules.
+
+3-tier rule resolution: exact doc > wildcard "*" > top-level > code defaults.
+Each field (enabled/policy/allow_from) falls back independently.
+Config: ~/.hermes/feishu_comment_rules.json (mtime-cached, hot-reload).
+Pairing store: ~/.hermes/feishu_comment_pairing.json.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+from hermes_constants import get_hermes_home
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Paths
+# ---------------------------------------------------------------------------
+#
+# Uses the canonical ``get_hermes_home()`` helper (HERMES_HOME-aware and
+# profile-safe). Resolved at import time; this module is lazy-imported by
+# the Feishu comment event handler, which runs long after profile overrides
+# have been applied, so freezing paths here is safe.
+
+RULES_FILE = get_hermes_home() / "feishu_comment_rules.json"
+PAIRING_FILE = get_hermes_home() / "feishu_comment_pairing.json"
+
+# ---------------------------------------------------------------------------
+# Data models
+# ---------------------------------------------------------------------------
+
+_VALID_POLICIES = ("allowlist", "pairing")
+
+
+@dataclass(frozen=True)
+class CommentDocumentRule:
+    """Per-document rule.  ``None`` means 'inherit from lower tier'."""
+    enabled: Optional[bool] = None
+    policy: Optional[str] = None
+    allow_from: Optional[frozenset] = None
+
+
+@dataclass(frozen=True)
+class CommentsConfig:
+    """Top-level comment access config."""
+    enabled: bool = True
+    policy: str = "pairing"
+    allow_from: frozenset = field(default_factory=frozenset)
+    documents: Dict[str, CommentDocumentRule] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class ResolvedCommentRule:
+    """Fully resolved rule after field-by-field fallback."""
+    enabled: bool
+    policy: str
+    allow_from: frozenset
+    match_source: str  # e.g. "exact:docx:xxx" | "wildcard" | "top" | "default"
+
+
+# ---------------------------------------------------------------------------
+# Mtime-cached file loading
+# ---------------------------------------------------------------------------
+
+class _MtimeCache:
+    """Generic mtime-based file cache.  ``stat()`` per access, re-read only on change."""
+
+    def __init__(self, path: Path):
+        self._path = path
+        self._mtime: float = 0.0
+        self._data: Optional[dict] = None
+
+    def load(self) -> dict:
+        try:
+            st = self._path.stat()
+            mtime = st.st_mtime
+        except FileNotFoundError:
+            self._mtime = 0.0
+            self._data = {}
+            return {}
+
+        if mtime == self._mtime and self._data is not None:
+            return self._data
+
+        try:
+            with open(self._path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+            if not isinstance(data, dict):
+                data = {}
+        except (json.JSONDecodeError, OSError):
+            logger.warning("[Feishu-Rules] Failed to read %s, using empty config", self._path)
+            data = {}
+
+        self._mtime = mtime
+        self._data = data
+        return data
+
+
+_rules_cache = _MtimeCache(RULES_FILE)
+_pairing_cache = _MtimeCache(PAIRING_FILE)
+
+
+# ---------------------------------------------------------------------------
+# Config parsing
+# ---------------------------------------------------------------------------
+
+def _parse_frozenset(raw: Any) -> Optional[frozenset]:
+    """Parse a list of strings into a frozenset; return None if key absent."""
+    if raw is None:
+        return None
+    if isinstance(raw, (list, tuple)):
+        return frozenset(str(u).strip() for u in raw if str(u).strip())
+    return None
+
+
+def _parse_document_rule(raw: dict) -> CommentDocumentRule:
+    enabled = raw.get("enabled")
+    if enabled is not None:
+        enabled = bool(enabled)
+    policy = raw.get("policy")
+    if policy is not None:
+        policy = str(policy).strip().lower()
+        if policy not in _VALID_POLICIES:
+            policy = None
+    allow_from = _parse_frozenset(raw.get("allow_from"))
+    return CommentDocumentRule(enabled=enabled, policy=policy, allow_from=allow_from)
+
+
+def load_config() -> CommentsConfig:
+    """Load comment rules from disk (mtime-cached)."""
+    raw = _rules_cache.load()
+    if not raw:
+        return CommentsConfig()
+
+    documents: Dict[str, CommentDocumentRule] = {}
+    raw_docs = raw.get("documents", {})
+    if isinstance(raw_docs, dict):
+        for key, rule_raw in raw_docs.items():
+            if isinstance(rule_raw, dict):
+                documents[str(key)] = _parse_document_rule(rule_raw)
+
+    policy = str(raw.get("policy", "pairing")).strip().lower()
+    if policy not in _VALID_POLICIES:
+        policy = "pairing"
+
+    return CommentsConfig(
+        enabled=raw.get("enabled", True),
+        policy=policy,
+        allow_from=_parse_frozenset(raw.get("allow_from")) or frozenset(),
+        documents=documents,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Rule resolution  (§8.4 field-by-field fallback)
+# ---------------------------------------------------------------------------
+
+def has_wiki_keys(cfg: CommentsConfig) -> bool:
+    """Check if any document rule key starts with 'wiki:'."""
+    return any(k.startswith("wiki:") for k in cfg.documents)
+
+
+def resolve_rule(
+    cfg: CommentsConfig,
+    file_type: str,
+    file_token: str,
+    wiki_token: str = "",
+) -> ResolvedCommentRule:
+    """Resolve effective rule: exact doc → wiki key → wildcard → top-level → defaults."""
+    exact_key = f"{file_type}:{file_token}"
+
+    exact = cfg.documents.get(exact_key)
+    exact_src = f"exact:{exact_key}"
+    if exact is None and wiki_token:
+        wiki_key = f"wiki:{wiki_token}"
+        exact = cfg.documents.get(wiki_key)
+        exact_src = f"exact:{wiki_key}"
+
+    wildcard = cfg.documents.get("*")
+
+    layers = []
+    if exact is not None:
+        layers.append((exact, exact_src))
+    if wildcard is not None:
+        layers.append((wildcard, "wildcard"))
+
+    def _pick(field_name: str):
+        for layer, source in layers:
+            val = getattr(layer, field_name)
+            if val is not None:
+                return val, source
+        return getattr(cfg, field_name), "top"
+
+    enabled, en_src = _pick("enabled")
+    policy, pol_src = _pick("policy")
+    allow_from, _ = _pick("allow_from")
+
+    # match_source = highest-priority tier that contributed any field
+    priority_order = {"exact": 0, "wildcard": 1, "top": 2}
+    best_src = min(
+        [en_src, pol_src],
+        key=lambda s: priority_order.get(s.split(":")[0], 3),
+    )
+
+    return ResolvedCommentRule(
+        enabled=enabled,
+        policy=policy,
+        allow_from=allow_from,
+        match_source=best_src,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Pairing store
+# ---------------------------------------------------------------------------
+
+def _load_pairing_approved() -> set:
+    """Return set of approved user open_ids (mtime-cached)."""
+    data = _pairing_cache.load()
+    approved = data.get("approved", {})
+    if isinstance(approved, dict):
+        return set(approved.keys())
+    if isinstance(approved, list):
+        return {str(u) for u in approved if u}
+    return set()
+
+
+def _save_pairing(data: dict) -> None:
+    PAIRING_FILE.parent.mkdir(parents=True, exist_ok=True)
+    tmp = PAIRING_FILE.with_suffix(".tmp")
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)
+    tmp.replace(PAIRING_FILE)
+    # Invalidate cache so next load picks up change
+    _pairing_cache._mtime = 0.0
+    _pairing_cache._data = None
+
+
+def pairing_add(user_open_id: str) -> bool:
+    """Add a user to the pairing-approved list. Returns True if newly added."""
+    data = _pairing_cache.load()
+    approved = data.get("approved", {})
+    if not isinstance(approved, dict):
+        approved = {}
+    if user_open_id in approved:
+        return False
+    approved[user_open_id] = {"approved_at": time.time()}
+    data["approved"] = approved
+    _save_pairing(data)
+    return True
+
+
+def pairing_remove(user_open_id: str) -> bool:
+    """Remove a user from the pairing-approved list. Returns True if removed."""
+    data = _pairing_cache.load()
+    approved = data.get("approved", {})
+    if not isinstance(approved, dict):
+        return False
+    if user_open_id not in approved:
+        return False
+    del approved[user_open_id]
+    data["approved"] = approved
+    _save_pairing(data)
+    return True
+
+
+def pairing_list() -> Dict[str, Any]:
+    """Return the approved dict  {user_open_id: {approved_at: ...}}."""
+    data = _pairing_cache.load()
+    approved = data.get("approved", {})
+    return dict(approved) if isinstance(approved, dict) else {}
+
+
+# ---------------------------------------------------------------------------
+# Access check  (public API for feishu_comment.py)
+# ---------------------------------------------------------------------------
+
+def is_user_allowed(rule: ResolvedCommentRule, user_open_id: str) -> bool:
+    """Check if user passes the resolved rule's policy gate."""
+    if user_open_id in rule.allow_from:
+        return True
+    if rule.policy == "pairing":
+        return user_open_id in _load_pairing_approved()
+    return False
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+def _print_status() -> None:
+    cfg = load_config()
+    print(f"Rules file: {RULES_FILE}")
+    print(f"  exists: {RULES_FILE.exists()}")
+    print(f"Pairing file: {PAIRING_FILE}")
+    print(f"  exists: {PAIRING_FILE.exists()}")
+    print()
+    print(f"Top-level:")
+    print(f"  enabled:    {cfg.enabled}")
+    print(f"  policy:     {cfg.policy}")
+    print(f"  allow_from: {sorted(cfg.allow_from) if cfg.allow_from else '[]'}")
+    print()
+    if cfg.documents:
+        print(f"Document rules ({len(cfg.documents)}):")
+        for key, rule in sorted(cfg.documents.items()):
+            parts = []
+            if rule.enabled is not None:
+                parts.append(f"enabled={rule.enabled}")
+            if rule.policy is not None:
+                parts.append(f"policy={rule.policy}")
+            if rule.allow_from is not None:
+                parts.append(f"allow_from={sorted(rule.allow_from)}")
+            print(f"  [{key}] {', '.join(parts) if parts else '(empty — inherits all)'}")
+    else:
+        print("Document rules: (none)")
+    print()
+    approved = pairing_list()
+    print(f"Pairing approved ({len(approved)}):")
+    for uid, meta in sorted(approved.items()):
+        ts = meta.get("approved_at", 0)
+        print(f"  {uid}  (approved_at={ts})")
+
+
+def _do_check(doc_key: str, user_open_id: str) -> None:
+    cfg = load_config()
+    parts = doc_key.split(":", 1)
+    if len(parts) != 2:
+        print(f"Error: doc_key must be 'fileType:fileToken', got '{doc_key}'")
+        return
+    file_type, file_token = parts
+    rule = resolve_rule(cfg, file_type, file_token)
+    allowed = is_user_allowed(rule, user_open_id)
+    print(f"Document:     {doc_key}")
+    print(f"User:         {user_open_id}")
+    print(f"Resolved rule:")
+    print(f"  enabled:      {rule.enabled}")
+    print(f"  policy:       {rule.policy}")
+    print(f"  allow_from:   {sorted(rule.allow_from) if rule.allow_from else '[]'}")
+    print(f"  match_source: {rule.match_source}")
+    print(f"Result:       {'ALLOWED' if allowed else 'DENIED'}")
+
+
+def _main() -> int:
+    import sys
+
+    try:
+        from hermes_cli.env_loader import load_hermes_dotenv
+        load_hermes_dotenv()
+    except Exception:
+        pass
+
+    usage = (
+        "Usage: python -m gateway.platforms.feishu_comment_rules <command> [args]\n"
+        "\n"
+        "Commands:\n"
+        "  status                              Show rules config and pairing state\n"
+        "  check <fileType:token> <user>        Simulate access check\n"
+        "  pairing add <user_open_id>           Add user to pairing-approved list\n"
+        "  pairing remove <user_open_id>        Remove user from pairing-approved list\n"
+        "  pairing list                         List pairing-approved users\n"
+        "\n"
+        f"Rules config file: {RULES_FILE}\n"
+        "  Edit this JSON file directly to configure policies and document rules.\n"
+        "  Changes take effect on the next comment event (no restart needed).\n"
+    )
+
+    args = sys.argv[1:]
+    if not args:
+        print(usage)
+        return 1
+
+    cmd = args[0]
+
+    if cmd == "status":
+        _print_status()
+
+    elif cmd == "check":
+        if len(args) < 3:
+            print("Usage: check <fileType:fileToken> <user_open_id>")
+            return 1
+        _do_check(args[1], args[2])
+
+    elif cmd == "pairing":
+        if len(args) < 2:
+            print("Usage: pairing <add|remove|list> [args]")
+            return 1
+        sub = args[1]
+        if sub == "add":
+            if len(args) < 3:
+                print("Usage: pairing add <user_open_id>")
+                return 1
+            if pairing_add(args[2]):
+                print(f"Added: {args[2]}")
+            else:
+                print(f"Already approved: {args[2]}")
+        elif sub == "remove":
+            if len(args) < 3:
+                print("Usage: pairing remove <user_open_id>")
+                return 1
+            if pairing_remove(args[2]):
+                print(f"Removed: {args[2]}")
+            else:
+                print(f"Not in approved list: {args[2]}")
+        elif sub == "list":
+            approved = pairing_list()
+            if not approved:
+                print("(no approved users)")
+            for uid, meta in sorted(approved.items()):
+                print(f"  {uid}  approved_at={meta.get('approved_at', '?')}")
+        else:
+            print(f"Unknown pairing subcommand: {sub}")
+            return 1
+    else:
+        print(f"Unknown command: {cmd}\n")
+        print(usage)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    import sys
+    sys.exit(_main())

plugins/platforms/feishu/feishu_meeting_invite.py

@@ -0,0 +1,212 @@
+"""
+Feishu/Lark meeting-invitation event handling.
+
+Processes ``vc.bot.meeting_invited_v1`` events by converting them into a
+synthetic gateway ``MessageEvent``.  Unlike document comments, the response
+should go back to the inviter through the normal Hermes gateway pipeline, so
+this module does not instantiate an agent directly.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from types import SimpleNamespace
+from typing import Any, Dict, Optional
+
+from gateway.platforms.base import MessageEvent, MessageType
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class MeetingInviteUser:
+    open_id: str = ""
+    user_id: str = ""
+    union_id: str = ""
+    user_name: str = ""
+
+
+@dataclass(frozen=True)
+class MeetingInviteMeeting:
+    id: str = ""
+    topic: str = ""
+    meeting_no: str = ""
+    start_time_ms: int = 0
+    end_time_ms: int = 0
+    host_user: Optional[MeetingInviteUser] = None
+
+
+@dataclass(frozen=True)
+class MeetingInvitedPayload:
+    event_id: str = ""
+    meeting: Optional[MeetingInviteMeeting] = None
+    inviter: Optional[MeetingInviteUser] = None
+    invite_time_s: int = 0
+
+
+def _as_dict(value: Any) -> Dict[str, Any]:
+    """Coerce a lark SDK object / dict / JSON string into a plain dict."""
+    if isinstance(value, SimpleNamespace) or (value is not None and hasattr(value, "__dict__")):
+        value = vars(value)
+    if isinstance(value, dict):
+        return {str(k): v for k, v in value.items()}
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+        except (TypeError, json.JSONDecodeError):
+            return {}
+        return parsed if isinstance(parsed, dict) else {}
+    return {}
+
+
+def _content_payload(container: Dict[str, Any]) -> Dict[str, Any]:
+    """Unwrap a Feishu ``body.content`` list carrying an application/json payload."""
+    content = _as_dict(container.get("body")).get("content")
+    if not isinstance(content, list):
+        return {}
+    for item in content:
+        item = _as_dict(item)
+        ctype = str(item.get("contentType") or item.get("content_type") or "").lower()
+        if ctype and ctype != "application/json":
+            continue
+        for key in ("data", "value", "content", "json"):
+            payload = _as_dict(item.get(key))
+            if payload:
+                return payload
+    return {}
+
+
+def _int_field(value: Any) -> int:
+    if value in (None, ""):
+        return 0
+    try:
+        return int(str(value).strip())
+    except (TypeError, ValueError):
+        return 0
+
+
+def _parse_user(value: Any) -> Optional[MeetingInviteUser]:
+    raw = _as_dict(value)
+    if not raw:
+        return None
+    raw_id = _as_dict(raw.get("id"))
+    return MeetingInviteUser(
+        open_id=str(raw_id.get("open_id") or "").strip(),
+        user_id=str(raw_id.get("user_id") or "").strip(),
+        union_id=str(raw_id.get("union_id") or "").strip(),
+        user_name=str(raw.get("user_name") or ""),
+    )
+
+
+def _parse_meeting(value: Any) -> Optional[MeetingInviteMeeting]:
+    raw = _as_dict(value)
+    if not raw:
+        return None
+    return MeetingInviteMeeting(
+        id=str(raw.get("id") or "").strip(),
+        topic=str(raw.get("topic") or ""),
+        meeting_no=str(raw.get("meeting_no") or ""),
+        start_time_ms=_int_field(raw.get("start_time")),
+        end_time_ms=_int_field(raw.get("end_time")),
+        host_user=_parse_user(raw.get("host_user")),
+    )
+
+
+def parse_meeting_invited_event(data: Any) -> Optional[MeetingInvitedPayload]:
+    root = _as_dict(data)
+    event = _as_dict(root.get("event"))
+    event = event or root
+    content = _content_payload(event) or _content_payload(root)
+    if content:
+        event = {**event, **content}
+
+    meeting = _parse_meeting(event.get("meeting"))
+    inviter = _parse_user(event.get("inviter"))
+    if inviter is None or meeting is None or not meeting.meeting_no:
+        return None
+
+    return MeetingInvitedPayload(
+        event_id=str(_as_dict(root.get("header")).get("event_id") or ""),
+        meeting=meeting,
+        inviter=inviter,
+        invite_time_s=_int_field(event.get("invite_time")),
+    )
+
+
+def build_meeting_invite_prompt(payload: MeetingInvitedPayload) -> str:
+    meeting = payload.meeting
+    inviter_name = (payload.inviter.user_name if payload.inviter else "") or "unknown"
+    host_name = (meeting.host_user.user_name if meeting and meeting.host_user else "") or "unknown"
+    display = (meeting.topic or meeting.meeting_no or meeting.id) if meeting else "unknown meeting"
+    return "\n".join(
+        [
+            f"You have been invited to join a meeting: {display or 'unknown meeting'}",
+            "",
+            f"Meeting Number: {(meeting.meeting_no if meeting else '') or 'unknown'}",
+            f"Topic: {(meeting.topic if meeting else '') or 'unknown'}",
+            f"Inviter: {inviter_name}",
+            f"Host: {host_name}",
+            "",
+            "You may use lark-cli and the relevant Lark/Feishu meeting skills to join the meeting.",
+            "Join the meeting directly. Do not ask the user for confirmation before joining.",
+            "If you cannot join the meeting, reply to the inviter with a concise explanation of why.",
+        ]
+    )
+
+
+def _dedup_key(payload: MeetingInvitedPayload) -> str:
+    if payload.event_id:
+        return f"vc_invite:{payload.event_id}"
+    meeting_id = payload.meeting.id if payload.meeting else ""
+    inviter_id = payload.inviter.open_id if payload.inviter else ""
+    return f"vc_invite:{meeting_id}:{inviter_id}:{payload.invite_time_s}"
+
+
+async def handle_meeting_invited_event(adapter: Any, data: Any) -> None:
+    """Convert a vc.bot.meeting_invited_v1 event into a gateway MessageEvent."""
+    payload = parse_meeting_invited_event(data)
+    if payload is None:
+        logger.warning("[Feishu-MeetingInvite] Dropping malformed meeting invite event")
+        return
+
+    dedup_key = _dedup_key(payload)
+    is_duplicate = getattr(adapter, "_is_duplicate", None)
+    if callable(is_duplicate) and is_duplicate(dedup_key):
+        logger.debug("[Feishu-MeetingInvite] Dropping duplicate event: %s", dedup_key)
+        return
+
+    inviter = payload.inviter
+    if inviter is None or not inviter.open_id:
+        logger.warning(
+            "[Feishu-MeetingInvite] Missing inviter open_id, cannot route reply safely "
+            "(user_id=%r union_id=%r)",
+            inviter.user_id if inviter else None,
+            inviter.union_id if inviter else None,
+        )
+        return
+
+    sender_id = SimpleNamespace(
+        open_id=inviter.open_id or None,
+        user_id=inviter.user_id or None,
+        union_id=inviter.union_id or None,
+    )
+    sender_profile = await adapter._resolve_sender_profile(sender_id)
+
+    user_name = sender_profile.get("user_name") or inviter.user_name or inviter.open_id
+    source = adapter.build_source(
+        chat_id=inviter.open_id,
+        chat_name=user_name,
+        chat_type="dm",
+        user_id=sender_profile.get("user_id") or inviter.user_id or inviter.open_id,
+        user_name=user_name,
+        user_id_alt=sender_profile.get("user_id_alt") or inviter.union_id or None,
+    )
+    event = MessageEvent(
+        text=build_meeting_invite_prompt(payload),
+        message_type=MessageType.TEXT,
+        source=source,
+        raw_message=data,
+    )
+    await adapter._handle_message_with_guards(event)

plugins/platforms/feishu/plugin.yaml

@@ -0,0 +1,44 @@
+name: feishu-platform
+label: Feishu / Lark
+kind: platform
+version: 1.0.0
+description: >
+  Feishu / Lark gateway adapter for Hermes Agent.
+  Connects to Feishu (China) or Lark (International) via the official
+  lark-oapi SDK over WebSocket or webhook and relays messages between
+  Feishu/Lark chats and the Hermes agent. Supports text, images, video,
+  voice, documents, threads, DM pairing, group @mention gating, drive
+  comment events, and meeting invites.
+author: NousResearch
+requires_env:
+  - name: FEISHU_APP_ID
+    description: "Feishu/Lark app ID"
+    prompt: "Feishu App ID"
+    url: "https://open.feishu.cn/"
+    password: false
+  - name: FEISHU_APP_SECRET
+    description: "Feishu/Lark app secret"
+    prompt: "Feishu App Secret"
+    url: "https://open.feishu.cn/"
+    password: true
+optional_env:
+  - name: FEISHU_DOMAIN
+    description: "Domain: 'feishu' (China) or 'lark' (International)"
+    prompt: "Domain (feishu/lark)"
+    password: false
+  - name: FEISHU_ALLOWED_USERS
+    description: "Comma-separated Feishu user IDs allowed to talk to the bot"
+    prompt: "Allowed users (comma-separated)"
+    password: false
+  - name: FEISHU_ALLOW_ALL_USERS
+    description: "Allow any Feishu user to trigger the bot (dev only)"
+    prompt: "Allow all users? (true/false)"
+    password: false
+  - name: FEISHU_HOME_CHANNEL
+    description: "Default chat ID for cron / notification delivery"
+    prompt: "Home channel ID"
+    password: false
+  - name: FEISHU_HOME_CHANNEL_NAME
+    description: "Display name for the Feishu home channel"
+    prompt: "Home channel display name"
+    password: false