Add tool input vs result separation diagnostics

Separately track TOOL_USE (inputs) and TOOL_RESULT (outputs) tokens
to enable more precise context optimization guidance.

New issue codes:
- TOOL_INPUT_BLOAT: tool inputs consume excessive context (compressible)
- TOOL_RESULT_DOMINATES: tool results dominate context (needs summarization)

Enhanced TOP_TOOL_CONTEXT_HOTSPOT with input/result breakdown in evidence.

Author: yanpgwang

src/context_profiler/analyzers/token_counter.py

@@ -18,6 +18,8 @@ def analyze(self, request: APIRequest) -> AnalyzerResult:
         by_role: dict[str, int] = defaultdict(int)
         by_content_type: dict[str, int] = defaultdict(int)
         by_tool_name: dict[str, int] = defaultdict(int)
+        by_tool_name_input: dict[str, int] = defaultdict(int)
+        by_tool_name_result: dict[str, int] = defaultdict(int)
         per_message: list[dict] = []
         total_tokens = 0
 
@@ -30,6 +32,10 @@ def analyze(self, request: APIRequest) -> AnalyzerResult:
                 by_content_type[block.block_type.value] += block.token_count
                 if block.tool_name:
                     by_tool_name[block.tool_name] += block.token_count
+                    if block.block_type == BlockType.TOOL_USE:
+                        by_tool_name_input[block.tool_name] += block.token_count
+                    elif block.block_type == BlockType.TOOL_RESULT:
+                        by_tool_name_result[block.tool_name] += block.token_count
 
             per_message.append({
                 "index": msg.index,
@@ -43,12 +49,17 @@ def analyze(self, request: APIRequest) -> AnalyzerResult:
 
         top_messages = sorted(per_message, key=lambda x: x["tokens"], reverse=True)[:10]
         top_tools = sorted(by_tool_name.items(), key=lambda x: x[1], reverse=True)[:10]
+        top_tools_input = sorted(by_tool_name_input.items(), key=lambda x: x[1], reverse=True)[:10]
+        top_tools_result = sorted(by_tool_name_result.items(), key=lambda x: x[1], reverse=True)[:10]
 
         tool_defs_detail = [
             {"name": t.name, "tokens": t.token_count}
             for t in sorted(request.tools, key=lambda t: t.token_count, reverse=True)
         ]
 
+        tool_use_tokens = by_content_type.get("tool_use", 0)
+        tool_result_tokens = by_content_type.get("tool_result", 0)
+
         summary = {
             "total_input_tokens": total_tokens,
             "message_tokens": total_tokens - tool_def_tokens,
@@ -57,7 +68,11 @@ def analyze(self, request: APIRequest) -> AnalyzerResult:
             "source_format": request.source_format,
             "by_role": dict(by_role),
             "by_content_type": dict(by_content_type),
+            "tool_use_tokens": tool_use_tokens,
+            "tool_result_tokens": tool_result_tokens,
             "top_tools_by_tokens": top_tools,
+            "top_tools_by_input_tokens": top_tools_input,
+            "top_tools_by_result_tokens": top_tools_result,
             "tool_definitions": tool_defs_detail,
         }
 

src/context_profiler/diagnostics.py

@@ -12,6 +12,10 @@
 
 _TOOL_USE_DOMINATES_RATIO = 0.5
 _TOOL_USE_DOMINATES_MIN_TOKENS = 100
+_TOOL_INPUT_BLOAT_RATIO = 0.3
+_TOOL_INPUT_BLOAT_MIN_TOKENS = 100
+_TOOL_RESULT_DOMINATES_RATIO = 0.4
+_TOOL_RESULT_DOMINATES_MIN_TOKENS = 100
 _TOP_TOOL_HOTSPOT_RATIO = 0.3
 _TOP_TOOL_HOTSPOT_MIN_TOKENS = 100
 _REPEATED_FIELD_MIN_WASTE_TOKENS = 500
@@ -61,6 +65,40 @@ def diagnose_result(result: ProfileResult, session: Session | None = None) -> di
                     "recommendation": "Consider externalizing large tool inputs or replacing bulky payloads with artifact references.",
                 })
 
+        # Separate tool input and result analysis
+        tool_use_tokens = summary.get("tool_use_tokens", 0) or 0
+        tool_result_tokens = summary.get("tool_result_tokens", 0) or 0
+
+        if total and tool_use_tokens >= _TOOL_INPUT_BLOAT_MIN_TOKENS:
+            tool_input_ratio = tool_use_tokens / total
+            if tool_input_ratio >= _TOOL_INPUT_BLOAT_RATIO:
+                issues.append({
+                    "code": "TOOL_INPUT_BLOAT",
+                    "severity": _severity_for_ratio(tool_input_ratio),
+                    "message": "Tool inputs (not results) consume a large share of context.",
+                    "evidence": {
+                        "tool_input_tokens": tool_use_tokens,
+                        "total_input_tokens": total,
+                        "ratio": tool_input_ratio,
+                    },
+                    "recommendation": "Tool inputs are often compressible. Consider using artifact references, shorter identifiers, or externalizing large payloads.",
+                })
+
+        if total and tool_result_tokens >= _TOOL_RESULT_DOMINATES_MIN_TOKENS:
+            tool_result_ratio = tool_result_tokens / total
+            if tool_result_ratio >= _TOOL_RESULT_DOMINATES_RATIO:
+                issues.append({
+                    "code": "TOOL_RESULT_DOMINATES",
+                    "severity": _severity_for_ratio(tool_result_ratio),
+                    "message": "Tool results (not inputs) dominate the context budget.",
+                    "evidence": {
+                        "tool_result_tokens": tool_result_tokens,
+                        "total_input_tokens": total,
+                        "ratio": tool_result_ratio,
+                    },
+                    "recommendation": "Tool results often contain real data that cannot be compressed. Consider summarizing large outputs, paginating results, or using streaming.",
+                })
+
         top_tools = summary.get("top_tools_by_tokens", [])
         if total and top_tools:
             top_tool_name, top_tool_tokens = top_tools[0]
@@ -69,13 +107,21 @@ def diagnose_result(result: ProfileResult, session: Session | None = None) -> di
                 top_tool_tokens >= _TOP_TOOL_HOTSPOT_MIN_TOKENS
                 and top_tool_ratio >= _TOP_TOOL_HOTSPOT_RATIO
             ):
+                # Enhanced message with input/result breakdown
+                top_tools_input = summary.get("top_tools_by_input_tokens", [])
+                top_tools_result = summary.get("top_tools_by_result_tokens", [])
+                input_tokens = dict(top_tools_input).get(top_tool_name, 0)
+                result_tokens = dict(top_tools_result).get(top_tool_name, 0)
+
                 issues.append({
                     "code": "TOP_TOOL_CONTEXT_HOTSPOT",
                     "severity": _severity_for_ratio(top_tool_ratio),
                     "message": f"{top_tool_name} is the largest visible tool context hotspot.",
                     "evidence": {
                         "tool_name": top_tool_name,
                         "tool_tokens": top_tool_tokens,
+                        "tool_input_tokens": input_tokens,
+                        "tool_result_tokens": result_tokens,
                         "total_input_tokens": total,
                         "ratio": top_tool_ratio,
                     },

tests/test_tool_input_result_separation.py

@@ -0,0 +1,73 @@
+"""Tests for tool input vs result token separation."""
+
+import json
+from pathlib import Path
+from click.testing import CliRunner
+from context_profiler.cli import main
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+
+def test_token_counter_separates_tool_input_and_result(tmp_path):
+    """Verify that token_counter tracks tool_use and tool_result separately."""
+    snapshot = FIXTURES / "repeated_tool_calls.json"
+    out = tmp_path / "analysis.json"
+    runner = CliRunner()
+    result = runner.invoke(main, ["analyze", str(snapshot), "-o", str(out)])
+    assert result.exit_code == 0
+
+    data = json.loads(out.read_text())
+    token_summary = data["analyzers"]["token_counter"]["summary"]
+
+    # Check that we have the new fields
+    assert "tool_use_tokens" in token_summary
+    assert "tool_result_tokens" in token_summary
+    assert "top_tools_by_input_tokens" in token_summary
+    assert "top_tools_by_result_tokens" in token_summary
+
+    # Verify the sum makes sense
+    by_content = token_summary["by_content_type"]
+    assert token_summary["tool_use_tokens"] == by_content.get("tool_use", 0)
+    assert token_summary["tool_result_tokens"] == by_content.get("tool_result", 0)
+
+
+def test_diagnose_detects_tool_input_bloat():
+    """Verify TOOL_INPUT_BLOAT issue is detected when tool inputs dominate."""
+    snapshot = FIXTURES / "repeated_tool_calls.json"
+    runner = CliRunner()
+    result = runner.invoke(main, ["diagnose", str(snapshot), "--format", "openai", "--json"])
+    assert result.exit_code == 0
+
+    data = json.loads(result.output)
+    issue_codes = [issue["code"] for issue in data["issues"]]
+
+    # This fixture has high tool_use content, so we should see the issue
+    if data.get("issues"):
+        # Check if TOOL_INPUT_BLOAT or TOOL_USE_DOMINATES_CONTEXT is present
+        assert "TOOL_INPUT_BLOAT" in issue_codes or "TOOL_USE_DOMINATES_CONTEXT" in issue_codes
+
+
+def test_top_tool_hotspot_includes_input_result_breakdown():
+    """Verify TOP_TOOL_CONTEXT_HOTSPOT evidence includes input/result breakdown."""
+    snapshot = FIXTURES / "repeated_tool_calls.json"
+    runner = CliRunner()
+    result = runner.invoke(main, ["diagnose", str(snapshot), "--format", "openai", "--json"])
+    assert result.exit_code == 0
+
+    data = json.loads(result.output)
+    hotspot_issues = [issue for issue in data["issues"] if issue["code"] == "TOP_TOOL_CONTEXT_HOTSPOT"]
+
+    if hotspot_issues:
+        issue = hotspot_issues[0]
+        evidence = issue["evidence"]
+
+        # Check that the evidence includes the new breakdown fields
+        assert "tool_input_tokens" in evidence
+        assert "tool_result_tokens" in evidence
+        assert "tool_tokens" in evidence
+
+        # Verify the breakdown sums to total (or close, accounting for rounding)
+        total = evidence["tool_tokens"]
+        input_tokens = evidence["tool_input_tokens"]
+        result_tokens = evidence["tool_result_tokens"]
+        assert input_tokens + result_tokens == total