From da451385392578d4d9b98cab6c7e39f2c2502891 Mon Sep 17 00:00:00 2001
From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Date: Wed, 2 Sep 2026 16:46:07 -0700
Subject: [PATCH 1/2] fix(session-naming): never call a provider this session
 did not select

hooks-session-naming picked its own provider and could silently borrow an
arbitrary one. Measured blast radius (model_performance-egh): 321 foreign
llm:response events across 12 evaluation capture roots, 158 of 549 root
sessions -- Anthropic-pinned cells emitting openai calls.

Two sites, one defect:

  __init__.py:482-504 (pre-fix) resolved model_role="fast" through the
  routing matrix, whose default matrix is openai, then matched the resolved
  name against mounted provider keys by substring. In an Anthropic-pinned
  cell that match fails.

  __init__.py:511-512 (pre-fix) then did
      fallback_key = next(iter(providers), None)
  -- an order-dependent, SILENT borrow of whichever provider instance
  happened to be first in the mount dict. That line is the leak.

Now:

  * _select_session_provider() returns the provider actually answering this
    session: the conversation.provider_pin pin when set, else the same
    priority ordering the streaming orchestrator uses (provider.priority,
    then config["priority"], default 100, ties by mount order). There is no
    next(iter(providers.values())) anywhere in this module.
  * A model_role candidate is honoured only when it is mounted here AND
    shares the session provider's get_info().id vendor. Same-vendor siblings
    (anthropic-sonnet -> anthropic-haiku) stay allowed; that is the intended
    cheap-model routing. Anything else is REFUSED with a WARNING naming both
    the refused provider and the one actually used, once per session.
  * An unprovable vendor (get_info missing or raising) fails closed --
    refused, not borrowed.
  * A pin whose provider is no longer mounted skips naming for that turn
    rather than answering on a provider the user never chose.

Seven regression tests in TestProviderPurity; six of them fail against the
pre-fix module. Default behaviour is otherwise unchanged: all 26 pre-existing
tests pass untouched.

Refs: model_performance-dgf, model_performance-egh
---
 modules/hooks-session-naming/README.md        |  40 ++-
 .../__init__.py                               | 300 ++++++++++++++----
 .../tests/test_session_naming.py              | 275 +++++++++++++++-
 3 files changed, 553 insertions(+), 62 deletions(-)

diff --git a/modules/hooks-session-naming/README.md b/modules/hooks-session-naming/README.md
index f74d6220..3ba6c644 100644
--- a/modules/hooks-session-naming/README.md
+++ b/modules/hooks-session-naming/README.md
@@ -41,24 +41,48 @@ hooks:
 matrix — the same mechanism used by the `delegate` tool and recipe agent steps.
 Session naming is a simple classification task; it does not need the priority model.
 
+**Naming never calls a provider this session did not select.** Every path below
+ends on either the session's own conversation provider or a *same-vendor* sibling
+of it. There is no arbitrary fallback.
+
 Resolution order:
 
 1. **`model_role`** — Resolved against the `model_role_resolver` capability
    (registered by whichever routing bundle is active — typically the
    matrix-based one shipped in `amplifier-bundle-routing-matrix`). Defaults to
-   `"fast"`.
-
-2. **Fallback** — `next(iter(providers.values()))` — the first/priority provider.
-   Used when `model_role` is `None`, or when resolution fails.
+   `"fast"`. A resolved candidate is **accepted only if** it is mounted in this
+   session **and** its `get_info().id` matches the vendor of the session's own
+   provider. Anything else is refused with a WARNING (once per session).
+
+2. **The session's own conversation provider** — the `conversation.provider_pin`
+   pin when one is set, otherwise the same priority ordering the streaming
+   orchestrator uses to pick the conversation provider (`provider.priority`,
+   then `provider.config["priority"]`, default 100, ties broken by mount order).
+   No model override is applied on this path.
+
+If the conversation is pinned to a provider that is no longer mounted, naming is
+**skipped** for that turn rather than run on some other provider.
+
+### Why the vendor check exists
+
+Session naming used to resolve `model_role` through the routing matrix (whose
+default matrix is openai) and, when the resolved name matched no mount, fall
+through to `next(iter(providers.values()))` — an order-dependent, silent borrow
+of whichever provider instance happened to be first in the mount dict. In an
+Anthropic-pinned evaluation cell that emitted openai calls into the session's
+event stream (321 foreign responses across 12 capture roots; see
+`model_performance-egh`). Same-vendor siblings (`anthropic-sonnet` →
+`anthropic-haiku`) remain allowed: that is the intended cheap-model routing.
 
 ### Optional Dependency: hooks-routing
 
 `hooks-routing` is an **optional runtime dependency**. The module degrades gracefully:
 
-- If `hooks-routing` is not installed, the module silently falls back to the priority
-  provider. No warning is emitted — falling back is the expected behaviour when the
-  routing module is absent.
-- To disable routing explicitly and always use the priority provider, set `model_role: null`.
+- If `hooks-routing` is not installed, the module falls back to the session's own
+  conversation provider (debug-logged). Falling back is the expected behaviour when
+  the routing module is absent.
+- To disable routing explicitly and always use the session's own provider, set
+  `model_role: null`.
 
 ## Async Behavior
 
diff --git a/modules/hooks-session-naming/amplifier_module_hooks_session_naming/__init__.py b/modules/hooks-session-naming/amplifier_module_hooks_session_naming/__init__.py
index fb7a2707..8a9c4510 100644
--- a/modules/hooks-session-naming/amplifier_module_hooks_session_naming/__init__.py
+++ b/modules/hooks-session-naming/amplifier_module_hooks_session_naming/__init__.py
@@ -26,7 +26,11 @@ class SessionNamingConfig:
     model_role routes naming to a cheap/fast model via the routing matrix.
     Defaults to "fast" — session naming is a simple classification task that
     does not need the priority/expensive model. Set to None to use the
-    priority provider explicitly.
+    session's own conversation provider explicitly.
+
+    Whatever model_role resolves to, naming only ever calls the session's own
+    conversation provider or a same-vendor sibling of it (see
+    ``SessionNamingHook._call_provider``).
     """
 
     initial_trigger_turn: int = 2
@@ -109,6 +113,9 @@ def __init__(self, coordinator: Any, config: SessionNamingConfig):
         # session, so without this a stable config gap would re-emit the
         # identical warning on every retry.
         self._role_fallback_warned: set[str] = set()
+        # Same dedup, for the "model_role resolved to a provider this session
+        # never selected — refusing to borrow it" WARNING.
+        self._cross_provider_refused: set[str] = set()
 
     async def on_orchestrator_complete(
         self, event: str, data: dict[str, Any]
@@ -460,19 +467,167 @@ def _truncate_content(self, content: str, max_len: int) -> str:
             truncated = truncated[:last_space]
         return truncated + "..."
 
+    @staticmethod
+    def _priority_of(provider: Any) -> float:
+        """Selection priority for one provider (lower wins, default 100).
+
+        Mirrors the streaming orchestrator's own rule (``provider.priority``,
+        then ``provider.config["priority"]``, then 100) so that the provider
+        this module picks for an unpinned session is *the same one answering
+        the conversation*, not an independent guess. Non-numeric values (a
+        test double's auto-attribute, a misconfigured string) are ignored
+        rather than crashing the comparison.
+        """
+        candidates = [getattr(provider, "priority", None)]
+        config = getattr(provider, "config", None)
+        if isinstance(config, dict):
+            candidates.append(config.get("priority"))
+        for value in candidates:
+            if isinstance(value, bool):
+                continue
+            if isinstance(value, (int, float)):
+                return float(value)
+        return 100.0
+
+    @staticmethod
+    def _vendor_of(provider: Any) -> str | None:
+        """Vendor identity of a provider via the kernel contract
+        ``get_info().id`` (e.g. ``"anthropic"``), lowercased.
+
+        Returns None when the vendor cannot be established — callers must
+        treat that as "cannot prove same vendor" and refuse, never as "no
+        conflict". Two mount names sharing an id (anthropic-sonnet /
+        anthropic-haiku) are the SAME vendor.
+        """
+        get_info = getattr(provider, "get_info", None)
+        if not callable(get_info):
+            return None
+        try:
+            info = get_info()
+        except Exception as e:  # pragma: no cover - defensive
+            logger.debug("get_info() failed while checking provider vendor: %s", e)
+            return None
+        vendor = getattr(info, "id", None)
+        if vendor is None and isinstance(info, dict):
+            vendor = info.get("id")
+        if isinstance(vendor, str) and vendor.strip():
+            return vendor.strip().lower()
+        return None
+
+    def _same_vendor(self, a: Any, b: Any) -> bool:
+        """True only when both vendors are known AND equal (fail closed)."""
+        if a is b:
+            return True
+        vendor_a = self._vendor_of(a)
+        vendor_b = self._vendor_of(b)
+        return bool(vendor_a and vendor_b and vendor_a == vendor_b)
+
+    def _select_session_provider(
+        self, providers: dict[str, Any]
+    ) -> tuple[str | None, Any | None]:
+        """The provider answering THIS session — never an arbitrary one.
+
+        1. The conversation-scope pin, when the ``conversation.provider_pin``
+           capability reports one. A pin naming a provider that is no longer
+           mounted returns ``(None, None)``: refuse, never fall through to
+           another provider the user did not choose.
+        2. Otherwise priority ordering, identical to the orchestrator's rule,
+           with insertion order breaking ties — so the result *is* the
+           session's own conversation provider rather than
+           ``next(iter(providers.values()))`` reached by coincidence.
+        """
+        get_capability = getattr(self.coordinator, "get_capability", None)
+        pinned: str | None = None
+        if callable(get_capability):
+            try:
+                pin_capability = get_capability("conversation.provider_pin")
+            except Exception as e:  # pragma: no cover - defensive
+                logger.debug("conversation.provider_pin lookup failed: %s", e)
+                pin_capability = None
+            current = getattr(pin_capability, "current", None)
+            if callable(current):
+                try:
+                    name = current()
+                except Exception as e:  # pragma: no cover - defensive
+                    logger.debug("conversation.provider_pin.current() failed: %s", e)
+                    name = None
+                if isinstance(name, str) and name:
+                    pinned = name
+
+        if pinned is not None:
+            provider = providers.get(pinned)
+            if provider is None:
+                logger.warning(
+                    "This conversation is pinned to provider %r, which is no"
+                    " longer mounted. Skipping session naming rather than"
+                    " naming on a provider this session never pinned.",
+                    pinned,
+                )
+                return None, None
+            return pinned, provider
+
+        ranked = [
+            (self._priority_of(provider), index, name, provider)
+            for index, (name, provider) in enumerate(providers.items())
+        ]
+        if not ranked:
+            return None, None
+        ranked.sort(key=lambda entry: (entry[0], entry[1]))
+        _, _, name, provider = ranked[0]
+        return name, provider
+
+    @staticmethod
+    def _match_resolved_provider(
+        providers: dict[str, Any], resolved_name: str
+    ) -> tuple[str | None, Any | None]:
+        """Mounted provider whose mount name contains the resolved name."""
+        if not isinstance(resolved_name, str) or not resolved_name:
+            return None, None
+        needle = resolved_name.lower()
+        for key, provider in providers.items():
+            if needle in key.lower():
+                return key, provider
+        return None, None
+
+    def _warn_once(self, session_id: str | None, seen: set[str], *args: Any) -> None:
+        """WARNING the first time per session, DEBUG on every repeat.
+
+        Naming retries every few turns for the life of a session, so a stable
+        configuration gap would otherwise re-emit the identical warning
+        forever.
+        """
+        warn_key = session_id or ""
+        if warn_key not in seen:
+            seen.add(warn_key)
+            logger.warning(*args)
+        else:
+            logger.debug(*args)
+
     async def _call_provider(
         self, prompt: str, session_id: str | None = None
     ) -> str | None:
         """Call the LLM provider to generate name/description.
 
+        THE PROVIDER IS NEVER ARBITRARY. Every path lands on either the
+        session's own conversation provider or a same-vendor sibling of it;
+        there is no ``next(iter(providers.values()))`` here. A session pinned
+        to one provider can never emit a naming call on another vendor: the
+        historical bug was that ``model_role`` resolved through the routing
+        matrix (whose default matrix is openai) and, failing to match a mount,
+        fell through to whichever provider instance happened to be first in
+        the mount dict — an order-dependent, silent cross-provider borrow.
+
         Resolution order (highest to lowest priority):
-          1. model_role — resolved via routing matrix (lazy import)
-          2. Fallback   — next(iter(providers.values()))
+          1. model_role — resolved via the ``model_role_resolver`` capability,
+             ACCEPTED ONLY IF the resolved provider is mounted here and is the
+             same vendor as the session's own provider.
+          2. The session's own conversation provider (pin, else priority
+             order), with no model override.
 
-        model_role resolution requires amplifier_module_hooks_routing. When that
-        module is not installed (no model_role_resolver capability registered
-        at all), logs a debug message and falls back to #2 — that fallback is
-        legitimate and intended.
+        model_role resolution requires a routing bundle. When none is
+        installed (no model_role_resolver capability registered at all), logs
+        a debug message and falls back to #2 — that fallback is legitimate
+        and intended.
 
         When a model_role_resolver IS registered and resolution itself raises
         (e.g. a transient provider API hiccup while listing models), the
@@ -489,11 +644,15 @@ async def _call_provider(
         transient error — retrying later changes nothing. Skipping silently
         in that case means session naming is a feature that quietly never
         runs, with only a log line nobody reads to explain why. So this case
-        falls back to #2 (the session's own default/priority provider)
-        instead of skipping, and logs a WARNING naming the unresolved role
-        and the provider substituted for it — once per session (via
+        falls back to #2 and logs a WARNING naming the unresolved role and
+        the provider substituted for it — once per session (via
         ``session_id``), since naming retries every few turns and repeating
         the identical warning on every retry would just be noise.
+
+        A resolved candidate that is NOT mounted here, or that belongs to a
+        different vendor than the session's own provider, is REFUSED the same
+        loud way: warn once, then name on the session's own provider with no
+        model override.
         """
         try:
             providers = self.coordinator.get("providers")
@@ -501,10 +660,20 @@ async def _call_provider(
                 logger.warning("No provider available for session naming")
                 return None
 
-            # Resolution order: model_role > priority provider
+            session_provider_name, session_provider = self._select_session_provider(
+                providers
+            )
+            if session_provider is None:
+                # _select_session_provider already logged the specific cause.
+                logger.debug("No session provider resolved for session naming")
+                return None
+
+            # Resolution order: model_role (same vendor only) > session provider
             provider = None
+            provider_name: str | None = None
             model_override: str | None = None
             role_had_no_candidates = False
+            refusal: tuple[str, str] | None = None
 
             if self.config.model_role:
                 # Look up the model_role_resolver capability registered by
@@ -519,7 +688,7 @@ async def _call_provider(
                 if resolver is None:
                     logger.debug(
                         "model_role %r set but no model_role_resolver capability"
-                        " registered, falling back to priority provider",
+                        " registered, falling back to the session's own provider",
                         self.config.model_role,
                     )
                 else:
@@ -538,49 +707,70 @@ async def _call_provider(
                     if resolved:
                         # ProviderPreference attrs: .provider, .model, .config
                         resolved_provider_name = resolved[0].provider
-                        model_override = resolved[0].model
-                        # Find the provider whose key contains the resolved name
-                        for key, p in providers.items():
-                            if resolved_provider_name.lower() in key.lower():
-                                provider = p
-                                break
+                        candidate_name, candidate = self._match_resolved_provider(
+                            providers, resolved_provider_name
+                        )
+                        if candidate is None:
+                            refusal = (
+                                str(resolved_provider_name),
+                                "no provider with that name is mounted in this session",
+                            )
+                        elif self._same_vendor(candidate, session_provider):
+                            provider = candidate
+                            provider_name = candidate_name
+                            model_override = resolved[0].model
+                        else:
+                            refusal = (
+                                str(candidate_name),
+                                "it is a different provider vendor than the one"
+                                " answering this session",
+                            )
                     else:
                         role_had_no_candidates = True
 
-            # Fallback: use first/priority provider. Reached when model_role
-            # is unset, no resolver capability is registered, OR the role
-            # resolved to no candidates (role_had_no_candidates, handled
-            # below with a loud warning instead of a silent substitution).
+            # Fall back to the session's OWN provider. Reached when model_role
+            # is unset, no resolver capability is registered, the role resolved
+            # to no candidates, or the resolved candidate was refused as
+            # foreign — the last two are announced loudly below rather than
+            # substituted silently.
             if provider is None:
-                fallback_key = next(iter(providers), None)
-                provider = providers.get(fallback_key) if fallback_key else None
-
-                if role_had_no_candidates and provider is not None:
-                    warn_key = session_id or ""
-                    if warn_key not in self._role_fallback_warned:
-                        self._role_fallback_warned.add(warn_key)
-                        logger.warning(
-                            "model_role %r resolved to no candidates; session"
-                            " naming is falling back to provider %r (the"
-                            " session's own default) instead of skipping."
-                            " This uses whatever model that provider is"
-                            " already configured with, which may be more"
-                            " expensive than intended — configure a %r"
-                            " candidate in the routing matrix to route naming"
-                            " to a cheap model instead. (Further occurrences"
-                            " this session are logged at DEBUG.)",
-                            self.config.model_role,
-                            fallback_key,
-                            self.config.model_role,
-                        )
-                    else:
-                        logger.debug(
-                            "model_role %r again resolved to no candidates;"
-                            " reusing fallback provider %r (already warned"
-                            " once this session)",
-                            self.config.model_role,
-                            fallback_key,
-                        )
+                provider = session_provider
+                provider_name = session_provider_name
+                model_override = None
+
+                if refusal is not None:
+                    refused_name, reason = refusal
+                    self._warn_once(
+                        session_id,
+                        self._cross_provider_refused,
+                        "model_role %r resolved to provider %r, but %s."
+                        " REFUSING to borrow it: session naming will run on"
+                        " %r, the provider answering this session. (Naming"
+                        " must never issue a call on a provider this session"
+                        " never selected. Further occurrences this session"
+                        " are logged at DEBUG.)",
+                        self.config.model_role,
+                        refused_name,
+                        reason,
+                        provider_name,
+                    )
+                elif role_had_no_candidates:
+                    self._warn_once(
+                        session_id,
+                        self._role_fallback_warned,
+                        "model_role %r resolved to no candidates; session"
+                        " naming is falling back to provider %r (the"
+                        " session's own conversation provider) instead of"
+                        " skipping. This uses whatever model that provider is"
+                        " already configured with, which may be more"
+                        " expensive than intended — configure a %r"
+                        " candidate in the routing matrix to route naming"
+                        " to a cheap model instead. (Further occurrences"
+                        " this session are logged at DEBUG.)",
+                        self.config.model_role,
+                        provider_name,
+                        self.config.model_role,
+                    )
 
             if not provider:
                 logger.warning("No provider available for session naming")
@@ -717,8 +907,12 @@ async def mount(
         max_retries: int (default: 3) - Max retries on defer
         model_role: str | None (default: "fast") - Model role resolved via routing matrix.
             Defaults to "fast" so naming uses a cheap model automatically.
-            Set to None to use the priority provider explicitly.
-            Falls back to priority provider silently when hooks-routing is not installed.
+            Set to None to use the session's own conversation provider explicitly.
+            A resolved candidate is honoured only when it is mounted in this
+            session AND shares the vendor of the session's own provider;
+            anything else is refused with a WARNING and naming runs on the
+            session's own provider. Falls back to that provider (debug-logged)
+            when no routing bundle is installed.
     """
     config = config or {}
 
diff --git a/modules/hooks-session-naming/tests/test_session_naming.py b/modules/hooks-session-naming/tests/test_session_naming.py
index 626cff16..deddd42c 100644
--- a/modules/hooks-session-naming/tests/test_session_naming.py
+++ b/modules/hooks-session-naming/tests/test_session_naming.py
@@ -34,16 +34,41 @@ def _make_mock_provider() -> MagicMock:
     return provider
 
 
+def _make_vendor_provider(vendor: str, *, priority: int | None = None) -> MagicMock:
+    """A mock provider that answers the kernel's ``get_info().id`` contract.
+
+    ``vendor`` is the provider id ("anthropic", "openai", ...) — two mount
+    names sharing an id are the same vendor.
+    """
+    provider = _make_mock_provider()
+    provider.get_info.return_value.id = vendor
+    if priority is not None:
+        provider.priority = priority
+    return provider
+
+
+def _make_pin(current: str | None):
+    """Duck-typed ``conversation.provider_pin`` capability mock."""
+    pin = MagicMock()
+    pin.current = MagicMock(return_value=current)
+    return pin
+
+
 def _make_coordinator(
     *,
     providers: dict | None = None,
     model_role_resolver=None,
+    provider_pin: str | None = None,
 ) -> MagicMock:
     """Return a coordinator mock wired for session-naming tests.
 
     ``model_role_resolver`` is the duck-typed capability the consumer code
     looks up via ``coordinator.get_capability("model_role_resolver")``.
     Pass ``None`` (default) to simulate "no routing bundle installed".
+
+    ``provider_pin`` is the mount name the ``conversation.provider_pin``
+    capability reports as pinned. ``None`` (default) means unpinned, which
+    is what a session without an explicit pin looks like.
     """
     coordinator = MagicMock()
     coordinator.session_state = {}
@@ -59,7 +84,12 @@ def _make_coordinator(
     coordinator.get = MagicMock(
         side_effect=lambda key: _providers if key == "providers" else None
     )
-    capabilities: dict = {"model_role_resolver": model_role_resolver}
+    capabilities: dict = {
+        "model_role_resolver": model_role_resolver,
+        "conversation.provider_pin": (
+            _make_pin(provider_pin) if provider_pin is not None else None
+        ),
+    }
     coordinator.get_capability = MagicMock(side_effect=capabilities.get)
     return coordinator
 
@@ -70,11 +100,13 @@ def _make_hook(
     model_role_resolver=None,
     model_role: str | None = None,
     initial_trigger_turn: int = 2,
+    provider_pin: str | None = None,
 ) -> SessionNamingHook:
     """Return a SessionNamingHook with mocked coordinator."""
     coordinator = _make_coordinator(
         providers=providers,
         model_role_resolver=model_role_resolver,
+        provider_pin=provider_pin,
     )
     config = SessionNamingConfig(
         initial_trigger_turn=initial_trigger_turn,
@@ -529,6 +561,247 @@ async def test_resolver_exception_logs_warning(self, caplog) -> None:
         assert warnings, "Expected a WARNING log when the resolver raises"
 
 
+# =============================================================================
+# Cross-provider purity: naming never calls a provider this session didn't pick
+# =============================================================================
+
+
+class TestProviderPurity:
+    """A session pinned to provider X must never emit a naming call on Y.
+
+    Measured leak this pins shut (model_performance-egh): the routing matrix
+    defaults to openai, so in an Anthropic-pinned session ``model_role="fast"``
+    resolved to an openai candidate, and the unmatched-candidate path fell
+    through to ``next(iter(providers.values()))`` — an order-dependent,
+    SILENT borrow of whichever provider instance happened to be first in the
+    mount dict. 321 foreign responses across 12 capture roots came from here.
+    """
+
+    @pytest.mark.asyncio
+    async def test_pinned_session_never_calls_foreign_vendor(self, caplog) -> None:
+        """Anthropic-pinned session + openai-resolving role → anthropic only.
+
+        The mount dict deliberately lists openai FIRST, so the old
+        ``next(iter(providers))`` fallback would have picked openai even
+        without the resolver ever matching.
+        """
+        openai_provider = _make_vendor_provider("openai")
+        anthropic_provider = _make_vendor_provider("anthropic")
+        providers = {
+            "openai-gpt-5": openai_provider,
+            "anthropic-sonnet": anthropic_provider,
+        }
+
+        resolver = _make_resolver(
+            return_value=[
+                ProviderPreference(provider="openai", model="gpt-5-mini", config={}),
+            ]
+        )
+        hook = _make_hook(
+            providers=providers,
+            model_role_resolver=resolver,
+            model_role="fast",
+            provider_pin="anthropic-sonnet",
+        )
+
+        with caplog.at_level("WARNING"):
+            result = await hook._call_provider("name this session", "session-pin")
+
+        assert not openai_provider.complete.called, (
+            "A session pinned to anthropic must NEVER emit a naming call on "
+            "openai — this is the cross-provider leak"
+        )
+        assert anthropic_provider.complete.called, (
+            "Naming must run on the session's own pinned provider"
+        )
+        assert result is not None
+
+        request = anthropic_provider.complete.call_args[0][0]
+        assert request.model is None, (
+            "A refused foreign candidate must not leave its model override "
+            "behind on the session's own provider"
+        )
+
+        warnings = [r.getMessage() for r in caplog.records if r.levelno >= 30]
+        assert warnings, "Refusing a foreign provider must be loud, not silent"
+        assert any("openai" in m for m in warnings), (
+            "The warning must name the provider that was refused"
+        )
+        assert any("anthropic-sonnet" in m for m in warnings), (
+            "The warning must name the provider actually used"
+        )
+
+    @pytest.mark.asyncio
+    async def test_same_vendor_sibling_is_allowed_with_model_override(self) -> None:
+        """anthropic-haiku for an anthropic-pinned session is NOT a leak.
+
+        Two mount names sharing a ``get_info().id`` are the same vendor, so
+        routing a cheap chore to a cheaper sibling model stays allowed — the
+        purity rule is about vendors, not about mount names.
+        """
+        sonnet = _make_vendor_provider("anthropic")
+        haiku = _make_vendor_provider("anthropic")
+        providers = {"anthropic-sonnet": sonnet, "anthropic-haiku": haiku}
+
+        resolver = _make_resolver(
+            return_value=[
+                ProviderPreference(
+                    provider="anthropic-haiku", model="claude-haiku-4-5", config={}
+                ),
+            ]
+        )
+        hook = _make_hook(
+            providers=providers,
+            model_role_resolver=resolver,
+            model_role="fast",
+            provider_pin="anthropic-sonnet",
+        )
+        await hook._call_provider("name this session", "session-sibling")
+
+        assert haiku.complete.called, "Same-vendor sibling must still be usable"
+        assert not sonnet.complete.called
+        assert haiku.complete.call_args[0][0].model == "claude-haiku-4-5"
+
+    @pytest.mark.asyncio
+    async def test_unknown_vendor_candidate_is_refused(self, caplog) -> None:
+        """Fail closed: a candidate whose vendor cannot be established is refused.
+
+        ``get_info()`` is the only contract for vendor identity. If it is
+        missing or unreadable, sameness cannot be PROVEN, and an unprovable
+        sameness is exactly how the leak got in.
+        """
+        session_provider = _make_vendor_provider("anthropic")
+        mystery = _make_mock_provider()
+        mystery.get_info = MagicMock(side_effect=RuntimeError("no info"))
+        providers = {
+            "anthropic-sonnet": session_provider,
+            "mystery-provider": mystery,
+        }
+
+        resolver = _make_resolver(
+            return_value=[
+                ProviderPreference(provider="mystery", model="who-knows", config={}),
+            ]
+        )
+        hook = _make_hook(
+            providers=providers,
+            model_role_resolver=resolver,
+            model_role="fast",
+            provider_pin="anthropic-sonnet",
+        )
+
+        with caplog.at_level("WARNING"):
+            await hook._call_provider("name this session", "session-unknown")
+
+        assert not mystery.complete.called, (
+            "An unprovable-vendor candidate must be refused, not borrowed"
+        )
+        assert session_provider.complete.called
+        assert [r for r in caplog.records if r.levelno >= 30]
+
+    @pytest.mark.asyncio
+    async def test_resolved_provider_not_mounted_is_refused(self, caplog) -> None:
+        """A candidate naming a provider that is not mounted here is refused."""
+        session_provider = _make_vendor_provider("anthropic")
+        providers = {"anthropic-sonnet": session_provider}
+
+        resolver = _make_resolver(
+            return_value=[
+                ProviderPreference(provider="gemini", model="flash", config={}),
+            ]
+        )
+        hook = _make_hook(
+            providers=providers,
+            model_role_resolver=resolver,
+            model_role="fast",
+        )
+
+        with caplog.at_level("WARNING"):
+            await hook._call_provider("name this session", "session-unmounted")
+
+        assert session_provider.complete.called
+        assert session_provider.complete.call_args[0][0].model is None
+        warnings = [r.getMessage() for r in caplog.records if r.levelno >= 30]
+        assert any("gemini" in m for m in warnings), (
+            "The warning must name the unmounted provider that was refused"
+        )
+
+    @pytest.mark.asyncio
+    async def test_unpinned_session_uses_priority_not_dict_order(self) -> None:
+        """Unpinned selection follows the orchestrator's priority rule.
+
+        The mount dict lists openai first; anthropic carries the better
+        (lower) priority, so the conversation is answered by anthropic — and
+        so must naming be. ``next(iter(providers))`` would have picked openai.
+        """
+        openai_provider = _make_vendor_provider("openai", priority=100)
+        anthropic_provider = _make_vendor_provider("anthropic", priority=10)
+        providers = {
+            "openai-gpt-5": openai_provider,
+            "anthropic-sonnet": anthropic_provider,
+        }
+
+        hook = _make_hook(providers=providers)
+        await hook._call_provider("name this session", "session-priority")
+
+        assert anthropic_provider.complete.called, (
+            "Naming must follow the same priority rule the orchestrator uses "
+            "to pick the conversation provider"
+        )
+        assert not openai_provider.complete.called
+
+    @pytest.mark.asyncio
+    async def test_stale_pin_refuses_instead_of_borrowing(self, caplog) -> None:
+        """A pin whose provider is gone must skip naming, not pick another."""
+        openai_provider = _make_vendor_provider("openai")
+        providers = {"openai-gpt-5": openai_provider}
+
+        hook = _make_hook(providers=providers, provider_pin="anthropic-sonnet")
+
+        with caplog.at_level("WARNING"):
+            result = await hook._call_provider("name this session", "session-stale")
+
+        assert result is None
+        assert not openai_provider.complete.called, (
+            "A stale pin must never fall through to whatever else is mounted"
+        )
+        warnings = [r.getMessage() for r in caplog.records if r.levelno >= 30]
+        assert any("anthropic-sonnet" in m for m in warnings)
+
+    @pytest.mark.asyncio
+    async def test_cross_provider_refusal_warns_once_per_session(
+        self, caplog
+    ) -> None:
+        """The refusal warning fires once per session, then drops to DEBUG."""
+        providers = {
+            "openai-gpt-5": _make_vendor_provider("openai"),
+            "anthropic-sonnet": _make_vendor_provider("anthropic"),
+        }
+        resolver = _make_resolver(
+            return_value=[
+                ProviderPreference(provider="openai", model="gpt-5-mini", config={}),
+            ]
+        )
+        hook = _make_hook(
+            providers=providers,
+            model_role_resolver=resolver,
+            model_role="fast",
+            provider_pin="anthropic-sonnet",
+        )
+
+        with caplog.at_level("DEBUG"):
+            await hook._call_provider("name this session", "session-repeat")
+            first = [r for r in caplog.records if r.levelno >= 30]
+            caplog.clear()
+            await hook._call_provider("name this session", "session-repeat")
+            second = [r for r in caplog.records if r.levelno >= 30]
+            second_debug = [r for r in caplog.records if r.levelno == 10]
+
+        assert first, "First refusal in a session must warn"
+        assert not second, "Second refusal in the SAME session must not re-warn"
+        assert second_debug, "Repeat refusals must still be logged at DEBUG"
+
+
 # =============================================================================
 # Task 7: Background naming call must not leak llm:stream_* events
 # =============================================================================

From 8bcbed43d20a817d1c8d3aae8041fa66c966032e Mon Sep 17 00:00:00 2001
From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Date: Wed, 2 Sep 2026 16:49:00 -0700
Subject: [PATCH 2/2] fix(session-naming): stamp naming's own llm:* events so
 scorers can exclude them

A provider emits llm:request / llm:response through the coordinator it was
mounted with -- the ROOT session's -- and the kernel stamps session_id and
parent_id defaults onto every event
(amplifier_core/session.py:88-91, set_default_fields(session_id, parent_id)).

Pre-fix, __init__.py:518-526 issued the naming call on that same coordinator,
so the hook's own calls landed in the session's events.jsonl with
parent_id: null and NO marker of any kind -- structurally indistinguishable
from the root agent's work. Every scorer in the model_performance program
counted them as root responses: 321 of 12,882 (2.49%), and in 20260902-4nd
that silently moved an Anthropic re-warm headline by 3.3 pp against a
pre-registered margin of +0.0016.

Every event a naming call emits now carries:

    {"purpose": "session-naming", "origin_module": "hooks-session-naming"}

hooks-logging copies unknown payload keys straight into the record's data
object, so excluding session naming is one predicate:

    select(.data.purpose != "session-naming")

Mechanism: a provider reads self.coordinator inside its own methods, so a
forwarding proxy cannot intercept it -- only a copy with its own coordinator
attribute can. _stamped_provider() returns a shallow copy carrying a
_NamingCoordinator (whose hooks stamp every emit and forward everything
else), built once per provider per session so a lazily-created SDK client is
not rebuilt every few turns. The shared provider instance is never mutated:
the foreground conversation's own events stay unstamped.

If a provider's events cannot be stamped (frozen instance, uncopyable), the
naming call is SKIPPED with a WARNING. An unattributable call is worse than a
missing session name.

Known, unchanged: the 10 s hard timeout at __init__.py:248-256 can leave a
stamped llm:request with no matching llm:response (one measured arm logged 15
naming requests and 13 responses). The stamp makes that orphan identifiable
rather than mysterious; the timeout itself is deliberately left alone.

Five tests in TestNamingEventAttribution; three fail against the pre-fix
module. Module version 0.1.2 -> 0.2.0.

Refs: model_performance-dgf, model_performance-egh
---
 modules/hooks-session-naming/README.md        |  33 ++++
 .../__init__.py                               | 142 +++++++++++++-
 modules/hooks-session-naming/pyproject.toml   |   2 +-
 .../tests/test_session_naming.py              | 178 ++++++++++++++++++
 4 files changed, 352 insertions(+), 3 deletions(-)

diff --git a/modules/hooks-session-naming/README.md b/modules/hooks-session-naming/README.md
index 3ba6c644..2efe473a 100644
--- a/modules/hooks-session-naming/README.md
+++ b/modules/hooks-session-naming/README.md
@@ -17,6 +17,7 @@ The module is entirely non-blocking: all LLM calls run as background asyncio tas
 - **Description updates**: Periodically updates the session description as the conversation evolves, only when scope meaningfully expands
 - **Smart context extraction**: Uses a bookend+sampling strategy for long conversations (first 3 turns, sampled middle, last 5 turns)
 - **Graceful deferral**: If the LLM signals insufficient context, retries on subsequent turns up to `max_retries` times
+- **Attributable**: Every `llm:*` event a naming call emits carries `data.purpose = "session-naming"`, so analyzers can exclude it from the session's own work (see [Event Attribution](#event-attribution))
 
 ## Configuration
 
@@ -74,6 +75,38 @@ event stream (321 foreign responses across 12 capture roots; see
 `model_performance-egh`). Same-vendor siblings (`anthropic-sonnet` →
 `anthropic-haiku`) remain allowed: that is the intended cheap-model routing.
 
+## Event Attribution
+
+A provider emits `llm:request` / `llm:response` through the coordinator it was
+mounted with — the session's own — and the kernel stamps `session_id` and
+`parent_id` defaults onto every event
+(`amplifier_core/session.py`: `set_default_fields(...)`). A background naming
+call therefore lands in the session's `events.jsonl` with `parent_id: null` and,
+before this module stamped them, nothing at all to distinguish it from the root
+agent's own turns.
+
+Every event a naming call emits now carries:
+
+```json
+{"purpose": "session-naming", "origin_module": "hooks-session-naming"}
+```
+
+Excluding session naming from an analysis is then one predicate:
+
+```jq
+select(.data.purpose != "session-naming")
+```
+
+The stamp is applied to a naming-only *view* of the provider (a shallow copy
+carrying a wrapping coordinator), built once per provider per session. The
+shared provider instance is never mutated, so the foreground conversation's own
+events are unaffected. If a provider's events cannot be stamped, the naming call
+is **skipped** with a WARNING rather than emitted unattributably.
+
+Note: the provider call has a 10 s hard timeout. A timed-out call can leave a
+stamped `llm:request` with no matching `llm:response` — the stamp is what makes
+that orphan identifiable rather than mysterious.
+
 ### Optional Dependency: hooks-routing
 
 `hooks-routing` is an **optional runtime dependency**. The module degrades gracefully:
diff --git a/modules/hooks-session-naming/amplifier_module_hooks_session_naming/__init__.py b/modules/hooks-session-naming/amplifier_module_hooks_session_naming/__init__.py
index 8a9c4510..69ae9744 100644
--- a/modules/hooks-session-naming/amplifier_module_hooks_session_naming/__init__.py
+++ b/modules/hooks-session-naming/amplifier_module_hooks_session_naming/__init__.py
@@ -6,6 +6,7 @@
 """
 
 import asyncio
+import copy
 import json
 import logging
 import re
@@ -18,6 +19,65 @@
 
 logger = logging.getLogger(__name__)
 
+# Provenance stamped onto every event this module's own LLM call emits.
+# The provider writes llm:request / llm:response into the SESSION'S event
+# stream through the coordinator it was mounted with, and the kernel adds
+# session_id / parent_id defaults -- so without a stamp a naming call is
+# structurally indistinguishable from the root agent's own work, and every
+# scorer reading events.jsonl counts it as a root response.
+NAMING_PURPOSE = "session-naming"
+NAMING_ORIGIN = "hooks-session-naming"
+
+
+class _NamingHooks:
+    """Hook-registry view that stamps naming provenance on every event.
+
+    Wraps the real registry: ``emit``/``emit_and_collect`` add
+    ``purpose``/``origin_module`` to the payload before it reaches the
+    registry, so the fields land in ``data`` in events.jsonl (hooks-logging
+    copies unknown payload keys straight through). Everything else is
+    forwarded untouched.
+    """
+
+    def __init__(self, hooks: Any):
+        self._hooks = hooks
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._hooks, name)
+
+    @staticmethod
+    def _stamp(data: Any) -> Any:
+        if not isinstance(data, dict):
+            return data
+        stamped = dict(data)
+        stamped["purpose"] = NAMING_PURPOSE
+        stamped["origin_module"] = NAMING_ORIGIN
+        return stamped
+
+    async def emit(self, event: str, data: Any = None) -> Any:
+        return await self._hooks.emit(event, self._stamp(data))
+
+    async def emit_and_collect(
+        self, event: str, data: Any = None, timeout: float | None = None
+    ) -> Any:
+        return await self._hooks.emit_and_collect(event, self._stamp(data), timeout)
+
+
+class _NamingCoordinator:
+    """Coordinator view whose ``hooks`` stamp naming provenance.
+
+    Handed to a provider *copy* (see ``SessionNamingHook._stamped_provider``)
+    so the provider's own ``self.coordinator.hooks.emit`` calls are tagged.
+    Every other coordinator attribute is forwarded to the real one.
+    """
+
+    def __init__(self, coordinator: Any):
+        self._coordinator = coordinator
+        self.hooks = _NamingHooks(coordinator.hooks)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._coordinator, name)
+
 
 @dataclass
 class SessionNamingConfig:
@@ -116,6 +176,14 @@ def __init__(self, coordinator: Any, config: SessionNamingConfig):
         # Same dedup, for the "model_role resolved to a provider this session
         # never selected — refusing to borrow it" WARNING.
         self._cross_provider_refused: set[str] = set()
+        # Same dedup, for the "cannot stamp this provider's events" WARNING.
+        self._unstampable_warned: set[str] = set()
+        # id(real provider) -> (real provider, stamped copy). The copy is made
+        # once per provider per session: providers create their SDK client
+        # lazily, so a fresh copy on every naming call would build a fresh
+        # client (and connection pool) every few turns. The real provider is
+        # held alongside so its id() cannot be recycled under us.
+        self._stamped_providers: dict[int, tuple[Any, Any]] = {}
 
     async def on_orchestrator_complete(
         self, event: str, data: dict[str, Any]
@@ -589,6 +657,53 @@ def _match_resolved_provider(
                 return key, provider
         return None, None
 
+    def _stamped_provider(self, provider: Any) -> Any | None:
+        """A view of ``provider`` whose emitted events carry naming provenance.
+
+        A provider emits ``llm:request`` / ``llm:response`` through the
+        coordinator it holds on ``self.coordinator`` — the ROOT session's
+        coordinator — and the kernel stamps ``session_id``/``parent_id``
+        defaults onto every event. So a naming call's events are otherwise
+        indistinguishable from the root agent's own, and every scorer reading
+        events.jsonl counts them as root responses.
+
+        Attribute reads inside the provider's own methods bind to its real
+        instance, so a forwarding proxy cannot intercept them — only a copy
+        with its own ``coordinator`` attribute can. The copy is shallow: the
+        SDK client, config and credentials are shared with the original.
+
+        Returns:
+            The stamped copy; the provider itself when it emits nothing
+            (no coordinator, so nothing can leak); or None when the copy
+            cannot be made or the coordinator cannot be swapped — the caller
+            must then SKIP the call rather than emit unattributable events
+            into the session's stream.
+        """
+        base = getattr(provider, "coordinator", None)
+        if base is None or not hasattr(base, "hooks"):
+            # Nothing is emitted through this provider, so nothing to stamp.
+            return provider
+        if isinstance(base, _NamingCoordinator):
+            return provider
+
+        cached = self._stamped_providers.get(id(provider))
+        if cached is not None and cached[0] is provider:
+            return cached[1]
+
+        try:
+            stamped = copy.copy(provider)
+            stamped.coordinator = _NamingCoordinator(base)
+        except Exception as e:
+            logger.debug("Could not build a stamped provider view: %s", e)
+            return None
+
+        if not isinstance(getattr(stamped, "coordinator", None), _NamingCoordinator):
+            # e.g. a frozen model that swallowed the assignment.
+            return None
+
+        self._stamped_providers[id(provider)] = (provider, stamped)
+        return stamped
+
     def _warn_once(self, session_id: str | None, seen: set[str], *args: Any) -> None:
         """WARNING the first time per session, DEBUG on every repeat.
 
@@ -776,6 +891,29 @@ async def _call_provider(
                 logger.warning("No provider available for session naming")
                 return None
 
+            # Attribution: the provider emits llm:request / llm:response into
+            # THIS session's event stream. Route those emits through a stamping
+            # coordinator so every one of them carries purpose="session-naming"
+            # and a scorer can exclude them from the root agent's own work.
+            # If the events cannot be stamped, SKIP the call — naming is a
+            # best-effort background chore, and an unattributable call is worse
+            # than a missing session name.
+            call_provider = self._stamped_provider(provider)
+            if call_provider is None:
+                self._warn_once(
+                    session_id,
+                    self._unstampable_warned,
+                    "Session naming cannot stamp provider %r's events with"
+                    " purpose=%r, so its llm:request/llm:response would be"
+                    " indistinguishable from this session's own work."
+                    " SKIPPING naming rather than emitting unattributable"
+                    " events. (Further occurrences this session are logged at"
+                    " DEBUG.)",
+                    provider_name,
+                    NAMING_PURPOSE,
+                )
+                return None
+
             # Make the request — model=None means use provider default.
             # metadata={"stream": False} signals to the provider that this is
             # a background utility call and must NOT take the streaming branch.
@@ -807,7 +945,7 @@ async def _call_provider(
             # Anthropic's "streaming is required for operations that may take
             # longer than 10 minutes" guard and makes naming fail every retry.
             # Providers without a thinking concept ignore this kwarg.
-            response = await provider.complete(request, extended_thinking=False)
+            response = await call_provider.complete(request, extended_thinking=False)
 
             if response and response.content:
                 # Extract text from content blocks
@@ -948,7 +1086,7 @@ async def mount(
 
     return {
         "name": "hooks-session-naming",
-        "version": "0.1.1",
+        "version": "0.2.0",
         "description": "Automatic session naming and description generation",
         "config": {
             "initial_trigger_turn": hook_config.initial_trigger_turn,
diff --git a/modules/hooks-session-naming/pyproject.toml b/modules/hooks-session-naming/pyproject.toml
index 309ff6cb..e1267615 100644
--- a/modules/hooks-session-naming/pyproject.toml
+++ b/modules/hooks-session-naming/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "amplifier-module-hooks-session-naming"
-version = "0.1.2"
+version = "0.2.0"
 description = "Automatic session naming and description generation"
 readme = "README.md"
 requires-python = ">=3.11"
diff --git a/modules/hooks-session-naming/tests/test_session_naming.py b/modules/hooks-session-naming/tests/test_session_naming.py
index deddd42c..f6f6512d 100644
--- a/modules/hooks-session-naming/tests/test_session_naming.py
+++ b/modules/hooks-session-naming/tests/test_session_naming.py
@@ -4,6 +4,7 @@
 
 import asyncio
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 from amplifier_foundation.spawn_utils import ProviderPreference
@@ -802,6 +803,183 @@ async def test_cross_provider_refusal_warns_once_per_session(
         assert second_debug, "Repeat refusals must still be logged at DEBUG"
 
 
+# =============================================================================
+# Attribution: naming's own llm:* events must be distinguishable from root work
+# =============================================================================
+
+
+class _EmittingProvider:
+    """A provider that emits llm:* the way real providers do.
+
+    Real providers emit through ``self.coordinator.hooks.emit`` — an attribute
+    read bound to their own instance — which is why attribution has to happen
+    on a provider view rather than via a forwarding proxy.
+    """
+
+    def __init__(self, vendor: str = "anthropic") -> None:
+        self.coordinator = None
+        self._vendor = vendor
+        self.complete_calls: list = []
+
+    def get_info(self):
+        return SimpleNamespace(id=self._vendor)
+
+    async def complete(self, request, **kwargs):
+        self.complete_calls.append(request)
+        await self.coordinator.hooks.emit(
+            "llm:request", {"provider": self._vendor, "model": "test-model"}
+        )
+        await self.coordinator.hooks.emit(
+            "llm:response",
+            {"provider": self._vendor, "model": "test-model", "status": "ok"},
+        )
+        return SimpleNamespace(
+            content=[
+                SimpleNamespace(
+                    text='{"action": "set", "name": "N", "description": "D"}'
+                )
+            ]
+        )
+
+
+def _llm_events(coordinator) -> list[tuple[str, dict]]:
+    """(event, data) pairs for llm:* events emitted on a coordinator mock."""
+    return [
+        (call.args[0], call.args[1])
+        for call in coordinator.hooks.emit.call_args_list
+        if call.args and str(call.args[0]).startswith("llm:")
+    ]
+
+
+class TestNamingEventAttribution:
+    """The hook's own LLM calls must never look like the root agent's.
+
+    Providers write llm:request/llm:response into the SESSION's event stream,
+    and the kernel stamps session_id/parent_id defaults onto every event
+    (amplifier_core/session.py: set_default_fields(session_id, parent_id)).
+    Pre-fix, a naming call was therefore recorded with parent_id: null and no
+    marker at all — 321 such responses were counted as root agent work by
+    every scorer in the model_performance program.
+    """
+
+    @pytest.mark.asyncio
+    async def test_naming_llm_events_carry_purpose_marker(self) -> None:
+        """Every llm:* event a naming call emits carries data.purpose."""
+        provider = _EmittingProvider()
+        hook = _make_hook(providers={"anthropic-sonnet": provider})
+        provider.coordinator = hook.coordinator
+
+        result = await hook._call_provider("name this session", "session-attr")
+        assert result is not None
+
+        events = _llm_events(hook.coordinator)
+        assert [name for name, _ in events] == ["llm:request", "llm:response"], (
+            "The naming call must actually have emitted provider events"
+        )
+        for name, data in events:
+            assert data.get("purpose") == "session-naming", (
+                f"{name} emitted by session naming must be excludable by a "
+                f"scorer; got {data!r}"
+            )
+            assert data.get("origin_module") == "hooks-session-naming"
+
+    @pytest.mark.asyncio
+    async def test_original_provider_is_not_mutated(self) -> None:
+        """Foreground calls through the same provider stay unstamped.
+
+        The stamp must live on a naming-only view. If it were applied to the
+        shared provider instance, the root agent's own events would start
+        claiming to be session naming — the same attribution bug, inverted.
+        """
+        provider = _EmittingProvider()
+        hook = _make_hook(providers={"anthropic-sonnet": provider})
+        root_coordinator = hook.coordinator
+        provider.coordinator = root_coordinator
+
+        await hook._call_provider("name this session", "session-attr")
+
+        assert provider.coordinator is root_coordinator, (
+            "The shared provider instance must be left exactly as it was"
+        )
+
+        root_coordinator.hooks.emit.reset_mock()
+        await provider.complete(object())
+        for _, data in _llm_events(root_coordinator):
+            assert "purpose" not in data, (
+                "A non-naming call through the same provider must not be "
+                "stamped as session naming"
+            )
+
+    @pytest.mark.asyncio
+    async def test_stamped_view_is_built_once_per_provider(self) -> None:
+        """Providers create SDK clients lazily; don't build a view per turn."""
+        provider = _EmittingProvider()
+        hook = _make_hook(providers={"anthropic-sonnet": provider})
+        provider.coordinator = hook.coordinator
+
+        await hook._call_provider("name this session", "session-attr")
+        await hook._call_provider("name this session", "session-attr")
+
+        assert len(hook._stamped_providers) == 1
+        assert hook._stamped_provider(provider) is hook._stamped_provider(provider)
+
+    @pytest.mark.asyncio
+    async def test_unstampable_provider_skips_rather_than_leaks(
+        self, caplog
+    ) -> None:
+        """If events cannot be stamped, skip the call — loudly.
+
+        An unattributable naming call is worse than a missing session name:
+        it silently contaminates whatever reads the event stream.
+        """
+
+        class _FrozenProvider:
+            """Read-only ``coordinator`` — the copy cannot be re-pointed."""
+
+            def __init__(self) -> None:
+                self._coordinator = None
+                self.complete_calls: list = []
+
+            @property
+            def coordinator(self):
+                return self._coordinator
+
+            def get_info(self):
+                return SimpleNamespace(id="anthropic")
+
+            async def complete(self, request, **kwargs):  # pragma: no cover
+                self.complete_calls.append(request)
+                raise AssertionError("must not be called")
+
+        provider = _FrozenProvider()
+        hook = _make_hook(providers={"anthropic-sonnet": provider})
+        provider._coordinator = hook.coordinator
+
+        with caplog.at_level("WARNING"):
+            result = await hook._call_provider("name this session", "session-frozen")
+
+        assert result is None
+        assert not provider.complete_calls, (
+            "Must not issue the call at all when its events cannot be stamped"
+        )
+        warnings = [r.getMessage() for r in caplog.records if r.levelno >= 30]
+        assert any("session-naming" in m for m in warnings), (
+            "Skipping for lack of attribution must be loud, not silent"
+        )
+
+    @pytest.mark.asyncio
+    async def test_provider_without_coordinator_still_names(self) -> None:
+        """A provider that emits nothing has nothing to leak — don't skip it."""
+        provider = _make_mock_provider()
+        provider.coordinator = None
+        hook = _make_hook(providers={"provider-1": provider})
+
+        result = await hook._call_provider("name this session", "session-none")
+
+        assert result is not None
+        assert provider.complete.called
+
+
 # =============================================================================
 # Task 7: Background naming call must not leak llm:stream_* events
 # =============================================================================