"""Tests for the Feishu interactive-card clarification module (outbound half). Rendering, ownership, one-time consumption and bounded registry lifecycle. """ from __future__ import annotations import asyncio import json import time from unittest.mock import AsyncMock, MagicMock import pytest from app.channels.feishu_card_interaction import ( MAX_CARD_OPTIONS, ClarificationCardRequest, PendingCardRegistry, build_button_card, ) from app.channels.message_bus import HUMAN_INPUT_RESPONSE_METADATA_KEY, InboundMessage, MessageBus from app.channels.store import JsonChannelStore def _metadata(payload: dict | None) -> dict: return {} if payload is None else {"human_input_request": payload} def _payload(**overrides) -> dict: base = { "version": 1, "kind": "human_input_request", "source": "ask_clarification", "request_id": "clarification:call_1", "question": "Pick one", "input_mode": "choice_with_other", "options": [ {"id": "option-1", "label": "Alpha", "value": "Alpha"}, {"id": "option-2", "label": "Beta", "value": "Beta"}, ], } base.update(overrides) return base class TestClarificationCardRequestExtraction: def test_absent_payload_returns_none(self): assert ClarificationCardRequest.from_metadata(_metadata(None)) is None def test_free_text_mode_returns_none(self): assert ClarificationCardRequest.from_metadata(_metadata(_payload(input_mode="free_text", options=[]))) is None def test_form_mode_returns_none(self): assert ClarificationCardRequest.from_metadata(_metadata(_payload(version=2, input_mode="form", fields=[{"key": "x"}]))) is None def test_empty_options_returns_none(self): assert ClarificationCardRequest.from_metadata(_metadata(_payload(options=[]))) is None def test_overflow_options_returns_none(self): options = [{"id": f"option-{i}", "label": f"L{i}", "value": f"L{i}"} for i in range(1, MAX_CARD_OPTIONS + 2)] assert ClarificationCardRequest.from_metadata(_metadata(_payload(options=options))) is None def test_choice_with_options_extracts(self): req = ClarificationCardRequest.from_metadata(_metadata(_payload())) assert req is not None assert req.request_id == "clarification:call_1" assert req.question == "Pick one" assert req.options == ("Alpha", "Beta") def test_custom_option_ids_pass_through_to_buttons(self): req = ClarificationCardRequest.from_metadata( _metadata( _payload( options=[ {"id": "approve", "label": "批准", "value": "批准"}, {"id": "reject", "label": "拒绝", "value": "拒绝"}, ] ) ) ) assert req is not None card = json.loads(build_button_card(req)) actions = next(e for e in card["elements"] if e["tag"] == "action")["actions"] assert [a["value"]["option_id"] for a in actions] == ["approve", "reject"] def test_malformed_payload_returns_none(self): assert ClarificationCardRequest.from_metadata(_metadata({"kind": "other"})) is None assert ClarificationCardRequest.from_metadata({"human_input_request": "not-a-dict"}) is None class TestBuildButtonCard: def test_card_preserves_context_before_question(self): req = ClarificationCardRequest.from_metadata(_metadata(_payload(question="是否继续？", context="这将**永久删除**生产表中的数据。\n请先确认已有备份。"))) assert req is not None card = json.loads(build_button_card(req)) markdown = next(element for element in card["elements"] if element["tag"] == "markdown") assert markdown["content"] == "这将**永久删除**生产表中的数据。\n请先确认已有备份。\n\n是否继续？" @pytest.mark.parametrize("context", [None, "", " \t\n ", 42, ["background"]]) def test_missing_or_invalid_context_keeps_question_only(self, context): req = ClarificationCardRequest.from_metadata(_metadata(_payload(context=context))) assert req is not None card = json.loads(build_button_card(req)) markdown = next(element for element in card["elements"] if element["tag"] == "markdown") assert markdown["content"] == "Pick one" def test_card_structure_binds_buttons_to_request(self): req = ClarificationCardRequest.from_metadata(_metadata(_payload())) assert req is not None assert req.option_ids == ("option-1", "option-2") card = json.loads(build_button_card(req)) assert card["config"]["update_multi"] is True assert card["header"]["title"]["content"] elements = card["elements"] markdown = next(e for e in elements if e["tag"] == "markdown") assert "Pick one" in markdown["content"] actions = next(e for e in elements if e["tag"] == "action")["actions"] assert len(actions) == 2 for i, (action, label) in enumerate(zip(actions, ("Alpha", "Beta")), start=1): assert action["tag"] == "button" assert action["text"]["content"] == label assert action["value"]["callback_token"] == req.callback_token assert action["value"]["option_id"] == f"option-{i}" assert action["value"]["option_value"] == label assert actions[0]["type"] == "primary" assert actions[1]["type"] == "default" class TestPendingCardRegistry: def _entry(self, request_id: str = "clarification:call_1", **overrides) -> dict: entry = { "callback_token": "callback-1", "request_id": request_id, "source": "ask_clarification", "thread_id": "thread-1", "chat_id": "chat-1", "originator_open_id": "ou_originator", "options": ("Alpha", "Beta"), "option_ids": ("option-1", "option-2"), } entry.update(overrides) return entry def test_register_then_consume_returns_entry_once(self): registry = PendingCardRegistry() registry.register(**self._entry(topic_id="topic-1", connection_id="conn-1", owner_user_id="deer-user-1")) entry, reason = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert reason is None and entry is not None assert entry["originator_open_id"] == "ou_originator" assert entry["topic_id"] == "topic-1" assert entry["connection_id"] == "conn-1" assert entry["owner_user_id"] == "deer-user-1" entry2, _ = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert entry2 is None # one-time def test_same_request_id_keeps_independent_cards(self): registry = PendingCardRegistry() registry.register(**self._entry()) registry.register(**self._entry(callback_token="callback-2", originator_open_id="ou_new")) first, _ = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert first is not None and first["originator_open_id"] == "ou_originator" entry, _ = registry.consume_if_originator("callback-2", "ou_new", option_id="option-1", option_value="Alpha") assert entry is not None assert entry["originator_open_id"] == "ou_new" def test_duplicate_callback_token_cannot_replace_a_pending_card(self): registry = PendingCardRegistry() registry.register(**self._entry()) with pytest.raises(ValueError, match="already owns"): registry.register(**self._entry(request_id="clarification:other")) entry, _ = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert entry is not None and entry["request_id"] == "clarification:call_1" def test_register_prunes_expired_cards_without_clicks(self, monkeypatch): clock = [100.0] monkeypatch.setattr("app.channels.feishu_card_interaction.time.time", lambda: clock[0]) registry = PendingCardRegistry(ttl_seconds=10) registry.register(**self._entry()) clock[0] += 11 registry.register(**self._entry(callback_token="callback-2")) assert len(registry._entries) == 1 entry, _ = registry.consume_if_originator("callback-2", "ou_originator", option_id="option-1", option_value="Alpha") assert entry is not None def test_capacity_retires_oldest_card_and_keeps_new_cards_usable(self): registry = PendingCardRegistry(max_entries=2) for index in range(3): registry.register(**self._entry(callback_token=f"callback-{index}")) oldest, _ = registry.consume_if_originator("callback-0", "ou_originator", option_id="option-1", option_value="Alpha") assert oldest is None for index in (1, 2): entry, _ = registry.consume_if_originator(f"callback-{index}", "ou_originator", option_id="option-1", option_value="Alpha") assert entry is not None def test_discard_thread_retires_all_matching_cards(self): registry = PendingCardRegistry() registry.register(**self._entry(callback_token="cb-1", thread_id="thread-target")) registry.register(**self._entry(callback_token="cb-2", thread_id="thread-target")) registry.register(**self._entry(callback_token="cb-3", thread_id="thread-other")) assert len(registry) == 3 pruned = registry.discard_thread("thread-target") assert pruned == 2 assert len(registry) == 1 entry, _ = registry.consume_if_originator("cb-3", "ou_originator", option_id="option-1", option_value="Alpha") assert entry is not None def test_prune_expired_removes_stale_cards_and_returns_count(self): registry = PendingCardRegistry(ttl_seconds=10) registry.register(**self._entry(callback_token="cb-fresh")) registry.register(**self._entry(callback_token="cb-stale")) registry._entries["cb-stale"]["created_at"] -= 15 cleaned = registry.prune_expired() assert cleaned == 1 assert len(registry) == 1 assert "cb-fresh" in registry._entries assert "cb-stale" not in registry._entries def test_get_pending_entry_provides_read_only_view(self): registry = PendingCardRegistry() registry.register(**self._entry(callback_token="cb-inspect", thread_id="th-100")) view = registry.get_pending_entry("cb-inspect") assert view is not None assert view["thread_id"] == "th-100" assert len(registry) == 1 consumed, _ = registry.consume_if_originator("cb-inspect", "ou_originator", option_id="option-1", option_value="Alpha") assert consumed is not None assert registry.get_pending_entry("cb-inspect") is None class TestFeishuChannelButtonCardWiring: """Channel-level wiring: when the flag and payload line up, final clarification outbounds go out as button cards; everything else is byte-identical to today's plain-text path.""" def _channel(self, *, enabled: bool, tmp_path): from unittest.mock import AsyncMock from app.channels.feishu import FeishuChannel from app.channels.message_bus import MessageBus channel = FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test", "card_interaction_enabled": enabled, "channel_store": JsonChannelStore(path=tmp_path / "routes.json")}) channel._reply_card = AsyncMock(return_value="om_card_final") channel._update_card = AsyncMock() channel._add_reaction = AsyncMock() channel._remember_thread_mapping = AsyncMock() return channel def _clarification_outbound(self, *, with_options: bool = True): from app.channels.message_bus import ( HUMAN_INPUT_REQUEST_METADATA_KEY, PENDING_CLARIFICATION_METADATA_KEY, OutboundMessage, ) metadata = { PENDING_CLARIFICATION_METADATA_KEY: True, "user_id": "ou_originator", "topic_id": "topic-1", "message_id": "om_src", } if with_options: metadata[HUMAN_INPUT_REQUEST_METADATA_KEY] = { "version": 1, "kind": "human_input_request", "source": "ask_clarification", "request_id": "clarification:call_1", "question": "Pick one", "input_mode": "choice_with_other", "options": [ {"id": "option-1", "label": "Alpha", "value": "Alpha"}, {"id": "option-2", "label": "Beta", "value": "Beta"}, ], } else: metadata[HUMAN_INPUT_REQUEST_METADATA_KEY] = { "version": 1, "kind": "human_input_request", "source": "ask_clarification", "request_id": "clarification:call_2", "question": "Say anything", "input_mode": "free_text", } return OutboundMessage( channel_name="feishu", chat_id="chat-1", thread_id="thread-1", text="Pick one", is_final=True, thread_ts="om_src", metadata=metadata, ) @pytest.mark.asyncio async def test_flag_on_options_renders_button_card_and_registers(self, tmp_path): channel = self._channel(enabled=True, tmp_path=tmp_path) await channel._send_card_message(self._clarification_outbound()) card_json = channel._reply_card.call_args.args[1] card = json.loads(card_json) assert any(e.get("tag") == "action" for e in card["elements"]) token = next(e for e in card["elements"] if e["tag"] == "action")["actions"][0]["value"]["callback_token"] entry, reason = channel._card_registry.consume_if_originator(token, "ou_originator", option_id="option-1", option_value="Alpha") assert reason is None and entry is not None assert entry["originator_open_id"] == "ou_originator" assert entry["thread_id"] == "thread-1" assert entry["card_message_id"] == "om_card_final" @pytest.mark.asyncio async def test_flag_off_falls_back_to_plain_card(self, tmp_path): channel = self._channel(enabled=False, tmp_path=tmp_path) await channel._send_card_message(self._clarification_outbound()) # Plain-text path: markdown text, no raw-card bypass, nothing registered. text_arg = channel._reply_card.call_args.args[1] assert "Pick one" in text_arg assert '"action"' not in text_arg assert channel._reply_card.call_args.kwargs.get("raw_card_content") is not True entry, _ = channel._card_registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert entry is None @pytest.mark.asyncio async def test_flag_on_free_text_falls_back_to_plain_card(self, tmp_path): channel = self._channel(enabled=True, tmp_path=tmp_path) await channel._send_card_message(self._clarification_outbound(with_options=False)) text_arg = channel._reply_card.call_args.args[1] assert '"action"' not in text_arg assert channel._reply_card.call_args.kwargs.get("raw_card_content") is not True entry, _ = channel._card_registry.consume_if_originator("clarification:call_2", "ou_originator", option_id="option-1", option_value="x") assert entry is None @pytest.mark.asyncio async def test_running_card_is_updated_into_button_card(self, tmp_path): channel = self._channel(enabled=True, tmp_path=tmp_path) channel._running_card_ids["om_src"] = "om_running" await channel._send_card_message(self._clarification_outbound()) card_json = channel._update_card.call_args.args[1] card = json.loads(card_json) assert any(e.get("tag") == "action" for e in card["elements"]) token = next(e for e in card["elements"] if e["tag"] == "action")["actions"][0]["value"]["callback_token"] entry, _ = channel._card_registry.consume_if_originator(token, "ou_originator", option_id="option-1", option_value="Alpha") assert entry is not None assert entry["card_message_id"] == "om_running" class TestConsumeIfOriginator: """Atomic validate-and-consume for card clicks. Identity mismatch must NOT consume the entry (the real originator may still click later).""" def _registry_with_entry(self): registry = PendingCardRegistry() registry.register( callback_token="callback-1", request_id="clarification:call_1", source="ask_clarification", thread_id="thread-1", chat_id="chat-1", originator_open_id="ou_originator", options=("Alpha", "Beta"), option_ids=("option-1", "option-2"), card_message_id="om_card_1", ) return registry def test_originator_consumes_and_gets_entry(self): registry = self._registry_with_entry() entry, reason = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert reason is None assert entry is not None assert entry["card_message_id"] == "om_card_1" # replayed click: entry already gone entry2, reason2 = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert entry2 is None assert reason2 == "unknown" def test_non_originator_rejected_without_consuming(self): registry = self._registry_with_entry() entry, reason = registry.consume_if_originator("callback-1", "ou_someone_else", option_id="option-1", option_value="Alpha") assert entry is None assert reason == "not_originator" # the real originator can still click afterwards entry2, reason2 = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert reason2 is None assert entry2 is not None def test_unknown_request_id(self): registry = self._registry_with_entry() entry, reason = registry.consume_if_originator("clarification:nope", "ou_originator", option_id="option-1", option_value="Alpha") assert entry is None assert reason == "unknown" def test_expired_entry_is_unknown(self): registry = self._registry_with_entry() future = time.time() + 31 * 60 entry, reason = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha", now=future) assert entry is None assert reason == "unknown" def test_unknown_option_value_rejected_without_consuming(self): registry = self._registry_with_entry() entry, reason = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="注入值") assert entry is None assert reason == "unknown" # the real button value still works afterwards entry2, reason2 = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert reason2 is None and entry2 is not None def _make_card_action_event(*, callback_token: str, option_value: str, option_id: str = "option-1", operator_open_id: str = "ou_originator"): """Build a minimal P2CardActionTrigger-shaped stand-in.""" from unittest.mock import MagicMock event = MagicMock() event.event.action.value = {"callback_token": callback_token, "option_id": option_id, "option_value": option_value} event.event.operator.open_id = operator_open_id return event class TestOnCardActionValidation: """Sync layer of the card-action callback: validation + toast, no state change.""" def _channel(self): from unittest.mock import MagicMock from app.channels.feishu import FeishuChannel from app.channels.message_bus import MessageBus channel = FeishuChannel(MessageBus(), {"app_id": "t", "app_secret": "t", "card_interaction_enabled": True}) channel._card_registry.register( callback_token="callback-1", request_id="clarification:call_1", source="ask_clarification", thread_id="thread-1", chat_id="chat-1", originator_open_id="ou_originator", options=("Alpha", "Beta"), option_ids=("option-1", "option-2"), card_message_id="om_card_1", ) channel._main_loop = MagicMock() channel._main_loop.is_running.return_value = True channel._submit_threadsafe_coroutine = MagicMock(side_effect=lambda coroutine, *args, **kwargs: (coroutine.close(), True)[1]) class _Toast: pass class _Resp: pass channel._CallBackToast = _Toast channel._P2CardActionTriggerResponse = _Resp return channel def test_non_originator_gets_refusal_toast_and_entry_survives(self): channel = self._channel() resp = channel._on_card_action(_make_card_action_event(callback_token="callback-1", option_id="option-1", option_value="Alpha", operator_open_id="ou_other")) assert resp.toast is not None assert "发起人" in resp.toast.content channel._submit_threadsafe_coroutine.assert_not_called() # entry not consumed: originator can still click entry, reason = channel._card_registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert reason is None and entry is not None def test_unknown_request_id_gets_stale_toast(self): channel = self._channel() resp = channel._on_card_action(_make_card_action_event(callback_token="unknown", option_id="option-1", option_value="Alpha")) assert resp.toast is not None assert "失效" in resp.toast.content channel._submit_threadsafe_coroutine.assert_not_called() def test_dispatch_failure_returns_warning_toast(self): channel = self._channel() channel._submit_threadsafe_coroutine = MagicMock(side_effect=lambda coroutine, *args, **kwargs: (coroutine.close(), False)[1]) resp = channel._on_card_action(_make_card_action_event(callback_token="callback-1", option_id="option-1", option_value="Alpha")) assert resp.toast is not None assert resp.toast.type == "warning" def test_originator_gets_ack_toast_and_dispatches_async_resolution(self): channel = self._channel() resp = channel._on_card_action(_make_card_action_event(callback_token="callback-1", option_id="option-1", option_value="Alpha")) assert resp.toast is not None assert "Alpha" in resp.toast.content channel._submit_threadsafe_coroutine.assert_called_once() # one-time: the entry is consumed by the click entry, reason = channel._card_registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") assert entry is None and reason == "unknown" class TestResolveCardClick: """Async layer: terminal card update + hidden synthetic inbound publication.""" @pytest.mark.asyncio async def test_publishes_hidden_inbound_and_marks_card_resolved(self): from unittest.mock import AsyncMock from app.channels.feishu import FeishuChannel from app.channels.message_bus import HUMAN_INPUT_RESPONSE_METADATA_KEY, MessageBus bus = MessageBus() channel = FeishuChannel(bus, {"app_id": "t", "app_secret": "t", "card_interaction_enabled": True}) channel._update_card = AsyncMock() published = [] async def capture(msg): published.append(msg) return True channel._publish_inbound_or_drop = capture entry = { "request_id": "clarification:call_1", "source": "ask_clarification", "thread_id": "thread-1", "chat_id": "chat-1", "originator_open_id": "ou_originator", "options": ("Alpha", "Beta"), "option_ids": ("option-1", "option-2"), "card_message_id": "om_card_1", "topic_id": "om_topic_1", "connection_id": "conn-1", "owner_user_id": "deer-user-1", } await channel._resolve_card_click(entry, option_id="option-1", option_value="Alpha", operator_open_id="ou_originator") # terminal card: no buttons, shows the choice assert channel._update_card.await_count == 1 terminal_text = channel._update_card.await_args.args[1] assert "Alpha" in terminal_text # synthetic inbound: hidden human-input response routed at the card message, # carrying the routing identity the manager keys on (topic + connection). assert len(published) == 1 inbound = published[0] assert inbound.chat_id == "chat-1" assert inbound.user_id == "ou_originator" assert inbound.thread_ts == "om_card_1" assert inbound.topic_id == "om_topic_1" assert inbound.connection_id == "conn-1" assert inbound.owner_user_id == "deer-user-1" response = inbound.metadata[HUMAN_INPUT_RESPONSE_METADATA_KEY] assert response["kind"] == "human_input_response" assert response["response_kind"] == "option" assert response["request_id"] == "clarification:call_1" assert response["value"] == "Alpha" assert inbound.metadata.get("hide_from_ui") is True @pytest.mark.asyncio async def test_dropped_inbound_leaves_card_unresolved(self): """If the inbound cannot be published, the card must not flip to a false resolved state.""" from unittest.mock import AsyncMock from app.channels.feishu import FeishuChannel from app.channels.message_bus import MessageBus channel = FeishuChannel(MessageBus(), {"app_id": "t", "app_secret": "t", "card_interaction_enabled": True}) channel._update_card = AsyncMock() channel._publish_inbound_or_drop = AsyncMock(return_value=False) entry = { "request_id": "clarification:call_1", "source": "ask_clarification", "thread_id": "thread-1", "chat_id": "chat-1", "originator_open_id": "ou_originator", "options": ("Alpha",), "card_message_id": "om_card_1", } await channel._resolve_card_click(entry, option_id="option-1", option_value="Alpha", operator_open_id="ou_originator") channel._update_card.assert_not_awaited() class TestCardClickEndToEndRouting: """Pin the contract the reviewer flagged: the synthetic inbound must carry the routing identity the manager's thread resolution keys on, so a card click resumes the clarified thread instead of opening a new one.""" def test_card_answer_resumes_existing_thread(self, tmp_path, monkeypatch): from app.channels.manager import ChannelManager from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialCipher from deerflow.persistence.engine import close_engine, get_session_factory, init_engine monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) async def go(): await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'channel-connections.db'}", sqlite_dir=str(tmp_path)) repo = ChannelConnectionRepository(get_session_factory(), cipher=ChannelCredentialCipher.from_key("test-channel-key")) conn = await repo.upsert_connection( owner_user_id="deer-user-1", provider="feishu", external_account_id="ou_originator", workspace_id="chat-1", ) # What the channel persisted when the button card was delivered: # the card's topic maps to the clarified thread. await repo.set_thread_id( connection_id=conn["id"], owner_user_id="deer-user-1", provider="feishu", external_conversation_id="chat-1", thread_id="thread-existing", external_topic_id="om_topic_1", ) bus = MessageBus() store = JsonChannelStore(path=tmp_path / "legacy-store.json") manager = ChannelManager(bus=bus, store=store, connection_repo=repo) mock_client = MagicMock() mock_client.threads.create = AsyncMock(return_value={"thread_id": "thread-existing"}) mock_client.threads.update = AsyncMock(return_value={"thread_id": "thread-existing"}) mock_client.threads.get = AsyncMock(return_value={"thread_id": "thread-existing"}) mock_client.runs.wait = AsyncMock(return_value={"messages": [{"type": "human", "content": "Alpha"}, {"type": "ai", "content": "resumed"}]}) # Feishu is a streaming channel: the run goes through runs.stream, # whose first positional argument is the thread id. class _Part: def __init__(self, event, data): self.event = event self.data = data async def _stream(*args, **kwargs): yield _Part("values", {"messages": [{"type": "human", "content": "Alpha"}, {"type": "ai", "content": "resumed"}], "artifacts": []}) mock_client.runs.stream = MagicMock(side_effect=lambda *args, **kwargs: _stream(*args, **kwargs)) manager._client = mock_client # The synthetic inbound exactly as _resolve_card_click builds it. inbound = InboundMessage( channel_name="feishu", chat_id="chat-1", user_id="ou_originator", owner_user_id="deer-user-1", connection_id=conn["id"], text="Alpha", thread_ts="om_card_1", topic_id="om_topic_1", metadata={ "hide_from_ui": True, HUMAN_INPUT_RESPONSE_METADATA_KEY: { "version": 1, "kind": "human_input_response", "source": "ask_clarification", "request_id": "clarification:call_1", "response_kind": "option", "option_id": "option-1", "value": "Alpha", }, }, ) await manager._handle_chat(inbound) # The clarified thread is reused — nothing is created anew. mock_client.threads.create.assert_not_called() assert mock_client.runs.stream.call_count == 1 assert mock_client.runs.stream.call_args.args[0] == "thread-existing" # And the posted human message carries the hidden human-input response. posted = mock_client.runs.stream.call_args.kwargs["input"]["messages"][0] assert posted["additional_kwargs"]["hide_from_ui"] is True assert posted["additional_kwargs"][HUMAN_INPUT_RESPONSE_METADATA_KEY]["value"] == "Alpha" try: asyncio.run(go()) finally: asyncio.run(close_engine()) def test_card_answer_passes_bound_identity_admission(self, tmp_path, monkeypatch): """Connections-enabled deployments (require_bound_identity=True) re-read the binding with workspace_id; Feishu binds connections with workspace_id=chat_id, so the synthetic inbound must carry it or every click is rejected as unbound.""" from app.channels.manager import ChannelManager from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialCipher from deerflow.persistence.engine import close_engine, get_session_factory, init_engine monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) async def go(): await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'cc.db'}", sqlite_dir=str(tmp_path)) repo = ChannelConnectionRepository(get_session_factory(), cipher=ChannelCredentialCipher.from_key("test-channel-key")) conn = await repo.upsert_connection( owner_user_id="deer-user-1", provider="feishu", external_account_id="ou_originator", workspace_id="chat-1", ) await repo.set_thread_id( connection_id=conn["id"], owner_user_id="deer-user-1", provider="feishu", external_conversation_id="chat-1", thread_id="thread-existing", external_topic_id="om_topic_1", ) bus = MessageBus() store = JsonChannelStore(path=tmp_path / "legacy-store.json") manager = ChannelManager(bus=bus, store=store, connection_repo=repo, require_bound_identity=True) mock_client = MagicMock() mock_client.threads.create = AsyncMock(return_value={"thread_id": "thread-existing"}) mock_client.threads.update = AsyncMock(return_value={"thread_id": "thread-existing"}) mock_client.threads.get = AsyncMock(return_value={"thread_id": "thread-existing"}) class _Part: def __init__(self, event, data): self.event = event self.data = data async def _stream(*args, **kwargs): yield _Part("values", {"messages": [{"type": "human", "content": "Alpha"}, {"type": "ai", "content": "resumed"}], "artifacts": []}) mock_client.runs.stream = MagicMock(side_effect=lambda *args, **kwargs: _stream(*args, **kwargs)) manager._client = mock_client # Build the inbound through the channel's real click-resolution path, # then hand it to the manager — this is the production seam. from app.channels.feishu import FeishuChannel channel = FeishuChannel(MessageBus(), {"app_id": "t", "app_secret": "t", "card_interaction_enabled": True}) channel._update_card = AsyncMock() produced = [] async def capture(msg): produced.append(msg) return True channel._publish_inbound_or_drop = capture entry = { "request_id": "clarification:call_1", "source": "ask_clarification", "thread_id": "thread-existing", "chat_id": "chat-1", "originator_open_id": "ou_originator", "options": ("Alpha", "Beta"), "option_ids": ("option-1", "option-2"), "card_message_id": "om_card_1", "topic_id": "om_topic_1", "connection_id": conn["id"], "owner_user_id": "deer-user-1", } await channel._resolve_card_click(entry, option_id="option-1", option_value="Alpha", operator_open_id="ou_originator") assert len(produced) == 1 await manager._handle_chat(produced[0]) mock_client.threads.create.assert_not_called() assert mock_client.runs.stream.call_count == 1 assert mock_client.runs.stream.call_args.args[0] == "thread-existing" try: asyncio.run(go()) finally: asyncio.run(close_engine()) class TestThreadReplacementRetiresCards: """Integration: ``/new`` retires pending cards for the old thread so stale button clicks are correctly rejected instead of lingering.""" def _channel_with_cards(self, *, thread_id: str = "thread-old"): from app.channels.feishu import FeishuChannel from app.channels.message_bus import MessageBus channel = FeishuChannel( MessageBus(), {"app_id": "t", "app_secret": "t", "card_interaction_enabled": True}, ) # Register two pending cards on the target thread and one on another. for token_suffix in ("a", "b"): channel._card_registry.register( callback_token=f"cb-{token_suffix}", request_id=f"req-{token_suffix}", source="ask_clarification", thread_id=thread_id, chat_id="chat-1", originator_open_id="ou_originator", options=("Alpha", "Beta"), option_ids=("option-1", "option-2"), card_message_id=f"om_card_{token_suffix}", ) channel._card_registry.register( callback_token="cb-other", request_id="req-other", source="ask_clarification", thread_id="thread-unrelated", chat_id="chat-1", originator_open_id="ou_originator", options=("Alpha",), option_ids=("option-1",), card_message_id="om_card_other", ) return channel def test_on_thread_replaced_retires_old_thread_cards(self): channel = self._channel_with_cards() assert len(channel._card_registry) == 3 channel.on_thread_replaced("thread-old") # The two cards for the replaced thread are gone. assert len(channel._card_registry) == 1 entry_a, _ = channel._card_registry.consume_if_originator("cb-a", "ou_originator", option_id="option-1", option_value="Alpha") assert entry_a is None entry_b, _ = channel._card_registry.consume_if_originator("cb-b", "ou_originator", option_id="option-1", option_value="Alpha") assert entry_b is None # The unrelated thread's card is unaffected. entry_other, reason = channel._card_registry.consume_if_originator("cb-other", "ou_originator", option_id="option-1", option_value="Alpha") assert reason is None and entry_other is not None def test_on_thread_replaced_is_noop_when_card_interaction_disabled(self): from app.channels.feishu import FeishuChannel from app.channels.message_bus import MessageBus channel = FeishuChannel( MessageBus(), {"app_id": "t", "app_secret": "t", "card_interaction_enabled": False}, ) channel._card_registry.register( callback_token="cb-1", request_id="req-1", source="ask_clarification", thread_id="thread-old", chat_id="chat-1", originator_open_id="ou_originator", options=("Alpha",), option_ids=("option-1",), card_message_id="om_card_1", ) channel.on_thread_replaced("thread-old") # Card interaction is off, so nothing is retired. assert len(channel._card_registry) == 1 def test_manager_new_command_notifies_channel(self, tmp_path): """The ``/new`` command path in ChannelManager looks up the old thread and calls ``channel.on_thread_replaced`` before replying.""" from unittest.mock import patch from app.channels.feishu import FeishuChannel from app.channels.manager import ChannelManager from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus from app.channels.store import JsonChannelStore async def go(): bus = MessageBus() store = JsonChannelStore(path=tmp_path / "routes.json") manager = ChannelManager(bus=bus, store=store) # Pre-seed the thread mapping so /new finds an old thread to replace. channel = FeishuChannel( MessageBus(), {"app_id": "t", "app_secret": "t", "card_interaction_enabled": True}, ) channel._card_registry.register( callback_token="cb-old", request_id="req-old", source="ask_clarification", thread_id="thread-old", chat_id="chat-1", originator_open_id="ou_originator", options=("Alpha",), option_ids=("option-1",), card_message_id="om_card_old", ) mock_client = MagicMock() mock_client.threads.create = AsyncMock(return_value={"thread_id": "thread-new"}) manager._client = mock_client # Simulate: _lookup_thread_id returns the old thread manager._lookup_thread_id = AsyncMock(return_value="thread-old") manager._store_thread_id = AsyncMock() inbound = InboundMessage( channel_name="feishu", chat_id="chat-1", user_id="ou_originator", text="/new", msg_type=InboundMessageType.COMMAND, ) # Patch get_channel_service to return our channel instance mock_service = MagicMock() mock_service.get_channel.return_value = channel with patch("app.channels.service.get_channel_service", return_value=mock_service): await manager._handle_command(inbound) # The old thread's card should be retired. assert len(channel._card_registry) == 0 entry, _ = channel._card_registry.consume_if_originator("cb-old", "ou_originator", option_id="option-1", option_value="Alpha") assert entry is None asyncio.run(go())