Fix streaming tool calls, compressed request bodies, beacon field names; bump to 0.5.13

- Fix LiteLLM stream_message: emit tool_use blocks from delta.tool_calls,
set stop_reason from finish_reason (fixes silent MCP tool call failures)
- Fix _convert_messages_for_litellm: convert Anthropic tool_result/tool_use
to OpenAI role=tool/tool_calls format (fixes 500 on tool round-trips)
- Fix proxy request body parsing: decompress zstd/gzip/deflate/brotli
Content-Encoding before JSON decode (fixes Codex UnicodeDecodeError crash)
- Fix telemetry beacon field names to match /stats endpoint
(tokens.total_before_compression, not tokens.original)
- Add dedupe_telemetry.py script for hourly Supabase row deduplication

headroom/__init__.py

@@ -153,7 +153,7 @@ from .transforms import (
     TransformPipeline,
 )
 
-__version__ = "0.5.12"
+__version__ = "0.5.13"
 
 __all__ = [
     # Main client

headroom/backends/litellm.py

@@ -474,8 +474,12 @@ class LiteLLMBackend(Backend):
     def _convert_messages_for_litellm(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
         """Convert Anthropic message format to LiteLLM/OpenAI format.
 
-        LiteLLM expects OpenAI-style messages but handles most Anthropic
-        content blocks automatically.
+        Anthropic and OpenAI have different representations for tool calls:
+        - Anthropic: assistant content blocks with type=tool_use, user content blocks with type=tool_result
+        - OpenAI: assistant message with tool_calls field, separate role=tool messages
+
+        This method converts Anthropic-style messages to OpenAI-style so LiteLLM
+        can send them to any provider.
         """
         converted = []
         for msg in messages:
@@ -489,24 +493,68 @@ class LiteLLMBackend(Backend):
 
             # Handle content blocks (Anthropic style)
             if isinstance(content, list):
-                # Check if it's simple text blocks only
+                # Separate blocks by type
                 text_parts = []
-                has_complex_content = False
+                tool_use_blocks = []
+                tool_result_blocks = []
 
                 for block in content:
-                    if isinstance(block, dict):
-                        if block.get("type") == "text":
-                            text_parts.append(block.get("text", ""))
-                        elif block.get("type") in ("tool_use", "tool_result", "image"):
-                            has_complex_content = True
-                            break
-
-                if not has_complex_content and text_parts:
-                    # Simple text - join into single string
+                    if not isinstance(block, dict):
+                        continue
+                    block_type = block.get("type", "")
+                    if block_type == "text":
+                        text_parts.append(block.get("text", ""))
+                    elif block_type == "tool_use":
+                        tool_use_blocks.append(block)
+                    elif block_type == "tool_result":
+                        tool_result_blocks.append(block)
+
+                # tool_result blocks → OpenAI "tool" role messages
+                if tool_result_blocks:
+                    # Include any text from the user message first
+                    if text_parts:
+                        converted.append({"role": "user", "content": "\n".join(text_parts)})
+                    for tr in tool_result_blocks:
+                        tr_content = tr.get("content", "")
+                        if isinstance(tr_content, list):
+                            tr_content = "\n".join(
+                                b.get("text", "") for b in tr_content if b.get("type") == "text"
+                            )
+                        converted.append(
+                            {
+                                "role": "tool",
+                                "tool_call_id": tr["tool_use_id"],
+                                "content": str(tr_content),
+                            }
+                        )
+                    continue
+
+                # tool_use blocks → OpenAI assistant message with tool_calls
+                if tool_use_blocks:
+                    assistant_msg: dict[str, Any] = {"role": "assistant"}
+                    if text_parts:
+                        assistant_msg["content"] = "\n".join(text_parts)
+                    else:
+                        assistant_msg["content"] = None
+                    assistant_msg["tool_calls"] = [
+                        {
+                            "id": tu["id"],
+                            "type": "function",
+                            "function": {
+                                "name": tu["name"],
+                                "arguments": json.dumps(tu.get("input", {})),
+                            },
+                        }
+                        for tu in tool_use_blocks
+                    ]
+                    converted.append(assistant_msg)
+                    continue
+
+                # Simple text only
+                if text_parts:
                     converted.append({"role": role, "content": "\n".join(text_parts)})
                 else:
-                    # Complex content - pass through (LiteLLM handles it)
-                    converted.append({"role": role, "content": content})
+                    converted.append({"role": role, "content": ""})
 
         return converted
 
@@ -662,7 +710,12 @@ class LiteLLMBackend(Backend):
         body: dict[str, Any],
         headers: dict[str, str],
     ) -> AsyncIterator[StreamEvent]:
-        """Stream message via LiteLLM."""
+        """Stream message via LiteLLM.
+
+        Translates OpenAI streaming chunks into Anthropic SSE events.
+        Handles both text content and tool_calls dynamically — block types
+        are emitted based on what LiteLLM actually returns, not hardcoded.
+        """
         original_model = body.get("model", "claude-3-5-sonnet-20241022")
         litellm_model = self.map_model_id(original_model)
 
@@ -691,6 +744,11 @@ class LiteLLMBackend(Backend):
                 system = body["system"]
                 if isinstance(system, str):
                     kwargs["messages"].insert(0, {"role": "system", "content": system})
+                elif isinstance(system, list):
+                    system_text = " ".join(
+                        s.get("text", "") if isinstance(s, dict) else str(s) for s in system
+                    )
+                    kwargs["messages"].insert(0, {"role": "system", "content": system_text})
 
             if self.provider == "bedrock" and self.region:
                 kwargs["aws_region_name"] = self.region
@@ -715,46 +773,126 @@ class LiteLLMBackend(Backend):
                 },
             )
 
-            # Emit content_block_start
-            yield StreamEvent(
-                event_type="content_block_start",
-                data={
-                    "type": "content_block_start",
-                    "index": 0,
-                    "content_block": {"type": "text", "text": ""},
-                },
-            )
-
-            # Stream content
+            # Stream content — blocks emitted dynamically based on response
             response = await acompletion(**kwargs)
             output_tokens = 0
+            current_block_index = -1
+            active_block_type: str | None = None  # "text" or "tool_use"
+            tool_block_map: dict[int, int] = {}  # litellm tc.index → SSE block index
+            stop_reason = "end_turn"
 
             async for chunk in response:
-                if hasattr(chunk, "choices") and chunk.choices:
-                    delta = chunk.choices[0].delta
-                    if hasattr(delta, "content") and delta.content:
+                if not hasattr(chunk, "choices") or not chunk.choices:
+                    continue
+
+                choice = chunk.choices[0]
+                delta = choice.delta
+
+                # Check finish_reason to set stop_reason
+                if choice.finish_reason == "tool_calls":
+                    stop_reason = "tool_use"
+                elif choice.finish_reason == "stop":
+                    stop_reason = "end_turn"
+                elif choice.finish_reason == "length":
+                    stop_reason = "max_tokens"
+
+                # Handle tool_calls in the delta
+                if hasattr(delta, "tool_calls") and delta.tool_calls:
+                    for tc in delta.tool_calls:
+                        idx = tc.index if tc.index is not None else 0
+                        if idx not in tool_block_map:
+                            # Close previous block if open
+                            if active_block_type is not None:
+                                yield StreamEvent(
+                                    event_type="content_block_stop",
+                                    data={
+                                        "type": "content_block_stop",
+                                        "index": current_block_index,
+                                    },
+                                )
+                            # Open a new tool_use block
+                            current_block_index += 1
+                            tool_block_map[idx] = current_block_index
+                            active_block_type = "tool_use"
+                            tool_id = tc.id or f"toolu_{uuid.uuid4().hex[:24]}"
+                            tool_name = tc.function.name if tc.function and tc.function.name else ""
+                            yield StreamEvent(
+                                event_type="content_block_start",
+                                data={
+                                    "type": "content_block_start",
+                                    "index": current_block_index,
+                                    "content_block": {
+                                        "type": "tool_use",
+                                        "id": tool_id,
+                                        "name": tool_name,
+                                        "input": {},
+                                    },
+                                },
+                            )
+
+                        # Emit argument deltas
+                        if tc.function and tc.function.arguments:
+                            block_idx = tool_block_map[idx]
+                            yield StreamEvent(
+                                event_type="content_block_delta",
+                                data={
+                                    "type": "content_block_delta",
+                                    "index": block_idx,
+                                    "delta": {
+                                        "type": "input_json_delta",
+                                        "partial_json": tc.function.arguments,
+                                    },
+                                },
+                            )
+                            output_tokens += 1
+
+                # Handle text content in the delta
+                elif hasattr(delta, "content") and delta.content:
+                    if active_block_type != "text":
+                        # Close previous block if open
+                        if active_block_type is not None:
+                            yield StreamEvent(
+                                event_type="content_block_stop",
+                                data={
+                                    "type": "content_block_stop",
+                                    "index": current_block_index,
+                                },
+                            )
+                        # Open a new text block
+                        current_block_index += 1
+                        active_block_type = "text"
                         yield StreamEvent(
-                            event_type="content_block_delta",
+                            event_type="content_block_start",
                             data={
-                                "type": "content_block_delta",
-                                "index": 0,
-                                "delta": {"type": "text_delta", "text": delta.content},
+                                "type": "content_block_start",
+                                "index": current_block_index,
+                                "content_block": {"type": "text", "text": ""},
                             },
                         )
-                        output_tokens += 1  # Rough estimate
 
-            # Emit content_block_stop
-            yield StreamEvent(
-                event_type="content_block_stop",
-                data={"type": "content_block_stop", "index": 0},
-            )
+                    yield StreamEvent(
+                        event_type="content_block_delta",
+                        data={
+                            "type": "content_block_delta",
+                            "index": current_block_index,
+                            "delta": {"type": "text_delta", "text": delta.content},
+                        },
+                    )
+                    output_tokens += 1
+
+            # Close the last open block
+            if active_block_type is not None:
+                yield StreamEvent(
+                    event_type="content_block_stop",
+                    data={"type": "content_block_stop", "index": current_block_index},
+                )
 
-            # Emit message_delta with stop reason
+            # Emit message_delta with correct stop reason
             yield StreamEvent(
                 event_type="message_delta",
                 data={
                     "type": "message_delta",
-                    "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+                    "delta": {"stop_reason": stop_reason, "stop_sequence": None},
                     "usage": {"output_tokens": output_tokens},
                 },
             )

headroom/proxy/server.py

@@ -538,6 +538,73 @@ MAX_SSE_BUFFER_SIZE = 10 * 1024 * 1024
 # Maximum message array length (prevents DoS from deeply nested payloads)
 MAX_MESSAGE_ARRAY_LENGTH = 10000
 
+
+async def _read_request_json(request: Request) -> dict[str, Any]:
+    """Read and parse JSON from a request, handling compressed bodies.
+
+    Clients like OpenAI Codex may send zstd, gzip, or deflate-compressed
+    request bodies.  Starlette's ``request.json()`` does not decompress
+    automatically, causing a UnicodeDecodeError on compressed bytes.
+
+    This helper inspects ``Content-Encoding``, decompresses if needed,
+    then JSON-decodes the result.  It raises ``ValueError`` on any
+    decompression or parse failure so callers can return a clean 400.
+    """
+    encoding = (request.headers.get("content-encoding") or "").lower().strip()
+    raw = await request.body()
+
+    if encoding in ("zstd", "zstandard"):
+        try:
+            import zstandard
+
+            dctx = zstandard.ZstdDecompressor()
+            raw = dctx.decompress(raw)
+        except ImportError:
+            # Auto-detect: if bytes start with zstd magic (0x28 0xb5 0x2f 0xfd), fail clearly
+            raise ValueError(
+                "Request body is zstd-compressed but the 'zstandard' package is not installed. "
+                "Install it with: pip install zstandard"
+            ) from None
+        except Exception as exc:
+            raise ValueError(f"Failed to decompress zstd request body: {exc}") from exc
+    elif encoding == "gzip":
+        import gzip as _gzip
+
+        try:
+            raw = _gzip.decompress(raw)
+        except Exception as exc:
+            raise ValueError(f"Failed to decompress gzip request body: {exc}") from exc
+    elif encoding == "deflate":
+        import zlib
+
+        try:
+            raw = zlib.decompress(raw)
+        except Exception as exc:
+            raise ValueError(f"Failed to decompress deflate request body: {exc}") from exc
+    elif encoding == "br":
+        try:
+            import brotli
+
+            raw = brotli.decompress(raw)
+        except ImportError:
+            raise ValueError(
+                "Request body is brotli-compressed but the 'brotli' package is not installed."
+            ) from None
+        except Exception as exc:
+            raise ValueError(f"Failed to decompress brotli request body: {exc}") from exc
+    elif encoding and encoding != "identity":
+        raise ValueError(f"Unsupported Content-Encoding: {encoding}")
+
+    # Decode and parse JSON
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise ValueError(f"Request body is not valid UTF-8 (possibly compressed?): {exc}") from exc
+
+    result: dict[str, Any] = json.loads(text)
+    return result
+
+
 # Maximum compression cache sessions (prevents unbounded memory growth)
 MAX_COMPRESSION_CACHE_SESSIONS = 500
 
@@ -2195,7 +2262,13 @@ class HeadroomProxy:
         """
         # Check bypass header
         if request.headers.get("x-headroom-bypass", "").lower() == "true":
-            body = await request.json()
+            try:
+                body = await _read_request_json(request)
+            except (json.JSONDecodeError, ValueError) as e:
+                return JSONResponse(
+                    status_code=400,
+                    content={"error": f"Invalid request body: {e!s}"},
+                )
             messages = body.get("messages", [])
             return JSONResponse(
                 {
@@ -2210,7 +2283,7 @@ class HeadroomProxy:
             )
 
         try:
-            body = await request.json()
+            body = await _read_request_json(request)
         except Exception:
             return JSONResponse(
                 status_code=400,
@@ -2327,15 +2400,15 @@ class HeadroomProxy:
 
         # Parse request
         try:
-            body = await request.json()
-        except json.JSONDecodeError as e:
+            body = await _read_request_json(request)
+        except (json.JSONDecodeError, ValueError) as e:
             return JSONResponse(
                 status_code=400,
                 content={
                     "type": "error",
                     "error": {
                         "type": "invalid_request_error",
-                        "message": f"Invalid JSON in request body: {e!s}",
+                        "message": f"Invalid request body: {e!s}",
                     },
                 },
             )
@@ -3288,15 +3361,15 @@ class HeadroomProxy:
 
         # Parse request
         try:
-            body = await request.json()
-        except json.JSONDecodeError as e:
+            body = await _read_request_json(request)
+        except (json.JSONDecodeError, ValueError) as e:
             return JSONResponse(
                 status_code=400,
                 content={
                     "type": "error",
                     "error": {
                         "type": "invalid_request_error",
-                        "message": f"Invalid JSON in request body: {e!s}",
+                        "message": f"Invalid request body: {e!s}",
                     },
                 },
             )
@@ -3726,14 +3799,14 @@ class HeadroomProxy:
 
         # Parse request
         try:
-            body = await request.json()
-        except json.JSONDecodeError as e:
+            body = await _read_request_json(request)
+        except (json.JSONDecodeError, ValueError) as e:
             return JSONResponse(
                 status_code=400,
                 content={
                     "error": {
                         "code": 400,
-                        "message": f"Invalid JSON in request body: {e!s}",
+                        "message": f"Invalid request body: {e!s}",
                         "status": "INVALID_ARGUMENT",
                     }
                 },
@@ -5138,13 +5211,13 @@ class HeadroomProxy:
 
         # Parse request
         try:
-            body = await request.json()
-        except json.JSONDecodeError as e:
+            body = await _read_request_json(request)
+        except (json.JSONDecodeError, ValueError) as e:
             return JSONResponse(
                 status_code=400,
                 content={
                     "error": {
-                        "message": f"Invalid JSON in request body: {e!s}",
+                        "message": f"Invalid request body: {e!s}",
                         "type": "invalid_request_error",
                         "code": "invalid_json",
                     }
@@ -5735,7 +5808,7 @@ class HeadroomProxy:
         request_id = await self._next_request_id()
 
         try:
-            body = await request.json()
+            body = await _read_request_json(request)
         except Exception as e:
             logger.error(f"[{request_id}] Failed to parse Databricks request body: {e}")
             return JSONResponse(
@@ -5789,13 +5862,13 @@ class HeadroomProxy:
 
         # Parse request
         try:
-            body = await request.json()
-        except json.JSONDecodeError as e:
+            body = await _read_request_json(request)
+        except (json.JSONDecodeError, ValueError) as e:
             return JSONResponse(
                 status_code=400,
                 content={
                     "error": {
-                        "message": f"Invalid JSON in request body: {e!s}",
+                        "message": f"Invalid request body: {e!s}",
                         "type": "invalid_request_error",
                         "code": "invalid_json",
                     }
@@ -6272,13 +6345,13 @@ class HeadroomProxy:
 
         # Parse request
         try:
-            body = await request.json()
-        except json.JSONDecodeError as e:
+            body = await _read_request_json(request)
+        except (json.JSONDecodeError, ValueError) as e:
             return JSONResponse(
                 status_code=400,
                 content={
                     "error": {
-                        "message": f"Invalid JSON in request body: {e!s}",
+                        "message": f"Invalid request body: {e!s}",
                         "type": "invalid_request_error",
                         "code": "invalid_json",
                     }
@@ -6434,13 +6507,13 @@ class HeadroomProxy:
 
         # Parse request
         try:
-            body = await request.json()
-        except json.JSONDecodeError as e:
+            body = await _read_request_json(request)
+        except (json.JSONDecodeError, ValueError) as e:
             return JSONResponse(
                 status_code=400,
                 content={
                     "error": {
-                        "message": f"Invalid JSON in request body: {e!s}",
+                        "message": f"Invalid request body: {e!s}",
                         "code": 400,
                     }
                 },
@@ -6698,13 +6771,13 @@ class HeadroomProxy:
 
         # Parse request
         try:
-            body = await request.json()
-        except json.JSONDecodeError as e:
+            body = await _read_request_json(request)
+        except (json.JSONDecodeError, ValueError) as e:
             return JSONResponse(
                 status_code=400,
                 content={
                     "error": {
-                        "message": f"Invalid JSON in request body: {e!s}",
+                        "message": f"Invalid request body: {e!s}",
                         "code": 400,
                     }
                 },
@@ -6767,13 +6840,13 @@ class HeadroomProxy:
 
         # Parse request
         try:
-            body = await request.json()
-        except json.JSONDecodeError as e:
+            body = await _read_request_json(request)
+        except (json.JSONDecodeError, ValueError) as e:
             return JSONResponse(
                 status_code=400,
                 content={
                     "error": {
-                        "message": f"Invalid JSON in request body: {e!s}",
+                        "message": f"Invalid request body: {e!s}",
                         "code": 400,
                     }
                 },

headroom/telemetry/beacon.py

@@ -208,8 +208,8 @@ class TelemetryBeacon:
         try:
             tokens = stats.get("tokens", {})
             total_req = stats.get("requests", {}).get("total", 1)
-            tokens_before = tokens.get("original", 0)
-            tokens_after = tokens.get("optimized", 0)
+            tokens_before = tokens.get("total_before_compression", 0)
+            tokens_after = tokens_before - tokens.get("saved", 0)
             payload.update(
                 {
                     "avg_tokens_before": round(tokens_before / max(total_req, 1)),
@@ -226,7 +226,7 @@ class TelemetryBeacon:
                 payload["compression_cache"] = {
                     "hit_rate": cc.get("hit_rate", 0),
                     "entries": cc.get("entries", 0),
-                    "avg_lookup_ns": cc.get("avg_lookup_ns", 0),
+                    "tokens_saved": cc.get("total_tokens_saved", 0),
                 }
         except Exception:
             logger.debug("Beacon: failed to extract cache stats", exc_info=True)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "headroom-ai"
-version = "0.5.12"
+version = "0.5.13"
 description = "The Context Optimization Layer for LLM Applications - Cut costs by 50-90%"
 readme = "README.md"
 license = "Apache-2.0"

scripts/dedupe_telemetry.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+"""Deduplicate proxy_telemetry_v2 rows by session_id.
+
+Keeps the latest entry (by created_at) per session and deletes older duplicates.
+Run manually or via cron:
+
+    # One-off
+    python scripts/dedupe_telemetry.py
+
+    # Hourly via crontab -e
+    0 * * * * cd /Users/tchopra/claude-projects/headroom && python scripts/dedupe_telemetry.py >> /tmp/dedupe_telemetry.log 2>&1
+
+    # Dry run (show what would be deleted)
+    python scripts/dedupe_telemetry.py --dry-run
+"""
+
+from __future__ import annotations
+
+import argparse
+from collections import defaultdict
+from datetime import datetime
+
+import requests
+
+# Supabase config (same anon key as beacon — INSERT/SELECT/DELETE via RLS)
+_SUPABASE_URL = "https://dtlllcsudcoasebbamcq.supabase.co"
+_SUPABASE_KEY = ".".join(
+    [
+        "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9",
+        "eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6ImR0bGxsY3N1ZGNvYXNlYmJhbWNxIiwicm9sZSI6ImFub24iLCJpYXQiOjE3NzM3MDc4NDUsImV4cCI6MjA4OTI4Mzg0NX0",
+        "h_C6dLQKa8BVc3upgEvulR4E0K4eiEViyddRMIylKjU",
+    ]
+)
+_TABLE = "proxy_telemetry_v2"
+_ENDPOINT = f"{_SUPABASE_URL}/rest/v1/{_TABLE}"
+_HEADERS = {
+    "apikey": _SUPABASE_KEY,
+    "Authorization": f"Bearer {_SUPABASE_KEY}",
+}
+
+
+def fetch_all_rows() -> list[dict]:
+    """Fetch id, session_id, created_at for all rows."""
+    rows = []
+    offset = 0
+    limit = 1000
+    while True:
+        resp = requests.get(
+            _ENDPOINT,
+            params={
+                "select": "id,session_id,created_at",
+                "order": "created_at.desc",
+                "offset": offset,
+                "limit": limit,
+            },
+            headers=_HEADERS,
+        )
+        resp.raise_for_status()
+        batch = resp.json()
+        if not batch:
+            break
+        rows.extend(batch)
+        if len(batch) < limit:
+            break
+        offset += limit
+    return rows
+
+
+def find_duplicates(rows: list[dict]) -> list[str]:
+    """Find row IDs to delete (all but the latest per session_id)."""
+    by_session: dict[str, list[dict]] = defaultdict(list)
+    for row in rows:
+        by_session[row["session_id"]].append(row)
+
+    ids_to_delete = []
+    for _session_id, entries in by_session.items():
+        if len(entries) <= 1:
+            continue
+        # Sort by created_at descending — keep first (latest), delete rest
+        entries.sort(key=lambda r: r["created_at"], reverse=True)
+        for entry in entries[1:]:
+            ids_to_delete.append(entry["id"])
+
+    return ids_to_delete
+
+
+def delete_rows(ids: list[str], dry_run: bool = False) -> int:
+    """Delete rows by ID. Returns count deleted."""
+    if dry_run or not ids:
+        return 0
+
+    deleted = 0
+    # PostgREST supports filtering by id in a list
+    batch_size = 50
+    for i in range(0, len(ids), batch_size):
+        batch = ids[i : i + batch_size]
+        id_filter = ",".join(batch)
+        resp = requests.delete(
+            _ENDPOINT,
+            params={"id": f"in.({id_filter})"},
+            headers={**_HEADERS, "Prefer": "return=minimal"},
+        )
+        resp.raise_for_status()
+        deleted += len(batch)
+
+    return deleted
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Deduplicate proxy_telemetry_v2 by session_id")
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Show what would be deleted without deleting",
+    )
+    args = parser.parse_args()
+
+    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    print(f"[{now}] Fetching telemetry rows...")
+
+    rows = fetch_all_rows()
+    print(f"  Total rows: {len(rows)}")
+
+    # Group stats
+    by_session: dict[str, int] = defaultdict(int)
+    for row in rows:
+        by_session[row["session_id"]] += 1
+    dup_sessions = {k: v for k, v in by_session.items() if v > 1}
+
+    if not dup_sessions:
+        print("  No duplicates found. Nothing to do.")
+        return
+
+    print(f"  Sessions with duplicates: {len(dup_sessions)}")
+    total_dups = sum(v - 1 for v in dup_sessions.values())
+    print(f"  Rows to delete: {total_dups}")
+
+    ids_to_delete = find_duplicates(rows)
+
+    if args.dry_run:
+        print("\n  [DRY RUN] Would delete these rows:")
+        for row in rows:
+            if row["id"] in ids_to_delete:
+                print(
+                    f"    {row['id']}  session={row['session_id'][:12]}...  created={row['created_at']}"
+                )
+        print(f"\n  [DRY RUN] Would keep {len(rows) - len(ids_to_delete)} rows")
+        return
+
+    deleted = delete_rows(ids_to_delete)
+    print(f"  Deleted: {deleted} duplicate rows")
+    print(f"  Remaining: {len(rows) - deleted} rows (1 per session)")
+
+
+if __name__ == "__main__":
+    main()

tests/test_backend_bugs.py

@@ -1,9 +1,11 @@
 """Tests for backend bug fixes in LiteLLM and any-llm integrations.
 
 Tests tool forwarding, tool argument parsing, streaming param forwarding,
-and Vertex AI model mapping.
+message conversion (tool_use/tool_result), streaming tool_calls, and
+Vertex AI model mapping.
 """
 
+import json
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -196,6 +198,336 @@ class TestLiteLLMToolsForwarding:
             assert isinstance(tool_block["input"], dict)
 
 
+# =============================================================================
+# Message Conversion: tool_use / tool_result (GitHub Issue — Bug 2)
+# =============================================================================
+
+
+class TestConvertMessagesToolBlocks:
+    """Test that _convert_messages_for_litellm converts Anthropic tool blocks to OpenAI format."""
+
+    def _make_backend(self):
+        with patch("headroom.backends.litellm._fetch_bedrock_inference_profiles", return_value={}):
+            return LiteLLMBackend(provider="openrouter")
+
+    def test_tool_result_converted_to_tool_role(self):
+        """Anthropic tool_result blocks must become role=tool messages."""
+        backend = self._make_backend()
+        messages = [
+            {"role": "user", "content": "Weather in Paris?"},
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": "toolu_01",
+                        "name": "get_weather",
+                        "input": {"city": "Paris"},
+                    },
+                ],
+            },
+            {
+                "role": "user",
+                "content": [
+                    {"type": "tool_result", "tool_use_id": "toolu_01", "content": "Sunny, 22C"},
+                ],
+            },
+        ]
+        converted = backend._convert_messages_for_litellm(messages)
+
+        # assistant message should have tool_calls
+        assistant = converted[1]
+        assert assistant["role"] == "assistant"
+        assert "tool_calls" in assistant
+        assert assistant["tool_calls"][0]["id"] == "toolu_01"
+        assert assistant["tool_calls"][0]["type"] == "function"
+        assert assistant["tool_calls"][0]["function"]["name"] == "get_weather"
+        assert json.loads(assistant["tool_calls"][0]["function"]["arguments"]) == {"city": "Paris"}
+
+        # tool_result should become role=tool
+        tool_msg = converted[2]
+        assert tool_msg["role"] == "tool"
+        assert tool_msg["tool_call_id"] == "toolu_01"
+        assert tool_msg["content"] == "Sunny, 22C"
+
+    def test_tool_result_with_list_content(self):
+        """tool_result with list content should be flattened to string."""
+        backend = self._make_backend()
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "tool_result",
+                        "tool_use_id": "toolu_02",
+                        "content": [
+                            {"type": "text", "text": "Line 1"},
+                            {"type": "text", "text": "Line 2"},
+                        ],
+                    },
+                ],
+            },
+        ]
+        converted = backend._convert_messages_for_litellm(messages)
+        assert converted[0]["role"] == "tool"
+        assert converted[0]["content"] == "Line 1\nLine 2"
+
+    def test_assistant_tool_use_with_text(self):
+        """Assistant message with both text and tool_use blocks."""
+        backend = self._make_backend()
+        messages = [
+            {
+                "role": "assistant",
+                "content": [
+                    {"type": "text", "text": "Let me check the weather."},
+                    {
+                        "type": "tool_use",
+                        "id": "toolu_03",
+                        "name": "get_weather",
+                        "input": {"city": "Tokyo"},
+                    },
+                ],
+            },
+        ]
+        converted = backend._convert_messages_for_litellm(messages)
+        assert len(converted) == 1
+        assert converted[0]["role"] == "assistant"
+        assert converted[0]["content"] == "Let me check the weather."
+        assert converted[0]["tool_calls"][0]["function"]["name"] == "get_weather"
+
+    def test_simple_text_messages_unchanged(self):
+        """Plain string messages pass through."""
+        backend = self._make_backend()
+        messages = [
+            {"role": "user", "content": "Hello"},
+            {"role": "assistant", "content": "Hi!"},
+        ]
+        converted = backend._convert_messages_for_litellm(messages)
+        assert converted == messages
+
+    def test_multiple_tool_results(self):
+        """Multiple tool_result blocks in one user message → multiple role=tool messages."""
+        backend = self._make_backend()
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "tool_result", "tool_use_id": "toolu_a", "content": "Result A"},
+                    {"type": "tool_result", "tool_use_id": "toolu_b", "content": "Result B"},
+                ],
+            },
+        ]
+        converted = backend._convert_messages_for_litellm(messages)
+        assert len(converted) == 2
+        assert converted[0]["role"] == "tool"
+        assert converted[0]["tool_call_id"] == "toolu_a"
+        assert converted[1]["role"] == "tool"
+        assert converted[1]["tool_call_id"] == "toolu_b"
+
+
+# =============================================================================
+# Streaming tool_calls (GitHub Issue — Bug 1)
+# =============================================================================
+
+
+class TestStreamMessageToolCalls:
+    """Test that stream_message emits tool_use blocks and correct stop_reason."""
+
+    @pytest.mark.asyncio
+    async def test_stream_emits_tool_use_blocks(self):
+        """Tool calls in streaming should produce content_block_start with type=tool_use."""
+
+        async def mock_stream():
+            # First chunk: tool call start (id + name)
+            tc = MagicMock()
+            tc.index = 0
+            tc.id = "toolu_stream_01"
+            tc.function = MagicMock()
+            tc.function.name = "get_weather"
+            tc.function.arguments = ""
+
+            chunk1 = MagicMock()
+            chunk1.choices = [
+                MagicMock(delta=MagicMock(content=None, tool_calls=[tc]), finish_reason=None)
+            ]
+            yield chunk1
+
+            # Second chunk: arguments delta
+            tc2 = MagicMock()
+            tc2.index = 0
+            tc2.id = None
+            tc2.function = MagicMock()
+            tc2.function.name = None
+            tc2.function.arguments = '{"city":"Paris"}'
+
+            chunk2 = MagicMock()
+            chunk2.choices = [
+                MagicMock(delta=MagicMock(content=None, tool_calls=[tc2]), finish_reason=None)
+            ]
+            yield chunk2
+
+            # Final chunk: finish_reason=tool_calls
+            chunk3 = MagicMock()
+            chunk3.choices = [
+                MagicMock(
+                    delta=MagicMock(content=None, tool_calls=None), finish_reason="tool_calls"
+                )
+            ]
+            yield chunk3
+
+        with (
+            patch("headroom.backends.litellm.acompletion", new_callable=AsyncMock) as mock_acomp,
+            patch("headroom.backends.litellm._fetch_bedrock_inference_profiles", return_value={}),
+        ):
+            mock_acomp.return_value = mock_stream()
+            backend = LiteLLMBackend(provider="openrouter")
+
+            events = []
+            async for event in backend.stream_message(
+                {
+                    "model": "test",
+                    "messages": [{"role": "user", "content": "weather?"}],
+                    "tools": [
+                        {
+                            "name": "get_weather",
+                            "description": "Get weather",
+                            "input_schema": {"type": "object"},
+                        }
+                    ],
+                },
+                {},
+            ):
+                events.append(event)
+
+        # Find content_block_start events
+        block_starts = [e for e in events if e.event_type == "content_block_start"]
+        assert len(block_starts) == 1
+        assert block_starts[0].data["content_block"]["type"] == "tool_use"
+        assert block_starts[0].data["content_block"]["id"] == "toolu_stream_01"
+        assert block_starts[0].data["content_block"]["name"] == "get_weather"
+
+        # Find input_json_delta events
+        json_deltas = [
+            e
+            for e in events
+            if e.event_type == "content_block_delta"
+            and e.data.get("delta", {}).get("type") == "input_json_delta"
+        ]
+        assert len(json_deltas) == 1
+        assert json_deltas[0].data["delta"]["partial_json"] == '{"city":"Paris"}'
+
+        # Check stop_reason is "tool_use"
+        msg_delta = [e for e in events if e.event_type == "message_delta"]
+        assert len(msg_delta) == 1
+        assert msg_delta[0].data["delta"]["stop_reason"] == "tool_use"
+
+    @pytest.mark.asyncio
+    async def test_stream_text_still_works(self):
+        """Pure text streaming should still work correctly."""
+
+        async def mock_stream():
+            chunk = MagicMock()
+            chunk.choices = [
+                MagicMock(delta=MagicMock(content="Hello!", tool_calls=None), finish_reason=None)
+            ]
+            yield chunk
+
+            chunk2 = MagicMock()
+            chunk2.choices = [
+                MagicMock(delta=MagicMock(content=None, tool_calls=None), finish_reason="stop")
+            ]
+            yield chunk2
+
+        with (
+            patch("headroom.backends.litellm.acompletion", new_callable=AsyncMock) as mock_acomp,
+            patch("headroom.backends.litellm._fetch_bedrock_inference_profiles", return_value={}),
+        ):
+            mock_acomp.return_value = mock_stream()
+            backend = LiteLLMBackend(provider="openrouter")
+
+            events = []
+            async for event in backend.stream_message(
+                {"model": "test", "messages": [{"role": "user", "content": "hi"}]},
+                {},
+            ):
+                events.append(event)
+
+        block_starts = [e for e in events if e.event_type == "content_block_start"]
+        assert len(block_starts) == 1
+        assert block_starts[0].data["content_block"]["type"] == "text"
+
+        text_deltas = [e for e in events if e.event_type == "content_block_delta"]
+        assert len(text_deltas) == 1
+        assert text_deltas[0].data["delta"]["text"] == "Hello!"
+
+        msg_delta = [e for e in events if e.event_type == "message_delta"]
+        assert msg_delta[0].data["delta"]["stop_reason"] == "end_turn"
+
+    @pytest.mark.asyncio
+    async def test_stream_text_then_tool(self):
+        """Text followed by tool call should produce two blocks."""
+
+        async def mock_stream():
+            # Text chunk
+            chunk1 = MagicMock()
+            chunk1.choices = [
+                MagicMock(
+                    delta=MagicMock(content="I'll check. ", tool_calls=None), finish_reason=None
+                )
+            ]
+            yield chunk1
+
+            # Tool call chunk
+            tc = MagicMock()
+            tc.index = 0
+            tc.id = "toolu_mixed"
+            tc.function = MagicMock()
+            tc.function.name = "search"
+            tc.function.arguments = '{"q":"test"}'
+
+            chunk2 = MagicMock()
+            chunk2.choices = [
+                MagicMock(delta=MagicMock(content=None, tool_calls=[tc]), finish_reason=None)
+            ]
+            yield chunk2
+
+            # Finish
+            chunk3 = MagicMock()
+            chunk3.choices = [
+                MagicMock(
+                    delta=MagicMock(content=None, tool_calls=None), finish_reason="tool_calls"
+                )
+            ]
+            yield chunk3
+
+        with (
+            patch("headroom.backends.litellm.acompletion", new_callable=AsyncMock) as mock_acomp,
+            patch("headroom.backends.litellm._fetch_bedrock_inference_profiles", return_value={}),
+        ):
+            mock_acomp.return_value = mock_stream()
+            backend = LiteLLMBackend(provider="openrouter")
+
+            events = []
+            async for event in backend.stream_message(
+                {"model": "test", "messages": [{"role": "user", "content": "hi"}]},
+                {},
+            ):
+                events.append(event)
+
+        block_starts = [e for e in events if e.event_type == "content_block_start"]
+        assert len(block_starts) == 2
+        assert block_starts[0].data["content_block"]["type"] == "text"
+        assert block_starts[1].data["content_block"]["type"] == "tool_use"
+
+        # Two content_block_stop events (one per block)
+        block_stops = [e for e in events if e.event_type == "content_block_stop"]
+        assert len(block_stops) == 2
+
+        # stop_reason should be tool_use
+        msg_delta = [e for e in events if e.event_type == "message_delta"]
+        assert msg_delta[0].data["delta"]["stop_reason"] == "tool_use"
+
+
 # =============================================================================
 # Streaming Params (Bugs 3-4)
 # =============================================================================