feat(pipeline): say whose request an OutcomeSnapshot is

headroom/pipeline.py

@@ -5,9 +5,10 @@
 import importlib.metadata
 import logging
 import os
-from collections.abc import Iterable
+from collections.abc import Iterable, Mapping
 from dataclasses import dataclass, field
 from enum import Enum
+from types import MappingProxyType
 from typing import Any, Protocol
 
 log = logging.getLogger(__name__)
@@ -86,6 +87,21 @@ class OutcomeSnapshot:
     #: is re-sent and recompressed every turn, and ``tokens_saved -
     #: novel_tokens_saved`` is the re-sent history removed again.
     novel_tokens_saved: int = 0
+    #: Whose request this was: ``{dimension: label}``, for now ``project`` and
+    #: ``client`` when known. Self-reported labels, not authenticated
+    #: identity. Read-only.
+    attribution: Mapping[str, str] = field(default_factory=dict)
+    #: Stable across every turn of one conversation, when the handler knows it.
+    conversation_key: str | None = None
+    #: Pre-compression input size on the same tokenizer as ``tokens_saved``.
+    original_tokens: int = 0
+    #: Estimated output tokens avoided by output shaping on this request.
+    output_tokens_saved: int = 0
+
+    def __post_init__(self) -> None:
+        # Frozen fields can still hold a mutable dict; freeze it so an
+        # extension cannot rewrite the labels another extension will read.
+        object.__setattr__(self, "attribution", MappingProxyType(dict(self.attribution or {})))
 
     @property
     def truncated(self) -> bool:
@@ -247,6 +263,17 @@ def __init__(
     def enabled(self) -> bool:
         return bool(self._extensions)
 
+    def add(self, extension: Any) -> None:
+        """Attach an extension after construction.
+
+        For a proxy extension (``headroom.proxy_extension``) that also needs
+        lifecycle events: it is installed after the proxy and its manager
+        exist, so it cannot be passed to the constructor. Adding the same
+        object twice is a no-op.
+        """
+        if all(existing is not extension for existing in self._extensions):
+            self._extensions.append(extension)
+
     def emit(
         self,
         stage: PipelineStage,

headroom/proxy/outcome.py

@@ -665,6 +665,14 @@ def _record_and_estimate() -> int:
                     transforms_applied=tuple(str(t) for t in (outcome.transforms_applied or ())),
                     tokens_saved=outcome.tokens_saved,
                     novel_tokens_saved=novel_tokens_saved,
+                    attribution={
+                        dimension: label
+                        for dimension, label in (("project", project), ("client", outcome.client))
+                        if label
+                    },
+                    conversation_key=outcome.conversation_key,
+                    original_tokens=outcome.original_tokens,
+                    output_tokens_saved=output_tokens_saved_est,
                 ),
             )
         except Exception:  # noqa: BLE001 - telemetry must never break a response

tests/test_pipeline_outcome_stage.py

@@ -99,6 +99,51 @@ def on_pipeline_event(self, event):
         event = mgr.emit(PipelineStage.OUTCOME_OBSERVED, operation="proxy.outcome", outcome=snap)
         assert event.outcome is snap
 
+    def test_snapshot_carries_attribution_and_conversation(self):
+        snap = OutcomeSnapshot(
+            attribution={"project": "web", "client": "codex"},
+            conversation_key="conv-a",
+            original_tokens=1_000,
+            output_tokens_saved=40,
+        )
+        assert dict(snap.attribution) == {"project": "web", "client": "codex"}
+        assert (snap.conversation_key, snap.original_tokens, snap.output_tokens_saved) == (
+            "conv-a",
+            1_000,
+            40,
+        )
+
+    def test_attribution_labels_are_read_only(self):
+        """Frozen fields can still hold a mutable dict; one extension must not
+        be able to rewrite the labels the next one reads."""
+        labels = {"project": "web"}
+        snap = OutcomeSnapshot(attribution=labels)
+        with pytest.raises(TypeError):
+            snap.attribution["project"] = "other"  # type: ignore[index]
+        labels["project"] = "changed-after"
+        assert snap.attribution["project"] == "web"
+
+
+class TestAdd:
+    def test_an_extension_added_later_receives_events(self):
+        mgr = PipelineExtensionManager(extensions=[], discover=False)
+        rec = _Recorder()
+        mgr.add(rec)
+        assert mgr.enabled
+        mgr.emit(
+            PipelineStage.OUTCOME_OBSERVED, operation="proxy.outcome", outcome=OutcomeSnapshot()
+        )
+        assert rec.stages() == [PipelineStage.OUTCOME_OBSERVED]
+
+    def test_adding_the_same_extension_twice_is_a_no_op(self):
+        rec = _Recorder()
+        mgr = PipelineExtensionManager(extensions=[rec], discover=False)
+        mgr.add(rec)
+        mgr.emit(
+            PipelineStage.OUTCOME_OBSERVED, operation="proxy.outcome", outcome=OutcomeSnapshot()
+        )
+        assert len(rec.seen) == 1
+
 
 class TestDiscoveryGate:
     """Installing a package must not silently start rewriting live requests."""

tests/test_proxy/test_conversation_savings.py

@@ -393,3 +393,27 @@ def on_pipeline_event(self, event: Any) -> None:
         (62_806, 62_806),
         (76_020, 13_214),
     ]
+
+
+@pytest.mark.asyncio
+async def test_the_snapshot_says_whose_request_it_was() -> None:
+    """Project, client, conversation and the pre-compression size reach the
+    extension, so it can attribute what it measures without patching core."""
+    from headroom.pipeline import PipelineExtensionManager, PipelineStage
+
+    seen: list[Any] = []
+
+    class _Recorder:
+        def on_pipeline_event(self, event: Any) -> None:
+            if event.stage is PipelineStage.OUTCOME_OBSERVED:
+                seen.append(event.outcome)
+
+    handler = _Handler()
+    handler.pipeline_extensions = PipelineExtensionManager(extensions=[_Recorder()], discover=False)
+    await emit_request_outcome(handler, _responses_outcome(project="web", client="codex"))
+    await emit_request_outcome(handler, _responses_outcome(request_id="req-2"))
+
+    assert dict(seen[0].attribution) == {"project": "web", "client": "codex"}
+    assert seen[0].conversation_key == "conv-a"
+    assert seen[0].original_tokens == 156_260
+    assert dict(seen[1].attribution) == {}