fix(proxy): surface CompressionDecision.passthrough_reason in tags

Adds ``CompressionDecision.apply_to_tags(tags)`` — a one-liner mutator
that stamps the passthrough reason into a tags dict for downstream
observability. Each migrated handler now calls
``_decision.apply_to_tags(tags)`` immediately after
``CompressionDecision.decide(...)``. The tags dict flows unchanged
into every downstream ``RequestOutcome(tags=tags, ...)`` construction,
which the funnel surfaces in ``RequestLog.tags`` — same mechanism the
funnel already uses for ``client``.

Dashboards can now slice passthrough traffic by cause:

  * tags["passthrough_reason"] == "bypass_header"
  * tags["passthrough_reason"] == "compression_disabled"
  * tags["passthrough_reason"] == "no_messages"
  * tags["passthrough_reason"] == "license_denied"

No-op when ``should_compress=True`` — compressing requests don't
carry the tag, so absence vs presence is itself the signal.

Bonus fix: ``handle_gemini_count_tokens`` was the one Gemini handler
that never pulled tags out of headers, so its emitted
``RequestOutcome`` reached the dashboard without any of the per-
request slicing keys. Added the missing ``tags = self._extract_tags
(request.headers)`` and threaded ``tags=tags`` into its outcome.

Closes the observability loop opened by PR #477: the four Gemini-
bypass-bug fixes are now visible in the request-log feed the moment
they fire.

.claude-plugin/marketplace.json

@@ -5,14 +5,14 @@
   },
   "metadata": {
     "description": "Headroom marketplace for Claude Code and GitHub Copilot CLI plugins.",
-    "version": "0.21.38"
+    "version": "0.22.0"
   },
   "plugins": [
     {
       "name": "headroom",
       "source": "./plugins/headroom-agent-hooks",
       "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
-      "version": "0.21.38",
+      "version": "0.22.0",
       "author": {
         "name": "Headroom Contributors",
         "url": "https://github.com/chopratejas/headroom"

.github/plugin/marketplace.json

@@ -5,14 +5,14 @@
   },
   "metadata": {
     "description": "Headroom marketplace for Claude Code and GitHub Copilot CLI plugins.",
-    "version": "0.21.38"
+    "version": "0.22.0"
   },
   "plugins": [
     {
       "name": "headroom",
       "source": "./plugins/headroom-agent-hooks",
       "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
-      "version": "0.21.38",
+      "version": "0.22.0",
       "author": {
         "name": "Headroom Contributors",
         "url": "https://github.com/chopratejas/headroom"

headroom/proxy/compression_decision.py

@@ -145,3 +145,24 @@ class CompressionDecision:
             license_allows=license_ok,
             has_messages=has_msgs,
         )
+
+    def apply_to_tags(self, tags: dict[str, str]) -> None:
+        """Stamp the passthrough reason into a tags dict for downstream
+        observability.
+
+        Mutates ``tags`` in place. No-op when ``should_compress=True``
+        (compressing requests don't carry a ``passthrough_reason`` tag —
+        absence vs presence is itself the signal).
+
+        Handler call pattern, after ``CompressionDecision.decide(...)``::
+
+            tags = self._extract_tags(headers)
+            _decision = CompressionDecision.decide(...)
+            _decision.apply_to_tags(tags)
+            # ... tags now carries passthrough_reason if applicable;
+            # every downstream RequestOutcome(tags=tags, ...) inherits
+            # it for free, which flows through emit_request_outcome()
+            # → RequestLog.tags → dashboard.
+        """
+        if self.passthrough_reason is not None:
+            tags["passthrough_reason"] = self.passthrough_reason

headroom/proxy/handlers/anthropic.py

@@ -893,6 +893,7 @@ class AnthropicHandlerMixin:
                 usage_reporter=self.usage_reporter,
                 messages=messages,
             )
+            _decision.apply_to_tags(tags)
             if not _decision.should_compress:
                 logger.info(
                     f"[{request_id}] Compression skipped: reason={_decision.passthrough_reason}"

headroom/proxy/handlers/gemini.py

@@ -330,6 +330,7 @@ class GeminiHandlerMixin:
             usage_reporter=self.usage_reporter,
             messages=messages,
         )
+        _decision.apply_to_tags(tags)
         if not _decision.should_compress:
             logger.info(
                 f"[{request_id}] Compression skipped: reason={_decision.passthrough_reason}"
@@ -642,6 +643,7 @@ class GeminiHandlerMixin:
             usage_reporter=self.usage_reporter,
             messages=messages,
         )
+        _decision.apply_to_tags(tags)
         if not _decision.should_compress:
             logger.info(
                 f"[{request_id}] Compression skipped: reason={_decision.passthrough_reason}"
@@ -876,12 +878,18 @@ class GeminiHandlerMixin:
         transforms_applied: list[str] = []
         optimized_messages = messages
 
+        # countTokens is the one Gemini handler that didn't pull tags
+        # out of headers; sibling handlers do and thread them into the
+        # outcome. Extract here so apply_to_tags below has a dict to
+        # mutate and the outcome at end-of-call inherits the tag.
+        tags = self._extract_tags(request.headers)
         _decision = CompressionDecision.decide(
             headers=request.headers,
             config=self.config,
             usage_reporter=self.usage_reporter,
             messages=messages,
         )
+        _decision.apply_to_tags(tags)
         if not _decision.should_compress:
             logger.info(
                 f"[{request_id}] Compression skipped: reason={_decision.passthrough_reason}"
@@ -960,6 +968,7 @@ class GeminiHandlerMixin:
                     attempted_input_tokens=compressed_tokens + tokens_saved,
                     total_latency_ms=total_latency,
                     transforms_applied=tuple(transforms_applied),
+                    tags=tags,
                     client=client,
                 )
             )

headroom/proxy/handlers/openai.py

@@ -1409,6 +1409,7 @@ class OpenAIHandlerMixin:
             usage_reporter=self.usage_reporter,
             messages=messages,
         )
+        _decision.apply_to_tags(tags)
         if not _decision.should_compress:
             logger.info(
                 f"[{request_id}] Compression skipped: reason={_decision.passthrough_reason}"

plugins/headroom-agent-hooks/.claude-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "headroom",
-  "version": "0.21.38",
+  "version": "0.22.0",
   "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
   "author": {
     "name": "Headroom Contributors",

plugins/headroom-agent-hooks/.github/plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "headroom",
-  "version": "0.21.38",
+  "version": "0.22.0",
   "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
   "author": {
     "name": "Headroom Contributors",

tests/test_compression_decision.py

@@ -380,3 +380,112 @@ def test_decide_with_missing_messages_field_on_body() -> None:
     d = CompressionDecision.decide(headers={}, config=_config(), usage_reporter=None, messages=None)
     assert d.should_compress is False
     assert d.passthrough_reason == "no_messages"
+
+
+# ── apply_to_tags: thread passthrough_reason into RequestOutcome.tags ─
+#
+# Handlers compute ``tags = self._extract_tags(headers)`` at entry and
+# pass that dict through to every downstream ``RequestOutcome``
+# construction. ``decision.apply_to_tags(tags)`` is a one-liner mutation
+# at the post-decision point that gives every downstream outcome the
+# ``passthrough_reason`` for free — no need to thread the decision
+# through five layers of helper calls. The outcome funnel then surfaces
+# ``tags["passthrough_reason"]`` in ``RequestLog.tags`` (dashboard
+# slicing) — same mechanism the funnel already uses for ``client``.
+
+
+def test_apply_to_tags_stamps_reason_when_passthrough() -> None:
+    """On a passthrough decision, ``apply_to_tags`` mutates the supplied
+    tags dict in place with ``passthrough_reason = <the reason>``. This
+    is the single integration point between the input-side decision and
+    the output-side ``RequestOutcome``."""
+    d = CompressionDecision.decide(
+        headers={"x-headroom-bypass": "true"},
+        config=_config(),
+        usage_reporter=None,
+        messages=_msgs(),
+    )
+    tags: dict[str, str] = {}
+    d.apply_to_tags(tags)
+    assert tags == {"passthrough_reason": "bypass_header"}
+
+
+def test_apply_to_tags_is_a_noop_when_compressing() -> None:
+    """When the decision is "compress" (``passthrough_reason is None``),
+    the tags dict must be left untouched — no spurious
+    ``passthrough_reason=None`` string entry, which would mislead any
+    dashboard that filters on tag presence."""
+    d = CompressionDecision.decide(
+        headers={}, config=_config(), usage_reporter=None, messages=_msgs()
+    )
+    assert d.should_compress is True
+    tags: dict[str, str] = {"client": "codex"}
+    d.apply_to_tags(tags)
+    assert tags == {"client": "codex"}  # untouched
+    assert "passthrough_reason" not in tags
+
+
+def test_apply_to_tags_preserves_pre_existing_entries() -> None:
+    """``apply_to_tags`` is a mutator over the existing tags dict, not
+    a replacement. Pre-existing entries (``client``, custom routing
+    tags, etc.) must survive unchanged."""
+    d = CompressionDecision.decide(
+        headers={}, config=_config(optimize=False), usage_reporter=None, messages=_msgs()
+    )
+    tags: dict[str, str] = {"client": "aider", "route": "alpha"}
+    d.apply_to_tags(tags)
+    assert tags == {
+        "client": "aider",
+        "route": "alpha",
+        "passthrough_reason": "compression_disabled",
+    }
+
+
+def test_apply_to_tags_for_every_passthrough_reason() -> None:
+    """Every passthrough reason name must round-trip through
+    ``apply_to_tags`` exactly — these strings are the dashboard's
+    slicing keys; a typo would silently break filtering."""
+    reason_to_inputs: dict[str, dict[str, Any]] = {
+        "bypass_header": {
+            "headers": {"x-headroom-bypass": "true"},
+            "config": _config(),
+            "usage_reporter": None,
+            "messages": _msgs(),
+        },
+        "compression_disabled": {
+            "headers": {},
+            "config": _config(optimize=False),
+            "usage_reporter": None,
+            "messages": _msgs(),
+        },
+        "no_messages": {
+            "headers": {},
+            "config": _config(),
+            "usage_reporter": None,
+            "messages": [],
+        },
+        "license_denied": {
+            "headers": {},
+            "config": _config(),
+            "usage_reporter": _usage_reporter(should_compress=False),
+            "messages": _msgs(),
+        },
+    }
+    for expected_reason, decide_kwargs in reason_to_inputs.items():
+        d = CompressionDecision.decide(**decide_kwargs)
+        tags: dict[str, str] = {}
+        d.apply_to_tags(tags)
+        assert tags.get("passthrough_reason") == expected_reason, expected_reason
+
+
+def test_apply_to_tags_overwrites_a_pre_existing_passthrough_reason() -> None:
+    """If a tag with the same key existed before (a contrived case —
+    handlers don't write this tag elsewhere), the decision overwrites
+    it. The decision is the canonical source of truth for this tag;
+    anything earlier was stale or wrong."""
+    d = CompressionDecision.decide(
+        headers={}, config=_config(optimize=False), usage_reporter=None, messages=_msgs()
+    )
+    tags: dict[str, str] = {"passthrough_reason": "stale_value"}
+    d.apply_to_tags(tags)
+    assert tags["passthrough_reason"] == "compression_disabled"