async def _generate_thread_entities(
        self, client: httpx.AsyncClient, processed_message_ids: set
    ) -> AsyncGenerator[ChunkEntity, None]:
        """Generate GmailThreadEntity objects and associated message entities."""
        logger.info("Starting thread entity generation")
        base_url = "https://gmail.googleapis.com/gmail/v1/users/me/threads"
        params = {"maxResults": 100}

        page_count = 0
        thread_count = 0

        while True:
            page_count += 1
            logger.info(f"Fetching thread list page #{page_count} with params: {params}")
            data = await self._get_with_auth(client, base_url, params=params)
            threads = data.get("threads", [])
            logger.info(f"Found {len(threads)} threads on page {page_count}")

            for thread_idx, thread_info in enumerate(threads):
                thread_count += 1
                thread_id = thread_info["id"]
                logger.info(f"Processing thread #{thread_idx + 1}/{len(threads)} (ID: {thread_id})")

                # Fetch full thread detail
                detail_url = f"{base_url}/{thread_id}"
                logger.info(f"Fetching full thread details from: {detail_url}")
                thread_data = await self._get_with_auth(client, detail_url)

                # Collect thread information
                snippet = thread_data.get("snippet", "")
                history_id = thread_data.get("historyId")
                message_list = thread_data.get("messages", [])

                logger.info(f"Thread {thread_id} contains {len(message_list)} messages")
                logger.debug(f"Thread snippet: {snippet[:50]}...")
                logger.debug(f"Thread history ID: {history_id}")

                # Calculate message count
                message_count = len(message_list)

                # Find last message date
                last_message_date = None
                if message_list:
                    logger.debug("Sorting messages by date to find most recent")
                    sorted_msgs = sorted(
                        message_list, key=lambda m: int(m.get("internalDate", 0)), reverse=True
                    )
                    last_message_date_ms = sorted_msgs[0].get("internalDate")
                    if last_message_date_ms:
                        last_message_date = datetime.utcfromtimestamp(
                            int(last_message_date_ms) / 1000
                        )
                        logger.debug(f"Last message date: {last_message_date}")

                # Get label IDs from first message if available
                label_ids = []
                if message_list:
                    label_ids = message_list[0].get("labelIds", [])
                    logger.debug(f"Thread labels: {label_ids}")

                # Create thread entity
                logger.info(f"Creating thread entity for thread ID: {thread_id}")
                thread_entity = GmailThreadEntity(
                    entity_id=f"thread_{thread_id}",  # Prefix to ensure uniqueness
                    breadcrumbs=[],  # Thread is top-level
                    snippet=snippet,
                    history_id=history_id,
                    message_count=message_count,
                    label_ids=label_ids,
                    last_message_date=last_message_date,
                )
                logger.debug(f"Thread entity created: {thread_entity.dict()}")
                yield thread_entity
                logger.info(f"Thread entity yielded: {thread_id}")

                # Create thread breadcrumb for messages
                thread_breadcrumb = Breadcrumb(
                    entity_id=f"thread_{thread_id}",  # Match the thread entity's ID
                    name=snippet[:50] + "..." if len(snippet) > 50 else snippet,
                    type="thread",
                )
                logger.debug(f"Created thread breadcrumb: {thread_breadcrumb}")

                # Process each message in the thread
                logger.info(f"Processing {len(message_list)} messages in thread {thread_id}")
                for msg_idx, message_data in enumerate(message_list):
                    msg_id = message_data.get("id", "unknown")

                    # Skip if we've already processed this message in another thread
                    if msg_id in processed_message_ids:
                        logger.info(
                            f"Skipping message {msg_id} in thread {thread_id} - already processed"
                        )
                        continue

                    msg_info = (
                        f"Processing message #{msg_idx + 1}/{len(message_list)} "
                        f"(ID: {msg_id}) in thread {thread_id}"
                    )
                    logger.info(msg_info)

                    # Mark this message as processed
                    processed_message_ids.add(msg_id)

                    msg_entity_count = 0
                    async for entity in self._process_message(
                        client, message_data, thread_id, thread_breadcrumb
                    ):
                        msg_entity_count += 1
                        entity_type = type(entity).__name__
                        logger.info(f"Yielding {entity_type} entity from message {msg_id}")
                        yield entity

                    logger.info(f"Yielded {msg_entity_count} entities for message {msg_id}")

            # Handle pagination
            next_page_token = data.get("nextPageToken")
            if not next_page_token:
                msg = (
                    f"No more pages to fetch. Processed {thread_count} threads "
                    f"in {page_count} pages."
                )
                logger.info(msg)
                break

            logger.info(f"Found next page token: {next_page_token[:10]}...")
            params["pageToken"] = next_page_token
Generate GmailThreadEntity objects and associated message entities.
backend/airweave/platform/sources/gmail.py