"""Conversation-shaping helpers: token counting + tool-aware truncation. Moved out of src/utils/clients.py as part of the migration into src/llm/. These are pure helpers with no orchestration dependencies. """ from __future__ import annotations import json import logging from typing import Any, cast from src.utils.tokens import estimate_tokens logger = logging.getLogger(__name__) def count_message_tokens(messages: list[dict[str, Any]]) -> int: """Count tokens in a list of messages using tiktoken.""" total = 0 for msg in messages: content = msg.get("content", "") if isinstance(content, str): total += estimate_tokens(content) elif isinstance(content, list): # Anthropic-style content blocks total += estimate_tokens(json.dumps(content)) if "parts" in msg: try: total += estimate_tokens(json.dumps(msg["parts"])) except TypeError: # Non-JSON-serializable content (e.g. bytes) — estimate from repr. total += estimate_tokens(str(msg["parts"])) return total def _is_tool_use_message(msg: dict[str, Any]) -> bool: """Check if a message contains tool calls (any format). Recognizes: - Anthropic: ``content`` is a list containing a ``{"type": "tool_use"}`` block. - Gemini: ``parts`` is a list containing a ``{"function_call": …}`` entry. - OpenAI: assistant message with a non-empty ``tool_calls`` field. """ content = msg.get("content") if isinstance(content, list): for block in cast(list[dict[str, Any]], content): if block.get("type") == "tool_use": return True parts = msg.get("parts") if isinstance(parts, list): for part in cast(list[dict[str, Any]], parts): if "function_call" in part: return True return bool(msg.get("tool_calls")) def _is_tool_result_message(msg: dict[str, Any]) -> bool: """Check if a message contains tool results (any format). Recognizes: - Anthropic: ``content`` is a list containing a ``{"type": "tool_result"}`` block. - Gemini: ``parts`` is a list containing a ``{"function_response": …}`` entry. - OpenAI: message with ``role == "tool"``. """ content = msg.get("content") if isinstance(content, list): for block in cast(list[dict[str, Any]], content): if block.get("type") == "tool_result": return True parts = msg.get("parts") if isinstance(parts, list): for part in cast(list[dict[str, Any]], parts): if "function_response" in part: return True return msg.get("role") == "tool" def _group_into_units( messages: list[dict[str, Any]], ) -> list[list[dict[str, Any]]]: """Group messages into logical conversation units. A unit is either: - A tool_use message + ALL consecutive tool_result messages that follow - A single non-tool message Keeps tool_use / tool_result pairs together so truncation never breaks them apart. """ units: list[list[dict[str, Any]]] = [] i = 0 while i < len(messages): msg = messages[i] if _is_tool_use_message(msg): j = i + 1 while j < len(messages) and _is_tool_result_message(messages[j]): j += 1 unit = messages[i:j] if len(unit) > 1: units.append(unit) i = j else: # Orphaned tool_use with no results — skip it. logger.debug(f"Skipping orphaned tool_use at index {i}") i += 1 elif _is_tool_result_message(msg): # Orphaned tool_result — skip it. logger.debug(f"Skipping orphaned tool_result at index {i}") i += 1 else: units.append([msg]) i += 1 return units def truncate_messages_to_fit( messages: list[dict[str, Any]], max_tokens: int, preserve_system: bool = True, ) -> list[dict[str, Any]]: """Truncate messages to fit within a token limit while maintaining valid structure. Strategy: 1. Group messages into units (tool_use + results together, or single messages) 2. Remove oldest units first to preserve recent context 3. Units stay intact so tool_use/tool_result pairs are never broken """ current_tokens = count_message_tokens(messages) if current_tokens <= max_tokens: return messages logger.info(f"Truncating: {current_tokens} tokens exceeds {max_tokens} limit") system_messages: list[dict[str, Any]] = [] conversation: list[dict[str, Any]] = [] for msg in messages: if msg.get("role") == "system" and preserve_system: system_messages.append(msg) else: conversation.append(msg) system_tokens = count_message_tokens(system_messages) available_tokens = max_tokens - system_tokens if available_tokens <= 0: logger.warning("System message exceeds max_input_tokens") return messages units = _group_into_units(conversation) if not units: logger.warning("No valid conversation units") return system_messages # Drop oldest units until conversation fits, but keep at least one unit so # we never erase the entire non-system conversation. while len(units) > 1: flat_messages = [m for unit in units for m in unit] if count_message_tokens(flat_messages) <= available_tokens: break removed_unit = units.pop(0) logger.debug( "Dropping conversation unit with " + f"{len(removed_unit)} messages " + f"(~{count_message_tokens(removed_unit)} tokens)" ) result = system_messages + [m for unit in units for m in unit] result_tokens = count_message_tokens(result) logger.info( f"Truncation complete: {current_tokens} → {result_tokens} tokens " + f"({len(messages)} → {len(result)} messages)" ) return result __all__ = [ "count_message_tokens", "truncate_messages_to_fit", ]