diff --git a/nanobot/channels/feishu.py b/nanobot/channels/feishu.py index 38155434..58a0b1c1 100644 --- a/nanobot/channels/feishu.py +++ b/nanobot/channels/feishu.py @@ -1420,16 +1420,11 @@ class FeishuChannel(BaseChannel): self.logger.warning("Error stream-updating card {}: {}", card_id, e) return False - def _close_streaming_mode_sync(self, card_id: str, sequence: int) -> bool: - """Turn off CardKit streaming_mode so the chat list preview exits the streaming placeholder. - - Per Feishu docs, streaming cards keep a generating-style summary in the session list until - streaming_mode is set to false via card settings (after final content update). - Sequence must strictly exceed the previous card OpenAPI operation on this entity. - """ + def _set_streaming_mode_sync(self, card_id: str, enabled: bool, sequence: int) -> bool: + """Set CardKit streaming_mode using a strictly increasing sequence.""" from lark_oapi.api.cardkit.v1 import SettingsCardRequest, SettingsCardRequestBody - settings_payload = json.dumps({"config": {"streaming_mode": False}}, ensure_ascii=False) + settings_payload = json.dumps({"config": {"streaming_mode": enabled}}, ensure_ascii=False) try: request = ( SettingsCardRequest.builder() @@ -1446,7 +1441,8 @@ class FeishuChannel(BaseChannel): response = self._client.cardkit.v1.card.settings(request) if not response.success(): self.logger.warning( - "Failed to close streaming on card {}: code={}, msg={}", + "Failed to set streaming={} on card {}: code={}, msg={}", + enabled, card_id, response.code, response.msg, @@ -1454,9 +1450,32 @@ class FeishuChannel(BaseChannel): return False return True except Exception as e: - self.logger.warning("Error closing streaming on card {}: {}", card_id, e) + self.logger.warning("Error setting streaming={} on card {}: {}", enabled, card_id, e) return False + def _close_streaming_mode_sync(self, card_id: str, sequence: int) -> bool: + """Turn off CardKit streaming_mode so the chat list preview exits the streaming placeholder. + + Per Feishu docs, streaming cards keep a generating-style summary in the session list until + streaming_mode is set to false via card settings (after final content update). + Sequence must strictly exceed the previous card OpenAPI operation on this entity. + """ + return self._set_streaming_mode_sync(card_id, False, sequence) + + def _stream_update_text_with_reopen_sync( + self, + card_id: str, + content: str, + sequence: int, + ) -> tuple[bool, int]: + if self._stream_update_text_sync(card_id, content, sequence): + return True, sequence + sequence += 1 + if not self._set_streaming_mode_sync(card_id, True, sequence): + return False, sequence + sequence += 1 + return self._stream_update_text_sync(card_id, content, sequence), sequence + async def send_delta( self, chat_id: str, delta: str, metadata: dict[str, Any] | None = None ) -> None: @@ -1499,22 +1518,37 @@ class FeishuChannel(BaseChannel): # back to sending a regular interactive card. if buf.card_id: buf.sequence += 1 - ok = await loop.run_in_executor( + ok, buf.sequence = await loop.run_in_executor( None, - self._stream_update_text_sync, + self._stream_update_text_with_reopen_sync, buf.card_id, buf.text, buf.sequence, ) if ok: buf.sequence += 1 - await loop.run_in_executor( + closed = await loop.run_in_executor( None, self._close_streaming_mode_sync, buf.card_id, buf.sequence, ) + if not closed: + buf.sequence += 1 + await loop.run_in_executor( + None, + self._close_streaming_mode_sync, + buf.card_id, + buf.sequence, + ) return + buf.sequence += 1 + await loop.run_in_executor( + None, + self._close_streaming_mode_sync, + buf.card_id, + buf.sequence, + ) self.logger.warning( "Streaming card {} final update failed, falling back to regular card", buf.card_id, @@ -1567,18 +1601,36 @@ class FeishuChannel(BaseChannel): ), ) if card_id: - buf.card_id = card_id - buf.sequence = 1 - await loop.run_in_executor( - None, self._stream_update_text_sync, card_id, buf.text, 1 + ok, sequence = await loop.run_in_executor( + None, self._stream_update_text_with_reopen_sync, card_id, buf.text, 1 ) - buf.last_edit = now + if ok: + buf.card_id = card_id + buf.sequence = sequence + buf.last_edit = now + else: + await loop.run_in_executor( + None, self._close_streaming_mode_sync, card_id, sequence + 1 + ) elif (now - buf.last_edit) >= self._STREAM_EDIT_INTERVAL: - buf.sequence += 1 - await loop.run_in_executor( - None, self._stream_update_text_sync, buf.card_id, buf.text, buf.sequence + ok, buf.sequence = await loop.run_in_executor( + None, + self._stream_update_text_with_reopen_sync, + buf.card_id, + buf.text, + buf.sequence + 1, ) - buf.last_edit = now + if ok: + buf.last_edit = now + else: + buf.sequence += 1 + await loop.run_in_executor( + None, + self._close_streaming_mode_sync, + buf.card_id, + buf.sequence, + ) + buf.card_id = None async def send(self, msg: OutboundMessage) -> None: """Send a message through Feishu, including media (images/files) if present.""" diff --git a/tests/channels/test_feishu_streaming.py b/tests/channels/test_feishu_streaming.py index 68232cb4..6810998b 100644 --- a/tests/channels/test_feishu_streaming.py +++ b/tests/channels/test_feishu_streaming.py @@ -113,6 +113,22 @@ class TestCloseStreamingMode: assert ch._close_streaming_mode_sync("card_1", 10) is False +class TestStreamUpdateWithReopen: + def test_reopens_streaming_mode_and_retries_update(self): + ch = _make_channel() + ch._client.cardkit.v1.card_element.content.side_effect = [ + _mock_content_response(False), + _mock_content_response(True), + ] + ch._client.cardkit.v1.card.settings.return_value = _mock_content_response(True) + + assert ch._stream_update_text_with_reopen_sync("card_1", "hello", 4) == (True, 6) + assert ch._client.cardkit.v1.card_element.content.call_count == 2 + settings_call = ch._client.cardkit.v1.card.settings.call_args[0][0] + assert settings_call.body.sequence == 5 + assert '"streaming_mode": true' in settings_call.body.settings + + class TestStreamUpdateText: def test_returns_true_on_success(self): ch = _make_channel() @@ -149,6 +165,24 @@ class TestSendDelta: ch._client.im.v1.message.create.assert_called_once() ch._client.cardkit.v1.card_element.content.assert_called_once() + @pytest.mark.asyncio + async def test_first_delta_closes_blank_card_when_initial_update_fails(self): + ch = _make_channel() + ch._client.cardkit.v1.card.create.return_value = _mock_create_card_response("card_new") + ch._client.im.v1.message.create.return_value = _mock_send_response("om_new") + ch._client.cardkit.v1.card_element.content.return_value = _mock_content_response(False) + ch._client.cardkit.v1.card.settings.return_value = _mock_content_response(True) + + await ch.send_delta("oc_chat1", "Hello ") + + buf = ch._stream_bufs["oc_chat1"] + assert buf.text == "Hello " + assert buf.card_id is None + assert ch._client.cardkit.v1.card_element.content.call_count == 2 + assert ch._client.cardkit.v1.card.settings.call_count == 2 + close_call = ch._client.cardkit.v1.card.settings.call_args_list[-1][0][0] + assert '"streaming_mode": false' in close_call.body.settings + @pytest.mark.asyncio async def test_group_delta_uses_create_when_reply_disabled(self): ch = _make_channel(reply_to_message=False) @@ -338,11 +372,29 @@ class TestSendDelta: await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True}) assert "oc_chat1" not in ch._stream_bufs - # Should NOT attempt to close streaming mode since update failed - ch._client.cardkit.v1.card.settings.assert_not_called() + assert ch._client.cardkit.v1.card.settings.call_count == 2 # Should fall back to sending a regular interactive card ch._client.im.v1.message.create.assert_called_once() + @pytest.mark.asyncio + async def test_stream_end_reopens_streaming_card_before_fallback(self): + ch = _make_channel() + ch._stream_bufs["oc_chat1"] = _FeishuStreamBuf( + text="Recovered content", card_id="card_1", sequence=3, last_edit=0.0, + ) + ch._client.cardkit.v1.card_element.content.side_effect = [ + _mock_content_response(False), + _mock_content_response(True), + ] + ch._client.cardkit.v1.card.settings.return_value = _mock_content_response(True) + + await ch.send_delta("oc_chat1", "", metadata={"_stream_end": True}) + + assert "oc_chat1" not in ch._stream_bufs + assert ch._client.cardkit.v1.card_element.content.call_count == 2 + assert ch._client.cardkit.v1.card.settings.call_count == 2 + ch._client.im.v1.message.create.assert_not_called() + @pytest.mark.asyncio async def test_stream_end_without_buf_is_noop(self): ch = _make_channel()