room.register_text_stream_handler(
        "agentChat", my_text_stream_handler(frame_buffer)
    )
    <http://logger.info|logger.info>("[tracking] Registered text stream handler for 'agentChat'")

    frame_processor_task = asyncio.create_task(
        frame_buffer_processor(frame_buffer, agent_identity, room.name),
        name=f"tracking-frame-buffer-{job_id}",
    )
    <http://logger.info|logger.info>("[tracking] Started frame_buffer_processor (decision logic + NATS publish)")


def my_text_stream_handler(frame_buffer):
    """
    Returns a handler for LiveKit text stream on the "agentChat" topic.
    Reassembles framed messages from OCI Vision and puts parsed JSON
    into the frame_buffer for downstream processing.
    """
    inflight_tasks: set[asyncio.Task] = set()

    def handler(text_reader, participant_identity: str):
        async def read_full_stream():
            global frame_insert_count, message_completed
            try:
                _prune_reassembly_buffers()
                logger.debug("Starting to read text stream for participant: %s", participant_identity)
                data = await text_reader.read_all()
                info = getattr(text_reader, "info", None)
                attrs = getattr(info, "attributes", None) if info else None
                meta_json = attrs.get("frame_meta") if attrs else None

                to_parse = None
                if meta_json:
                    try:
                        meta = json.loads(meta_json)
                        message_id: str = meta["message_id"]
                        frame_index: int = meta["frame_index"]
                        total_frames: int = meta["total_frames"]

                        buffers = message_buffers[message_id]
                        previous = buffers.get(frame_index)
                        buffers[frame_index] = data
                        message_totals[message_id] = total_frames
                        message_started_at.setdefault(message_id, datetime.now())
                        message_bytes[message_id] = (
                            message_bytes.get(message_id, 0)
                            - (_payload_size(previous) if previous is not None else 0)
                            + _payload_size(data)
                        )
                        _prune_reassembly_buffers()
                        if message_id not in message_buffers:
                            return

                        if len(buffers) == total_frames:
                            ordered = "".join(buffers[i] for i in range(total_frames))
                            message_completed += 1
                            message_buffers.pop(message_id, None)
                            message_totals.pop(message_id, None)
                            message_started_at.pop(message_id, None)
                            message_bytes.pop(message_id, None)
                            logger.debug("Completed reassembly of message_id: %s", message_id)
                            to_parse = ordered
                        else:
                            return
                    except Exception as e:
                        logger.warning("Failed framed reassembly; dropping message: %s", e)
                        return
                else:
                    return

                if isinstance(to_parse, (bytes, bytearray)):
                    to_parse = to_parse.decode("utf-8", "replace")

                if isinstance(to_parse, str):
                    try:
                        parsed_data = json.loads(to_parse)
                    except json.JSONDecodeError:
                        try:
                            parsed_data = ast.literal_eval(to_parse)
                        except (ValueError, SyntaxError) as e:
                            logger.warning("Parse error (non-JSON/non-literal); dropping: %s", e)
                            return
                else:
                    parsed_data = to_parse

                buffer_data = {"parsed_data": parsed_data}

                q_before = frame_buffer.qsize()
                if q_before >= FRAME_BUFFER_LIMIT or frame_buffer.full():
                    try:
                        frame_buffer.get_nowait()
                        frame_buffer.task_done()
                        logger.debug(
                            "Evicted oldest message (qsize_before=%d, limit=%d, MAX_FPS=%d)",
                            q_before, FRAME_BUFFER_LIMIT, MAX_FPS,
                        )
                    except asyncio.QueueEmpty:
                        logger.debug("Buffer eviction skipped; queue emptied by consumer")

                try:
                    frame_buffer.put_nowait(buffer_data)
                    frame_insert_count += 1
                    if handler.first_insert_time is None:
                        handler.first_insert_time = time.perf_counter()
                        <http://logger.info|logger.info>("First insert at: %s", datetime.now())
                except asyncio.QueueFull:
                    logger.debug("Queue full after eviction attempt; dropping newest message")

            except Exception as e:
                logger.error("Error processing text stream: %s", e)

        task = asyncio.create_task(read_full_stream())
        inflight_tasks.add(task)
        task.add_done_callback(inflight_tasks.discard)

    handler.first_insert_time = None
    return handler



def frame_buffer_processor(frame_buffer, agent_identity, room_name):
    """
    Returns a coroutine that consumes frames from frame_buffer and runs
    publish_track.process_frame (decision logic + NATS publish) for each.
    """
    async def processor():
        global frame_get_count

        while True:
            message = await frame_buffer.get()
            frame_get_count += 1

            # TEMP DIAGNOSTIC: log buffer sizes every 50 frames (~5s at 10 FPS).
            if frame_get_count % 50 == 0:
                logger.warning(
                    "[mem] frame_buffer.qsize=%d | message_buffers=%d totals=%d started=%d "
                    "bytes=%d completed=%d dropped=%d insert=%d get=%d",
                    frame_buffer.qsize(),
                    len(message_buffers), len(message_totals), len(message_started_at),
                    sum(message_bytes.values()),
                    message_completed, message_dropped, frame_insert_count, frame_get_count,
                )

            try:
                parsed_data = message["parsed_data"]
                await process_frame(parsed_data, agent_identity, room_name)
            except Exception as e:
                logger.error("Error processing frame: %s", e)
            finally:
                frame_buffer.task_done()

    return processor()