diff --git a/mellea/backends/adapters/_core.py b/mellea/backends/adapters/_core.py
index 298c9fd49..5351d12f6 100644
--- a/mellea/backends/adapters/_core.py
+++ b/mellea/backends/adapters/_core.py
@@ -9,9 +9,9 @@
 - :class:`IOContract` — ABC for prompt building and output parsing
 - :class:`WeightsBinding` — pluggable ABC for weights lifecycle management
 
-Also provides three stub :class:`WeightsBinding` subclasses
-(:class:`LocalFileBinding`, :class:`EmbeddedBinding`,
-:class:`ServerMediatedBinding`) and :class:`AdapterSchemaMismatchError`.
+Also provides :class:`LocalFileBinding`, two stub :class:`WeightsBinding`
+subclasses (:class:`EmbeddedBinding`, :class:`ServerMediatedBinding`), and
+:class:`AdapterSchemaMismatchError`.
 
 Note:
     The existing :class:`~mellea.backends.adapters.adapter.Adapter` ABC in
@@ -24,12 +24,21 @@
 
 import abc
 import json
+import threading
+import time
 import warnings
 from dataclasses import dataclass
-from typing import Literal
+from typing import TYPE_CHECKING, ClassVar, Literal
 
-from ...core import Component
+from ...core import Component, MelleaLogger
+from ...helpers.event_loop_helper import _run_async_in_thread
+from ...plugins.manager import has_plugins, invoke_hook
+from ...plugins.types import HookType
 from .capabilities import KNOWN_CAPABILITIES
+from .catalog import AdapterType, fetch_intrinsic_metadata
+
+if TYPE_CHECKING:
+    from .adapter import AdapterMixin
 
 _PHASE_2_NOT_IMPLEMENTED = (
     "{cls} is a Phase 0 stub; implementation lands in Epic #929 Phase 2."
@@ -193,8 +202,14 @@ class WeightsBinding(abc.ABC):
 
     Concrete implementations are expected to document any deviations from this
     contract (e.g. servers that prepare-and-activate atomically).
+
+    Attributes:
+        binding_type (ClassVar[str]): Weight-binding reality identifier used in
+            adapter-function telemetry (e.g. `"local_file"`).
     """
 
+    binding_type: ClassVar[str] = "unknown"
+
     @abc.abstractmethod
     def prepare(self) -> None:
         """Prepare the weights for activation (e.g. download or stage them)."""
@@ -217,32 +232,358 @@ def release(self) -> None:
 
 
 class LocalFileBinding(WeightsBinding):
-    """Stub binding for locally stored adapter weights."""
+    """Weights binding for the LocalFile/PEFT reality (Epic #929 Phase 2).
 
-    def prepare(self) -> None:
-        raise NotImplementedError(
-            _PHASE_2_NOT_IMPLEMENTED.format(cls="LocalFileBinding")
+    Downloads LoRA/aLoRA adapter weights from a Hugging Face Hub repository and
+    loads them into a PEFT-capable backend (e.g. `LocalHFBackend`) via the
+    `AdapterMixin` verb contract.
+
+    `prepare()` is session-scoped: call `bind_backend()` once, then `prepare()`.
+    `activate()`/`deactivate()` are call-scoped, typically driven by
+    `AdapterMixin.adapter_scope`.
+    `release()` is terminal.
+
+    Attributes:
+        name (str): Adapter function name (e.g. `"answerability"`).
+        adapter_type (AdapterType): The LoRA variant.
+        repo_id (str): Hugging Face Hub repository containing the adapter weights.
+        revision (str | None): Git revision (branch, tag, or commit SHA) to
+            download, or `None` to use the catalogue's pinned revision for
+            `name` — resolved lazily, so it stays correct if the catalogue is
+            re-pinned. Pass `"main"` explicitly to opt into tracking latest.
+        backend (AdapterMixin | None): Backend this binding is registered
+            with, set by `prepare()` and cleared by `release()`.
+        path (str | None): Local filesystem path to the downloaded adapter
+            weights, set by `prepare()` and cleared by `release()`.
+    """
+
+    binding_type: ClassVar[str] = "local_file"
+
+    def __init__(
+        self,
+        name: str = "",
+        adapter_type: AdapterType = AdapterType.LORA,
+        repo_id: str = "",
+        revision: str | None = None,
+    ) -> None:
+        """Constructs a LocalFileBinding.
+
+        Args:
+            name: Adapter function name (e.g. `"answerability"`).
+            adapter_type: The LoRA variant.
+            repo_id: Hugging Face Hub repository containing the adapter weights.
+            revision: Git revision (branch, tag, or commit SHA) to download, or
+                `None` to use the catalogue's pinned revision for `name`. Pass
+                `"main"` explicitly to opt into tracking latest.
+        """
+        self.name = name
+        self.adapter_type = adapter_type
+        self.repo_id = repo_id
+        self.revision = revision
+        self.backend: AdapterMixin | None = None
+        self.path: str | None = None
+        self._staged_backend: AdapterMixin | None = None
+        self._loaded = False
+        self._active = False
+        self._released = False
+        # Keeps one binding from being released while it registers or loads.
+        self._lifecycle_lock = threading.Lock()
+
+    @property
+    def qualified_name(self) -> str:
+        """Backend-facing adapter identifier, e.g. `"answerability_lora"`."""
+        return f"{self.name}_{self.adapter_type.value}"
+
+    def resolved_revision(self) -> str:
+        """Returns the revision to download, resolving `None` via the catalogue.
+
+        A `revision` of `None` means "whatever the catalogue has pinned for this
+        adapter function". Resolving here rather than in `__init__` keeps a
+        long-lived binding correct across a catalogue re-pin, and keeps
+        construction free of catalogue lookups.
+
+        Returns:
+            The git revision (branch, tag, or commit SHA) to download.
+
+        Raises:
+            ValueError: `revision` is `None` and `name` is not a registered
+                adapter function, so there is no pinned revision to fall back on.
+        """
+        if self.revision is not None:
+            return self.revision
+        return fetch_intrinsic_metadata(self.name).revision
+
+    def get_local_hf_path(self, base_model_name: str) -> str:
+        """Downloads (or reuses a cached copy of) the adapter weights.
+
+        Args:
+            base_model_name: Base model the adapter is being loaded against.
+
+        Returns:
+            Filesystem path to the local copy of the adapter weights.
+
+        Raises:
+            ValueError: `revision` is `None` and `name` is not a registered
+                adapter function.
+        """
+        from ...formatters.granite import intrinsics
+
+        return str(
+            intrinsics.obtain_lora(
+                self.name,
+                base_model_name,
+                self.repo_id,
+                revision=self.resolved_revision(),
+                alora=self.adapter_type is AdapterType.ALORA,
+            )
         )
 
-    def activate(self) -> None:
-        raise NotImplementedError(
-            _PHASE_2_NOT_IMPLEMENTED.format(cls="LocalFileBinding")
+    @classmethod
+    def from_catalog(cls, name: str) -> "LocalFileBinding":
+        """Builds a `LocalFileBinding` from the adapter function catalog.
+
+        Args:
+            name: Adapter function name registered in the catalog.
+
+        Returns:
+            A `LocalFileBinding` configured with the catalog's pinned
+            `repo_id`, `revision`, and first-listed adapter type.
+
+        Raises:
+            ValueError: `name` is not a registered adapter function.
+        """
+        metadata = fetch_intrinsic_metadata(name)
+        return cls(
+            name=name,
+            adapter_type=metadata.adapter_types[0],
+            repo_id=metadata.repo_id,
+            revision=metadata.revision,
         )
 
+    def bind_backend(self, backend: "AdapterMixin") -> None:
+        """Stages the backend that `prepare()` will register this binding with.
+
+        Args:
+            backend: The backend to register with on the next `prepare()` call.
+
+        Raises:
+            RuntimeError: This binding has already been `release()`d, or is
+                registered with a different backend.
+        """
+        with self._lifecycle_lock:
+            if self._released:
+                raise RuntimeError(
+                    "LocalFileBinding.bind_backend() called after release(): "
+                    "release() is terminal per the WeightsBinding contract and does "
+                    "not revive the binding. Construct a new LocalFileBinding instead."
+                )
+            if self.backend is not None and backend is not self.backend:
+                raise RuntimeError(
+                    "LocalFileBinding.bind_backend() cannot change the backend after "
+                    "registration. Release this binding and construct a new one instead."
+                )
+            self._staged_backend = backend
+
+    def prepare(self) -> None:
+        """Downloads the adapter weights and loads them into the staged backend.
+
+        Idempotent: a no-op once already prepared. Retryable: if a previous
+        call registered with the backend but failed during the weights load
+        (e.g. a transient download/load failure), the next call retries only
+        the load rather than re-registering — registration already succeeded
+        and re-attempting it would hit the backend's own duplicate-registration
+        guard.
+
+        The `prepare` phase duration reported to
+        `ADAPTER_FUNCTION_PHASE_COMPLETE` spans the whole operation, **including
+        the Hugging Face download** — `add_adapter` calls `get_local_hf_path`,
+        which can take seconds on a cache miss. That is deliberate (it is the
+        wall-clock cost of preparing), but worth stating, since a phase added
+        later may not want the same boundary.
+
+        Raises:
+            RuntimeError: `bind_backend()` was not called first, `name` is empty,
+                the binding was already `release()`d, or the backend refused
+                the registration.
+        """
+        started_at = time.monotonic()
+        with self._lifecycle_lock:
+            if self._released:
+                raise RuntimeError(
+                    "LocalFileBinding.prepare() called after release(): release() is "
+                    "terminal per the WeightsBinding contract and does not revive the "
+                    "binding. Construct a new LocalFileBinding instead."
+                )
+            if self.backend is not None and self._loaded:
+                return
+            if self.backend is None:
+                if self._staged_backend is None:
+                    raise RuntimeError(
+                        "LocalFileBinding.prepare() requires bind_backend() to be called first."
+                    )
+                if not self.name:
+                    raise RuntimeError(
+                        "LocalFileBinding.prepare() requires a non-empty name. A "
+                        "default-constructed LocalFileBinding() is an unconfigured "
+                        "placeholder — build one with LocalFileBinding.from_catalog(name) "
+                        "instead."
+                    )
+
+                self._staged_backend.add_adapter(self)
+                # `add_adapter` signals success by setting `.backend`; it has early-return
+                # paths (notably: a different object already registered under this
+                # `qualified_name`) that log a warning and leave it unset. Without this
+                # check `prepare()` would go on to load the *other* adapter's weights and
+                # leave `.backend` None, so a later `activate()` would raise "requires
+                # prepare() to be called first" despite `prepare()` having run. Fail here
+                # instead, where the cause is still visible.
+                if self.backend is None:
+                    raise RuntimeError(
+                        f"Backend refused to register adapter {self.qualified_name!r}; see the "
+                        "backend's warning log. Either another adapter is already registered "
+                        "under this qualified name, or this binding was previously released — "
+                        "`release()` is terminal and does not free the name for re-use "
+                        "(see #1528)."
+                    )
+            # `load_peft_adapter` mutates the backend's underlying PEFT model, the
+            # same shared state `activate_peft_adapter`/`deactivate_peft_adapter`
+            # document "must be called while holding `_generation_lock`" for.
+            # `prepare()`/`release()` aren't driven through `adapter_scope`, so
+            # nothing else takes this lock on their behalf.
+            with self.backend._adapter_activation_lock():
+                self.backend.load_peft_adapter(self.qualified_name)
+            self._loaded = True
+        self._fire_phase_complete("prepare", time.monotonic() - started_at)
+
+    def activate(self) -> None:
+        """Selects already-loaded adapter weights for generation.
+
+        Raises:
+            RuntimeError: `prepare()` was not called first, or called but did
+                not complete (registered with the backend but the weights
+                load itself failed or hasn't been retried yet).
+        """
+        backend = self.backend
+        if backend is None or not self._loaded:
+            raise RuntimeError(
+                "LocalFileBinding.activate() requires prepare() to be called first."
+            )
+        with backend._adapter_activation_lock():
+            if self.backend is not backend or not self._loaded:
+                raise RuntimeError(
+                    "LocalFileBinding.activate() requires prepare() to be called first."
+                )
+            backend.activate_peft_adapter(self.qualified_name)
+            self._active = True
+
     def deactivate(self) -> None:
-        raise NotImplementedError(
-            _PHASE_2_NOT_IMPLEMENTED.format(cls="LocalFileBinding")
-        )
+        """Deselects the adapter so generation uses the base model.
+
+        Raises:
+            RuntimeError: `prepare()` was not called first, or called but did
+                not complete (registered with the backend but the weights
+                load itself failed or hasn't been retried yet).
+        """
+        backend = self.backend
+        if backend is None or not self._loaded:
+            raise RuntimeError(
+                "LocalFileBinding.deactivate() requires prepare() to be called first."
+            )
+        with backend._adapter_activation_lock():
+            if self.backend is not backend or not self._loaded:
+                raise RuntimeError(
+                    "LocalFileBinding.deactivate() requires prepare() to be called first."
+                )
+            backend.deactivate_peft_adapter(self.qualified_name)
+            self._active = False
 
     def release(self) -> None:
-        raise NotImplementedError(
-            _PHASE_2_NOT_IMPLEMENTED.format(cls="LocalFileBinding")
+        """Unloads the adapter's weights from the backend and clears local state.
+
+        Idempotent: a no-op if never prepared, or already released. Terminal, per
+        the `WeightsBinding` contract — enforced: `bind_backend()` and
+        `prepare()` both raise `RuntimeError` if called after `release()`,
+        rather than silently reviving the binding on a new backend.
+
+        Does **not** fully deregister. `unload_peft_adapter` removes the adapter
+        from the backend's *loaded* set, but the backend's *registered* set
+        (`_added_adapters` on `LocalHFBackend`) keeps its entry, because
+        `add_adapter` has no inverse verb. So the `qualified_name` stays claimed
+        for the backend's lifetime and no later binding can register under it.
+        Tracked in #1528, which also asks whether re-registration should be
+        supported at all given the terminal contract.
+
+        Raises:
+            RuntimeError: The binding is active; call `deactivate()` before
+                releasing its weights.
+        """
+        with self._lifecycle_lock:
+            if self._released:
+                return
+            backend = self.backend
+            if backend is None:
+                self._staged_backend = None
+                self._released = True
+                return
+
+            # See the matching comment in `prepare()`: this mutates the same
+            # shared PEFT model state `activate_peft_adapter`/
+            # `deactivate_peft_adapter` require the lock for.
+            with backend._adapter_activation_lock():
+                if self.backend is not backend:
+                    return
+                if self._active:
+                    raise RuntimeError(
+                        "LocalFileBinding.release() requires deactivate() to be called first."
+                    )
+                backend.unload_peft_adapter(self.qualified_name)
+                self.backend = None
+                self.path = None
+                self._staged_backend = None
+                self._loaded = False
+                self._active = False
+                self._released = True
+
+    def _fire_phase_complete(self, phase: str, duration_s: float) -> None:
+        """Fires `adapter_function_phase_complete` for a phase this binding owns.
+
+        Only `"prepare"` is fired from here: `"activate"`/`"deactivate"` are
+        owned by `AdapterMixin.adapter_scope`, and `"release"` has no phase
+        metric in the `AdapterFunctionPhaseCompletePayload` contract (Epic #929
+        Phase 1, issue #1140).
+
+        Args:
+            phase: Lifecycle phase name; must be a valid
+                `AdapterFunctionPhaseCompletePayload.phase` value.
+            duration_s: Wall-clock duration of the phase, in seconds.
+        """
+        if not has_plugins(HookType.ADAPTER_FUNCTION_PHASE_COMPLETE):
+            return
+
+        from ...plugins.hooks.adapter_function import (
+            AdapterFunctionPhaseCompletePayload,
         )
 
+        try:
+            payload = AdapterFunctionPhaseCompletePayload(
+                name=self.name, phase=phase, duration_ms=duration_s * 1000.0
+            )
+            hook_coro = invoke_hook(HookType.ADAPTER_FUNCTION_PHASE_COMPLETE, payload)
+            _run_async_in_thread(hook_coro)
+        except Exception:
+            MelleaLogger.get_logger().warning(
+                f"adapter_function_phase_complete hook dispatch failed for {self.name!r} "
+                f"during {phase!r}; ignoring so it does not turn a completed phase "
+                "into an operation failure.",
+                exc_info=True,
+            )
+
 
 class EmbeddedBinding(WeightsBinding):
     """Stub binding for weights embedded in a model artifact."""
 
+    binding_type: ClassVar[str] = "embedded"
+
     def prepare(self) -> None:
         raise NotImplementedError(
             _PHASE_2_NOT_IMPLEMENTED.format(cls="EmbeddedBinding")
@@ -267,6 +608,8 @@ def release(self) -> None:
 class ServerMediatedBinding(WeightsBinding):
     """Stub binding for server-managed adapter weights."""
 
+    binding_type: ClassVar[str] = "server_mediated"
+
     def prepare(self) -> None:
         raise NotImplementedError(
             _PHASE_2_NOT_IMPLEMENTED.format(cls="ServerMediatedBinding")
@@ -304,3 +647,16 @@ class Adapter:
     identity: Identity
     io_contract: IOContract
     weights: WeightsBinding
+
+    # NOTE(#1516): a construction-time cross-check that `weights.adapter_type`
+    # agrees with `identity.adapter_type` was tried here and backed out. It is the
+    # right invariant — the two feed different lookup paths (registration and the
+    # verbs key on the binding's `qualified_name`; `_find_adapter` scans on the
+    # identity) and both return `None` on a miss, so a disagreement surfaces as
+    # "adapter not found" far from its cause. But it cannot be enforced yet: the
+    # ten module-level `Adapter` constants in `stdlib/components/intrinsic/rag.py`
+    # and `guardian.py` pair an `alora` identity with a bare, deliberately
+    # unconfigured `LocalFileBinding()` that defaults to LoRA. Every catalogue
+    # entry supports both types, so those are placeholders rather than genuine
+    # conflicts, and the check fired on "not configured yet". Enforce it once
+    # #1516 gives those constants real bindings.
diff --git a/mellea/backends/adapters/adapter.py b/mellea/backends/adapters/adapter.py
index e00eeab8a..0e5bb2e3f 100644
--- a/mellea/backends/adapters/adapter.py
+++ b/mellea/backends/adapters/adapter.py
@@ -18,14 +18,26 @@
 import contextlib
 import pathlib
 import re
+import time
 import warnings
+from collections.abc import Callable
 from typing import Literal, TypeAlias, TypeVar, cast
 
 import yaml
 
-from ...core import Backend
+from ...core import Backend, MelleaLogger
 from ...formatters.granite import intrinsics as intrinsics
-from ._core import Adapter as _AdapterCore, Identity, IOContract, WeightsBinding
+from ...helpers.event_loop_helper import _run_async_in_thread
+from ...plugins.manager import has_plugins, invoke_hook
+from ...plugins.types import HookType
+from ._core import (
+    Adapter as _AdapterCore,
+    AdapterSchemaMismatchError,
+    Identity,
+    IOContract,
+    LocalFileBinding,
+    WeightsBinding,
+)
 from .catalog import AdapterType, fetch_intrinsic_metadata
 
 
@@ -225,12 +237,11 @@ def __init__(
                 f"{adapter_type} not supported"
             )
             is_alora = self.adapter_type == AdapterType.ALORA
-            # TODO(phase-2.2): pass revision=self.intrinsic_metadata.revision
-            # once revision-aware prepare() is merged (issue #1141 / epic #929).
             config_file = intrinsics.obtain_io_yaml(
                 self.intrinsic_name,
                 self.base_model_name,
                 self.intrinsic_metadata.repo_id,
+                revision=self.intrinsic_metadata.revision,
                 alora=is_alora,
             )
         if config_file:
@@ -283,13 +294,12 @@ def download_and_get_path(self, base_model_name: str) -> str:
             a path to the files
         """
         is_alora = self.adapter_type == AdapterType.ALORA
-        # TODO(phase-2.2): pass revision=self.intrinsic_metadata.revision once
-        # revision-aware prepare() is merged (issue #1141 / epic #929).
         return str(
             intrinsics.obtain_lora(
                 self.intrinsic_name,
                 base_model_name,
                 self.intrinsic_metadata.repo_id,
+                revision=self.intrinsic_metadata.revision,
                 alora=is_alora,
             )
         )
@@ -326,6 +336,102 @@ def get_adapter_for_intrinsic(
     return adapter
 
 
+def _fire_phase_complete_hook(name: str, phase: str, duration_ms: float) -> None:
+    """Fire the `adapter_function_phase_complete` metric hook for a phase that already ran.
+
+    Split out of `_run_adapter_phase` so a caller that must guarantee cleanup
+    after a phase's side effect — e.g. `adapter_scope` guaranteeing
+    `deactivate()` runs once `activate()` has succeeded — can run the side
+    effect and this hook fire under separate exception handling. A hook-dispatch
+    failure is logged and ignored: observability must not turn a completed
+    lifecycle phase into an operation failure.
+
+    Args:
+        name: Adapter function name, used as the metric's `name` field.
+        phase: Lifecycle phase name; must be a valid
+            `AdapterFunctionPhaseCompletePayload.phase` value.
+        duration_ms: Wall-clock duration of the phase, in milliseconds.
+    """
+    if not has_plugins(HookType.ADAPTER_FUNCTION_PHASE_COMPLETE):
+        return
+    from ...plugins.hooks.adapter_function import AdapterFunctionPhaseCompletePayload
+
+    payload = AdapterFunctionPhaseCompletePayload(
+        name=name, phase=phase, duration_ms=duration_ms
+    )
+    try:
+        hook_coro = invoke_hook(HookType.ADAPTER_FUNCTION_PHASE_COMPLETE, payload)
+        _run_async_in_thread(hook_coro)
+    except Exception:
+        MelleaLogger.get_logger().warning(
+            f"adapter_function_phase_complete hook dispatch failed for {name!r} "
+            f"during {phase!r}; ignoring so it does not turn a completed phase "
+            "into an operation failure.",
+            exc_info=True,
+        )
+
+
+def _run_adapter_phase(name: str, phase: str, phase_fn: Callable[[], None]) -> None:
+    """Run one lifecycle phase and fire its phase-complete metric hook.
+
+    Fires the hook only; it does not open a span. Span production belongs to a
+    plugin (#1464, #1466), not to code under `mellea/backends/`.
+
+    The hook fires **only when the phase succeeds**, matching the name of
+    `ADAPTER_FUNCTION_PHASE_COMPLETE`: a phase that raised did not complete. If
+    `phase_fn` raises, the exception propagates and no phase event is emitted, so
+    a consumer reconciling phase counts against invocation counts will see the
+    failure only at invocation level, where `outcome` and `error` carry it.
+
+    Args:
+        name: Adapter function name, used as the metric's `name` field.
+        phase: Lifecycle phase name; must be a valid
+            `AdapterFunctionPhaseCompletePayload.phase` value.
+        phase_fn: The zero-argument callable implementing the phase (e.g.
+            `adapter.weights.activate`).
+    """
+    started_at = time.monotonic()
+    phase_fn()
+    _fire_phase_complete_hook(name, phase, (time.monotonic() - started_at) * 1000.0)
+
+
+def _fire_invocation_complete(
+    *,
+    name: str,
+    revision: str | None,
+    binding_type: str,
+    adapter_type: str,
+    outcome: Literal["success", "schema_error", "error"],
+    error: BaseException | None,
+) -> None:
+    """Fire the `adapter_function_invocation_complete` metric hook.
+
+    Args:
+        name: Adapter function name.
+        revision: Catalog revision of the adapter, or `None` if unpinned.
+        binding_type: Weight-binding reality the adapter ran under.
+        adapter_type: Adapter mechanism (e.g. `"lora"`, `"alora"`).
+        outcome: Invocation outcome.
+        error: The exception raised during invocation, or `None` on success.
+    """
+    if not has_plugins(HookType.ADAPTER_FUNCTION_INVOCATION_COMPLETE):
+        return
+    from ...plugins.hooks.adapter_function import (
+        AdapterFunctionInvocationCompletePayload,
+    )
+
+    payload = AdapterFunctionInvocationCompletePayload(
+        name=name,
+        revision=revision,
+        binding_type=binding_type,
+        adapter_type=adapter_type,
+        outcome=outcome,
+        error=error,
+    )
+    hook_coro = invoke_hook(HookType.ADAPTER_FUNCTION_INVOCATION_COMPLETE, payload)
+    _run_async_in_thread(hook_coro)
+
+
 # The full adapter-input surface `add_adapter` advertises. The legacy abc
 # `Adapter` (LocalFile/PEFT) and the core dataclass adapter (`_AdapterCore`,
 # Embedded/ServerMediated) are disjoint hierarchies, so the accepted type is
@@ -333,7 +439,7 @@ def get_adapter_for_intrinsic(
 # adapter realities they do not implement — the same "reject unsupported reality"
 # contract the reality-specific verbs use. See the module note on the mixin-vs-
 # generic trade-off for why this is a runtime, not a type-parameter, guarantee.
-AdapterInput: TypeAlias = Adapter | _AdapterCore
+AdapterInput: TypeAlias = Adapter | _AdapterCore | LocalFileBinding
 
 
 class AdapterMixin(Backend, abc.ABC):
@@ -429,6 +535,58 @@ def unload_peft_adapter(self, adapter_qualified_name: str) -> None:
             f"Backend type {type(self)} does not support unload_peft_adapter()."
         )
 
+    def activate_peft_adapter(self, adapter_qualified_name: str) -> None:
+        """Switch a previously loaded PEFT adapter on for subsequent generation.
+
+        LocalFile/PEFT reality only (e.g. a locally hosted Hugging Face
+        model). The adapter must have been loaded via `load_peft_adapter`
+        before calling this method.
+
+        Args:
+            adapter_qualified_name (str): The `adapter.qualified_name` of the
+                adapter to activate.
+
+        Raises:
+            NotImplementedError: If this backend's adapter reality is not
+                LocalFile/PEFT.
+        """
+        raise NotImplementedError(
+            f"Backend type {type(self)} does not support activate_peft_adapter()."
+        )
+
+    def deactivate_peft_adapter(self, adapter_qualified_name: str) -> None:
+        """Switch off any active PEFT adapter so generation uses the base model.
+
+        LocalFile/PEFT reality only (e.g. a locally hosted Hugging Face
+        model).
+
+        Args:
+            adapter_qualified_name (str): The `adapter.qualified_name` of the
+                adapter to deactivate. Accepted for symmetry with
+                `activate_peft_adapter`; the underlying primitive clears all
+                active PEFT adapters regardless of name.
+
+        Raises:
+            NotImplementedError: If this backend's adapter reality is not
+                LocalFile/PEFT.
+        """
+        raise NotImplementedError(
+            f"Backend type {type(self)} does not support deactivate_peft_adapter()."
+        )
+
+    def _adapter_activation_lock(
+        self,
+    ) -> contextlib.AbstractContextManager[bool | None]:
+        """Exclusivity lock to hold while calling activate/deactivate verbs.
+
+        Default is a no-op (`contextlib.nullcontext()`). Backends whose
+        activation verbs mutate shared, non-thread-safe state (e.g.
+        `LocalHFBackend`'s underlying PEFT model) override this to return
+        their own lock, so callers like `LocalFileBinding.activate()` get
+        the same exclusivity `_generate_with_adapter_lock` relies on.
+        """
+        return contextlib.nullcontext()
+
     def render_controls(self, adapter_qualified_name: str, active: bool) -> None:
         """Render or clear the control tokens for a baked-in embedded adapter.
 
@@ -537,19 +695,166 @@ def resolve_adapter(self, name: str) -> _AdapterCore:
         if found is not None:
             return found
 
+        # `_find_adapter` only matches `_AdapterCore` entries. If registration
+        # above silently failed because a `LocalFileBinding` already claims a
+        # colliding qualified name (they share `f"{name}_{type}"` with
+        # `IntrinsicAdapter`), say so — the alternative is an opaque KeyError
+        # that gives no hint the two registration paths collided.
+        added = getattr(self, "_added_adapters", {})
+        blocking = next(
+            (
+                v
+                for k, v in added.items()
+                if k.startswith(f"{name}_") and not isinstance(v, _AdapterCore)
+            ),
+            None,
+        )
+        if blocking is not None:
+            blocking_name = getattr(blocking, "qualified_name", None)
+            raise KeyError(
+                f"Adapter {name!r} not found after registration: a "
+                f"{type(blocking).__name__} is already registered under "
+                f"{blocking_name!r}, which collides with {name!r}'s auto-registration "
+                "path. LocalFileBinding and resolve_adapter()/intrinsic-helper "
+                "registrations share the same qualified-name key space on this "
+                "backend and cannot both claim it."
+            )
+
         raise KeyError(f"Adapter {name!r} not found after registration")
 
     @contextlib.contextmanager
     def adapter_scope(self, adapter: "_AdapterCore | None"):  # type: ignore[type-arg]
         """Context manager wrapping adapter activation and deactivation.
 
-        Phase 1 stub — yields immediately (no-op). Phase 2 (see epic #929) wires
-        in `adapter.weights.activate()` and `adapter.weights.deactivate()`.
+        A no-op when `adapter` is `None`. Otherwise: activates
+        `adapter.weights`, yields, then always deactivates — even if the `with`
+        body raises. Each phase fires `ADAPTER_FUNCTION_PHASE_COMPLETE`, and
+        `ADAPTER_FUNCTION_INVOCATION_COMPLETE` fires on the way out, carrying the
+        overall outcome.
+
+        This method fires hooks only; it does not open spans. Span production is a
+        plugin's job (see #1464 for the rule and #1466 for the adapter-function
+        spans), and the `ADAPTER_FUNCTION_*` family currently has no start hook for
+        a plugin to open a span on. See `docs/dev/adapter_observability.md` for the
+        metric schema.
+
+        `deactivate()` is guarded on `activate()`'s own side effect having
+        completed, not on the activate phase's hook dispatch also succeeding.
+        If a plugin subscribed to `ADAPTER_FUNCTION_PHASE_COMPLETE` raises after
+        `activate()` already flipped the adapter on, `deactivate()` still runs —
+        telemetry must not be able to strand the adapter active.
+
+        Not atomic across the whole scope: `_adapter_activation_lock()` is
+        held only inside each of `activate()`/`deactivate()`'s own verb calls
+        (see `LocalFileBinding.activate`), not for the `with` body in between.
+        Two concurrent `adapter_scope()` calls on one backend can therefore
+        interleave — one thread's body can run while a different adapter is
+        active, activated by another thread's call. Widening the lock to span
+        the whole scope was tried and reverted: it deadlocks the moment the
+        body does real async generation (confirmed against
+        `test_local_file_e2e.py`), because that work runs on the shared
+        event-loop thread while this thread holds the lock — a same-thread
+        `RLock` doesn't help across threads. No caller combines concurrent
+        `adapter_scope()` calls today (nothing outside tests calls it at all),
+        so this is latent, not reachable — but #1465 (wiring real generation
+        through this scope) has to solve the atomicity and the threading
+        interaction together, not layer one fix on top of the other.
 
         Args:
-            adapter: The adapter to activate, or `None` (no-op in Phase 1).
+            adapter: The adapter to activate, or `None` (no-op).
+
+        Raises:
+            BaseException: An error raised by activation, the `with` body, or
+                deactivation. If both the body and deactivation fail, the body
+                error remains primary and the deactivation error is chained.
         """
-        yield
+        if adapter is None:
+            yield
+            return
+
+        name = adapter.identity.name
+        # Prefer `resolved_revision()` over the raw `.revision` attribute: a
+        # lazily-resolved binding (`revision=None`) still downloads and runs
+        # against the catalogue's pinned SHA, so reporting the unresolved
+        # `None` would mislabel an effectively-pinned invocation as unpinned.
+        # `resolved_revision()` only exists on `LocalFileBinding`, not the
+        # `WeightsBinding` base, so both the lookup and the call are guarded.
+        revision: str | None
+        if isinstance(adapter.weights, LocalFileBinding):
+            try:
+                revision = adapter.weights.resolved_revision()
+            except Exception:
+                revision = adapter.weights.revision
+        else:
+            revision = cast(str | None, getattr(adapter.weights, "revision", None))
+        binding_type = adapter.weights.binding_type
+        adapter_type = adapter.identity.adapter_type
+
+        outcome: Literal["success", "schema_error", "error"] = "success"
+        exception: BaseException | None = None
+        activated = False
+        body_exception: BaseException | None = None
+        try:
+            started_at = time.monotonic()
+            try:
+                adapter.weights.activate()
+                activated = True
+                _fire_phase_complete_hook(
+                    name, "activate", (time.monotonic() - started_at) * 1000.0
+                )
+                try:
+                    yield
+                except BaseException as exc:
+                    body_exception = exc
+                    raise
+            finally:
+                if activated:
+                    try:
+                        _run_adapter_phase(
+                            name, "deactivate", adapter.weights.deactivate
+                        )
+                    except BaseException as deactivate_exc:
+                        if body_exception is None:
+                            raise
+                        body_exception.add_note(
+                            "Adapter deactivation also failed: "
+                            f"{type(deactivate_exc).__name__}: {deactivate_exc}"
+                        )
+        except AdapterSchemaMismatchError as exc:
+            # Distinct from a generic error: this is the schema-drift signal the
+            # `parse_failures` counter exists to detect, so collapsing it into
+            # "error" would leave that counter permanently at zero. Reachable
+            # today — `adapter_scope` is public, so a caller can parse inside the
+            # scope — and it becomes the common case once #1465 moves generation
+            # and parsing in here.
+            outcome = "schema_error"
+            exception = exc
+            raise
+        except BaseException as exc:
+            outcome = "error"
+            exception = exc
+            raise
+        finally:
+            # A hook-dispatch failure here must not replace or mask the real
+            # outcome computed above — that would turn a clean `with` block
+            # into a thrown error, or swap a genuine body exception for a
+            # telemetry-plumbing one. Log and swallow instead.
+            try:
+                _fire_invocation_complete(
+                    name=name,
+                    revision=revision,
+                    binding_type=binding_type,
+                    adapter_type=adapter_type,
+                    outcome=outcome,
+                    error=exception,
+                )
+            except Exception:
+                MelleaLogger.get_logger().warning(
+                    f"adapter_function_invocation_complete hook dispatch failed for "
+                    f"{name!r}; ignoring so it doesn't mask the real outcome "
+                    f"({outcome!r}).",
+                    exc_info=True,
+                )
 
     def _find_adapter(
         self, capability: str, adapter_types: tuple[str, ...] | None = None
diff --git a/mellea/backends/adapters/catalog.py b/mellea/backends/adapters/catalog.py
index 488abfb69..4889d9112 100644
--- a/mellea/backends/adapters/catalog.py
+++ b/mellea/backends/adapters/catalog.py
@@ -72,9 +72,7 @@ class IntrinsicsCatalogEntry(pydantic.BaseModel):
         revision (str): Hugging Face revision — branch name, tag, or commit SHA.
             Catalogue entries pin to commit SHAs by convention so loads are
             reproducible; the validator itself only requires a non-empty string.
-            Note: this field is stored in the catalogue but not yet forwarded to
-            the Hugging Face download call; wiring it through is deferred to a
-            subsequent phase of the adapter-lifecycle epic (#929).
+            The revision is forwarded to Hugging Face download calls.
         adapter_types (tuple[AdapterType, ...]): Adapter types known to be
             available for this adapter function; defaults to
             `(AdapterType.LORA, AdapterType.ALORA)`.
diff --git a/mellea/backends/huggingface.py b/mellea/backends/huggingface.py
index 0d66a92a8..870e9817d 100644
--- a/mellea/backends/huggingface.py
+++ b/mellea/backends/huggingface.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import dataclasses
 import datetime
 import functools
@@ -76,6 +77,7 @@
 from ..telemetry.context import generate_request_id, with_context
 from ._options import resolve_model_options
 from .adapters import AdapterMixin, IntrinsicAdapter, LocalHFAdapter
+from .adapters._core import LocalFileBinding
 from .adapters.adapter import AdapterInput
 from .backend import FormatterBackend
 from .cache import Cache, SimpleLRUCache
@@ -445,8 +447,8 @@ def __init__(
         )
 
         # Adapters can be made known to the backend (added) and loaded.
-        self._added_adapters: dict[str, LocalHFAdapter] = {}
-        self._loaded_adapters: dict[str, LocalHFAdapter] = {}
+        self._added_adapters: dict[str, LocalHFAdapter | LocalFileBinding] = {}
+        self._loaded_adapters: dict[str, LocalHFAdapter | LocalFileBinding] = {}
 
         self._generation_lock = threading.Lock()
         """Used to force generation requests to be non-concurrent. Necessary for preventing issues with adapters."""
@@ -565,19 +567,9 @@ def _generate_with_adapter_lock(
         with self._generation_lock:
             if adapter_name != "":
                 self.load_peft_adapter(adapter_name)
-                self._model.set_adapter(adapter_name)
+                self.activate_peft_adapter(adapter_name)
             else:
-                try:
-                    # `._model.disable_adapters()` doesn't seem to actually disable them or
-                    # remove them from the model's list of `.active_adapters()`.
-                    self._model.set_adapter([])
-                except ValueError as e:
-                    # If no weights have been loaded, the model will raise a ValueError:
-                    # `ValueError("No adapter loaded. Please load an adapter first.")`
-                    if "No adapter loaded" in str(e):
-                        pass
-                    else:
-                        raise e
+                self.deactivate_peft_adapter(adapter_name)
 
             _assert_correct_adapters(adapter_name, self._model)
             out = generate_func(*args, **kwargs)
@@ -2005,15 +1997,16 @@ def add_adapter(self, adapter: AdapterInput) -> None:
 
         Args:
             adapter (AdapterInput): The adapter to register. Must be a
-                `LocalHFAdapter`; other adapter realities are rejected.
+                `LocalHFAdapter` or `LocalFileBinding`; other adapter realities
+                are rejected.
 
         Raises:
-            TypeError: If `adapter` is not a `LocalHFAdapter`.
+            TypeError: If `adapter` is not a `LocalHFAdapter` or `LocalFileBinding`.
             Exception: If `adapter` has already been added to a different backend.
         """
-        if not isinstance(adapter, LocalHFAdapter):
+        if not isinstance(adapter, (LocalHFAdapter, LocalFileBinding)):
             raise TypeError(
-                f"LocalHFBackend requires a LocalHFAdapter; got "
+                f"LocalHFBackend requires a LocalHFAdapter or LocalFileBinding; got "
                 f"{type(adapter).__name__}."
             )
         if adapter.backend is not None:
@@ -2027,9 +2020,15 @@ def add_adapter(self, adapter: AdapterInput) -> None:
                     f"adapter {adapter.name} with type {adapter.adapter_type} has already been added to backend {adapter.backend}"
                 )
 
-        if self._added_adapters.get(adapter.qualified_name) is not None:
+        existing = self._added_adapters.get(adapter.qualified_name)
+        if existing is not None:
             MelleaLogger.get_logger().warning(
-                f"Client code attempted to add {adapter.name} with type {adapter.adapter_type} but {adapter.name} was already added to {self.__class__}. The backend is refusing to do this, because adapter loading is not idempotent."
+                f"Client code attempted to add {adapter.name} with type {adapter.adapter_type} "
+                f"but {adapter.qualified_name!r} is already registered as a "
+                f"{type(existing).__name__} on {self.__class__.__name__}. The backend is "
+                "refusing to do this, because adapter loading is not idempotent. "
+                "LocalFileBinding and IntrinsicAdapter/resolve_adapter() registrations "
+                "share this qualified-name key space and cannot both claim the same name."
             )
             return None
 
@@ -2103,6 +2102,69 @@ def unload_peft_adapter(self, adapter_qualified_name: str) -> None:
         # Remove the adapter from the list of loaded adapters.
         del self._loaded_adapters[adapter.qualified_name]
 
+    def activate_peft_adapter(self, adapter_qualified_name: str) -> None:
+        """Switch a previously loaded PEFT adapter on for subsequent generation.
+
+        Must be called while holding `_generation_lock`.
+
+        Args:
+            adapter_qualified_name (str): The `adapter.qualified_name` of the adapter
+                to activate.
+        """
+        self._model.set_adapter(adapter_qualified_name)
+
+    def deactivate_peft_adapter(self, adapter_qualified_name: str) -> None:
+        """Switch off any active PEFT adapter so generation uses the base model.
+
+        Must be called while holding `_generation_lock`.
+
+        Args:
+            adapter_qualified_name (str): The `adapter.qualified_name` of the adapter
+                to deactivate. Accepted for symmetry with `activate_peft_adapter`; the
+                underlying primitive clears all active PEFT adapters regardless of
+                name.
+
+        Raises:
+            ValueError: If the underlying PEFT model raises `ValueError` for a
+                reason other than "no adapter loaded" (which is treated as a
+                no-op, since deactivating is already a no-op in that case).
+        """
+        try:
+            # `._model.disable_adapters()` doesn't seem to actually disable them or
+            # remove them from the model's list of `.active_adapters()`.
+            self._model.set_adapter([])
+        except ValueError as e:
+            # If no weights have been loaded, the model will raise a ValueError:
+            # `ValueError("No adapter loaded. Please load an adapter first.")`
+            if "No adapter loaded" not in str(e):
+                raise e
+
+    def _adapter_activation_lock(
+        self,
+    ) -> contextlib.AbstractContextManager[bool | None]:
+        """Reuse `_generation_lock` for exclusivity around adapter activation.
+
+        `activate_peft_adapter`/`deactivate_peft_adapter` document "must be called
+        while holding `_generation_lock`" as a precondition, and they have two
+        callers:
+
+        1. `_generate_with_adapter_lock`, which takes `_generation_lock` itself.
+        2. `LocalFileBinding.activate()`/`.deactivate()` (driven by
+           `AdapterMixin.adapter_scope()`), which holds no lock of its own.
+
+        This exists for caller 2 — so it is not a duplicate of the lock caller 1
+        takes, it is the only thing satisfying the precondition on that path.
+
+        TODO(#1465): `_generation_lock` is a plain non-reentrant `threading.Lock`.
+        Nothing nests these two callers today, because generation still runs
+        outside `adapter_scope`. When #1465 moves the model call inside the scope,
+        `activate()` will re-acquire this lock while `_generate_with_adapter_lock`
+        already holds it and deadlock. #1465 owns the fix (make it reentrant, or
+        restructure so the scope takes the lock once); it must not be resolved by
+        dropping the lock here, which would leave caller 2's precondition unmet.
+        """
+        return self._generation_lock
+
     def list_adapters(self) -> list[str]:
         """List the qualified names of all adapters registered with this backend.
 
diff --git a/mellea/helpers/event_loop_helper.py b/mellea/helpers/event_loop_helper.py
index 6ade28d22..426866acd 100644
--- a/mellea/helpers/event_loop_helper.py
+++ b/mellea/helpers/event_loop_helper.py
@@ -78,20 +78,56 @@ def __call__(self, co: Coroutine[Any, Any, R]) -> R:
         The caller's `contextvars` snapshot is applied inside the new Task so
         contextvar-backed state is visible to the coroutine. One-way:
         mutations inside the Task don't leak back.
-        """
-        self._reinit_if_forked()
-        if self._event_loop == get_current_event_loop():
-            # If this gets called from the same event loop, launch in a separate thread to prevent blocking.
-            return _EventLoopHandler()(co)
-
-        parent_ctx = contextvars.copy_context()
 
-        async def _wrapped() -> R:
-            for var, value in parent_ctx.items():
-                var.set(value)
-            return await co
-
-        return asyncio.run_coroutine_threadsafe(_wrapped(), self._event_loop).result()
+        `_reinit_if_forked`/`get_current_event_loop` run before the coroutine
+        is scheduled, so an exception there (or a failure to schedule) would
+        otherwise leave `co` unawaited and emit a "coroutine was never
+        awaited" warning on top of whatever raised. `co.close()` covers that:
+        if scheduling never succeeded, `co` hasn't started and closing it is
+        a clean no-op; if `.result()` raised the task's own exception instead,
+        the future is only marked done after the scheduled task has fully
+        finished running `co`, so closing an already-completed coroutine is
+        also a no-op.
+
+        Deliberately `except Exception`, not `BaseException`: a `Future.result()`
+        with no timeout blocks on a `threading.Condition`, and a `KeyboardInterrupt`
+        delivered to this (calling) thread can unblock that wait before the task
+        on the event-loop thread has actually finished — `co` may still be running
+        there. Closing it from here at that point would run `co`'s `GeneratorExit`
+        handling in this thread while the event loop thread is concurrently
+        stepping the same frame, which is not safe. Excluding
+        `KeyboardInterrupt`/`SystemExit` means that rare case leaks the
+        unawaited-coroutine warning instead of risking that race.
+
+        `_wrapped()`'s own coroutine object needs the same treatment as `co`:
+        if scheduling fails before `run_coroutine_threadsafe` hands it to the
+        loop, `_wrapped()` never starts and never gets to `await co` — leaking
+        both `_wrapped()`'s and (via the outer `except`) `co`'s "coroutine was
+        never awaited" warnings otherwise.
+        """
+        wrapped_co: Coroutine[Any, Any, R] | None = None
+        try:
+            self._reinit_if_forked()
+            if self._event_loop == get_current_event_loop():
+                # If this gets called from the same event loop, launch in a separate thread to prevent blocking.
+                return _EventLoopHandler()(co)
+
+            parent_ctx = contextvars.copy_context()
+
+            async def _wrapped() -> R:
+                for var, value in parent_ctx.items():
+                    var.set(value)
+                return await co
+
+            wrapped_co = _wrapped()
+            return asyncio.run_coroutine_threadsafe(
+                wrapped_co, self._event_loop
+            ).result()
+        except Exception:
+            co.close()
+            if wrapped_co is not None:
+                wrapped_co.close()
+            raise
 
 
 # Instantiate this class once. It will not be re-instantiated.
diff --git a/mellea/telemetry/metrics_plugins.py b/mellea/telemetry/metrics_plugins.py
index 741ad30d4..0c515a128 100644
--- a/mellea/telemetry/metrics_plugins.py
+++ b/mellea/telemetry/metrics_plugins.py
@@ -476,10 +476,8 @@ class AdapterFunctionMetricsPlugin(
     """Records adapter function invocation and phase-duration metrics.
 
     Hooks into `adapter_function_invocation_complete` and
-    `adapter_function_phase_complete`. No production call site fires these
-    hooks yet — real `prepare`/`activate`/`generate`/`parse`/`deactivate`
-    wiring lands with the LocalFileBinding and EmbeddedBinding lifecycle work
-    (Epic #929 Phase 2 follow-ups).
+    `adapter_function_phase_complete`. `phase` is one of the values in
+    `AdapterFunctionPhaseCompletePayload`'s `phase` field.
     """
 
     @hook("adapter_function_invocation_complete", mode=PluginMode.FIRE_AND_FORGET)
diff --git a/test/backends/test_adapters/__init__.py b/test/backends/test_adapters/__init__.py
new file mode 100644
index 000000000..4cdd3f38b
--- /dev/null
+++ b/test/backends/test_adapters/__init__.py
@@ -0,0 +1,2 @@
+# Copyright IBM Corp. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
diff --git a/test/backends/test_adapters/_hook_capture.py b/test/backends/test_adapters/_hook_capture.py
new file mode 100644
index 000000000..67429030c
--- /dev/null
+++ b/test/backends/test_adapters/_hook_capture.py
@@ -0,0 +1,90 @@
+# Copyright IBM Corp. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared helper for asserting on the adapter-function hooks (Epic #929).
+
+Single home for the hook-capture idiom, so the patching contract below is stated
+once. Not a `conftest.py` fixture: callers need to wrap a *specific* block inside
+a test (the integration tests capture only the `adapter_scope` section, not the
+whole test), which a fixture cannot express.
+
+Assertions here are on **hooks, not spans**. `adapter_scope` fires hooks and never
+opens a span — #1464 documents that rule, #1466 adds the spans from a plugin.
+"""
+
+import contextlib
+from collections.abc import Iterator
+from unittest.mock import MagicMock, patch
+
+_TARGET = "mellea.backends.adapters.adapter"
+
+
+@contextlib.contextmanager
+def capture_adapter_hooks() -> Iterator[MagicMock]:
+    """Capture the hook payloads fired inside the block.
+
+    Patches three things, each for a distinct reason:
+
+    - **`has_plugins` pinned `True`.** It is already `True` under pytest —
+      `test/conftest.py`'s `auto_register_acceptance_sets` is `autouse`,
+      session-scoped, and registers a plugin for every `HookType`
+      (`test/plugins/_acceptance_sets.py`). Pinning it removes the dependency on
+      that ambient registration.
+    - **`invoke_hook` replaced with `new_callable=MagicMock`.** Load-bearing:
+      `invoke_hook` is an `async def`, so a bare `patch()` auto-creates an
+      `AsyncMock`. Calling an `AsyncMock` returns a coroutine, and if a
+      `side_effect` returns a coroutine of its own, *that* inner coroutine becomes
+      the outer one's result and is never awaited — surfacing as
+      `PytestUnraisableExceptionWarning: coroutine ... was never awaited`. Note
+      `-W error::RuntimeWarning` does **not** catch it; use
+      `-W error::pytest.PytestUnraisableExceptionWarning`. Forcing a sync
+      `MagicMock` means no coroutine exists to leak.
+    - **`_run_async_in_thread` patched out.** Real dispatch works fine; it is
+      simply not needed to read the payloads, and skipping it keeps these tests
+      off the shared event loop.
+
+    Yields:
+        The `invoke_hook` mock. Use `hook_payloads()` to read what it recorded.
+    """
+    with (
+        patch(f"{_TARGET}.has_plugins", return_value=True),
+        patch(f"{_TARGET}.invoke_hook", new_callable=MagicMock) as mock_invoke,
+        patch(f"{_TARGET}._run_async_in_thread"),
+    ):
+        yield mock_invoke
+
+
+def hook_payloads(mock_invoke: MagicMock) -> list:
+    """Returns the payload argument of every recorded `invoke_hook` call, in order.
+
+    Args:
+        mock_invoke: The mock yielded by `capture_adapter_hooks`.
+
+    Returns:
+        Each call's payload, ordered as fired.
+    """
+    return [call.args[1] for call in mock_invoke.call_args_list]
+
+
+def phase_payloads(mock_invoke: MagicMock) -> list:
+    """Returns only the phase-complete payloads.
+
+    Args:
+        mock_invoke: The mock yielded by `capture_adapter_hooks`.
+
+    Returns:
+        The recorded `AdapterFunctionPhaseCompletePayload`s, ordered as fired.
+    """
+    return [p for p in hook_payloads(mock_invoke) if hasattr(p, "phase")]
+
+
+def invocation_payloads(mock_invoke: MagicMock) -> list:
+    """Returns only the invocation-complete payloads.
+
+    Args:
+        mock_invoke: The mock yielded by `capture_adapter_hooks`.
+
+    Returns:
+        The recorded `AdapterFunctionInvocationCompletePayload`s, ordered as fired.
+    """
+    return [p for p in hook_payloads(mock_invoke) if hasattr(p, "outcome")]
diff --git a/test/backends/test_adapters/test_adapter.py b/test/backends/test_adapters/test_adapter.py
index 8cf216cf3..142ea4fdd 100644
--- a/test/backends/test_adapters/test_adapter.py
+++ b/test/backends/test_adapters/test_adapter.py
@@ -2,10 +2,12 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import pathlib
+from unittest.mock import patch
 
 import pytest
 
 from mellea.backends.adapters import IntrinsicAdapter
+from mellea.backends.adapters.catalog import fetch_intrinsic_metadata
 
 
 # The backend tests handle most of the adapter testing. Do a basic test here
@@ -20,5 +22,41 @@ def test_adapter_init():
     assert adapter.config["parameters"]["max_completion_tokens"] == 6
 
 
+def test_init_forwards_pinned_revision_to_obtain_io_yaml():
+    """Regression guard (issue #1141): when no config_file/config_dict is given,
+    __init__ must forward the catalog's pinned revision to obtain_io_yaml, not the
+    default `"main"`.
+    """
+    pinned_revision = fetch_intrinsic_metadata("answerability").revision
+
+    with patch(
+        "mellea.formatters.granite.intrinsics.obtain_io_yaml"
+    ) as mock_obtain_io_yaml:
+        mock_obtain_io_yaml.return_value = (
+            pathlib.Path(__file__).parent / "intrinsics-data" / "answerability.yaml"
+        )
+        IntrinsicAdapter("answerability", base_model_name="granite-3.3-8b-instruct")
+
+    assert mock_obtain_io_yaml.call_args.kwargs["revision"] == pinned_revision
+    assert pinned_revision != "main"
+
+
+def test_download_and_get_path_forwards_pinned_revision_to_obtain_lora():
+    """Regression guard (issue #1141): download_and_get_path must forward the
+    catalog's pinned revision to obtain_lora, not the default `"main"`.
+    """
+    dir_file = pathlib.Path(__file__).parent.joinpath("intrinsics-data")
+    answerability_file = f"{dir_file}/answerability.yaml"
+    adapter = IntrinsicAdapter("answerability", config_file=answerability_file)
+    pinned_revision = adapter.intrinsic_metadata.revision
+
+    with patch("mellea.formatters.granite.intrinsics.obtain_lora") as mock_obtain_lora:
+        mock_obtain_lora.return_value = pathlib.Path("/fake/adapter/path")
+        adapter.download_and_get_path("granite-3.3-8b-instruct")
+
+    assert mock_obtain_lora.call_args.kwargs["revision"] == pinned_revision
+    assert pinned_revision != "main"
+
+
 if __name__ == "__main__":
     pytest.main([__file__])
diff --git a/test/backends/test_adapters/test_adapter_mixin.py b/test/backends/test_adapters/test_adapter_mixin.py
index ba99c8992..21384946d 100644
--- a/test/backends/test_adapters/test_adapter_mixin.py
+++ b/test/backends/test_adapters/test_adapter_mixin.py
@@ -4,9 +4,9 @@
 """Unit tests for the narrowed AdapterMixin verb contract (Epic #929 Phase 2, issue #1140).
 
 Verifies that:
-  - the four reality-specific verbs (`load_peft_adapter`, `unload_peft_adapter`,
-    `render_controls`, `set_request_adapter`) raise `NotImplementedError` by
-    default on the mixin
+  - the reality-specific verbs (`load_peft_adapter`, `unload_peft_adapter`,
+    `activate_peft_adapter`, `deactivate_peft_adapter`, `render_controls`,
+    `set_request_adapter`) raise `NotImplementedError` by default on the mixin
   - each concrete backend overrides only the verb(s) matching its own adapter
     reality, leaving the others on the default (raising) implementation
 """
@@ -22,6 +22,8 @@
 _REALITY_SPECIFIC_VERBS = (
     "load_peft_adapter",
     "unload_peft_adapter",
+    "activate_peft_adapter",
+    "deactivate_peft_adapter",
     "render_controls",
     "set_request_adapter",
 )
@@ -39,9 +41,11 @@ def test_default_reality_specific_verb_raises_not_implemented(verb):
 
 
 def test_hf_backend_overrides_only_peft_verbs():
-    """LocalHFBackend (LocalFile/PEFT reality) overrides load/unload_peft_adapter only."""
+    """LocalHFBackend (LocalFile/PEFT reality) overrides the PEFT verbs only."""
     assert "load_peft_adapter" in vars(LocalHFBackend)
     assert "unload_peft_adapter" in vars(LocalHFBackend)
+    assert "activate_peft_adapter" in vars(LocalHFBackend)
+    assert "deactivate_peft_adapter" in vars(LocalHFBackend)
     assert "render_controls" not in vars(LocalHFBackend)
     assert "set_request_adapter" not in vars(LocalHFBackend)
 
@@ -58,3 +62,26 @@ def test_no_backend_implements_server_mediated_reality():
     """set_request_adapter has no concrete implementation anywhere yet."""
     assert "set_request_adapter" not in vars(LocalHFBackend)
     assert "set_request_adapter" not in vars(OpenAIBackend)
+
+
+def test_default_adapter_activation_lock_is_a_noop():
+    """The mixin default is a no-op context manager, not a real lock.
+
+    Backends whose activation verbs mutate shared, non-thread-safe state
+    override this; a backend that doesn't (nothing to protect) gets a
+    `nullcontext` for free rather than having to implement one.
+    """
+    mock_backend = MagicMock(spec=AdapterMixin)
+
+    with AdapterMixin._adapter_activation_lock(mock_backend):
+        pass
+
+
+def test_hf_backend_overrides_adapter_activation_lock():
+    """LocalHFBackend overrides the lock (it has shared PEFT model state to protect)."""
+    assert "_adapter_activation_lock" in vars(LocalHFBackend)
+
+
+def test_openai_backend_does_not_override_adapter_activation_lock():
+    """OpenAIBackend has no PEFT model state, so it keeps the mixin's no-op default."""
+    assert "_adapter_activation_lock" not in vars(OpenAIBackend)
diff --git a/test/backends/test_adapters/test_adapter_scope.py b/test/backends/test_adapters/test_adapter_scope.py
new file mode 100644
index 000000000..85e3af14f
--- /dev/null
+++ b/test/backends/test_adapters/test_adapter_scope.py
@@ -0,0 +1,382 @@
+# Copyright IBM Corp. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the real AdapterMixin.adapter_scope implementation (issue #1141).
+
+Exercises activate/deactivate ordering and exception-safety using a fake
+WeightsBinding double — no real backend or model required. `adapter_scope` fires
+metric hooks only and opens no spans, so no exporter is involved; the hook
+dispatch safely no-ops when no plugins are registered.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from mellea.backends.adapters._core import (
+    Adapter,
+    AdapterSchemaMismatchError,
+    Identity,
+    IOContract,
+    LocalFileBinding,
+)
+from mellea.backends.adapters.adapter import AdapterMixin
+from mellea.backends.adapters.catalog import AdapterType
+from mellea.core import Component
+from test.backends.test_adapters._hook_capture import (
+    capture_adapter_hooks,
+    invocation_payloads,
+)
+
+
+class _Contract(IOContract):
+    def build_prompt(self, **kwargs: object) -> Component:
+        raise NotImplementedError
+
+    def parse(self, raw: str) -> dict[str, object]:
+        return {}
+
+
+def _make_adapter():
+    weights = MagicMock(spec=LocalFileBinding)
+    weights.binding_type = "local_file"
+    weights.revision = "abc123"
+    # `MagicMock(spec=LocalFileBinding)` passes `isinstance(weights,
+    # LocalFileBinding)`, so `adapter_scope` calls `resolved_revision()` on it
+    # (real `LocalFileBinding`s resolve lazily); stub it to match `.revision`
+    # since these tests aren't exercising lazy resolution itself.
+    weights.resolved_revision.return_value = "abc123"
+    identity = Identity(name="answerability", adapter_type="lora")
+    adapter = Adapter(identity=identity, io_contract=_Contract(), weights=weights)
+    return adapter, weights
+
+
+def test_adapter_scope_activates_then_deactivates_in_order():
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, weights = _make_adapter()
+    calls = []
+    weights.activate.side_effect = lambda: calls.append("activate")
+    weights.deactivate.side_effect = lambda: calls.append("deactivate")
+
+    with AdapterMixin.adapter_scope(mock_backend, adapter):
+        calls.append("body")
+
+    assert calls == ["activate", "body", "deactivate"]
+
+
+def test_adapter_scope_deactivates_even_when_body_raises():
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, weights = _make_adapter()
+
+    with pytest.raises(RuntimeError, match="boom"):
+        with AdapterMixin.adapter_scope(mock_backend, adapter):
+            raise RuntimeError("boom")
+
+    weights.activate.assert_called_once()
+    weights.deactivate.assert_called_once()
+
+
+def test_adapter_scope_preserves_body_error_when_deactivate_also_raises():
+    """The body's exception remains primary when cleanup also fails."""
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, weights = _make_adapter()
+    body_error = ValueError("body failed")
+    original_cause = KeyError("original cause")
+    deactivate_error = RuntimeError("deactivation failed")
+    weights.deactivate.side_effect = deactivate_error
+
+    with capture_adapter_hooks() as mock_invoke:
+        with pytest.raises(ValueError, match="body failed") as exc_info:
+            with AdapterMixin.adapter_scope(mock_backend, adapter):
+                raise body_error from original_cause
+
+    assert exc_info.value is body_error
+    assert exc_info.value.__cause__ is original_cause
+    assert any(
+        "RuntimeError: deactivation failed" in note for note in exc_info.value.__notes__
+    )
+    weights.deactivate.assert_called_once()
+    invocations = invocation_payloads(mock_invoke)
+    assert [p.outcome for p in invocations] == ["error"]
+    assert invocations[0].error is body_error
+
+
+def test_adapter_scope_deactivates_even_when_activate_raises():
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, weights = _make_adapter()
+    weights.activate.side_effect = RuntimeError("activation failed")
+
+    with pytest.raises(RuntimeError, match="activation failed"):
+        with AdapterMixin.adapter_scope(mock_backend, adapter):
+            pytest.fail("body must not run when activate() raises")
+
+    weights.deactivate.assert_not_called()
+
+
+@pytest.mark.parametrize("failing_phase", ["activate", "deactivate"])
+def test_adapter_scope_ignores_phase_hook_dispatch_failures(failing_phase: str):
+    """A phase-hook failure must not break an otherwise successful scope."""
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, weights = _make_adapter()
+    body_ran = False
+
+    def _raise_on_phase_hook(hook_type: object, payload: object) -> None:
+        if getattr(payload, "phase", None) == failing_phase:
+            raise RuntimeError("plugin dispatch blew up")
+
+    with (
+        patch("mellea.backends.adapters.adapter.has_plugins", return_value=True),
+        patch(
+            "mellea.plugins.hooks.adapter_function.AdapterFunctionPhaseCompletePayload",
+            side_effect=lambda **kwargs: SimpleNamespace(**kwargs),
+        ),
+        patch(
+            "mellea.backends.adapters.adapter.invoke_hook",
+            side_effect=_raise_on_phase_hook,
+        ),
+    ):
+        with AdapterMixin.adapter_scope(mock_backend, adapter):
+            body_ran = True
+
+    assert body_ran
+    weights.activate.assert_called_once()
+    weights.deactivate.assert_called_once()
+
+
+def test_adapter_scope_propagates_deactivate_error_over_body_success():
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, weights = _make_adapter()
+    weights.deactivate.side_effect = RuntimeError("deactivation failed")
+
+    with capture_adapter_hooks() as mock_invoke:
+        with pytest.raises(RuntimeError, match="deactivation failed"):
+            with AdapterMixin.adapter_scope(mock_backend, adapter):
+                pass
+
+    weights.activate.assert_called_once()
+    weights.deactivate.assert_called_once()
+
+    # A body that succeeded does not make the invocation a success: the failure
+    # came from deactivate, and the invocation must still report it.
+    invocations = invocation_payloads(mock_invoke)
+    assert [p.outcome for p in invocations] == ["error"]
+    assert isinstance(invocations[0].error, RuntimeError)
+
+
+def test_adapter_scope_reports_schema_mismatch_as_schema_error():
+    """An AdapterSchemaMismatchError is `schema_error`, not a generic `error`.
+
+    `mellea.adapter_function.parse_failures` increments only on `schema_error`, so
+    collapsing this into `error` would leave that counter permanently at zero.
+    """
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, _ = _make_adapter()
+
+    with capture_adapter_hooks() as mock_invoke:
+        with pytest.raises(AdapterSchemaMismatchError):
+            with AdapterMixin.adapter_scope(mock_backend, adapter):
+                raise AdapterSchemaMismatchError(
+                    "answerability",
+                    frozenset({"wrong_key"}),
+                    frozenset({"answerability"}),
+                )
+
+    invocations = invocation_payloads(mock_invoke)
+    assert [p.outcome for p in invocations] == ["schema_error"]
+    assert isinstance(invocations[0].error, AdapterSchemaMismatchError)
+
+
+def test_adapter_scope_reports_other_exceptions_as_error():
+    """Anything that is not a schema mismatch stays `error`."""
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, _ = _make_adapter()
+
+    with capture_adapter_hooks() as mock_invoke:
+        with pytest.raises(RuntimeError, match="boom"):
+            with AdapterMixin.adapter_scope(mock_backend, adapter):
+                raise RuntimeError("boom")
+
+    assert [p.outcome for p in invocation_payloads(mock_invoke)] == ["error"]
+
+
+def test_phase_hook_not_fired_when_the_phase_itself_fails():
+    """A phase that raised did not complete, so no phase event is emitted.
+
+    `ADAPTER_FUNCTION_PHASE_COMPLETE` means the phase finished. The failure is
+    reported once, at invocation level, where `outcome`/`error` carry it — so a
+    consumer reconciling phase counts against invocation counts sees one
+    invocation error and no phase event, not both.
+    """
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, weights = _make_adapter()
+    weights.activate.side_effect = RuntimeError("activation failed")
+
+    with capture_adapter_hooks() as mock_invoke:
+        with pytest.raises(RuntimeError, match="activation failed"):
+            with AdapterMixin.adapter_scope(mock_backend, adapter):
+                pytest.fail("body must not run when activate() raises")
+
+    payloads = [c.args[1] for c in mock_invoke.call_args_list]
+    assert [p for p in payloads if hasattr(p, "phase")] == []
+
+    invocations = invocation_payloads(mock_invoke)
+    assert [p.outcome for p in invocations] == ["error"]
+
+
+def test_adapter_scope_reports_resolved_revision_not_raw_none():
+    """A lazily-resolved binding (revision=None) must report its resolved pin, not None.
+
+    Regression guard: `adapter_scope` used to read the raw `.revision`
+    attribute, which is `None` for a `LocalFileBinding(name=..., revision=None)`
+    even though the binding downloads and runs against a concrete catalogue
+    pin. Reporting `None` mislabels an effectively-pinned invocation as
+    unpinned in telemetry.
+    """
+    mock_backend = MagicMock(spec=AdapterMixin)
+    binding = LocalFileBinding(name="answerability")  # revision=None, lazily resolved
+    identity = Identity(name="answerability", adapter_type="lora")
+    adapter = Adapter(identity=identity, io_contract=_Contract(), weights=binding)
+    binding.activate = MagicMock()
+    binding.deactivate = MagicMock()
+    assert binding.revision is None
+
+    with capture_adapter_hooks() as mock_invoke:
+        with AdapterMixin.adapter_scope(mock_backend, adapter):
+            pass
+
+    invocations = invocation_payloads(mock_invoke)
+    assert len(invocations) == 1
+    assert invocations[0].revision == binding.resolved_revision()
+    assert invocations[0].revision != "main"
+
+
+def test_adapter_scope_swallows_invocation_hook_failure_on_clean_run():
+    """A failing invocation-complete hook must not turn a clean run into an error.
+
+    Regression guard: `_fire_invocation_complete` used to be called unguarded
+    in the outer `finally`. If its hook dispatch raised, that exception
+    replaced the (successful, no-exception) outcome of an otherwise-clean
+    `with` block — telemetry turning success into failure.
+    """
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, weights = _make_adapter()
+
+    with patch(
+        "mellea.backends.adapters.adapter._fire_invocation_complete",
+        side_effect=RuntimeError("invocation hook dispatch blew up"),
+    ):
+        with AdapterMixin.adapter_scope(mock_backend, adapter):
+            pass  # must not raise despite the hook failing on exit
+
+    weights.activate.assert_called_once()
+    weights.deactivate.assert_called_once()
+
+
+def test_adapter_scope_invocation_hook_failure_does_not_mask_body_exception():
+    """A failing invocation-complete hook must not replace the body's real exception.
+
+    Regression guard: when both the body and the invocation hook raise, the
+    caller must still see the body's exception, not the hook's.
+    """
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter, weights = _make_adapter()
+
+    with patch(
+        "mellea.backends.adapters.adapter._fire_invocation_complete",
+        side_effect=RuntimeError("invocation hook dispatch blew up"),
+    ):
+        with pytest.raises(ValueError, match="the real failure"):
+            with AdapterMixin.adapter_scope(mock_backend, adapter):
+                raise ValueError("the real failure")
+
+    weights.deactivate.assert_called_once()
+
+
+def test_adapter_scope_is_not_atomic_across_concurrent_calls():
+    """Known limitation, not a guarantee: two concurrent `adapter_scope()`
+    calls on one backend can interleave.
+
+    `_adapter_activation_lock()` is held only inside each of
+    `activate()`/`deactivate()`'s own verb calls, not across the `with` body
+    in between — so a second thread's full activate-body-deactivate cycle can
+    run while the first thread's body is still executing, leaving the first
+    thread's body observing a different adapter (or none) active.
+
+    Widening the lock to span the whole scope was tried and reverted: it
+    deadlocks the real async generation path (see the docstring note on
+    `adapter_scope`). This test pins today's actual (non-atomic) behaviour so
+    it doesn't get silently "fixed" back to interleaving by an unrelated
+    change, or silently broken worse. #1465 owns making this atomic, together
+    with the threading model for real generation.
+    """
+    import threading
+
+    class _FakeBackend:
+        def __init__(self) -> None:
+            self._lock = threading.Lock()
+            self.active: str | None = None
+
+        def _adapter_activation_lock(self):
+            return self._lock
+
+        def activate_peft_adapter(self, name: str) -> None:
+            self.active = name
+
+        def deactivate_peft_adapter(self, name: str) -> None:
+            self.active = None
+
+    def _make(backend: _FakeBackend, name: str):
+        binding = LocalFileBinding(name=name, adapter_type=AdapterType.LORA)
+        binding.backend = backend  # type: ignore[assignment]
+        binding._loaded = (
+            True  # bypass prepare(); this test is about activate/deactivate
+        )
+        identity = Identity(name=name, adapter_type="lora")
+        return Adapter(identity=identity, io_contract=_Contract(), weights=binding)
+
+    backend = _FakeBackend()
+    a1 = _make(backend, "adapter_one")
+    a2 = _make(backend, "adapter_two")
+
+    observed_inside_a1_body: list[str | None] = []
+    a1_activated = threading.Event()
+    a2_done = threading.Event()
+
+    def thread1() -> None:
+        with AdapterMixin.adapter_scope(backend, a1):
+            a1_activated.set()
+            a2_done.wait(timeout=2)
+            observed_inside_a1_body.append(backend.active)
+
+    def thread2() -> None:
+        a1_activated.wait(timeout=2)
+        with AdapterMixin.adapter_scope(backend, a2):
+            pass
+        a2_done.set()
+
+    t1 = threading.Thread(target=thread1)
+    t2 = threading.Thread(target=thread2)
+    t1.start()
+    t2.start()
+    t1.join(timeout=5)
+    t2.join(timeout=5)
+
+    # If adapter_scope were atomic, this would still be "adapter_one_lora".
+    # It isn't: thread2's full cycle ran to completion (and deactivated)
+    # while thread1's body was still executing.
+    assert observed_inside_a1_body == [None]
+
+
+def test_adapter_scope_noop_when_adapter_is_none():
+    mock_backend = MagicMock(spec=AdapterMixin)
+
+    entered = False
+    with AdapterMixin.adapter_scope(mock_backend, None):
+        entered = True
+
+    assert entered
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])
diff --git a/test/backends/test_adapters/test_core_types.py b/test/backends/test_adapters/test_core_types.py
index 604c022ce..d3751d891 100644
--- a/test/backends/test_adapters/test_core_types.py
+++ b/test/backends/test_adapters/test_core_types.py
@@ -119,9 +119,7 @@ def deactivate(self) -> None:
         PartialBinding()  # type: ignore[abstract]
 
 
-@pytest.mark.parametrize(
-    "cls", [LocalFileBinding, EmbeddedBinding, ServerMediatedBinding]
-)
+@pytest.mark.parametrize("cls", [EmbeddedBinding, ServerMediatedBinding])
 @pytest.mark.parametrize("verb", ["prepare", "activate", "deactivate", "release"])
 def test_stub_binding_subclasses_raise_not_implemented(cls, verb):
     binding = cls()
@@ -129,6 +127,12 @@ def test_stub_binding_subclasses_raise_not_implemented(cls, verb):
         getattr(binding, verb)()
 
 
+def test_local_file_binding_not_a_phase_0_stub():
+    # LocalFileBinding graduated out of the stub set in Epic #929 Phase 2
+    # (issue #1141) — see test_local_file_binding.py for its real behavior.
+    assert LocalFileBinding.prepare is not EmbeddedBinding.prepare
+
+
 def test_adapter_schema_mismatch_error_format():
     observed = frozenset({"key_a", "key_b"})
     expected = frozenset({"key_a", "key_c"})
diff --git a/test/backends/test_adapters/test_local_file_binding.py b/test/backends/test_adapters/test_local_file_binding.py
new file mode 100644
index 000000000..37b10ebf7
--- /dev/null
+++ b/test/backends/test_adapters/test_local_file_binding.py
@@ -0,0 +1,530 @@
+# Copyright IBM Corp. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for LocalFileBinding (Epic #929 Phase 2, issue #1141).
+
+Uses a fake AdapterMixin-conforming backend double throughout — no real HF
+model or network access.
+"""
+
+import threading
+from collections.abc import Coroutine
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from mellea.backends.adapters._core import LocalFileBinding
+from mellea.backends.adapters.catalog import AdapterType, fetch_intrinsic_metadata
+
+
+def _fake_backend():
+    """A minimal AdapterMixin-conforming double."""
+    backend = MagicMock()
+    backend.add_adapter.side_effect = lambda binding: setattr(
+        binding, "backend", backend
+    )
+    return backend
+
+
+def test_construction_defaults():
+    binding = LocalFileBinding()
+    assert binding.name == ""
+    assert binding.adapter_type is AdapterType.LORA
+    assert binding.repo_id == ""
+    # `None`, not "main": a default-constructed binding must not silently opt into
+    # tracking-latest. `None` defers to the catalogue's pinned revision.
+    assert binding.revision is None
+    assert binding.backend is None
+    assert binding.path is None
+
+
+def test_resolved_revision_falls_back_to_catalogue_pin():
+    pinned = fetch_intrinsic_metadata("answerability").revision
+    binding = LocalFileBinding(name="answerability")
+    assert binding.revision is None
+    assert binding.resolved_revision() == pinned
+    assert binding.resolved_revision() != "main"
+
+
+def test_resolved_revision_honours_explicit_main_override():
+    binding = LocalFileBinding(name="answerability", revision="main")
+    assert binding.resolved_revision() == "main"
+
+
+def test_resolved_revision_unknown_name_raises():
+    binding = LocalFileBinding(name="not-a-real-adapter-function")
+    with pytest.raises(ValueError, match="Unknown intrinsic name"):
+        binding.resolved_revision()
+
+
+def test_prepare_rejects_unconfigured_binding():
+    backend = _fake_backend()
+    binding = LocalFileBinding()
+    binding.bind_backend(backend)
+    with pytest.raises(RuntimeError, match="requires a non-empty name"):
+        binding.prepare()
+
+
+def test_qualified_name():
+    binding = LocalFileBinding(name="answerability", adapter_type=AdapterType.ALORA)
+    assert binding.qualified_name == "answerability_alora"
+
+
+def test_from_catalog_uses_pinned_metadata():
+    metadata = fetch_intrinsic_metadata("answerability")
+
+    binding = LocalFileBinding.from_catalog("answerability")
+
+    assert binding.name == "answerability"
+    assert binding.repo_id == metadata.repo_id
+    assert binding.revision == metadata.revision
+    assert binding.revision != "main"
+    assert binding.adapter_type == metadata.adapter_types[0]
+
+
+def test_from_catalog_unknown_name_raises():
+    with pytest.raises(ValueError, match="Unknown intrinsic name"):
+        LocalFileBinding.from_catalog("not-a-real-adapter-function")
+
+
+def test_prepare_without_bind_backend_raises():
+    binding = LocalFileBinding(name="answerability")
+    with pytest.raises(RuntimeError, match="bind_backend"):
+        binding.prepare()
+
+
+def test_prepare_registers_and_loads_on_staged_backend():
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+
+    binding.prepare()
+
+    backend.add_adapter.assert_called_once_with(binding)
+    backend.load_peft_adapter.assert_called_once_with(binding.qualified_name)
+    assert binding.backend is backend
+
+
+def test_prepare_is_idempotent():
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+
+    binding.prepare()
+    binding.prepare()
+
+    backend.add_adapter.assert_called_once()
+    backend.load_peft_adapter.assert_called_once()
+
+
+def test_prepare_and_release_are_linearized_before_registration():
+    backend = _fake_backend()
+    registration_started = threading.Event()
+    allow_registration = threading.Event()
+    release_finished = threading.Event()
+    errors: list[BaseException] = []
+
+    def register(binding: LocalFileBinding) -> None:
+        registration_started.set()
+        allow_registration.wait(timeout=1)
+        binding.backend = backend
+
+    def release(binding: LocalFileBinding) -> None:
+        try:
+            binding.release()
+        except BaseException as exc:
+            errors.append(exc)
+        finally:
+            release_finished.set()
+
+    backend.add_adapter.side_effect = register
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+
+    prepare_thread = threading.Thread(target=binding.prepare)
+    prepare_thread.start()
+    assert registration_started.wait(timeout=1)
+
+    release_thread = threading.Thread(target=release, args=(binding,))
+    release_thread.start()
+    try:
+        assert not release_finished.wait(timeout=0.1)
+    finally:
+        allow_registration.set()
+
+    prepare_thread.join(timeout=1)
+    release_thread.join(timeout=1)
+
+    assert not prepare_thread.is_alive()
+    assert not release_thread.is_alive()
+    assert not errors
+    backend.load_peft_adapter.assert_called_once_with(binding.qualified_name)
+    backend.unload_peft_adapter.assert_called_once_with(binding.qualified_name)
+    assert binding._released
+    assert binding.backend is None
+    assert not binding._loaded
+
+
+def test_bind_backend_rejects_a_different_backend_after_registration():
+    backend = _fake_backend()
+    other_backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+
+    with pytest.raises(RuntimeError, match="cannot change the backend"):
+        binding.bind_backend(other_backend)
+
+    assert binding.backend is backend
+    assert binding._staged_backend is backend
+
+
+def test_prepare_ignores_phase_hook_dispatch_failure():
+    """A prepare hook failure must not make successfully loaded weights unusable."""
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+
+    with (
+        patch("mellea.backends.adapters._core.has_plugins", return_value=True),
+        patch(
+            "mellea.plugins.hooks.adapter_function.AdapterFunctionPhaseCompletePayload",
+            side_effect=lambda **kwargs: SimpleNamespace(**kwargs),
+        ),
+        patch(
+            "mellea.backends.adapters._core.invoke_hook",
+            side_effect=RuntimeError("plugin dispatch blew up"),
+        ),
+    ):
+        binding.prepare()
+
+    assert binding._loaded
+    binding.activate()
+    backend.load_peft_adapter.assert_called_once_with(binding.qualified_name)
+    backend.activate_peft_adapter.assert_called_once_with(binding.qualified_name)
+
+
+def test_prepare_retries_only_the_load_after_a_load_failure():
+    """A failed load must be retryable without re-registering.
+
+    Regression guard: `add_adapter` sets `.backend` (registration) before
+    `prepare()` calls `load_peft_adapter` (the load). If the load raised,
+    `.backend` was already non-None, so the old idempotency guard
+    (`if self.backend is not None: return`) made every retry a silent no-op —
+    the caller got no error and no adapter, forever. The fix tracks the load
+    separately from registration so a retry redoes only the failed step.
+    """
+    backend = _fake_backend()
+    backend.load_peft_adapter.side_effect = [
+        RuntimeError("transient load failure"),
+        None,
+    ]
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+
+    with pytest.raises(RuntimeError, match="transient load failure"):
+        binding.prepare()
+
+    # Registration succeeded (that's why .backend is set); the load did not.
+    # A binding in this state must not look "already prepared".
+    assert binding.backend is backend
+    with pytest.raises(RuntimeError, match="prepare"):
+        binding.activate()
+
+    binding.prepare()  # retry: must not re-register, must retry the load
+
+    backend.add_adapter.assert_called_once()
+    assert backend.load_peft_adapter.call_count == 2
+    binding.activate()
+    backend.activate_peft_adapter.assert_called_once_with(binding.qualified_name)
+
+
+def test_bind_backend_after_release_raises():
+    """release() is terminal: bind_backend() must not silently revive the binding."""
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+    binding.release()
+
+    other_backend = _fake_backend()
+    with pytest.raises(RuntimeError, match="release"):
+        binding.bind_backend(other_backend)
+
+
+def test_prepare_after_release_raises():
+    """release() is terminal: prepare() must not silently revive the binding."""
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+    binding.release()
+
+    # Bypass bind_backend()'s own guard to confirm prepare() enforces this too.
+    binding._staged_backend = _fake_backend()
+    with pytest.raises(RuntimeError, match="release"):
+        binding.prepare()
+
+
+def test_activate_without_prepare_raises():
+    binding = LocalFileBinding(name="answerability")
+    with pytest.raises(RuntimeError, match="prepare"):
+        binding.activate()
+
+
+def test_deactivate_without_prepare_raises():
+    binding = LocalFileBinding(name="answerability")
+    with pytest.raises(RuntimeError, match="prepare"):
+        binding.deactivate()
+
+
+def test_activate_delegates_to_backend_verb():
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+
+    binding.activate()
+
+    backend.activate_peft_adapter.assert_called_once_with(binding.qualified_name)
+
+
+def test_deactivate_delegates_to_backend_verb():
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+
+    binding.deactivate()
+
+    backend.deactivate_peft_adapter.assert_called_once_with(binding.qualified_name)
+
+
+def test_activate_holds_the_backends_activation_lock():
+    """`activate()` must hold whatever lock `_adapter_activation_lock()` returns.
+
+    `activate_peft_adapter`/`deactivate_peft_adapter` document "must be called
+    while holding `_generation_lock`" as a precondition on the backend side;
+    `_adapter_activation_lock()` is the only thing satisfying that precondition
+    on this path (`adapter_scope` holds no lock of its own). A real
+    `threading.Lock` proves it's actually held during the call, not just
+    entered-and-exited around a no-op.
+    """
+    backend = _fake_backend()
+    lock = threading.Lock()
+    backend._adapter_activation_lock.return_value = lock
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+
+    observed_locked = {}
+    backend.activate_peft_adapter.side_effect = lambda _name: (
+        observed_locked.setdefault("during_call", lock.locked())
+    )
+
+    binding.activate()
+
+    assert observed_locked["during_call"] is True
+    assert not lock.locked()
+
+
+def test_deactivate_holds_the_backends_activation_lock():
+    backend = _fake_backend()
+    lock = threading.Lock()
+    backend._adapter_activation_lock.return_value = lock
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+
+    observed_locked = {}
+    backend.deactivate_peft_adapter.side_effect = lambda _name: (
+        observed_locked.setdefault("during_call", lock.locked())
+    )
+
+    binding.deactivate()
+
+    assert observed_locked["during_call"] is True
+    assert not lock.locked()
+
+
+def test_release_without_prepare_is_noop():
+    binding = LocalFileBinding(name="answerability")
+    binding.release()  # must not raise
+
+
+def test_release_after_bind_before_prepare_clears_staged_backend():
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+
+    binding.release()
+
+    assert binding._staged_backend is None
+    assert binding._released
+    backend.unload_peft_adapter.assert_not_called()
+
+
+def test_release_unloads_and_clears_state():
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+
+    binding.release()
+
+    backend.unload_peft_adapter.assert_called_once_with(binding.qualified_name)
+    assert binding.backend is None
+    assert binding.path is None
+    assert binding._staged_backend is None
+
+
+def test_release_requires_deactivation_after_activation():
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+    binding.activate()
+
+    with pytest.raises(RuntimeError, match="requires deactivate"):
+        binding.release()
+
+    backend.unload_peft_adapter.assert_not_called()
+    binding.deactivate()
+    binding.release()
+
+    backend.unload_peft_adapter.assert_called_once_with(binding.qualified_name)
+
+
+def test_release_cannot_race_activation_after_the_backend_selects_weights():
+    backend = _fake_backend()
+    lock = threading.Lock()
+    activate_exited_lock = threading.Event()
+    allow_activate_to_finish = threading.Event()
+    delayed_exits = [0]
+
+    class _ActivationLock:
+        def __enter__(self) -> None:
+            lock.acquire()
+
+        def __exit__(self, *_args: object) -> None:
+            lock.release()
+            if delayed_exits[0]:
+                delayed_exits[0] -= 1
+                activate_exited_lock.set()
+                allow_activate_to_finish.wait()
+
+    backend._adapter_activation_lock.return_value = _ActivationLock()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+    delayed_exits[0] = 1
+
+    activate_thread = threading.Thread(target=binding.activate)
+    activate_thread.start()
+    assert activate_exited_lock.wait(timeout=1)
+
+    release_error: list[BaseException] = []
+
+    def release() -> None:
+        try:
+            binding.release()
+        except BaseException as exc:
+            release_error.append(exc)
+
+    release_thread = threading.Thread(target=release)
+    release_thread.start()
+    allow_activate_to_finish.set()
+    activate_thread.join(timeout=1)
+    release_thread.join(timeout=1)
+
+    assert not activate_thread.is_alive()
+    assert not release_thread.is_alive()
+    assert len(release_error) == 1
+    assert isinstance(release_error[0], RuntimeError)
+    assert binding.backend is backend
+    assert binding._active
+    backend.unload_peft_adapter.assert_not_called()
+
+
+def test_release_retries_after_unload_failure():
+    backend = _fake_backend()
+    backend.unload_peft_adapter.side_effect = [
+        RuntimeError("transient unload failure"),
+        None,
+    ]
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+    binding.path = "/fake/adapter"
+
+    with pytest.raises(RuntimeError, match="transient unload failure"):
+        binding.release()
+
+    assert not binding._released
+    assert binding.backend is backend
+    assert binding.path == "/fake/adapter"
+    assert binding._staged_backend is backend
+    assert binding._loaded
+    binding.activate()
+    binding.deactivate()
+
+    binding.release()
+
+    assert backend.unload_peft_adapter.call_count == 2
+    assert binding._released
+    assert binding.backend is None
+    assert binding.path is None
+    assert binding._staged_backend is None
+    assert not binding._loaded
+
+
+def test_release_is_idempotent():
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+
+    binding.release()
+    binding.release()
+
+    backend.unload_peft_adapter.assert_called_once()
+
+
+def test_prepare_fires_phase_complete_metric_when_plugins_present():
+    pytest.importorskip("cpex", reason="cpex not installed — install mellea[hooks]")
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+
+    with (
+        patch("mellea.backends.adapters._core.has_plugins", return_value=True),
+        patch("mellea.backends.adapters._core._run_async_in_thread") as mock_run,
+    ):
+        binding.prepare()
+
+    mock_run.assert_called_once()
+    hook_coro = mock_run.call_args.args[0]
+    assert isinstance(hook_coro, Coroutine)
+    hook_coro.close()
+
+
+def test_release_does_not_fire_phase_complete_metric():
+    # "release" is not a valid AdapterFunctionPhaseCompletePayload.phase value.
+    pytest.importorskip("cpex", reason="cpex not installed — install mellea[hooks]")
+    backend = _fake_backend()
+    binding = LocalFileBinding(name="answerability")
+    binding.bind_backend(backend)
+    binding.prepare()
+
+    with (
+        patch("mellea.backends.adapters._core.has_plugins", return_value=True),
+        patch("mellea.backends.adapters._core._run_async_in_thread") as mock_run,
+    ):
+        binding.release()
+
+    mock_run.assert_not_called()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])
diff --git a/test/backends/test_adapters/test_local_file_e2e.py b/test/backends/test_adapters/test_local_file_e2e.py
new file mode 100644
index 000000000..a6602f9d1
--- /dev/null
+++ b/test/backends/test_adapters/test_local_file_e2e.py
@@ -0,0 +1,116 @@
+# Copyright IBM Corp. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Real e2e test: LocalFileBinding's lifecycle against a real PEFT adapter.
+
+Downloads the real "answerability" adapter from Hugging Face and loads it onto
+a real Granite base model via LocalHFBackend — no mocking of the HF download,
+PEFT machinery, or model. Requires GPU and network/Hub access; not expected to
+run in CI or in sandboxes without hardware access (see test/README.md).
+
+`adapter_scope()` is asserted to really flip the real PEFT model's active
+adapter set. `generate_from_context()` on a plain `CBlock` is only a
+smoke-test that generation still succeeds afterwards — it does not run
+through the activated adapter, since the standard generation path always
+deactivates adapters first (`_generate_with_adapter_lock("", ...)`); wiring
+that path onto `adapter_scope` is deferred to #1465.
+
+Assertions are structural/functional only (adapter registered, real model
+reports it active, generation succeeds, adapter cleanly released), per
+test/README.md's e2e rules — no assertions on generated text content.
+"""
+
+import os
+
+import pytest
+
+torch = pytest.importorskip("torch", reason="torch not installed — install mellea[hf]")
+
+from test.predicates import require_gpu
+
+pytestmark = [
+    pytest.mark.huggingface,
+    pytest.mark.e2e,
+    pytest.mark.slow,
+    require_gpu(min_vram_gb=20),
+    pytest.mark.skipif(
+        int(os.environ.get("CICD", 0)) == 1,
+        reason="Skipping HuggingFace e2e tests in CI",
+    ),
+]
+
+from mellea.backends import model_ids
+from mellea.backends.adapters._core import (
+    Adapter,
+    Identity,
+    IOContract,
+    LocalFileBinding,
+)
+from mellea.backends.huggingface import LocalHFBackend
+from mellea.core import CBlock, Component
+from mellea.stdlib.context import SimpleContext
+from test.conftest import cleanup_gpu_backend, hf_skip
+
+
+class _Contract(IOContract):
+    def build_prompt(self, **kwargs: object) -> Component:
+        raise NotImplementedError
+
+    def parse(self, raw: str) -> dict[str, object]:
+        return {}
+
+
+@pytest.fixture
+def backend():
+    with hf_skip():
+        backend = LocalHFBackend(model_id=model_ids.IBM_GRANITE_4_1_3B)
+    yield backend
+    cleanup_gpu_backend(backend, backend_name="local_file_e2e")
+
+
+@pytest.mark.asyncio
+async def test_local_file_binding_full_lifecycle_against_real_model(backend):
+    binding = LocalFileBinding.from_catalog("answerability")
+    # adapter_type must agree with the binding: `from_catalog` takes
+    # `metadata.adapter_types[0]`, which for `answerability` is LoRA. Hardcoding
+    # "alora" here made the identity contradict the weights actually loaded.
+    identity = Identity(
+        name="answerability",
+        adapter_type=binding.adapter_type.value,
+        capability="answerability",
+    )
+    adapter = Adapter(identity=identity, io_contract=_Contract(), weights=binding)
+
+    with hf_skip():
+        binding.bind_backend(backend)
+        binding.prepare()
+
+    assert binding.backend is backend
+    assert binding.qualified_name in backend.list_adapters()
+
+    ctx = SimpleContext().add(CBlock("Is the sky blue?"))
+    with backend.adapter_scope(adapter):
+        # Confirms activate() really flipped the real PEFT model's active
+        # adapter — the generate call below does not run through it (see
+        # module docstring), so this is the only in-scope proof of activation.
+        assert binding.qualified_name in backend._model.active_adapters()  # type: ignore[union-attr]
+
+        mot, _ = await backend.generate_from_context(
+            CBlock("Is the sky blue?"), ctx, model_options={}
+        )
+        value = await mot.avalue()
+
+    assert binding.qualified_name not in backend._model.active_adapters()  # type: ignore[union-attr]
+    assert isinstance(value, str)
+    assert len(value) > 0
+
+    binding.release()
+    assert binding.backend is None
+    # list_adapters() reports everything ever registered via add_adapter,
+    # regardless of load state — release() only reverses the load, so check
+    # the loaded-adapters bookkeeping directly instead.
+    assert binding.qualified_name not in backend._loaded_adapters
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])
diff --git a/test/backends/test_adapters/test_local_file_integration.py b/test/backends/test_adapters/test_local_file_integration.py
new file mode 100644
index 000000000..a369ffad7
--- /dev/null
+++ b/test/backends/test_adapters/test_local_file_integration.py
@@ -0,0 +1,165 @@
+# Copyright IBM Corp. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Integration test: LocalFileBinding's full lifecycle through a real LocalHFBackend.
+
+A real `LocalHFBackend` instance is used; only the Hugging Face download
+(`intrinsics.obtain_lora`) and the underlying PEFT model (`_model`) are mocked,
+per test/README.md's definition of `integration` — real framework/library objects
+wired together, external network and model weights mocked at the outer boundary.
+
+Telemetry assertions are on the fired **hooks**, not on spans: `adapter_scope`
+fires hooks and deliberately opens no spans (#1464 documents the rule, #1466 adds
+the spans from a plugin).
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+torch = pytest.importorskip("torch", reason="torch not installed — install mellea[hf]")
+pytest.importorskip(
+    "transformers", reason="transformers not installed — install mellea[hf]"
+)
+pytest.importorskip(
+    "llguidance", reason="llguidance not installed — install mellea[hf]"
+)
+
+from mellea.backends.adapters._core import (
+    Adapter,
+    Identity,
+    IOContract,
+    LocalFileBinding,
+)
+from mellea.backends.adapters.catalog import fetch_intrinsic_metadata
+from mellea.backends.huggingface import LocalHFBackend
+from mellea.core import Component
+from test.backends.test_adapters._hook_capture import (
+    capture_adapter_hooks,
+    hook_payloads,
+)
+
+pytestmark = pytest.mark.integration
+
+
+class _Contract(IOContract):
+    def build_prompt(self, **kwargs: object) -> Component:
+        raise NotImplementedError
+
+    def parse(self, raw: str) -> dict[str, object]:
+        return {}
+
+
+def _make_backend() -> LocalHFBackend:
+    mock_tok = MagicMock(eos_token_id=0, vocab_size=32000)
+    mock_tok._tokenizer = MagicMock()
+    mock_tok._tokenizer.get_vocab_size.return_value = 32000
+    mock_tok.__len__ = MagicMock(return_value=32000)
+    mock_model = MagicMock(vocab_size=32000)
+    with (
+        patch("mellea.backends.huggingface.llguidance") as mock_llg,
+        patch("mellea.backends.huggingface.set_seed"),
+    ):
+        mock_llg.hf.from_tokenizer.return_value = MagicMock(vocab_size=32000)
+        return LocalHFBackend(
+            model_id="ibm-granite/granite-3.3-8b-instruct",
+            custom_config=(mock_tok, mock_model, torch.device("cpu")),
+        )
+
+
+def _make_binding() -> LocalFileBinding:
+    metadata = fetch_intrinsic_metadata("answerability")
+    return LocalFileBinding(
+        name="answerability",
+        adapter_type=metadata.adapter_types[0],
+        repo_id=metadata.repo_id,
+        revision=metadata.revision,
+    )
+
+
+def _make_adapter(binding: LocalFileBinding) -> Adapter:
+    # adapter_type must agree with the binding: `from_catalog` takes
+    # `metadata.adapter_types[0]`, which for `answerability` is LoRA. Hardcoding
+    # "alora" here made the identity contradict the weights actually loaded.
+    identity = Identity(
+        name="answerability",
+        adapter_type=binding.adapter_type.value,
+        capability="answerability",
+    )
+    return Adapter(identity=identity, io_contract=_Contract(), weights=binding)
+
+
+def test_prepare_activate_deactivate_release_full_lifecycle():
+    backend = _make_backend()
+    binding = _make_binding()
+    adapter = _make_adapter(binding)
+
+    with patch(
+        "mellea.formatters.granite.intrinsics.obtain_lora",
+        return_value="/fake/local/adapter/path",
+    ) as mock_obtain_lora:
+        binding.bind_backend(backend)
+        binding.prepare()
+
+        assert binding.backend is backend
+        assert binding.qualified_name in backend.list_adapters()
+        mock_obtain_lora.assert_called_once()
+        assert mock_obtain_lora.call_args.kwargs["revision"] == binding.revision
+
+        with capture_adapter_hooks() as mock_invoke:
+            with backend.adapter_scope(adapter):
+                backend._model.set_adapter.assert_called_with(binding.qualified_name)  # type: ignore[union-attr]
+
+            backend._model.set_adapter.assert_called_with([])  # type: ignore[union-attr]
+
+        binding.release()
+
+    backend._model.delete_adapter.assert_called_once_with(binding.qualified_name)  # type: ignore[union-attr]
+    assert binding.backend is None
+
+    recorded = hook_payloads(mock_invoke)
+    phases = [p.phase for p in recorded if hasattr(p, "phase")]
+    assert phases == ["activate", "deactivate"]
+
+    invocations = [p for p in recorded if hasattr(p, "outcome")]
+    assert len(invocations) == 1
+    assert invocations[0].outcome == "success"
+    assert invocations[0].name == "answerability"
+    assert invocations[0].binding_type == "local_file"
+    assert invocations[0].adapter_type == binding.adapter_type.value
+
+
+def test_deactivate_runs_even_when_generation_body_raises():
+    backend = _make_backend()
+    binding = _make_binding()
+    adapter = _make_adapter(binding)
+
+    with patch(
+        "mellea.formatters.granite.intrinsics.obtain_lora",
+        return_value="/fake/local/adapter/path",
+    ):
+        binding.bind_backend(backend)
+        binding.prepare()
+
+        with capture_adapter_hooks() as mock_invoke:
+            with pytest.raises(RuntimeError, match="generation failed"):
+                with backend.adapter_scope(adapter):
+                    raise RuntimeError("generation failed")
+
+        backend._model.set_adapter.assert_called_with([])  # type: ignore[union-attr]
+        binding.release()
+
+    # deactivate still ran, and the invocation is reported as an error carrying
+    # the original exception — the behaviour the span status used to assert.
+    recorded = hook_payloads(mock_invoke)
+    phases = [p.phase for p in recorded if hasattr(p, "phase")]
+    assert "deactivate" in phases
+
+    invocations = [p for p in recorded if hasattr(p, "outcome")]
+    assert len(invocations) == 1
+    assert invocations[0].outcome == "error"
+    assert isinstance(invocations[0].error, RuntimeError)
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])
diff --git a/test/backends/test_adapters/test_shims.py b/test/backends/test_adapters/test_shims.py
index 8b4a39b08..98801a75d 100644
--- a/test/backends/test_adapters/test_shims.py
+++ b/test/backends/test_adapters/test_shims.py
@@ -16,7 +16,7 @@
 import pytest
 
 from mellea.backends.adapters import Adapter, EmbeddedIntrinsicAdapter, IntrinsicAdapter
-from mellea.backends.adapters._core import Identity
+from mellea.backends.adapters._core import Identity, LocalFileBinding
 from mellea.backends.adapters.adapter import AdapterMixin
 from mellea.backends.adapters.catalog import AdapterType, IntrinsicsCatalogEntry
 
@@ -229,6 +229,29 @@ def test_adapter_scope_is_noop():
         pass  # must not raise
 
 
+def test_adapter_scope_raises_for_a_shim_backed_adapter():
+    """adapter_scope now activates real weights, so a shim-backed adapter raises.
+
+    Deliberate behaviour change from Phase 1 (issue #1140), where `adapter_scope`
+    was `yield` unconditionally regardless of `adapter.weights`. `resolve_adapter()`
+    still returns `IntrinsicAdapter`/`LocalHFAdapter` shims carrying
+    `_ShimWeightsBinding`, whose `.activate()` raises `NotImplementedError` — so
+    `with backend.adapter_scope(backend.resolve_adapter(name)):` goes from a
+    no-op to a hard failure for every adapter the public API currently hands
+    out. Nothing in the codebase calls `adapter_scope` with a resolved adapter
+    yet (#1465 is the tracked cutover), but this pins the change as
+    deliberate rather than incidental — if #1465 needs `adapter_scope` to
+    tolerate shim/unprepared bindings instead, that decision should update
+    this test, not silently contradict it.
+    """
+    mock_backend = MagicMock(spec=AdapterMixin)
+    adapter = _make_intrinsic_adapter("answerability")
+
+    with pytest.raises(NotImplementedError, match="Phase 2"):
+        with AdapterMixin.adapter_scope(mock_backend, adapter):
+            pytest.fail("body must not run when the shim's activate() raises")
+
+
 def test_resolve_adapter_returns_existing_by_capability():
     """resolve_adapter must return an already-registered adapter without creating a new one."""
     existing = _make_intrinsic_adapter("answerability")
@@ -263,6 +286,45 @@ def test_find_adapter_honours_type_preference_order():
     assert result is alora, "alora must win over lora regardless of insertion order"
 
 
+def test_resolve_adapter_names_the_conflict_when_a_binding_blocks_registration():
+    """resolve_adapter's KeyError should name the collision, not just say "not found".
+
+    Regression guard: a `LocalFileBinding` registered under the same
+    qualified-name key space `resolve_adapter` auto-registers into silently
+    blocks the new `IntrinsicAdapter` (the backend's duplicate-key guard
+    refuses it). `_find_adapter` can't see the `LocalFileBinding` either (not
+    an `_AdapterCore`), so without this check the failure surfaced as a bare
+    "Adapter 'answerability' not found after registration" with no hint of
+    what actually occupied the name.
+    """
+    binding = LocalFileBinding(name="answerability", adapter_type=AdapterType.LORA)
+    mock_backend = MagicMock(spec=AdapterMixin)
+    mock_backend.base_model_name = "ibm-granite/granite-4.1-3b"
+    mock_backend._uses_embedded_adapters = False
+    mock_backend._added_adapters = {binding.qualified_name: binding}
+    mock_backend._find_adapter.side_effect = lambda cap, types=None: (
+        AdapterMixin._find_adapter(mock_backend, cap, types)
+    )
+    # Simulates the backend's real duplicate-key guard refusing the new
+    # IntrinsicAdapter: registration is attempted but nothing new lands in
+    # `_added_adapters`.
+    mock_backend.add_adapter.side_effect = lambda a: None
+
+    with (
+        patch(
+            "mellea.backends.adapters.adapter.fetch_intrinsic_metadata",
+            return_value=_MOCK_CATALOG_ENTRY,
+        ),
+        patch(
+            "mellea.backends.adapters.adapter.intrinsics.obtain_io_yaml",
+            return_value="/fake/adapter.yaml",
+        ),
+        patch("builtins.open", mock_open(read_data="key: value")),
+    ):
+        with pytest.raises(KeyError, match=r"LocalFileBinding.*answerability_lora"):
+            AdapterMixin.resolve_adapter(mock_backend, "answerability")
+
+
 def test_resolve_adapter_raises_without_base_model():
     """resolve_adapter must raise ValueError when the backend has no model ID."""
     mock_backend = MagicMock(spec=AdapterMixin)
diff --git a/test/backends/test_huggingface_unit.py b/test/backends/test_huggingface_unit.py
index e96375061..beaf52187 100644
--- a/test/backends/test_huggingface_unit.py
+++ b/test/backends/test_huggingface_unit.py
@@ -239,9 +239,93 @@ def test_generate_with_adapter_lock_calls_load_peft_adapter():
         backend._generate_with_adapter_lock("my_adapter", lambda: "output")
 
     mock_load.assert_called_once_with("my_adapter")
+    # Deliberately no `_model.set_adapter` assertion. Since #1141 that call is
+    # reached via `activate_peft_adapter` rather than inlined here, so asserting
+    # it would make this test an unannounced guard for the delegation chain --
+    # failing on a change to `activate_peft_adapter` without naming it. The chain
+    # is covered by `test_generate_with_adapter_lock_uses_activate_deactivate_verbs`
+    # and the verb itself by `test_activate_peft_adapter_calls_set_adapter`.
+
+
+def test_generate_with_adapter_lock_uses_activate_deactivate_verbs():
+    """_generate_with_adapter_lock delegates to the new activate/deactivate verbs
+    rather than calling `_model.set_adapter` directly (Epic #929 Phase 2, issue #1141).
+    """
+    backend = _make_backend()
+    backend._model.active_adapters.return_value = ["my_adapter"]  # type: ignore[union-attr]
+
+    with (
+        patch.object(backend, "load_peft_adapter"),
+        patch.object(backend, "activate_peft_adapter") as mock_activate,
+        patch.object(backend, "deactivate_peft_adapter") as mock_deactivate,
+    ):
+        backend._generate_with_adapter_lock("my_adapter", lambda: "output")
+
+    mock_activate.assert_called_once_with("my_adapter")
+    mock_deactivate.assert_not_called()
+
+    backend._model.active_adapters.return_value = []  # type: ignore[union-attr]
+    with (
+        patch.object(backend, "activate_peft_adapter") as mock_activate,
+        patch.object(backend, "deactivate_peft_adapter") as mock_deactivate,
+    ):
+        backend._generate_with_adapter_lock("", lambda: "output")
+
+    mock_activate.assert_not_called()
+    mock_deactivate.assert_called_once_with("")
+
+
+def test_activate_peft_adapter_calls_set_adapter():
+    """activate_peft_adapter() is a thin wrapper over `_model.set_adapter`."""
+    backend = _make_backend()
+
+    backend.activate_peft_adapter("my_adapter")
+
     backend._model.set_adapter.assert_called_once_with("my_adapter")  # type: ignore[union-attr]
 
 
+def test_deactivate_peft_adapter_calls_set_adapter_empty():
+    """deactivate_peft_adapter() clears active adapters via `_model.set_adapter([])`."""
+    backend = _make_backend()
+
+    backend.deactivate_peft_adapter("my_adapter")
+
+    backend._model.set_adapter.assert_called_once_with([])  # type: ignore[union-attr]
+
+
+def test_deactivate_peft_adapter_swallows_no_adapter_loaded_error():
+    """deactivate_peft_adapter() is a no-op if the model has no adapter loaded yet."""
+    backend = _make_backend()
+    backend._model.set_adapter.side_effect = ValueError(  # type: ignore[union-attr]
+        "No adapter loaded. Please load an adapter first."
+    )
+
+    backend.deactivate_peft_adapter("my_adapter")  # must not raise
+
+
+def test_deactivate_peft_adapter_reraises_other_value_errors():
+    """deactivate_peft_adapter() only swallows the specific 'no adapter loaded' error."""
+    backend = _make_backend()
+    backend._model.set_adapter.side_effect = ValueError("some other failure")  # type: ignore[union-attr]
+
+    with pytest.raises(ValueError, match="some other failure"):
+        backend.deactivate_peft_adapter("my_adapter")
+
+
+def test_adapter_activation_lock_is_the_generation_lock():
+    """`_adapter_activation_lock()` reuses `_generation_lock`, not a separate lock.
+
+    `LocalFileBinding.activate()`/`.deactivate()` (driven by `adapter_scope()`)
+    hold no lock of their own and rely on this method for the exclusivity
+    `_generate_with_adapter_lock` otherwise gets from holding `_generation_lock`
+    directly. If this ever returned a different lock, the two callers would no
+    longer be mutually exclusive.
+    """
+    backend = _make_backend()
+
+    assert backend._adapter_activation_lock() is backend._generation_lock
+
+
 def test_list_adapters_reflects_registration_not_just_loading():
     """list_adapters() must include adapters registered via add_adapter, even
     if they've never been loaded (aligns HF's semantics with OpenAI's).
diff --git a/test/helpers/test_event_loop_helper.py b/test/helpers/test_event_loop_helper.py
index dd8c23ee4..10219ebc7 100644
--- a/test/helpers/test_event_loop_helper.py
+++ b/test/helpers/test_event_loop_helper.py
@@ -2,7 +2,10 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import contextvars
+import gc
 import multiprocessing
+import warnings
+from unittest import mock
 
 import pytest
 
@@ -82,6 +85,75 @@ async def testing() -> int:
     assert elh.__event_loop_handler is not None
 
 
+def test_run_async_in_thread_closes_coroutine_on_scheduling_failure():
+    """A failure before the coroutine is scheduled must close it, not leak it."""
+    closed = False
+
+    async def never_scheduled() -> None:
+        nonlocal closed
+        closed = True
+
+    co = never_scheduled()
+
+    with pytest.raises(RuntimeError, match="boom"):
+        with mock.patch.object(
+            elh, "get_current_event_loop", side_effect=RuntimeError("boom")
+        ):
+            elh._run_async_in_thread(co)
+
+    # A closed-but-never-started coroutine raises StopIteration internally,
+    # which close() swallows; the body (setting `closed`) never runs.
+    assert closed is False
+    with pytest.raises(RuntimeError):
+        co.send(None)
+
+
+def test_run_async_in_thread_closes_wrapped_coroutine_on_scheduling_failure():
+    """A scheduling failure must also close the internal `_wrapped()` coroutine.
+
+    Regression guard: closing `co` alone isn't enough — `_wrapped()`'s own
+    coroutine object is created before `run_coroutine_threadsafe` is called, so
+    if scheduling itself raises, `_wrapped()` never starts and never gets to
+    `await co`, leaking a second "coroutine was never awaited" warning distinct
+    from `co`'s.
+    """
+
+    async def caller_coroutine() -> None:
+        return None
+
+    co = caller_coroutine()
+
+    with mock.patch.object(
+        elh.asyncio,
+        "run_coroutine_threadsafe",
+        side_effect=RuntimeError("scheduling failed"),
+    ):
+        with pytest.raises(RuntimeError, match="scheduling failed"):
+            elh._run_async_in_thread(co)
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        del co
+        gc.collect()
+
+    unawaited = [
+        w
+        for w in caught
+        if issubclass(w.category, RuntimeWarning) and "never awaited" in str(w.message)
+    ]
+    assert unawaited == [], f"leaked unawaited coroutine(s): {unawaited}"
+
+
+def test_run_async_in_thread_reraises_coroutine_exception():
+    """Exceptions raised by the coroutine itself still propagate normally."""
+
+    async def boom() -> None:
+        raise ValueError("from inside the coroutine")
+
+    with pytest.raises(ValueError, match="from inside the coroutine"):
+        elh._run_async_in_thread(boom())
+
+
 def test_event_loop_handler_with_forking():
     """Importing mellea before fork must not crash the child process."""