diff --git a/cli/feishu_cli.py b/cli/feishu_cli.py new file mode 100644 index 00000000..7c857bb9 --- /dev/null +++ b/cli/feishu_cli.py @@ -0,0 +1,174 @@ +"""``deepcode feishu`` — run the Feishu approval bridge. + +The bridge itself lives in :mod:`integrations.feishu_approver`; this module is +only the CLI surface so the integration has a discoverable, documented entry +point instead of a bare ``python -m`` invocation. The heavy ``lark_oapi`` +import stays inside :func:`integrations.feishu_approver.__main__.main` so a +configuration error (or ``--help``) never pays the SDK import cost. +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +from pathlib import Path + + +def _pid_file() -> Path: + """Where the detached bridge records its pid. + + Fixed location derived from ``DEEPCODE_HOME`` (falling back to + ``~/.deepcode``), never the process cwd — a bridge spawned from another + working directory must still be visible to ``status``/``stop``. + """ + home = os.environ.get("DEEPCODE_HOME") or str(Path.home() / ".deepcode") + return Path(home) / "feishu-approver.pid" + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="deepcode feishu", + description="Run the Feishu approval bridge.", + ) + commands = parser.add_subparsers(dest="command") + + start = commands.add_parser("start", help="Run the bridge in the foreground.") + start.add_argument( + "--detach", + action="store_true", + help="Spawn the bridge in the background and return immediately.", + ) + + commands.add_parser("status", help="Report whether a detached bridge is running.") + commands.add_parser("stop", help="Stop a bridge started with --detach.") + return parser + + +def _pid_alive(pid: int) -> bool: + """True when *pid* names a live process (POSIX and Windows).""" + if pid <= 0: + return False + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except OSError: + # Windows raises OSError for "exists but not ours"; anything other + # than a clean lookup is treated as running so stop() can still try. + return True + return True + + +def _read_pid() -> int | None: + try: + return int(_pid_file().read_text(encoding="utf-8").strip()) + except (OSError, ValueError): + return None + + +def _write_pid(pid: int) -> None: + path = _pid_file() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(str(pid), encoding="utf-8") + + +def _clear_pid() -> None: + try: + _pid_file().unlink() + except OSError: + pass + + +def _run_foreground() -> int: + # Deferred so `--help` and a config error stay lark_oapi-free. + from integrations.feishu_approver.__main__ import main as bridge_main + + return bridge_main() + + +def _start_detached() -> int: + existing = _read_pid() + if existing is not None and _pid_alive(existing): + print(f"feishu bridge already running (pid {existing})", file=sys.stderr) + return 1 + + # Run from the repo root (same convention as the app_server service) so + # ``-m integrations.feishu_approver`` resolves the package regardless of + # the caller's cwd. + from app_server.service_state import SERVICE_WORKING_DIRECTORY + + # The bridge already mirrors its logs to ``FEISHU_APPROVER_LOG``; capture + # the detached stdio there too (else a ``pythonw``/detached run drops the + # startup/config-error output into a black hole). + log_path = os.environ.get("FEISHU_APPROVER_LOG", "").strip() + stdout_target = subprocess.DEVNULL + log_handle = None + if log_path: + Path(log_path).parent.mkdir(parents=True, exist_ok=True) + log_handle = open(log_path, "ab") # noqa: SIM115 + stdout_target = log_handle + try: + proc = subprocess.Popen( + [sys.executable, "-m", "integrations.feishu_approver"], + stdout=stdout_target, + stderr=subprocess.STDOUT, + stdin=subprocess.DEVNULL, + cwd=str(SERVICE_WORKING_DIRECTORY), + close_fds=True, + ) + finally: + if log_handle is not None: + log_handle.close() + _write_pid(proc.pid) + print(f"feishu bridge started in background (pid {proc.pid})") + return 0 + + +def _status() -> int: + pid = _read_pid() + if pid is None: + print(json.dumps({"running": False, "reason": "no pid file"})) + return 3 + if _pid_alive(pid): + print(json.dumps({"running": True, "pid": pid})) + return 0 + print(json.dumps({"running": False, "pid": pid, "reason": "stale pid file"})) + _clear_pid() + return 3 + + +def _stop() -> int: + pid = _read_pid() + if pid is None or not _pid_alive(pid): + print("feishu bridge is not running", file=sys.stderr) + _clear_pid() + return 3 + try: + os.kill(pid, 15) # SIGTERM; on Windows this is a best-effort terminate. + except OSError as exc: + print(f"failed to signal pid {pid}: {exc}", file=sys.stderr) + return 1 + _clear_pid() + print(f"feishu bridge stopped (pid {pid})") + return 0 + + +def run(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + if args.command == "start": + if args.detach: + return _start_detached() + return _run_foreground() + if args.command == "status": + return _status() + if args.command == "stop": + return _stop() + _parser().print_help() + return 0 + + +if __name__ == "__main__": + raise SystemExit(run()) diff --git a/deepcode.py b/deepcode.py index e42be18d..e44d1e79 100755 --- a/deepcode.py +++ b/deepcode.py @@ -167,6 +167,10 @@ def main(): from cli.automation_cli import run as automation_run raise SystemExit(automation_run(sys.argv[2:])) + elif sys.argv[1] == "feishu": + from cli.feishu_cli import run as feishu_run + + raise SystemExit(feishu_run(sys.argv[2:])) elif sys.argv[1] in ["--help", "-h", "help"]: print_banner() @@ -240,6 +244,10 @@ def row(cmd, desc): ), row("deepcode provider list", "Show model connections"), row("deepcode service status", "Check the background service"), + row( + "deepcode feishu start", + "Run the Feishu approval bridge", + ), "", "Run 'deepcode --help' for command options.", "Run 'deepcode chat --help' for TUI options.", diff --git a/integrations/__init__.py b/integrations/__init__.py new file mode 100644 index 00000000..4307fc30 --- /dev/null +++ b/integrations/__init__.py @@ -0,0 +1 @@ +"""Optional integrations that extend DeepCode at the edges (E1).""" diff --git a/integrations/feishu_approver/__init__.py b/integrations/feishu_approver/__init__.py new file mode 100644 index 00000000..f4bdceaa --- /dev/null +++ b/integrations/feishu_approver/__init__.py @@ -0,0 +1,31 @@ +"""Feishu approval bridge: a phone-side approver for DeepCode runs. + +This package is a *separate process* that connects to a running DeepCode +app_server over its native RPC, relays ``approval.requested`` prompts to Feishu +as interactive cards, and writes decisions back via ``approval/respond``. It +makes no changes to the core; it is an edge extension. + +Run it with:: + + python -m integrations.feishu_approver +""" + +from __future__ import annotations + +__all__ = ["ApprovalBridge", "BridgeSettings", "load_settings"] + + +def __getattr__(name: str): # lazy re-export; lark_oapi import stays off the hot path + if name == "ApprovalBridge": + from .bridge import ApprovalBridge + + return ApprovalBridge + if name == "BridgeSettings": + from .config import BridgeSettings + + return BridgeSettings + if name == "load_settings": + from .config import load_settings + + return load_settings + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/integrations/feishu_approver/__main__.py b/integrations/feishu_approver/__main__.py new file mode 100644 index 00000000..eec3f7e9 --- /dev/null +++ b/integrations/feishu_approver/__main__.py @@ -0,0 +1,112 @@ +"""Entry point: ``python -m integrations.feishu_approver``. + +Reads configuration from the environment (see ``config.py`` for the variables), +connects to the local DeepCode app_server, and starts relaying approval prompts +to Feishu. Runs until interrupted; Ctrl-C stops it cleanly. + +Example (POSIX):: + + export FEISHU_APP_ID=cli_xxx + export FEISHU_APP_SECRET=xxx + export FEISHU_OPEN_IDS=ou_xxx + python -m integrations.feishu_approver +""" + +from __future__ import annotations + +import asyncio +import logging +import signal +import sys + +from integrations.feishu_approver.config import load_settings + +try: # Optional convenience: plain env vars still work without python-dotenv. + from dotenv import load_dotenv + + load_dotenv() +except ImportError: + pass + + +def _install_signal_handlers(bridge, loop: asyncio.AbstractEventLoop) -> None: + for sig in (signal.SIGINT, signal.SIGTERM): + try: + loop.add_signal_handler(sig, bridge.stop) + except (NotImplementedError, RuntimeError): + # Windows / non-main thread: fall back to the default KeyboardInterrupt. + pass + + +def _install_file_logging() -> None: + """Mirror logs to ``FEISHU_APPROVER_LOG`` when set. + + ``pythonw.exe`` (Windows scheduled task) discards stderr, so without this a + crashed bridge leaves no traceback anywhere. A logging failure here must + never take the bridge down, so it is swallowed (stderr still works under a + real console). + """ + import os + + path = os.environ.get("FEISHU_APPROVER_LOG", "").strip() + if not path: + return + try: + os.makedirs(os.path.dirname(path) or ".", exist_ok=True) + handler = logging.FileHandler(path, encoding="utf-8") + handler.setFormatter( + logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") + ) + logging.getLogger().addHandler(handler) + except OSError: + logging.getLogger("feishu-approver").exception( + "failed to open FEISHU_APPROVER_LOG=%s", path + ) + + +async def _amain() -> int: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + _install_file_logging() + log = logging.getLogger("feishu-approver") + + try: + settings = load_settings() + except ValueError as exc: + log.error("Configuration error: %s", exc) + return 2 + + # Imported here so a config error does not pay the lark_oapi import cost. + from integrations.feishu_approver.bridge import ApprovalBridge + + bridge = ApprovalBridge(settings) + loop = asyncio.get_running_loop() + _install_signal_handlers(bridge, loop) + + log.info( + "Starting Feishu approval bridge (timeout=%ss action=%s, open_ids=%d, chat_ids=%d)", + settings.timeout_seconds, + settings.timeout_action, + len(settings.open_ids), + len(settings.chat_ids), + ) + try: + await bridge.run() + except Exception: + log.exception("Feishu approval bridge crashed") + return 1 + log.info("Feishu approval bridge stopped") + return 0 + + +def main() -> int: + try: + return asyncio.run(_amain()) + except KeyboardInterrupt: + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/integrations/feishu_approver/bridge.py b/integrations/feishu_approver/bridge.py new file mode 100644 index 00000000..cfc70d19 --- /dev/null +++ b/integrations/feishu_approver/bridge.py @@ -0,0 +1,499 @@ +"""Core of the Feishu approval bridge. + +Runs as a **separate process** that: + +1. connects to a running DeepCode app_server over its native RPC (``NativeRpcClient``, + authenticated via ``ServiceFiles`` discovery + bearer token, starting the + service when it is down); +2. listens for the ``approval.requested`` live notification, which carries the + durable ``approval_view``, and sweeps for approvals missed while the bridge + was down; +3. pushes an interactive Feishu card (allow once / allow session / deny) to the + allowlisted ``open_id`` owners via the Feishu long-lived websocket (no public IP); +4. translates a card click into an ``approval/respond`` RPC — the durable SQLite + transition is authoritative, so a phone decision is exactly as valid as a + terminal one. + +Timeout policy (``FEISHU_APPROVER_TIMEOUT_ACTION``): an approval left +unanswered is auto-resolved after ``timeout_seconds`` (``0`` disables) — +``deny`` fails closed, ``approve_once``/``approve_session`` auto-默许 so an +unattended phone never stalls a running turn. A *card delivery* failure +still denies immediately: infrastructure errors must not silently allow +risky operations through. Clicks from users outside the ``open_id`` +allowlist are ignored. Cards carry only the opaque ``approval_id``. + +``lark_oapi`` is imported lazily (it is slow to import) so this module stays +unit-testable with a mock connector. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import threading +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from app_server.native_client import NativeRpcClient +from app_server.service_state import ServiceFiles +from core.persistence.database import default_database_path +from integrations.feishu_approver.config import BridgeSettings + +logger = logging.getLogger(__name__) + +PROTOCOL_VERSION = "1.0" + +# "automation" marks this as a machine client rather than a human session. +_CLIENT_INFO = {"name": "feishu-approver", "version": "1", "surface": "automation"} + +DECISION_APPROVE_ONCE = "approved_once" +DECISION_APPROVE_SESSION = "approved_session" +DECISION_DENY = "denied" + +# Card button ``value.action`` -> decision string accepted by approval/respond. +_ACTIONS: dict[str, str] = { + "approve_once": DECISION_APPROVE_ONCE, + "approve_session": DECISION_APPROVE_SESSION, + "deny": DECISION_DENY, +} + +_DECISION_LABELS = { + DECISION_APPROVE_ONCE: "Approved once", + DECISION_APPROVE_SESSION: "Approved for session", + DECISION_DENY: "Denied", +} + +# settings.timeout_action -> decision sent when nobody answers in time. +_TIMEOUT_DECISIONS: dict[str, str] = { + "deny": DECISION_DENY, + "approve_once": DECISION_APPROVE_ONCE, + "approve_session": DECISION_APPROVE_SESSION, +} + + +def _new_rpc_client( + files: ServiceFiles, notify: Callable[[dict], None] +) -> NativeRpcClient: + """Factory seam so tests can inject a fake RPC client.""" + return NativeRpcClient(files, notify=notify) + + +class ApprovalBridge: + """Owns one RPC connection to app_server and one Feishu long connection.""" + + def __init__( + self, + settings: BridgeSettings, + *, + client_factory: Callable[ + [ServiceFiles, Callable[[dict], None]], Any + ] = _new_rpc_client, + ): + self.settings = settings + self._client_factory = client_factory + + # approval_id -> {"approval": view, "messageId": str}. Added when a + # prompt card is sent, atomically taken on click/timeout/deny. + self._pending: dict[str, dict[str, Any]] = {} + self._pending_lock = threading.Lock() + + # Timers for timeout auto-resolution, keyed by approval_id. + self._timers: dict[str, asyncio.TimerHandle] = {} + + # Bridges the Feishu callback thread (sync) onto the RPC event loop. + self._loop: asyncio.AbstractEventLoop | None = None + self._rpc: Any = None + database = ( + Path(settings.database_path).expanduser() + if settings.database_path + else default_database_path() + ) + self._files = ServiceFiles(database) + self._lark_client: Any = None # lark.Client (sync, for im.v1.message.create) + self._ws: Any = None # lark.ws.Client + self._stop = asyncio.Event() + + # ------------------------------------------------------------------ RPC + + async def _rpc_request(self, method: str, params: dict[str, Any]) -> Any: + """One app_server RPC round-trip.""" + rpc = self._rpc + if rpc is None: + raise RuntimeError("Feishu bridge RPC client not started") + return await rpc.request(method, params) + + async def _send_message( + self, receive_id_type: str, receive_id: str, msg_type: str, content: str + ) -> str: + """Create one Feishu message, return its id.""" + body = { + "receive_id": receive_id, + "msg_type": msg_type, + "content": content, + } + return await asyncio.to_thread(self._create_message, receive_id_type, body) + + def _on_notify(self, message: dict[str, Any]) -> None: + """Runs on the RPC read loop; must never raise.""" + try: + method = message.get("method") + params = message.get("params") or {} + if method == "approval.requested": + approval = (params.get("payload") or {}).get("approval") or {} + if approval.get("status") != "pending": + return + self._schedule(self._handle_approval(approval, params)) + return + except Exception: + logger.exception("feishu-approver: failed to handle notification") + + def _schedule(self, coro) -> None: + if self._loop is None or self._loop.is_closed(): + coro.close() + return + asyncio.run_coroutine_threadsafe(coro, self._loop) + + async def _connect_rpc(self) -> None: + """Connect (starting the service if needed); notify drives live events.""" + initialize = {"protocolVersion": PROTOCOL_VERSION, "clientInfo": _CLIENT_INFO} + self._rpc = self._client_factory(self._files, self._on_notify) + info = await self._rpc.connect(initialize, start=True) + logger.info( + "feishu-approver: connected to app_server %s", info.get("serviceInfo") + ) + + async def _respond( + self, approval_id: str, decision: str, note: str | None = None + ) -> bool: + """Send approval/respond. Returns False if already resolved elsewhere.""" + params: dict[str, Any] = {"approvalId": approval_id, "decision": decision} + if note: + params["message"] = note + try: + await self._rpc.request("approval/respond", params) + return True + except Exception as exc: # noqa: BLE001 - a lost race surfaces as an RPC error + logger.info( + "feishu-approver: respond %s for %s failed (likely already resolved): %s", + decision, + approval_id, + exc, + ) + return False + + # ------------------------------------------------------------- lifecycle + + async def run(self) -> None: + """Run until cancelled: connect, sweep pending, then pump events.""" + self._loop = asyncio.get_running_loop() + # Start Feishu first: once the RPC connection is up, approval.requested + # can arrive at any moment, and a prompt must never race a missing + # _lark_client (that would fail-closed deny a legitimate request). + self._start_feishu() + await self._connect_rpc() + await self._sweep_pending() + try: + await self._stop.wait() + finally: + await self._close() + + def stop(self) -> None: + """Ask the bridge to shut down (thread-safe).""" + if self._loop is not None and not self._loop.is_closed(): + self._loop.call_soon_threadsafe(self._stop.set) + + async def _close(self) -> None: + for timer in list(self._timers.values()): + timer.cancel() + self._timers.clear() + if self._rpc is not None: + try: + await self._rpc.close() + except Exception: + logger.debug("feishu-approver: rpc close raised", exc_info=True) + + # ------------------------------------------------------------- sweeping + + async def _sweep_pending(self) -> None: + """Recover approvals requested while the bridge was down.""" + try: + threads = await self._rpc.request("thread/list", {}) + for thread in threads.get("threads", []): + if thread.get("status") != "waiting": + continue + turns = await self._rpc.request( + "turn/list", {"threadId": thread["id"], "state": "active"} + ) + for turn in turns.get("turns", []): + snapshot = await self._rpc.request( + "turn/read", {"turnId": turn["id"]} + ) + for approval in snapshot.get("approvals", []): + if approval.get("status") == "pending": + await self._handle_approval(approval, {}) + except Exception: + logger.exception("feishu-approver: pending sweep failed") + + # ------------------------------------------------------------- approvals + + async def _handle_approval( + self, approval: dict[str, Any], event: dict[str, Any] + ) -> None: + approval_id = str(approval.get("id") or "") + if not approval_id: + return + with self._pending_lock: + if approval_id in self._pending: + return # already prompted (sweep + live event race) + self._pending[approval_id] = { + "approval": approval, + "threadId": event.get("threadId") or approval.get("threadId"), + } + try: + await self._send_card(approval) + except Exception: + logger.exception("feishu-approver: failed to send card for %s", approval_id) + with self._pending_lock: + self._pending.pop(approval_id, None) + # Fail closed so a cardless prompt does not hang the run. + await self._respond(approval_id, DECISION_DENY, "card delivery failed") + return + self._arm_timeout(approval_id) + + def _arm_timeout(self, approval_id: str) -> None: + """Auto-resolve via ``settings.timeout_action`` if nobody answers.""" + seconds = self.settings.timeout_seconds + if seconds <= 0 or self._loop is None: + return + timer = self._loop.call_later(seconds, self._on_timeout, approval_id) + with self._pending_lock: + self._timers[approval_id] = timer + + def _take_pending(self, approval_id: str) -> dict[str, Any] | None: + """Atomically pop a pending entry and cancel its timeout timer.""" + with self._pending_lock: + entry = self._pending.pop(approval_id, None) + timer = self._timers.pop(approval_id, None) + if timer is not None: + timer.cancel() + return entry + + def _on_timeout(self, approval_id: str) -> None: + """Runs on the event loop; apply the configured timeout action.""" + entry = self._take_pending(approval_id) + if entry is None: + return + message_id = str(entry.get("messageId") or "") + decision = _TIMEOUT_DECISIONS[self.settings.timeout_action] + self._schedule( + self._finish( + approval_id, decision, "timeout", by="auto", message_id=message_id + ) + ) + + async def _finish( + self, + approval_id: str, + decision: str, + note: str, + *, + by: str, + message_id: str = "", + ) -> bool: + """Record a decision and (best-effort) patch the card to show it.""" + ok = await self._respond(approval_id, decision, note) + logger.info( + "feishu-approver: decision %s recorded for %s (ok=%s) by=%s", + decision, + approval_id, + ok, + by, + ) + if ok and message_id: + try: + await self._patch_card(message_id, approval_id, decision, by) + except Exception: + logger.debug("feishu-approver: card patch failed", exc_info=True) + return ok + + async def _send_card(self, approval: dict[str, Any]) -> None: + from integrations.feishu_approver.feishu_card import build_prompt_card + + card = build_prompt_card( + approval, + timeout_seconds=self.settings.timeout_seconds, + timeout_action=self.settings.timeout_action, + ) + receive_id_type = "chat_id" if self.settings.chat_ids else "open_id" + targets = self.settings.chat_ids or self.settings.open_ids + if not targets: + raise RuntimeError("no Feishu recipients configured") + body = { + "receive_id": targets[0], + "msg_type": "interactive", + "content": json.dumps(card, ensure_ascii=False), + } + message_id = await asyncio.to_thread( + self._create_message, receive_id_type, body + ) + with self._pending_lock: + entry = self._pending.get(str(approval.get("id") or "")) + if entry is not None: + entry["messageId"] = message_id + logger.info( + "feishu-approver: prompt card sent approval=%s message=%s", + approval.get("id"), + message_id, + ) + + def _create_message(self, receive_id_type: str, body: dict[str, Any]) -> str: + from lark_oapi.api.im.v1 import CreateMessageRequest, CreateMessageRequestBody + + if self._lark_client is None: + raise RuntimeError("Feishu client not started") + + request_body = ( + CreateMessageRequestBody.builder() + .receive_id(body["receive_id"]) + .msg_type(body["msg_type"]) + .content(body["content"]) + .build() + ) + request = ( + CreateMessageRequest.builder() + .receive_id_type(receive_id_type) + .request_body(request_body) + .build() + ) + response = self._lark_client.im.v1.message.create(request) + if not response.success(): + raise RuntimeError( + f"Feishu message create failed: {response.code} {response.msg}" + ) + return str(getattr(response.data, "message_id", "") or "") + + async def _patch_card( + self, message_id: str, approval_id: str, decision: str, by: str + ) -> None: + """Replace the prompt card with a resolved card (cosmetic, best-effort).""" + from lark_oapi.api.im.v1 import PatchMessageRequest, PatchMessageRequestBody + + from integrations.feishu_approver.feishu_card import build_resolved_card + + if not message_id: + return + card = build_resolved_card(approval_id, decision=decision, by=by) + body = ( + PatchMessageRequestBody.builder() + .content(json.dumps(card, ensure_ascii=False)) + .build() + ) + request = ( + PatchMessageRequest.builder() + .message_id(message_id) + .request_body(body) + .build() + ) + response = await asyncio.to_thread( + self._lark_client.im.v1.message.patch, request + ) + if not response.success(): + logger.debug( + "feishu-approver: patch failed for %s: %s %s", + approval_id, + response.code, + response.msg, + ) + + # ------------------------------------------------------------- feishu + + def _start_feishu(self) -> None: + """Start the Feishu long connection and build the sync API client.""" + import lark_oapi as lark + + # Warm the sync IM API now: its first import costs seconds, and the + # first approval prompt must not pay that latency while a tool waits. + import lark_oapi.api.im.v1 + import lark_oapi.ws.client as lark_ws_client + from lark_oapi.event.dispatcher_handler import EventDispatcherHandler + + # lark's ws client pins a module-level event loop at import time. We + # import it from inside the bridge's running loop, so without this it + # would capture *our* loop and the ws thread would die with + # "This event loop is already running". Give it a loop of its own. + lark_ws_client.loop = asyncio.new_event_loop() + + self._lark_client = ( + lark.Client.builder() + .app_id(self.settings.app_id) + .app_secret(self.settings.app_secret) + .build() + ) + handler = ( + EventDispatcherHandler.builder("", "") + .register_p2_card_action_trigger(self._on_card_action) + .build() + ) + self._ws = lark.ws.Client( + self.settings.app_id, + self.settings.app_secret, + event_handler=handler, + log_level=lark.LogLevel.WARNING, + ) + thread = threading.Thread( + target=self._ws.start, name="feishu-approver-ws", daemon=True + ) + thread.start() + logger.info("feishu-approver: Feishu long connection starting") + + def _on_card_action(self, event: Any) -> Any: + """Runs on the Feishu callback thread; returns a toast within 3s.""" + data = getattr(event, "event", event) + operator = getattr(data, "operator", None) + action = getattr(data, "action", None) + open_id = getattr(operator, "open_id", None) or "" + value = getattr(action, "value", None) or {} + approval_id = str(value.get("approval_id") or "") + action_name = str(value.get("action") or "") + + if open_id not in self.settings.open_ids: + return self._toast("error", "Not authorized to approve.") + decision = _ACTIONS.get(action_name) + if decision is None or not approval_id: + return self._toast("error", "Unknown action.") + + entry = self._take_pending(approval_id) + if entry is None: + return self._toast("warning", "Already resolved or unknown.") + message_id = str(entry.get("messageId") or "") + logger.info( + "feishu-approver: card click approval=%s action=%s by=%s", + approval_id, + action_name, + open_id, + ) + + self._schedule( + self._finish( + approval_id, + decision, + f"feishu:{action_name}", + by=open_id, + message_id=message_id, + ) + ) + return self._toast("success", f"{_DECISION_LABELS[decision]}.") + + @staticmethod + def _toast(toast_type: str, content: str) -> Any: + from lark_oapi.event.callback.model.p2_card_action_trigger import ( + P2CardActionTriggerResponse, + ) + + # construct.parse() requires a plain dict for object-typed fields; it + # re-checks isinstance(dict) and raises UnmarshalException otherwise. + return P2CardActionTriggerResponse( + {"toast": {"type": toast_type, "content": content}} + ) diff --git a/integrations/feishu_approver/config.py b/integrations/feishu_approver/config.py new file mode 100644 index 00000000..378deab8 --- /dev/null +++ b/integrations/feishu_approver/config.py @@ -0,0 +1,123 @@ +"""Configuration loader for the Feishu approval bridge. + +All secrets come from environment variables (or a .env file) so nothing +sensitive is ever committed. Placeholders of the form ``${ENV_VAR}`` are +resolved from the environment. + +Environment variables +--------------------- +``FEISHU_APP_ID`` Feishu self-built app ID. +``FEISHU_APP_SECRET`` Feishu self-built app secret. +``FEISHU_OPEN_IDS`` Comma-separated ``open_id`` allowlist (the only + users whose card clicks count as a decision). +``FEISHU_APPROVER_TIMEOUT`` Seconds before an unanswered prompt is auto-resolved + (default ``600``; ``0`` disables the timeout). +``FEISHU_APPROVER_TIMEOUT_ACTION`` + What "auto-resolve" means: ``deny`` (fail-closed, + the safe default), ``approve_once``, or + ``approve_session`` (let the run continue and stop + re-asking for the same tool in this session). +``FEISHU_CHAT_IDS`` Optional comma-separated ``chat_id`` targets to + push prompt cards to. Leave empty to DM each + allowlisted user instead. +``DEEPCODE_DATABASE`` Optional path to the DeepCode SQLite database; the + bridge auto-discovers the app_server from it. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + + +def _split_ids(raw: str | None) -> tuple[str, ...]: + if not raw: + return () + return tuple( + part.strip() for part in raw.replace("\n", ",").split(",") if part.strip() + ) + + +def _coerce_timeout(raw: str | None) -> int: + if raw is None or raw.strip() == "": + return 600 + try: + value = int(raw) + except ValueError as exc: + raise ValueError( + f"FEISHU_APPROVER_TIMEOUT must be an integer, got {raw!r}" + ) from exc + if value < 0: + raise ValueError("FEISHU_APPROVER_TIMEOUT must be >= 0") + return value + + +# Accepted FEISHU_APPROVER_TIMEOUT_ACTION values -> canonical names. +_TIMEOUT_ACTIONS: dict[str, str] = { + "deny": "deny", + "approve_once": "approve_once", + "approve_session": "approve_session", +} + + +def _coerce_timeout_action(raw: str | None) -> str: + if raw is None or raw.strip() == "": + return "deny" + value = raw.strip().lower() + if value not in _TIMEOUT_ACTIONS: + raise ValueError( + "FEISHU_APPROVER_TIMEOUT_ACTION must be one of " + f"{sorted(_TIMEOUT_ACTIONS)}, got {raw!r}" + ) + return _TIMEOUT_ACTIONS[value] + + +@dataclass(frozen=True) +class BridgeSettings: + """Validated configuration for the Feishu approval bridge.""" + + app_id: str + app_secret: str + open_ids: tuple[str, ...] = field(default_factory=tuple) + timeout_seconds: int = 600 + timeout_action: str = "deny" + chat_ids: tuple[str, ...] = field(default_factory=tuple) + database_path: str | None = None + + def validate(self) -> None: + """Raise ``ValueError`` when required settings are missing or empty.""" + if not self.app_id: + raise ValueError("FEISHU_APP_ID is required") + if not self.app_secret: + raise ValueError("FEISHU_APP_SECRET is required") + if not self.open_ids: + raise ValueError( + "FEISHU_OPEN_IDS is required (comma-separated open_id allowlist); " + "without it no card click can produce a decision" + ) + if self.timeout_action not in _TIMEOUT_ACTIONS.values(): + raise ValueError( + "FEISHU_APPROVER_TIMEOUT_ACTION must be one of " + f"{sorted(_TIMEOUT_ACTIONS.values())}, got {self.timeout_action!r}" + ) + + +def load_settings(environ: dict[str, str] | None = None) -> BridgeSettings: + """Build :class:`BridgeSettings` from the environment. + + ``environ`` defaults to ``os.environ``; inject a dict in tests. + """ + env = os.environ if environ is None else environ + settings = BridgeSettings( + app_id=env.get("FEISHU_APP_ID", "").strip(), + app_secret=env.get("FEISHU_APP_SECRET", "").strip(), + open_ids=_split_ids(env.get("FEISHU_OPEN_IDS")), + timeout_seconds=_coerce_timeout(env.get("FEISHU_APPROVER_TIMEOUT")), + timeout_action=_coerce_timeout_action( + env.get("FEISHU_APPROVER_TIMEOUT_ACTION") + ), + chat_ids=_split_ids(env.get("FEISHU_CHAT_IDS")), + database_path=env.get("DEEPCODE_DATABASE") or None, + ) + settings.validate() + return settings diff --git a/integrations/feishu_approver/feishu_card.py b/integrations/feishu_approver/feishu_card.py new file mode 100644 index 00000000..0a75b43d --- /dev/null +++ b/integrations/feishu_approver/feishu_card.py @@ -0,0 +1,204 @@ +"""Interactive Feishu card construction for approval prompts. + +The card is deliberately minimal: a header colour encodes urgency, the body +summarises the pending tool call, and three action buttons carry the decision. +Only ``approval_id`` rides along in the button ``value`` — never secrets or +full arguments — so a leaked card cannot leak more than the opaque id. + +Card JSON follows the Feishu "card 1.0" schema (``config.wide_screen_mode``, +``header.template``, ``elements`` with ``tag: "action"`` + ``button``). The +bridge never imports the SDK here so the card can be unit-tested offline. +""" + +from __future__ import annotations + +from typing import Any + +# ``approval_view`` category -> Feishu header colour. Unknown categories fall +# back to the neutral "blue". Colours are cosmetic only; authorization is +# decided by the open_id allowlist, never by the card colour. +_CATEGORY_COLORS: dict[str, str] = { + # Canonical ApprovalCategory values written by core's approval service. + "command": "orange", + "file_write": "blue", + "network": "red", + "external_tool": "orange", + "destructive": "red", + # Legacy/alternate spellings kept so older payloads still colour sensibly. + "file": "blue", + "secret": "red", + "permission": "orange", +} + +# Button ``value.action`` -> decision string accepted by the ``approval/respond`` +# RPC. Kept explicit (not derived) so an accidental rename is caught by tests. +DECISIONS: dict[str, str] = { + "approve_once": "approved_once", + "approve_session": "approved_session", + "deny": "denied", +} + + +def _truncate(text: str, limit: int = 600) -> str: + if len(text) <= limit: + return text + return text[: limit - 1] + "…" + + +def _button(value: dict[str, str], text: str, style: str | None) -> dict[str, Any]: + button: dict[str, Any] = { + "tag": "button", + "text": {"tag": "plain_text", "content": text}, + "type": "primary" if style is None else "default", + "value": value, + } + if style is not None: + button["type"] = style + return button + + +def _timeout_note(timeout_seconds: int, timeout_action: str) -> str: + """Human-readable auto-resolution note shown in the card footer.""" + if timeout_seconds <= 0: + return "No auto-resolution." + if timeout_action == "approve_session": + return f"Auto-approved for this session in {timeout_seconds}s if unanswered." + if timeout_action == "approve_once": + return f"Auto-approved once in {timeout_seconds}s if unanswered." + return f"Auto-denied in {timeout_seconds}s." + + +def build_prompt_card( + approval: dict[str, Any], + *, + timeout_seconds: int, + timeout_action: str = "deny", +) -> dict[str, Any]: + """Build the interactive approval card. + + ``approval`` is the ``approval_view`` payload from the ``approval.requested`` + event (``{id, category, status, request, ...}``). ``request`` is expected to + be a dict with ``toolName`` (core's canonical camelCase; ``tool_name`` is + accepted as a legacy fallback) / ``arguments`` / ``reason``. + """ + approval_id = str(approval.get("id", "")) + request = approval.get("request") or {} + category = str(approval.get("category", "unknown")) + tool_name = str( + request.get("toolName") or request.get("tool_name") or "unknown tool" + ) + arguments = request.get("arguments") or {} + reason = str(request.get("reason") or "").strip() + + # Pretty-print arguments compactly; a long JSON blob is truncated so the + # card stays readable on a phone screen. + import json + + try: + args_text = json.dumps(arguments, ensure_ascii=False, default=str) + except (TypeError, ValueError): + args_text = str(arguments) + args_text = _truncate(args_text, 500) + + timeout_note = _timeout_note(timeout_seconds, timeout_action) + elements: list[dict[str, Any]] = [ + { + "tag": "div", + "text": { + "tag": "lark_md", + "content": f"**Tool:** `{_truncate(tool_name, 120)}`\n**Category:** `{category}`", + }, + }, + { + "tag": "div", + "text": { + "tag": "lark_md", + "content": f"**Arguments:**\n`{_truncate(args_text, 500)}`", + }, + }, + ] + if reason: + elements.append( + { + "tag": "div", + "text": { + "tag": "lark_md", + "content": f"**Reason:** {_truncate(reason, 400)}", + }, + } + ) + elements.append({"tag": "hr"}) + elements.append( + { + "tag": "note", + "elements": [ + { + "tag": "plain_text", + "content": f"{timeout_note} Only allowlisted open_ids count.", + }, + ], + } + ) + elements.append( + { + "tag": "action", + "actions": [ + _button( + {"approval_id": approval_id, "action": "approve_once"}, + "Approve once", + "primary", + ), + _button( + {"approval_id": approval_id, "action": "approve_session"}, + "Approve for session", + None, + ), + _button( + {"approval_id": approval_id, "action": "deny"}, "Deny", "danger" + ), + ], + } + ) + + return { + "config": {"wide_screen_mode": True}, + "header": { + "template": _CATEGORY_COLORS.get(category, "blue"), + "title": {"tag": "plain_text", "content": "DeepCode approval required"}, + }, + "elements": elements, + } + + +def build_resolved_card(approval_id: str, *, decision: str, by: str) -> dict[str, Any]: + """Build the follow-up card shown after a decision is recorded.""" + label = { + "approved_once": "Approved (once)", + "approved_session": "Approved (session)", + "denied": "Denied", + }.get(decision, decision) + return { + "config": {"wide_screen_mode": True}, + "header": { + "template": "green" if decision.startswith("approved") else "red", + "title": {"tag": "plain_text", "content": f"DeepCode approval: {label}"}, + }, + "elements": [ + { + "tag": "div", + "text": { + "tag": "lark_md", + "content": f"Decision: **{label}**\nBy: `{by}`", + }, + }, + { + "tag": "note", + "elements": [ + { + "tag": "plain_text", + "content": f"Approval `{approval_id}` · decision `{decision}`", + } + ], + }, + ], + } diff --git a/setup.py b/setup.py index 65eba3cb..c41705e3 100644 --- a/setup.py +++ b/setup.py @@ -71,6 +71,11 @@ def read_requirements(): extras_require={ "server": [], "advanced-documents": ["docling>=2.113.0"], + # The Feishu approval bridge (integrations/feishu_approver, + # `deepcode feishu start`). Optional so the core install stays lean; + # `lark_oapi` is imported lazily only when the bridge actually runs. + # Install with `pip install .[feishu]`. + "feishu": ["lark-oapi>=1.4.0"], "test": [ "pytest>=8,<10", "pytest-asyncio>=1,<2", diff --git a/tests/test_feishu_approver.py b/tests/test_feishu_approver.py new file mode 100644 index 00000000..c0b10867 --- /dev/null +++ b/tests/test_feishu_approver.py @@ -0,0 +1,354 @@ +"""Closed-loop test for the Feishu approval bridge with a mock RPC client. + +Covers the full decision path without touching a real DeepCode service or a +real Feishu tenant: + +* ``approval.requested`` push -> prompt card sent +* allowlisted card click -> ``approval/respond`` approved_session +* unlisted card click -> ignored (no respond) +* unanswered prompt -> ``timeout_action`` decision on timeout + (default fail-closed ``denied``) +* card-send failure -> fail-closed ``denied`` (no hang) +* startup sweep -> missed pending approval re-prompted +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + +import pytest + +from integrations.feishu_approver.bridge import ( + DECISION_APPROVE_SESSION, + DECISION_DENY, + ApprovalBridge, +) +from integrations.feishu_approver.config import BridgeSettings + +ME = "ou_me" +OTHER = "ou_stranger" + + +def make_bridge( + *, timeout_seconds: int = 0, timeout_action: str = "deny" +) -> tuple[ApprovalBridge, FakeRpc, "FakeFeishu"]: + """Build a bridge wired to a FakeRpc and a fake Feishu sender. + + The FakeRpc is created here (not inside ``factory``) so the test and the + bridge share one instance: sweep fixtures and ``responded`` must be visible + to both sides. + """ + settings = BridgeSettings( + app_id="cli_test", + app_secret="secret", + open_ids=(ME,), + timeout_seconds=timeout_seconds, + timeout_action=timeout_action, + chat_ids=(), + database_path=None, + ) + rpc = FakeRpc(lambda _: None) + + def factory(files, notify): + rpc.notify = notify + return rpc + + bridge = ApprovalBridge(settings, client_factory=factory) + fake = FakeFeishu() + # Replace the three lark-backed methods with fakes that mirror their + # contracts but never import or construct lark SDK types, so these tests + # run in a CI env that does not install the optional ``lark_oapi`` extra. + bridge._create_message = fake.create # type: ignore[method-assign] + bridge._patch_card = fake.patch # type: ignore[method-assign] + bridge._toast = fake.toast # type: ignore[method-assign] + # Bypass the real Feishu long connection; tests drive clicks directly. + bridge._start_feishu = lambda: None # type: ignore[method-assign] + return bridge, rpc, fake + + +def approval_view(approval_id: str, *, status: str = "pending") -> dict: + return { + "id": approval_id, + "threadId": "thr_1", + "turnId": "turn_1", + "status": status, + "request": { + # Real core payloads are camelCase (see ApprovalService.request); + # the old snake_case fixture masked the card's toolName bug. + "toolName": "bash", + "arguments": {"command": "ls"}, + "reason": "test", + }, + } + + +def notify_frame(approval: dict) -> dict: + return { + "method": "approval.requested", + "params": {"threadId": approval["threadId"], "payload": {"approval": approval}}, + } + + +def card_event(approval_id: str, *, open_id: str = ME, action: str = "approve_session"): + """Fabricate the lark P2CardActionTrigger shape the handler reads.""" + return SimpleNamespace( + event=SimpleNamespace( + operator=SimpleNamespace(open_id=open_id), + action=SimpleNamespace( + value={"approval_id": approval_id, "action": action} + ), + ) + ) + + +class FakeRpc: + """Records requests and lets tests drive the notify callback.""" + + def __init__(self, notify): + self.notify = notify + self.calls: list[tuple[str, dict]] = [] + self.responded: list[dict] = [] + self.fail_request = False + self.sweep_threads: list[dict] = [] + self.sweep_turns: dict[str, list[dict]] = {} + self.sweep_snapshots: dict[str, dict] = {} + + async def connect(self, initialize, *, start=False): + return {"serviceInfo": {"instance": "test"}} + + async def request(self, method, params): + self.calls.append((method, params)) + if self.fail_request: + raise RuntimeError("fake rpc failure") + if method == "approval/respond": + self.responded.append(params) + return { + "approval": {"id": params["approvalId"], "status": params["decision"]} + } + if method == "thread/list": + return {"threads": self.sweep_threads} + if method == "turn/list": + return {"turns": self.sweep_turns.get(params.get("threadId"), [])} + if method == "turn/read": + return self.sweep_snapshots.get(params.get("turnId"), {"approvals": []}) + return {} + + async def close(self): + return None + + +class FakeFeishu: + """Stands in for the three lark-backed methods the bridge calls. + + The real ``_create_message``/``_patch_card``/``_toast`` build ``lark_oapi`` + request/response objects; ``lark_oapi`` is an optional extra that a CI env + may not have installed. These fakes replace those methods entirely (wired up + in ``make_bridge``) so the tests exercise the bridge's decision logic without + importing or constructing any lark SDK types. + """ + + def __init__(self): + self.created: list[dict] = [] + self.patched: list[dict] = [] + self.toasts: list[dict] = [] + self.fail_create = False + + def create(self, receive_id_type: str, body: dict) -> str: + """Mirrors ``ApprovalBridge._create_message``'s contract: return a + message_id, or raise to trigger the bridge's fail-closed deny.""" + if self.fail_create: + raise RuntimeError("fake card send failure") + self.created.append(body) + return f"om_{len(self.created)}" + + async def patch( + self, message_id: str, approval_id: str, decision: str, by: str + ) -> None: + """Mirrors ``ApprovalBridge._patch_card``'s contract: best-effort, no raise.""" + self.patched.append( + { + "message_id": message_id, + "approval_id": approval_id, + "decision": decision, + "by": by, + } + ) + + def toast(self, toast_type: str, content: str) -> SimpleNamespace: + """Mirrors ``ApprovalBridge._toast``'s shape: a response object with a + ``.toast`` accessor carrying ``type`` and ``content``.""" + self.toasts.append({"type": toast_type, "content": content}) + return SimpleNamespace(toast=SimpleNamespace(type=toast_type, content=content)) + + +async def settle(bridge: ApprovalBridge, *, rounds: int = 50): + """Let scheduled coroutines and to_thread card sends finish.""" + for _ in range(rounds): + await asyncio.sleep(0.01) + + +@pytest.mark.asyncio +async def test_push_then_click_approves(): + bridge, rpc, fake = make_bridge() + bridge._loop = asyncio.get_running_loop() + await bridge._connect_rpc() + + view = approval_view("apr_1") + bridge._on_notify(notify_frame(view)) + await settle(bridge) + + assert "apr_1" in bridge._pending + assert len(fake.created) == 1 + + toast = bridge._on_card_action(card_event("apr_1", action="approve_session")) + await settle(bridge) + + assert rpc.responded == [ + { + "approvalId": "apr_1", + "decision": DECISION_APPROVE_SESSION, + "message": "feishu:approve_session", + } + ] + assert "apr_1" not in bridge._pending + assert toast.toast.type == "success" + # The prompt card was patched to a resolved card. + assert len(fake.patched) == 1 + + +@pytest.mark.asyncio +async def test_unlisted_click_is_ignored(): + bridge, rpc, _ = make_bridge() + bridge._loop = asyncio.get_running_loop() + await bridge._connect_rpc() + + bridge._on_notify(notify_frame(approval_view("apr_2"))) + await settle(bridge) + + toast = bridge._on_card_action(card_event("apr_2", open_id=OTHER)) + await settle(bridge) + + assert rpc.responded == [] + assert "apr_2" in bridge._pending # still awaiting an authorized decision + assert toast.toast.type == "error" + + +@pytest.mark.asyncio +async def test_double_click_second_is_ignored(): + bridge, rpc, _ = make_bridge() + bridge._loop = asyncio.get_running_loop() + await bridge._connect_rpc() + + bridge._on_notify(notify_frame(approval_view("apr_3"))) + await settle(bridge) + + first = bridge._on_card_action(card_event("apr_3", action="approve_once")) + second = bridge._on_card_action(card_event("apr_3", action="deny")) + await settle(bridge) + + assert first.toast.type == "success" + assert second.toast.type == "warning" + decisions = [c["decision"] for c in rpc.responded] + assert decisions == ["approved_once"] + + +@pytest.mark.asyncio +async def test_timeout_denies(): + bridge, rpc, _ = make_bridge(timeout_seconds=1) + bridge._loop = asyncio.get_running_loop() + await bridge._connect_rpc() + + bridge._on_notify(notify_frame(approval_view("apr_4"))) + await settle(bridge) + assert "apr_4" in bridge._pending + + # Let the 1s timer fire. + await asyncio.sleep(1.2) + await settle(bridge) + + assert rpc.responded == [ + {"approvalId": "apr_4", "decision": DECISION_DENY, "message": "timeout"} + ] + assert "apr_4" not in bridge._pending + + +@pytest.mark.asyncio +async def test_timeout_approves_session_when_configured(): + """timeout_action=approve_session auto-allows instead of denying.""" + bridge, rpc, _ = make_bridge(timeout_seconds=1, timeout_action="approve_session") + bridge._loop = asyncio.get_running_loop() + await bridge._connect_rpc() + + bridge._on_notify(notify_frame(approval_view("apr_6"))) + await settle(bridge) + assert "apr_6" in bridge._pending + + # Let the 1s timer fire. + await asyncio.sleep(1.2) + await settle(bridge) + + assert rpc.responded == [ + { + "approvalId": "apr_6", + "decision": DECISION_APPROVE_SESSION, + "message": "timeout", + } + ] + assert "apr_6" not in bridge._pending + + +@pytest.mark.asyncio +async def test_no_timeout_when_click_wins_race(): + bridge, rpc, _ = make_bridge(timeout_seconds=1) + bridge._loop = asyncio.get_running_loop() + await bridge._connect_rpc() + + bridge._on_notify(notify_frame(approval_view("apr_5"))) + await settle(bridge) + bridge._on_card_action(card_event("apr_5", action="approve_once")) + await settle(bridge) + + # Wait past the timeout window; the cancelled timer must not deny. + await asyncio.sleep(1.2) + await settle(bridge) + + assert [c["decision"] for c in rpc.responded] == ["approved_once"] + + +@pytest.mark.asyncio +async def test_card_send_failure_denies(): + bridge, rpc, fake = make_bridge() + bridge._loop = asyncio.get_running_loop() + await bridge._connect_rpc() + fake.fail_create = True + + bridge._on_notify(notify_frame(approval_view("apr_6"))) + await settle(bridge) + + assert rpc.responded == [ + { + "approvalId": "apr_6", + "decision": DECISION_DENY, + "message": "card delivery failed", + } + ] + assert "apr_6" not in bridge._pending + + +@pytest.mark.asyncio +async def test_sweep_recovers_missed_pending(): + bridge, rpc, fake = make_bridge() + bridge._loop = asyncio.get_running_loop() + await bridge._connect_rpc() + + missed = approval_view("apr_7") + rpc.sweep_threads = [{"id": "thr_1", "status": "waiting"}] + rpc.sweep_turns = {"thr_1": [{"id": "turn_1"}]} + rpc.sweep_snapshots = {"turn_1": {"approvals": [missed]}} + + await bridge._sweep_pending() + await settle(bridge) + + assert "apr_7" in bridge._pending + assert len(fake.created) == 1