MessageManager class interface - handles persistence, injections, and context engineering.

class MessageManager:
    """Conversation state management with smart loading and auto-persistence"""

    def __init__(self, session_id, user_id, storage, injections=None):
        self.session_id = session_id
        self.user_id = user_id
        self.storage = storage  # HybridStorage (Redis + PostgreSQL)
        self.injections = injections or []
        self._messages = []  # In-memory working set

    async def load(self, token_budget=180000):
        """Load messages within token budget

        Order: injections → load from storage → trim to budget
        """

    async def add_user_message(self, content):
        """Add user message, estimate tokens, auto-save to storage"""

    async def add_assistant_message(self, content_blocks, usage):
        """Add assistant message, use accurate tokens from API, auto-save"""

    def to_api_format(self, keep_full_detail_for_last_n=30):
        """Convert to API format, strip metadata, apply context engineering

        Old interactions: Strip tool_use/tool_result/thinking blocks
        Recent interactions: Keep full detail
        """

    def get_for_display(self):
        """Filter for frontend - show user msgs + assistant text only"""

    async def save_loop_state(self, step, awaiting_tool, tool_use_id):
        """Save loop state for interactive tool pause"""

    async def load_loop_state(self):
        """Load saved state for resume"""

    async def clear_loop_state(self):
        """Clear state after resumption"""

    # Context manager support
    async def __aenter__(self): return self
    async def __aexit__(self, *args): pass

Token tracking: User messages use estimation (1 token ≈ 4 chars), assistant messages use accurate count from API response. Both auto-save to storage.

async def add_user_message(self, content):
    """Add user message with estimated tokens, auto-save"""
    # Estimate: 1 token ≈ 4 chars (acceptable for budget queries)
    estimated_tokens = len(content) // 4

    msg = {
        "sequence": self._get_next_sequence(),
        "role": "user",
        "content": content,  # str or list (tool results)
        "tokens": estimated_tokens,
        "tokens_estimated": True,
        "created_at": datetime.now().isoformat()
    }

    self._messages.append(msg)
    await self.storage.save_message(self.session_id, self.user_id, msg)


async def add_assistant_message(self, content_blocks, usage):
    """Add assistant message with accurate tokens from API, auto-save"""
    # Use ACCURATE tokens from API response
    accurate_tokens = usage["output_tokens"]

    # Handle Pydantic models from Anthropic SDK
    if content_blocks and hasattr(content_blocks[0], 'model_dump'):
        content_blocks = [block.model_dump() for block in content_blocks]

    msg = {
        "sequence": self._get_next_sequence(),
        "role": "assistant",
        "content": content_blocks,  # Keep thinking blocks
        "tokens": accurate_tokens,
        "tokens_estimated": False,
        "created_at": datetime.now().isoformat()
    }

    self._messages.append(msg)
    await self.storage.save_message(self.session_id, self.user_id, msg)


def _get_next_sequence(self):
    """Get next sequence number"""
    if not self._messages:
        return 1
    return self._messages[-1]["sequence"] + 1

Why this works: Estimation for user input is fast and accurate enough (±10-20% error acceptable). API provides exact tokens for assistant output. Avoids complex reconciliation.

Three key operations: load from storage with budget, transform to API format with context engineering, filter for frontend display.

async def load(self, token_budget=180000):
    """Load messages within token budget

    Order: Execute injections → Load from storage → Trim to budget
    """
    # Step 1: Apply injections (system context)
    for inject_fn in self.injections:
        injected_msg = await inject_fn()
        self._messages.append(injected_msg)

    # Step 2: Load from storage (Redis first, PostgreSQL fallback)
    history = await self.storage.load_recent_messages(
        session_id=self.session_id,
        limit=100
    )

    # Step 3: Trim to token budget
    total_tokens = sum(m["tokens"] for m in self._messages)
    for msg in history:
        if total_tokens + msg["tokens"] > token_budget:
            break
        self._messages.append(msg)
        total_tokens += msg["tokens"]

    return self


def to_api_format(self, keep_full_detail_for_last_n=30):
    """Transform to API format with progressive detail reduction

    Context engineering:
    - Old interactions: Strip tool_use/tool_result/thinking (keep text only)
    - Recent interactions: Keep full detail for reasoning continuity
    """
    # Group into interactions (user input → assistant → tool results)
    interactions = self._group_into_interactions()

    if len(interactions) <= keep_full_detail_for_last_n:
        # Few interactions - keep everything
        return self._messages_to_api_format(self._messages)

    # Split: old interactions vs recent
    old_count = len(interactions) - keep_full_detail_for_last_n
    old = interactions[:old_count]
    recent = interactions[old_count:]

    # Strip old interactions (remove tool blocks)
    stripped = []
    for interaction in old:
        for msg in interaction:
            # Keep only text blocks from assistant messages
            # Remove tool_result user messages entirely
            stripped_msg = self._strip_tool_blocks(msg)
            if stripped_msg:
                stripped.append(stripped_msg)

    # Combine: stripped old + full recent
    all_messages = stripped + [msg for interaction in recent for msg in interaction]

    # Convert to API format (strip metadata, clean extra fields)
    return self._messages_to_api_format(all_messages)


def _messages_to_api_format(self, messages):
    """Strip metadata, remove extra fields from tool_result blocks"""
    api_messages = []
    for msg in messages:
        api_msg = {"role": msg["role"], "content": msg["content"]}

        # Clean tool_result blocks (remove display_data, artifact_data, etc.)
        if isinstance(api_msg["content"], list):
            cleaned_content = []
            for block in api_msg["content"]:
                if block.get("type") == "tool_result":
                    # Only allowed fields: type, tool_use_id, content, is_error
                    cleaned_block = {
                        "type": "tool_result",
                        "tool_use_id": block["tool_use_id"],
                        "content": block["content"]
                    }
                    if "is_error" in block:
                        cleaned_block["is_error"] = block["is_error"]
                    cleaned_content.append(cleaned_block)
                else:
                    cleaned_content.append(block)
            api_msg["content"] = cleaned_content

        api_messages.append(api_msg)
    return api_messages

Key insight: Tool blocks (tool_use, tool_result) consume tokens but add little value after the first few interactions. Stripping them from old messages saves 50-70% tokens while preserving conversation flow.