Merge pull request #68 from KunalLohtia/feat/langgraph-compress-tool-messages

OSS_PR_STRATEGY.md

@@ -15,7 +15,7 @@ Contribute to popular LangChain ecosystem repos to demonstrate Headroom's value
 - **What**: Core LangGraph framework
 - **PR**: Add `compress_tool_messages` pre-model hook example
 - **Issues it addresses**: #3717 (ToolMessage overflow), #11405 (agent token limit), #2140 (127K tokens from plugin)
-- **Status**: TODO
+- **Status**: DONE — `compress_tool_messages()` and `create_compress_tool_messages_node()` in `headroom/integrations/langchain/langgraph.py`
 
 ### Priority 3: `langchain-ai/deepagents` (~17K stars)
 - **What**: LangChain's coding agent (like Claude Code but OSS)

docs/langchain.md

@@ -340,6 +340,59 @@ print(f"Tokens saved: {llm.get_metrics()['tokens_saved']}")
 
 ---
 
+### Example 1b: LangGraph Custom Graph with compress_tool_messages Node
+
+If you're building a custom LangGraph `StateGraph` (instead of using `create_react_agent`),
+you can insert a compression node between tools and the agent. This compresses all
+`ToolMessage` content in the graph state before the LLM sees it.
+
+```python
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import HumanMessage
+from langgraph.graph import StateGraph, MessagesState, START, END
+from headroom.integrations.langchain import create_compress_tool_messages_node
+
+# Define your agent and tools nodes
+def agent_node(state: MessagesState):
+    llm = ChatOpenAI(model="gpt-4o")
+    response = llm.invoke(state["messages"])
+    return {"messages": [response]}
+
+def tools_node(state: MessagesState):
+    # Your tool execution logic here
+    ...
+
+# Build the graph with a compression step
+graph = StateGraph(MessagesState)
+graph.add_node("agent", agent_node)
+graph.add_node("tools", tools_node)
+graph.add_node("compress", create_compress_tool_messages_node(
+    min_tokens_to_compress=100,  # Only compress outputs > ~100 tokens
+))
+
+# Wire: tools -> compress -> agent (instead of tools -> agent directly)
+graph.add_edge(START, "agent")
+graph.add_edge("tools", "compress")
+graph.add_edge("compress", "agent")
+# ... add conditional edges from agent to tools/END as needed
+
+app = graph.compile()
+result = app.invoke({"messages": [HumanMessage(content="Find sales data")]})
+```
+
+You can also use `compress_tool_messages` directly as a standalone function:
+
+```python
+from headroom.integrations.langchain import compress_tool_messages
+
+# Compress ToolMessages in any list of LangChain messages
+result = compress_tool_messages(messages, min_tokens_to_compress=100)
+compressed_messages = result.messages
+print(f"Saved {result.total_tokens_saved} tokens across {result.messages_compressed} messages")
+```
+
+---
+
 ### Example 2: RAG Pipeline with Document Filtering
 
 ```python

headroom/integrations/langchain/__init__.py

@@ -7,6 +7,8 @@ This package provides seamless integration with LangChain, including:
 - HeadroomToolWrapper: Tool output compression for agents
 - StreamingMetricsTracker: Token counting during streaming
 - HeadroomLangSmithCallbackHandler: LangSmith trace enrichment
+- compress_tool_messages: LangGraph pre-model hook for ToolMessage compression
+- create_compress_tool_messages_node: LangGraph node factory
 
 Example:
     from langchain_openai import ChatOpenAI
@@ -21,7 +23,6 @@ Example:
 Install: pip install headroom[langchain]
 """
 
-# Core chat model wrapper
 # Agent tool wrapping
 from .agents import (
     HeadroomToolWrapper,
@@ -31,6 +32,8 @@ from .agents import (
     reset_tool_metrics,
     wrap_tools_with_headroom,
 )
+
+# Core chat model wrapper
 from .chat_model import (
     HeadroomCallbackHandler,
     HeadroomChatModel,
@@ -40,6 +43,15 @@ from .chat_model import (
     optimize_messages,
 )
 
+# LangGraph integration
+from .langgraph import (
+    CompressToolMessagesConfig,
+    CompressToolMessagesResult,
+    ToolMessageCompressionMetrics,
+    compress_tool_messages,
+    create_compress_tool_messages_node,
+)
+
 # LangSmith integration
 from .langsmith import (
     HeadroomLangSmithCallbackHandler,
@@ -93,6 +105,12 @@ __all__ = [
     "wrap_tools_with_headroom",
     "get_tool_metrics",
     "reset_tool_metrics",
+    # LangGraph
+    "compress_tool_messages",
+    "create_compress_tool_messages_node",
+    "CompressToolMessagesConfig",
+    "CompressToolMessagesResult",
+    "ToolMessageCompressionMetrics",
     # LangSmith
     "HeadroomLangSmithCallbackHandler",
     "is_langsmith_available",

headroom/integrations/langchain/langgraph.py

@@ -0,0 +1,399 @@
+"""LangGraph integration for Headroom tool message compression.
+
+This module provides a compress_tool_messages utility and a LangGraph-compatible
+node factory for compressing ToolMessage content before it reaches the LLM,
+solving context bloat from large tool outputs (JSON arrays, DB results, logs).
+
+Addresses:
+- LangGraph Issue #3717 (ToolMessage overflow)
+- LangChain Issue #11405 (agent token limit)
+- LangChain Issue #2140 (127K tokens from plugin)
+
+Example:
+    from langgraph.graph import StateGraph, MessagesState
+    from headroom.integrations.langchain.langgraph import (
+        compress_tool_messages,
+        create_compress_tool_messages_node,
+    )
+
+    # Option 1: Use as a LangGraph node
+    graph = StateGraph(MessagesState)
+    graph.add_node("agent", agent_node)
+    graph.add_node("tools", tool_node)
+    graph.add_node("compress", create_compress_tool_messages_node())
+    graph.add_edge("tools", "compress")
+    graph.add_edge("compress", "agent")
+
+    # Option 2: Use as a standalone function
+    compressed = compress_tool_messages(messages)
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+from uuid import uuid4
+
+# LangChain imports - optional dependencies
+try:
+    from langchain_core.messages import BaseMessage, ToolMessage
+
+    LANGCHAIN_AVAILABLE = True
+except ImportError:
+    LANGCHAIN_AVAILABLE = False
+    BaseMessage = object  # type: ignore[misc,assignment]
+    ToolMessage = object  # type: ignore[misc,assignment]
+
+from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig
+
+logger = logging.getLogger(__name__)
+
+
+def _check_langchain_available() -> None:
+    """Raise ImportError if LangChain is not installed."""
+    if not LANGCHAIN_AVAILABLE:
+        raise ImportError(
+            "LangChain is required for this integration. "
+            "Install with: pip install headroom[langchain] "
+            "or: pip install langchain-core"
+        )
+
+
+def _estimate_tokens(text: str) -> int:
+    """Estimate token count using ~4 characters per token heuristic."""
+    if not text:
+        return 0
+    return len(text) // 4
+
+
+@dataclass
+class ToolMessageCompressionMetrics:
+    """Metrics from compressing a single ToolMessage."""
+
+    request_id: str
+    timestamp: datetime
+    tool_call_id: str
+    tokens_before: int
+    tokens_after: int
+    tokens_saved: int
+    savings_percent: float
+    was_compressed: bool
+    skip_reason: str | None = None
+
+
+@dataclass
+class CompressToolMessagesConfig:
+    """Configuration for compress_tool_messages.
+
+    Attributes:
+        min_tokens_to_compress: Minimum estimated token count in a ToolMessage
+            before compression is applied. Default 100.
+        preserve_errors: If True, skip compression on ToolMessages whose content
+            contains error indicators. Default True.
+        error_indicators: Strings that indicate a ToolMessage contains an error.
+    """
+
+    min_tokens_to_compress: int = 100
+    preserve_errors: bool = True
+    error_indicators: tuple[str, ...] = ('"error"', '"ERROR"', "Error:", "Traceback")
+
+
+@dataclass
+class CompressToolMessagesResult:
+    """Result from compress_tool_messages including metrics."""
+
+    messages: list[Any]  # list[BaseMessage] but Any for when langchain not installed
+    metrics: list[ToolMessageCompressionMetrics] = field(default_factory=list)
+
+    @property
+    def total_tokens_saved(self) -> int:
+        """Total tokens saved across all compressed messages."""
+        return sum(m.tokens_saved for m in self.metrics if m.was_compressed)
+
+    @property
+    def messages_compressed(self) -> int:
+        """Number of messages that were actually compressed."""
+        return sum(1 for m in self.metrics if m.was_compressed)
+
+
+class _CrusherSingleton:
+    """Thread-safe lazy singleton for SmartCrusher."""
+
+    def __init__(self, min_tokens: int) -> None:
+        self._crusher: SmartCrusher | None = None
+        self._min_tokens = min_tokens
+        self._lock = threading.Lock()
+
+    def get(self) -> SmartCrusher:
+        if self._crusher is None:
+            with self._lock:
+                if self._crusher is None:
+                    config = SmartCrusherConfig(
+                        min_tokens_to_crush=self._min_tokens,
+                    )
+                    self._crusher = SmartCrusher(config=config)
+        return self._crusher
+
+
+# Module-level singleton, lazily initialized on first call
+_crusher_singleton: _CrusherSingleton | None = None
+_crusher_lock = threading.Lock()
+
+
+def _get_crusher(min_tokens: int) -> SmartCrusher:
+    """Get or create the module-level SmartCrusher singleton."""
+    global _crusher_singleton
+    if _crusher_singleton is None:
+        with _crusher_lock:
+            if _crusher_singleton is None:
+                _crusher_singleton = _CrusherSingleton(min_tokens)
+    return _crusher_singleton.get()
+
+
+def _should_skip(
+    content: str,
+    config: CompressToolMessagesConfig,
+) -> str | None:
+    """Check if a ToolMessage should skip compression.
+
+    Returns skip reason string, or None if it should be compressed.
+    """
+    if not content:
+        return "empty_content"
+
+    tokens = _estimate_tokens(content)
+    if tokens < config.min_tokens_to_compress:
+        return f"below_threshold:{tokens}<{config.min_tokens_to_compress}"
+
+    if config.preserve_errors:
+        for indicator in config.error_indicators:
+            if indicator in content:
+                return "error_content_preserved"
+
+    return None
+
+
+def compress_tool_messages(
+    messages: list[BaseMessage],  # type: ignore[type-arg]
+    *,
+    min_tokens_to_compress: int = 100,
+    preserve_errors: bool = True,
+    config: CompressToolMessagesConfig | None = None,
+) -> CompressToolMessagesResult:
+    """Compress ToolMessage content in a list of LangChain messages.
+
+    Iterates through messages, finds ToolMessages with large content,
+    and compresses them using SmartCrusher. Non-tool messages are
+    returned unchanged. tool_call_id is always preserved.
+
+    Args:
+        messages: List of LangChain BaseMessage objects.
+        min_tokens_to_compress: Minimum estimated tokens to trigger compression.
+        preserve_errors: If True, skip ToolMessages containing error indicators.
+        config: Full configuration object (overrides other kwargs if provided).
+
+    Returns:
+        CompressToolMessagesResult with compressed messages and metrics.
+
+    Example:
+        from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        messages = [
+            HumanMessage(content="Get sales data"),
+            AIMessage(content="", tool_calls=[{"id": "call_1", "name": "db", "args": {}}]),
+            ToolMessage(content='[{"row": 1}, {"row": 2}, ...]', tool_call_id="call_1"),
+        ]
+
+        result = compress_tool_messages(messages)
+        print(f"Saved {result.total_tokens_saved} tokens")
+        compressed_messages = result.messages
+    """
+    _check_langchain_available()
+
+    if config is None:
+        config = CompressToolMessagesConfig(
+            min_tokens_to_compress=min_tokens_to_compress,
+            preserve_errors=preserve_errors,
+        )
+
+    crusher = _get_crusher(config.min_tokens_to_compress)
+    result_messages: list[BaseMessage] = []
+    metrics: list[ToolMessageCompressionMetrics] = []
+
+    for msg in messages:
+        if not isinstance(msg, ToolMessage):
+            result_messages.append(msg)
+            continue
+
+        content = msg.content if isinstance(msg.content, str) else str(msg.content)
+        request_id = str(uuid4())
+
+        # Check if we should skip
+        skip_reason = _should_skip(content, config)
+        if skip_reason:
+            result_messages.append(msg)
+            tokens = _estimate_tokens(content)
+            metrics.append(
+                ToolMessageCompressionMetrics(
+                    request_id=request_id,
+                    timestamp=datetime.now(timezone.utc),
+                    tool_call_id=getattr(msg, "tool_call_id", "unknown"),
+                    tokens_before=tokens,
+                    tokens_after=tokens,
+                    tokens_saved=0,
+                    savings_percent=0.0,
+                    was_compressed=False,
+                    skip_reason=skip_reason,
+                )
+            )
+            logger.debug(
+                "Skipping ToolMessage %s compression: %s",
+                getattr(msg, "tool_call_id", "unknown"),
+                skip_reason,
+            )
+            continue
+
+        # Compress
+        tokens_before = _estimate_tokens(content)
+        try:
+            crush_result = crusher.crush(content=content, query="")
+            compressed_text = crush_result.compressed
+            was_modified = crush_result.was_modified
+        except Exception as e:
+            logger.warning(
+                "Compression failed for ToolMessage %s: %s. Keeping original.",
+                getattr(msg, "tool_call_id", "unknown"),
+                str(e),
+            )
+            result_messages.append(msg)
+            metrics.append(
+                ToolMessageCompressionMetrics(
+                    request_id=request_id,
+                    timestamp=datetime.now(timezone.utc),
+                    tool_call_id=getattr(msg, "tool_call_id", "unknown"),
+                    tokens_before=tokens_before,
+                    tokens_after=tokens_before,
+                    tokens_saved=0,
+                    savings_percent=0.0,
+                    was_compressed=False,
+                    skip_reason=f"compression_error:{type(e).__name__}",
+                )
+            )
+            continue
+
+        tokens_after = _estimate_tokens(compressed_text)
+
+        if was_modified and tokens_after < tokens_before:
+            # Create new ToolMessage with compressed content, preserving tool_call_id
+            compressed_msg = ToolMessage(
+                content=compressed_text,
+                tool_call_id=msg.tool_call_id,
+            )
+            result_messages.append(compressed_msg)
+            tokens_saved = tokens_before - tokens_after
+
+            metrics.append(
+                ToolMessageCompressionMetrics(
+                    request_id=request_id,
+                    timestamp=datetime.now(timezone.utc),
+                    tool_call_id=msg.tool_call_id,
+                    tokens_before=tokens_before,
+                    tokens_after=tokens_after,
+                    tokens_saved=tokens_saved,
+                    savings_percent=(tokens_saved / tokens_before * 100)
+                    if tokens_before > 0
+                    else 0.0,
+                    was_compressed=True,
+                )
+            )
+
+            logger.info(
+                "Compressed ToolMessage %s: %d -> %d tokens (%.1f%% saved)",
+                msg.tool_call_id,
+                tokens_before,
+                tokens_after,
+                (tokens_saved / tokens_before * 100) if tokens_before > 0 else 0,
+            )
+        else:
+            # Compression didn't help, keep original
+            result_messages.append(msg)
+            metrics.append(
+                ToolMessageCompressionMetrics(
+                    request_id=request_id,
+                    timestamp=datetime.now(timezone.utc),
+                    tool_call_id=msg.tool_call_id,
+                    tokens_before=tokens_before,
+                    tokens_after=tokens_before,
+                    tokens_saved=0,
+                    savings_percent=0.0,
+                    was_compressed=False,
+                    skip_reason="no_reduction",
+                )
+            )
+
+    return CompressToolMessagesResult(messages=result_messages, metrics=metrics)
+
+
+def create_compress_tool_messages_node(
+    *,
+    min_tokens_to_compress: int = 100,
+    preserve_errors: bool = True,
+    config: CompressToolMessagesConfig | None = None,
+) -> Any:
+    """Create a LangGraph node that compresses ToolMessages in graph state.
+
+    Returns a function compatible with LangGraph's StateGraph that reads
+    messages from state, compresses ToolMessages, and returns updated state.
+
+    Args:
+        min_tokens_to_compress: Minimum estimated tokens to trigger compression.
+        preserve_errors: If True, skip ToolMessages containing error indicators.
+        config: Full configuration object (overrides other kwargs if provided).
+
+    Returns:
+        A callable suitable for use as a LangGraph node.
+
+    Example:
+        from langgraph.graph import StateGraph, MessagesState
+
+        graph = StateGraph(MessagesState)
+        graph.add_node("agent", agent_node)
+        graph.add_node("tools", tool_node)
+        graph.add_node("compress", create_compress_tool_messages_node(
+            min_tokens_to_compress=200,
+        ))
+
+        # Wire: tools -> compress -> agent
+        graph.add_edge("tools", "compress")
+        graph.add_edge("compress", "agent")
+    """
+    _check_langchain_available()
+
+    if config is None:
+        config = CompressToolMessagesConfig(
+            min_tokens_to_compress=min_tokens_to_compress,
+            preserve_errors=preserve_errors,
+        )
+
+    def compress_node(state: dict[str, Any]) -> dict[str, Any]:
+        """LangGraph node that compresses ToolMessages in state.
+
+        Args:
+            state: LangGraph state dict containing a "messages" key.
+
+        Returns:
+            Updated state dict with compressed messages.
+        """
+        messages = state.get("messages", [])
+        if not messages:
+            return state
+
+        result = compress_tool_messages(messages, config=config)
+
+        return {"messages": result.messages}
+
+    return compress_node

tests/test_integrations/langchain/test_langgraph.py

@@ -0,0 +1,342 @@
+"""Tests for LangGraph tool message compression integration.
+
+Tests cover:
+1. compress_tool_messages - Compresses large ToolMessages in a message list
+2. create_compress_tool_messages_node - LangGraph node factory
+3. CompressToolMessagesConfig - Configuration options
+4. CompressToolMessagesResult - Result with metrics
+5. ToolMessageCompressionMetrics - Per-message metrics
+"""
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Check if LangChain is available
+try:
+    from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, ToolMessage
+
+    LANGCHAIN_AVAILABLE = True
+except ImportError:
+    LANGCHAIN_AVAILABLE = False
+
+# Skip all tests if LangChain not installed
+pytestmark = pytest.mark.skipif(not LANGCHAIN_AVAILABLE, reason="LangChain not installed")
+
+
+def _make_large_tool_output(num_items: int = 200) -> str:
+    """Generate a large JSON array string that will trigger compression."""
+    items = [{"id": i, "name": f"item_{i}", "value": i * 1.5, "status": "ok"} for i in range(num_items)]
+    return json.dumps(items)
+
+
+def _make_messages_with_tool_output(tool_content: str, tool_call_id: str = "call_1") -> list:
+    """Create a typical message sequence with a tool call and result."""
+    return [
+        HumanMessage(content="Get the data"),
+        AIMessage(content="", tool_calls=[{"id": tool_call_id, "name": "search", "args": {}}]),
+        ToolMessage(content=tool_content, tool_call_id=tool_call_id),
+    ]
+
+
+class TestCompressToolMessages:
+    """Tests for the compress_tool_messages function."""
+
+    def test_compresses_large_tool_message(self):
+        """Large ToolMessage content should be compressed."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        large_output = _make_large_tool_output(200)
+        messages = _make_messages_with_tool_output(large_output)
+
+        result = compress_tool_messages(messages)
+
+        # Should have same number of messages
+        assert len(result.messages) == 3
+        # ToolMessage should be smaller
+        compressed_content = result.messages[2].content
+        assert len(compressed_content) < len(large_output)
+
+    def test_preserves_small_tool_messages(self):
+        """Small ToolMessages should not be compressed."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        small_output = '{"result": "ok"}'
+        messages = _make_messages_with_tool_output(small_output)
+
+        result = compress_tool_messages(messages)
+
+        # Content should be unchanged
+        assert result.messages[2].content == small_output
+        assert result.messages_compressed == 0
+
+    def test_preserves_non_tool_messages(self):
+        """HumanMessage and AIMessage should pass through unchanged."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        large_output = _make_large_tool_output(200)
+        messages = _make_messages_with_tool_output(large_output)
+
+        result = compress_tool_messages(messages)
+
+        assert isinstance(result.messages[0], HumanMessage)
+        assert result.messages[0].content == "Get the data"
+        assert isinstance(result.messages[1], AIMessage)
+        tool_call = result.messages[1].tool_calls[0]
+        assert tool_call["id"] == "call_1"
+        assert tool_call["name"] == "search"
+        assert tool_call["args"] == {}
+
+    def test_preserves_tool_call_id(self):
+        """Compressed ToolMessages must keep their tool_call_id."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        large_output = _make_large_tool_output(200)
+        messages = _make_messages_with_tool_output(large_output, tool_call_id="call_abc123")
+
+        result = compress_tool_messages(messages)
+
+        tool_msg = result.messages[2]
+        assert isinstance(tool_msg, ToolMessage)
+        assert tool_msg.tool_call_id == "call_abc123"
+
+    def test_preserves_error_content_by_default(self):
+        """ToolMessages with error indicators should be skipped by default."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        # Large content but contains error indicator
+        error_output = json.dumps({
+            "error": "Database connection failed",
+            "details": "x" * 2000,
+        })
+        messages = _make_messages_with_tool_output(error_output)
+
+        result = compress_tool_messages(messages)
+
+        # Should be unchanged — error preserved
+        assert result.messages[2].content == error_output
+        assert result.metrics[0].skip_reason == "error_content_preserved"
+
+    def test_compresses_error_content_when_disabled(self):
+        """Error content should be compressed when preserve_errors=False."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        error_output = json.dumps({
+            "error": "fail",
+            "data": [{"id": i} for i in range(200)],
+        })
+        messages = _make_messages_with_tool_output(error_output)
+
+        result = compress_tool_messages(messages, preserve_errors=False)
+
+        # Should have attempted compression (no error_content_preserved skip)
+        assert result.metrics[0].skip_reason != "error_content_preserved"
+
+    def test_handles_empty_messages(self):
+        """Empty message list should return empty result."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        result = compress_tool_messages([])
+
+        assert result.messages == []
+        assert result.metrics == []
+        assert result.total_tokens_saved == 0
+
+    def test_handles_no_tool_messages(self):
+        """Message list with no ToolMessages should pass through."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        messages = [
+            HumanMessage(content="Hello"),
+            AIMessage(content="Hi there!"),
+        ]
+
+        result = compress_tool_messages(messages)
+
+        assert len(result.messages) == 2
+        assert result.messages[0].content == "Hello"
+        assert result.messages[1].content == "Hi there!"
+        assert result.metrics == []
+
+    def test_multiple_tool_messages(self):
+        """Should compress multiple ToolMessages independently."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        large_output_1 = _make_large_tool_output(200)
+        large_output_2 = _make_large_tool_output(150)
+
+        messages = [
+            HumanMessage(content="Get all data"),
+            AIMessage(
+                content="",
+                tool_calls=[
+                    {"id": "call_1", "name": "search", "args": {}},
+                    {"id": "call_2", "name": "database", "args": {}},
+                ],
+            ),
+            ToolMessage(content=large_output_1, tool_call_id="call_1"),
+            ToolMessage(content=large_output_2, tool_call_id="call_2"),
+        ]
+
+        result = compress_tool_messages(messages)
+
+        assert len(result.messages) == 4
+        # Both tool messages should have their correct tool_call_ids
+        assert result.messages[2].tool_call_id == "call_1"
+        assert result.messages[3].tool_call_id == "call_2"
+
+    def test_min_tokens_to_compress_config(self):
+        """Custom min_tokens_to_compress should be respected."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        # Content that's ~100 tokens (400 chars) — below a 200 token threshold
+        medium_output = json.dumps({"data": "x" * 400})
+        messages = _make_messages_with_tool_output(medium_output)
+
+        result = compress_tool_messages(messages, min_tokens_to_compress=200)
+
+        # Should be skipped due to being below threshold
+        assert result.metrics[0].was_compressed is False
+        assert "below_threshold" in (result.metrics[0].skip_reason or "")
+
+
+class TestCompressToolMessagesResult:
+    """Tests for CompressToolMessagesResult properties."""
+
+    def test_total_tokens_saved(self):
+        """total_tokens_saved should sum across compressed metrics."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        large_output = _make_large_tool_output(200)
+        messages = _make_messages_with_tool_output(large_output)
+
+        result = compress_tool_messages(messages)
+
+        assert result.total_tokens_saved >= 0
+        # If compression happened, tokens_saved should be positive
+        if result.messages_compressed > 0:
+            assert result.total_tokens_saved > 0
+
+    def test_messages_compressed_count(self):
+        """messages_compressed should count actually compressed messages."""
+        from headroom.integrations.langchain.langgraph import compress_tool_messages
+
+        messages = [
+            HumanMessage(content="test"),
+            ToolMessage(content='{"small": true}', tool_call_id="call_1"),
+        ]
+
+        result = compress_tool_messages(messages)
+
+        assert result.messages_compressed == 0
+
+
+class TestCompressToolMessagesConfig:
+    """Tests for CompressToolMessagesConfig."""
+
+    def test_config_object(self):
+        """Config object should override kwargs."""
+        from headroom.integrations.langchain.langgraph import (
+            CompressToolMessagesConfig,
+            compress_tool_messages,
+        )
+
+        config = CompressToolMessagesConfig(
+            min_tokens_to_compress=500,
+            preserve_errors=False,
+        )
+
+        medium_output = json.dumps({"data": "x" * 800})
+        messages = _make_messages_with_tool_output(medium_output)
+
+        result = compress_tool_messages(messages, config=config)
+
+        # ~200 tokens, below the 500 threshold
+        assert result.metrics[0].was_compressed is False
+
+    def test_default_config(self):
+        """Default config should have sensible defaults."""
+        from headroom.integrations.langchain.langgraph import CompressToolMessagesConfig
+
+        config = CompressToolMessagesConfig()
+        assert config.min_tokens_to_compress == 100
+        assert config.preserve_errors is True
+
+
+class TestCreateCompressToolMessagesNode:
+    """Tests for the LangGraph node factory."""
+
+    def test_returns_callable(self):
+        """Factory should return a callable node function."""
+        from headroom.integrations.langchain.langgraph import create_compress_tool_messages_node
+
+        node = create_compress_tool_messages_node()
+        assert callable(node)
+
+    def test_node_reads_messages_from_state(self):
+        """Node should read messages from state dict and return updated state."""
+        from headroom.integrations.langchain.langgraph import create_compress_tool_messages_node
+
+        large_output = _make_large_tool_output(200)
+        state = {
+            "messages": _make_messages_with_tool_output(large_output),
+        }
+
+        node = create_compress_tool_messages_node()
+        result_state = node(state)
+
+        assert "messages" in result_state
+        assert len(result_state["messages"]) == 3
+        # ToolMessage should be compressed
+        assert len(result_state["messages"][2].content) < len(large_output)
+
+    def test_node_preserves_tool_call_id(self):
+        """Node should preserve tool_call_id on compressed messages."""
+        from headroom.integrations.langchain.langgraph import create_compress_tool_messages_node
+
+        large_output = _make_large_tool_output(200)
+        state = {
+            "messages": [
+                HumanMessage(content="test"),
+                AIMessage(content="", tool_calls=[{"id": "call_xyz", "name": "db", "args": {}}]),
+                ToolMessage(content=large_output, tool_call_id="call_xyz"),
+            ],
+        }
+
+        node = create_compress_tool_messages_node()
+        result_state = node(state)
+
+        assert result_state["messages"][2].tool_call_id == "call_xyz"
+
+    def test_node_handles_empty_state(self):
+        """Node should handle empty messages gracefully."""
+        from headroom.integrations.langchain.langgraph import create_compress_tool_messages_node
+
+        node = create_compress_tool_messages_node()
+        result_state = node({"messages": []})
+
+        assert result_state == {"messages": []}
+
+    def test_node_handles_missing_messages_key(self):
+        """Node should handle state without messages key."""
+        from headroom.integrations.langchain.langgraph import create_compress_tool_messages_node
+
+        node = create_compress_tool_messages_node()
+        result_state = node({})
+
+        assert "messages" not in result_state or result_state.get("messages") == []
+
+    def test_node_with_custom_config(self):
+        """Node should respect custom configuration."""
+        from headroom.integrations.langchain.langgraph import create_compress_tool_messages_node
+
+        node = create_compress_tool_messages_node(min_tokens_to_compress=10000)
+
+        large_output = _make_large_tool_output(200)
+        state = {"messages": _make_messages_with_tool_output(large_output)}
+
+        result_state = node(state)
+
+        # With very high threshold, nothing should be compressed
+        assert result_state["messages"][2].content == large_output