diff --git a/README.md b/README.md index 89b076757..73ab4430a 100644 --- a/README.md +++ b/README.md @@ -453,6 +453,7 @@ Notes: - `assistant_id: lead_agent` calls the default LangGraph assistant directly. - If `assistant_id` is set to a custom agent name, DeerFlow still routes through `lead_agent` and injects that value as `agent_name`, so the custom agent's SOUL/config takes effect for IM channels. - IM channel workers call Gateway's LangGraph-compatible API internally and automatically attach process-local internal auth plus the CSRF cookie/header pair required for thread and run creation. +- Feishu/Lark now queues rapid follow-up messages per mapped DeerFlow `thread_id` instead of immediately surfacing the generic busy reply, and topic replies keep a per-message card with a compact source-message preview across queued/running/final patches. Set the corresponding API keys in your `.env` file: diff --git a/backend/AGENTS.md b/backend/AGENTS.md index a9142d43f..0b12af195 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -485,7 +485,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk **Components**: - `message_bus.py` - Async pub/sub hub (`InboundMessage` → queue → dispatcher; `OutboundMessage` → callbacks → channels) - `store.py` - JSON-file persistence mapping `channel_name:chat_id[:topic_id]` → `thread_id` (keys are `channel:chat` for root conversations and `channel:chat:topic` for threaded conversations) -- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout` +- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout` - `base.py` - Abstract `Channel` base class (start/stop/send lifecycle) - `service.py` - Manages lifecycle of all configured channels from `config.yaml` - `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` registers the "Working on it..." placeholder as the stream target and edits it in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured) @@ -502,7 +502,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk 5. Feishu/Telegram chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`) 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound 6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound. The agent posts its own reply mid-run via `gh` from the sandbox. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub). -7. Feishu channel sends one running reply card up front, then patches the same card for each outbound update (card JSON sets `config.update_multi=true` for Feishu's patch API requirement) +7. Feishu channel sends one running reply card up front, then patches the same card for each outbound update (card JSON sets `config.update_multi=true` for Feishu's patch API requirement). Messages already sent inside an existing Feishu topic carry a compact source-message preview in that card, and queued same-thread follow-ups patch their own source message's card from queued → running → final without falling back to the generic busy reply. 8. Telegram streaming: the "Working on it..." placeholder message is registered as the stream target; non-final updates `editMessageText` it in place (channel-side throttle: 1s in private chats, 3s in groups due to Telegram's 20 msg/min group cap; 4096-char truncation; rate-limited updates dropped); the final update performs the last edit and splits >4096 texts into follow-up messages 9. DingTalk AI Card mode (when `card_template_id` configured): `runs.stream()` → create card with initial text → stream updates via `PUT /v1.0/card/streaming` → finalize on `is_final=True`. Falls back to `sampleMarkdown` if card creation or streaming fails 10. For commands (`/new`, `/status`, `/models`, `/memory`, `/goal`, `/help`): handle locally or query Gateway API diff --git a/backend/README.md b/backend/README.md index e02559f83..95b89235d 100644 --- a/backend/README.md +++ b/backend/README.md @@ -128,9 +128,9 @@ FastAPI application providing REST endpoints for frontend integration: ### IM Channels -The IM bridge supports Feishu, Slack, and Telegram. Slack and Telegram still use the final `runs.wait()` response path, while Feishu now streams through `runs.stream(["messages-tuple", "values"])` and updates a single in-thread card in place. +The IM bridge supports Feishu, Slack, and Telegram. Slack and Telegram still use the final `runs.wait()` response path, while Feishu now streams through `runs.stream(["messages-tuple", "values"])`, serializes rapid same-thread turns inside the channel manager, and updates a single in-thread card per source message in place. -For Feishu card updates, DeerFlow stores the running card's `message_id` per inbound message and patches that same card until the run finishes, preserving the existing `OK` / `DONE` reaction flow. +For Feishu card updates, DeerFlow stores the running card's `message_id` per inbound message and patches that same card until the run finishes, preserving the existing `OK` / `DONE` reaction flow. When a follow-up arrives inside an existing Feishu topic while another turn is still running, the later message now waits on the mapped DeerFlow `thread_id`, receives a queued/running card on that exact source message, and keeps a compact source-message blockquote in subsequent patches so rapid consecutive questions remain distinguishable. --- diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py index 1eeed7e7f..3c00aa9dc 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -29,6 +29,7 @@ from deerflow.sandbox.sandbox_provider import get_sandbox_provider logger = logging.getLogger(__name__) PENDING_CLARIFICATION_TTL_SECONDS = 30 * 60 FEISHU_INBOUND_BATCH_WINDOW_SECONDS = 0.75 +SOURCE_PREVIEW_METADATA_KEY = "feishu_source_preview" def _is_feishu_command(text: str) -> bool: @@ -85,6 +86,45 @@ class FeishuChannel(Channel): def _pending_key(chat_id: str, user_id: str) -> tuple[str, str]: return (chat_id, user_id) + @staticmethod + def _should_include_source_preview( + *, + chat_type: str | None, + root_id: str | None, + parent_id: str | None, + thread_id: str | None, + ) -> bool: + if chat_type == "p2p": + return False + return bool(root_id or parent_id or thread_id) + + @staticmethod + def _compact_source_preview(text: str) -> str | None: + stripped = text.strip() + if not stripped: + return None + + lines = [line.strip() for line in stripped.splitlines() if line.strip()] + if not lines: + return None + preview = "\n".join(lines[:3]) + if len(preview) > 240: + preview = preview[:237].rstrip() + "..." + return preview + + @classmethod + def _compose_card_text(cls, text: str, metadata: dict[str, Any] | None = None) -> str: + preview = None + if isinstance(metadata, dict): + raw_preview = metadata.get(SOURCE_PREVIEW_METADATA_KEY) + if isinstance(raw_preview, str) and raw_preview.strip(): + preview = raw_preview.strip() + if not preview: + return text + + quoted_preview = "\n".join(f"> {line}" for line in preview.splitlines()) + return f"{quoted_preview}\n\n{text}" + @property def supports_streaming(self) -> bool: return True @@ -494,9 +534,15 @@ class FeishuChannel(Channel): self._background_tasks.discard(task) self._log_task_error(task, name, msg_id) - async def _create_running_card(self, source_message_id: str, text: str) -> str | None: + async def _create_running_card( + self, + source_message_id: str, + text: str, + *, + metadata: dict[str, Any] | None = None, + ) -> str | None: """Create the running card and cache its message ID when available.""" - running_card_id = await self._reply_card(source_message_id, text) + running_card_id = await self._reply_card(source_message_id, self._compose_card_text(text, metadata)) if running_card_id: self._running_card_ids[source_message_id] = running_card_id logger.info("[Feishu] running card created: source=%s card=%s", source_message_id, running_card_id) @@ -504,7 +550,13 @@ class FeishuChannel(Channel): logger.warning("[Feishu] running card creation returned no message_id for source=%s, subsequent updates will fall back to new replies", source_message_id) return running_card_id - def _ensure_running_card_started(self, source_message_id: str, text: str = "thinking...") -> asyncio.Task | None: + def _ensure_running_card_started( + self, + source_message_id: str, + text: str = "thinking...", + *, + metadata: dict[str, Any] | None = None, + ) -> asyncio.Task | None: """Start running-card creation once per source message.""" running_card_id = self._running_card_ids.get(source_message_id) if running_card_id: @@ -514,7 +566,7 @@ class FeishuChannel(Channel): if running_card_task: return running_card_task - running_card_task = asyncio.create_task(self._create_running_card(source_message_id, text)) + running_card_task = asyncio.create_task(self._create_running_card(source_message_id, text, metadata=metadata)) self._running_card_tasks[source_message_id] = running_card_task running_card_task.add_done_callback(lambda done_task, mid=source_message_id: self._finalize_running_card_task(mid, done_task)) return running_card_task @@ -524,21 +576,31 @@ class FeishuChannel(Channel): self._running_card_tasks.pop(source_message_id, None) self._log_task_error(task, "create_running_card", source_message_id) - async def _ensure_running_card(self, source_message_id: str, text: str = "thinking...") -> str | None: + async def _ensure_running_card( + self, + source_message_id: str, + text: str = "thinking...", + *, + metadata: dict[str, Any] | None = None, + ) -> str | None: """Ensure the in-thread running card exists and track its message ID.""" running_card_id = self._running_card_ids.get(source_message_id) if running_card_id: return running_card_id - running_card_task = self._ensure_running_card_started(source_message_id, text) + running_card_task = self._ensure_running_card_started( + source_message_id, + text, + metadata=metadata, + ) if running_card_task is None: return self._running_card_ids.get(source_message_id) return await running_card_task - async def _send_running_reply(self, message_id: str) -> None: + async def _send_running_reply(self, message_id: str, *, metadata: dict[str, Any] | None = None) -> None: """Reply to a message in-thread with a running card.""" try: - await self._ensure_running_card(message_id) + await self._ensure_running_card(message_id, metadata=metadata) except Exception: logger.exception("[Feishu] failed to send running reply for message %s", message_id) @@ -556,8 +618,9 @@ class FeishuChannel(Channel): running_card_id = await running_card_task if running_card_id: + card_text = self._compose_card_text(msg.text, msg.metadata) try: - await self._update_card(running_card_id, msg.text) + await self._update_card(running_card_id, card_text) except Exception: if not msg.is_final: raise @@ -565,7 +628,7 @@ class FeishuChannel(Channel): "[Feishu] failed to patch running card %s, falling back to final reply", running_card_id, ) - fallback_card_id = await self._reply_card(source_message_id, msg.text) + fallback_card_id = await self._reply_card(source_message_id, card_text) self._remember_thread_mapping(msg, source_message_id, fallback_card_id) self._remember_pending_clarification(msg, fallback_card_id) else: @@ -573,7 +636,10 @@ class FeishuChannel(Channel): self._remember_pending_clarification(msg, running_card_id) logger.info("[Feishu] running card updated: source=%s card=%s", source_message_id, running_card_id) elif msg.is_final: - final_card_id = await self._reply_card(source_message_id, msg.text) + final_card_id = await self._reply_card( + source_message_id, + self._compose_card_text(msg.text, msg.metadata), + ) self._remember_thread_mapping(msg, source_message_id, final_card_id) self._remember_pending_clarification(msg, final_card_id) elif awaited_running_card_task: @@ -582,7 +648,11 @@ class FeishuChannel(Channel): source_message_id, ) else: - created_card_id = await self._ensure_running_card(source_message_id, msg.text) + created_card_id = await self._ensure_running_card( + source_message_id, + msg.text, + metadata=msg.metadata, + ) self._remember_thread_mapping(msg, source_message_id, created_card_id) if msg.is_final: @@ -843,7 +913,7 @@ class FeishuChannel(Channel): for reaction_message_id in reaction_message_ids: reaction_task = asyncio.create_task(self._add_reaction(reaction_message_id, "OK")) self._track_background_task(reaction_task, name="add_reaction", msg_id=reaction_message_id) - self._ensure_running_card_started(msg_id) + self._ensure_running_card_started(msg_id, metadata=inbound.metadata) await self.bus.publish_inbound(inbound) async def _attach_connection_identity(self, inbound: InboundMessage) -> InboundMessage: @@ -1014,6 +1084,27 @@ class FeishuChannel(Channel): self._ensure_pending_thread_mapping(chat_id, sender_id, pending) resolved_from_pending = True + source_preview = None + if self._should_include_source_preview( + chat_type=chat_type, + root_id=root_id, + parent_id=parent_id, + thread_id=feishu_thread_id, + ): + source_preview = self._compact_source_preview(text) + + metadata = { + "message_id": msg_id, + "root_id": root_id, + "parent_id": parent_id, + "thread_id": feishu_thread_id, + "topic_id": topic_id, + "user_id": sender_id, + RESOLVED_FROM_PENDING_CLARIFICATION_METADATA_KEY: resolved_from_pending, + } + if source_preview: + metadata[SOURCE_PREVIEW_METADATA_KEY] = source_preview + inbound = self._make_inbound( chat_id=chat_id, user_id=sender_id, @@ -1021,15 +1112,7 @@ class FeishuChannel(Channel): msg_type=msg_type, thread_ts=msg_id, files=files_list, - metadata={ - "message_id": msg_id, - "root_id": root_id, - "parent_id": parent_id, - "thread_id": feishu_thread_id, - "topic_id": topic_id, - "user_id": sender_id, - RESOLVED_FROM_PENDING_CLARIFICATION_METADATA_KEY: resolved_from_pending, - }, + metadata=metadata, ) inbound.topic_id = topic_id diff --git a/backend/app/channels/feishu_run_policy.py b/backend/app/channels/feishu_run_policy.py new file mode 100644 index 000000000..14add3a60 --- /dev/null +++ b/backend/app/channels/feishu_run_policy.py @@ -0,0 +1,15 @@ +"""Per-run policy registration for the Feishu channel.""" + +from __future__ import annotations + +from app.channels.run_policy import CHANNEL_RUN_POLICY, ChannelRunPolicy + + +def register_policy() -> None: + """Register Feishu's queue-same-thread behavior in the shared policy map.""" + CHANNEL_RUN_POLICY["feishu"] = ChannelRunPolicy( + serialize_thread_runs=True, + ) + + +register_policy() diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index 9153bb776..ba8f47fae 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -17,6 +17,7 @@ from urllib.parse import quote import httpx from langgraph_sdk.errors import ConflictError +from app.channels import feishu_run_policy as _feishu_run_policy # noqa: F401 from app.channels.commands import KNOWN_CHANNEL_COMMANDS from app.channels.message_bus import ( PENDING_CLARIFICATION_METADATA_KEY, @@ -29,6 +30,10 @@ from app.channels.message_bus import ( from app.channels.run_policy import CHANNEL_RUN_POLICY, ChannelRunPolicy from app.channels.store import ChannelStore from app.gateway.csrf_middleware import CSRF_COOKIE_NAME, CSRF_HEADER_NAME, generate_csrf_token + +# Import built-in channel run-policy registrars eagerly so direct +# ChannelManager construction sees the same policy map as gateway bootstrap. +from app.gateway.github import run_policy as _github_run_policy # noqa: F401 from app.gateway.internal_auth import create_internal_auth_headers from deerflow.config.agents_config import load_agent_config from deerflow.config.paths import make_safe_user_id @@ -177,6 +182,14 @@ class _BoundIdentityRejection: outbound_owner_user_id: str | None = None +@dataclass(slots=True) +class _SerializedThreadRunState: + """Per-thread lock state for channels that queue same-thread turns.""" + + lock: asyncio.Lock + waiters: int = 0 + + def _is_thread_busy_error(exc: BaseException | None) -> bool: if exc is None: return False @@ -814,6 +827,9 @@ class ChannelManager: # Per-conversation locks so concurrent inbound messages for the same # chat don't race to create duplicate threads (see _get_or_create_thread). self._thread_create_locks: dict[tuple[str, str, str | None], asyncio.Lock] = {} + # Per-thread run locks for channels that want in-manager serialization + # instead of surfacing the runtime's generic busy reply. + self._serialized_thread_runs: dict[tuple[str, str], _SerializedThreadRunState] = {} self._skill_storage: SkillStorage | None = None self._csrf_token = generate_csrf_token() self._semaphore: asyncio.Semaphore | None = None @@ -841,6 +857,57 @@ class ChannelManager: user_layer = _as_dict(users_layer.get(msg.user_id)) return channel_layer, user_layer + def _begin_serialized_thread_run( + self, + *, + channel_name: str, + thread_id: str, + ) -> tuple[_SerializedThreadRunState | None, bool]: + policy = CHANNEL_RUN_POLICY.get(channel_name) + if policy is None or not policy.serialize_thread_runs: + return None, False + + key = (channel_name, thread_id) + state = self._serialized_thread_runs.get(key) + if state is None: + state = _SerializedThreadRunState(lock=asyncio.Lock()) + self._serialized_thread_runs[key] = state + queued = state.lock.locked() + state.waiters += 1 + return state, queued + + def _finish_serialized_thread_run( + self, + *, + channel_name: str, + thread_id: str, + state: _SerializedThreadRunState | None, + lock_acquired: bool, + ) -> None: + if state is None: + return + + if lock_acquired: + state.lock.release() + state.waiters -= 1 + if state.waiters == 0 and not state.lock.locked(): + self._serialized_thread_runs.pop((channel_name, thread_id), None) + + async def _publish_progress_update(self, msg: InboundMessage, thread_id: str, text: str) -> None: + await self.bus.publish_outbound( + OutboundMessage( + channel_name=msg.channel_name, + chat_id=msg.chat_id, + thread_id=thread_id, + text=text, + is_final=False, + thread_ts=msg.thread_ts, + connection_id=msg.connection_id, + owner_user_id=msg.owner_user_id, + metadata=_response_metadata(msg.metadata), + ) + ) + def _resolve_run_params(self, msg: InboundMessage, thread_id: str) -> tuple[str, dict[str, Any], dict[str, Any]]: channel_layer, user_layer = self._resolve_session_layer(msg) @@ -1439,6 +1506,50 @@ class ChannelManager: logger.info("[Manager] reusing thread: thread_id=%s for topic_id=%s", thread_id, msg.topic_id) await self._update_thread_channel_metadata(client, msg, thread_id) + serial_state, queued = self._begin_serialized_thread_run( + channel_name=msg.channel_name, + thread_id=thread_id, + ) + serial_lock_acquired = False + try: + if queued: + await self._publish_progress_update( + msg, + thread_id, + "Queued behind another request in this conversation. I’ll start working on this as soon as it finishes.", + ) + if serial_state is not None: + await serial_state.lock.acquire() + serial_lock_acquired = True + if queued: + await self._publish_progress_update(msg, thread_id, "thinking...") + await self._handle_chat_on_thread( + client, + msg, + thread_id, + extra_context=extra_context, + storage_user_id=storage_user_id, + ) + finally: + self._finish_serialized_thread_run( + channel_name=msg.channel_name, + thread_id=thread_id, + state=serial_state, + lock_acquired=serial_lock_acquired, + ) + + async def _handle_chat_on_thread( + self, + client, + msg: InboundMessage, + thread_id: str, + *, + extra_context: dict[str, Any] | None = None, + storage_user_id: str | None = None, + ) -> None: + if storage_user_id is None: + storage_user_id = _channel_storage_user_id(msg) + assistant_id, run_config, run_context = self._resolve_run_params(msg, thread_id) # Apply per-channel policy: credentials provider (e.g. GitHub diff --git a/backend/app/channels/run_policy.py b/backend/app/channels/run_policy.py index c78fbf27d..3e91a2290 100644 --- a/backend/app/channels/run_policy.py +++ b/backend/app/channels/run_policy.py @@ -76,6 +76,14 @@ class ChannelRunPolicy: fires. Defaults to False (the safe default for an interactive IM channel that depends on the manager to publish the agent's reply). + serialize_thread_runs: When True, the manager serializes + same-thread inbound turns for this channel instead of + surfacing the runtime's generic busy-thread error. This is + useful for chat surfaces like Feishu topics where rapid + follow-up messages should queue behind the active turn while + unrelated DeerFlow threads continue concurrently. Defaults + to False so existing channels keep the runtime's native + multitask behavior unless they opt in explicitly. """ is_interactive: bool = True @@ -83,6 +91,7 @@ class ChannelRunPolicy: credentials_provider: Callable[[InboundMessage, dict[str, Any]], Awaitable[None]] | None = None requires_bound_identity: bool = True fire_and_forget: bool = False + serialize_thread_runs: bool = False # Channel name → policy. Channels absent from this map fall through to diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 4cf30c075..a9124e9c3 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -1816,6 +1816,232 @@ class TestChannelManager: _run(go()) + def test_handle_feishu_same_thread_messages_queue_instead_of_busy(self, monkeypatch): + from app.channels.manager import THREAD_BUSY_MESSAGE, ChannelManager + + monkeypatch.setattr("app.channels.manager.STREAM_UPDATE_MIN_INTERVAL_SECONDS", 0.0) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store) + + outbound_received = [] + + async def capture_outbound(msg): + outbound_received.append(msg) + + bus.subscribe_outbound(capture_outbound) + + first_started = asyncio.Event() + release_first = asyncio.Event() + second_started = asyncio.Event() + + async def _stream(thread_id, assistant_id, *, input, **kwargs): # noqa: ARG001 + prompt = input["messages"][0]["content"] + if prompt == "first": + first_started.set() + await release_first.wait() + yield _make_stream_part( + "values", + { + "messages": [ + {"type": "human", "content": "first"}, + {"type": "ai", "content": "First done"}, + ], + "artifacts": [], + }, + ) + return + + second_started.set() + yield _make_stream_part( + "values", + { + "messages": [ + {"type": "human", "content": "second"}, + {"type": "ai", "content": "Second done"}, + ], + "artifacts": [], + }, + ) + + mock_client = _make_mock_langgraph_client(thread_id="feishu-thread-1") + mock_client.runs.stream = MagicMock(side_effect=_stream) + manager._client = mock_client + + await manager.start() + + await bus.publish_inbound( + InboundMessage( + channel_name="feishu", + chat_id="chat1", + user_id="user1", + text="first", + topic_id="topic-1", + thread_ts="om-source-1", + ) + ) + await _wait_for(first_started.is_set) + + await bus.publish_inbound( + InboundMessage( + channel_name="feishu", + chat_id="chat1", + user_id="user1", + text="second", + topic_id="topic-1", + thread_ts="om-source-2", + ) + ) + + await _wait_for(lambda: any(message.thread_ts == "om-source-2" and message.text.startswith("Queued behind another request") for message in outbound_received)) + assert second_started.is_set() is False + + release_first.set() + await _wait_for(second_started.is_set) + await _wait_for(lambda: len([message for message in outbound_received if message.is_final]) == 2) + await manager.stop() + + assert all(message.text != THREAD_BUSY_MESSAGE for message in outbound_received) + second_turn = [message for message in outbound_received if message.thread_ts == "om-source-2"] + assert second_turn[0].text.startswith("Queued behind another request") + assert any(message.text == "thinking..." for message in second_turn if message.is_final is False) + assert second_turn[-1].text == "Second done" + assert mock_client.runs.stream.call_count == 2 + + _run(go()) + + def test_handle_feishu_queue_waiter_cleanup_on_cancelled_progress_publish(self): + from app.channels.manager import ChannelManager + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store) + + msg = InboundMessage( + channel_name="feishu", + chat_id="chat1", + user_id="user1", + text="second", + topic_id="topic-1", + thread_ts="om-source-2", + ) + + thread_id = "feishu-thread-1" + serial_state, _ = manager._begin_serialized_thread_run( + channel_name="feishu", + thread_id=thread_id, + ) + assert serial_state is not None + await serial_state.lock.acquire() + + manager._get_client = MagicMock(return_value=object()) + manager._get_or_create_thread = AsyncMock(return_value=(thread_id, False)) + manager._update_thread_channel_metadata = AsyncMock() + manager._publish_progress_update = AsyncMock(side_effect=asyncio.CancelledError()) + manager._handle_chat_on_thread = AsyncMock() + + with pytest.raises(asyncio.CancelledError): + await manager._handle_chat(msg, bound_identity_checked=True) + + leaked_state = manager._serialized_thread_runs.get(("feishu", thread_id)) + assert leaked_state is serial_state + assert leaked_state.waiters == 1 + assert leaked_state.lock.locked() is True + manager._handle_chat_on_thread.assert_not_awaited() + + manager._finish_serialized_thread_run( + channel_name="feishu", + thread_id=thread_id, + state=serial_state, + lock_acquired=True, + ) + assert ("feishu", thread_id) not in manager._serialized_thread_runs + + _run(go()) + + def test_handle_feishu_different_threads_can_stream_concurrently(self, monkeypatch): + from app.channels.manager import ChannelManager + + monkeypatch.setattr("app.channels.manager.STREAM_UPDATE_MIN_INTERVAL_SECONDS", 0.0) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store) + + first_started = asyncio.Event() + second_started = asyncio.Event() + release_streams = asyncio.Event() + + async def create_thread(**kwargs): + topic_id = kwargs["metadata"]["channel_source"]["topic_id"] + return {"thread_id": f"thread-{topic_id}"} + + async def _stream(thread_id, assistant_id, *, input, **kwargs): # noqa: ARG001 + if thread_id == "thread-topic-a": + first_started.set() + elif thread_id == "thread-topic-b": + second_started.set() + await release_streams.wait() + yield _make_stream_part( + "values", + { + "messages": [ + {"type": "human", "content": input["messages"][0]["content"]}, + {"type": "ai", "content": f"done:{thread_id}"}, + ], + "artifacts": [], + }, + ) + + mock_client = _make_mock_langgraph_client() + mock_client.threads.create = AsyncMock(side_effect=create_thread) + mock_client.runs.stream = MagicMock(side_effect=_stream) + manager._client = mock_client + + outbound_received = [] + + async def capture_outbound(msg): + outbound_received.append(msg) + + bus.subscribe_outbound(capture_outbound) + + await manager.start() + await bus.publish_inbound( + InboundMessage( + channel_name="feishu", + chat_id="chat1", + user_id="user1", + text="first", + topic_id="topic-a", + thread_ts="om-source-a", + ) + ) + await bus.publish_inbound( + InboundMessage( + channel_name="feishu", + chat_id="chat1", + user_id="user1", + text="second", + topic_id="topic-b", + thread_ts="om-source-b", + ) + ) + + await _wait_for(first_started.is_set) + await _wait_for(second_started.is_set) + release_streams.set() + await _wait_for(lambda: len([message for message in outbound_received if message.is_final]) == 2) + await manager.stop() + + assert mock_client.runs.stream.call_count == 2 + assert not any(message.text.startswith("Queued behind another request") for message in outbound_received) + + _run(go()) + def test_handle_command_help(self): from app.channels.manager import ChannelManager @@ -3123,6 +3349,16 @@ class TestGithubFireAndForget: from app.channels.run_policy import ChannelRunPolicy assert ChannelRunPolicy().fire_and_forget is False + assert ChannelRunPolicy().serialize_thread_runs is False + + def test_feishu_channel_policy_opts_into_serialized_thread_runs(self): + """Feishu's queue-same-thread behavior should be policy-driven.""" + import app.channels.feishu_run_policy # noqa: F401 + from app.channels.run_policy import CHANNEL_RUN_POLICY + + feishu_policy = CHANNEL_RUN_POLICY.get("feishu") + assert feishu_policy is not None + assert feishu_policy.serialize_thread_runs is True def test_github_channel_policy_opts_into_fire_and_forget(self): """The GitHub channel must register ``fire_and_forget=True``. This is @@ -4358,6 +4594,48 @@ class TestFeishuChannel: _run(go()) + def test_prepare_inbound_topic_reply_includes_source_preview(self): + from app.channels.feishu import SOURCE_PREVIEW_METADATA_KEY, FeishuChannel + + async def go(): + bus = MessageBus() + bus.publish_inbound = AsyncMock() + channel = FeishuChannel(bus, config={}) + + reply_started = asyncio.Event() + release_reply = asyncio.Event() + + async def slow_reply(message_id: str, text: str) -> str: + reply_started.set() + await release_reply.wait() + return "om-running-card" + + channel._add_reaction = AsyncMock() + channel._reply_card = AsyncMock(side_effect=slow_reply) + + inbound = InboundMessage( + channel_name="feishu", + chat_id="chat-1", + user_id="user-1", + text="follow-up question", + thread_ts="om-source-msg", + metadata={SOURCE_PREVIEW_METADATA_KEY: "follow-up question"}, + ) + + prepare_task = asyncio.create_task(channel._prepare_inbound("om-source-msg", inbound)) + + await _wait_for(lambda: bus.publish_inbound.await_count == 1) + await _wait_for(reply_started.is_set) + + preview_text = channel._reply_card.await_args.args[1] + assert preview_text == "> follow-up question\n\nthinking..." + + await prepare_task + release_reply.set() + await _wait_for(lambda: channel._running_card_ids.get("om-source-msg") == "om-running-card") + + _run(go()) + def test_prepare_inbound_and_send_share_running_card_task(self): from app.channels.feishu import FeishuChannel @@ -4497,6 +4775,74 @@ class TestFeishuChannel: _run(go()) + def test_streaming_updates_preserve_source_preview(self): + from lark_oapi.api.im.v1 import ( + CreateMessageReactionRequest, + CreateMessageReactionRequestBody, + Emoji, + PatchMessageRequest, + PatchMessageRequestBody, + ReplyMessageRequest, + ReplyMessageRequestBody, + ) + + from app.channels.feishu import SOURCE_PREVIEW_METADATA_KEY, FeishuChannel + + async def go(): + bus = MessageBus() + channel = FeishuChannel(bus, config={}) + + channel._api_client = MagicMock() + channel._ReplyMessageRequest = ReplyMessageRequest + channel._ReplyMessageRequestBody = ReplyMessageRequestBody + channel._PatchMessageRequest = PatchMessageRequest + channel._PatchMessageRequestBody = PatchMessageRequestBody + channel._CreateMessageReactionRequest = CreateMessageReactionRequest + channel._CreateMessageReactionRequestBody = CreateMessageReactionRequestBody + channel._Emoji = Emoji + + reply_response = MagicMock() + reply_response.data.message_id = "om-running-card" + channel._api_client.im.v1.message.reply = MagicMock(return_value=reply_response) + channel._api_client.im.v1.message.patch = MagicMock() + channel._api_client.im.v1.message_reaction.create = MagicMock() + + metadata = {SOURCE_PREVIEW_METADATA_KEY: "What changed in the last run?"} + + await channel._send_running_reply("om-source-msg", metadata=metadata) + await channel.send( + OutboundMessage( + channel_name="feishu", + chat_id="chat-1", + thread_id="thread-1", + text="Queued behind another request", + is_final=False, + thread_ts="om-source-msg", + metadata=metadata, + ) + ) + await channel.send( + OutboundMessage( + channel_name="feishu", + chat_id="chat-1", + thread_id="thread-1", + text="Answer ready", + is_final=True, + thread_ts="om-source-msg", + metadata=metadata, + ) + ) + + reply_request = channel._api_client.im.v1.message.reply.call_args.args[0] + first_patch_request = channel._api_client.im.v1.message.patch.call_args_list[0].args[0] + final_patch_request = channel._api_client.im.v1.message.patch.call_args_list[1].args[0] + + assert json.loads(reply_request.body.content)["elements"][0]["content"] == "> What changed in the last run?\n\nthinking..." + assert json.loads(first_patch_request.body.content)["elements"][0]["content"] == "> What changed in the last run?\n\nQueued behind another request" + assert json.loads(final_patch_request.body.content)["elements"][0]["content"] == "> What changed in the last run?\n\nAnswer ready" + + _run(go()) + class TestWeComChannel: def test_publish_ws_inbound_starts_stream_and_publishes_message(self, monkeypatch): diff --git a/backend/tests/test_feishu_parser.py b/backend/tests/test_feishu_parser.py index 4dbf00b0c..e409055ba 100644 --- a/backend/tests/test_feishu_parser.py +++ b/backend/tests/test_feishu_parser.py @@ -328,7 +328,13 @@ def test_feishu_batches_top_level_file_messages_from_same_user(monkeypatch): assert inbound.metadata["message_id"] == "msg_file_1" assert inbound.metadata["topic_id"] == "msg_file_1" assert inbound.metadata["batched_message_ids"] == ["msg_file_1", "msg_file_2"] - channel._ensure_running_card_started.assert_called_once_with("msg_file_1") + channel._ensure_running_card_started.assert_called_once() + assert channel._ensure_running_card_started.call_args.args == ("msg_file_1",) + assert channel._ensure_running_card_started.call_args.kwargs["metadata"]["message_id"] == "msg_file_1" + assert channel._ensure_running_card_started.call_args.kwargs["metadata"]["batched_message_ids"] == [ + "msg_file_1", + "msg_file_2", + ] assert [call.args for call in channel._add_reaction.call_args_list] == [ ("msg_file_1", "OK"), ("msg_file_2", "OK"), @@ -390,6 +396,8 @@ def test_feishu_file_batch_window_expiry_starts_new_topic(monkeypatch): assert second.files == [{"file_key": "file_b"}] assert channel._ensure_running_card_started.call_args_list[0].args == ("msg_file_1",) assert channel._ensure_running_card_started.call_args_list[1].args == ("msg_file_2",) + assert channel._ensure_running_card_started.call_args_list[0].kwargs["metadata"]["message_id"] == "msg_file_1" + assert channel._ensure_running_card_started.call_args_list[1].kwargs["metadata"]["message_id"] == "msg_file_2" _run(go())