diff --git a/headroom/__init__.py b/headroom/__init__.py
index 8c14ea748..01bc2633b 100644
--- a/headroom/__init__.py
+++ b/headroom/__init__.py
@@ -119,6 +119,7 @@ __all__ = [
     "DiffArtifact",
     "RequestMetrics",
     "SimulationResult",
+    "MessageDecision",
     "TransformDiff",
     "TransformResult",
     "WasteSignals",
@@ -223,6 +224,7 @@ _LAZY_EXPORTS: dict[str, tuple[str, str]] = {
     "DiffArtifact": ("headroom.config", "DiffArtifact"),
     "RequestMetrics": ("headroom.config", "RequestMetrics"),
     "SimulationResult": ("headroom.config", "SimulationResult"),
+    "MessageDecision": ("headroom.config", "MessageDecision"),
     "TransformDiff": ("headroom.config", "TransformDiff"),
     "TransformResult": ("headroom.config", "TransformResult"),
     "WasteSignals": ("headroom.config", "WasteSignals"),
diff --git a/headroom/compress.py b/headroom/compress.py
index 2d1a786b6..fc7841c1e 100644
--- a/headroom/compress.py
+++ b/headroom/compress.py
@@ -57,15 +57,19 @@ Examples:
 from __future__ import annotations
 
 import logging
+import os
 import threading
 from dataclasses import dataclass, field, replace
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from .agent_savings import apply_agent_savings_profile
 from .observability import get_otel_metrics
 from .pipeline import PipelineExtensionManager, PipelineStage, summarize_routing_markers
 from .utils import extract_user_query as _extract_user_query
 
+if TYPE_CHECKING:
+    from .config import MessageDecision
+
 logger = logging.getLogger(__name__)
 
 
@@ -146,6 +150,14 @@ class CompressConfig:
     savings_profile: str | None = None
     """Named high-savings profile, e.g. 'agent-90' for Codex/Claude/Cursor."""
 
+    diagnostics: bool = False
+    """Collect per-message compression decisions.  Also enabled by the
+    ``HEADROOM_DIAGNOSTICS=1`` environment variable.  When True, the returned
+    :class:`CompressResult` carries a ``diagnostics`` list of
+    :class:`~headroom.config.MessageDecision` objects — one per message —
+    describing which action was taken (compressed, protected, skipped …) and
+    how many tokens were spent before/after."""
+
 
 @dataclass
 class CompressResult:
@@ -166,6 +178,7 @@ class CompressResult:
     tokens_saved: int = 0
     compression_ratio: float = 0.0
     transforms_applied: list[str] = field(default_factory=list)
+    diagnostics: list[MessageDecision] | None = None
 
 
 def compress(
@@ -224,6 +237,8 @@ def compress(
     if cfg.savings_profile:
         apply_agent_savings_profile(cfg, cfg.savings_profile)
 
+    collect_diagnostics = cfg.diagnostics or os.environ.get("HEADROOM_DIAGNOSTICS", "") == "1"
+
     pipeline = _get_pipeline()
     pipeline_extensions = PipelineExtensionManager(hooks=hooks, discover=False)
 
@@ -266,6 +281,7 @@ def compress(
             min_tokens_to_compress=cfg.min_tokens_to_compress,
             kompress_model=cfg.kompress_model,
             frozen_message_count=cfg.frozen_message_count,
+            collect_diagnostics=collect_diagnostics,
         )
 
         tokens_before = result.tokens_before
@@ -344,6 +360,7 @@ def compress(
             tokens_saved=tokens_saved,
             compression_ratio=ratio,
             transforms_applied=result.transforms_applied,
+            diagnostics=result.message_decisions if collect_diagnostics else None,
         )
 
     except Exception as e:
diff --git a/headroom/config.py b/headroom/config.py
index b1dc607eb..fdde6c8ab 100644
--- a/headroom/config.py
+++ b/headroom/config.py
@@ -790,6 +790,22 @@ class CachePrefixMetrics:
     previous_hash: str | None = None  # Previous hash for comparison (None = first request)
 
 
+@dataclass
+class MessageDecision:
+    """Per-message compression decision recorded in diagnostics mode.
+
+    Collected by ContentRouter when ``collect_diagnostics=True`` is passed
+    as a kwarg (set automatically when ``CompressConfig.diagnostics`` is True
+    or ``HEADROOM_DIAGNOSTICS=1`` is set in the environment).
+    """
+
+    message_index: int
+    role: str
+    tokens_before: int
+    tokens_after: int
+    action: str
+
+
 @dataclass
 class TransformResult:
     """Output of a transform operation."""
@@ -804,6 +820,7 @@ class TransformResult:
     cache_metrics: CachePrefixMetrics | None = None  # Populated by CacheAligner
     timing: dict[str, float] = field(default_factory=dict)  # transform_name → ms
     waste_signals: WasteSignals | None = None  # Detected waste in original messages
+    message_decisions: list[MessageDecision] = field(default_factory=list)
 
     @property
     def transforms_summary(self) -> dict[str, int]:
diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py
index 70002c29f..9532df438 100644
--- a/headroom/transforms/content_router.py
+++ b/headroom/transforms/content_router.py
@@ -53,6 +53,7 @@ from typing import Any
 from ..config import (
     DEFAULT_EXCLUDE_TOOLS,
     DEFAULT_VERBATIM_EXCLUDE_TOOLS,
+    MessageDecision,
     ReadLifecycleConfig,
     RelevanceScorerConfig,
     TransformResult,
@@ -5009,6 +5010,10 @@ class ContentRouter(Transform):
         }
         compressed_details: list[str] = []  # e.g. ["code_aware:0.72", "kompress:0.65"]
 
+        # Per-message diagnostics (collected only when collect_diagnostics=True)
+        collect_diagnostics = bool(kwargs.get("collect_diagnostics", False))
+        _diag: dict[int, str] = {}  # slot_index → action string
+
         # Check for analysis intent in the most recent user message
         analysis_intent = False
         if self.config.protect_analysis_context:
@@ -5110,6 +5115,8 @@ class ContentRouter(Transform):
                 else:
                     # Frozen — byte-identical to preserve the prefix cache.
                     result_slots[i] = message
+                    if collect_diagnostics:
+                        _diag[i] = "passthrough:frozen"
                     continue
 
             role = message.get("role", "")
@@ -5141,12 +5148,16 @@ class ContentRouter(Transform):
                 )
                 result_slots[i] = transformed_message
                 route_counts["content_blocks"] += 1
+                if collect_diagnostics:
+                    _diag[i] = "compressed:content_blocks"
                 continue
 
             # Skip non-string content (other types)
             if not isinstance(content, str):
                 result_slots[i] = message
                 route_counts["non_string"] += 1
+                if collect_diagnostics:
+                    _diag[i] = "passthrough:non_string"
                 continue
 
             # A headroom_retrieve result IS already-retrieved, original CCR content --
@@ -5166,6 +5177,8 @@ class ContentRouter(Transform):
                 result_slots[i] = message
                 transforms_applied.append("router:excluded:ccr_retrieve")
                 route_counts["ccr_retrieve"] += 1
+                if collect_diagnostics:
+                    _diag[i] = "protected:ccr_retrieve"
                 continue
 
             # Skip OpenAI-style tool messages for excluded tools
@@ -5178,6 +5191,8 @@ class ContentRouter(Transform):
                         result_slots[i] = message
                         transforms_applied.append("router:excluded:tool")
                         route_counts["excluded_tool"] += 1
+                        if collect_diagnostics:
+                            _diag[i] = "protected:excluded_tool_verbatim"
                         continue
                     if messages_from_end <= read_protection_window:
                         # Protected from lossy compression — but grep/log/json
@@ -5190,11 +5205,15 @@ class ContentRouter(Transform):
                             route_counts["excluded_tool_lossless"] = (
                                 route_counts.get("excluded_tool_lossless", 0) + 1
                             )
+                            if collect_diagnostics:
+                                _diag[i] = f"compressed:lossless_{kind}"
                             continue
                         # Recent — protect as before
                         result_slots[i] = message
                         transforms_applied.append("router:excluded:tool")
                         route_counts["excluded_tool"] += 1
+                        if collect_diagnostics:
+                            _diag[i] = "protected:excluded_tool_recent"
                         continue
                     # Old excluded-tool output — fall through to compression
                     # (the LLM is unlikely to need exact content from this far back,
@@ -5213,6 +5232,8 @@ class ContentRouter(Transform):
                     route_counts["bash_lossless_search"] = (
                         route_counts.get("bash_lossless_search", 0) + 1
                     )
+                    if collect_diagnostics:
+                        _diag[i] = "compressed:bash_lossless_search"
                     continue
 
             # Read protection (ROLE / SHAPE-AGNOSTIC). An observation produced by
@@ -5240,11 +5261,15 @@ class ContentRouter(Transform):
                         route_counts["read_kompress_exp"] = (
                             route_counts.get("read_kompress_exp", 0) + 1
                         )
+                        if collect_diagnostics:
+                            _diag[i] = "compressed:read_kompress_exp"
                         continue
                     result_slots[i] = message
                     transforms_applied.append("router:read_protected")
                     route_counts.setdefault("read_protected", 0)
                     route_counts["read_protected"] += 1
+                    if collect_diagnostics:
+                        _diag[i] = "protected:read_output"
                     continue
 
             # Protection 1: Never compress user messages (unless overridden)
@@ -5252,6 +5277,8 @@ class ContentRouter(Transform):
                 result_slots[i] = message
                 transforms_applied.append("router:protected:user_message")
                 route_counts["user_msg"] += 1
+                if collect_diagnostics:
+                    _diag[i] = "protected:user_message"
                 continue
 
             # Protection 1b: Never compress system/developer messages unless
@@ -5261,12 +5288,16 @@ class ContentRouter(Transform):
                 transforms_applied.append(f"router:protected:{role}_message")
                 route_counts.setdefault("system_msg", 0)
                 route_counts["system_msg"] += 1
+                if collect_diagnostics:
+                    _diag[i] = f"protected:{role}_message"
                 continue
 
             if not content or tokenizer.count_text(content) < min_tokens:
                 # Skip small content
                 result_slots[i] = message
                 route_counts["small"] += 1
+                if collect_diagnostics:
+                    _diag[i] = "passthrough:small"
                 continue
 
             # Protection: failed tool calls / error outputs stay verbatim
@@ -5285,6 +5316,8 @@ class ContentRouter(Transform):
                 transforms_applied.append("router:protected:error_output")
                 route_counts.setdefault("error_protected", 0)
                 route_counts["error_protected"] += 1
+                if collect_diagnostics:
+                    _diag[i] = "protected:error_output"
                 continue
 
             # Detect content type for protection decisions. Even when the
@@ -5303,6 +5336,8 @@ class ContentRouter(Transform):
                 result_slots[i] = message
                 transforms_applied.append("router:protected:recent_code")
                 route_counts["recent_code"] += 1
+                if collect_diagnostics:
+                    _diag[i] = "protected:recent_code"
                 continue
 
             # Protection 3: Don't compress CODE when analysis intent detected
@@ -5310,6 +5345,8 @@ class ContentRouter(Transform):
                 result_slots[i] = message
                 transforms_applied.append("router:protected:analysis_context")
                 route_counts["analysis_ctx"] += 1
+                if collect_diagnostics:
+                    _diag[i] = "protected:analysis_context"
                 continue
 
             # Compression pinning: if this message was already compressed
@@ -5320,6 +5357,8 @@ class ContentRouter(Transform):
                 result_slots[i] = message
                 route_counts.setdefault("already_compressed", 0)
                 route_counts["already_compressed"] += 1
+                if collect_diagnostics:
+                    _diag[i] = "passthrough:already_compressed"
                 continue
 
             # Route and compress based on content detection
@@ -5347,6 +5386,8 @@ class ContentRouter(Transform):
                 route_counts["ratio_too_high"] += 1
                 route_counts.setdefault("cache_hit", 0)
                 route_counts["cache_hit"] += 1
+                if collect_diagnostics:
+                    _diag[i] = "passthrough:cache_skip"
                 continue
 
             # Tier 2: result cache — reuse compressed output
@@ -5386,10 +5427,14 @@ class ContentRouter(Transform):
                         # Net-cost gate: mutation would cost more in cache
                         # invalidation than it saves — leave untouched.
                         result_slots[i] = message
+                        if collect_diagnostics:
+                            _diag[i] = "passthrough:netcost_blocked"
                     else:
                         result_slots[i] = {**message, "content": cached_compressed}
                         transforms_applied.append(f"router:{cached_strategy}:{cached_ratio:.2f}")
                         compressed_details.append(f"{cached_strategy}:{cached_ratio:.2f}")
+                        if collect_diagnostics:
+                            _diag[i] = f"compressed:{cached_strategy}:{cached_ratio:.2f}"
                         # Freeze the "compress" verdict so future turns skip the
                         # min_ratio re-check above and never downgrade it.
                         if freeze_decision:
@@ -5411,6 +5456,8 @@ class ContentRouter(Transform):
                     self._cache.move_to_skip(content_key)
                     result_slots[i] = message
                     route_counts["ratio_too_high"] += 1
+                    if collect_diagnostics:
+                        _diag[i] = "passthrough:ratio_too_high"
                 route_counts.setdefault("cache_hit", 0)
                 route_counts["cache_hit"] += 1
                 continue
@@ -5535,6 +5582,8 @@ class ContentRouter(Transform):
                         route_counts["lossy_unrecoverable_skipped"] = (
                             route_counts.get("lossy_unrecoverable_skipped", 0) + 1
                         )
+                        if collect_diagnostics:
+                            _diag[slot_idx] = "passthrough:lossy_unrecoverable"
                         continue
                     # Compressed — store in result cache. The cache is still
                     # warmed when the net-cost gate blocks the slot: the
@@ -5562,12 +5611,18 @@ class ContentRouter(Transform):
                         write_multiplier=netcost_write_multiplier,
                     ):
                         result_slots[slot_idx] = message
+                        if collect_diagnostics:
+                            _diag[slot_idx] = "passthrough:netcost_blocked"
                         continue
                     result_slots[slot_idx] = {**message, "content": result.compressed}
                     transforms_applied.append(
                         f"router:{result.strategy_used.value}:{accept_ratio:.2f}"
                     )
                     compressed_details.append(f"{result.strategy_used.value}:{accept_ratio:.2f}")
+                    if collect_diagnostics:
+                        _diag[slot_idx] = (
+                            f"compressed:{result.strategy_used.value}:{accept_ratio:.2f}"
+                        )
                     if slot_idx in frozen_unlock_slots:
                         transforms_applied.append("router:netcost_frozen_unlock")
                         route_counts.setdefault("netcost_frozen_unlocked", 0)
@@ -5577,6 +5632,8 @@ class ContentRouter(Transform):
                     self._cache.mark_skip(content_key)
                     result_slots[slot_idx] = message
                     route_counts["ratio_too_high"] += 1
+                    if collect_diagnostics:
+                        _diag[slot_idx] = "passthrough:ratio_too_high"
                     # Caveat (1): only freeze a "skip" verdict when the ML model
                     # is actually ready. A passthrough caused purely by a still-
                     # loading ModernBERT must stay re-evaluable on later turns,
@@ -5689,6 +5746,32 @@ class ContentRouter(Transform):
             except Exception as e:  # pragma: no cover - defensive
                 logger.debug("Router observer raised (non-fatal): %s", e)
 
+        # Build per-message diagnostics when collect_diagnostics=True
+        message_decisions: list[MessageDecision] = []
+        if collect_diagnostics:
+            for idx, orig_msg in enumerate(messages):
+                slot = result_slots[idx] if idx < len(result_slots) else None
+                if slot is None:
+                    continue
+                orig_content = orig_msg.get("content", "")
+                result_content = slot.get("content", "")
+                tok_before = (
+                    tokenizer.count_text(orig_content) if isinstance(orig_content, str) else 0
+                )
+                tok_after = (
+                    tokenizer.count_text(result_content) if isinstance(result_content, str) else 0
+                )
+                action = _diag.get(idx, "compressed:unknown")
+                message_decisions.append(
+                    MessageDecision(
+                        message_index=idx,
+                        role=orig_msg.get("role", ""),
+                        tokens_before=tok_before,
+                        tokens_after=tok_after,
+                        action=action,
+                    )
+                )
+
         all_transforms = lifecycle_transforms + transforms_applied
         return TransformResult(
             messages=transformed_messages,
@@ -5698,6 +5781,7 @@ class ContentRouter(Transform):
             markers_inserted=lifecycle_ccr_hashes,
             warnings=warnings,
             timing=compressor_timing,
+            message_decisions=message_decisions,
         )
 
     def _lossless_compact_excluded(self, content: Any) -> tuple[str, str] | None:
diff --git a/headroom/transforms/pipeline.py b/headroom/transforms/pipeline.py
index 305e888f6..3348a5b7e 100644
--- a/headroom/transforms/pipeline.py
+++ b/headroom/transforms/pipeline.py
@@ -317,6 +317,7 @@ class TransformPipeline:
             all_markers: list[str] = []
             all_warnings: list[str] = []
             all_timing: dict[str, float] = {}  # transform_name → ms
+            all_message_decisions: list[Any] = []
 
             # Track transform diffs if enabled
             transform_diffs: list[TransformDiff] = []
@@ -401,6 +402,8 @@ class TransformPipeline:
                     all_markers.extend(result.markers_inserted)
                     all_warnings.extend(result.warnings)
                     all_timing[transform.name] = duration_ms
+                    if result.message_decisions:
+                        all_message_decisions.extend(result.message_decisions)
 
                     # Merge sub-transform timing (e.g. ContentRouter's per-compressor breakdown)
                     if result.timing:
@@ -543,6 +546,7 @@ class TransformPipeline:
             diff_artifact=diff_artifact,
             timing=all_timing,
             waste_signals=waste_signals,
+            message_decisions=all_message_decisions,
         )
 
     def simulate(
diff --git a/tests/test_compress_diagnostics.py b/tests/test_compress_diagnostics.py
new file mode 100644
index 000000000..bb8dda521
--- /dev/null
+++ b/tests/test_compress_diagnostics.py
@@ -0,0 +1,303 @@
+"""Tests for per-message compression diagnostics (issue #2855).
+
+Verifies that CompressConfig(diagnostics=True) — and the HEADROOM_DIAGNOSTICS=1
+env var — populate CompressResult.diagnostics with per-message MessageDecision
+objects, and that the compress() function correctly threads collect_diagnostics
+through to the pipeline and collects the results.
+
+ContentRouter integration tests (which need the compiled headroom._core extension)
+run in CI; this file tests the compress() ↔ pipeline contract via a mock pipeline.
+"""
+
+from __future__ import annotations
+
+import importlib
+import os
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from headroom.compress import CompressConfig, compress
+from headroom.config import MessageDecision, TransformResult
+
+# ---------------------------------------------------------------------------
+# Mock pipeline factory
+# ---------------------------------------------------------------------------
+
+
+def _noop_result(messages: list[dict]) -> TransformResult:
+    """TransformResult that passes messages through unchanged, no decisions."""
+    tokens = sum(len(str(m.get("content", ""))) for m in messages)
+    return TransformResult(
+        messages=messages,
+        tokens_before=tokens,
+        tokens_after=tokens,
+        transforms_applied=["router:noop"],
+    )
+
+
+def _result_with_decisions(
+    messages: list[dict], decisions: list[MessageDecision]
+) -> TransformResult:
+    """TransformResult that includes per-message decisions."""
+    tokens = sum(len(str(m.get("content", ""))) for m in messages)
+    return TransformResult(
+        messages=messages,
+        tokens_before=tokens,
+        tokens_after=tokens,
+        transforms_applied=["router:noop"],
+        message_decisions=decisions,
+    )
+
+
+class _MockPipeline:
+    """Pipeline stub that captures kwargs and returns a configurable result."""
+
+    def __init__(self, decisions: list[MessageDecision] | None = None):
+        self.last_kwargs: dict = {}
+        self._decisions = decisions or []
+
+    def apply(self, messages, model, **kwargs):
+        self.last_kwargs = kwargs
+        return _result_with_decisions(messages, self._decisions)
+
+
+def _fake_otel():
+    return SimpleNamespace(record_compression_failure=lambda **kw: None)
+
+
+# ---------------------------------------------------------------------------
+# MessageDecision dataclass
+# ---------------------------------------------------------------------------
+
+
+def test_message_decision_dataclass():
+    """MessageDecision is a dataclass with the expected fields."""
+    dec = MessageDecision(
+        message_index=0,
+        role="user",
+        tokens_before=100,
+        tokens_after=60,
+        action="compressed:kompress:0.60",
+    )
+    assert dec.message_index == 0
+    assert dec.role == "user"
+    assert dec.tokens_before == 100
+    assert dec.tokens_after == 60
+    assert dec.action == "compressed:kompress:0.60"
+
+
+def test_message_decision_importable_from_headroom():
+    """MessageDecision is accessible via the headroom top-level package."""
+    import headroom
+
+    assert hasattr(headroom, "MessageDecision")
+    cls = headroom.MessageDecision
+    dec = cls(
+        message_index=1,
+        role="assistant",
+        tokens_before=50,
+        tokens_after=50,
+        action="passthrough:small",
+    )
+    assert dec.message_index == 1
+
+
+# ---------------------------------------------------------------------------
+# TransformResult carries message_decisions
+# ---------------------------------------------------------------------------
+
+
+def test_transform_result_has_message_decisions_field():
+    """TransformResult has a message_decisions field defaulting to []."""
+    tr = TransformResult(
+        messages=[],
+        tokens_before=0,
+        tokens_after=0,
+        transforms_applied=[],
+    )
+    assert hasattr(tr, "message_decisions")
+    assert tr.message_decisions == []
+
+
+def test_transform_result_message_decisions_populated():
+    """TransformResult stores MessageDecision objects when provided."""
+    dec = MessageDecision(
+        message_index=0, role="user", tokens_before=10, tokens_after=10, action="passthrough:small"
+    )
+    tr = TransformResult(
+        messages=[],
+        tokens_before=0,
+        tokens_after=0,
+        transforms_applied=[],
+        message_decisions=[dec],
+    )
+    assert len(tr.message_decisions) == 1
+    assert tr.message_decisions[0] is dec
+
+
+# ---------------------------------------------------------------------------
+# CompressConfig.diagnostics field
+# ---------------------------------------------------------------------------
+
+
+def test_compress_config_diagnostics_defaults_false():
+    """CompressConfig.diagnostics is False by default."""
+    cfg = CompressConfig()
+    assert cfg.diagnostics is False
+
+
+def test_compress_config_diagnostics_can_be_set():
+    """CompressConfig(diagnostics=True) stores True."""
+    cfg = CompressConfig(diagnostics=True)
+    assert cfg.diagnostics is True
+
+
+# ---------------------------------------------------------------------------
+# compress() does NOT collect diagnostics by default
+# ---------------------------------------------------------------------------
+
+
+def test_compress_result_diagnostics_none_by_default(monkeypatch):
+    """CompressResult.diagnostics is None when diagnostics is not requested."""
+    compress_module = importlib.import_module("headroom.compress")
+    mock_pipeline = _MockPipeline()
+    monkeypatch.setattr(compress_module, "_get_pipeline", lambda: mock_pipeline)
+    monkeypatch.setattr(compress_module, "get_otel_metrics", _fake_otel)
+
+    messages = [{"role": "user", "content": "hello"}]
+    result = compress(messages, model="gpt-4o")
+
+    assert result.diagnostics is None
+    assert mock_pipeline.last_kwargs.get("collect_diagnostics") is False
+
+
+# ---------------------------------------------------------------------------
+# compress() passes collect_diagnostics when CompressConfig.diagnostics=True
+# ---------------------------------------------------------------------------
+
+
+def test_compress_passes_collect_diagnostics_when_config_true(monkeypatch):
+    """compress() passes collect_diagnostics=True to pipeline.apply() when config.diagnostics=True."""
+    compress_module = importlib.import_module("headroom.compress")
+    mock_pipeline = _MockPipeline()
+    monkeypatch.setattr(compress_module, "_get_pipeline", lambda: mock_pipeline)
+    monkeypatch.setattr(compress_module, "get_otel_metrics", _fake_otel)
+
+    messages = [{"role": "user", "content": "hello"}]
+    cfg = CompressConfig(diagnostics=True)
+    compress(messages, model="gpt-4o", config=cfg)
+
+    assert mock_pipeline.last_kwargs.get("collect_diagnostics") is True
+
+
+def test_compress_result_diagnostics_list_when_config_true(monkeypatch):
+    """CompressResult.diagnostics is a list (possibly empty) when config.diagnostics=True."""
+    compress_module = importlib.import_module("headroom.compress")
+    mock_pipeline = _MockPipeline(decisions=[])
+    monkeypatch.setattr(compress_module, "_get_pipeline", lambda: mock_pipeline)
+    monkeypatch.setattr(compress_module, "get_otel_metrics", _fake_otel)
+
+    messages = [{"role": "user", "content": "hello"}]
+    cfg = CompressConfig(diagnostics=True)
+    result = compress(messages, model="gpt-4o", config=cfg)
+
+    assert result.diagnostics is not None
+    assert isinstance(result.diagnostics, list)
+
+
+def test_compress_result_diagnostics_contains_decisions(monkeypatch):
+    """CompressResult.diagnostics contains the MessageDecision objects from the pipeline."""
+    compress_module = importlib.import_module("headroom.compress")
+    expected = [
+        MessageDecision(
+            message_index=0,
+            role="user",
+            tokens_before=10,
+            tokens_after=10,
+            action="protected:user_message",
+        ),
+        MessageDecision(
+            message_index=1,
+            role="tool",
+            tokens_before=200,
+            tokens_after=80,
+            action="compressed:kompress:0.40",
+        ),
+    ]
+    mock_pipeline = _MockPipeline(decisions=expected)
+    monkeypatch.setattr(compress_module, "_get_pipeline", lambda: mock_pipeline)
+    monkeypatch.setattr(compress_module, "get_otel_metrics", _fake_otel)
+
+    messages = [
+        {"role": "user", "content": "x" * 40},
+        {"role": "tool", "tool_call_id": "t1", "content": "y" * 800},
+    ]
+    cfg = CompressConfig(diagnostics=True)
+    result = compress(messages, model="gpt-4o", config=cfg)
+
+    assert result.diagnostics == expected
+
+
+# ---------------------------------------------------------------------------
+# HEADROOM_DIAGNOSTICS=1 env var
+# ---------------------------------------------------------------------------
+
+
+def test_env_var_enables_diagnostics(monkeypatch):
+    """HEADROOM_DIAGNOSTICS=1 enables collect_diagnostics without config flag."""
+    compress_module = importlib.import_module("headroom.compress")
+    mock_pipeline = _MockPipeline()
+    monkeypatch.setattr(compress_module, "_get_pipeline", lambda: mock_pipeline)
+    monkeypatch.setattr(compress_module, "get_otel_metrics", _fake_otel)
+
+    messages = [{"role": "user", "content": "hello"}]
+    with patch.dict(os.environ, {"HEADROOM_DIAGNOSTICS": "1"}):
+        result = compress(messages, model="gpt-4o")
+
+    assert mock_pipeline.last_kwargs.get("collect_diagnostics") is True
+    assert result.diagnostics is not None
+
+
+def test_env_var_absent_keeps_diagnostics_none(monkeypatch):
+    """When HEADROOM_DIAGNOSTICS is unset, result.diagnostics stays None."""
+    compress_module = importlib.import_module("headroom.compress")
+    mock_pipeline = _MockPipeline()
+    monkeypatch.setattr(compress_module, "_get_pipeline", lambda: mock_pipeline)
+    monkeypatch.setattr(compress_module, "get_otel_metrics", _fake_otel)
+
+    messages = [{"role": "user", "content": "hello"}]
+    env = {k: v for k, v in os.environ.items() if k != "HEADROOM_DIAGNOSTICS"}
+    with patch.dict(os.environ, env, clear=True):
+        result = compress(messages, model="gpt-4o")
+
+    assert result.diagnostics is None
+
+
+def test_env_var_overrides_config_false(monkeypatch):
+    """HEADROOM_DIAGNOSTICS=1 enables diagnostics even when CompressConfig.diagnostics=False."""
+    compress_module = importlib.import_module("headroom.compress")
+    mock_pipeline = _MockPipeline()
+    monkeypatch.setattr(compress_module, "_get_pipeline", lambda: mock_pipeline)
+    monkeypatch.setattr(compress_module, "get_otel_metrics", _fake_otel)
+
+    messages = [{"role": "user", "content": "hello"}]
+    cfg = CompressConfig(diagnostics=False)
+    with patch.dict(os.environ, {"HEADROOM_DIAGNOSTICS": "1"}):
+        result = compress(messages, model="gpt-4o", config=cfg)
+
+    assert result.diagnostics is not None
+    assert mock_pipeline.last_kwargs.get("collect_diagnostics") is True
+
+
+# ---------------------------------------------------------------------------
+# diagnostics=None when optimization is disabled
+# ---------------------------------------------------------------------------
+
+
+def test_no_diagnostics_when_optimize_false():
+    """compress(optimize=False) returns empty CompressResult without calling pipeline."""
+    cfg = CompressConfig(diagnostics=True)
+    messages = [{"role": "user", "content": "hello"}]
+    result = compress(messages, model="gpt-4o", optimize=False, config=cfg)
+    # optimize=False early-returns before calling the pipeline
+    assert result.diagnostics is None
diff --git a/tests/test_config.py b/tests/test_config.py
index 403ec386f..65d0835c0 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -336,6 +336,7 @@ class TestTransformResult:
             "cache_metrics",
             "timing",
             "waste_signals",
+            "message_decisions",
         }
         assert field_names == expected_fields