Remove hardcoded source hint system from ContentRouter

ContentRouter now routes purely based on content analysis instead of
relying on hardcoded tool name mappings. This makes the router work
with any MCP tool regardless of naming convention.

Changes:
- Remove generate_source_hint() function and _strategy_from_hint() method
- Remove source_hint parameter from compress() method
- Remove _get_tool_source_hint() from IntelligentContextManager
- Update tests to remove source hint test cases
- Update docs to document content detection approach

docs/transforms.md

@@ -516,15 +516,14 @@ router = ContentRouter(config)
 ### Example
 
 ```python
-from headroom.transforms import ContentRouter, generate_source_hint
+from headroom.transforms import ContentRouter
 
 router = ContentRouter()
 
-# With source hint for high-confidence routing
-hint = generate_source_hint(tool_name="grep", file_path="src/auth.py")
-result = router.compress(content, source_hint=hint)
+# Router auto-detects content type and routes to optimal compressor
+result = router.compress(content)
 
-print(result.strategy)  # CompressionStrategy.SEARCH or CODE_AWARE
+print(result.strategy_used)  # CompressionStrategy.CODE_AWARE, SMART_CRUSHER, etc.
 print(result.routing_log)  # List of routing decisions
 ```
 
@@ -540,22 +539,17 @@ print(result.routing_log)  # List of routing decisions
 | LLMLINGUA | Any (max compression) | LLMLinguaCompressor |
 | PASSTHROUGH | Small content | None |
 
-### Source Hints
+### Content Detection
 
-Use source hints for accurate routing:
+The router automatically detects content types by analyzing the content itself:
 
-```python
-from headroom.transforms import generate_source_hint
-
-# From tool invocation
-hint = generate_source_hint(tool_name="Read", file_path="main.py")
+- **Source code**: Detected by syntax patterns, indentation, keywords
+- **JSON arrays**: Detected by JSON structure with array elements
+- **Search results**: Detected by `file:line:` patterns
+- **Log output**: Detected by timestamp and log level patterns
+- **Plain text**: Fallback for prose content
 
-# From file extension
-hint = generate_source_hint(file_path="components/Button.tsx")
-
-# From explicit tool
-hint = generate_source_hint(tool_name="Grep")  # Routes to SEARCH
-```
+No manual hints required - the router inspects content directly.
 
 ---
 

headroom/transforms/__init__.py

@@ -57,7 +57,6 @@ from .content_router import (
     ContentRouter,
     ContentRouterConfig,
     RouterCompressionResult,
-    generate_source_hint,
 )
 
 __all__ = [
@@ -101,7 +100,6 @@ __all__ = [
     "ContentRouterConfig",
     "RouterCompressionResult",
     "CompressionStrategy",
-    "generate_source_hint",
     # Other transforms
     "CacheAligner",
     "RollingWindow",

headroom/transforms/content_router.py

@@ -360,51 +360,6 @@ def _extract_json_block(lines: list[str], start: int) -> tuple[str | None, int]:
     return None, start
 
 
-def generate_source_hint(tool_name: str, tool_input: dict[str, Any]) -> str:
-    """Generate a source hint from tool metadata.
-
-    This enables higher-confidence routing decisions.
-
-    Args:
-        tool_name: Name of the tool that produced the output.
-        tool_input: Input parameters to the tool.
-
-    Returns:
-        Source hint string (e.g., "file:auth.py", "tool:grep").
-    """
-    # File read operations
-    if tool_name in ("Read", "read_file", "cat", "ReadFile"):
-        file_path = tool_input.get("file_path", tool_input.get("path", ""))
-        if file_path:
-            return f"file:{file_path}"
-
-    # Search operations
-    if tool_name in ("Grep", "grep", "ripgrep", "rg", "search", "Search"):
-        return "tool:grep"
-
-    # Glob operations
-    if tool_name in ("Glob", "glob", "find"):
-        return "tool:glob"
-
-    # Build/test operations
-    if tool_name == "Bash":
-        command = str(tool_input.get("command", ""))
-        if any(cmd in command for cmd in ["pytest", "npm test", "cargo test", "go test"]):
-            return "tool:pytest"
-        if any(cmd in command for cmd in ["npm run build", "cargo build", "make"]):
-            return "tool:build"
-        if "git diff" in command:
-            return "tool:git-diff"
-        if "git log" in command:
-            return "tool:git-log"
-
-    # Web fetch
-    if tool_name in ("WebFetch", "fetch", "curl", "WebSearch"):
-        return "tool:web"
-
-    return ""
-
-
 class ContentRouter(Transform):
     """Intelligent router that selects optimal compression strategy.
 
@@ -462,15 +417,12 @@ class ContentRouter(Transform):
     def compress(
         self,
         content: str,
-        source_hint: str | None = None,
         context: str = "",
     ) -> RouterCompressionResult:
-        """Compress content using optimal strategy.
+        """Compress content using optimal strategy based on content detection.
 
         Args:
             content: Content to compress.
-            source_hint: Optional hint about content source.
-                Examples: "file:auth.py", "tool:grep", "tool:pytest"
             context: Optional context for relevance-aware compression.
 
         Returns:
@@ -484,87 +436,31 @@ class ContentRouter(Transform):
                 routing_log=[],
             )
 
-        # Determine strategy
-        strategy = self._determine_strategy(content, source_hint)
+        # Determine strategy from content analysis
+        strategy = self._determine_strategy(content)
 
         if strategy == CompressionStrategy.MIXED:
             return self._compress_mixed(content, context)
         else:
             return self._compress_pure(content, strategy, context)
 
-    def _determine_strategy(
-        self,
-        content: str,
-        source_hint: str | None,
-    ) -> CompressionStrategy:
-        """Determine the compression strategy.
+    def _determine_strategy(self, content: str) -> CompressionStrategy:
+        """Determine the compression strategy from content analysis.
 
         Args:
             content: Content to analyze.
-            source_hint: Optional source hint.
 
         Returns:
             Selected compression strategy.
         """
-        # 1. Source hint takes priority
-        if source_hint:
-            strategy = self._strategy_from_hint(source_hint)
-            if strategy:
-                return strategy
-
-        # 2. Check for mixed content
+        # 1. Check for mixed content
         if is_mixed_content(content):
             return CompressionStrategy.MIXED
 
-        # 3. Detect content type
+        # 2. Detect content type from content itself
         detection = detect_content_type(content)
         return self._strategy_from_detection(detection)
 
-    def _strategy_from_hint(self, hint: str) -> CompressionStrategy | None:
-        """Get strategy from source hint.
-
-        Args:
-            hint: Source hint string.
-
-        Returns:
-            Strategy if determinable, None otherwise.
-        """
-        hint_lower = hint.lower()
-
-        # File hints
-        if hint_lower.startswith("file:"):
-            file_path = hint_lower[5:]
-            if file_path.endswith((".py", ".pyw")):
-                return CompressionStrategy.CODE_AWARE
-            if file_path.endswith((".js", ".jsx", ".ts", ".tsx", ".mjs")):
-                return CompressionStrategy.CODE_AWARE
-            if file_path.endswith((".go", ".rs", ".java", ".c", ".cpp", ".h", ".hpp")):
-                return CompressionStrategy.CODE_AWARE
-            if file_path.endswith(".json"):
-                return CompressionStrategy.SMART_CRUSHER
-            if file_path.endswith((".md", ".txt", ".rst")):
-                return CompressionStrategy.TEXT
-            if file_path.endswith((".log", ".out")):
-                return CompressionStrategy.LOG
-
-        # Tool hints
-        if hint_lower.startswith("tool:"):
-            tool = hint_lower[5:]
-            if tool in ("grep", "rg", "ripgrep", "ag", "search"):
-                return CompressionStrategy.SEARCH
-            if tool in ("pytest", "jest", "cargo-test", "go-test", "npm-test"):
-                return CompressionStrategy.LOG
-            if tool in ("build", "make", "cargo-build", "npm-build"):
-                return CompressionStrategy.LOG
-            if tool in ("git-diff", "diff"):
-                return CompressionStrategy.DIFF
-
-        # Direct strategy hints (used by _process_content_blocks for tool_result)
-        if hint_lower == "json_array":
-            return CompressionStrategy.SMART_CRUSHER
-
-        return None
-
     def _strategy_from_detection(self, detection: Any) -> CompressionStrategy:
         """Get strategy from content detection result.
 
@@ -897,14 +793,13 @@ class ContentRouter(Transform):
         Args:
             messages: Messages to transform.
             tokenizer: Tokenizer for counting.
-            **kwargs: Additional arguments (context, source_hints).
+            **kwargs: Additional arguments (context).
 
         Returns:
             TransformResult with routed and compressed messages.
         """
         tokens_before = sum(tokenizer.count_text(str(m.get("content", ""))) for m in messages)
         context = kwargs.get("context", "")
-        source_hints = kwargs.get("source_hints", {})  # message_id -> hint
 
         transformed_messages: list[dict[str, Any]] = []
         transforms_applied: list[str] = []
@@ -945,9 +840,6 @@ class ContentRouter(Transform):
                 transformed_messages.append(message)
                 continue
 
-            # Get source hint if available
-            source_hint = source_hints.get(i) or source_hints.get(str(i))
-
             # Detect content type for protection decisions
             detection = detect_content_type(content)
             is_code = detection.content_type == ContentType.SOURCE_CODE
@@ -969,8 +861,8 @@ class ContentRouter(Transform):
                 transforms_applied.append("router:protected:analysis_context")
                 continue
 
-            # Route and compress
-            result = self.compress(content, source_hint=source_hint, context=context)
+            # Route and compress based on content detection
+            result = self.compress(content, context=context)
 
             if result.compression_ratio < 0.9:
                 transformed_messages.append({**message, "content": result.compressed})
@@ -1013,8 +905,6 @@ class ContentRouter(Transform):
         Returns:
             Transformed message with compressed content blocks.
         """
-        import json
-
         new_blocks = []
         any_compressed = False
 
@@ -1031,33 +921,7 @@ class ContentRouter(Transform):
 
                 # Only process string content
                 if isinstance(tool_content, str) and len(tool_content) > 500:
-                    # Try to detect if it's JSON array data (SmartCrusher target)
-                    try:
-                        parsed = json.loads(tool_content)
-                        if isinstance(parsed, list) and len(parsed) > 10:
-                            # Route to SmartCrusher for arrays
-                            result = self.compress(
-                                tool_content,
-                                source_hint="json_array",
-                                context=context,
-                            )
-                            if result.compression_ratio < 0.9:
-                                new_blocks.append(
-                                    {
-                                        **block,
-                                        "content": result.compressed,
-                                    }
-                                )
-                                transforms_applied.append(
-                                    f"router:tool_result:{result.strategy_used.value}"
-                                )
-                                any_compressed = True
-                                continue
-                    except (json.JSONDecodeError, TypeError):
-                        # Not JSON, try general compression
-                        pass
-
-                    # Try general compression for large non-JSON content
+                    # Compress using content detection (will auto-detect JSON arrays, etc.)
                     result = self.compress(tool_content, context=context)
                     if result.compression_ratio < 0.9:
                         new_blocks.append({**block, "content": result.compressed})
@@ -1140,15 +1004,13 @@ class ContentRouter(Transform):
 
 def route_and_compress(
     content: str,
-    source_hint: str | None = None,
     context: str = "",
 ) -> str:
     """Convenience function for one-off routing and compression.
 
     Args:
         content: Content to compress.
-        source_hint: Optional source hint.
-        context: Optional context.
+        context: Optional context for relevance-aware compression.
 
     Returns:
         Compressed content.
@@ -1157,5 +1019,5 @@ def route_and_compress(
         >>> compressed = route_and_compress(mixed_content)
     """
     router = ContentRouter()
-    result = router.compress(content, source_hint=source_hint, context=context)
+    result = router.compress(content, context=context)
     return result.compressed

headroom/transforms/intelligent_context.py

@@ -444,16 +444,8 @@ class IntelligentContextManager(Transform):
             # Focus on tool messages (highest compression potential)
             if role == "tool" and isinstance(content, str) and len(content) > 100:
                 try:
-                    # Get source hint from tool call context
-                    tool_call_id = msg.get("tool_call_id", "")
-                    source_hint = self._get_tool_source_hint(messages, tool_call_id)
-
-                    # Compress using ContentRouter
-                    result = router.compress(
-                        content,
-                        source_hint=source_hint,
-                        context="",  # No specific context in compress-first mode
-                    )
+                    # Compress using ContentRouter (auto-detects content type)
+                    result = router.compress(content)
 
                     # Check if compression was effective
                     if result.compression_ratio < 0.9:  # At least 10% savings
@@ -500,46 +492,6 @@ class IntelligentContextManager(Transform):
 
         return compressed_messages, transforms_applied, total_tokens_saved
 
-    def _get_tool_source_hint(self, messages: list[dict[str, Any]], tool_call_id: str) -> str:
-        """Extract source hint from the tool call that produced this result.
-
-        Args:
-            messages: List of all messages.
-            tool_call_id: The ID of the tool call to find.
-
-        Returns:
-            Source hint string (e.g., "tool:grep", "file:main.py").
-        """
-        if not tool_call_id:
-            return ""
-
-        # Find the assistant message with this tool call
-        for msg in messages:
-            if msg.get("role") == "assistant" and msg.get("tool_calls"):
-                for tc in msg.get("tool_calls", []):
-                    if tc.get("id") == tool_call_id:
-                        func = tc.get("function", {})
-                        tool_name = func.get("name", "")
-
-                        # Import here to avoid circular imports
-                        try:
-                            import json
-
-                            from .content_router import generate_source_hint
-
-                            args_str = func.get("arguments", "{}")
-                            try:
-                                args = (
-                                    json.loads(args_str) if isinstance(args_str, str) else args_str
-                                )
-                            except json.JSONDecodeError:
-                                args = {}
-
-                            return generate_source_hint(tool_name, args)
-                        except ImportError:
-                            return f"tool:{tool_name}" if tool_name else ""
-        return ""
-
     def _compress_content_blocks(
         self,
         content_blocks: list[Any],

tests/test_transforms/test_content_router.py

@@ -4,7 +4,6 @@ Comprehensive tests covering:
 - ContentRouterConfig: Configuration validation and defaults
 - ContentRouter: Core routing functionality
 - Strategy detection: Code, JSON, search, logs, text
-- Source hint parsing: File paths, tool names
 - Mixed content handling: Split, route, reassemble
 - Transform interface: apply(), should_apply() methods
 """
@@ -18,7 +17,6 @@ from headroom.transforms.content_router import (
     ContentRouterConfig,
     RouterCompressionResult,
     RoutingDecision,
-    generate_source_hint,
 )
 
 # =============================================================================
@@ -279,51 +277,6 @@ class TestRouterCompressionResult:
         assert result.savings_percentage == 0.0
 
 
-# =============================================================================
-# TestSourceHintGeneration
-# =============================================================================
-
-
-class TestSourceHintGeneration:
-    """Tests for generate_source_hint function."""
-
-    def test_file_path_hint_with_file_path(self):
-        """File path generates file: hint."""
-        hint = generate_source_hint("Read", {"file_path": "/src/auth.py"})
-        assert hint == "file:/src/auth.py"
-
-    def test_file_path_hint_with_path(self):
-        """File path with 'path' key also works."""
-        hint = generate_source_hint("read_file", {"path": "/src/auth.py"})
-        assert hint == "file:/src/auth.py"
-
-    def test_grep_tool_hint(self):
-        """Grep tool generates tool:grep hint."""
-        hint = generate_source_hint("Grep", {"pattern": "def.*"})
-        assert hint == "tool:grep"
-
-    def test_glob_tool_hint(self):
-        """Glob tool generates tool:glob hint."""
-        hint = generate_source_hint("Glob", {"pattern": "*.py"})
-        assert hint == "tool:glob"
-
-    def test_bash_test_command(self):
-        """Bash with test command generates tool:pytest hint."""
-        hint = generate_source_hint("Bash", {"command": "pytest tests/"})
-        assert hint == "tool:pytest"
-
-    def test_bash_build_command(self):
-        """Bash with build command generates tool:build hint."""
-        hint = generate_source_hint("Bash", {"command": "npm run build"})
-        assert "tool:" in hint
-
-    def test_unknown_tool(self):
-        """Unknown tool returns generic tool hint."""
-        hint = generate_source_hint("CustomTool", {"arg": "value"})
-        # Should return some hint, not crash
-        assert hint is not None
-
-
 # =============================================================================
 # TestStrategyDetection
 # =============================================================================
@@ -335,54 +288,32 @@ class TestStrategyDetection:
     def test_detect_python_code(self, router):
         """Python code is detected."""
         code = generate_python_code(5)
-        strategy = router._determine_strategy(code, None)
+        strategy = router._determine_strategy(code)
         # Should be either CODE_AWARE or fallback
         assert strategy in CompressionStrategy
 
     def test_detect_json_content(self, router):
         """JSON content is detected."""
         json_data = generate_json_data(20)
-        strategy = router._determine_strategy(json_data, None)
+        strategy = router._determine_strategy(json_data)
         assert strategy in CompressionStrategy
 
     def test_detect_search_results(self, router):
         """Search/grep results are detected."""
         search_results = generate_search_results(10)
-        strategy = router._determine_strategy(search_results, None)
+        strategy = router._determine_strategy(search_results)
         assert strategy in CompressionStrategy
 
     def test_detect_log_output(self, router):
         """Build/test logs are detected."""
         logs = generate_log_output(30)
-        strategy = router._determine_strategy(logs, None)
+        strategy = router._determine_strategy(logs)
         assert strategy in CompressionStrategy
 
     def test_detect_plain_text(self, router):
         """Plain text detection."""
         text = "This is just plain text without any special formatting."
-        strategy = router._determine_strategy(text, None)
-        assert strategy in CompressionStrategy
-
-
-# =============================================================================
-# TestSourceHintRouting
-# =============================================================================
-
-
-class TestSourceHintRouting:
-    """Tests for source hint-based routing."""
-
-    def test_file_hint_routes_appropriately(self, router):
-        """file:*.py hint influences routing."""
-        content = generate_python_code(3)
-        strategy = router._determine_strategy(content, "file:/src/auth.py")
-        # Should return a valid strategy
-        assert strategy in CompressionStrategy
-
-    def test_grep_hint_routes_appropriately(self, router):
-        """tool:grep hint influences routing."""
-        content = generate_search_results(5)
-        strategy = router._determine_strategy(content, "tool:grep")
+        strategy = router._determine_strategy(text)
         assert strategy in CompressionStrategy
 
 
@@ -430,14 +361,6 @@ class TestContentRouter:
         assert result.original == content
         assert result.strategy_used is not None
 
-    def test_compress_with_source_hint(self, router):
-        """Source hint influences routing decision."""
-        content = generate_python_code(5)
-        result = router.compress(content, source_hint="file:/src/auth.py")
-
-        # Should return a valid result
-        assert isinstance(result, RouterCompressionResult)
-
     def test_name_property(self, router):
         """Router has correct name."""
         assert router.name == "content_router"
@@ -515,7 +438,7 @@ class TestCompressorDisabling:
         code = generate_python_code(10)
 
         # Should not crash
-        result = router.compress(code, source_hint="file:/src/auth.py")
+        result = router.compress(code)
         assert result is not None
 
     def test_config_accepts_disable_search_compression(self):
@@ -528,7 +451,7 @@ class TestCompressorDisabling:
         search_results = generate_search_results(10)
 
         # Should not crash
-        result = router.compress(search_results, source_hint="tool:grep")
+        result = router.compress(search_results)
         assert result is not None
 
     def test_config_accepts_disable_log_compression(self):
@@ -541,7 +464,7 @@ class TestCompressorDisabling:
         logs = generate_log_output(30)
 
         # Should not crash
-        result = router.compress(logs, source_hint="tool:pytest")
+        result = router.compress(logs)
         assert result is not None
 
 
@@ -570,18 +493,6 @@ class TestEdgeCases:
         result = router.compress(content)
         assert result is not None
 
-    def test_null_source_hint(self, router):
-        """Null source hint doesn't crash."""
-        content = generate_python_code(5)
-        result = router.compress(content, source_hint=None)
-        assert result is not None
-
-    def test_empty_source_hint(self, router):
-        """Empty source hint doesn't crash."""
-        content = generate_python_code(5)
-        result = router.compress(content, source_hint="")
-        assert result is not None
-
 
 # =============================================================================
 # TestRoutingLog

tests/test_transforms/test_intelligent_context.py

@@ -1017,49 +1017,6 @@ class TestCompressFirstStrategy:
         router2 = manager._get_content_router()
         assert router is router2
 
-    def test_source_hint_extraction(self):
-        """Source hints should be extracted from tool calls."""
-        manager = IntelligentContextManager()
-
-        messages = [
-            {
-                "role": "assistant",
-                "tool_calls": [
-                    {
-                        "id": "call_1",
-                        "function": {
-                            "name": "Read",
-                            "arguments": '{"file_path": "/src/main.py"}',
-                        },
-                    }
-                ],
-            },
-            {
-                "role": "assistant",
-                "tool_calls": [
-                    {
-                        "id": "call_2",
-                        "function": {
-                            "name": "Grep",
-                            "arguments": '{"pattern": "def"}',
-                        },
-                    }
-                ],
-            },
-        ]
-
-        # Test file read hint
-        hint1 = manager._get_tool_source_hint(messages, "call_1")
-        assert "file:" in hint1 or hint1 == ""  # May not have content_router import
-
-        # Test grep hint
-        hint2 = manager._get_tool_source_hint(messages, "call_2")
-        assert "grep" in hint2.lower() or hint2 == ""
-
-        # Test unknown tool call
-        hint3 = manager._get_tool_source_hint(messages, "unknown")
-        assert hint3 == ""
-
 
 class TestCompressFirstWithContentBlocks:
     """Tests for COMPRESS_FIRST with Anthropic-style content blocks."""