diff --git a/packages/apps/src/microsoft_teams/apps/http_stream.py b/packages/apps/src/microsoft_teams/apps/http_stream.py index cf52b321..f33588d3 100644 --- a/packages/apps/src/microsoft_teams/apps/http_stream.py +++ b/packages/apps/src/microsoft_teams/apps/http_stream.py @@ -399,6 +399,9 @@ async def _send(self, to_send: Union[TypingActivityInput, MessageActivityInput]) else: res = await self._client.conversations.create_activity(self._ref.conversation.id, to_send) + if to_send.id: + res = res.model_copy(update={"id": to_send.id}) + return SentActivity.merge(to_send, res) except HTTPStatusError as e: # Various error codes are used for streaming. diff --git a/packages/apps/tests/test_http_stream.py b/packages/apps/tests/test_http_stream.py index 892f6ad4..dd8a93d3 100644 --- a/packages/apps/tests/test_http_stream.py +++ b/packages/apps/tests/test_http_stream.py @@ -50,6 +50,14 @@ def mock_api_client(self): async def mock_send(conversation_id, activity): client.send_call_count += 1 client.sent_activities.append(activity) + + stream_entities = [ + entity for entity in activity.entities or [] if entity.type == "streaminfo" and entity.stream_id + ] + stream_id = next((entity.stream_id for entity in stream_entities), None) + if stream_id: + return SentActivity(id="DO_NOT_USE_PLACEHOLDER_ID", activity_params=activity) + return SentActivity(id=f"activity-{client.send_call_count}", activity_params=activity) client.conversations.create_activity = mock_send @@ -504,6 +512,45 @@ async def test_close_returns_none_when_canceled(self, mock_api_client, conversat result = await stream.close() assert result is None + @pytest.mark.asyncio + async def test_close_returns_stream_id_when_final_response_has_placeholder( + self, mock_api_client, conversation_reference, patch_loop_call_later + ): + """The final streaming response is empty, so close() must retain the first chunk's id.""" + loop = asyncio.get_running_loop() + patcher, scheduled = patch_loop_call_later(loop) + close_activities: list[SentActivity] = [] + close_event = asyncio.Event() + + async def handle_close(activity: SentActivity) -> None: + close_activities.append(activity) + close_event.set() + + async def mock_create(conversation_id, activity): + if any( + entity.type == "streaminfo" and entity.stream_type == "final" for entity in (activity.entities or []) + ): + return SentActivity(id="DO_NOT_USE_PLACEHOLDER_ID", activity_params=activity) + return SentActivity(id="stream-1", activity_params=activity) + + mock_api_client.conversations.create_activity = mock_create + + with patcher: + stream = HttpStream(mock_api_client, conversation_reference) + stream.on_close(handle_close) + + stream.emit("Streamed content") + await asyncio.sleep(0) + await self._run_scheduled_flushes(scheduled) + + result = await stream.close() + + assert result is not None + assert result.id == "stream-1" + await asyncio.wait_for(close_event.wait(), timeout=1) + assert close_activities == [result] + assert close_activities[0].id == "stream-1" + @pytest.mark.asyncio async def test_final_activity_last_wins(self, mock_api_client, conversation_reference, patch_loop_call_later): """When multiple MessageActivityInputs are emitted, the last one's non-text fields are used.""" @@ -780,6 +827,7 @@ async def test_close_waits_for_flush_to_complete(self, mock_api_client, conversa result = await close_task assert result is not None + assert result.id == "activity-1" assert mock_api_client.send_call_count == 1 assert mock_api_client.sent_activities[0].text == "Response text"