fix(vertexai): record async stream text after the stream ends - #4542

import asyncio
from types import SimpleNamespace
from unittest.mock import patch
from opentelemetry.instrumentation.vertexai import _abuild_from_streaming_response
from opentelemetry.semconv._incubating.attributes import gen_ai_attributes as GenAIAttributes

class RecordingSpan:
    def __init__(self):
        self.attributes = {}
        self.statuses = []
        self.ended = False
    def is_recording(self):
        return True
    def set_attribute(self, name, value):
        self.attributes[name] = value
    def set_status(self, status):
        self.statuses.append(status)
    def end(self):
        self.ended = True

def _chunk(text, usage_metadata=None):
    return SimpleNamespace(text=text, usage_metadata=usage_metadata)

async def _stream_chunks():
    usage_metadata = SimpleNamespace(total_token_count=7, candidates_token_count=4, prompt_token_count=3, cached_content_token_count=None)
    yield _chunk('hello ')
    yield _chunk('world', usage_metadata=usage_metadata)

async def main():
    span = RecordingSpan()
    with patch('opentelemetry.instrumentation.vertexai.should_emit_events', return_value=False), patch('opentelemetry.instrumentation.vertexai.span_utils.should_send_prompts', return_value=True):
        response = _abuild_from_streaming_response(span=span, event_logger=None, response=_stream_chunks(), llm_model='gemini-2.5-pro')
        observed = [chunk.text async for chunk in response]
    print('observed', observed)
    print('completion_attr_type', type(span.attributes[f"{GenAIAttributes.GEN_AI_COMPLETION}.0.content"]).__name__)
    print('completion_attr_repr', repr(span.attributes[f"{GenAIAttributes.GEN_AI_COMPLETION}.0.content"]))
    print('ended', span.ended)

asyncio.run(main())