diff --git a/backend/package/yuxi/channel/approval/__init__.py b/backend/package/yuxi/channel/approval/__init__.py
new file mode 100644
index 00000000..d3ae27cc
--- /dev/null
+++ b/backend/package/yuxi/channel/approval/__init__.py
@@ -0,0 +1,30 @@
+from yuxi.channel.approval.delivery import DeliveryResult, deliver_approval_request
+from yuxi.channel.approval.engine import ApprovalEngine
+from yuxi.channel.approval.native_runtime import (
+ NativeApprovalCard,
+ NativeApprovalRenderer,
+ get_native_renderer,
+)
+from yuxi.channel.approval.parser import ParseResult, parse_approval_reply, parse_channel_approval_reply
+from yuxi.channel.approval.templates import (
+ ApprovalTemplateConfig,
+ ApprovalTemplateRenderer,
+ TemplateStyle,
+ get_template_renderer,
+)
+
+__all__ = [
+ "ApprovalEngine",
+ "ApprovalTemplateConfig",
+ "ApprovalTemplateRenderer",
+ "DeliveryResult",
+ "NativeApprovalCard",
+ "NativeApprovalRenderer",
+ "ParseResult",
+ "TemplateStyle",
+ "deliver_approval_request",
+ "get_native_renderer",
+ "get_template_renderer",
+ "parse_approval_reply",
+ "parse_channel_approval_reply",
+]
diff --git a/backend/package/yuxi/channel/approval/delivery.py b/backend/package/yuxi/channel/approval/delivery.py
new file mode 100644
index 00000000..936e8ad5
--- /dev/null
+++ b/backend/package/yuxi/channel/approval/delivery.py
@@ -0,0 +1,139 @@
+"""审批投递适配器 — 将审批请求通过渠道消息投递给审批人"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass, field
+
+from yuxi.channel.protocols import (
+ ApprovalDecision,
+ ApprovalRequest,
+ OutboundProtocol,
+)
+from yuxi.channel.plugins.registry import ChannelPluginRegistry
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_TIMEOUT = 300
+POLL_INTERVAL = 3
+
+
+@dataclass
+class DeliveryResult:
+ success: bool
+ request_id: str
+ decision: ApprovalDecision | None = None
+ decided_by: str | None = None
+ reason: str | None = None
+ error: str | None = None
+ elapsed_seconds: float = 0.0
+ delivery_targets: list[str] = field(default_factory=list)
+
+
+async def deliver_approval_request(
+ request: ApprovalRequest,
+ approver_peer_ids: list[str],
+ *,
+ timeout: float = DEFAULT_TIMEOUT,
+) -> DeliveryResult:
+ """将审批请求投递到审批人，等待审批结果。
+
+ 投递时格式化为结构化审批消息，包含「批准/拒绝」操作指引。
+ 并发发送通知给所有审批人，轮询解析审批回复。
+
+ 返回 DeliveryResult 包含投递过程详情。
+ """
+ started_at = time.monotonic()
+
+ plugin = ChannelPluginRegistry.get(request.channel_type)
+ if plugin is None:
+ return _delivery_expired(request.id, "渠道插件未注册", started_at)
+
+ if not isinstance(plugin, OutboundProtocol):
+ return _delivery_expired(request.id, "渠道不支持消息发送", started_at)
+
+ notify_text = _format_approval_message(request)
+
+ send_tasks = [
+ _send_notification(plugin, peer_id, notify_text, request)
+ for peer_id in approver_peer_ids
+ ]
+ send_results = await asyncio.gather(*send_tasks)
+ delivery_targets = [peer_id for peer_id, ok in send_results if ok]
+
+ if not delivery_targets:
+ return _delivery_expired(request.id, "所有审批通知发送失败", started_at)
+
+ deadline = time.monotonic() + timeout
+ poll_count = 0
+ while time.monotonic() < deadline:
+ poll_count += 1
+ await asyncio.sleep(POLL_INTERVAL)
+
+ try:
+ if callable(getattr(plugin, "check_approval_status", None)):
+ result = await plugin.check_approval_status({}, request.id)
+ if result is not None:
+ logger.info(
+ "审批结果: req=%s, decision=%s, poll=%d, elapsed=%.1fs",
+ request.id,
+ result.decision,
+ poll_count,
+ time.monotonic() - started_at,
+ )
+ return DeliveryResult(
+ success=True,
+ request_id=request.id,
+ decision=result.decision,
+ decided_by=result.decided_by,
+ reason=result.reason,
+ elapsed_seconds=time.monotonic() - started_at,
+ delivery_targets=delivery_targets,
+ )
+ except Exception:
+ logger.exception("轮询审批状态异常: req=%s", request.id)
+
+ return _delivery_expired(request.id, "审批等待超时", started_at, delivery_targets)
+
+
+async def _send_notification(
+ plugin,
+ peer_id: str,
+ notify_text: str,
+ request: ApprovalRequest,
+) -> tuple[str, bool]:
+ try:
+ await plugin.send_text(peer_id, notify_text, account_id=request.account_id)
+ return peer_id, True
+ except Exception:
+ logger.exception(
+ "发送审批通知失败: channel=%s, peer=%s", request.channel_type, peer_id
+ )
+ return peer_id, False
+
+
+def _format_approval_message(request: ApprovalRequest) -> str:
+ from yuxi.channel.approval.templates import get_template_renderer
+
+ renderer = get_template_renderer("default")
+ return renderer.render_request(request)
+
+
+def _delivery_expired(
+ request_id: str,
+ reason: str,
+ started_at: float,
+ delivery_targets: list[str] | None = None,
+) -> DeliveryResult:
+ return DeliveryResult(
+ success=False,
+ request_id=request_id,
+ decision=ApprovalDecision.EXPIRED,
+ decided_by="system",
+ reason=reason,
+ elapsed_seconds=time.monotonic() - started_at,
+ delivery_targets=delivery_targets or [],
+ error=reason,
+ )
diff --git a/backend/package/yuxi/channel/approval/engine.py b/backend/package/yuxi/channel/approval/engine.py
new file mode 100644
index 00000000..5cd7376d
--- /dev/null
+++ b/backend/package/yuxi/channel/approval/engine.py
@@ -0,0 +1,272 @@
+import asyncio
+import logging
+import time
+
+from yuxi.channel.protocols import (
+ ApprovalAction,
+ ApprovalDecision,
+ ApprovalProtocol,
+ ApprovalRequest,
+ ApprovalResult,
+)
+from yuxi.channel.plugins.registry import ChannelPluginRegistry
+
+logger = logging.getLogger(__name__)
+
+_TERMINAL_DECISIONS = frozenset({
+ ApprovalDecision.APPROVED,
+ ApprovalDecision.DENIED,
+ ApprovalDecision.CANCELLED,
+ ApprovalDecision.EXPIRED,
+})
+
+
+class ApprovalEngine:
+ def __init__(self):
+ self._pending: dict[str, ApprovalRequest] = {}
+ self._results: dict[str, ApprovalResult] = {}
+ self._events: dict[str, asyncio.Event] = {}
+ self._cleanup_task: asyncio.Task | None = None
+
+ @property
+ def pending_count(self) -> int:
+ return len(self._pending)
+
+ @property
+ def results_count(self) -> int:
+ return len(self._results)
+
+ def get_pending_request(self, request_id: str) -> ApprovalRequest | None:
+ return self._pending.get(request_id)
+
+ def get_result(self, request_id: str) -> ApprovalResult | None:
+ return self._results.get(request_id)
+
+ def _is_terminal(self, request_id: str) -> bool:
+ existing = self._results.get(request_id)
+ return existing is not None and existing.decision in _TERMINAL_DECISIONS
+
+ def cancel_request(self, request_id: str, reason: str = "已取消") -> bool:
+ existing = self._results.get(request_id)
+ if existing is not None:
+ logger.warning(
+ "审批请求已达终态，无法取消: req=%s, current=%s",
+ request_id,
+ existing.decision.value,
+ )
+ return False
+ if request_id not in self._pending:
+ return False
+ del self._pending[request_id]
+ result = ApprovalResult(
+ request_id=request_id,
+ decision=ApprovalDecision.CANCELLED,
+ reason=reason,
+ )
+ self._results[request_id] = result
+ self._signal(request_id)
+ logger.info("审批请求已取消: req=%s, reason=%s", request_id, reason)
+ return True
+
+ def register_decision(
+ self,
+ request_id: str,
+ decision: ApprovalDecision,
+ decided_by: str = "external",
+ reason: str | None = None,
+ ) -> bool:
+ existing = self._results.get(request_id)
+ if existing is not None:
+ logger.warning(
+ "审批请求已达终态，无法重新决策: req=%s, current=%s, attempted=%s",
+ request_id,
+ existing.decision.value,
+ decision.value,
+ )
+ return False
+ request = self._pending.get(request_id)
+ if request is None:
+ return False
+ if not request.approver_ids:
+ logger.warning(
+ "审批身份校验失败: req=%s, 审批人列表为空",
+ request_id,
+ )
+ return False
+ if decided_by not in request.approver_ids:
+ logger.warning(
+ "审批身份校验失败: req=%s, decided_by=%s 不在审批人列表 %s 中",
+ request_id,
+ decided_by,
+ request.approver_ids,
+ )
+ return False
+ del self._pending[request_id]
+ result = ApprovalResult(
+ request_id=request_id,
+ decision=decision,
+ decided_by=decided_by,
+ reason=reason,
+ )
+ self._results[request_id] = result
+ self._signal(request_id)
+ logger.info(
+ "审批决策已注册: req=%s, decision=%s, decided_by=%s, reason=%s",
+ request_id,
+ decision.value,
+ decided_by,
+ reason,
+ )
+ return True
+
+ def cleanup_expired(self, max_age_seconds: float = 300) -> int:
+ removed = 0
+ now = time.monotonic()
+ stale = [
+ rid for rid, req in self._pending.items()
+ if now - req.created_at > max_age_seconds and rid not in self._results
+ ]
+ for rid in stale:
+ del self._pending[rid]
+ result = ApprovalResult(
+ request_id=rid,
+ decision=ApprovalDecision.EXPIRED,
+ reason="清理过期审批",
+ )
+ self._results[rid] = result
+ self._signal(rid)
+ removed += 1
+ if removed > 0:
+ logger.info("清理过期审批请求: count=%d", removed)
+ return removed
+
+ def cleanup_results(self, max_age_seconds: float = 600) -> int:
+ removed = 0
+ now = time.monotonic()
+ stale = [rid for rid, res in self._results.items() if now - res.decided_at > max_age_seconds]
+ for rid in stale:
+ del self._results[rid]
+ removed += 1
+ if removed > 0:
+ logger.info("清理过期审批结果: count=%d", removed)
+ return removed
+
+ def start_cleanup_task(self, interval: float = 60) -> None:
+ if self._cleanup_task is not None and not self._cleanup_task.done():
+ return
+ self._cleanup_task = asyncio.create_task(self._run_cleanup_loop(interval))
+
+ def stop_cleanup_task(self) -> None:
+ if self._cleanup_task is not None and not self._cleanup_task.done():
+ self._cleanup_task.cancel()
+
+ async def _run_cleanup_loop(self, interval: float) -> None:
+ try:
+ while True:
+ await asyncio.sleep(interval)
+ self.cleanup_expired()
+ self.cleanup_results()
+ except asyncio.CancelledError:
+ pass
+
+ def _signal(self, request_id: str) -> None:
+ event = self._events.pop(request_id, None)
+ if event is not None:
+ event.set()
+
+ async def request_approval(
+ self,
+ channel_type: str,
+ account_id: str,
+ config: dict,
+ action: ApprovalAction,
+ initiator_peer_id: str,
+ description: str,
+ context: dict,
+ ) -> ApprovalRequest | None:
+ plugin = ChannelPluginRegistry.get(channel_type)
+ if plugin is None or not isinstance(plugin, ApprovalProtocol):
+ return None
+
+ need = await plugin.check_approval_required(config, action, initiator_peer_id)
+ if not need:
+ return None
+
+ request = await plugin.create_approval_request(
+ config,
+ action,
+ initiator_peer_id,
+ description,
+ context,
+ )
+ approvers = plugin.get_approver_ids(config)
+ if not approvers:
+ logger.warning(
+ "审批请求创建失败: channel=%s, 审批人列表为空",
+ channel_type,
+ )
+ return None
+ request.approver_ids = list(approvers)
+ sent = await plugin.send_approval_notification(config, request, approvers)
+ if not sent:
+ logger.warning(
+ "审批通知发送失败: req=%s, channel=%s, approvers=%s",
+ request.id,
+ channel_type,
+ approvers,
+ )
+
+ self._pending[request.id] = request
+ return request
+
+ async def wait_for_decision(
+ self, channel_type: str, config: dict, request_id: str, timeout: float = 300
+ ) -> ApprovalResult:
+ existing = self._results.get(request_id)
+ if existing is not None:
+ return existing
+
+ event = self._events.get(request_id)
+ if event is None:
+ event = asyncio.Event()
+ self._events[request_id] = event
+
+ loop = asyncio.get_running_loop()
+ deadline = loop.time() + timeout
+ while True:
+ remaining = deadline - loop.time()
+ if remaining <= 0:
+ break
+
+ plugin_result = await self._check_and_update(channel_type, config, request_id)
+ if plugin_result is not None:
+ self._pending.pop(request_id, None)
+ self._results[request_id] = plugin_result
+ self._events.pop(request_id, None)
+ return plugin_result
+
+ try:
+ await asyncio.wait_for(event.wait(), timeout=min(remaining, 5))
+ except asyncio.TimeoutError:
+ continue
+
+ stored = self._results.get(request_id)
+ if stored is not None:
+ self._events.pop(request_id, None)
+ return stored
+
+ self._events.pop(request_id, None)
+ expired = ApprovalResult(
+ request_id=request_id,
+ decision=ApprovalDecision.EXPIRED,
+ reason="审批超时",
+ )
+ self._pending.pop(request_id, None)
+ self._results[request_id] = expired
+ return expired
+
+ async def _check_and_update(self, channel_type: str, config: dict, request_id: str) -> ApprovalResult | None:
+ plugin = ChannelPluginRegistry.get(channel_type)
+ if plugin is None or not isinstance(plugin, ApprovalProtocol):
+ return None
+ return await plugin.check_approval_status(config, request_id)
diff --git a/backend/package/yuxi/channel/approval/native_runtime.py b/backend/package/yuxi/channel/approval/native_runtime.py
new file mode 100644
index 00000000..a6b80164
--- /dev/null
+++ b/backend/package/yuxi/channel/approval/native_runtime.py
@@ -0,0 +1,373 @@
+"""审批原生运行时 — 通过渠道原生交互式卡片渲染审批 UI
+
+相比 delivery.py 的纯文本投递，native_runtime 使用渠道的交互式卡片消息
+渲染审批按钮，提供更好的用户体验。
+
+支持的渠道卡片格式：
+- 飞书: interactive card (lark_md + button actions)
+- 钉钉: action_card (Markdown + btns)
+- 其他渠道: 降级为纯文本投递
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from typing import Any
+
+from yuxi.channel.protocols import (
+ ApprovalDecision,
+ ApprovalRequest,
+ ApprovalResult,
+)
+from yuxi.channel.plugins.registry import ChannelPluginRegistry
+
+logger = logging.getLogger(__name__)
+
+FEISHU = "feishu"
+DINGTALK = "dingtalk"
+DEFAULT_TIMEOUT_MINUTES = 5
+DEFAULT_TIMEOUT = 300
+POLL_INTERVAL = 3
+
+
+@dataclass
+class NativeApprovalCard:
+ channel_type: str
+ card_content: dict
+ request_id: str
+
+ def to_dict(self) -> dict:
+ return {
+ "channel_type": self.channel_type,
+ "card_content": self.card_content,
+ "request_id": self.request_id,
+ }
+
+
+class NativeApprovalRenderer:
+ def render_card(self, request: ApprovalRequest, channel_type: str) -> NativeApprovalCard:
+ if channel_type == FEISHU:
+ card = _build_feishu_approval_card(request)
+ elif channel_type == DINGTALK:
+ card = _build_dingtalk_approval_card(request)
+ else:
+ card = _build_fallback_card(request)
+ return NativeApprovalCard(
+ channel_type=channel_type,
+ card_content=card,
+ request_id=request.id,
+ )
+
+ def render_resolved_card(self, result: ApprovalResult, request: ApprovalRequest | None, channel_type: str) -> dict:
+ if channel_type == FEISHU:
+ return _build_feishu_resolved_card(result, request)
+ elif channel_type == DINGTALK:
+ return _build_dingtalk_resolved_card(result, request)
+ return _build_fallback_resolved_card(result, request)
+
+ async def deliver_native_approval(
+ self,
+ request: ApprovalRequest,
+ approver_peer_ids: list[str],
+ *,
+ timeout: float = DEFAULT_TIMEOUT,
+ ) -> ApprovalResult:
+ """通过原生交互式卡片投递审批请求。
+
+ 1. 构建渠道原生卡片
+ 2. 通过 send_card 发送给审批人
+ 3. 轮询等待审批结果
+ 4. 获得结果后更新卡片为已处理状态
+
+ 若渠道不支持 send_card，降级为纯文本投递。
+ """
+ from yuxi.channel.approval.delivery import deliver_approval_request
+
+ plugin = ChannelPluginRegistry.get(request.channel_type)
+ if plugin is None:
+ return _native_expired(request.id, "渠道插件未注册")
+
+ if not _supports_card(plugin):
+ logger.info("渠道 %s 不支持卡片消息，降级为纯文本投递", request.channel_type)
+ return await deliver_approval_request(request, approver_peer_ids, timeout=timeout)
+
+ card = self.render_card(request, request.channel_type)
+ sent_message_ids: dict[str, str] = {}
+
+ for peer_id in approver_peer_ids:
+ try:
+ raw = await plugin.send_card(peer_id, card.card_content)
+ msg_id = _extract_msg_id(raw)
+ if msg_id:
+ sent_message_ids[peer_id] = msg_id
+ except Exception:
+ logger.exception(
+ "发送审批卡片失败: channel=%s, peer=%s",
+ request.channel_type,
+ peer_id,
+ )
+
+ if not sent_message_ids:
+ return _native_expired(request.id, "所有审批通知发送失败")
+
+ loop = asyncio.get_running_loop()
+ deadline = loop.time() + timeout
+ try:
+ while loop.time() < deadline:
+ await asyncio.sleep(POLL_INTERVAL)
+
+ if callable(getattr(plugin, "check_approval_status", None)):
+ result = await plugin.check_approval_status({}, request.id)
+ if result is not None:
+ resolved_card = self.render_resolved_card(result, request, request.channel_type)
+ await _edit_cards(plugin, sent_message_ids, resolved_card)
+ return result
+
+ expired = _native_expired(request.id, "审批等待超时")
+ resolved_card = self.render_resolved_card(expired, request, request.channel_type)
+ await _edit_cards(plugin, sent_message_ids, resolved_card)
+ return expired
+ except asyncio.CancelledError:
+ cancelled = ApprovalResult(
+ request_id=request.id,
+ decision=ApprovalDecision.CANCELLED,
+ decided_by="system",
+ reason="审批被取消",
+ )
+ resolved_card = self.render_resolved_card(cancelled, request, request.channel_type)
+ await _edit_cards(plugin, sent_message_ids, resolved_card)
+ raise
+
+
+_native_renderer = NativeApprovalRenderer()
+
+
+def get_native_renderer() -> NativeApprovalRenderer:
+ return _native_renderer
+
+
+def _supports_card(plugin: Any) -> bool:
+ has_send = callable(getattr(plugin, "send_card", None))
+ has_edit = (
+ callable(getattr(plugin, "edit_card", None))
+ or callable(getattr(plugin, "update_card", None))
+ )
+ return has_send and has_edit
+
+
+def _extract_msg_id(raw: Any) -> str | None:
+ if raw is None:
+ return None
+ if isinstance(raw, str):
+ return raw
+ if isinstance(raw, dict):
+ return raw.get("msg_id") or raw.get("message_id")
+ return None
+
+
+async def _edit_cards(plugin: Any, sent_message_ids: dict[str, str], resolved_card: dict) -> None:
+ edit_fn = getattr(plugin, "edit_card", None) or getattr(plugin, "update_card", None)
+ if edit_fn is None:
+ return
+ for peer_id, msg_id in sent_message_ids.items():
+ try:
+ await edit_fn(peer_id, msg_id, resolved_card)
+ except Exception:
+ logger.debug("更新审批卡片失败: %s/%s", peer_id, msg_id)
+
+
+def _native_expired(request_id: str, reason: str) -> ApprovalResult:
+ return ApprovalResult(
+ request_id=request_id,
+ decision=ApprovalDecision.EXPIRED,
+ decided_by="system",
+ reason=reason,
+ )
+
+
+# ── 飞书交互式卡片 ──────────────────────────────────────
+
+
+def _build_feishu_approval_card(request: ApprovalRequest) -> dict:
+ timeout_minutes = DEFAULT_TIMEOUT_MINUTES
+ return {
+ "config": {"wide_screen_mode": True},
+ "header": {
+ "title": {"tag": "plain_text", "content": "🔐 审批请求"},
+ "template": "blue",
+ },
+ "elements": [
+ {
+ "tag": "div",
+ "fields": [
+ {"is_short": True, "text": {"tag": "lark_md", "content": f"**请求 ID**\n`{request.id[:12]}...`"}},
+ {"is_short": True, "text": {"tag": "lark_md", "content": f"**操作类型**\n{request.action.value}"}},
+ ],
+ },
+ {
+ "tag": "div",
+ "text": {"tag": "lark_md", "content": f"**描述**\n{request.description}"},
+ },
+ {
+ "tag": "div",
+ "fields": [
+ {
+ "is_short": True,
+ "text": {"tag": "lark_md", "content": f"**发起人**\n{request.initiator_peer_id}"},
+ },
+ {"is_short": True, "text": {"tag": "lark_md", "content": f"**有效期**\n{timeout_minutes} 分钟"}},
+ ],
+ },
+ {"tag": "hr"},
+ {
+ "tag": "action",
+ "actions": [
+ {
+ "tag": "button",
+ "text": {"tag": "lark_md", "content": "✅ 批准"},
+ "type": "primary",
+ "value": f'{{"action":"approve","req":"{request.id}"}}',
+ },
+ {
+ "tag": "button",
+ "text": {"tag": "lark_md", "content": "❌ 拒绝"},
+ "type": "danger",
+ "value": f'{{"action":"deny","req":"{request.id}"}}',
+ },
+ ],
+ },
+ ],
+ }
+
+
+def _build_feishu_resolved_card(result: ApprovalResult, request: ApprovalRequest | None) -> dict:
+ if result.decision == ApprovalDecision.APPROVED:
+ template = "green"
+ title = "✅ 审批已通过"
+ reason_text = f"**通过原因**: {result.reason}" if result.reason else ""
+ elif result.decision == ApprovalDecision.DENIED:
+ template = "red"
+ title = "❌ 审批已拒绝"
+ reason_text = f"**拒绝原因**: {result.reason}" if result.reason else ""
+ elif result.decision == ApprovalDecision.CANCELLED:
+ template = "grey"
+ title = "🚫 审批已取消"
+ reason_text = f"**取消原因**: {result.reason}" if result.reason else ""
+ else:
+ template = "grey"
+ title = "⏰ 审批已过期"
+ reason_text = f"**原因**: {result.reason}" if result.reason else ""
+
+ elements: list[dict] = [
+ {
+ "tag": "div",
+ "text": {"tag": "lark_md", "content": f"请求 ID: `{result.request_id[:12]}...`"},
+ },
+ ]
+
+ if result.decided_by:
+ elements.append(
+ {
+ "tag": "div",
+ "text": {"tag": "lark_md", "content": f"决定人: {result.decided_by}"},
+ }
+ )
+
+ if reason_text:
+ elements.append(
+ {
+ "tag": "div",
+ "text": {"tag": "lark_md", "content": reason_text},
+ }
+ )
+
+ if request:
+ elements.insert(
+ 0,
+ {
+ "tag": "div",
+ "text": {"tag": "lark_md", "content": f"原请求: {request.description}"},
+ },
+ )
+
+ return {
+ "config": {"wide_screen_mode": True},
+ "header": {
+ "title": {"tag": "plain_text", "content": title},
+ "template": template,
+ },
+ "elements": elements,
+ }
+
+
+# ── 钉钉 Action Card ────────────────────────────────────
+
+
+def _build_dingtalk_approval_card(request: ApprovalRequest) -> dict:
+ timeout_minutes = DEFAULT_TIMEOUT_MINUTES
+ markdown = (
+ f"## 🔐 审批请求\n\n"
+ f"**请求 ID**: `{request.id[:12]}...`\n\n"
+ f"**操作类型**: {request.action.value}\n\n"
+ f"**描述**: {request.description}\n\n"
+ f"**发起人**: {request.initiator_peer_id}\n\n"
+ f"**有效期**: {timeout_minutes} 分钟\n\n"
+ f"> 请点击下方按钮做出决定"
+ )
+ return {
+ "title": "🔐 审批请求",
+ "text": markdown,
+ "btnOrientation": "1",
+ "btns": [
+ {
+ "title": "✅ 批准",
+ "actionURL": f"dingtalk://yuxi/approval/approve?req={request.id}",
+ },
+ {
+ "title": "❌ 拒绝",
+ "actionURL": f"dingtalk://yuxi/approval/deny?req={request.id}",
+ },
+ ],
+ }
+
+
+def _build_dingtalk_resolved_card(result: ApprovalResult, request: ApprovalRequest | None) -> dict:
+ if result.decision == ApprovalDecision.APPROVED:
+ title = "✅ 审批已通过"
+ elif result.decision == ApprovalDecision.DENIED:
+ title = "❌ 审批已拒绝"
+ elif result.decision == ApprovalDecision.CANCELLED:
+ title = "🚫 审批已取消"
+ else:
+ title = "⏰ 审批已过期"
+
+ lines = [f"## {title}", "", f"**请求 ID**: `{result.request_id[:12]}...`"]
+
+ if result.decided_by:
+ lines.append(f"**决定人**: {result.decided_by}")
+ if result.reason:
+ lines.append(f"**原因**: {result.reason}")
+
+ markdown = "\n\n".join(lines)
+ return {"title": title, "text": markdown}
+
+
+# ── 降级卡片（纯文本格式的卡片消息）────────────────────
+
+
+def _build_fallback_card(request: ApprovalRequest) -> dict:
+ from yuxi.channel.approval.templates import get_template_renderer
+
+ renderer = get_template_renderer("default")
+ text = renderer.render_request(request)
+ return {"text": text, "type": "plain_text"}
+
+
+def _build_fallback_resolved_card(result: ApprovalResult, request: ApprovalRequest | None) -> dict:
+ from yuxi.channel.approval.templates import get_template_renderer
+
+ renderer = get_template_renderer("default")
+ text = renderer.render_result(result, request)
+ return {"text": text, "type": "plain_text"}
diff --git a/backend/package/yuxi/channel/approval/parser.py b/backend/package/yuxi/channel/approval/parser.py
new file mode 100644
index 00000000..1338f454
--- /dev/null
+++ b/backend/package/yuxi/channel/approval/parser.py
@@ -0,0 +1,229 @@
+"""审批响应解析器 — 解析审批人回复文本中的批准/拒绝信号"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from typing import Any
+
+from yuxi.channel.protocols import ApprovalDecision
+
+APPROVE_PATTERNS = [
+ re.compile(r"^批准|^同意|^通过|^approve|^yes|^ok|^可以|^确认"),
+ re.compile(r"批准|同意|通过|approve|yes|ok", re.IGNORECASE),
+]
+
+DENY_PATTERNS = [
+ re.compile(r"^拒绝|^驳回|^deny|^no(?!t)\b|^不行"),
+ re.compile(r"拒绝|驳回|deny|\bno(?!t)\b", re.IGNORECASE),
+]
+
+# 当 "no" 后面紧跟这些词时，视为否定表达而非拒绝信号
+_NO_EXCLUDE_SUFFIX = re.compile(r"\s+(?:approve|agree|ok|yes|pass|可以|批准|同意|通过)", re.IGNORECASE)
+
+CANCEL_PATTERNS = [
+ re.compile(r"^取消|^撤销|^cancel|^撤回"),
+ re.compile(r"取消|cancel", re.IGNORECASE),
+]
+
+# 否定词模式 — 要求紧邻匹配关键词前（允许可选空白）
+# 包括独立的否定词 + 中文复合否定词（不能、不会、没法、无法等）
+_NEGATION_PATTERNS = [
+ re.compile(r"不\s*$"),
+ re.compile(r"没\s*$"),
+ re.compile(r"未\s*$"),
+ re.compile(r"别\s*$"),
+ re.compile(r"勿\s*$"),
+ re.compile(r"(?:不能|不会|不要|不可|不准|不想|不愿|不敢|不必|不用)\s*$"),
+ re.compile(r"(?:没法|没有|无法)\s*$"),
+ re.compile(r"(?:不可以|不应该|不可能|不允许|不准许|不值得)\s*$"),
+ re.compile(r"non[-\s]?$", re.IGNORECASE),
+ re.compile(r"not\s*$", re.IGNORECASE),
+ re.compile(r"no\s*$", re.IGNORECASE),
+ re.compile(r"never\s*$", re.IGNORECASE),
+ re.compile(r"don'?t\s*$", re.IGNORECASE),
+ re.compile(r"doesn'?t\s*$", re.IGNORECASE),
+ re.compile(r"didn'?t\s*$", re.IGNORECASE),
+ re.compile(r"won'?t\s*$", re.IGNORECASE),
+ re.compile(r"wouldn'?t\s*$", re.IGNORECASE),
+ re.compile(r"can'?t\s*$", re.IGNORECASE),
+ re.compile(r"couldn'?t\s*$", re.IGNORECASE),
+ re.compile(r"shouldn'?t\s*$", re.IGNORECASE),
+]
+
+# 转折/让步词 — 当这些词出现在批准关键词之后时，降低置信度或视为拒绝
+_CONCESSION_PATTERNS = [
+ re.compile(r"[,，。；\n]\s*但是"),
+ re.compile(r"[,，。；\n]\s*但"),
+ re.compile(r"[,，。；\n]\s*不过"),
+ re.compile(r"[,，。；\n]\s*然而"),
+ re.compile(r"[,，。；\n]\s*只是"),
+ re.compile(r"[,，。；\n]\s*except"),
+ re.compile(r"[,，。；\n]\s*but"),
+ re.compile(r"[,，。；\n]\s*however"),
+]
+
+
+def _is_negated(text: str, match_start: int) -> bool:
+ """检查匹配位置前是否紧邻否定词。
+
+ 只检查匹配位置之前的有限文本片段（最多20个字符），
+ 要求否定词紧邻关键词前（中间仅允许空白字符），
+ 避免"不是不行"中"不行"被误判，也避免远距离否定词干扰。
+ 同时避免否定词属于前一句子的情况。
+ """
+ if match_start <= 0:
+ return False
+ window_start = max(0, match_start - 20)
+ prefix = text[window_start:match_start]
+
+ sentence_breaks = [". ", "。", "? ", "？", "! ", "！", "\n"]
+ for breaker in sentence_breaks:
+ idx = prefix.rfind(breaker)
+ if idx >= 0:
+ prefix = prefix[idx + len(breaker):]
+
+ for neg_pattern in _NEGATION_PATTERNS:
+ if neg_pattern.search(prefix):
+ return True
+ return False
+
+
+def _has_concession_after(text: str, match_end: int) -> bool:
+ """检查匹配位置后是否出现转折/让步词。"""
+ if match_end >= len(text):
+ return False
+ suffix = text[match_end:]
+ for con_pattern in _CONCESSION_PATTERNS:
+ if con_pattern.search(suffix):
+ return True
+ return False
+
+
+@dataclass
+class ParseResult:
+ decision: ApprovalDecision | None
+ reason: str | None = None
+ confidence: float = 0.0
+ matched_pattern: str | None = None
+
+
+@dataclass
+class ApprovalContext:
+ channel_type: str
+ account_id: str
+ request_id: str
+ approver_id: str
+ reply_text: str
+ reply_at: float = 0.0
+ metadata: dict[str, Any] = field(default_factory=dict)
+
+
+def parse_approval_reply(
+ reply_text: str,
+ *,
+ strict: bool = False,
+) -> ParseResult:
+ """解析审批回复文本，识别审批决定。
+
+ 优先级: 批准 > 拒绝 > 取消
+ 严格模式下仅匹配行首关键字（如「批准」「拒绝」）。
+ 非严格模式下会检测否定词前缀，避免"不批准"被误判为批准。
+ """
+ original = reply_text.strip()
+ cleaned = original.lower()
+
+ if strict:
+ for pattern in APPROVE_PATTERNS[:1]:
+ match = pattern.match(cleaned)
+ if match and not _is_negated(cleaned, match.start()):
+ if _has_concession_after(cleaned, match.end()):
+ return ParseResult(
+ decision=ApprovalDecision.DENIED,
+ confidence=0.75,
+ matched_pattern=pattern.pattern,
+ reason=original,
+ )
+ return ParseResult(
+ decision=ApprovalDecision.APPROVED,
+ confidence=0.95,
+ matched_pattern=pattern.pattern,
+ )
+ for pattern in DENY_PATTERNS[:1]:
+ match = pattern.match(cleaned)
+ if match and not _is_negated(cleaned, match.start()):
+ return ParseResult(
+ decision=ApprovalDecision.DENIED,
+ confidence=0.95,
+ matched_pattern=pattern.pattern,
+ reason=original,
+ )
+ for pattern in CANCEL_PATTERNS[:1]:
+ match = pattern.match(cleaned)
+ if match and not _is_negated(cleaned, match.start()):
+ return ParseResult(
+ decision=ApprovalDecision.CANCELLED,
+ confidence=0.95,
+ matched_pattern=pattern.pattern,
+ reason=original,
+ )
+ return ParseResult(decision=None, confidence=0.0)
+
+ for pattern in APPROVE_PATTERNS:
+ match = pattern.search(cleaned)
+ if match and not _is_negated(cleaned, match.start()):
+ if _has_concession_after(cleaned, match.end()):
+ return ParseResult(
+ decision=ApprovalDecision.DENIED,
+ confidence=0.70,
+ matched_pattern=pattern.pattern,
+ reason=original,
+ )
+ return ParseResult(
+ decision=ApprovalDecision.APPROVED,
+ confidence=0.85,
+ matched_pattern=pattern.pattern,
+ )
+
+ for pattern in DENY_PATTERNS:
+ match = pattern.search(cleaned)
+ if match and not _is_negated(cleaned, match.start()):
+ matched_text = match.group().lower()
+ if matched_text == "no" and _NO_EXCLUDE_SUFFIX.match(cleaned[match.end() :]):
+ continue
+ return ParseResult(
+ decision=ApprovalDecision.DENIED,
+ confidence=0.85,
+ matched_pattern=pattern.pattern,
+ reason=original,
+ )
+
+ for pattern in CANCEL_PATTERNS:
+ match = pattern.search(cleaned)
+ if match and not _is_negated(cleaned, match.start()):
+ return ParseResult(
+ decision=ApprovalDecision.CANCELLED,
+ confidence=0.70,
+ matched_pattern=pattern.pattern,
+ reason="审批人取消",
+ )
+
+ return ParseResult(decision=None, confidence=0.0)
+
+
+async def parse_channel_approval_reply(
+ channel_type: str,
+ reply_text: str,
+ *,
+ strict: bool = False,
+) -> ParseResult:
+ """渠道感知的审批回复解析。
+
+ 各渠道可自定义解析逻辑。当前统一使用文本模式匹配，
+ 预留渠道扩展点。
+ """
+ result = parse_approval_reply(reply_text, strict=strict)
+ if result.decision is not None:
+ return result
+
+ return result
diff --git a/backend/package/yuxi/channel/approval/templates.py b/backend/package/yuxi/channel/approval/templates.py
new file mode 100644
index 00000000..81206aa0
--- /dev/null
+++ b/backend/package/yuxi/channel/approval/templates.py
@@ -0,0 +1,164 @@
+"""审批展示模板 — 可自定义的审批信息展示格式"""
+
+from __future__ import annotations
+
+import html as _html
+import logging
+import re
+from dataclasses import dataclass
+from enum import StrEnum
+
+from yuxi.channel.protocols import ApprovalAction, ApprovalDecision, ApprovalRequest, ApprovalResult
+
+logger = logging.getLogger(__name__)
+
+_MD_ESCAPE_RE = re.compile(r"([\\`*_{}\[\]()#+\-.!])")
+
+
+class TemplateStyle(StrEnum):
+ PLAIN = "plain"
+ MARKDOWN = "markdown"
+ HTML = "html"
+
+
+@dataclass
+class ApprovalTemplateConfig:
+ title: str = "审批请求"
+ show_request_id: bool = True
+ show_action: bool = True
+ show_description: bool = True
+ show_initiator: bool = True
+ show_expiry: bool = True
+ style: TemplateStyle = TemplateStyle.PLAIN
+
+ approve_label: str = "批准"
+ deny_label: str = "拒绝"
+ timeout_minutes: int = 5
+
+
+_DEFAULT_CONFIG = ApprovalTemplateConfig()
+
+
+class ApprovalTemplateRenderer:
+ def __init__(self, config: ApprovalTemplateConfig | None = None):
+ self._config = config or _DEFAULT_CONFIG
+
+ @property
+ def config(self) -> ApprovalTemplateConfig:
+ return self._config
+
+ def render_request(self, request: ApprovalRequest) -> str:
+ cfg = self._config
+ lines: list[str] = [f"🔐 {self._bold(cfg.title)}"]
+
+ if cfg.show_request_id:
+ lines.append(f"请求 ID: {self._code(request.id)}")
+ if cfg.show_action:
+ lines.append(f"操作类型: {_action_label(request.action)}")
+ if cfg.show_description:
+ lines.append(f"描述: {self._escape(request.description)}")
+ if cfg.show_initiator:
+ lines.append(f"发起人: {self._escape(request.initiator_peer_id)}")
+ if cfg.show_expiry and cfg.timeout_minutes > 0:
+ lines.append(f"请在 {cfg.timeout_minutes} 分钟内回复")
+
+ lines.append("")
+ lines.append(f"回复「{self._bold(cfg.approve_label)}」或「{self._bold(cfg.deny_label)}」做出决定")
+ return "\n".join(lines)
+
+ def render_result(self, result: ApprovalResult, request: ApprovalRequest | None = None) -> str:
+ if result.decision == ApprovalDecision.APPROVED:
+ icon = "✅"
+ text = "审批已通过"
+ elif result.decision == ApprovalDecision.DENIED:
+ icon = "❌"
+ text = "审批已拒绝"
+ elif result.decision == ApprovalDecision.CANCELLED:
+ icon = "🚫"
+ text = "审批已取消"
+ elif result.decision == ApprovalDecision.EXPIRED:
+ icon = "⏰"
+ text = "审批已过期"
+ else:
+ icon = "❓"
+ text = "审批状态未知"
+
+ lines = [f"{icon} {self._bold(text)}"]
+ lines.append(f"请求 ID: {self._code(result.request_id)}")
+ if result.reason:
+ lines.append(f"原因: {self._escape(result.reason)}")
+ if result.decided_by:
+ lines.append(f"决定人: {self._escape(result.decided_by)}")
+ return "\n".join(lines)
+
+ def render_expiry_warning(self, request: ApprovalRequest, remaining_seconds: int) -> str:
+ request_id = self._code(request.id)
+ if remaining_seconds <= 0:
+ return f"⏳ 审批请求 {request_id} 已过期。"
+ if remaining_seconds < 60:
+ return f"⏳ 审批请求 {request_id} 将在 {remaining_seconds} 秒后过期，请尽快处理。"
+ minutes = remaining_seconds // 60
+ return f"⏳ 审批请求 {request_id} 将在 {minutes} 分钟后过期，请尽快处理。"
+
+ def _bold(self, text: str) -> str:
+ style = self._config.style
+ if style == TemplateStyle.MARKDOWN:
+ return f"**{text}**"
+ elif style == TemplateStyle.HTML:
+ return f"{_html.escape(text)}"
+ return text
+
+ def _code(self, text: str) -> str:
+ style = self._config.style
+ if style == TemplateStyle.MARKDOWN:
+ return f"`{text}`"
+ elif style == TemplateStyle.HTML:
+ return f"{_html.escape(text)}"
+ return f"'{text}'"
+
+ def _escape(self, text: str) -> str:
+ style = self._config.style
+ if style == TemplateStyle.HTML:
+ return _html.escape(text)
+ elif style == TemplateStyle.MARKDOWN:
+ return _MD_ESCAPE_RE.sub(r"\\\1", text)
+ return text
+
+
+def _action_label(action: ApprovalAction) -> str:
+ labels: dict[ApprovalAction, str] = {
+ ApprovalAction.APPROVE_EXEC: "执行操作",
+ ApprovalAction.APPROVE_READ: "读取操作",
+ }
+ return labels.get(action, action.value)
+
+
+_renderer_cache: dict[str, ApprovalTemplateRenderer] = {}
+
+
+def get_template_renderer(style: str = "default") -> ApprovalTemplateRenderer:
+ if style not in _renderer_cache:
+ if style == "default":
+ _renderer_cache[style] = ApprovalTemplateRenderer()
+ elif style == "compact":
+ _renderer_cache[style] = ApprovalTemplateRenderer(
+ ApprovalTemplateConfig(
+ title="审批",
+ show_request_id=False,
+ show_action=False,
+ show_initiator=False,
+ show_expiry=False,
+ )
+ )
+ elif style == "detailed":
+ _renderer_cache[style] = ApprovalTemplateRenderer(
+ ApprovalTemplateConfig(
+ title="操作审批请求",
+ style=TemplateStyle.MARKDOWN,
+ timeout_minutes=10,
+ )
+ )
+ else:
+ logger.warning("未知模板风格 '%s'，使用默认配置", style)
+ _renderer_cache[style] = ApprovalTemplateRenderer()
+ return _renderer_cache[style]