diff --git a/backend/.env.example b/backend/.env.example index 77fdd1c4..9c87b802 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -18,3 +18,4 @@ STAFFDECK_ROLE="all" WECHAT_ILINK_BASE_URL="https://ilinkai.weixin.qq.com" CHANNEL_DELIVERY_POLL_SECONDS="1.0" CHANNEL_DELIVERY_MAX_ATTEMPTS="8" +CHANNEL_RICH_RENDER_ENABLED="true" diff --git a/backend/app/channels/adapters/dingtalk.py b/backend/app/channels/adapters/dingtalk.py index 662e17f1..aaebe4d1 100644 --- a/backend/app/channels/adapters/dingtalk.py +++ b/backend/app/channels/adapters/dingtalk.py @@ -24,6 +24,13 @@ stream_download_with_limit, ) from app.channels.crypto import decrypt_channel_secret +from app.channels.markdown_render import ( + ensure_code_fences, + extract_dingtalk_title, + has_markdown, + split_markdown_by_lines, +) +from app.config import get_settings from app.db import engine from app.db.models import ChannelBinding @@ -543,12 +550,31 @@ def send( expires_ms = int(target.get("session_webhook_expired_time") or 0) if expires_ms and expires_ms <= int(datetime.now(tz=UTC).timestamp() * 1000): raise DingTalkPermanentError("钉钉会话回复地址已过期") + rich_enabled = bool(get_settings().channel_rich_render_enabled) + use_rich = rich_enabled and has_markdown(text) + if use_rich: + chunks = split_markdown_by_lines(text, DINGTALK_TEXT_LIMIT) + if not chunks: + chunks = [text] + else: + chunks = split_channel_text(text, DINGTALK_TEXT_LIMIT) try: with self._client_factory() as client: - for chunk in split_channel_text(text, DINGTALK_TEXT_LIMIT): + for chunk in chunks: + if use_rich: + fenced = ensure_code_fences(chunk) + body = { + "msgtype": "markdown", + "markdown": { + "title": extract_dingtalk_title(fenced), + "text": fenced, + }, + } + else: + body = {"msgtype": "text", "text": {"content": chunk}} response = client.post( webhook, - json={"msgtype": "text", "text": {"content": chunk}}, + json=body, headers={"Content-Type": "application/json"}, ) data = response.json() diff --git a/backend/app/channels/adapters/feishu.py b/backend/app/channels/adapters/feishu.py index 0e67a347..e591637e 100644 --- a/backend/app/channels/adapters/feishu.py +++ b/backend/app/channels/adapters/feishu.py @@ -9,12 +9,20 @@ import httpx from app.channels.adapters.base import ( + CHANNEL_TEXT_LIMIT, ChannelInboundAttachment, register_channel_adapter, split_channel_text, stream_download_with_limit, ) from app.channels.crypto import decrypt_channel_secret +from app.channels.markdown_render import ( + has_markdown, + parse_markdown, + render_feishu_post, + split_markdown_by_lines, +) +from app.config import get_settings from app.db.models import ChannelBinding FEISHU_API_BASE = "https://open.feishu.cn/open-apis" @@ -202,6 +210,8 @@ def _request( response = client.post(url, json=body, **request_kwargs) elif method == "GET": response = client.get(url, **request_kwargs) + elif method == "PATCH": + response = client.patch(url, json=body, **request_kwargs) else: response = client.delete(url, **request_kwargs) except (httpx.TimeoutException, httpx.NetworkError) as exc: @@ -387,6 +397,71 @@ def download_media( return response.content raise FeishuPermanentError("飞书 token 刷新后仍无法下载附件") + def create_card( + self, + binding: ChannelBinding, + target: dict[str, Any], + card_json: dict[str, Any], + *, + idempotency_key: str, + ) -> str: + """发送一张交互式卡片，返回 message_id。 + + 复用 send() 的目标解析逻辑：有 message_id 走 reply，否则按 receive_id 投递。 + idempotency_key 生成稳定 uuid，保证重试不重复发卡。 + """ + key = str(idempotency_key or "").strip() + if not key: + raise FeishuPermanentError("飞书卡片创建缺少幂等键") + message_id = str(target.get("message_id") or "").strip() + receive_id = str(target.get("receive_id") or "").strip() + receive_id_type = str(target.get("receive_id_type") or "").strip() + if not message_id and (not receive_id or not receive_id_type): + raise FeishuPermanentError("飞书卡片投递目标无效") + body: dict[str, Any] = { + "msg_type": "interactive", + "content": json.dumps(card_json, ensure_ascii=False), + "uuid": self._uuid(key, 0), + } + if message_id: + body["reply_in_thread"] = bool(target.get("reply_in_thread")) + data = self._post( + binding, + f"{FEISHU_API_BASE}/im/v1/messages/{message_id}/reply", + params=None, + body=body, + ) + else: + body["receive_id"] = receive_id + data = self._post( + binding, + f"{FEISHU_API_BASE}/im/v1/messages", + params={"receive_id_type": receive_id_type}, + body=body, + ) + created_id = str((data.get("data") or {}).get("message_id") or "").strip() + if not created_id: + raise FeishuTransientError("飞书卡片创建响应缺少 message_id") + return created_id + + def update_card( + self, + binding: ChannelBinding, + message_id: str, + card_json: dict[str, Any], + ) -> None: + """PATCH 更新已发送卡片的 content。""" + message_id = str(message_id or "").strip() + if not message_id: + raise FeishuPermanentError("飞书卡片更新缺少 message_id") + self._request( + binding, + "PATCH", + f"{FEISHU_API_BASE}/im/v1/messages/{message_id}", + params=None, + body={"content": json.dumps(card_json, ensure_ascii=False)}, + ) + def send( self, binding: ChannelBinding, @@ -405,12 +480,28 @@ def send( receive_id_type = str(target.get("receive_id_type") or "").strip() if not message_id and (not receive_id or not receive_id_type): raise FeishuPermanentError("飞书投递目标无效") - for index, chunk in enumerate(split_channel_text(text)): - body: dict[str, Any] = { - "msg_type": "text", - "content": json.dumps({"text": chunk}, ensure_ascii=False), - "uuid": self._uuid(key, index), - } + rich_enabled = bool(get_settings().channel_rich_render_enabled) + use_rich = rich_enabled and has_markdown(text) + if use_rich: + chunks = split_markdown_by_lines(text, CHANNEL_TEXT_LIMIT) + if not chunks: + chunks = [text] + else: + chunks = split_channel_text(text) + for index, chunk in enumerate(chunks): + if use_rich: + post_content = render_feishu_post(parse_markdown(chunk)) + body: dict[str, Any] = { + "msg_type": "post", + "content": json.dumps(post_content, ensure_ascii=False), + "uuid": self._uuid(key, index), + } + else: + body = { + "msg_type": "text", + "content": json.dumps({"text": chunk}, ensure_ascii=False), + "uuid": self._uuid(key, index), + } if message_id: body["reply_in_thread"] = bool(target.get("reply_in_thread")) self._post( diff --git a/backend/app/channels/feishu_path.md b/backend/app/channels/feishu_path.md new file mode 100644 index 00000000..8cde8c2a --- /dev/null +++ b/backend/app/channels/feishu_path.md @@ -0,0 +1,117 @@ +# 完整调用链路 +以用户给飞书机器人发一条 p2p 文本消息为例（群里 @ 机器人类似，多一步 mention 清洗）。 + +## 阶段一：子进程接收并归一化（feishu_runtime.py 子进程内） + +1. 飞书 WebSocket 推送事件帧 → ProductionClient._handle_data_frame（feishu_runtime.py:180）给 watchdog 挂 token，调父类处理。 +2. 事件分发 → EventDispatcherHandler（由 _build_event_dispatcher 构造，feishu_runtime.py:107）匹配到 p2_im_message_receive_v1，调用 receive 回调（feishu_runtime.py:154）。 +3. 归一化 → _normalize_event(event, bot_open_id)（feishu_runtime.py:26）： +- 校验 header.app_id / tenant_key / message_id / sender.open_id +- 过滤掉 bot 自身、非 text 类型、空文本 +- 群消息要求 @ 到本机器人，并剔除 @ key +- 构造 ChannelInbound（channel="feishu"、event_id=message_id、from_user_id=open_id、session_id=chat_id 或 open_id、text） +- 同时构造 target dict（message_id / receive_id_type / receive_id） +4. 落库暂存 → stage_feishu_inbound(...)（service_feishu_inbox.py:59）： +- 校验 binding 活跃 + revision + external_account_key + provider_tenant_key + identity_scope_key（首次写入钉住租户） +- encode_replay_envelope 把 ChannelInbound 序列化进 envelope（含 app_id/tenant_key） +- 写入 ChannelInboundEvent 表，status="received"，(binding_id, event_id) 唯一约束防重 +- 返回 StageResult，STAGED 时带 event_pk +5. 通知父进程 → control.emit("INBOX_STAGED", event_pk=...)（feishu_runtime.py:171）通过 Pipe 发给 supervisor。 + +## 阶段二：父进程唤醒 intake（feishu_process.py → service_intake.py） + + 6. supervisor 收到事件 → _accept_event（feishu_process.py:215）匹配 binding_id + config_revision + child_nonce + pid，写入 record.events 队列；遇到 INBOX_STAGED 调 wake_staged_inbound_worker()（service_intake.py:1064），_staged_inbound_wake.set()。 + 7. intake 守护线程 → run_staged_inbound_daemon（service_intake.py:1068）被唤醒，查 ChannelInboundEvent 表 status="received" 的行，对每个 event_pk 调 process_staged_inbound（service_intake.py:967）。 + 8. claim 抢占 → claim_staged_inbound（service_intake.py:167）用 UPDATE ... WHERE status="received" 原子地把状态置 processing 并写 processor_run_id，rowcount==1 才继续；失败/并发被别的线程抢走则跳过。 + 9. 解码 envelope → _decode_and_validate_staged_event（service_intake.py:1125）重新校验 binding.external_account_key / provider_tenant_key / identity_scope_key 与 envelope 中的 app_id / tenant_key 一致，还原出 ChannelInbound。 +10. reaction 标记 → _stage_received_reaction（service_intake.py:465）登记一条 ChannelDelivery(kind="reaction_add") 给原消息挂"处理中"标记（飞书支持 reaction 时）。 +11. → process_inbound(binding, inbound, staged_event_pk=event_pk)（service_intake.py:707）。 + +## 阶段三：身份/会话锚定 + 对话轮（service_intake.py 主线程，持 _session_lock） + +12. 指令拦截 → parse_command(inbound.text)（service_intake.py:728），若是 /bind、/员工 等指令直接 _stage_notice 回复并标记 event done，不进 AgentLoop。 +13. 身份解析 → external_identity_for_message + resolve_or_provision_user（service_intake.py:858）按 open_id + scope 找或创建本地 User。 +14. 去重 → _client_turn_seen_in_conv / _user_message_with_client_turn_exists 防崩溃恢复重放。 +15. agent 路由 → resolve_current_agent（service_intake.py:875）取当前指针员工；maybe_auto_route（service_intake.py:878）可选 LLM 意图分发切换 agent。 +16. 会话锚定 → find_or_create_channel_session（service_intake.py:881）按 binding + external_conv_id + agent_id 找或建 ChatSession，并把 target 写进 channel_target_json。 +17. typing 上行 → _send_wechat_typing(..., state=1)（service_intake.py:940）通过 outbox 登记一条"对方正在输入"投递（飞书会走对应渠道的 send）。 +18. 构建请求 → ChatTurnRequest(tenant_id, session_id, agent_id, user_id, message=_message_text(...), channel="feishu", client_turn_id=inbound.event_id)（service_intake.py:931）。 +18a. trace 卡片初始化（仅 channel_feishu_trace_enabled 开启且 binding.channel=="feishu"）→ FeishuTraceStreamer.start()（feishu_trace.py）调 adapter.create_card 创建"正在执行"交互式卡片（msg_type=interactive），保存 message_id；失败仅记日志不阻塞 turn。streamer.on_event 作为 event_sink 传入 AgentLoop。 +19. 执行对话轮 → AgentLoop(db, event_sink=streamer.on_event).handle_turn(request)（service_intake.py:962）→ HarnessV2Engine.run（harness_v2_engine.py:88）： +- 会话锁 acquire_harness_session + lease +- turn_store.claim 抢 turn（防并发同 client_turn_id 重放） +- _append_message(role="user") 落 Message 表 +- _conversation_context 取历史消息 +- planner.plan 生成 plan（含 router_decision） +- 跑 task frame / tool calls / materialize_task_attachments（若有附件，当前飞书入站 attachments=[]，这条链目前不走） +- LLM 产出回复 → _append_message(role="assistant") +- events.record 记 assistant_message_created 等，每个事件同步回调 event_sink → FeishuTraceStreamer.on_event → 重建 trace 行 → 节流 PATCH 更新"正在执行"卡片（仅 trace 开关开启时，见 18a） +20. typing 下行 → finally 里 _send_wechat_typing(..., state=2) 撤销输入态。 +20a. trace 卡片定格 → turn 正常结束调 streamer.finish() 把卡片定格为"执行完成"（green）；异常路径调 streamer.abort() 定格为"执行失败"（red）。卡片更新失败仅记日志。 +21. event 收尾 → event 置 done + processed_at（service_intake.py:959），提交。异常路径置 failed + _stage_error_notice。 + +## 阶段四：回复投递（service_outbox.py 守护线程） + +22. 回复登记：HarnessV2Engine 在产出 assistant 消息时，由 ResponseGenerator/delivery 桥接把回复文本登记为一条 ChannelDelivery(kind="reply", status="pending", target_json=)（service_outbox.py:62 / :214）。 +23. delivery 守护轮询 → run_delivery_daemon（service_outbox.py:657）→ _run_delivery_lane → _deliver_due（service_outbox.py:227）捞 pending 且到点的投递。 +24. 单条投递 → _deliver_one_locked（service_outbox.py:388）： +- 校验 binding active、session 与 binding 账号一致 +- get_channel_adapter("feishu") 取飞书适配器 +- adapter.send(binding, target, text, idempotency_key=...)（service_outbox.py:506）调用飞书 im/v1/messages 发送回复（target.message_id 用于回复式消息，或 receive_id 发新消息） +- 成功置 delivered；失败按 retryable + max_attempts 重试或置 failed +25. reaction 收尾：reply 投递成功后，若该 event 之前挂过"处理中" reaction，_event_has_delivered_response 为真，登记一条 kind="reaction_remove" 清掉标记（service_outbox.py:495）。 +链路全景图 +飞书用户发消息 + │ + ▼ +飞书 WS ──> 子进程 ProductionClient._handle_data_frame [feishu_runtime.py:180] + │ └> EventDispatcherHandler → receive [feishu_runtime.py:154] + │ └> _normalize_event [feishu_runtime.py:26] + │ └> stage_feishu_inbound [service_feishu_inbox.py:59] + │ └> 写 ChannelInboundEvent(received) + │ └> control.emit("INBOX_STAGED", event_pk) [feishu_runtime.py:171] + ▼ +父进程 FeishuProcessSupervisor._accept_event [feishu_process.py:215] + └> wake_staged_inbound_worker() [service_intake.py:1064] + ▼ +intake 守护线程 run_staged_inbound_daemon [service_intake.py:1068] + └> process_staged_inbound(event_pk) [service_intake.py:967] + ├> claim_staged_inbound (UPDATE→processing) [service_intake.py:167] + ├> _decode_and_validate_staged_event [service_intake.py:1125] + ├> _stage_received_reaction [service_intake.py:465] + └> process_inbound(binding, inbound, event_pk) [service_intake.py:707] + ├> parse_command（指令直接 _stage_notice 返回） + ├> resolve_or_provision_user + ├> resolve_current_agent / maybe_auto_route + ├> find_or_create_channel_session + ├> _send_wechat_typing(state=1) + ├> FeishuTraceStreamer.start()（trace 开关开启时） [feishu_trace.py] + │ └> adapter.create_card → 创建"正在执行"卡片，保存 message_id + ├> AgentLoop.handle_turn(ChatTurnRequest, event_sink=streamer.on_event) [agent_loop.py:136] + │ └> HarnessV2Engine.run [harness_v2_engine.py:88] + │ ├> acquire_harness_session + turn_store.claim + │ ├> _append_message(user) + │ ├> planner.plan + router_decision + │ ├> TaskFrame / tools / LLM + │ ├> events.record → event_sink → streamer.on_event → 节流 PATCH 更新卡片 + │ ├> materialize_task_attachments（当前 attachments=[] 不走） + │ └> _append_message(assistant) → 登记 ChannelDelivery(reply,pending) + ├> streamer.finish() / streamer.abort()（trace 开关开启时） + ├> _send_wechat_typing(state=2) + └> event → done + ▼ +delivery 守护线程 run_delivery_daemon [service_outbox.py:657] + └> _deliver_due → _deliver_one_locked [service_outbox.py:388] + └> adapter.send(binding, target, reply_text) [service_outbox.py:506] + └> 飞书 im/v1/messages 发回复 + └> 成功 → delivered；失败 → 重试/failed + └> 若有 reaction → 登记 reaction_remove 清标记 + ▼ +用户在飞书看到机器人回复 +关键设计点 +- 子进程隔离：每个 binding 一个 spawn 子进程跑 WS 长连接，崩溃/backoff/重启不影响主进程，由 FeishuProcessSupervisor 监控。 +- durable inbox：事件先落 ChannelInboundEvent(received) 再异步处理，即使进程在处理中崩溃，重启后 sweep_stale_inbound_events 会把 processing 遗留行回收重跑（service_intake.py:1243）。 +- 幂等：claim_staged_inbound 用 UPDATE WHERE status="received" 原子抢占，(binding_id, event_id) 唯一约束防重投递；turn 层 turn_store.claim 再防一次。 +- 串行：_session_lock(session_id) 保证同一会话同一时刻只有一个 turn 在跑。 +- 附件断点：当前 _normalize_event 在 message_type != "text" 时直接返回 None 丢弃，这正是 channel-attachments-plan.md 要改造的入口点。 +- 实时 trace 卡片：channel_feishu_trace_enabled 开启时，intake worker 在 handle_turn 前创建一张"正在执行"交互式卡片，通过 EventLog.event_sink 钩子把每个 trace 事件（SOP 匹配/步骤/工具/知识检索）实时渲染为可读行并节流 PATCH 更新卡片，结束后定格为完成/失败状态。卡片是"进度展示"，与正文回复（outbox 幂等投递）互不影响；卡片创建/更新失败仅记日志不阻塞 turn。开关关闭或非飞书渠道走原路径不创建卡片。 \ No newline at end of file diff --git a/backend/app/channels/feishu_trace.py b/backend/app/channels/feishu_trace.py new file mode 100644 index 00000000..09da9f46 --- /dev/null +++ b/backend/app/channels/feishu_trace.py @@ -0,0 +1,421 @@ +from __future__ import annotations + +import logging +import queue +import threading +import time +from datetime import UTC, datetime +from typing import Any + +from app.config import get_settings +from app.db.models import ChannelBinding, Skill + +logger = logging.getLogger(__name__) + +# 卡片更新最小间隔（秒）：规避飞书消息更新限流。 +_MIN_UPDATE_INTERVAL = 1.0 +# 单张卡片最多展示的步骤行数，超出截断尾部历史。 +_MAX_LINES = 60 +# 后台 worker 线程 join 超时（秒）：finish/abort 后等待最终卡片更新。 +_WORKER_JOIN_TIMEOUT = 8.0 + + +class _SinkEvent: + """轻量 AgentEvent 替身，仅供 _event_trace_lines 渲染使用。 + + EventLog.record 的 sink 收到的是 (event_type, payload_dict)，而 + _event_trace_lines 读取 event.event_type / event.payload_json / event.id / + event.created_at 四个字段。这里用一个最小对象补齐，避免构造完整 ORM 行。 + """ + + __slots__ = ("created_at", "event_type", "id", "payload_json") + + def __init__(self, event_type: str, payload: dict[str, Any]) -> None: + self.event_type = event_type + self.payload_json = payload + self.id = str(payload.get("turn_id") or payload.get("user_message_id") or "") + self.created_at = datetime.now(tz=UTC) + + +def _load_skill_names(db, tenant_id: str) -> dict[str, str]: + from sqlmodel import select + + rows = db.exec(select(Skill).where(Skill.tenant_id == tenant_id)).all() + return {row.skill_id: row.name for row in rows} + + +class FeishuTraceStreamer: + """飞书渠道实时执行步骤卡片流式器。 + + 生命周期： + start() → 后台创建"正在执行"卡片，保存 message_id + on_event → 累积 trace 行，节流后后台 PATCH 更新卡片 + finish() → 定格为完成状态，等待后台 worker 排空 + abort() → 异常路径定格为失败状态 + + 所有 HTTP I/O 在后台 worker 线程执行，on_event 不阻塞调用方 + （即 AgentLoop 主线程）。start/finish/abort 同样非阻塞， + finish/abort 会 join worker 最多 _WORKER_JOIN_TIMEOUT 秒以确保 + 最终卡片更新被尝试。 + + 全程 try/except 隔离：卡片创建/更新失败仅记日志，绝不抛出，不影响 turn + 成功与正文回复投递。 + """ + + def __init__( + self, + binding: ChannelBinding, + target: dict[str, Any], + turn_id: str, + *, + adapter: Any | None = None, + skill_names: dict[str, str] | None = None, + db=None, + min_update_interval: float = _MIN_UPDATE_INTERVAL, + ) -> None: + self._binding = binding + self._target = dict(target or {}) + self._turn_id = str(turn_id or "").strip() + self._adapter = adapter + self._skill_names = dict(skill_names or {}) + self._db = db + self._min_update_interval = max(0.1, float(min_update_interval)) + self._message_id: str | None = None + self._lines: list[dict] = [] + self._skill_hint: str | None = None + self._lock = threading.Lock() + self._last_update_at = 0.0 + self._dirty = False + self._finished = False + self._started = False + self._final_state: str | None = None + self._draining = False + self._card_created = False + + # 后台 worker + self._task_queue: queue.Queue[_Task | None] = queue.Queue() + self._worker: threading.Thread | None = None + self._worker_started = False + + # ---- 后台 worker ---- + + def _start_worker(self) -> None: + if self._worker_started: + return + self._worker_started = True + self._worker = threading.Thread( + target=self._worker_loop, name="feishu-trace-worker", daemon=True + ) + self._worker.start() + + def _worker_loop(self) -> None: + while True: + try: + task = self._task_queue.get(timeout=0.05) + except queue.Empty: + if self._draining: + return + continue + if task is None: + return + try: + task.execute(self) + except Exception: + logger.exception( + "飞书 trace worker 任务执行失败 binding=%s turn=%s", + self._binding.id, + self._turn_id, + ) + + def _stop_worker(self, *, timeout: float = _WORKER_JOIN_TIMEOUT) -> None: + if not self._worker_started or self._worker is None: + return + self._draining = True + self._worker.join(timeout=timeout) + self._worker_started = False + self._worker = None + + # ---- adapter / skill names ---- + + def _ensure_adapter(self): + if self._adapter is not None: + return self._adapter + from app.channels.adapters.base import get_channel_adapter + + self._adapter = get_channel_adapter("feishu") + return self._adapter + + def _ensure_skill_names(self) -> dict[str, str]: + if self._skill_names or self._db is None: + return self._skill_names + try: + self._skill_names = _load_skill_names(self._db, self._binding.tenant_id) + except Exception: + logger.exception("飞书 trace 流式器加载技能名称失败 tenant=%s", self._binding.tenant_id) + return self._skill_names + + # ---- 生命周期 ---- + + def start(self) -> None: + if self._started: + return + self._started = True + self._start_worker() + self._task_queue.put(_CreateCardTask()) + + def on_event(self, event_type: str, payload: dict[str, Any]) -> None: + if self._finished: + return + try: + self._ingest_event(event_type, payload) + self._maybe_enqueue_patch() + except Exception: + logger.exception( + "飞书 trace 事件处理失败 binding=%s turn=%s event=%s", + self._binding.id, + self._turn_id, + event_type, + ) + + def finish(self) -> None: + if self._finished: + return + self._finished = True + self._final_state = "completed" + with self._lock: + for line in self._lines: + if line.get("state") == "running": + line["state"] = "completed" + if self._message_id: + self._task_queue.put(_PatchCardTask(state="completed", force=True)) + self._stop_worker() + + def abort(self, reason: str | None = None) -> None: + if self._finished: + return + self._finished = True + self._final_state = "failed" + with self._lock: + for line in self._lines: + if line.get("state") == "running": + line["state"] = "failed" + if self._message_id: + self._task_queue.put(_PatchCardTask(state="failed", force=True)) + self._stop_worker() + logger.info( + "飞书 trace 流式器中止 binding=%s turn=%s reason=%s", + self._binding.id, + self._turn_id, + reason, + ) + + # ---- 事件处理 ---- + + def _ingest_event(self, event_type: str, payload: dict[str, Any]) -> None: + if event_type == "router_decision_created": + target_skill_id = str(payload.get("target_skill_id") or "").strip() + if target_skill_id: + self._skill_hint = target_skill_id + + from app.api.chat import _event_trace_lines + + sink_event = _SinkEvent(event_type, payload) + lines = _event_trace_lines(sink_event, self._ensure_skill_names(), self._skill_hint) + if not lines: + skill_context = _skill_context_from_payload(event_type, payload, self._skill_hint) + if skill_context: + self._skill_hint = skill_context + return + with self._lock: + for line in lines: + _upsert_line(self._lines, line) + if len(self._lines) > _MAX_LINES: + self._lines = self._lines[-_MAX_LINES:] + self._dirty = True + + skill_context = _skill_context_from_payload(event_type, payload, self._skill_hint) + if skill_context: + self._skill_hint = skill_context + + def _maybe_enqueue_patch(self) -> None: + if not self._message_id: + return + with self._lock: + if not self._dirty: + return + now = time.monotonic() + if (now - self._last_update_at) < self._min_update_interval: + return + self._dirty = False + self._last_update_at = now + lines_snapshot = list(self._lines) + self._task_queue.put(_PatchCardTask(lines=lines_snapshot, state="running", force=False)) + + # ---- 卡片操作（在 worker 线程执行）---- + + def _do_create_card(self) -> None: + try: + adapter = self._ensure_adapter() + card = self._render_card(state="running") + idempotency_key = f"feishu-trace:{self._binding.id}:{self._turn_id}" + self._message_id = adapter.create_card( + self._binding, self._target, card, idempotency_key=idempotency_key + ) + except Exception: + logger.exception( + "飞书 trace 卡片创建失败 binding=%s turn=%s", self._binding.id, self._turn_id + ) + self._message_id = None + finally: + self._card_created = True + # 卡片创建成功后，处理累积行或最终状态 + if self._message_id: + if self._final_state is not None: + # finish/abort 已被调用，直接发送最终状态 + with self._lock: + lines_snapshot = list(self._lines) + self._task_queue.put( + _PatchCardTask(lines=lines_snapshot, state=self._final_state, force=True) + ) + else: + with self._lock: + if self._dirty and self._lines: + self._dirty = False + self._last_update_at = time.monotonic() + lines_snapshot = list(self._lines) + else: + lines_snapshot = None + if lines_snapshot is not None: + self._task_queue.put( + _PatchCardTask(lines=lines_snapshot, state="running", force=False) + ) + + def _do_patch_card(self, lines: list[dict] | None, *, state: str, force: bool) -> None: + if not self._message_id: + return + try: + adapter = self._ensure_adapter() + if lines is None: + with self._lock: + lines = list(self._lines) + card = self._render_card(lines=lines, state=state) + adapter.update_card(self._binding, self._message_id, card) + except Exception: + logger.exception( + "飞书 trace 卡片更新失败 binding=%s message_id=%s", + self._binding.id, + self._message_id, + ) + + # ---- 卡片渲染 ---- + + def _render_card( + self, + *, + lines: list[dict] | None = None, + state: str = "running", + ) -> dict[str, Any]: + header_title = "正在思考…" + header_template = "blue" + if state == "completed": + header_title = "执行完成" + header_template = "green" + elif state == "failed": + header_title = "执行失败" + header_template = "red" + + elements: list[dict[str, Any]] = [] + display_lines = lines if lines is not None else [] + for line in display_lines: + elements.append(_line_to_card_element(line)) + if not display_lines: + elements.append({"tag": "div", "text": {"tag": "plain_text", "content": "等待执行步骤…"}}) + + return { + "config": {"wide_screen_mode": True}, + "header": { + "title": {"tag": "plain_text", "content": header_title}, + "template": header_template, + }, + "elements": elements, + } + + +# ---- 后台任务 ---- + + +class _Task: + """worker 线程执行的抽象任务。""" + + def execute(self, streamer: FeishuTraceStreamer) -> None: + raise NotImplementedError + + +class _CreateCardTask(_Task): + def execute(self, streamer: FeishuTraceStreamer) -> None: + streamer._do_create_card() + + +class _PatchCardTask(_Task): + __slots__ = ("force", "lines", "state") + + def __init__(self, *, lines: list[dict] | None = None, state: str = "running", force: bool = False) -> None: + self.lines = lines + self.state = state + self.force = force + + def execute(self, streamer: FeishuTraceStreamer) -> None: + streamer._do_patch_card(self.lines, state=self.state, force=self.force) + + +def _line_to_card_element(line: dict) -> dict[str, Any]: + text = str(line.get("text") or "").strip() + detail = str(line.get("detail") or "").strip() + state = str(line.get("state") or "").strip() + icon = _state_icon(state) + content_parts = [f"{icon} {text}" if icon else text] + if detail: + content_parts.append(detail) + content = "\n".join(part for part in content_parts if part) + return {"tag": "div", "text": {"tag": "lark_md", "content": content}} + + +def _state_icon(state: str) -> str: + if state == "completed": + return "✅" + if state == "failed": + return "❌" + if state == "running": + return "⏳" + return "" + + +def _upsert_line(lines: list[dict], line: dict) -> None: + line_id = str(line.get("id") or "").strip() + if line_id: + for index, existing in enumerate(lines): + if str(existing.get("id") or "") == line_id: + lines[index] = {**existing, **line} + return + lines.append(line) + + +def _skill_context_from_payload( + event_type: str, + payload: dict[str, Any], + skill_hint: str | None, +) -> str | None: + if event_type in {"skill_started", "skill_resumed", "skill_step_changed"}: + to_skill_id = str(payload.get("to_skill_id") or "").strip() + from_skill_id = str(payload.get("from_skill_id") or "").strip() + return to_skill_id or from_skill_id or skill_hint or None + return None + + +def is_feishu_trace_enabled(binding: ChannelBinding | None) -> bool: + if not binding or binding.channel != "feishu": + return False + if not get_settings().channel_feishu_trace_enabled: + return False + config = binding.config_json or {} + return not (isinstance(config, dict) and config.get("trace_enabled") is False) diff --git a/backend/app/channels/markdown_render.py b/backend/app/channels/markdown_render.py new file mode 100644 index 00000000..50457979 --- /dev/null +++ b/backend/app/channels/markdown_render.py @@ -0,0 +1,743 @@ +"""Markdown 子集解析器 + 飞书 post 富文本渲染。 + +设计目标（见 channel-render-plan-feishu-dingtalk.md §3.2）： +- 零新增依赖，手写解析器覆盖受控子集； +- 输出通用块模型供飞书渲染器消费（钉钉原生 markdown 直接透传，不走块模型）； +- `has_markdown(text)` 做语法检测，纯文本返回 False 以走原 text 路径，避免回归。 + +覆盖子集：标题 / 粗体 / 斜体 / 行内代码 / 围栏代码块 / 链接 / 有序无序列表 / 引用 / 分隔线。 +不支持的（表格、HTML、不闭合围栏等）按纯文本处理，绝不抛异常。 +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field + + +@dataclass +class Span: + """行内文本片段。styles 为空集合表示普通文本。""" + + text: str + styles: frozenset[str] = field(default_factory=frozenset) + href: str = "" + + +@dataclass +class Heading: + level: int + spans: list[Span] + + +@dataclass +class Paragraph: + spans: list[Span] + + +@dataclass +class CodeBlock: + language: str + text: str + + +@dataclass +class ListItem: + ordered: bool + index: int + spans: list[Span] + + +@dataclass +class Quote: + spans: list[Span] + + +@dataclass +class ThematicBreak: + pass + + +@dataclass +class TableBlock: + """表格降级：按纯文本行保留，飞书 post 不支持表格。""" + + lines: list[str] + + +Block = Heading | Paragraph | CodeBlock | ListItem | Quote | ThematicBreak | TableBlock + + +# 行级正则 +_HEADING_RE = re.compile(r"^(#{1,6})\s+(.*?)\s*#*\s*$") +_FENCE_RE = re.compile(r"^(`{3,}|~{3,})\s*([\w+-]*)\s*$") +_UNORDERED_RE = re.compile(r"^\s*[-*+]\s+(.*)$") +_ORDERED_RE = re.compile(r"^\s*(\d+)\.\s+(.*)$") +_QUOTE_RE = re.compile(r"^\s*>\s?(.*)$") +_HR_RE = re.compile(r"^\s*([-*_])\1{2,}\s*$") +_TABLE_SEP_RE = re.compile(r"^\s*\|?[\s:|-]+\|?\s*$") +_INDENT_CODE_RE = re.compile(r"^( |\t)(.*)$") + +# 顶格代码起始行检测：def/class/import/from/if __name__/function/func 等明显的代码关键字 +_CODE_START_RE = re.compile( + r"^(def |class |import |from \S+ import |if __name__|async def |@)" +) + +# 行内正则 +_INLINE_CODE_RE = re.compile(r"`([^`\n]+)`") +_BOLD_RE = re.compile(r"\*\*([^*\n]+?)\*\*|__([^_\n]+?)__") +_ITALIC_RE = re.compile(r"(?\s?\S"), # 引用 + re.compile(r"(?m)^\s{0,3}([-*_])\1{2,}\s*$"), # 分隔线 + re.compile(r"(?m)^\s*```"), # 围栏代码块 + re.compile(r"(?m)^(def |class |import |from \S+ import |if __name__|async def |@)"), # 顶格代码 + re.compile(r"\*\*[^*\n]+\*\*|__[^_\n]+__"), # 粗体 + re.compile(r"`[^`\n]+`"), # 行内代码 + re.compile(r"\[[^\]]*\]\([^)\s]+\)"), # 链接 +] + + +def has_markdown(text: str) -> bool: + """检测文本是否含 markdown 语法标记。 + + 纯文本（含普通破折号、单星号装饰但不构成语法）应返回 False，避免外观变化。 + 斜体 `*x*` / `_x_` 不参与检测，因其误判率高（如 "a * b"、文件名 a_b_c）。 + """ + if not text: + return False + return any(pattern.search(text) for pattern in _MD_DETECT_PATTERNS) + + +def _is_markdown_block_line(line: str) -> bool: + """判断一行是否是 markdown 块级语法行（标题/围栏/列表/引用/分隔线）。 + + 用于顶格代码块收集时判断后续行是否属于代码还是 markdown 语法。 + """ + return bool( + _HEADING_RE.match(line) + or _FENCE_RE.match(line) + or _HR_RE.match(line) + or _UNORDERED_RE.match(line) + or _ORDERED_RE.match(line) + or _QUOTE_RE.match(line) + ) + + +def _is_code_continuation(line: str) -> bool: + """判断一行是否可以作为顶格代码块的续行。 + + 保守策略：接受缩进行、代码起始行、以 # 开头的注释行， + 以及顶格的赋值/函数调用行（含 = 或以 print/return/await/yield 开头）。 + 其他顶格行（自然语言段落等）视为代码块结束。 + """ + if not line.strip(): + return False + if _INDENT_CODE_RE.match(line): + return True + if _CODE_START_RE.match(line): + return True + # # 开头的行在代码上下文中是注释，不是标题 + if re.match(r"^#\s", line): + return True + # 顶格赋值行：var = ... / var: type = ... + stripped = line.strip() + if re.match(r"^\w[\w.]*\s*[:=]", stripped): + return True + # 顶格函数调用行：print(...) / foo(...) / await ... + if re.match(r"^(print|return|await|yield|raise|break|continue)\b", stripped): + return True + return bool(re.match(r"^\w[\w.]*\s*\(", stripped)) + + +def parse_markdown(text: str) -> list[Block]: + """把 markdown 文本解析为块模型列表。不抛异常，无法解析的行降级为 Paragraph。""" + if not text: + return [] + lines = text.split("\n") + blocks: list[Block] = [] + i = 0 + n = len(lines) + while i < n: + line = lines[i] + # 围栏代码块 + fence_match = _FENCE_RE.match(line) + if fence_match: + fence_marker = fence_match.group(1) + fence_char = re.escape(fence_marker[0]) + language = fence_match.group(2) or "" + code_lines: list[str] = [] + i += 1 + while i < n: + cur = lines[i] + if re.match(rf"^\s*{fence_char}{{3,}}\s*$", cur): + i += 1 + break + code_lines.append(cur) + i += 1 + # 不闭合围栏：把已收集的行作为代码块返回，飞书仍可渲染 + blocks.append(CodeBlock(language=language, text="\n".join(code_lines))) + continue + + # 分隔线 + if _HR_RE.match(line): + blocks.append(ThematicBreak()) + i += 1 + continue + + # 表格（含分隔行 |---|） + if "|" in line and i + 1 < n and _TABLE_SEP_RE.match(lines[i + 1]): + table_lines: list[str] = [] + table_lines.append(line.strip()) + i += 1 + table_lines.append(lines[i].strip()) # 分隔行 + i += 1 + while i < n and "|" in lines[i] and lines[i].strip(): + table_lines.append(lines[i].strip()) + i += 1 + blocks.append(TableBlock(lines=table_lines)) + continue + + # 标题 + heading_match = _HEADING_RE.match(line) + if heading_match: + level = len(heading_match.group(1)) + content = heading_match.group(2) + spans = _parse_inline(content) + blocks.append(Heading(level=level, spans=spans)) + i += 1 + continue + + # 引用（连续行合并为单个 Quote） + quote_match = _QUOTE_RE.match(line) + if quote_match: + quote_text_parts: list[str] = [quote_match.group(1)] + i += 1 + while i < n: + qm = _QUOTE_RE.match(lines[i]) + if not qm: + break + quote_text_parts.append(qm.group(1)) + i += 1 + spans = _parse_inline(" ".join(part for part in quote_text_parts if part)) + blocks.append(Quote(spans=spans)) + continue + + # 无序列表（连续项各自成块，便于飞书分行） + unordered_match = _UNORDERED_RE.match(line) + if unordered_match: + spans = _parse_inline(unordered_match.group(1)) + blocks.append(ListItem(ordered=False, index=0, spans=spans)) + i += 1 + continue + + # 有序列表 + ordered_match = _ORDERED_RE.match(line) + if ordered_match: + idx = int(ordered_match.group(1)) + spans = _parse_inline(ordered_match.group(2)) + blocks.append(ListItem(ordered=True, index=idx, spans=spans)) + i += 1 + continue + + # 空行 + if not line.strip(): + i += 1 + continue + + # 缩进代码块（4 空格或 tab）：连续缩进行（含中间空行）合并为代码块 + if _INDENT_CODE_RE.match(line): + code_lines: list[str] = [] + while i < n: + cur = lines[i] + m = _INDENT_CODE_RE.match(cur) + if m: + code_lines.append(m.group(2)) + i += 1 + elif not cur.strip(): + # 收集连续空行，看后面是否还有缩进行 + blank_start = i + while i < n and not lines[i].strip(): + i += 1 + if i < n and _INDENT_CODE_RE.match(lines[i]): + code_lines.extend([""] * (i - blank_start)) + else: + i = blank_start + break + else: + break + blocks.append(CodeBlock(language="", text="\n".join(code_lines))) + continue + + # 顶格代码块：以 def/class/import/from import/if __name__/async def/@decorator 开头 + # 收集该行及后续行，直到遇到空行后明显非代码的内容 + if _CODE_START_RE.match(line): + code_lines = [line] + i += 1 + while i < n: + cur = lines[i] + if not cur.strip(): + # 空行：向前看，如果后续行仍是代码，则保留空行继续 + blank_start = i + while i < n and not lines[i].strip(): + i += 1 + if i < n and _is_code_continuation(lines[i]): + code_lines.extend([""] * (i - blank_start)) + else: + i = blank_start + break + elif _is_code_continuation(cur): + code_lines.append(cur) + i += 1 + else: + break + blocks.append(CodeBlock(language="", text="\n".join(code_lines))) + continue + + # 普通段落（连续非空非块行合并） + para_lines = [line] + i += 1 + while i < n: + cur = lines[i] + if not cur.strip(): + break + if ( + _HEADING_RE.match(cur) + or _FENCE_RE.match(cur) + or _HR_RE.match(cur) + or _UNORDERED_RE.match(cur) + or _ORDERED_RE.match(cur) + or _QUOTE_RE.match(cur) + or _INDENT_CODE_RE.match(cur) + or _CODE_START_RE.match(cur) + ): + break + para_lines.append(cur) + i += 1 + spans = _parse_inline("\n".join(para_lines)) + blocks.append(Paragraph(spans=spans)) + return blocks + + +def _parse_inline(text: str) -> list[Span]: + """解析行内标记：粗体 / 斜体 / 行内代码 / 链接。 + + 采用 token 扫描法：按最早出现的标记切分，递归处理。代码片段内的内容不二次解析。 + """ + if not text: + return [] + # 先抽取行内代码片段为占位，避免其内部被粗体/斜体/链接误解析 + code_segments: list[str] = [] + + def _stash_code(match: re.Match[str]) -> str: + code_segments.append(match.group(1)) + return f"\x00CODE{len(code_segments) - 1}\x00" + + work = _INLINE_CODE_RE.sub(_stash_code, text) + spans = _parse_inline_recursive(work, code_segments, set()) + return spans + + +def _parse_inline_recursive( + text: str, code_segments: list[str], styles: frozenset[str] +) -> list[Span]: + """递归解析行内标记。""" + spans: list[Span] = [] + pos = 0 + # 合并所有可能的起始标记，按位置排序处理 + patterns = [ + ("bold", _BOLD_RE), + ("italic", _ITALIC_RE), + ("link", _LINK_RE), + ] + while pos < len(text): + earliest: tuple[int, str, re.Match[str]] | None = None + for kind, pattern in patterns: + match = pattern.search(text, pos) + if match and (earliest is None or match.start() < earliest[0]): + earliest = (match.start(), kind, match) + # 代码占位回填 + code_placeholder_match = _CODE_PLACEHOLDER_RE.search(text, pos) + if code_placeholder_match: + cp_start = code_placeholder_match.start() + if earliest is None or cp_start < earliest[0]: + earliest = (cp_start, "code", code_placeholder_match) + + if earliest is None: + # 剩余纯文本 + rest = text[pos:] + if rest: + spans.append(Span(text=_restore_code(rest, code_segments), styles=styles)) + break + + start, kind, match = earliest + # 前导文本 + if start > pos: + leading = text[pos:start] + if leading: + spans.append(Span(text=_restore_code(leading, code_segments), styles=styles)) + + if kind == "code": + idx = int(match.group(1)) + spans.append( + Span(text=code_segments[idx], styles=styles | frozenset({"code"})) + ) + pos = match.end() + elif kind == "bold": + inner = match.group(1) if match.group(1) is not None else match.group(2) + spans.extend( + _parse_inline_recursive(inner, code_segments, styles | frozenset({"bold"})) + ) + pos = match.end() + elif kind == "italic": + inner = match.group(1) if match.group(1) is not None else match.group(2) + spans.extend( + _parse_inline_recursive(inner, code_segments, styles | frozenset({"italic"})) + ) + pos = match.end() + elif kind == "link": + label = match.group(1) + href = match.group(2) + label_spans = _parse_inline_recursive(label, code_segments, styles) + if label_spans: + for sp in label_spans: + sp.href = href + spans.extend(label_spans) + else: + spans.append(Span(text=href, styles=styles, href=href)) + pos = match.end() + return spans + + +def _restore_code(text: str, code_segments: list[str]) -> str: + """把代码占位符还原为实际代码文本。""" + return _CODE_PLACEHOLDER_RE.sub( + lambda m: code_segments[int(m.group(1))], + text, + ) + + +# --------------------------------------------------------------------------- +# 飞书 post 富文本渲染 +# --------------------------------------------------------------------------- + +def render_feishu_post(blocks: list[Block]) -> dict: + """把块模型渲染为飞书 post 消息的 content 结构（zh_cn 包裹）。 + + 返回形如： + {"zh_cn": {"title": "", "content": [[{tag...}, ...], ...]}} + + 每个块对应 content 数组中的一个"行"（tag 数组）。 + """ + content_rows: list[list[dict]] = [] + for block in blocks: + row = _block_to_feishu_row(block) + if row is not None: + content_rows.append(row) + return {"zh_cn": {"title": "", "content": content_rows}} + + +def _span_to_feishu_tag(span: Span) -> dict: + styles = span.styles + style_flags = [] + if "bold" in styles: + style_flags.append("bold") + if "italic" in styles: + style_flags.append("italic") + if span.href: + tag: dict = {"tag": "a", "text": span.text, "href": span.href} + elif "code" in styles: + tag = {"tag": "text", "text": span.text, "un_escape": False, "style": ["code"]} + return tag + else: + tag = {"tag": "text", "text": span.text, "un_escape": False} + if style_flags: + tag["style"] = style_flags + return tag + + +def _spans_to_feishu_tags(spans: list[Span]) -> list[dict]: + tags: list[dict] = [] + for span in spans: + if not span.text and not span.href: + continue + tags.append(_span_to_feishu_tag(span)) + return tags + + +def _block_to_feishu_row(block: Block) -> list[dict] | None: + if isinstance(block, Heading): + tags = _spans_to_feishu_tags(block.spans) + for tag in tags: + existing = tag.get("style") or [] + tag["style"] = list(dict.fromkeys(["bold", *existing])) + return tags or [{"tag": "text", "text": "", "un_escape": False}] + if isinstance(block, Paragraph): + tags = _spans_to_feishu_tags(block.spans) + if not tags: + return [{"tag": "text", "text": "", "un_escape": False}] + # 段落内若含换行（多行合并），拆成多行 + return _split_paragraph_newlines(tags) + if isinstance(block, Quote): + tags = _spans_to_feishu_tags(block.spans) + for tag in tags: + tag["text"] = f"｜{tag.get('text', '')}" + return tags or [{"tag": "text", "text": "｜", "un_escape": False}] + if isinstance(block, ListItem): + prefix = f"{block.index}. " if block.ordered else "• " + tags = _spans_to_feishu_tags(block.spans) + if tags: + first = tags[0] + first["text"] = f"{prefix}{first.get('text', '')}" + else: + tags = [{"tag": "text", "text": prefix, "un_escape": False}] + return tags + if isinstance(block, CodeBlock): + return [ + { + "tag": "code_block", + "language": block.language or "", + "text": block.text, + } + ] + if isinstance(block, ThematicBreak): + return [{"tag": "text", "text": "———", "un_escape": False}] + if isinstance(block, TableBlock): + # 表格降级为纯文本行 + return [{"tag": "text", "text": "\n".join(block.lines), "un_escape": False}] + return None + + +def _split_paragraph_newlines(tags: list[dict]) -> list[dict]: + """段落 spans 内若含 \n，拆成多行 tag（飞书 post 一行内不渲染换行）。""" + out: list[dict] = [] + for tag in tags: + text = tag.get("text", "") + if "\n" not in text: + out.append(tag) + continue + parts = text.split("\n") + for part in parts: + new_tag = dict(tag) + new_tag["text"] = part + out.append(new_tag) + return out if out else [{"tag": "text", "text": "", "un_escape": False}] + + +def _detect_fence(line: str) -> tuple[str, str] | None: + """若 line 是围栏起始/闭合行，返回 (fence_marker, language)；否则返回 None。 + + language 仅在起始行有意义，闭合行时为空字符串。 + """ + match = _FENCE_RE.match(line) + if not match: + return None + return match.group(1), match.group(2) or "" + + +def _is_fence_close(line: str, fence_marker: str) -> bool: + """判断 line 是否是对应围栏的闭合行。""" + fence_char = re.escape(fence_marker[0]) + return bool(re.match(rf"^\s*{fence_char}{{3,}}\s*$", line)) + + +def split_markdown_by_lines(text: str, limit: int) -> list[str]: + """按行边界切分 markdown，保证不破坏围栏代码块。 + + 用于富文本路径的分块：优先在空行处切分，其次在普通行边界。 + 关键约束：当切分点落在围栏代码块内部时，会在前一段末尾补上闭合围栏、 + 在后一段开头重新打开同语言的围栏，使每个 chunk 都是合法且自包含的 Markdown， + 飞书 parse_markdown 与钉钉 ensure_code_fences 均可独立正确处理。 + + 单行超限时硬切该行（此时围栏已在行外，不会出现半截围栏问题）。 + """ + if not text: + return [] + if len(text) <= limit: + return [text] + lines = text.split("\n") + chunks: list[str] = [] + current_lines: list[str] = [] + current_len = 0 + # 当前处于围栏代码块内时的状态：None 表示不在代码块内 + fence_state: tuple[str, str] | None = None # (fence_marker, language) + + def _flush() -> None: + nonlocal current_lines, current_len + if not current_lines: + return + chunk_lines = list(current_lines) + # 若切分时处于围栏代码块内部，补上闭合围栏使本段自包含 + if fence_state is not None: + chunk_lines.append(fence_state[0]) + chunks.append("\n".join(chunk_lines).rstrip("\n")) + current_lines = [] + current_len = 0 + + for line in lines: + line_with_newline = line + "\n" + line_len = len(line_with_newline) + + # 检测围栏状态变化（在长度判断之前，确保 fence_state 准确） + if fence_state is None: + fence_open = _detect_fence(line) + else: + fence_open = None + + # 单行超长：先 flush 已累积内容，再硬切该行。 + # 注意：若该行是围栏起始行本身超长，不会发生（围栏行很短）； + # 若是代码块内的一行超长，flush 会闭合围栏，然后硬切该代码行。 + if line_len > limit: + _flush() + remaining = line + if fence_state is not None: + # 围栏内的超长行：每个硬切片段都需包裹围栏，确保独立可渲染为代码块 + fence_marker, lang = fence_state + open_fence = f"{fence_marker}{lang}" if lang else fence_marker + while len(remaining) > limit: + chunk_content = remaining[:limit] + chunks.append(f"{open_fence}\n{chunk_content}\n{fence_marker}") + remaining = remaining[limit:] + if remaining: + reopen = open_fence + current_lines = [reopen, remaining] + current_len = len(current_lines[0]) + 1 + len(remaining) + 1 + else: + while len(remaining) > limit: + chunks.append(remaining[:limit]) + remaining = remaining[limit:] + if remaining: + current_lines = [remaining] + current_len = len(remaining) + 1 + # 围栏状态不因普通代码行变化 + continue + + if current_len + line_len > limit: + _flush() + # 新段开头：若仍处于围栏代码块内，重新打开围栏 + if fence_state is not None: + fence_marker, lang = fence_state + open_fence = f"{fence_marker}{lang}" if lang else fence_marker + current_lines = [open_fence, line] + current_len = len(open_fence) + 1 + line_len + else: + current_lines = [line] + current_len = line_len + else: + current_lines.append(line) + current_len += line_len + + # 更新围栏状态 + if fence_open is not None: + fence_state = fence_open + elif fence_state is not None and _is_fence_close(line, fence_state[0]): + fence_state = None + + _flush() + return [c for c in chunks if c] + + +def extract_dingtalk_title(text: str, *, max_length: int = 20) -> str: + """从 markdown 文本提取钉钉 markdown 消息的 title。 + + 规则：首个 `#` 标题文本 → 否则首个非空行 → 否则默认 "消息"；截断 ≤ max_length 字。 + """ + if not text or not text.strip(): + return "消息" + for line in text.split("\n"): + match = _HEADING_RE.match(line) + if match: + title = match.group(2).strip() + if title: + return title[:max_length] + for line in text.split("\n"): + stripped = line.strip() + if stripped: + return stripped[:max_length] + return "消息" + + +def ensure_code_fences(text: str) -> str: + """为缺少围栏的代码块补上 ``` 围栏，使钉钉等原生 markdown 渲染器正确识别。 + + 利用 parse_markdown 的块识别能力定位代码块区间，然后检查原始文本中 + 对应位置是否已有围栏；没有则插入 ``` 围栏。 + """ + if not text or not text.strip(): + return text + blocks = parse_markdown(text) + lines = text.split("\n") + # 收集需要加围栏的代码块行区间 [start, end)（0-based，原始行号） + fence_ranges: list[tuple[int, int, str]] = [] + idx = 0 + for block in blocks: + if not isinstance(block, CodeBlock): + continue + # 跳过已有围栏的代码块（``` 或 ~~~ 开头） + # 找到该代码块在原始行中的起始位置 + block_text_lines = block.text.split("\n") if block.text else [] + block_line_count = len(block_text_lines) + # 从 idx 开始搜索代码块的起始行 + start = _find_block_start(lines, idx, block, block_text_lines) + if start is None: + continue + end = start + block_line_count + # 检查是否已有围栏（前一非空行是 ``` 或 ~~~） + if _has_fence_before(lines, start): + idx = end + continue + language = block.language or "" + fence_ranges.append((start, end, language)) + idx = end + + if not fence_ranges: + return text + + # 从后往前插入围栏，避免行号偏移 + result_lines = list(lines) + for start, end, language in reversed(fence_ranges): + fence_open = f"```{language}" if language else "```" + # 插入闭合围栏（在 end 位置，即代码块最后一行之后） + result_lines.insert(end, "```") + # 插入开启围栏（在 start 位置） + result_lines.insert(start, fence_open) + return "\n".join(result_lines) + + +def _find_block_start( + lines: list[str], from_idx: int, block: CodeBlock, block_lines: list[str] +) -> int | None: + """在 lines 中从 from_idx 开始查找 CodeBlock 对应的起始行号。""" + if not block_lines: + return None + first_code_line = block_lines[0].rstrip() + for i in range(from_idx, len(lines)): + if lines[i].rstrip() == first_code_line: + # 验证后续行是否匹配 + match = True + for j, bl in enumerate(block_lines): + if i + j >= len(lines): + match = False + break + if lines[i + j].rstrip() != bl.rstrip(): + match = False + break + if match: + return i + return None + + +def _has_fence_before(lines: list[str], code_start: int) -> bool: + """检查代码块起始行之前是否已有围栏标记（``` 或 ~~~）。""" + for i in range(code_start - 1, -1, -1): + if not lines[i].strip(): + continue + return bool(_FENCE_RE.match(lines[i])) + return False diff --git a/backend/app/channels/service_intake.py b/backend/app/channels/service_intake.py index f18a5d46..26e1614b 100644 --- a/backend/app/channels/service_intake.py +++ b/backend/app/channels/service_intake.py @@ -958,10 +958,23 @@ def process_inbound( attachments=attachments, ) _send_wechat_typing(binding, inbound.from_user_id, inbound.context_token, 1, db_engine=use_engine) + from app.channels.feishu_trace import FeishuTraceStreamer, is_feishu_trace_enabled + + trace_streamer: FeishuTraceStreamer | None = None + if is_feishu_trace_enabled(binding): + trace_streamer = FeishuTraceStreamer( + binding, + target, + inbound.event_id, + db=db, + ) + trace_streamer.start() try: - AgentLoop(db).handle_turn(request) + AgentLoop(db, event_sink=trace_streamer.on_event if trace_streamer else None).handle_turn(request) except Exception as exc: logger.exception("渠道入站处理失败 binding=%s event=%s", binding.id, inbound.event_id) + if trace_streamer: + trace_streamer.abort(str(exc)[:200]) db.rollback() event = db.get(ChannelInboundEvent, event_id) chat_session = db.get(ChatSession, session_id) @@ -974,6 +987,9 @@ def process_inbound( _stage_error_notice(db, binding, chat_session) db.commit() return False + else: + if trace_streamer: + trace_streamer.finish() finally: _send_wechat_typing(binding, inbound.from_user_id, inbound.context_token, 2, db_engine=use_engine) event.status = "done" diff --git a/backend/app/config.py b/backend/app/config.py index 3a16054e..84c4bb0f 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -41,6 +41,13 @@ class Settings(BaseSettings): # 钉钉 emotion 接口的表情常量与所需权限尚未真机验证，验证通过前默认关闭： # 否则常量失效或权限未开时，每条入站消息都会留下一条失败的 reaction 投递。 channel_dingtalk_reaction_enabled: bool = False + # 出站富文本渲染开关：开启时飞书走 post 富文本、钉钉走 markdown 消息； + # 关闭时两者回退为纯 text 消息，用于快速回退。 + channel_rich_render_enabled: bool = True + # 飞书渠道实时执行步骤卡片开关：开启后飞书对话在执行过程中创建并实时更新 + # 一张独立卡片展示智能体每一步（SOP/工具/知识检索），与正文回复互不影响。 + # 仅影响飞书渠道；关闭时退化为仅发最终回复。 + channel_feishu_trace_enabled: bool = True model_config = SettingsConfigDict( env_file=_os.environ.get("ULTRARAG_DOTENV", ".env"), diff --git a/backend/app/core/agent_loop.py b/backend/app/core/agent_loop.py index d8a82599..d4813099 100644 --- a/backend/app/core/agent_loop.py +++ b/backend/app/core/agent_loop.py @@ -119,9 +119,14 @@ def __init__(self, code: str, message: str): class AgentLoop: - def __init__(self, db: Session) -> None: + def __init__( + self, + db: Session, + *, + event_sink: Callable[[str, dict[str, Any]], None] | None = None, + ) -> None: self.db = db - self.events = EventLog(db) + self.events = EventLog(db, event_sink=event_sink) self.runtime = SkillRuntime() self.response_generator = ResponseGenerator() self.memory = MemoryService(db) diff --git a/backend/app/observability/event_log.py b/backend/app/observability/event_log.py index 9ef11880..fcc5c2cf 100644 --- a/backend/app/observability/event_log.py +++ b/backend/app/observability/event_log.py @@ -1,15 +1,25 @@ from __future__ import annotations +import logging +from collections.abc import Callable from typing import Any from sqlmodel import Session from app.db.models import AgentEvent +logger = logging.getLogger(__name__) + class EventLog: - def __init__(self, db: Session): + def __init__( + self, + db: Session, + *, + event_sink: Callable[[str, dict[str, Any]], None] | None = None, + ): self.db = db + self._event_sink = event_sink self._turn_id: str | None = None self._client_turn_id: str | None = None @@ -31,4 +41,9 @@ def record(self, tenant_id: str, session_id: str, event_type: str, payload: dict payload_json=traced_payload, ) self.db.add(event) + if self._event_sink is not None: + try: + self._event_sink(event_type, traced_payload) + except Exception: + logger.exception("event_sink 调用失败 event_type=%s", event_type) return event diff --git a/backend/tests/test_channel_autoroute.py b/backend/tests/test_channel_autoroute.py index 56ab622b..855738c4 100644 --- a/backend/tests/test_channel_autoroute.py +++ b/backend/tests/test_channel_autoroute.py @@ -364,7 +364,7 @@ def _p2p_message(event_id: str, text: str) -> dict: class RecordingAgentLoop: calls: list = [] - def __init__(self, db): + def __init__(self, db, *, event_sink=None): self.db = db def handle_turn(self, request): diff --git a/backend/tests/test_channel_bind.py b/backend/tests/test_channel_bind.py index 6ad08e95..1a870032 100644 --- a/backend/tests/test_channel_bind.py +++ b/backend/tests/test_channel_bind.py @@ -160,7 +160,7 @@ def _make_lazy_account(engine) -> User: class RecordingAgentLoop: calls: list = [] - def __init__(self, db): + def __init__(self, db, *, event_sink=None): self.db = db def handle_turn(self, request): diff --git a/backend/tests/test_channel_dingtalk.py b/backend/tests/test_channel_dingtalk.py index 8716b12d..7861fd5c 100644 --- a/backend/tests/test_channel_dingtalk.py +++ b/backend/tests/test_channel_dingtalk.py @@ -10,6 +10,7 @@ DINGTALK_ACK_EMOTION_ID, DINGTALK_ACK_EMOTION_NAME, DINGTALK_REACTION_HANDLE, + DINGTALK_TEXT_LIMIT, DingTalkAdapter, DingTalkPermanentError, DingTalkTokenProvider, @@ -713,3 +714,161 @@ def test_envelope_round_trips_attachments_as_dataclass() -> None: assert decoded.attachments[0].download_params["download_code"] == "dc_001" assert decoded.attachments[1].filename == "report.xlsx" assert decoded.attachments[1].kind == "file" + + +# --------------------------------------------------------------------------- +# 富文本（markdown）渲染 — channel-render-plan §5.3 +# --------------------------------------------------------------------------- + + +class _WebhookClient: + """记录所有 webhook POST 请求的假 client。""" + + def __init__(self, response=None): + self.calls = [] + self._response = response or _Response(200, {"errcode": 0}) + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def post(self, url, json=None, headers=None, **_kwargs): + self.calls.append({"url": url, "body": json, "headers": headers or {}}) + return self._response + + +def _send_binding(): + return ChannelBinding( + tenant_id="t", + agent_id="a", + channel="dingtalk", + config_json={"client_id": "client-1"}, + credentials_enc=encrypt_channel_secret("secret"), + ) + + +def _send_target(): + return {"session_webhook": "https://oapi.dingtalk.com/robot/send?session=x"} + + +def test_dingtalk_markdown_render_uses_markdown_msgtype(): + client = _WebhookClient() + adapter = DingTalkAdapter(client_factory=lambda: client) + adapter.send(_send_binding(), _send_target(), "**粗体** 列表", idempotency_key="d1") + assert len(client.calls) == 1 + body = client.calls[0]["body"] + assert body["msgtype"] == "markdown" + assert body["markdown"]["text"] == "**粗体** 列表" + assert body["markdown"]["title"] + + +def test_dingtalk_markdown_title_from_heading(): + client = _WebhookClient() + adapter = DingTalkAdapter(client_factory=lambda: client) + adapter.send(_send_binding(), _send_target(), "# 周报\n正文", idempotency_key="d2") + body = client.calls[0]["body"] + assert body["markdown"]["title"] == "周报" + + +def test_dingtalk_markdown_title_from_first_line_when_no_heading(): + client = _WebhookClient() + adapter = DingTalkAdapter(client_factory=lambda: client) + # 含 markdown 语法（粗体）但无标题，title 取首行截断 + adapter.send(_send_binding(), _send_target(), "**首行**\n第二行", idempotency_key="d3") + body = client.calls[0]["body"] + assert body["msgtype"] == "markdown" + assert body["markdown"]["title"] == "**首行**" + + +def test_dingtalk_plain_text_still_uses_text_msgtype(): + client = _WebhookClient() + adapter = DingTalkAdapter(client_factory=lambda: client) + adapter.send(_send_binding(), _send_target(), "hello world", idempotency_key="d4") + body = client.calls[0]["body"] + assert body["msgtype"] == "text" + assert body["text"]["content"] == "hello world" + + +def test_dingtalk_rich_render_disabled_falls_back_to_text(monkeypatch): + settings = get_settings().model_copy(update={"channel_rich_render_enabled": False}) + monkeypatch.setattr("app.channels.adapters.dingtalk.get_settings", lambda: settings) + client = _WebhookClient() + adapter = DingTalkAdapter(client_factory=lambda: client) + adapter.send(_send_binding(), _send_target(), "**粗体**", idempotency_key="d5") + body = client.calls[0]["body"] + assert body["msgtype"] == "text" + assert body["text"]["content"] == "**粗体**" + + +def test_dingtalk_markdown_long_text_chunked_with_titles(): + client = _WebhookClient() + adapter = DingTalkAdapter(client_factory=lambda: client) + long_md = "\n".join(f"# 标题{i}\n内容{i}" for i in range(800)) + adapter.send(_send_binding(), _send_target(), long_md, idempotency_key="d6") + assert len(client.calls) >= 2 + for call in client.calls: + assert call["body"]["msgtype"] == "markdown" + assert call["body"]["markdown"]["title"] + assert len(call["body"]["markdown"]["text"]) <= DINGTALK_TEXT_LIMIT + + +def test_dingtalk_markdown_rejects_untrusted_webhook(): + client = _WebhookClient() + adapter = DingTalkAdapter(client_factory=lambda: client) + with pytest.raises(DingTalkPermanentError): + adapter.send( + _send_binding(), + {"session_webhook": "https://attacker.example/steal"}, + "**x**", + idempotency_key="d7", + ) + assert client.calls == [] + + +def test_dingtalk_markdown_rejects_expired_webhook(): + client = _WebhookClient() + adapter = DingTalkAdapter(client_factory=lambda: client) + target = { + "session_webhook": "https://oapi.dingtalk.com/robot/send?session=x", + "session_webhook_expired_time": 1, + } + with pytest.raises(DingTalkPermanentError, match="过期"): + adapter.send(_send_binding(), target, "**x**", idempotency_key="d8") + assert client.calls == [] + + +def test_dingtalk_markdown_send_failure_is_permanent(): + client = _WebhookClient(_Response(400, {"errcode": 1})) + adapter = DingTalkAdapter(client_factory=lambda: client) + with pytest.raises(DingTalkPermanentError): + adapter.send(_send_binding(), _send_target(), "**x**", idempotency_key="d9") + + +def test_dingtalk_markdown_5xx_is_transient(): + client = _WebhookClient(_Response(503, {"errcode": 0})) + adapter = DingTalkAdapter(client_factory=lambda: client) + with pytest.raises(DingTalkTransientError): + adapter.send(_send_binding(), _send_target(), "**x**", idempotency_key="d10") + + +def test_dingtalk_overlong_fenced_code_block_chunks_are_balanced(): + """回归 P1：超长围栏代码块切分后，每个发送 chunk 必须围栏平衡（合法 Markdown）。""" + client = _WebhookClient() + adapter = DingTalkAdapter(client_factory=lambda: client) + code_lines = [f"line_{i} = {i}" for i in range(300)] + text = "```python\n" + "\n".join(code_lines) + "\n```" + adapter.send(_send_binding(), _send_target(), text, idempotency_key="d-long-code") + assert len(client.calls) >= 2 + for call in client.calls: + body = call["body"] + assert body["msgtype"] == "markdown" + md_text = body["markdown"]["text"] + fence_count = sum( + 1 for line in md_text.split("\n") if line.strip().startswith("```") + ) + assert fence_count % 2 == 0, "fenced code block split across messages is unbalanced" + assert len(md_text) <= DINGTALK_TEXT_LIMIT + + diff --git a/backend/tests/test_channel_intake.py b/backend/tests/test_channel_intake.py index 498437f4..590474a3 100644 --- a/backend/tests/test_channel_intake.py +++ b/backend/tests/test_channel_intake.py @@ -90,7 +90,7 @@ class RecordingAgentLoop: calls: list = [] error: Exception | None = None - def __init__(self, db): + def __init__(self, db, *, event_sink=None): self.db = db def handle_turn(self, request): diff --git a/backend/tests/test_channel_routing.py b/backend/tests/test_channel_routing.py index e1e34a63..0ce59656 100644 --- a/backend/tests/test_channel_routing.py +++ b/backend/tests/test_channel_routing.py @@ -100,7 +100,7 @@ def _group_message(event_id: str, text: str, group_id: str = "room_123456") -> d class RecordingAgentLoop: calls: list = [] - def __init__(self, db): + def __init__(self, db, *, event_sink=None): self.db = db def handle_turn(self, request): diff --git a/backend/tests/test_channel_scope.py b/backend/tests/test_channel_scope.py index a54ac535..4b9e120c 100644 --- a/backend/tests/test_channel_scope.py +++ b/backend/tests/test_channel_scope.py @@ -94,7 +94,7 @@ def _wecom_inbound(event_id: str, text: str, *, userid: str = "zhangsan", group: class RecordingAgentLoop: calls: list = [] - def __init__(self, db): + def __init__(self, db, *, event_sink=None): self.db = db def handle_turn(self, request): diff --git a/backend/tests/test_channel_wecom.py b/backend/tests/test_channel_wecom.py index cec2d19f..48fa54f5 100644 --- a/backend/tests/test_channel_wecom.py +++ b/backend/tests/test_channel_wecom.py @@ -778,7 +778,7 @@ def test_reconcile_does_not_stop_stream_while_revision_initializes(monkeypatch) class RecordingAgentLoop: calls: list = [] - def __init__(self, db): + def __init__(self, db, *, event_sink=None): self.db = db def handle_turn(self, request): @@ -1570,7 +1570,7 @@ def test_delete_rejects_while_durable_turn_is_running(monkeypatch) -> None: release_turn = threading.Event() class BlockingAgentLoop: - def __init__(self, db): + def __init__(self, db, *, event_sink=None): self.db = db def handle_turn(self, request): diff --git a/backend/tests/test_event_log_sink.py b/backend/tests/test_event_log_sink.py new file mode 100644 index 00000000..d37898bc --- /dev/null +++ b/backend/tests/test_event_log_sink.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +from sqlalchemy.pool import StaticPool +from sqlmodel import Session, SQLModel, create_engine + +from app.observability.event_log import EventLog + + +def _test_db() -> Session: + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + SQLModel.metadata.create_all(engine) + return Session(engine) + + +def test_record_invokes_event_sink_with_traced_payload() -> None: + received: list[tuple[str, dict]] = [] + + def sink(event_type: str, payload: dict) -> None: + received.append((event_type, dict(payload))) + + with _test_db() as db: + events = EventLog(db, event_sink=sink) + events.bind_turn("turn_1", "client_turn_1") + events.record( + "tenant_demo", + "session_test", + "step_result", + {"reply": "ok"}, + ) + + assert len(received) == 1 + event_type, payload = received[0] + assert event_type == "step_result" + assert payload["reply"] == "ok" + assert payload["turn_id"] == "turn_1" + assert payload["user_message_id"] == "turn_1" + assert payload["client_turn_id"] == "client_turn_1" + + +def test_event_sink_none_does_not_raise() -> None: + with _test_db() as db: + events = EventLog(db) + events.bind_turn("turn_1") + event = events.record( + "tenant_demo", + "session_test", + "step_result", + {"reply": "ok"}, + ) + assert event.event_type == "step_result" + assert event.payload_json["turn_id"] == "turn_1" + + +def test_event_sink_exception_does_not_propagate() -> None: + def broken_sink(event_type: str, payload: dict) -> None: + raise RuntimeError("sink exploded") + + with _test_db() as db: + events = EventLog(db, event_sink=broken_sink) + events.bind_turn("turn_1") + event = events.record( + "tenant_demo", + "session_test", + "step_result", + {"reply": "ok"}, + ) + assert event.event_type == "step_result" diff --git a/backend/tests/test_feishu_adapter.py b/backend/tests/test_feishu_adapter.py index 94331780..3cc8c851 100644 --- a/backend/tests/test_feishu_adapter.py +++ b/backend/tests/test_feishu_adapter.py @@ -1,9 +1,10 @@ from __future__ import annotations -from concurrent.futures import ThreadPoolExecutor -from datetime import timedelta +import hashlib import json import threading +from concurrent.futures import ThreadPoolExecutor +from datetime import timedelta from types import SimpleNamespace import httpx @@ -13,13 +14,13 @@ from app.channels.adapters.feishu import ( FeishuAdapter, FeishuPermanentError, - FeishuTransientError, FeishuTokenProvider, + FeishuTransientError, validate_feishu_credentials, ) from app.channels.crypto import encrypt_channel_secret -from app.channels.service_outbox import run_delivery_daemon from app.channels.feishu_runtime import _build_event_dispatcher, _normalize_event +from app.channels.service_outbox import run_delivery_daemon from app.db.models import ChannelBinding, ChannelDelivery, Tenant, utc_now @@ -39,6 +40,9 @@ def post(self, url, **kwargs): def get(self, url, **kwargs): return self.handler(url, {**kwargs, "_method": "GET"}) + def patch(self, url, **kwargs): + return self.handler(url, {**kwargs, "_method": "PATCH"}) + def delete(self, url, **kwargs): return self.handler(url, {**kwargs, "_method": "DELETE"}) @@ -940,3 +944,286 @@ def test_normalize_post_message_without_locale_wrapper() -> None: assert len(inbound.attachments) == 1 assert inbound.attachments[0].media_id == "img_v3_nolocale" assert inbound.text == "查询假期余额" + + +# --------------------------------------------------------------------------- +# 富文本（post）渲染 — channel-render-plan §5.2 +# --------------------------------------------------------------------------- + + +def _rich_handler(calls): + def handler(url, kwargs): + if "/auth/" in url: + return _response(200, {"code": 0, "tenant_access_token": "token", "expire": 7200}, url) + calls.append((url, kwargs)) + return _response(200, {"code": 0}, url) + return handler + + +def test_rich_post_render_for_markdown_text() -> None: + calls = [] + adapter = FeishuAdapter(client_factory=lambda: FakeClient(_rich_handler(calls))) + adapter.send( + _binding(), + {"message_id": "om_source"}, + "# 标题\n\n**粗体** 与 [链接](https://x.com)", + idempotency_key="rich-1", + ) + send = calls[0] + assert send[1]["json"]["msg_type"] == "post" + content = json.loads(send[1]["json"]["content"]) + rows = content["zh_cn"]["content"] + # 第一行标题，应含 bold style + assert any("bold" in (tag.get("style") or []) for tag in rows[0]) + # 存在链接 tag + flat_tags = [tag for row in rows for tag in row] + assert any(tag.get("tag") == "a" and tag.get("href") == "https://x.com" for tag in flat_tags) + + +def test_rich_post_code_block_tag() -> None: + calls = [] + adapter = FeishuAdapter(client_factory=lambda: FakeClient(_rich_handler(calls))) + adapter.send( + _binding(), + {"message_id": "om_source"}, + "```python\nprint(1)\n```", + idempotency_key="rich-code", + ) + content = json.loads(calls[0][1]["json"]["content"]) + tag = content["zh_cn"]["content"][0][0] + assert tag["tag"] == "code_block" + assert tag["language"] == "python" + assert tag["text"] == "print(1)" + + +def test_plain_text_still_uses_text_msg_type() -> None: + calls = [] + adapter = FeishuAdapter(client_factory=lambda: FakeClient(_rich_handler(calls))) + adapter.send(_binding(), {"message_id": "om_source"}, "hello", idempotency_key="t1") + assert calls[0][1]["json"]["msg_type"] == "text" + assert json.loads(calls[0][1]["json"]["content"]) == {"text": "hello"} + + +def test_rich_render_disabled_falls_back_to_text(monkeypatch) -> None: + from app.config import get_settings + settings = get_settings().model_copy(update={"channel_rich_render_enabled": False}) + monkeypatch.setattr("app.channels.adapters.feishu.get_settings", lambda: settings) + calls = [] + adapter = FeishuAdapter(client_factory=lambda: FakeClient(_rich_handler(calls))) + adapter.send( + _binding(), + {"message_id": "om_source"}, + "**粗体**", + idempotency_key="rich-off", + ) + assert calls[0][1]["json"]["msg_type"] == "text" + assert json.loads(calls[0][1]["json"]["content"]) == {"text": "**粗体**"} + + +def test_rich_long_markdown_chunks_use_distinct_stable_uuids() -> None: + calls = [] + + def handler(url, kwargs): + if "/auth/" in url: + return _response(200, {"code": 0, "tenant_access_token": "token", "expire": 7200}, url) + calls.append(kwargs["json"]) + return _response(200, {"code": 0}, url) + + adapter = FeishuAdapter(client_factory=lambda: FakeClient(handler)) + target = {"message_id": "om_source"} + # 构造超长 markdown：多行，每行含粗体，触发分块 + long_md = "\n".join(f"**第{i}行**" for i in range(500)) + adapter.send(_binding(), target, long_md, idempotency_key="long-rich") + first_uuids = [body["uuid"] for body in calls] + assert len(first_uuids) >= 2 + assert len(set(first_uuids)) == len(first_uuids) # 互不相同 + # 全部走 post + assert all(body["msg_type"] == "post" for body in calls) + # 幂等：重发 uuid 稳定 + calls.clear() + adapter.send(_binding(), target, long_md, idempotency_key="long-rich") + assert [body["uuid"] for body in calls] == first_uuids + + +def test_rich_render_path_refreshes_token_on_401() -> None: + tokens = iter(["token-old", "token-new"]) + send_count = 0 + + def handler(url, kwargs): + nonlocal send_count + if "/auth/" in url: + return _response( + 200, + {"code": 0, "tenant_access_token": next(tokens), "expire": 7200}, + url, + ) + send_count += 1 + if send_count == 1: + return _response(401, {"code": 99991663}, url) + assert kwargs["headers"]["Authorization"] == "Bearer token-new" + return _response(200, {"code": 0}, url) + + def factory(): + return FakeClient(handler) + adapter = FeishuAdapter(client_factory=factory) + adapter.send( + _binding(), + {"message_id": "om_source"}, + "**粗体**", + idempotency_key="rich-401", + ) + assert send_count == 2 + + +def test_rich_create_message_uses_receive_id() -> None: + calls = [] + adapter = FeishuAdapter(client_factory=lambda: FakeClient(_rich_handler(calls))) + adapter.send( + _binding(), + {"receive_id": "ou_user", "receive_id_type": "open_id"}, + "# 标题", + idempotency_key="rich-create", + ) + send = calls[0] + assert send[1]["params"] == {"receive_id_type": "open_id"} + assert send[1]["json"]["receive_id"] == "ou_user" + assert send[1]["json"]["msg_type"] == "post" + + +def test_rich_overlong_fenced_code_block_each_chunk_has_code_block() -> None: + """回归 P1：超长围栏代码块切分后，每个发送 chunk 应含闭合的 code_block tag。 + + 避免围栏跨消息断裂导致代码块被当普通文本渲染。 + """ + calls = [] + adapter = FeishuAdapter(client_factory=lambda: FakeClient(_rich_handler(calls))) + code_lines = [f"line_{i} = {i}" for i in range(300)] + text = "```python\n" + "\n".join(code_lines) + "\n```" + adapter.send( + _binding(), + {"message_id": "om_source"}, + text, + idempotency_key="rich-long-code", + ) + assert len(calls) >= 2 + all_code_texts: list[str] = [] + for _url, kwargs in calls: + assert kwargs["json"]["msg_type"] == "post" + content = json.loads(kwargs["json"]["content"]) + rows = content["zh_cn"]["content"] + flat_tags = [tag for row in rows for tag in row] + assert any(tag.get("tag") == "code_block" for tag in flat_tags), ( + "chunk missing code_block tag after split" + ) + for tag in flat_tags: + if tag.get("tag") == "code_block": + all_code_texts.append(tag["text"]) + recovered = "\n".join(all_code_texts) + assert "line_0 = 0" in recovered + assert "line_299 = 299" in recovered + + +def test_create_card_reply_returns_message_id() -> None: + calls = [] + + def handler(url, kwargs): + if "/auth/" in url: + return _response(200, {"code": 0, "tenant_access_token": "token", "expire": 7200}, url) + calls.append((url, kwargs)) + return _response( + 200, + {"code": 0, "data": {"message_id": "om_card_new"}}, + url, + ) + + adapter = FeishuAdapter(client_factory=lambda: FakeClient(handler)) + card = {"config": {}, "header": {"title": {"tag": "plain_text", "content": "test"}}, "elements": []} + message_id = adapter.create_card( + _binding(), + {"message_id": "om_source"}, + card, + idempotency_key="card-key-1", + ) + assert message_id == "om_card_new" + url, kwargs = calls[0] + assert url.endswith("/im/v1/messages/om_source/reply") + assert kwargs["json"]["msg_type"] == "interactive" + assert kwargs["json"]["uuid"] == hashlib.sha256(b"card-key-1:0").hexdigest()[:40] + assert json.loads(kwargs["json"]["content"]) == card + + +def test_create_card_with_receive_id() -> None: + calls = [] + + def handler(url, kwargs): + if "/auth/" in url: + return _response(200, {"code": 0, "tenant_access_token": "token", "expire": 7200}, url) + calls.append((url, kwargs)) + return _response(200, {"code": 0, "data": {"message_id": "om_card_new"}}, url) + + adapter = FeishuAdapter(client_factory=lambda: FakeClient(handler)) + message_id = adapter.create_card( + _binding(), + {"receive_id": "ou_user", "receive_id_type": "open_id"}, + {"elements": []}, + idempotency_key="card-key-2", + ) + assert message_id == "om_card_new" + url, kwargs = calls[0] + assert url.endswith("/im/v1/messages") + assert kwargs["params"] == {"receive_id_type": "open_id"} + assert kwargs["json"]["receive_id"] == "ou_user" + + +def test_create_card_missing_idempotency_key_raises() -> None: + adapter = FeishuAdapter(client_factory=lambda: FakeClient(lambda *_a, **_kw: None)) + with pytest.raises(FeishuPermanentError, match="幂等键"): + adapter.create_card(_binding(), {"message_id": "om"}, {}, idempotency_key="") + + +def test_create_card_missing_response_message_id_raises_transient() -> None: + def handler(url, _kwargs): + if "/auth/" in url: + return _response(200, {"code": 0, "tenant_access_token": "token", "expire": 7200}, url) + return _response(200, {"code": 0, "data": {}}, url) + + adapter = FeishuAdapter(client_factory=lambda: FakeClient(handler)) + with pytest.raises(FeishuTransientError, match="message_id"): + adapter.create_card(_binding(), {"message_id": "om"}, {}, idempotency_key="k") + + +def test_update_card_sends_patch_request() -> None: + calls = [] + + def handler(url, kwargs): + if "/auth/" in url: + return _response(200, {"code": 0, "tenant_access_token": "token", "expire": 7200}, url) + calls.append((url, kwargs)) + return _response(200, {"code": 0}, url) + + adapter = FeishuAdapter(client_factory=lambda: FakeClient(handler)) + card = {"elements": [{"tag": "div"}]} + adapter.update_card(_binding(), "om_card_123", card) + assert len(calls) == 1 + url, kwargs = calls[0] + assert url.endswith("/im/v1/messages/om_card_123") + assert kwargs["_method"] == "PATCH" + assert json.loads(kwargs["json"]["content"]) == card + + +def test_update_card_missing_message_id_raises() -> None: + adapter = FeishuAdapter(client_factory=lambda: FakeClient(lambda *_a, **_kw: None)) + with pytest.raises(FeishuPermanentError, match="message_id"): + adapter.update_card(_binding(), "", {}) + + +def test_update_card_429_is_transient() -> None: + def handler(url, _kwargs): + if "/auth/" in url: + return _response(200, {"code": 0, "tenant_access_token": "token", "expire": 7200}, url) + return _response(429, {"code": 1}, url) + + adapter = FeishuAdapter(client_factory=lambda: FakeClient(handler)) + with pytest.raises(FeishuTransientError, match="暂时不可用"): + adapter.update_card(_binding(), "om_card", {}) + diff --git a/backend/tests/test_feishu_durable_inbox.py b/backend/tests/test_feishu_durable_inbox.py index 0a530f82..76da5e39 100644 --- a/backend/tests/test_feishu_durable_inbox.py +++ b/backend/tests/test_feishu_durable_inbox.py @@ -213,7 +213,7 @@ def test_daemon_recovers_received_event_without_memory_notification( calls = [] class FakeAgentLoop: - def __init__(self, db): + def __init__(self, db, *, event_sink=None): self.db = db def handle_turn(self, request): @@ -290,7 +290,7 @@ def test_group_event_creates_group_session_and_preserves_sender_context( requests = [] class FakeAgentLoop: - def __init__(self, db): + def __init__(self, db, *, event_sink=None): self.db = db def handle_turn(self, request): @@ -441,7 +441,7 @@ def test_startup_sweep_decodes_stale_feishu_processing_envelope( calls = [] class FakeAgentLoop: - def __init__(self, db): + def __init__(self, db, *, event_sink=None): self.db = db def handle_turn(self, request): diff --git a/backend/tests/test_feishu_trace_streamer.py b/backend/tests/test_feishu_trace_streamer.py new file mode 100644 index 00000000..808833fa --- /dev/null +++ b/backend/tests/test_feishu_trace_streamer.py @@ -0,0 +1,224 @@ +from __future__ import annotations + +import time +from types import SimpleNamespace + +from app.channels.adapters.feishu import FeishuPermanentError +from app.channels.feishu_trace import FeishuTraceStreamer, _SinkEvent, is_feishu_trace_enabled + + +def _binding(channel: str = "feishu", config: dict | None = None) -> SimpleNamespace: + return SimpleNamespace( + id="chan_feishu", + tenant_id="tenant_a", + channel=channel, + config_json=config if config is not None else {}, + ) + + +class FakeAdapter: + def __init__(self, *, fail_create: bool = False, fail_update: bool = False) -> None: + self.create_calls: list[dict] = [] + self.update_calls: list[dict] = [] + self.fail_create = fail_create + self.fail_update = fail_update + + def create_card(self, binding, target, card_json, *, idempotency_key) -> str: + self.create_calls.append( + {"binding": binding, "target": target, "card": card_json, "key": idempotency_key} + ) + if self.fail_create: + raise FeishuPermanentError("create failed") + return "om_card_123" + + def update_card(self, binding, message_id, card_json) -> None: + self.update_calls.append( + {"binding": binding, "message_id": message_id, "card": card_json} + ) + if self.fail_update: + raise FeishuPermanentError("update failed") + + +def _make_streamer( + *, + adapter: FakeAdapter | None = None, + min_update_interval: float = 0.0, +) -> FeishuTraceStreamer: + return FeishuTraceStreamer( + _binding(), + {"message_id": "om_source"}, + "turn_1", + adapter=adapter or FakeAdapter(), + min_update_interval=min_update_interval, + ) + + +def _wait_for_card(streamer: FeishuTraceStreamer, timeout: float = 1.0) -> None: + """等待后台 worker 完成卡片创建（无论成功或失败）。""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if streamer._card_created: + return + time.sleep(0.005) + + +def _wait_for_updates(adapter: FakeAdapter, count: int, timeout: float = 2.0) -> None: + """等待 adapter 收到指定数量的 update 调用。""" + deadline = time.monotonic() + timeout + while len(adapter.update_calls) < count and time.monotonic() < deadline: + time.sleep(0.005) + + +def test_start_creates_card_and_saves_message_id() -> None: + adapter = FakeAdapter() + streamer = _make_streamer(adapter=adapter) + streamer.start() + _wait_for_card(streamer) + streamer.finish() + assert streamer._message_id == "om_card_123" + assert len(adapter.create_calls) == 1 + call = adapter.create_calls[0] + assert call["key"] == "feishu-trace:chan_feishu:turn_1" + assert call["target"] == {"message_id": "om_source"} + header = call["card"]["header"] + assert "正在" in header["title"]["content"] + + +def test_start_failure_does_not_raise_and_disables_updates() -> None: + adapter = FakeAdapter(fail_create=True) + streamer = _make_streamer(adapter=adapter) + streamer.start() + _wait_for_card(streamer) + streamer.on_event("step_result", {"turn_id": "turn_1", "reply": "ok"}) + streamer.finish() + assert len(adapter.update_calls) == 0 + + +def test_on_event_renders_line_and_patches_card() -> None: + adapter = FakeAdapter() + streamer = _make_streamer(adapter=adapter) + streamer.start() + _wait_for_card(streamer) + + streamer.on_event( + "router_decision_created", + {"turn_id": "turn_1", "user_intent": "退款", "reason": "匹配退款SOP"}, + ) + streamer.finish() + + assert len(adapter.update_calls) >= 1 + last_card = adapter.update_calls[-1]["card"] + elements = last_card["elements"] + texts = [el["text"]["content"] for el in elements] + assert any("判断意图" in t for t in texts) + assert last_card["header"]["template"] == "green" + + +def test_throttle_merges_rapid_events() -> None: + adapter = FakeAdapter() + streamer = _make_streamer(adapter=adapter, min_update_interval=10.0) + streamer.start() + _wait_for_card(streamer) + + streamer.on_event("router_decision_created", {"turn_id": "t1"}) + streamer.on_event("step_result", {"turn_id": "t1", "next_step_id": "s2"}) + streamer.on_event("tool_call_started", {"turn_id": "t1", "name": "lookup"}) + + throttled_updates = len(adapter.update_calls) + assert throttled_updates <= 1 + + streamer.finish() + final_updates = len(adapter.update_calls) + assert final_updates > throttled_updates + last_card = adapter.update_calls[-1]["card"] + elements = last_card["elements"] + assert len(elements) >= 2 + + +def test_update_failure_does_not_raise() -> None: + adapter = FakeAdapter(fail_update=True) + streamer = _make_streamer(adapter=adapter) + streamer.start() + _wait_for_card(streamer) + streamer.on_event("router_decision_created", {"turn_id": "t1"}) + streamer.finish() + + +def test_finish_marks_running_lines_completed() -> None: + adapter = FakeAdapter() + streamer = _make_streamer(adapter=adapter) + streamer.start() + _wait_for_card(streamer) + streamer.on_event("tool_call_started", {"turn_id": "t1", "name": "lookup"}) + streamer.finish() + + last_card = adapter.update_calls[-1]["card"] + assert last_card["header"]["template"] == "green" + + +def test_abort_marks_failed_state() -> None: + adapter = FakeAdapter() + streamer = _make_streamer(adapter=adapter) + streamer.start() + _wait_for_card(streamer) + streamer.on_event("tool_call_started", {"turn_id": "t1", "name": "lookup"}) + streamer.abort("boom") + + last_card = adapter.update_calls[-1]["card"] + assert last_card["header"]["template"] == "red" + + +def test_on_event_after_finish_is_ignored() -> None: + adapter = FakeAdapter() + streamer = _make_streamer(adapter=adapter) + streamer.start() + _wait_for_card(streamer) + streamer.finish() + updates_before = len(adapter.update_calls) + streamer.on_event("step_result", {"turn_id": "t1"}) + assert len(adapter.update_calls) == updates_before + + +def test_sink_event_construction() -> None: + event = _SinkEvent("step_result", {"turn_id": "t1", "reply": "ok"}) + assert event.event_type == "step_result" + assert event.payload_json["turn_id"] == "t1" + assert event.id == "t1" + + +def test_is_feishu_trace_enabled() -> None: + assert is_feishu_trace_enabled(_binding(channel="feishu")) is True + assert is_feishu_trace_enabled(_binding(channel="wechat")) is False + assert is_feishu_trace_enabled(_binding(channel="feishu", config={"trace_enabled": False})) is False + assert is_feishu_trace_enabled(None) is False + + +def test_on_event_does_not_block_when_card_not_created() -> None: + """卡片尚未创建时 on_event 应立即返回，不阻塞 AgentLoop。""" + adapter = FakeAdapter() + streamer = _make_streamer(adapter=adapter, min_update_interval=0.0) + streamer.start() + # 不等待卡片创建——立即发送事件 + streamer.on_event("router_decision_created", {"turn_id": "t1"}) + streamer.on_event("step_result", {"turn_id": "t1"}) + # finish 会 join worker，确保所有排队任务完成 + streamer.finish() + assert len(adapter.create_calls) == 1 + # 卡片创建后应有一次最终更新 + assert len(adapter.update_calls) >= 1 + last_card = adapter.update_calls[-1]["card"] + assert last_card["header"]["template"] == "green" + + +def test_finish_before_card_created_still_sends_final_state() -> None: + """finish 在卡片创建前调用：worker 先建卡再发最终状态。""" + adapter = FakeAdapter() + streamer = _make_streamer(adapter=adapter) + streamer.start() + streamer.on_event("router_decision_created", {"turn_id": "t1"}) + # 立即 finish，不等卡片创建 + streamer.finish() + assert len(adapter.create_calls) == 1 + assert len(adapter.update_calls) >= 1 + last_card = adapter.update_calls[-1]["card"] + assert last_card["header"]["template"] == "green" diff --git a/backend/tests/test_markdown_render.py b/backend/tests/test_markdown_render.py new file mode 100644 index 00000000..9c43288c --- /dev/null +++ b/backend/tests/test_markdown_render.py @@ -0,0 +1,595 @@ +from __future__ import annotations + +import signal + +from app.channels.markdown_render import ( + CodeBlock, + Heading, + ListItem, + Paragraph, + Quote, + TableBlock, + ThematicBreak, + ensure_code_fences, + extract_dingtalk_title, + has_markdown, + parse_markdown, + render_feishu_post, + split_markdown_by_lines, +) + +# --------------------------------------------------------------------------- +# has_markdown +# --------------------------------------------------------------------------- + + +def test_has_markdown_detects_common_syntax(): + assert has_markdown("# 标题") + assert has_markdown("**bold**") + assert has_markdown("`code`") + assert has_markdown("[link](http://x)") + assert has_markdown("- 列表项") + assert has_markdown("1. 有序项") + assert has_markdown("> 引用") + assert has_markdown("```\ncode\n```") + assert has_markdown("---") + + +def test_has_markdown_false_for_plain_text(): + assert not has_markdown("hello world") + assert not has_markdown("a * b = c") + assert not has_markdown("file_name_with_underscores") + assert not has_markdown("普通中文回复") + assert not has_markdown("") + # 单星号装饰不构成斜体语法（已被 _ITALIC_RE 排除检测），但 * 列表项会命中 + assert not has_markdown("价格 * 3 = 9") + # 破折号不是分隔线（少于3个） + assert not has_markdown("a - b") + + +# --------------------------------------------------------------------------- +# parse_markdown +# --------------------------------------------------------------------------- + + +def test_parse_heading(): + blocks = parse_markdown("## 标题二") + assert len(blocks) == 1 + assert isinstance(blocks[0], Heading) + assert blocks[0].level == 2 + assert blocks[0].spans[0].text == "标题二" + + +def test_parse_bold_and_italic(): + blocks = parse_markdown("**粗** 和 *斜*") + para = blocks[0] + assert isinstance(para, Paragraph) + texts = [(s.text, set(s.styles)) for s in para.spans] + assert ("粗", {"bold"}) in texts + assert ("斜", {"italic"}) in texts + + +def test_parse_inline_code(): + blocks = parse_markdown("用 `printf` 输出") + spans = blocks[0].spans + code_span = next(s for s in spans if "code" in s.styles) + assert code_span.text == "printf" + + +def test_parse_multiple_inline_code_no_infinite_loop(): + """Regression: multiple inline code segments caused an infinite loop because + the code-placeholder regex searched text[pos:] but used the relative match.end() + as the absolute pos, never advancing past the second placeholder.""" + def _handler(signum, frame): + raise TimeoutError("parse_markdown did not complete in time") + + signal.signal(signal.SIGALRM, _handler) + signal.alarm(5) + try: + blocks = parse_markdown("a `b` c `d` e") + finally: + signal.alarm(0) + spans = blocks[0].spans + code_spans = [s for s in spans if "code" in s.styles] + assert len(code_spans) == 2 + assert code_spans[0].text == "b" + assert code_spans[1].text == "d" + + +def test_parse_code_in_bold_not_reparsed(): + blocks = parse_markdown("**`x`**") + spans = blocks[0].spans + # 粗体包裹，内部代码不应被二次拆为 code span，而是粗体文本 + assert any(set(s.styles) == {"bold"} and s.text == "`x`" or s.text == "x" + for s in spans) + + +def test_parse_fenced_code_block_with_language(): + blocks = parse_markdown("```python\nprint(1)\nprint(2)\n```") + assert len(blocks) == 1 + assert isinstance(blocks[0], CodeBlock) + assert blocks[0].language == "python" + assert blocks[0].text == "print(1)\nprint(2)" + + +def test_parse_fenced_code_block_no_language(): + blocks = parse_markdown("```\nraw\n```") + assert isinstance(blocks[0], CodeBlock) + assert blocks[0].language == "" + assert blocks[0].text == "raw" + + +def test_parse_unclosed_fence_does_not_raise(): + blocks = parse_markdown("```\nunclosed code") + assert isinstance(blocks[0], CodeBlock) + assert "unclosed code" in blocks[0].text + + +def test_parse_indented_code_block(): + blocks = parse_markdown(" print(1)\n print(2)") + assert isinstance(blocks[0], CodeBlock) + assert blocks[0].text == "print(1)\nprint(2)" + + +def test_parse_indented_code_block_with_blank_lines(): + text = " def f():\n return 1\n\n print(f())" + blocks = parse_markdown(text) + assert len(blocks) == 1 + assert isinstance(blocks[0], CodeBlock) + assert "return 1" in blocks[0].text + assert "print(f())" in blocks[0].text + + +def test_parse_indented_code_block_double_blank_merges(): + text = " line1\n\n\n line2" + blocks = parse_markdown(text) + assert len(blocks) == 1 + assert isinstance(blocks[0], CodeBlock) + assert blocks[0].text == "line1\n\n\nline2" + + +def test_parse_indented_code_block_followed_by_paragraph(): + text = " code line\n\nparagraph text" + blocks = parse_markdown(text) + assert isinstance(blocks[0], CodeBlock) + assert blocks[0].text == "code line" + assert isinstance(blocks[1], Paragraph) + assert blocks[1].spans[0].text == "paragraph text" + + +def test_parse_indented_code_block_after_paragraph(): + text = "intro text\n\n code here" + blocks = parse_markdown(text) + assert isinstance(blocks[0], Paragraph) + assert blocks[0].spans[0].text == "intro text" + assert isinstance(blocks[1], CodeBlock) + assert blocks[1].text == "code here" + + +def test_parse_toplevel_code_block_def(): + text = "def f():\n return 1" + blocks = parse_markdown(text) + assert isinstance(blocks[0], CodeBlock) + assert "def f():" in blocks[0].text + assert "return 1" in blocks[0].text + + +def test_parse_toplevel_code_block_with_following_paragraph(): + text = "def f():\n return 1\n\n这是说明文字。" + blocks = parse_markdown(text) + assert isinstance(blocks[0], CodeBlock) + assert "def f():" in blocks[0].text + assert isinstance(blocks[1], Paragraph) + assert blocks[1].spans[0].text == "这是说明文字。" + + +def test_parse_toplevel_code_block_includes_assignment_and_print(): + text = ( + "def f():\n" + " return 1\n" + "\n" + "result = f()\n" + "print(result)\n" + "# output: 1\n" + "\n" + "说明文字。" + ) + blocks = parse_markdown(text) + assert isinstance(blocks[0], CodeBlock) + assert "def f():" in blocks[0].text + assert "result = f()" in blocks[0].text + assert "print(result)" in blocks[0].text + assert "# output: 1" in blocks[0].text + assert isinstance(blocks[1], Paragraph) + assert blocks[1].spans[0].text == "说明文字。" + + +def test_parse_toplevel_code_block_hash_comment_not_heading(): + text = "def f():\n return 1\n\n# [1, 2, 3]\n\n说明。" + blocks = parse_markdown(text) + assert isinstance(blocks[0], CodeBlock) + assert "# [1, 2, 3]" in blocks[0].text + assert not any(isinstance(b, Heading) for b in blocks) + + +def test_parse_link(): + blocks = parse_markdown("[StaffDeck](https://staffdeck.ai)") + spans = blocks[0].spans + link = next(s for s in spans if s.href) + assert link.text == "StaffDeck" + assert link.href == "https://staffdeck.ai" + + +def test_parse_unordered_list(): + blocks = parse_markdown("- 项一\n- 项二") + assert all(isinstance(b, ListItem) for b in blocks) + assert blocks[0].ordered is False + assert blocks[0].spans[0].text == "项一" + assert blocks[1].spans[0].text == "项二" + + +def test_parse_ordered_list_keeps_index(): + blocks = parse_markdown("1. 第一\n2. 第二") + assert blocks[0].ordered is True + assert blocks[0].index == 1 + assert blocks[1].index == 2 + + +def test_parse_quote(): + blocks = parse_markdown("> 引用一\n> 引用二") + assert len(blocks) == 1 + assert isinstance(blocks[0], Quote) + assert "引用一" in blocks[0].spans[0].text + assert "引用二" in blocks[0].spans[0].text + + +def test_parse_thematic_break(): + blocks = parse_markdown("---") + assert isinstance(blocks[0], ThematicBreak) + + +def test_parse_table_degrades_to_text(): + md = "| a | b |\n|---|---|\n| 1 | 2 |" + blocks = parse_markdown(md) + assert len(blocks) == 1 + assert isinstance(blocks[0], TableBlock) + assert len(blocks[0].lines) == 3 + + +def test_parse_html_tags_treated_as_text(): + blocks = parse_markdown("") + para = blocks[0] + assert isinstance(para, Paragraph) + assert "") + post = render_feishu_post(blocks) + tag = post["zh_cn"]["content"][0][0] + assert tag["tag"] == "text" + assert "