feat(transforms): attribute read_lifecycle + smart_crush tags (#249)

## What

Enrich the `transforms_applied` tags emitted by `ReadLifecycleManager`
and `SmartCrusher` so each tag carries the specific target it acted on,
instead of being an opaque counter:

- `read_lifecycle:<state>` -> `read_lifecycle:<state>:<file_path>`
- `smart_crush:<n>` -> `smart_crush:<n>:<tool1,tool2,...>` (tool names
resolved from the assistant's `tool_calls` / `tool_use` metadata; falls
back to `smart_crush:<n>` when no name resolves)

Downstream UIs can then show *what* a compression acted on (which file
was a stale read, which tools had their output crushed), not just that
it happened.

## Note on the rebase

The original revision targeted `ToolCrusher` / `tool_crush:<n>`. That
transform has since been retired and replaced by the Rust-backed
`SmartCrusher` (which emits `smart_crush:<n>`), so the tool-name
attribution moved to `smart_crusher.py`. The `read_lifecycle` half is
unchanged.

## Response-header compatibility

`x-headroom-transforms` is built as `",".join(transforms_applied)`. A
tag containing a comma (tool-name lists; file paths) would make that
header ambiguous to split back into tags. To keep the header backward
compatible, `header_safe_transforms` (`headroom/proxy/cost.py`)
collapses the enriched tags back to their legacy counter shape **for the
header only** -- the full enriched detail still flows through the
structured `transforms_applied` list (dashboards, request logs, activity
feed). Applied at all three header sites (openai / anthropic / gemini
handlers).

Paths containing `:` survive in `transforms_applied` because consumers
bound their split to 3 parts.

## Tests

- `tests/test_transforms/test_read_lifecycle.py` -- OpenAI + Anthropic
tag shape, colon-in-path preservation
- `tests/test_transforms/test_smart_crusher_attribution.py` -- OpenAI +
Anthropic tool-name resolution, dedup, no-name fallback, id/name-missing
skips
- `tests/test_proxy/test_header_safe_transforms.py` -- header
normalization keeps the joined header unambiguous (incl. comma-in-path)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

headroom/proxy/cost.py

@@ -84,6 +84,31 @@ def _summarize_transforms(transforms: list[str]) -> str:
     return " ".join(parts)
 
 
+def header_safe_transforms(transforms: list[str]) -> list[str]:
+    """Strip enriched detail so each tag is safe in the comma-joined header.
+
+    ``x-headroom-transforms`` is built as ``",".join(transforms_applied)``, so a
+    tag must not itself contain a comma or the header can't be split back into
+    tags. The enriched ``read_lifecycle:<state>:<path>`` and
+    ``smart_crush:<n>:<names>`` tags carry comma-bearing detail (file paths may
+    contain commas; tool-name lists are comma-separated), so collapse them back
+    to their legacy counter shape for the header. Full detail stays in the
+    structured ``transforms_applied`` list (dashboards, request logs, the
+    desktop activity feed) — only the opaque header is normalized.
+    """
+    safe: list[str] = []
+    for t in transforms:
+        if t.startswith("smart_crush:"):
+            parts = t.split(":")
+            safe.append(f"smart_crush:{parts[1]}" if len(parts) >= 2 else t)
+        elif t.startswith("read_lifecycle:"):
+            parts = t.split(":")
+            safe.append(f"read_lifecycle:{parts[1]}" if len(parts) >= 2 else t)
+        else:
+            safe.append(t)
+    return safe
+
+
 def build_prefix_cache_stats(
     metrics: PrometheusMetrics,
     cost_tracker: CostTracker | None,

headroom/proxy/handlers/anthropic.py

@@ -2372,7 +2372,11 @@ class AnthropicHandlerMixin:
                     response_headers["x-headroom-tokens-saved"] = str(tokens_saved)
                     response_headers["x-headroom-model"] = model
                     if transforms_applied:
-                        response_headers["x-headroom-transforms"] = ",".join(transforms_applied)
+                        from headroom.proxy.cost import header_safe_transforms
+
+                        response_headers["x-headroom-transforms"] = ",".join(
+                            header_safe_transforms(transforms_applied)
+                        )
                     if cache_hit:
                         response_headers["x-headroom-cached"] = "true"
                     if _compression_failed:

headroom/proxy/handlers/gemini.py

@@ -672,7 +672,11 @@ class GeminiHandlerMixin:
                 response_headers["x-headroom-tokens-saved"] = str(tokens_saved)
                 response_headers["x-headroom-model"] = model
                 if transforms_applied:
-                    response_headers["x-headroom-transforms"] = ",".join(transforms_applied)
+                    from headroom.proxy.cost import header_safe_transforms
+
+                    response_headers["x-headroom-transforms"] = ",".join(
+                        header_safe_transforms(transforms_applied)
+                    )
                 if cache_read_tokens > 0:
                     response_headers["x-headroom-cached"] = "true"
                 if _compression_failed:

headroom/proxy/handlers/openai.py

@@ -45,7 +45,7 @@ from headroom.copilot_auth import apply_copilot_api_auth, build_copilot_upstream
 from headroom.pipeline import PipelineStage, summarize_routing_markers
 from headroom.proxy.auth_mode import classify_auth_mode, classify_client
 from headroom.proxy.compression_decision import CompressionDecision
-from headroom.proxy.cost import _summarize_transforms
+from headroom.proxy.cost import _summarize_transforms, header_safe_transforms
 from headroom.proxy.outcome import RequestOutcome
 from headroom.proxy.project_context import classify_project, set_current_project
 
@@ -2557,7 +2557,9 @@ class OpenAIHandlerMixin:
                 response_headers["x-headroom-tokens-saved"] = str(tokens_saved)
                 response_headers["x-headroom-model"] = model
                 if transforms_applied:
-                    response_headers["x-headroom-transforms"] = ",".join(transforms_applied)
+                    response_headers["x-headroom-transforms"] = ",".join(
+                        header_safe_transforms(transforms_applied)
+                    )
                 if cache_read_tokens > 0:
                     response_headers["x-headroom-cached"] = "true"
                 if _compression_failed:

headroom/transforms/read_lifecycle.py

@@ -65,6 +65,16 @@ class ReadClassification:
     content_size: int
 
 
+def _format_read_lifecycle_transform(classification: ReadClassification) -> str:
+    """Format a read_lifecycle transform tag including the source file path.
+
+    Shape: ``read_lifecycle:<state>:<file_path>``. Consumers splitting on ``:``
+    must bound the split to 3 parts so paths containing ``:`` are preserved.
+    """
+    path = classification.file_path or ""
+    return f"read_lifecycle:{classification.state.value}:{path}"
+
+
 @dataclass
 class ReadLifecycleResult:
     """Output of lifecycle management pass."""
@@ -381,7 +391,7 @@ class ReadLifecycleManager:
                     replaced, marker, ccr_hash = self._replace_content(content, classification)
                     if replaced:
                         result_messages.append({**msg, "content": marker})
-                        transforms.append(f"read_lifecycle:{classification.state.value}")
+                        transforms.append(_format_read_lifecycle_transform(classification))
                         if ccr_hash:
                             ccr_hashes.append(ccr_hash)
                         bytes_before += len(content.encode("utf-8"))
@@ -435,7 +445,7 @@ class ReadLifecycleManager:
                 replaced, marker, ccr_hash = self._replace_content(tool_content, classification)
                 if replaced:
                     new_blocks.append({**block, "content": marker})
-                    transforms.append(f"read_lifecycle:{classification.state.value}")
+                    transforms.append(_format_read_lifecycle_transform(classification))
                     if ccr_hash:
                         ccr_hashes.append(ccr_hash)
                     any_replaced = True

headroom/transforms/smart_crusher.py

@@ -89,6 +89,47 @@ def strip_ccr_sentinels(items: Any) -> Any:
     return [x for x in items if not is_ccr_sentinel(x)]
 
 
+# ─── Tool-name attribution ────────────────────────────────────────────────
+
+
+def _build_tool_name_index(messages: list[dict[str, Any]]) -> dict[str, str]:
+    """Map tool_call_id/tool_use_id → tool name across OpenAI + Anthropic formats.
+
+    Skips entries where id or name is missing; those calls still crush, but
+    won't contribute a tool-name to the ``smart_crush`` tag.
+    """
+    index: dict[str, str] = {}
+    for msg in messages:
+        if msg.get("role") != "assistant":
+            continue
+        for tc in msg.get("tool_calls") or []:
+            tc_id = tc.get("id")
+            name = (tc.get("function") or {}).get("name")
+            if tc_id and name:
+                index[tc_id] = name
+        content = msg.get("content")
+        if isinstance(content, list):
+            for block in content:
+                if not isinstance(block, dict) or block.get("type") != "tool_use":
+                    continue
+                bid = block.get("id")
+                name = block.get("name")
+                if bid and name:
+                    index[bid] = name
+    return index
+
+
+def _format_smart_crush_transform(count: int, tool_names: list[str]) -> str:
+    """Format ``smart_crush:<count>[:<name1,name2,...>]``.
+
+    Names are included when known so consumers can show what was crushed. Empty
+    names fall back to the count-only form for backwards compatibility.
+    """
+    if tool_names:
+        return f"smart_crush:{count}:{','.join(tool_names)}"
+    return f"smart_crush:{count}"
+
+
 # ─── Public dataclasses ───────────────────────────────────────────────────
 
 
@@ -827,6 +868,16 @@ class SmartCrusher(Transform):
         crushed_count = 0
         frozen_message_count = kwargs.get("frozen_message_count", 0)
 
+        crushed_tool_names: list[str] = []
+        seen_tool_names: set[str] = set()
+        tool_names_by_id = _build_tool_name_index(result_messages)
+
+        def _record(tool_id: str | None) -> None:
+            name = tool_names_by_id.get(tool_id or "")
+            if name and name not in seen_tool_names:
+                seen_tool_names.add(name)
+                crushed_tool_names.append(name)
+
         for msg_idx, msg in enumerate(result_messages):
             if msg_idx < frozen_message_count:
                 continue
@@ -844,6 +895,7 @@ class SmartCrusher(Transform):
                             marker = create_tool_digest_marker(compute_short_hash(content))
                             msg["content"] = crushed + "\n" + marker
                             crushed_count += 1
+                            _record(msg.get("tool_call_id"))
                             markers_inserted.append(marker)
                             if info:
                                 transforms_applied.append(f"smart:{info}")
@@ -870,13 +922,16 @@ class SmartCrusher(Transform):
                         marker = create_tool_digest_marker(compute_short_hash(tool_content))
                         content[i]["content"] = crushed + "\n" + marker
                         crushed_count += 1
+                        _record(block.get("tool_use_id"))
                         markers_inserted.append(marker)
                         if info:
                             transforms_applied.append(f"smart:{info}")
                         self._notify_observer(tokens, tokenizer.count_text(crushed))
 
         if crushed_count > 0:
-            transforms_applied.insert(0, f"smart_crush:{crushed_count}")
+            transforms_applied.insert(
+                0, _format_smart_crush_transform(crushed_count, crushed_tool_names)
+            )
 
         tokens_after = tokenizer.count_messages(result_messages)
 

tests/test_proxy/test_header_safe_transforms.py

@@ -0,0 +1,38 @@
+"""`header_safe_transforms` keeps the comma-joined transforms header splittable.
+
+`x-headroom-transforms` is built as ``",".join(transforms_applied)``. The
+enriched ``read_lifecycle:<state>:<path>`` and ``smart_crush:<n>:<names>`` tags
+can contain commas, which would make that header ambiguous; the helper collapses
+them back to their legacy counter shape so each header token stays comma-free.
+"""
+
+from __future__ import annotations
+
+from headroom.proxy.cost import header_safe_transforms
+
+
+def test_strips_smart_crush_tool_names():
+    assert header_safe_transforms(["smart_crush:2:Bash,Grep"]) == ["smart_crush:2"]
+
+
+def test_strips_read_lifecycle_path():
+    assert header_safe_transforms(["read_lifecycle:stale:/src/App.tsx"]) == ["read_lifecycle:stale"]
+
+
+def test_strips_read_lifecycle_path_with_comma():
+    # A path containing a comma is exactly the case that would corrupt the header.
+    assert header_safe_transforms(["read_lifecycle:superseded:/tmp/a,b/x.py"]) == [
+        "read_lifecycle:superseded"
+    ]
+
+
+def test_passes_through_legacy_and_unrelated_tags():
+    tags = ["smart_crush:3", "read_lifecycle:stale", "router:excluded:tool", "smart:lossless:table"]
+    assert header_safe_transforms(tags) == tags
+
+
+def test_joined_header_remains_unambiguous():
+    tags = ["smart_crush:2:Bash,Grep", "read_lifecycle:stale:/a,b.py", "router:excluded:tool"]
+    header = ",".join(header_safe_transforms(tags))
+    # One token per tag — no stray commas leaking in from enriched detail.
+    assert header.split(",") == ["smart_crush:2", "read_lifecycle:stale", "router:excluded:tool"]

tests/test_transforms/test_read_lifecycle.py

@@ -506,6 +506,80 @@ class TestTransformTracking:
         stale_transforms = [t for t in result.transforms_applied if "stale" in t]
         assert len(stale_transforms) == 2  # Both reads are stale
 
+    def test_transform_tag_includes_file_path_openai(self):
+        """OpenAI-format tag shape is ``read_lifecycle:<state>:<file_path>``."""
+        config = ReadLifecycleConfig(enabled=True)
+        mgr = ReadLifecycleManager(config)
+        messages = [
+            make_openai_read("r1", "/src/app.py"),
+            make_openai_tool_result("r1", LARGE_CONTENT),
+            make_openai_edit("e1", "/src/app.py"),
+            make_openai_tool_result("e1", "done"),
+        ]
+
+        result = mgr.apply(messages)
+        assert "read_lifecycle:stale:/src/app.py" in result.transforms_applied
+
+    def test_transform_tag_includes_file_path_anthropic(self):
+        """Anthropic-format tag shape matches OpenAI tag shape."""
+        config = ReadLifecycleConfig(enabled=True)
+        mgr = ReadLifecycleManager(config)
+        messages = [
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": "r1",
+                        "name": "Read",
+                        "input": {"file_path": "/src/notes.md"},
+                    }
+                ],
+            },
+            {
+                "role": "user",
+                "content": [{"type": "tool_result", "tool_use_id": "r1", "content": LARGE_CONTENT}],
+            },
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": "e1",
+                        "name": "Edit",
+                        "input": {
+                            "file_path": "/src/notes.md",
+                            "old_string": "old",
+                            "new_string": "new",
+                        },
+                    }
+                ],
+            },
+            {
+                "role": "user",
+                "content": [{"type": "tool_result", "tool_use_id": "e1", "content": "done"}],
+            },
+        ]
+
+        result = mgr.apply(messages)
+        assert "read_lifecycle:stale:/src/notes.md" in result.transforms_applied
+
+    def test_transform_tag_preserves_colons_in_path(self):
+        """Paths containing ``:`` survive — consumers must bound their split."""
+        config = ReadLifecycleConfig(enabled=True)
+        mgr = ReadLifecycleManager(config)
+        weird_path = "/tmp/has:colon/file.py"
+        messages = [
+            make_openai_read("r1", weird_path),
+            make_openai_tool_result("r1", LARGE_CONTENT),
+            make_openai_edit("e1", weird_path),
+            make_openai_tool_result("e1", "done"),
+        ]
+
+        result = mgr.apply(messages)
+        tag = next(t for t in result.transforms_applied if t.startswith("read_lifecycle:stale"))
+        assert tag.split(":", 2) == ["read_lifecycle", "stale", weird_path]
+
 
 class TestNoFilePathHandling:
     """Reads without parseable file_path should be left alone."""

tests/test_transforms/test_smart_crusher_attribution.py

@@ -0,0 +1,170 @@
+"""Tool-name attribution on the ``smart_crush`` transform tag.
+
+When `SmartCrusher` crushes tool outputs it enriches the
+``smart_crush:<count>`` tag with the names of the tools whose output was
+crushed: ``smart_crush:<count>:<name1,name2,...>``. Names are resolved
+from the assistant's ``tool_calls`` (OpenAI) / ``tool_use`` blocks
+(Anthropic). When no name resolves, the tag falls back to the legacy
+count-only shape.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from headroom import OpenAIProvider, Tokenizer
+
+
+def _build_extension() -> None:
+    try:
+        from headroom._core import SmartCrusher  # noqa: F401
+    except ImportError:
+        pytest.skip(
+            "headroom._core not built — run `bash scripts/build_rust_extension.sh`",
+            allow_module_level=True,
+        )
+
+
+_build_extension()
+
+_provider = OpenAIProvider()
+
+
+def get_tokenizer(model: str = "gpt-4o") -> Tokenizer:
+    token_counter = _provider.get_token_counter(model)
+    return Tokenizer(token_counter, model)
+
+
+def _make_crusher():
+    from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig
+
+    return SmartCrusher(SmartCrusherConfig(min_tokens_to_crush=10))
+
+
+# Uniform / tabular payloads the Rust crusher reliably compacts.
+_LARGE_A = {"items": [{"id": i, "v": "x" * 10} for i in range(40)]}
+_LARGE_B = {"rows": list(range(200))}
+
+
+class TestSmartCrushAttribution:
+    def test_transform_tag_includes_tool_names_openai(self):
+        """Tag shape is ``smart_crush:<count>:<name1,name2>`` for OpenAI format."""
+        messages = [
+            {
+                "role": "assistant",
+                "content": None,
+                "tool_calls": [
+                    {
+                        "id": "c1",
+                        "type": "function",
+                        "function": {"name": "Bash", "arguments": "{}"},
+                    },
+                    {
+                        "id": "c2",
+                        "type": "function",
+                        "function": {"name": "Grep", "arguments": "{}"},
+                    },
+                ],
+            },
+            {"role": "tool", "tool_call_id": "c1", "content": json.dumps(_LARGE_A)},
+            {"role": "tool", "tool_call_id": "c2", "content": json.dumps(_LARGE_B)},
+        ]
+
+        result = _make_crusher().apply(messages, get_tokenizer())
+
+        tags = [t for t in result.transforms_applied if t.startswith("smart_crush:")]
+        assert len(tags) == 1
+        parts = tags[0].split(":", 2)
+        assert parts[0] == "smart_crush"
+        assert parts[1] == "2"
+        # Order follows first-crushed-first.
+        assert parts[2] == "Bash,Grep"
+
+    def test_transform_tag_includes_tool_names_anthropic(self):
+        """Anthropic tool_use blocks feed the tool-name index."""
+        messages = [
+            {
+                "role": "assistant",
+                "content": [{"type": "tool_use", "id": "u1", "name": "Read", "input": {}}],
+            },
+            {
+                "role": "user",
+                "content": [
+                    {"type": "tool_result", "tool_use_id": "u1", "content": json.dumps(_LARGE_A)},
+                ],
+            },
+        ]
+
+        result = _make_crusher().apply(messages, get_tokenizer())
+
+        assert "smart_crush:1:Read" in result.transforms_applied
+
+    def test_transform_tag_dedupes_repeated_tool(self):
+        """Same tool crushed twice shows once in the tag."""
+        messages = [
+            {
+                "role": "assistant",
+                "content": None,
+                "tool_calls": [
+                    {
+                        "id": "c1",
+                        "type": "function",
+                        "function": {"name": "Bash", "arguments": "{}"},
+                    },
+                    {
+                        "id": "c2",
+                        "type": "function",
+                        "function": {"name": "Bash", "arguments": "{}"},
+                    },
+                ],
+            },
+            {"role": "tool", "tool_call_id": "c1", "content": json.dumps(_LARGE_A)},
+            {"role": "tool", "tool_call_id": "c2", "content": json.dumps(_LARGE_A)},
+        ]
+
+        result = _make_crusher().apply(messages, get_tokenizer())
+
+        assert "smart_crush:2:Bash" in result.transforms_applied
+
+    def test_tool_name_index_skips_entries_missing_id_or_name(self):
+        """tool_calls / tool_use blocks missing id or name are skipped, other
+        blocks (text, etc.) are skipped, and the tag still reflects the entries
+        that DO have both."""
+        messages = [
+            {
+                "role": "assistant",
+                "content": [
+                    {"type": "tool_use", "name": "NamelessRead"},  # no id → skipped
+                    {"type": "tool_use", "id": "u0"},  # no name → skipped
+                    {"type": "text", "text": "thinking..."},  # not tool_use → skipped
+                    {"type": "tool_use", "id": "u1", "name": "Grep", "input": {}},  # good
+                ],
+                "tool_calls": [
+                    {"id": "", "function": {"name": "Empty"}},  # no id → skipped
+                    {"id": "c1", "function": {"name": ""}},  # no name → skipped
+                ],
+            },
+            {
+                "role": "user",
+                "content": [
+                    {"type": "tool_result", "tool_use_id": "u1", "content": json.dumps(_LARGE_A)},
+                ],
+            },
+        ]
+
+        result = _make_crusher().apply(messages, get_tokenizer())
+
+        assert "smart_crush:1:Grep" in result.transforms_applied
+
+    def test_transform_tag_falls_back_when_no_names(self):
+        """Crushed tool with no resolvable name keeps legacy ``smart_crush:<n>`` shape."""
+        # No assistant message → no name index entries.
+        messages = [
+            {"role": "tool", "tool_call_id": "orphan", "content": json.dumps(_LARGE_A)},
+        ]
+
+        result = _make_crusher().apply(messages, get_tokenizer())
+
+        assert "smart_crush:1" in result.transforms_applied