diff --git a/application/__init__.py b/application/__init__.py
new file mode 100644
index 0000000..24d12d6
--- /dev/null
+++ b/application/__init__.py
@@ -0,0 +1,12 @@
+"""Application layer - pure Python use-case orchestration.
+
+Sits between ``presentation/`` (Qt widgets) and ``domain/`` (entities). A module
+here answers "what has to happen, in what order" for one use case - route a
+turn, run a conversation - without knowing whether a human, a scheduler or a
+test triggered it.
+
+Hard rule (ADR-001 I1/I3, enforced by ``scripts/check_imports.py``): no
+PySide6/PyQt imports and no reach into ``presentation/``/``ui/``. Results travel
+back up through plain-Python callbacks; turning those into Qt signals is the
+presentation layer's job.
+"""
diff --git a/application/conversations/__init__.py b/application/conversations/__init__.py
new file mode 100644
index 0000000..d4a4b1c
--- /dev/null
+++ b/application/conversations/__init__.py
@@ -0,0 +1,5 @@
+"""Conversation use case: the lifecycle of one agent turn (EPIC R04)."""
+
+from .conversation_application_service import ConversationApplicationService
+
+__all__ = ["ConversationApplicationService"]
diff --git a/application/model_routing/__init__.py b/application/model_routing/__init__.py
new file mode 100644
index 0000000..5a96359
--- /dev/null
+++ b/application/model_routing/__init__.py
@@ -0,0 +1,12 @@
+"""Model routing use case: pick the best-fit model for one turn (EPIC R03)."""
+
+from .routing_application_service import (
+    RoutingApplicationService,
+    RoutingDecision,
+    RoutingMode,
+    is_valid_mode,
+    normalize_mode,
+)
+
+__all__ = ["RoutingApplicationService", "RoutingDecision", "RoutingMode",
+           "normalize_mode", "is_valid_mode"]
diff --git a/application/model_routing/routing_application_service.py b/application/model_routing/routing_application_service.py
new file mode 100644
index 0000000..40eed00
--- /dev/null
+++ b/application/model_routing/routing_application_service.py
@@ -0,0 +1,353 @@
+"""RoutingApplicationService - one routing flow for every surface (R03-T03).
+
+Before this service, the same routing algorithm existed three times:
+
+* ``ui/chat_panel.py::_apply_routing``       (Cowork chat)
+* ``ui/co4e_tab.py::_apply_co4e_routing``    (Co4E studio)
+* ``ui/folder_tab.py::_ai_apply_routing``    (AI-Edit)
+
+The three copies had already drifted - each one resolves the "current model"
+differently and each one has its own private notion of what to do when the user
+declines - and every one of them lives inside a Qt widget, so none of the logic
+could be tested without building a window.
+
+This module is the single implementation. It is pure Python: no Qt import, no
+config access, no network. The presentation layer supplies a confirm callback
+and renders the notice; everything else happens here.
+
+Modes (:class:`RoutingMode`)
+----------------------------
+* ``OFF``      - never switch. The user's pinned model always wins.
+* ``AUTO``     - switch silently when the best candidate clears the gain threshold.
+* ``MANUAL``   - propose the switch and switch only if the confirm callback approves.
+* ``FALLBACK`` - never switch pre-emptively; switch only AFTER the current model
+  fails, to the next-best candidate. This is the mode a user wants when they
+  trust their own model choice but still want the turn to survive an outage.
+
+Migration note (ADR-001 section 4): the scoring/ranking engine is NOT rewritten.
+This service depends on the small :class:`RoutingPort` interface, and production
+wires the existing, already-tested ``core.routing.service.RoutingService`` into
+it. Tests wire a fake.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Callable, List, Optional, Protocol, Sequence, Tuple
+
+
+class RoutingMode(str, Enum):
+    """Per-surface routing behaviour.
+
+    The first three values match ``core.routing.models.SwitchMode`` string for
+    string, so a mode read from the existing config round-trips unchanged.
+    """
+
+    OFF = "off"
+    AUTO = "auto"
+    MANUAL = "manual"
+    FALLBACK = "fallback"
+
+    @classmethod
+    def parse(cls, raw: Any) -> "RoutingMode":
+        """Best-effort parse of a config value.
+
+        Unknown or empty values become ``OFF``: routing is an optimisation, and
+        the safe reading of a corrupt setting is "leave the user's model alone"
+        rather than "silently move their work to another model".
+        """
+        try:
+            return cls(str(raw or "off").strip().lower())
+        except ValueError:
+            return cls.OFF
+
+
+@dataclass(frozen=True)
+class RoutingDecision:
+    """The outcome of routing one turn - an immutable instruction for the caller.
+
+    ``provider``/``model`` are ALWAYS filled with what the turn should actually
+    run on, switched or not, so a call site never has to re-derive the fallback
+    itself (the bug that made the three UI copies diverge).
+    """
+
+    mode: RoutingMode
+    provider: str
+    model: str
+    switched: bool = False
+    task_type: str = ""
+    score_gain: float = 0.0
+    reason: str = ""
+    declined: bool = False  # Manual mode: a switch was offered and refused
+    # What the turn would have run on without routing. Carried so the Manual
+    # confirm dialog can show "from X to Y" without re-deriving the current
+    # model itself - re-deriving it differently per screen is exactly how the
+    # three legacy copies drifted apart.
+    previous_provider: str = ""
+    previous_model: str = ""
+
+    @property
+    def should_notify(self) -> bool:
+        """True when the UI should show the "switched model" notice - i.e. only
+        when a switch really happened."""
+        return self.switched
+
+    def target(self) -> Tuple[str, str]:
+        """``(provider, model)`` to run this turn on."""
+        return self.provider, self.model
+
+    @property
+    def from_model(self) -> str:
+        """Candidate key (``provider/model``) of the model being switched away
+        from, or "" when nothing was selected yet.
+
+        Named to match ``core.routing.models.SwitchDecision`` so the existing
+        Manual-mode dialog (``ui/routing_toggle.py::confirm_switch``) accepts
+        this object unchanged - the dialog moves to the new shape in EPIC R08.
+        """
+        if not self.previous_model:
+            return ""
+        return f"{self.previous_provider}/{self.previous_model}"
+
+    @property
+    def to_model(self) -> str:
+        """Candidate key (``provider/model``) of the model to run on. See
+        :attr:`from_model` for why the name matches the legacy decision."""
+        return f"{self.provider}/{self.model}" if self.model else ""
+
+
+def is_valid_mode(raw: Any) -> bool:
+    """True when ``raw`` names a mode the routing service understands.
+
+    Distinct from :func:`normalize_mode` because callers need to tell "the user
+    chose off" apart from "this stored value is unrecognised" - the per-workspace
+    lookup falls back to the global setting only in the second case.
+    """
+    try:
+        RoutingMode(str(raw or "").strip().lower())
+    except ValueError:
+        return False
+    return True
+
+
+def normalize_mode(raw: Any) -> str:
+    """Canonical mode string for persistence, or ``"off"`` when unrecognised.
+
+    Exists so the mode vocabulary is defined exactly once. It used to be
+    hard-coded as a ``("off", "auto", "manual")`` tuple in four separate places
+    (config.py twice, state.py twice); adding FALLBACK meant finding all four,
+    and missing one silently downgraded the user's choice back to "off".
+    """
+    return RoutingMode.parse(raw).value
+
+
+class RoutingPort(Protocol):
+    """The slice of the routing engine this service needs.
+
+    Declared as a Protocol so the application layer states its requirement
+    without importing the implementation - which is what lets the whole service
+    be tested against a 20-line fake, and lets ``core.routing`` be replaced later
+    without touching this file.
+    """
+
+    def route(self, surface: str, prompt: str, current_provider: str, current_model: str,
+              *, mode_override: Optional[str] = None,
+              required_capabilities: Optional[List[str]] = None,
+              task_type: Optional[Any] = None) -> Any:
+        """Return a route result exposing ``should_switch``, ``target()``,
+        ``task_type`` and ``decision``."""
+
+
+# Presentation supplies this to ask the human. Receives the proposal so the
+# dialog can explain it; returns True to approve. Manual mode only.
+ConfirmFn = Callable[[RoutingDecision], bool]
+
+
+class RoutingApplicationService:
+    """Decides which provider/model one turn runs on.
+
+    Args:
+        router: the scoring engine (see :class:`RoutingPort`).
+        mode_reader: ``surface -> mode string``; production passes the per-workspace
+            lookup ``AppContext.project_routing_mode``. Injected rather than read
+            from config here so this layer stays free of config plumbing that
+            EPIC R02 is rewriting in parallel.
+    """
+
+    def __init__(self, router: RoutingPort,
+                 mode_reader: Optional[Callable[[str], str]] = None) -> None:
+        self._router = router
+        self._mode_reader = mode_reader
+
+    # -- main entry point -------------------------------------------------- #
+    def route_turn(
+        self,
+        surface: str,
+        prompt: str,
+        current_provider: str,
+        current_model: str,
+        *,
+        mode: Optional[str] = None,
+        confirm: Optional[ConfirmFn] = None,
+        required_capabilities: Optional[Sequence[str]] = None,
+        task_type: Optional[Any] = None,
+    ) -> RoutingDecision:
+        """Decide what to run this turn on. Never raises.
+
+        A routing failure must never block a message: any unexpected error
+        degrades to "keep the current model", which is exactly what all three
+        legacy copies did with a bare ``except`` - made explicit and testable here.
+        """
+        resolved_mode = RoutingMode.parse(mode if mode is not None else self._read_mode(surface))
+        keep = self._keep(resolved_mode, current_provider, current_model,
+                          reason="routing off - keeping current model")
+
+        # An empty prompt carries no signal to classify, so routing cannot make a
+        # meaningful choice; the same guard exists in all three legacy copies.
+        if resolved_mode is RoutingMode.OFF or not (prompt or "").strip():
+            return keep
+
+        # FALLBACK never switches up front - it only reacts to a failure, which
+        # the caller reports through fallback_after_failure().
+        if resolved_mode is RoutingMode.FALLBACK:
+            return self._keep(resolved_mode, current_provider, current_model,
+                              reason="fallback mode - switching only after a failure")
+
+        try:
+            result = self._router.route(
+                surface, prompt, current_provider, current_model,
+                mode_override=resolved_mode.value,
+                required_capabilities=list(required_capabilities) if required_capabilities else None,
+                task_type=task_type,
+            )
+        except Exception:  # noqa: BLE001 - routing must never break a turn
+            return self._keep(resolved_mode, current_provider, current_model,
+                              reason="routing engine failed - keeping current model")
+
+        proposal = self._to_decision(result, resolved_mode, current_provider, current_model)
+        if not proposal.switched:
+            return proposal
+
+        # Manual mode: the proposal only becomes a switch once a human approves.
+        if resolved_mode is RoutingMode.MANUAL:
+            if confirm is None or not self._ask(confirm, proposal):
+                return self._keep(resolved_mode, current_provider, current_model,
+                                  reason="switch declined - keeping current model",
+                                  task_type=proposal.task_type, declined=True)
+        return proposal
+
+    # -- failure recovery -------------------------------------------------- #
+    def fallback_after_failure(
+        self,
+        surface: str,
+        prompt: str,
+        failed_provider: str,
+        failed_model: str,
+        *,
+        mode: Optional[str] = None,
+        required_capabilities: Optional[Sequence[str]] = None,
+        task_type: Optional[Any] = None,
+    ) -> Optional[RoutingDecision]:
+        """Pick a replacement after ``failed_provider/failed_model`` failed.
+
+        Returns None when there is nothing to fall back to, so the caller can
+        surface the original error instead of retrying forever. Available in
+        AUTO and FALLBACK; OFF and MANUAL keep the user's model on failure too,
+        because silently moving work to another model is exactly what those two
+        modes exist to prevent.
+        """
+        resolved_mode = RoutingMode.parse(mode if mode is not None else self._read_mode(surface))
+        if resolved_mode not in (RoutingMode.AUTO, RoutingMode.FALLBACK):
+            return None
+
+        try:
+            # Asked in AUTO so the engine ranks candidates rather than short-
+            # circuiting on FALLBACK's "never switch up front" rule; the failed
+            # model is passed as current so any positive gain beats it.
+            result = self._router.route(
+                surface, prompt, failed_provider, failed_model,
+                mode_override=RoutingMode.AUTO.value,
+                required_capabilities=list(required_capabilities) if required_capabilities else None,
+                task_type=task_type,
+            )
+        except Exception:  # noqa: BLE001 - a broken router must not mask the real error
+            return None
+
+        decision = self._to_decision(result, resolved_mode, failed_provider, failed_model)
+        # A "switch" back to the model that just failed would retry the outage.
+        if not decision.switched or (decision.provider, decision.model) == (failed_provider, failed_model):
+            return None
+        return RoutingDecision(
+            mode=resolved_mode, provider=decision.provider, model=decision.model,
+            switched=True, task_type=decision.task_type, score_gain=decision.score_gain,
+            reason=f"{failed_provider}/{failed_model} failed - falling back to "
+                   f"{decision.provider}/{decision.model}",
+            previous_provider=failed_provider, previous_model=failed_model,
+        )
+
+    # -- internals --------------------------------------------------------- #
+    def _read_mode(self, surface: str) -> str:
+        """Per-surface mode from the injected reader ('off' when none supplied)."""
+        if self._mode_reader is None:
+            return RoutingMode.OFF.value
+        try:
+            return self._mode_reader(surface) or RoutingMode.OFF.value
+        except Exception:  # noqa: BLE001 - a config read must not break a turn
+            return RoutingMode.OFF.value
+
+    @staticmethod
+    def _keep(mode: RoutingMode, provider: str, model: str, *, reason: str,
+              task_type: str = "", declined: bool = False) -> RoutingDecision:
+        """A no-switch decision that still names the model to run on."""
+        return RoutingDecision(mode=mode, provider=provider, model=model, switched=False,
+                               task_type=task_type, reason=reason, declined=declined,
+                               previous_provider=provider, previous_model=model)
+
+    @staticmethod
+    def _ask(confirm: ConfirmFn, proposal: RoutingDecision) -> bool:
+        """Run the confirm callback, treating any failure as "declined".
+
+        The callback opens a modal dialog in production; if that raises (window
+        already closing, for instance) the safe answer is to keep the user's own
+        model rather than to switch without consent.
+        """
+        try:
+            return bool(confirm(proposal))
+        except Exception:  # noqa: BLE001
+            return False
+
+    @staticmethod
+    def _to_decision(result: Any, mode: RoutingMode,
+                     current_provider: str, current_model: str) -> RoutingDecision:
+        """Translate the engine's route result into a :class:`RoutingDecision`.
+
+        Defensive about the result shape on purpose: this is the seam between the
+        new layer and a legacy module still under refactor, and a missing
+        attribute must degrade to "keep current model" instead of raising into
+        the middle of a chat turn.
+        """
+        inner = getattr(result, "decision", None)
+        task_type = getattr(getattr(result, "task_type", None), "value", "") or ""
+        gain = float(getattr(inner, "score_gain", 0.0) or 0.0)
+        reason = str(getattr(inner, "reason", "") or "")
+
+        target = None
+        if getattr(result, "should_switch", False):
+            getter = getattr(result, "target", None)
+            target = getter() if callable(getter) else None
+
+        if not target:
+            return RoutingDecision(mode=mode, provider=current_provider, model=current_model,
+                                   switched=False, task_type=task_type, score_gain=gain,
+                                   reason=reason or "no better model - keeping current",
+                                   previous_provider=current_provider,
+                                   previous_model=current_model)
+
+        provider, model = target
+        return RoutingDecision(mode=mode, provider=provider or current_provider, model=model,
+                               switched=True, task_type=task_type, score_gain=gain, reason=reason,
+                               previous_provider=current_provider, previous_model=current_model)
+
+
+__all__ = ["RoutingApplicationService", "RoutingDecision", "RoutingMode",
+           "RoutingPort", "normalize_mode", "is_valid_mode"]
diff --git a/config.py b/config.py
index 7b29aac..c506e07 100644
--- a/config.py
+++ b/config.py
@@ -552,19 +552,25 @@ class AppConfig:
         return d
 
     def routing_mode_for(self, surface: str) -> str:
-        """Effective Off/Auto/Manual mode for a chat surface.
+        """Effective Off/Auto/Manual/Fallback mode for a chat surface.
+
+        A per-surface override wins; an empty override falls back to the global
+        ``switch_mode``. The value is validated through
+        ``application.model_routing.normalize_mode`` so the accepted vocabulary
+        is defined in exactly one place (R03-T03) - it used to be a literal
+        tuple repeated here and in state.py, and adding a mode to one copy but
+        not the others silently downgraded the user's choice to "off"."""
+        from .application.model_routing import normalize_mode
 
-        A per-surface override ("auto"/"manual"/"off") wins; an empty override
-        falls back to the global ``switch_mode``."""
         routing = self.routing
         override = (routing.get("surface_modes", {}) or {}).get(surface, "")
-        mode = override or routing.get("switch_mode", "off")
-        return mode if mode in ("off", "auto", "manual") else "off"
+        return normalize_mode(override or routing.get("switch_mode", "off"))
 
     def set_routing_mode_for(self, surface: str, mode: str) -> None:
-        """Persist a chat surface's Off/Auto/Manual toggle selection."""
-        mode = mode if mode in ("off", "auto", "manual") else "off"
-        self.routing.setdefault("surface_modes", {})[surface] = mode
+        """Persist a chat surface's routing toggle selection."""
+        from .application.model_routing import normalize_mode
+
+        self.routing.setdefault("surface_modes", {})[surface] = normalize_mode(mode)
         self.save()
 
     @property
diff --git a/docs/refactor/Refactoring_Checklist.md b/docs/refactor/Refactoring_Checklist.md
index 789cdd3..41ceff8 100644
--- a/docs/refactor/Refactoring_Checklist.md
+++ b/docs/refactor/Refactoring_Checklist.md
@@ -62,18 +62,18 @@
 * **Team chịu trách nhiệm**: 🔵 **Team Duy** (Chủ trì)
 * **Mục tiêu**: Hợp nhất logic routing bị phân tán thành `RoutingApplicationService` độc lập Qt; chuẩn hóa catalog nhà cung cấp.
 
-- [ ] **R03-T01 (Team Duy)**: Xây dựng bộ Contract Tests chuẩn hóa cho các Provider từ `providers/base.py` ➔ `tests/contracts/test_providers.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R03-T02 (Team Duy)**: Xây dựng `ProviderDescriptor` và `ProviderRegistry` tập trung từ `providers/factory.py` ➔ `domain/models/provider_descriptor.py` & `infrastructure/providers/provider_registry.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R03-T03 (Team Duy)**: Xây dựng `RoutingApplicationService` độc lập với Qt từ `core/routing/` ➔ `application/model_routing/routing_application_service.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R03-T04 (Team Duy)**: Di chuyển luồng gọi routing từ `ui/chat_panel.py#L638` sang `RoutingApplicationService`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R03-T05 (Team Duy)**: Di chuyển luồng gọi routing từ `ui/co4e_tab.py` và `ui/folder_tab.py` sang `RoutingApplicationService`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R03-T06 (Team Duy)**: Tách logic ghi nhận token usage ra khỏi Provider, chuyển thành `UsageEventSink` ➔ `infrastructure/telemetry/usage_sink.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
+- [x] **R03-T01 (Team Duy)**: Xây dựng bộ Contract Tests chuẩn hóa cho các Provider từ `providers/base.py` ➔ `tests/contracts/test_providers.py`  
+  *Start: `2026-08-21 10:10` | End: `2026-08-21 10:12`*
+- [x] **R03-T02 (Team Duy)**: Xây dựng `ProviderDescriptor` và `ProviderRegistry` tập trung từ `providers/factory.py` ➔ `domain/models/provider_descriptor.py` & `infrastructure/providers/provider_registry.py`  
+  *Start: `2026-08-21 10:06` | End: `2026-08-21 10:10`*
+- [x] **R03-T03 (Team Duy)**: Xây dựng `RoutingApplicationService` độc lập với Qt từ `core/routing/` ➔ `application/model_routing/routing_application_service.py`  
+  *Start: `2026-08-21 10:12` | End: `2026-08-21 10:15`*
+- [x] **R03-T04 (Team Duy)**: Di chuyển luồng gọi routing từ `ui/chat_panel.py#L638` sang `RoutingApplicationService`  
+  *Start: `2026-08-21 10:17` | End: `2026-08-21 10:20`*
+- [x] **R03-T05 (Team Duy)**: Di chuyển luồng gọi routing từ `ui/co4e_tab.py` và `ui/folder_tab.py` sang `RoutingApplicationService`  
+  *Start: `2026-08-21 10:20` | End: `2026-08-21 10:22`*
+- [x] **R03-T06 (Team Duy)**: Tách logic ghi nhận token usage ra khỏi Provider, chuyển thành `UsageEventSink` ➔ `infrastructure/telemetry/usage_sink.py`  
+  *Start: `2026-08-21 10:15` | End: `2026-08-21 10:17`*
 
 ---
 
diff --git a/domain/__init__.py b/domain/__init__.py
new file mode 100644
index 0000000..608b96a
--- /dev/null
+++ b/domain/__init__.py
@@ -0,0 +1,12 @@
+"""Domain layer - pure Python entities, value objects and events.
+
+The innermost layer of the 4-tier architecture (see
+``docs/architecture/ADR-001-layered-architecture.md``). Modules here describe
+WHAT the application is about - a turn of conversation, a model candidate, an
+agent event - and depend on nothing but the standard library.
+
+Hard rule (ADR-001 I1/I2, enforced by ``scripts/check_imports.py``): no imports
+of PySide6/PyQt, and no imports from ``application/``, ``infrastructure/``,
+``presentation/`` or the legacy ``core/``/``ui/`` packages. That is what keeps
+this layer testable in milliseconds and reusable from a headless scheduler.
+"""
diff --git a/domain/agents/__init__.py b/domain/agents/__init__.py
new file mode 100644
index 0000000..dee54b3
--- /dev/null
+++ b/domain/agents/__init__.py
@@ -0,0 +1,2 @@
+"""Domain entities for one agent turn: the request snapshot and the event
+stream it produces (EPIC R04)."""
diff --git a/domain/models/__init__.py b/domain/models/__init__.py
new file mode 100644
index 0000000..d9b090e
--- /dev/null
+++ b/domain/models/__init__.py
@@ -0,0 +1,5 @@
+"""Domain models: provider/model catalogue value objects (EPIC R03)."""
+
+from .provider_descriptor import ProviderCapability, ProviderDescriptor
+
+__all__ = ["ProviderDescriptor", "ProviderCapability"]
diff --git a/domain/models/provider_descriptor.py b/domain/models/provider_descriptor.py
new file mode 100644
index 0000000..aee772c
--- /dev/null
+++ b/domain/models/provider_descriptor.py
@@ -0,0 +1,171 @@
+"""ProviderDescriptor - the declarative catalogue entry for one model provider (R03-T02).
+
+Today the knowledge of "what a provider is" is scattered across three places
+that must be edited together and can silently drift apart:
+
+* ``providers/factory.py::_REGISTRY``  - name -> implementation class
+* ``config.py::DEFAULT_CONFIG["providers"]`` - default base_url / model / api_key
+* ``config.py::PROVIDER_LABELS``       - the human label shown in Settings
+
+Adding a provider means remembering all three; forgetting one produces a
+provider that exists but has no label, or a label with no implementation. This
+value object folds those facts into a single immutable description that the
+registry (``infrastructure/providers/provider_registry.py``) and the UI can both
+read, so a new provider is declared once.
+
+Pure domain code: stdlib only, no Qt, no network, no config access. It describes
+a provider; building one is infrastructure's job.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Dict, FrozenSet, List, Mapping, Optional, Tuple
+
+
+class ProviderCapability(str, Enum):
+    """What a provider can do, as advertised by its descriptor.
+
+    Kept as a closed enum rather than free-form strings so a typo
+    (``"vison"``) fails at import time instead of silently disabling a feature
+    at runtime. Inherits ``str`` so existing dict/JSON code that compares against
+    plain strings keeps working during the migration.
+    """
+
+    STREAMING = "streaming"      # can stream answer fragments through on_text
+    TOOLS = "tools"              # can be given a ToolSpec catalogue and call tools
+    VISION = "vision"            # accepts image content blocks (see providers/base.py)
+    REASONING = "reasoning"      # emits a separate private "thinking" stream
+    MODEL_LISTING = "model_listing"  # list_models() returns a real catalogue
+
+
+@dataclass(frozen=True)
+class ProviderDescriptor:
+    """An immutable description of one provider the app can talk to.
+
+    Attributes:
+        id: the config key, e.g. ``"openai_compat"``. Also the ``provider`` half
+            of a routing candidate key (``provider/model_id``).
+        label: human-readable name for Settings and the model picker.
+        protocol: which wire format this provider speaks. Several ids share one
+            protocol - ``ollama``, ``github_copilot`` and ``codex`` are all
+            OpenAI-compatible endpoints - which is exactly why protocol and id
+            must be separate fields.
+        default_model: the model used when the user has not chosen one.
+        capabilities: what the provider supports (see :class:`ProviderCapability`).
+        requires_api_key: whether an empty ``api_key`` makes it unusable.
+        requires_base_url: whether an empty ``base_url`` makes it unusable.
+        local: True when the endpoint runs on the user's own machine. Routing
+            treats local models as zero-cost, and the security layer treats them
+            as not leaving the machine, so this is a real behavioural flag and
+            not just documentation.
+        notes: free-form remark shown in Settings (e.g. "paste a Copilot token").
+    """
+
+    id: str
+    label: str
+    protocol: str
+    default_model: str = ""
+    capabilities: FrozenSet[ProviderCapability] = field(default_factory=frozenset)
+    requires_api_key: bool = True
+    requires_base_url: bool = True
+    local: bool = False
+    notes: str = ""
+
+    # -- capability queries ---------------------------------------------- #
+    def supports(self, capability: ProviderCapability) -> bool:
+        """True when this provider advertises ``capability``."""
+        return capability in self.capabilities
+
+    @property
+    def supports_vision(self) -> bool:
+        """Mirrors ``providers.base.Provider.supports_vision`` so callers can ask
+        the descriptor (no instance, no network) before building a provider."""
+        return self.supports(ProviderCapability.VISION)
+
+    @property
+    def supports_tools(self) -> bool:
+        """True when this provider can run an agent turn with tools. A provider
+        without it can still chat, but must never be routed a tool-using task."""
+        return self.supports(ProviderCapability.TOOLS)
+
+    def capability_names(self) -> List[str]:
+        """Capabilities as sorted plain strings - the shape the routing layer's
+        ``required_capabilities`` filter and the assessment store both use."""
+        return sorted(c.value for c in self.capabilities)
+
+    # -- configuration validation ---------------------------------------- #
+    def missing_settings(self, conf: Mapping[str, Any]) -> List[str]:
+        """Which required config keys are absent or blank in ``conf``.
+
+        Returned as a list (not a bool) so Settings can tell the user exactly
+        what to fill in, instead of a generic "not configured". A provider that
+        needs nothing returns an empty list.
+        """
+        missing: List[str] = []
+        if self.requires_api_key and not str(conf.get("api_key", "") or "").strip():
+            missing.append("api_key")
+        if self.requires_base_url and not str(conf.get("base_url", "") or "").strip():
+            missing.append("base_url")
+        return missing
+
+    def is_configured(self, conf: Mapping[str, Any]) -> bool:
+        """True when ``conf`` carries everything this provider needs to run."""
+        return not self.missing_settings(conf)
+
+    def resolve_model(self, conf: Optional[Mapping[str, Any]] = None,
+                      requested: str = "") -> str:
+        """Pick the model id for a call: explicit request, else configured, else
+        this descriptor's default.
+
+        Centralised here because the same three-step fallback is currently
+        re-implemented at every call site (chat panel, Co4E, AI-edit, scheduler),
+        and each of them gets the precedence subtly different.
+        """
+        if requested:
+            return requested
+        configured = str((conf or {}).get("model", "") or "").strip()
+        return configured or self.default_model
+
+    def describe(self, conf: Optional[Mapping[str, Any]] = None) -> str:
+        """One-line summary for logs and the Settings row, e.g.
+        ``"anthropic:claude-sonnet-4-6 (Anthropic Claude)"``."""
+        return f"{self.id}:{self.resolve_model(conf)} ({self.label})"
+
+    def candidate_key(self, model_id: str) -> str:
+        """The ``provider/model_id`` identity the routing layer keys on.
+
+        Defined here so the domain owns the format; ``core.routing.models`` has
+        its own ``candidate_key()`` helper producing the identical string, and
+        keeping them equal is what lets the new registry and the existing
+        assessment store share one keyspace during the migration.
+        """
+        return f"{self.id}/{model_id}"
+
+    def to_dict(self) -> Dict[str, Any]:
+        """JSON-safe projection, for persisting a catalogue snapshot or sending
+        the descriptor to a UI layer that must not import domain types."""
+        return {
+            "id": self.id,
+            "label": self.label,
+            "protocol": self.protocol,
+            "default_model": self.default_model,
+            "capabilities": self.capability_names(),
+            "requires_api_key": self.requires_api_key,
+            "requires_base_url": self.requires_base_url,
+            "local": self.local,
+            "notes": self.notes,
+        }
+
+
+def split_candidate_key(key: str) -> Tuple[str, str]:
+    """Inverse of :meth:`ProviderDescriptor.candidate_key`.
+
+    Splits on the FIRST ``/`` only: some gateways expose model ids that contain
+    a slash (``org/model``), and splitting on the last one would corrupt them.
+    """
+    provider, _, model_id = key.partition("/")
+    return provider, model_id
+
+
+__all__ = ["ProviderCapability", "ProviderDescriptor", "split_candidate_key"]
diff --git a/infrastructure/__init__.py b/infrastructure/__init__.py
new file mode 100644
index 0000000..80d70e7
--- /dev/null
+++ b/infrastructure/__init__.py
@@ -0,0 +1,7 @@
+"""Infrastructure layer - adapters to the outside world.
+
+Concrete implementations of what the inner layers only describe: HTTP calls to
+model gateways, the OS keyring, the filesystem, subprocesses, telemetry sinks.
+May import ``domain/`` (to speak its types) and third-party libraries, but never
+``presentation/``/``ui/``.
+"""
diff --git a/infrastructure/providers/__init__.py b/infrastructure/providers/__init__.py
new file mode 100644
index 0000000..0b63699
--- /dev/null
+++ b/infrastructure/providers/__init__.py
@@ -0,0 +1,5 @@
+"""Provider adapters and the central provider catalogue (EPIC R03)."""
+
+from .provider_registry import ProviderRegistry, default_registry
+
+__all__ = ["ProviderRegistry", "default_registry"]
diff --git a/infrastructure/providers/provider_registry.py b/infrastructure/providers/provider_registry.py
new file mode 100644
index 0000000..1c0d56e
--- /dev/null
+++ b/infrastructure/providers/provider_registry.py
@@ -0,0 +1,207 @@
+"""ProviderRegistry - the one place a provider is declared (R03-T02).
+
+Replaces the three-way split between ``providers/factory.py::_REGISTRY``,
+``config.py::DEFAULT_CONFIG["providers"]`` and ``config.py::PROVIDER_LABELS``
+with a single catalogue of :class:`ProviderDescriptor` objects plus the
+implementation class each one maps to.
+
+Adding a provider is now one entry in :data:`BUILT_IN_PROVIDERS` (declarative
+facts) and one line in :data:`_IMPLEMENTATIONS` (which class speaks that
+protocol) - see ``docs/governance/contributor-recipes.md`` (R10-T04).
+
+Migration note (strangler fig, ADR-001 section 4): this registry does not
+re-implement any provider. It builds the SAME classes ``providers/factory.py``
+builds, so both entry points stay behaviourally identical while call sites move
+over one at a time.
+"""
+from __future__ import annotations
+
+from typing import Any, Dict, Iterable, List, Mapping, Optional
+
+from cowork_local.domain.models.provider_descriptor import (
+    ProviderCapability,
+    ProviderDescriptor,
+)
+from cowork_local.providers.base import Provider, ProviderError
+
+_CAP = ProviderCapability
+
+# Every provider the app ships with, described once.
+#
+# The capability sets are deliberately conservative: a capability listed here is
+# one the adapter genuinely implements today. Claiming VISION for a provider
+# whose chat() cannot translate an image block would route an image turn into a
+# guaranteed failure, so an unimplemented capability must stay off the list.
+BUILT_IN_PROVIDERS: tuple = (
+    ProviderDescriptor(
+        id="openai_compat",
+        label="OpenAI-compatible (Internal Gateway)",
+        protocol="openai_compat",
+        default_model="gpt-4o-mini",
+        capabilities=frozenset({_CAP.STREAMING, _CAP.TOOLS, _CAP.VISION,
+                                _CAP.REASONING, _CAP.MODEL_LISTING}),
+        notes="Any endpoint speaking the OpenAI Chat Completions protocol.",
+    ),
+    ProviderDescriptor(
+        id="anthropic",
+        label="Anthropic Claude",
+        protocol="anthropic",
+        default_model="claude-sonnet-4-6",
+        capabilities=frozenset({_CAP.STREAMING, _CAP.TOOLS, _CAP.VISION,
+                                _CAP.MODEL_LISTING}),
+    ),
+    ProviderDescriptor(
+        id="ollama",
+        label="Ollama (local models)",
+        protocol="openai_compat",
+        default_model="llama3.1",
+        capabilities=frozenset({_CAP.STREAMING, _CAP.TOOLS, _CAP.REASONING,
+                                _CAP.MODEL_LISTING}),
+        # Ollama ignores the key, but the OpenAI client layer requires a value,
+        # so the default config ships a placeholder rather than an empty string.
+        requires_api_key=False,
+        local=True,
+        notes="Runs on this machine - no data leaves the device, no token cost.",
+    ),
+    ProviderDescriptor(
+        id="github_copilot",
+        label="GitHub Copilot",
+        protocol="openai_compat",
+        default_model="gpt-4o",
+        capabilities=frozenset({_CAP.STREAMING, _CAP.TOOLS, _CAP.MODEL_LISTING}),
+        notes="Paste a Copilot token as the API key.",
+    ),
+    ProviderDescriptor(
+        id="codex",
+        label="OpenAI (Codex / GPT)",
+        protocol="openai_compat",
+        default_model="gpt-4o-mini",
+        capabilities=frozenset({_CAP.STREAMING, _CAP.TOOLS, _CAP.VISION,
+                                _CAP.REASONING, _CAP.MODEL_LISTING}),
+    ),
+)
+
+
+def _implementations() -> Dict[str, type]:
+    """Protocol -> adapter class.
+
+    Imported lazily inside the function because ``providers/anthropic.py`` and
+    ``providers/openai_compat.py`` pull in ``requests`` at import time; keeping
+    that out of module import means a test that only inspects descriptors pays
+    no import cost at all.
+    """
+    from cowork_local.providers.anthropic import AnthropicProvider
+    from cowork_local.providers.openai_compat import OpenAICompatProvider
+
+    return {
+        "openai_compat": OpenAICompatProvider,
+        "anthropic": AnthropicProvider,
+    }
+
+
+class ProviderRegistry:
+    """Catalogue of known providers + the factory that instantiates them.
+
+    Intentionally holds no config and no app context: it is a pure lookup table
+    plus a build step, so it can be constructed in a test with a custom
+    descriptor list and no application running.
+    """
+
+    def __init__(self, descriptors: Optional[Iterable[ProviderDescriptor]] = None) -> None:
+        # Dict preserves declaration order (Python 3.7+), which is the order
+        # Settings lists providers in - so the catalogue order is data, not luck.
+        self._by_id: Dict[str, ProviderDescriptor] = {
+            d.id: d for d in (descriptors if descriptors is not None else BUILT_IN_PROVIDERS)
+        }
+
+    # -- catalogue queries ------------------------------------------------ #
+    def ids(self) -> List[str]:
+        """Known provider ids, in declaration order."""
+        return list(self._by_id)
+
+    def all(self) -> List[ProviderDescriptor]:
+        """Every descriptor, in declaration order."""
+        return list(self._by_id.values())
+
+    def get(self, provider_id: str) -> Optional[ProviderDescriptor]:
+        """The descriptor for ``provider_id``, or None when unknown.
+
+        Returns None rather than raising because the caller is often reacting to
+        a config file that may name a provider from a newer version; the UI
+        should be able to skip it, not crash.
+        """
+        return self._by_id.get(provider_id)
+
+    def require(self, provider_id: str) -> ProviderDescriptor:
+        """Like :meth:`get` but raises :class:`ProviderError` when unknown.
+
+        Same error type ``providers/factory.py::build_provider`` already raises,
+        so callers that migrate to the registry keep their existing except clause.
+        """
+        descriptor = self._by_id.get(provider_id)
+        if descriptor is None:
+            known = ", ".join(self._by_id) or "(none)"
+            raise ProviderError(f"Unsupported provider: {provider_id} (known: {known})")
+        return descriptor
+
+    def labels(self) -> Dict[str, str]:
+        """``{id: label}`` - the drop-in replacement for ``config.PROVIDER_LABELS``."""
+        return {d.id: d.label for d in self._by_id.values()}
+
+    def supporting(self, capability: ProviderCapability) -> List[ProviderDescriptor]:
+        """Every descriptor advertising ``capability`` - used to answer "which
+        providers could serve this turn?" before any of them is built."""
+        return [d for d in self._by_id.values() if d.supports(capability)]
+
+    def configured(self, providers_conf: Mapping[str, Mapping[str, Any]]
+                   ) -> List[ProviderDescriptor]:
+        """Descriptors whose config section is complete enough to actually call.
+
+        ``providers_conf`` is ``AppConfig.data["providers"]``. Passing the raw
+        mapping (not the AppConfig object) keeps this layer independent of the
+        config implementation, which EPIC R02 is rewriting in parallel.
+        """
+        return [d for d in self._by_id.values()
+                if d.is_configured(providers_conf.get(d.id, {}) or {})]
+
+    # -- construction ----------------------------------------------------- #
+    def build(self, provider_id: str, conf: Mapping[str, Any],
+              model: str = "") -> Provider:
+        """Instantiate the adapter for ``provider_id``.
+
+        ``model`` overrides the configured model for this instance only - that is
+        how the routing layer runs one turn on a different model without mutating
+        the user's saved settings.
+        """
+        descriptor = self.require(provider_id)
+        impl = _implementations().get(descriptor.protocol)
+        if impl is None:  # pragma: no cover - only reachable via a bad descriptor
+            raise ProviderError(
+                f"Provider '{provider_id}' declares unknown protocol "
+                f"'{descriptor.protocol}'."
+            )
+        # Copy before mutating: conf is the caller's live config dict, and
+        # writing the routed model into it would silently change the user's
+        # saved default for every later turn.
+        resolved = dict(conf or {})
+        resolved["model"] = descriptor.resolve_model(conf, model)
+        instance = impl(resolved)
+        # The adapter class is shared by several ids (three of them are
+        # OpenAI-compatible), so its class-level `name` cannot identify which
+        # provider this is. Stamping the instance keeps usage records, audit
+        # entries and routing candidate keys attributed to the right provider.
+        instance.name = descriptor.id
+        return instance
+
+    def describe(self, provider_id: str, conf: Optional[Mapping[str, Any]] = None) -> str:
+        """One-line description used in logs and error messages."""
+        return self.require(provider_id).describe(conf)
+
+
+# Shared default instance. Callers that need the built-in catalogue use this
+# instead of constructing a registry each time; tests build their own with an
+# explicit descriptor list.
+default_registry = ProviderRegistry()
+
+
+__all__ = ["ProviderRegistry", "BUILT_IN_PROVIDERS", "default_registry"]
diff --git a/infrastructure/telemetry/__init__.py b/infrastructure/telemetry/__init__.py
new file mode 100644
index 0000000..e0b3461
--- /dev/null
+++ b/infrastructure/telemetry/__init__.py
@@ -0,0 +1,21 @@
+"""Telemetry sinks: where token usage and turn metrics are recorded (EPIC R03)."""
+
+from .usage_sink import (
+    NullUsageSink,
+    RecordingUsageSink,
+    UsageEvent,
+    UsageEventSink,
+    UsageTrackerSink,
+    default_sink,
+    set_default_sink,
+)
+
+__all__ = [
+    "UsageEvent",
+    "UsageEventSink",
+    "UsageTrackerSink",
+    "NullUsageSink",
+    "RecordingUsageSink",
+    "default_sink",
+    "set_default_sink",
+]
diff --git a/infrastructure/telemetry/usage_sink.py b/infrastructure/telemetry/usage_sink.py
new file mode 100644
index 0000000..4323475
--- /dev/null
+++ b/infrastructure/telemetry/usage_sink.py
@@ -0,0 +1,229 @@
+"""UsageEventSink - where a turn's token usage goes (R03-T06).
+
+Today each provider records its own usage inline, in the middle of the streaming
+loop::
+
+    # providers/openai_compat.py
+    def _record_usage(self, messages, text_parts, tool_acc, usage_seen):
+        from ..core import usage_tracker as ut
+        ...
+        ut.record(self.name, self.model, ...)
+
+Three problems with that shape:
+
+1. **Hidden side effect.** ``chat()`` looks like a pure request/response call but
+   also writes to the Dashboard's store, so a test of a provider silently
+   appends rows to the developer's real usage history.
+2. **Duplicated estimation.** The "no usage block from the server, so estimate
+   at ~4 chars/token" fallback is copy-pasted per provider and can drift.
+3. **One hard-wired destination.** Usage can only ever go to
+   ``core.usage_tracker``; a run that wants to bill a workflow, or a test that
+   wants to assert on token counts, has nowhere to plug in.
+
+This module introduces the seam: providers build a :class:`UsageEvent` and hand
+it to a :class:`UsageEventSink`. Production wires :class:`UsageTrackerSink`
+(same destination, same numbers as before); tests wire
+:class:`RecordingUsageSink` or :class:`NullUsageSink`.
+"""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Protocol, Sequence
+
+logger = logging.getLogger("cowork_local.telemetry")
+
+# Rough characters-per-token ratio used when the gateway sends no usage block.
+# Matches the constant behaviour of ``core.usage_tracker.estimate_tokens`` so
+# moving the estimation here does not change a single recorded number.
+_CHARS_PER_TOKEN = 4
+
+
+@dataclass(frozen=True)
+class UsageEvent:
+    """Token usage for exactly one provider round trip.
+
+    ``estimated`` marks a record derived from text length rather than reported by
+    the server. The Dashboard shows the two differently, and conflating them
+    would make cost figures look more precise than they are.
+    """
+
+    provider: str
+    model: str
+    input_tokens: int = 0
+    output_tokens: int = 0
+    cached_tokens: int = 0
+    estimated: bool = False
+
+    @property
+    def total_tokens(self) -> int:
+        """Input + output. Cached tokens are a subset of input, not an addition,
+        so adding them here would double-count a cache hit."""
+        return self.input_tokens + self.output_tokens
+
+    def to_dict(self) -> Dict[str, Any]:
+        """JSON-safe projection for logs and for sinks that persist raw events."""
+        return {
+            "provider": self.provider,
+            "model": self.model,
+            "input_tokens": self.input_tokens,
+            "output_tokens": self.output_tokens,
+            "cached_tokens": self.cached_tokens,
+            "estimated": self.estimated,
+        }
+
+
+class UsageEventSink(Protocol):
+    """Anything that can absorb a :class:`UsageEvent`.
+
+    Implementations MUST NOT raise: telemetry is observability, and a failure to
+    record usage must never abort the turn that produced it.
+    """
+
+    def record(self, event: UsageEvent) -> None:
+        """Absorb one usage event."""
+
+
+class NullUsageSink:
+    """Discards everything. The default for tests and headless tooling, so a
+    unit test never writes into the developer's real usage history."""
+
+    def record(self, event: UsageEvent) -> None:  # noqa: D102 - see protocol
+        return None
+
+
+class RecordingUsageSink:
+    """Keeps events in memory so a test can assert on what was recorded."""
+
+    def __init__(self) -> None:
+        self.events: List[UsageEvent] = []
+
+    def record(self, event: UsageEvent) -> None:  # noqa: D102 - see protocol
+        self.events.append(event)
+
+    @property
+    def total_tokens(self) -> int:
+        """Sum across every recorded event."""
+        return sum(e.total_tokens for e in self.events)
+
+
+class UsageTrackerSink:
+    """Forwards to ``core.usage_tracker`` - the Dashboard's store.
+
+    This is the production sink and the only place that still knows about the
+    legacy tracker module, which is what lets EPIC R10 replace the storage
+    without touching a single provider.
+    """
+
+    def __init__(self, tracker: Optional[Any] = None) -> None:
+        # Injectable for tests; imported lazily otherwise because the tracker
+        # touches the config directory at import time.
+        self._tracker = tracker
+
+    def _resolve(self) -> Any:
+        if self._tracker is None:
+            from cowork_local.core import usage_tracker
+
+            self._tracker = usage_tracker
+        return self._tracker
+
+    def record(self, event: UsageEvent) -> None:
+        """Write the event to the usage tracker, swallowing any failure.
+
+        The bare except mirrors the behaviour this replaces (each provider
+        already wrapped its ``ut.record`` call in ``try/except: pass``) but logs
+        at debug level instead of discarding the reason entirely, so a broken
+        Dashboard store can at least be diagnosed.
+        """
+        try:
+            self._resolve().record(
+                event.provider, event.model,
+                event.input_tokens, event.output_tokens, event.cached_tokens,
+                estimated=event.estimated,
+            )
+        except Exception:  # noqa: BLE001 - telemetry must never break a turn
+            logger.debug("usage sink: failed to record %s", event.to_dict(), exc_info=True)
+
+
+def estimate_tokens(text: str) -> int:
+    """Approximate token count for ``text`` (~4 characters per token).
+
+    Deliberately identical to ``core.usage_tracker.estimate_tokens`` so that
+    moving estimation into this layer changes no recorded number. Duplicated
+    rather than imported to keep this module free of the legacy dependency;
+    :class:`UsageTrackerSink` is the only bridge back to it.
+    """
+    return max(0, len(text or "") // _CHARS_PER_TOKEN)
+
+
+def estimated_event(provider: str, model: str, sent: str, received: str) -> UsageEvent:
+    """Build an estimated :class:`UsageEvent` from the raw text of a round trip.
+
+    Used when the gateway sends no usage block - most self-hosted OpenAI-compatible
+    servers and Ollama do not.
+    """
+    return UsageEvent(
+        provider=provider, model=model,
+        input_tokens=estimate_tokens(sent),
+        output_tokens=estimate_tokens(received),
+        cached_tokens=0,
+        estimated=True,
+    )
+
+
+def openai_usage_event(provider: str, model: str, usage: Dict[str, Any]) -> UsageEvent:
+    """Build a reported :class:`UsageEvent` from an OpenAI-style usage block."""
+    details = usage.get("prompt_tokens_details") or {}
+    return UsageEvent(
+        provider=provider, model=model,
+        input_tokens=int(usage.get("prompt_tokens", 0) or 0),
+        output_tokens=int(usage.get("completion_tokens", 0) or 0),
+        cached_tokens=int(details.get("cached_tokens", 0) or 0),
+        estimated=False,
+    )
+
+
+def anthropic_usage_event(provider: str, model: str, usage: Dict[str, Any]) -> UsageEvent:
+    """Build a reported :class:`UsageEvent` from Anthropic's usage accumulator.
+
+    Anthropic reports input tokens on ``message_start`` and output tokens on
+    ``message_delta``, so ``providers/anthropic.py`` accumulates them into a dict
+    keyed ``in``/``out``/``cache`` - this reads that shape.
+    """
+    return UsageEvent(
+        provider=provider, model=model,
+        input_tokens=int(usage.get("in", 0) or 0),
+        output_tokens=int(usage.get("out", 0) or 0),
+        cached_tokens=int(usage.get("cache", 0) or 0),
+        estimated=False,
+    )
+
+
+# The sink providers use unless one is injected. A module-level default keeps
+# the change to the provider classes to a single attribute, and lets a test swap
+# the destination process-wide with one monkeypatch.
+default_sink: UsageEventSink = UsageTrackerSink()
+
+
+def set_default_sink(sink: UsageEventSink) -> UsageEventSink:
+    """Replace the process-wide default sink; returns the previous one so a
+    caller (or fixture) can restore it."""
+    global default_sink
+    previous = default_sink
+    default_sink = sink
+    return previous
+
+
+__all__ = [
+    "UsageEvent",
+    "UsageEventSink",
+    "UsageTrackerSink",
+    "NullUsageSink",
+    "RecordingUsageSink",
+    "estimate_tokens",
+    "estimated_event",
+    "openai_usage_event",
+    "anthropic_usage_event",
+    "default_sink",
+    "set_default_sink",
+]
diff --git a/providers/anthropic.py b/providers/anthropic.py
index 0d63437..5addb00 100644
--- a/providers/anthropic.py
+++ b/providers/anthropic.py
@@ -292,21 +292,19 @@ class AnthropicProvider(Provider):
                 args = {"_raw": b["json"]}
             tool_calls.append({"id": b["id"], "name": b["name"], "arguments": args})
 
-        # Dashboard usage event — real counts from the stream's usage events,
-        # else a ~4 chars/token estimate. Never breaks the turn.
-        try:
-            from ..core import usage_tracker as ut
+        # Dashboard usage event — real counts from the stream's usage events
+        # (input arrives on message_start, output on message_delta), else a
+        # ~4 chars/token estimate. Delivery is the sink's job (R03-T06), so this
+        # only translates Anthropic's wire shape into a canonical UsageEvent.
+        from ..infrastructure.telemetry import usage_sink as telemetry
 
-            if usage_seen:
-                ut.record(self.name, self.model, usage_seen.get("in", 0),
-                          usage_seen.get("out", 0), usage_seen.get("cache", 0))
-            else:
-                sent = json.dumps(payload.get("messages", []), ensure_ascii=False)
-                got = "".join(text_parts) + "".join(b["json"] for b in blocks.values())
-                ut.record(self.name, self.model, ut.estimate_tokens(sent),
-                          ut.estimate_tokens(got), 0, estimated=True)
-        except Exception:  # noqa: BLE001
-            pass
+        if usage_seen:
+            event = telemetry.anthropic_usage_event(self.name, self.model, usage_seen)
+        else:
+            sent = json.dumps(payload.get("messages", []), ensure_ascii=False)
+            got = "".join(text_parts) + "".join(b["json"] for b in blocks.values())
+            event = telemetry.estimated_event(self.name, self.model, sent, got)
+        self._emit_usage(event)
 
         return {"role": "assistant", "content": "".join(text_parts), "tool_calls": tool_calls}
 
diff --git a/providers/base.py b/providers/base.py
index a4d9f9c..b693a57 100644
--- a/providers/base.py
+++ b/providers/base.py
@@ -224,6 +224,12 @@ class Provider:
         # silently swallowing the error — Settings' "Test connection" / "Load
         # models" surfaces this so "model won't load" has a concrete reason.
         self.last_error = ""
+        # Where this provider's token usage goes (R03-T06). None means "the
+        # process-wide default sink", resolved lazily in _emit_usage so that a
+        # test can swap the destination without rebuilding every provider.
+        # Set it per instance to bill one run somewhere else (a workflow, a
+        # scheduled task) without touching global state.
+        self.usage_sink = None
 
     def chat(
         self,
@@ -274,6 +280,24 @@ class Provider:
             return True, f"OK — {len(models)} model(s) available."
         return False, "No models returned. Check base_url/API key and network access."
 
+    # -- telemetry -----------------------------------------------------
+    def _emit_usage(self, event) -> None:
+        """Hand one ``UsageEvent`` to this provider's usage sink.
+
+        Never raises: recording how many tokens a turn cost must not be able to
+        fail the turn itself. Falls back to the process-wide default sink so
+        existing call sites keep reporting to the Dashboard exactly as before
+        (see infrastructure/telemetry/usage_sink.py)."""
+        try:
+            sink = self.usage_sink
+            if sink is None:
+                from ..infrastructure.telemetry import usage_sink as telemetry
+
+                sink = telemetry.default_sink
+            sink.record(event)
+        except Exception:  # noqa: BLE001 — telemetry is never worth a failed turn
+            pass
+
     # -- shared helpers ------------------------------------------------
     @staticmethod
     def _is_cancelled(cancel) -> bool:
diff --git a/providers/openai_compat.py b/providers/openai_compat.py
index c45083f..664978f 100644
--- a/providers/openai_compat.py
+++ b/providers/openai_compat.py
@@ -268,22 +268,25 @@ class OpenAICompatProvider(Provider):
     def _record_usage(self, messages, text_parts, tool_acc, usage_seen) -> None:
         """One Dashboard usage event per turn: real counts when the server's
         final chunk carried a "usage" block, a ~4 chars/token estimate
-        otherwise. Never breaks the turn."""
-        try:
-            from ..core import usage_tracker as ut
+        otherwise.
 
-            if usage_seen:
-                ut.record(self.name, self.model,
-                          usage_seen.get("prompt_tokens", 0),
-                          usage_seen.get("completion_tokens", 0),
-                          (usage_seen.get("prompt_tokens_details") or {}).get("cached_tokens", 0))
-            else:
-                sent = json.dumps(self._to_api_messages(messages), ensure_ascii=False)
-                got = "".join(text_parts) + "".join(s["args"] for s in tool_acc.values())
-                ut.record(self.name, self.model, ut.estimate_tokens(sent),
-                          ut.estimate_tokens(got), 0, estimated=True)
-        except Exception:  # noqa: BLE001
-            pass
+        Building the event and delivering it are now separate concerns (R03-T06):
+        this method only translates THIS provider's wire shape into a canonical
+        ``UsageEvent``; where it ends up is the sink's decision, so a test can
+        assert on token counts without writing to the real Dashboard store."""
+        from ..infrastructure.telemetry import usage_sink as telemetry
+
+        if usage_seen:
+            event = telemetry.openai_usage_event(self.name, self.model, usage_seen)
+        else:
+            # No usage block from the gateway (self-hosted servers and Ollama
+            # never send one) - fall back to estimating from the raw text of
+            # both directions, tool-call arguments included since the model was
+            # billed for generating them.
+            sent = json.dumps(self._to_api_messages(messages), ensure_ascii=False)
+            got = "".join(text_parts) + "".join(s["args"] for s in tool_acc.values())
+            event = telemetry.estimated_event(self.name, self.model, sent, got)
+        self._emit_usage(event)
 
     def list_models(self):
         self.last_error = ""
diff --git a/state.py b/state.py
index 98ab7a1..fc1d35e 100644
--- a/state.py
+++ b/state.py
@@ -52,7 +52,16 @@ class AppContext:
         # own event loop), so concurrent model calls never needed serializing.
         self._conn_lock = threading.Lock()
         self._routing_service = None       # lazy RoutingService (Auto Model Routing)
+        # Lazy RoutingApplicationService (R03-T03) — the Qt-free decision layer
+        # every chat surface now routes through. Wraps _routing_service, which
+        # stays the scoring/ranking engine underneath.
+        self._routing_application = None
         self._routing_lock = threading.Lock()
+        # A SEPARATE lock for the application service: building it calls
+        # routing(), which takes _routing_lock. threading.Lock is not
+        # reentrant, so sharing one lock across both accessors deadlocks the
+        # first caller instead of just serialising them.
+        self._routing_app_lock = threading.Lock()
         # The workspace (project) currently selected in the Workspace screen.
         # Per-workspace modes (routing + auto-run) resolve against THIS project
         # so each workspace keeps its own modes. Updated by WorkspaceTab on
@@ -79,16 +88,28 @@ class AppContext:
         workspace keep its own routing mode."""
         project = self._current_project()
         if project is not None:
+            # Validated through the single mode vocabulary (R03-T03) rather
+            # than a literal tuple, so a workspace can store any mode the
+            # routing service understands - including "fallback", whose
+            # on-screen toggle arrives in EPIC R08.
+            from .application.model_routing import is_valid_mode, normalize_mode
+
             mode = (project.routing_modes or {}).get(surface, "")
-            if mode in ("off", "auto", "manual"):
-                return mode
+            # Only a RECOGNISED override wins; an empty or corrupt value falls
+            # through to the global setting, exactly as before. Validation goes
+            # through the routing vocabulary (R03-T03) instead of a literal
+            # tuple, so a new mode works everywhere the moment it is defined.
+            if is_valid_mode(mode):
+                return normalize_mode(mode)
         return self.config.routing_mode_for(surface)
 
     def set_project_routing_mode(self, surface: str, mode: str) -> None:
         """Persist a surface's routing mode for the ACTIVE workspace. With no
         workspace selected, falls back to the global setting so behaviour
         outside a project stays global."""
-        mode = mode if mode in ("off", "auto", "manual") else "off"
+        from .application.model_routing import normalize_mode
+
+        mode = normalize_mode(mode)
         project = self._current_project()
         if project is None:
             self.config.set_routing_mode_for(surface, mode)
@@ -144,6 +165,34 @@ class AppContext:
                     self._routing_service = RoutingService(self)
         return self._routing_service
 
+    def routing_application(self):
+        """The shared :class:`RoutingApplicationService` (R03-T03).
+
+        This is what UI code should call: it owns the Off/Auto/Manual/Fallback
+        policy, the confirm handshake and the never-raise guarantee, while
+        :meth:`routing` remains the scoring engine underneath. Chat, Co4E and
+        AI-Edit all go through this one object, so a change to routing policy is
+        made once instead of three times.
+
+        Built lazily and memoised for the same reason as :meth:`routing`: the
+        pending-switch registry and assessment store must be shared app-wide."""
+        if self._routing_application is None:
+            # Resolve the engine BEFORE taking this lock: routing() takes
+            # _routing_lock, and nesting the two acquisitions is what makes the
+            # ordering fragile in the first place.
+            engine = self.routing()
+            with self._routing_app_lock:
+                if self._routing_application is None:
+                    from .application.model_routing import RoutingApplicationService
+
+                    self._routing_application = RoutingApplicationService(
+                        engine,
+                        # Per-workspace mode lookup, so each workspace keeps its
+                        # own routing behaviour (see project_routing_mode).
+                        mode_reader=self.project_routing_mode,
+                    )
+        return self._routing_application
+
     def build_active_provider(self):
         """Construct the currently selected provider (called inside workers)."""
         return self.build_provider_for(self.config.active_provider)
diff --git a/tests/contracts/__init__.py b/tests/contracts/__init__.py
new file mode 100644
index 0000000..33b58a9
--- /dev/null
+++ b/tests/contracts/__init__.py
@@ -0,0 +1,8 @@
+"""Contract tests: one shared behaviour suite every implementation must satisfy.
+
+Unlike unit tests (which test one module in isolation) a contract test is
+parametrised over EVERY implementation of an interface, so a newly added
+provider either satisfies the same promises as the existing ones or the suite
+goes red on the day it is added - not months later, in production, on the one
+code path that assumed the promise held.
+"""
diff --git a/tests/contracts/test_providers.py b/tests/contracts/test_providers.py
new file mode 100644
index 0000000..be2d137
--- /dev/null
+++ b/tests/contracts/test_providers.py
@@ -0,0 +1,342 @@
+"""Provider contract suite (R03-T01).
+
+Every provider - the two real adapters and the test double - must honour the
+same promises declared in ``providers/base.py``:
+
+1. ``chat()`` returns the canonical assistant message
+   ``{"role": "assistant", "content": str, "tool_calls": [...]}``.
+2. Answer text is streamed through ``on_text`` and equals the returned content.
+3. Private reasoning goes to ``on_reasoning`` ONLY - it must never leak into the
+   answer, or a reasoning model's chain of thought ends up persisted in history.
+4. Tool calls come back as ``{"id", "name", "arguments": dict}`` with arguments
+   already parsed - callers must never have to json.loads() them.
+5. A failure raises ``ProviderError`` and nothing else, so one except clause in
+   the agent loop covers every provider.
+
+The real adapters are exercised WITHOUT network access by replacing
+``Provider._request`` with a canned SSE response - which is exactly the seam
+``providers/base.py`` documents for its TLS retry, so no production code needed
+changing to make this testable.
+"""
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Optional
+
+import pytest
+
+from cowork_local.domain.models.provider_descriptor import ProviderCapability
+from cowork_local.infrastructure.providers.provider_registry import (
+    BUILT_IN_PROVIDERS,
+    ProviderRegistry,
+)
+from cowork_local.providers.anthropic import AnthropicProvider
+from cowork_local.providers.base import Provider, ProviderError, ToolSpec
+from cowork_local.providers.openai_compat import OpenAICompatProvider
+from tests.fakes import FakeProvider, ScriptedTurn
+
+
+class _StubResponse:
+    """Minimal stand-in for a streamed ``requests.Response``.
+
+    Only the members the provider code actually touches are implemented; adding
+    more would invite tests that pass against the stub but not against requests.
+    """
+
+    def __init__(self, lines: List[str], status_code: int = 200, text: str = "") -> None:
+        self._lines = lines
+        self.status_code = status_code
+        self.text = text
+        self.headers: Dict[str, str] = {}
+        self.encoding = "utf-8"
+        self.closed = False
+
+    def iter_lines(self, decode_unicode: bool = False):
+        yield from self._lines
+
+    def close(self) -> None:
+        self.closed = True
+
+    def json(self) -> Any:
+        return json.loads(self.text or "{}")
+
+
+def _sse(*payloads: Dict[str, Any]) -> List[str]:
+    """Render payloads as SSE ``data:`` lines, the wire shape both adapters parse."""
+    return [f"data: {json.dumps(p)}" for p in payloads]
+
+
+@pytest.fixture
+def canned(monkeypatch):
+    """Return a helper that makes every provider request answer with ``lines``."""
+
+    def _install(lines: List[str], status_code: int = 200, text: str = "") -> Dict[str, Any]:
+        seen: Dict[str, Any] = {}
+
+        def fake_request(self, method, url, **kwargs):
+            # Capture the outgoing payload so tests can assert on how the
+            # canonical message list was translated to the provider's wire format.
+            seen["method"] = method
+            seen["url"] = url
+            seen["json"] = kwargs.get("json")
+            return _StubResponse(lines, status_code=status_code, text=text)
+
+        monkeypatch.setattr(Provider, "_request", fake_request, raising=True)
+        return seen
+
+    return _install
+
+
+# --------------------------------------------------------------------------- #
+# Shared base-class behaviour every provider inherits
+# --------------------------------------------------------------------------- #
+def _providers_under_test() -> List[Provider]:
+    """One instance of each implementation, configured but never called."""
+    conf = {"base_url": "https://example.invalid/v1", "api_key": "k", "model": "m"}
+    return [
+        OpenAICompatProvider(dict(conf)),
+        AnthropicProvider(dict(conf)),
+        FakeProvider(),
+    ]
+
+
+@pytest.mark.parametrize("provider", _providers_under_test(), ids=lambda p: type(p).__name__)
+def test_every_provider_exposes_the_base_contract(provider):
+    assert isinstance(provider, Provider)
+    assert callable(provider.chat)
+    assert callable(provider.list_models)
+    assert callable(provider.test_connection)
+    # `name` identifies the provider in usage records and audit entries; an
+    # implementation that forgot to set it would silently report as "base".
+    assert provider.name and provider.name != "base"
+    assert isinstance(provider.supports_vision, bool)
+    assert provider.describe() == f"{provider.name}:{provider.model}"
+
+
+@pytest.mark.parametrize("provider", _providers_under_test(), ids=lambda p: type(p).__name__)
+def test_strip_think_removes_inline_reasoning_from_a_final_answer(provider):
+    """Safety net for gateways that fold reasoning into the content stream: the
+    answer stored in history must never contain a <think> block."""
+    assert provider.strip_think("<think>secret</think>Answer") == "Answer"
+    assert provider.strip_think("Plain answer") == "Plain answer"
+
+
+def test_tool_spec_translates_to_both_wire_formats():
+    """One ToolSpec must render for both protocols - this is what lets the agent
+    loop build its tool catalogue once and reuse it across providers."""
+    spec = ToolSpec(name="save_file", description="Write a file",
+                    parameters={"type": "object", "properties": {}})
+
+    openai_shape = spec.to_openai()
+    anthropic_shape = spec.to_anthropic()
+
+    assert openai_shape["type"] == "function"
+    assert openai_shape["function"]["name"] == "save_file"
+    assert openai_shape["function"]["parameters"] == spec.parameters
+    # Anthropic names the same field `input_schema`; the values must stay equal,
+    # otherwise the same tool would validate differently per provider.
+    assert anthropic_shape["name"] == "save_file"
+    assert anthropic_shape["input_schema"] == spec.parameters
+
+
+# --------------------------------------------------------------------------- #
+# Streaming contract - real adapters, canned transport
+# --------------------------------------------------------------------------- #
+def test_openai_compat_streams_text_and_returns_canonical_message(canned):
+    canned(_sse(
+        {"choices": [{"delta": {"content": "Hel"}}]},
+        {"choices": [{"delta": {"content": "lo"}}]},
+    ) + ["data: [DONE]"])
+    provider = OpenAICompatProvider({"base_url": "https://x.invalid/v1",
+                                     "api_key": "k", "model": "m"})
+    chunks: List[str] = []
+
+    result = provider.chat([{"role": "user", "content": "hi"}], on_text=chunks.append)
+
+    assert "".join(chunks) == "Hello"
+    assert result["role"] == "assistant"
+    assert result["content"] == "Hello"
+    assert result["tool_calls"] == []
+
+
+def test_openai_compat_keeps_reasoning_out_of_the_answer(canned):
+    canned(_sse(
+        {"choices": [{"delta": {"reasoning_content": "hmm..."}}]},
+        {"choices": [{"delta": {"content": "42"}}]},
+    ) + ["data: [DONE]"])
+    provider = OpenAICompatProvider({"base_url": "https://x.invalid/v1",
+                                     "api_key": "k", "model": "m"})
+    text: List[str] = []
+    reasoning: List[str] = []
+
+    result = provider.chat([{"role": "user", "content": "q"}],
+                           on_text=text.append, on_reasoning=reasoning.append)
+
+    assert reasoning == ["hmm..."]
+    assert result["content"] == "42"
+    assert "hmm" not in result["content"]
+
+
+def test_openai_compat_returns_tool_calls_with_parsed_arguments(canned):
+    """Arguments arrive as a JSON string split across chunks; the contract says
+    the caller receives a ready-to-use dict."""
+    canned(_sse(
+        {"choices": [{"delta": {"tool_calls": [
+            {"index": 0, "id": "call_a", "function": {"name": "save_file",
+                                                      "arguments": '{"filename":'}}]}}]},
+        {"choices": [{"delta": {"tool_calls": [
+            {"index": 0, "function": {"arguments": '"a.md"}'}}]}}]},
+    ) + ["data: [DONE]"])
+    provider = OpenAICompatProvider({"base_url": "https://x.invalid/v1",
+                                     "api_key": "k", "model": "m"})
+
+    result = provider.chat([{"role": "user", "content": "save it"}])
+
+    assert len(result["tool_calls"]) == 1
+    call = result["tool_calls"][0]
+    assert call["id"] == "call_a"
+    assert call["name"] == "save_file"
+    assert call["arguments"] == {"filename": "a.md"}
+
+
+def test_anthropic_streams_text_and_returns_canonical_message(canned):
+    canned(_sse(
+        {"type": "content_block_delta", "index": 0,
+         "delta": {"type": "text_delta", "text": "Hel"}},
+        {"type": "content_block_delta", "index": 0,
+         "delta": {"type": "text_delta", "text": "lo"}},
+        {"type": "message_stop"},
+    ))
+    provider = AnthropicProvider({"base_url": "https://x.invalid",
+                                  "api_key": "k", "model": "m"})
+    chunks: List[str] = []
+
+    result = provider.chat([{"role": "user", "content": "hi"}], on_text=chunks.append)
+
+    assert "".join(chunks) == "Hello"
+    assert result["content"] == "Hello"
+    assert result["role"] == "assistant"
+
+
+def test_anthropic_keeps_extended_thinking_out_of_the_answer(canned):
+    canned(_sse(
+        {"type": "content_block_delta", "index": 0,
+         "delta": {"type": "thinking_delta", "thinking": "reasoning..."}},
+        {"type": "content_block_delta", "index": 0,
+         "delta": {"type": "text_delta", "text": "42"}},
+        {"type": "message_stop"},
+    ))
+    provider = AnthropicProvider({"base_url": "https://x.invalid",
+                                 "api_key": "k", "model": "m"})
+    reasoning: List[str] = []
+
+    result = provider.chat([{"role": "user", "content": "q"}], on_reasoning=reasoning.append)
+
+    assert reasoning == ["reasoning..."]
+    assert result["content"] == "42"
+
+
+def test_anthropic_returns_tool_calls_with_parsed_arguments(canned):
+    canned(_sse(
+        {"type": "content_block_start", "index": 0,
+         "content_block": {"type": "tool_use", "id": "toolu_1", "name": "save_file"}},
+        {"type": "content_block_delta", "index": 0,
+         "delta": {"type": "input_json_delta", "partial_json": '{"filename":"a.md"}'}},
+        {"type": "message_stop"},
+    ))
+    provider = AnthropicProvider({"base_url": "https://x.invalid",
+                                 "api_key": "k", "model": "m"})
+
+    result = provider.chat([{"role": "user", "content": "save"}])
+
+    assert result["tool_calls"] == [
+        {"id": "toolu_1", "name": "save_file", "arguments": {"filename": "a.md"}}
+    ]
+
+
+@pytest.mark.parametrize("factory", [
+    lambda: OpenAICompatProvider({"base_url": "https://x.invalid/v1", "api_key": "k", "model": "m"}),
+    lambda: AnthropicProvider({"base_url": "https://x.invalid", "api_key": "k", "model": "m"}),
+], ids=["openai_compat", "anthropic"])
+def test_transport_failure_surfaces_as_provider_error(canned, factory):
+    """Every failure mode must arrive as ProviderError so the agent loop needs
+    exactly one except clause, whichever provider is active."""
+    canned([], status_code=500, text="boom")
+
+    with pytest.raises(ProviderError):
+        factory().chat([{"role": "user", "content": "hi"}])
+
+
+def test_fake_provider_satisfies_the_same_streaming_contract():
+    """The double is only useful as a stand-in if it keeps the same promises the
+    real adapters are held to above."""
+    provider = FakeProvider([ScriptedTurn(text="Hello", reasoning="hmm")])
+    text: List[str] = []
+    reasoning: List[str] = []
+
+    result = provider.chat([{"role": "user", "content": "hi"}],
+                           on_text=text.append, on_reasoning=reasoning.append)
+
+    assert "".join(text) == result["content"] == "Hello"
+    assert reasoning == ["hmm"]
+    assert result["role"] == "assistant"
+    assert result["tool_calls"] == []
+
+
+def test_fake_provider_raises_provider_error_like_the_real_ones():
+    provider = FakeProvider([ScriptedTurn(error="gateway exploded")])
+
+    with pytest.raises(ProviderError):
+        provider.chat([{"role": "user", "content": "hi"}])
+
+
+# --------------------------------------------------------------------------- #
+# Registry <-> implementation agreement
+# --------------------------------------------------------------------------- #
+@pytest.mark.parametrize("descriptor", BUILT_IN_PROVIDERS, ids=lambda d: d.id)
+def test_every_descriptor_builds_a_working_provider(descriptor):
+    """A descriptor that cannot be built is a catalogue lying to the UI: Settings
+    would list the provider and selecting it would fail at the first message."""
+    registry = ProviderRegistry()
+    conf = {"base_url": "https://x.invalid/v1", "api_key": "k"}
+
+    provider = registry.build(descriptor.id, conf)
+
+    assert isinstance(provider, Provider)
+    # The id, not the shared adapter class name: three descriptors map onto
+    # OpenAICompatProvider, and usage/audit records must still tell them apart.
+    assert provider.name == descriptor.id
+    assert provider.model == descriptor.default_model
+
+
+@pytest.mark.parametrize("descriptor", BUILT_IN_PROVIDERS, ids=lambda d: d.id)
+def test_declared_vision_capability_matches_the_implementation(descriptor):
+    """``supports_vision`` decides whether an image block may be sent. A
+    descriptor claiming vision for an adapter that cannot translate the block
+    would route image turns into a guaranteed failure."""
+    provider = ProviderRegistry().build(descriptor.id, {"base_url": "u", "api_key": "k"})
+
+    if descriptor.supports(ProviderCapability.VISION):
+        assert provider.supports_vision is True
+
+
+def test_registry_build_never_mutates_the_caller_config():
+    """The routing layer runs one turn on a different model; if build() wrote
+    that model back into the config dict it was handed, the override would
+    silently become the user's saved default."""
+    registry = ProviderRegistry()
+    conf = {"base_url": "u", "api_key": "k", "model": "configured-model"}
+
+    provider = registry.build("openai_compat", conf, model="routed-model")
+
+    assert provider.model == "routed-model"
+    assert conf["model"] == "configured-model"
+
+
+def test_registry_rejects_an_unknown_provider_with_provider_error():
+    with pytest.raises(ProviderError) as excinfo:
+        ProviderRegistry().build("does_not_exist", {})
+
+    # The message lists what IS known, so a typo in config is fixable from the
+    # error alone without opening the source.
+    assert "openai_compat" in str(excinfo.value)
diff --git a/tests/unit/test_routing_application_service.py b/tests/unit/test_routing_application_service.py
new file mode 100644
index 0000000..22c329a
--- /dev/null
+++ b/tests/unit/test_routing_application_service.py
@@ -0,0 +1,346 @@
+"""Unit tests for :mod:`application.model_routing` (R03-T03).
+
+These run against a hand-written fake router rather than ``core.routing``: the
+point of the service is the DECISION policy around the engine (mode handling,
+the manual confirm, never-raise behaviour, failure fallback), and mixing in the
+real scorer would test the wrong thing and drag the suite over its time budget.
+
+No Qt, no config, no network - the whole file runs in milliseconds, which is the
+concrete payoff of moving this logic out of ``ui/chat_panel.py``.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, List, Optional, Tuple
+
+import pytest
+
+from cowork_local.application.model_routing import (
+    RoutingApplicationService,
+    RoutingDecision,
+    RoutingMode,
+)
+
+
+# --------------------------------------------------------------------------- #
+# Test doubles shaped like core.routing's RouteResult / SwitchDecision
+# --------------------------------------------------------------------------- #
+@dataclass
+class _TaskType:
+    value: str
+
+
+@dataclass
+class _Decision:
+    score_gain: float = 0.0
+    reason: str = ""
+
+
+@dataclass
+class _RouteResult:
+    should_switch: bool
+    to: Optional[Tuple[str, str]] = None
+    task_type: Any = None
+    decision: Any = None
+
+    def target(self) -> Optional[Tuple[str, str]]:
+        return self.to
+
+
+class _FakeRouter:
+    """Records every route() call and replays a canned result."""
+
+    def __init__(self, result: Any = None, raises: bool = False) -> None:
+        self._result = result or _RouteResult(should_switch=False, decision=_Decision())
+        self._raises = raises
+        self.calls: List[dict] = []
+
+    def route(self, surface, prompt, current_provider, current_model, **kwargs):
+        self.calls.append({"surface": surface, "prompt": prompt,
+                           "provider": current_provider, "model": current_model, **kwargs})
+        if self._raises:
+            raise RuntimeError("assessment store is corrupt")
+        return self._result
+
+
+def _switch_to(provider: str, model: str, gain: float = 0.2, task: str = "coding") -> _RouteResult:
+    return _RouteResult(
+        should_switch=True, to=(provider, model), task_type=_TaskType(task),
+        decision=_Decision(score_gain=gain, reason=f"{task} fit beats current by {gain}"),
+    )
+
+
+# --------------------------------------------------------------------------- #
+# Mode parsing
+# --------------------------------------------------------------------------- #
+@pytest.mark.parametrize("raw,expected", [
+    ("off", RoutingMode.OFF),
+    ("AUTO", RoutingMode.AUTO),
+    (" manual ", RoutingMode.MANUAL),
+    ("fallback", RoutingMode.FALLBACK),
+])
+def test_parse_accepts_the_config_spellings(raw, expected):
+    assert RoutingMode.parse(raw) is expected
+
+
+@pytest.mark.parametrize("raw", ["", None, "nonsense", 0])
+def test_parse_degrades_unknown_values_to_off(raw):
+    """A corrupt setting must leave the user's own model alone rather than
+    silently moving their work onto another model."""
+    assert RoutingMode.parse(raw) is RoutingMode.OFF
+
+
+# --------------------------------------------------------------------------- #
+# OFF
+# --------------------------------------------------------------------------- #
+def test_off_never_consults_the_engine():
+    router = _FakeRouter(_switch_to("anthropic", "claude"))
+    service = RoutingApplicationService(router)
+
+    decision = service.route_turn("cowork", "hi", "openai_compat", "gpt-4o-mini",
+                                  mode="off")
+
+    assert router.calls == []          # not even scored: OFF costs nothing
+    assert decision.switched is False
+    assert decision.target() == ("openai_compat", "gpt-4o-mini")
+
+
+def test_blank_prompt_is_never_routed():
+    """An empty message carries no signal to classify; all three legacy copies
+    guarded this and the guard has to survive the move."""
+    router = _FakeRouter(_switch_to("anthropic", "claude"))
+    service = RoutingApplicationService(router)
+
+    decision = service.route_turn("cowork", "   ", "openai_compat", "m", mode="auto")
+
+    assert router.calls == []
+    assert decision.switched is False
+
+
+# --------------------------------------------------------------------------- #
+# AUTO
+# --------------------------------------------------------------------------- #
+def test_auto_switches_silently_and_reports_the_target():
+    router = _FakeRouter(_switch_to("anthropic", "claude-sonnet-4-6", gain=0.31))
+    service = RoutingApplicationService(router)
+
+    decision = service.route_turn("cowork", "write a function", "openai_compat", "gpt-4o-mini",
+                                  mode="auto")
+
+    assert decision.switched is True
+    assert decision.target() == ("anthropic", "claude-sonnet-4-6")
+    assert decision.task_type == "coding"
+    assert decision.score_gain == pytest.approx(0.31)
+    assert decision.should_notify is True
+
+
+def test_auto_keeps_the_current_model_when_no_candidate_wins():
+    router = _FakeRouter(_RouteResult(should_switch=False, decision=_Decision(reason="no gain")))
+    service = RoutingApplicationService(router)
+
+    decision = service.route_turn("cowork", "hello", "openai_compat", "gpt-4o-mini", mode="auto")
+
+    assert decision.switched is False
+    # The decision still names a model to run on, so the call site never has to
+    # re-derive the fallback itself - the exact drift the three copies suffered.
+    assert decision.target() == ("openai_compat", "gpt-4o-mini")
+    assert decision.should_notify is False
+
+
+def test_auto_never_asks_for_confirmation():
+    router = _FakeRouter(_switch_to("anthropic", "claude"))
+    asked: List[RoutingDecision] = []
+    service = RoutingApplicationService(router)
+
+    service.route_turn("cowork", "q", "openai_compat", "m", mode="auto",
+                       confirm=lambda d: asked.append(d) or True)
+
+    assert asked == []
+
+
+# --------------------------------------------------------------------------- #
+# MANUAL
+# --------------------------------------------------------------------------- #
+def test_manual_switches_only_after_the_user_approves():
+    router = _FakeRouter(_switch_to("anthropic", "claude"))
+    service = RoutingApplicationService(router)
+    seen: List[RoutingDecision] = []
+
+    def confirm(proposal: RoutingDecision) -> bool:
+        seen.append(proposal)
+        return True
+
+    decision = service.route_turn("cowork", "q", "openai_compat", "m",
+                                  mode="manual", confirm=confirm)
+
+    assert decision.switched is True
+    assert decision.target() == ("anthropic", "claude")
+    # The dialog is handed the full proposal so it can explain the trade-off.
+    assert seen[0].model == "claude"
+    assert seen[0].score_gain > 0
+
+
+def test_manual_keeps_the_current_model_when_declined():
+    router = _FakeRouter(_switch_to("anthropic", "claude"))
+    service = RoutingApplicationService(router)
+
+    decision = service.route_turn("cowork", "q", "openai_compat", "gpt-4o-mini",
+                                  mode="manual", confirm=lambda d: False)
+
+    assert decision.switched is False
+    assert decision.declined is True
+    assert decision.target() == ("openai_compat", "gpt-4o-mini")
+
+
+def test_manual_without_a_confirm_callback_does_not_switch():
+    """A headless caller (scheduler) has nobody to ask, so Manual must behave as
+    "not approved" rather than as "approved by default"."""
+    router = _FakeRouter(_switch_to("anthropic", "claude"))
+    service = RoutingApplicationService(router)
+
+    decision = service.route_turn("cowork", "q", "openai_compat", "m", mode="manual")
+
+    assert decision.switched is False
+    assert decision.declined is True
+
+
+def test_a_confirm_dialog_that_raises_counts_as_declined():
+    """If the modal blows up (window closing mid-turn) the safe reading is that
+    the user did NOT consent to running on another model."""
+    router = _FakeRouter(_switch_to("anthropic", "claude"))
+    service = RoutingApplicationService(router)
+
+    def confirm(_proposal):
+        raise RuntimeError("dialog destroyed")
+
+    decision = service.route_turn("cowork", "q", "openai_compat", "m",
+                                  mode="manual", confirm=confirm)
+
+    assert decision.switched is False
+
+
+# --------------------------------------------------------------------------- #
+# FALLBACK
+# --------------------------------------------------------------------------- #
+def test_fallback_does_not_switch_up_front():
+    """The whole point of the mode: honour the user's model choice until it
+    actually fails."""
+    router = _FakeRouter(_switch_to("anthropic", "claude"))
+    service = RoutingApplicationService(router)
+
+    decision = service.route_turn("cowork", "q", "openai_compat", "gpt-4o-mini",
+                                  mode="fallback")
+
+    assert router.calls == []
+    assert decision.switched is False
+    assert decision.target() == ("openai_compat", "gpt-4o-mini")
+
+
+def test_fallback_switches_after_a_failure():
+    router = _FakeRouter(_switch_to("anthropic", "claude", gain=0.4))
+    service = RoutingApplicationService(router)
+
+    decision = service.fallback_after_failure("cowork", "q", "openai_compat", "gpt-4o-mini",
+                                              mode="fallback")
+
+    assert decision is not None
+    assert decision.switched is True
+    assert decision.target() == ("anthropic", "claude")
+    assert "failed" in decision.reason
+
+
+def test_fallback_never_returns_the_model_that_just_failed():
+    """Retrying the model that just went down would spin on the outage."""
+    router = _FakeRouter(_switch_to("openai_compat", "gpt-4o-mini"))
+    service = RoutingApplicationService(router)
+
+    assert service.fallback_after_failure(
+        "cowork", "q", "openai_compat", "gpt-4o-mini", mode="fallback") is None
+
+
+def test_fallback_returns_none_when_there_is_no_alternative():
+    router = _FakeRouter(_RouteResult(should_switch=False, decision=_Decision()))
+    service = RoutingApplicationService(router)
+
+    assert service.fallback_after_failure("cowork", "q", "openai_compat", "m",
+                                          mode="auto") is None
+
+
+@pytest.mark.parametrize("mode", ["off", "manual"])
+def test_off_and_manual_do_not_auto_recover_from_a_failure(mode):
+    """Both modes exist to keep the user in control of which model runs their
+    work; moving it on failure would break that promise silently."""
+    router = _FakeRouter(_switch_to("anthropic", "claude"))
+    service = RoutingApplicationService(router)
+
+    assert service.fallback_after_failure("cowork", "q", "openai_compat", "m",
+                                          mode=mode) is None
+
+
+# --------------------------------------------------------------------------- #
+# Robustness - routing must never break a chat turn
+# --------------------------------------------------------------------------- #
+def test_engine_failure_degrades_to_keeping_the_current_model():
+    service = RoutingApplicationService(_FakeRouter(raises=True))
+
+    decision = service.route_turn("cowork", "q", "openai_compat", "gpt-4o-mini", mode="auto")
+
+    assert decision.switched is False
+    assert decision.target() == ("openai_compat", "gpt-4o-mini")
+
+
+def test_engine_failure_during_fallback_returns_none():
+    """A broken router must not mask the original provider error with its own."""
+    service = RoutingApplicationService(_FakeRouter(raises=True))
+
+    assert service.fallback_after_failure("cowork", "q", "p", "m", mode="auto") is None
+
+
+def test_a_malformed_route_result_is_treated_as_no_switch():
+    """The engine is a legacy module still under refactor; a missing attribute
+    must degrade, not raise into the middle of a turn."""
+    class _Garbage:
+        should_switch = True   # claims a switch but exposes no target()
+
+    service = RoutingApplicationService(_FakeRouter(_Garbage()))
+
+    decision = service.route_turn("cowork", "q", "openai_compat", "m", mode="auto")
+
+    assert decision.switched is False
+    assert decision.target() == ("openai_compat", "m")
+
+
+# --------------------------------------------------------------------------- #
+# Per-surface mode lookup
+# --------------------------------------------------------------------------- #
+def test_mode_is_read_per_surface_when_not_passed_explicitly():
+    """Each screen has its own Off/Auto/Manual toggle, and workspaces override
+    it - so the surface, not a global setting, decides."""
+    router = _FakeRouter(_switch_to("anthropic", "claude"))
+    modes = {"cowork": "auto", "ai_edit": "off"}
+    service = RoutingApplicationService(router, mode_reader=modes.get)
+
+    assert service.route_turn("cowork", "q", "p", "m").switched is True
+    assert service.route_turn("ai_edit", "q", "p", "m").switched is False
+
+
+def test_a_failing_mode_reader_falls_back_to_off():
+    def broken(_surface):
+        raise KeyError("config not loaded yet")
+
+    service = RoutingApplicationService(_FakeRouter(_switch_to("a", "b")),
+                                        mode_reader=broken)
+
+    assert service.route_turn("cowork", "q", "p", "m").switched is False
+
+
+def test_required_capabilities_are_passed_through_to_the_engine():
+    """An image turn must only be routed to a vision-capable model; the filter
+    has to reach the scorer or the constraint is silently dropped."""
+    router = _FakeRouter()
+    service = RoutingApplicationService(router)
+
+    service.route_turn("cowork", "describe this", "p", "m", mode="auto",
+                       required_capabilities=["vision"])
+
+    assert router.calls[0]["required_capabilities"] == ["vision"]
diff --git a/tests/unit/test_routing_wiring.py b/tests/unit/test_routing_wiring.py
new file mode 100644
index 0000000..e653f11
--- /dev/null
+++ b/tests/unit/test_routing_wiring.py
@@ -0,0 +1,112 @@
+"""Integration test for the AppContext routing wiring (R03-T04 / R03-T05).
+
+The three chat surfaces now call ``ctx.routing_application()`` instead of each
+carrying their own copy of the routing algorithm. The unit tests cover the
+policy; this file covers the WIRING, which unit tests with a fake router cannot
+see:
+
+* the service is built and memoised on the context
+* it reads the per-workspace mode through ``project_routing_mode``
+* the legacy ``core.routing.RoutingService`` is what sits underneath it
+* ``fallback`` survives a round trip through the per-workspace mode store
+
+Still Qt-free: ``AppContext`` itself imports no widgets, and the config is
+written into a tmp dir so nothing touches ``~/.cowork_local``.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from cowork_local.application.model_routing import (
+    RoutingApplicationService,
+    RoutingMode,
+)
+from cowork_local.config import AppConfig
+from cowork_local.state import AppContext
+
+
+@pytest.fixture
+def ctx(tmp_path: Path) -> AppContext:
+    """An AppContext backed by a throwaway config file."""
+    return AppContext(AppConfig.load(tmp_path / "config.json"))
+
+
+def test_routing_application_is_built_and_memoised(ctx):
+    """One instance per app: the pending-switch registry underneath it must be
+    shared by every surface, so a second call has to return the same object."""
+    first = ctx.routing_application()
+
+    assert isinstance(first, RoutingApplicationService)
+    assert ctx.routing_application() is first
+
+
+def test_the_legacy_engine_sits_underneath_the_new_service():
+    """Strangler-fig check (ADR-001 section 4): the scoring engine is reused, not
+    reimplemented. If this ever stops holding, the assessment scores the
+    scheduler probes would no longer be the ones routing decisions use."""
+    from cowork_local.core.routing.service import RoutingService
+
+    config = AppConfig.load(Path("does-not-exist.json"))
+    context = AppContext(config)
+
+    service = context.routing_application()
+
+    assert isinstance(service._router, RoutingService)
+    assert service._router is context.routing()
+
+
+def test_mode_is_read_through_the_per_workspace_lookup(ctx, monkeypatch):
+    seen = []
+
+    def fake_mode(surface: str) -> str:
+        seen.append(surface)
+        return "off"
+
+    monkeypatch.setattr(ctx, "project_routing_mode", fake_mode)
+    # Built after the patch so the service captures the patched reader.
+    service = RoutingApplicationService(ctx.routing(), mode_reader=ctx.project_routing_mode)
+
+    decision = service.route_turn("co4e", "hello", "openai_compat", "gpt-4o-mini")
+
+    assert seen == ["co4e"]
+    assert decision.switched is False
+
+
+def test_routing_off_by_default_leaves_the_selected_model_alone(ctx):
+    """Default config has routing off on every surface, so a fresh install must
+    never move a turn to another model."""
+    decision = ctx.routing_application().route_turn(
+        "cowork", "write a function", "openai_compat", "gpt-4o-mini")
+
+    assert decision.mode is RoutingMode.OFF
+    assert decision.switched is False
+    assert decision.target() == ("openai_compat", "gpt-4o-mini")
+
+
+@pytest.mark.parametrize("mode", ["off", "auto", "manual", "fallback"])
+def test_every_mode_survives_a_round_trip_through_the_config(ctx, mode):
+    """``fallback`` is new (R03-T03); the per-surface store used to whitelist
+    only three values and would have silently downgraded it to "off"."""
+    ctx.set_project_routing_mode("cowork", mode)
+
+    assert ctx.project_routing_mode("cowork") == mode
+
+
+def test_an_unknown_mode_still_falls_back_to_off(ctx):
+    ctx.set_project_routing_mode("cowork", "turbo")
+
+    assert ctx.project_routing_mode("cowork") == "off"
+
+
+def test_a_real_route_call_never_raises_without_any_assessments(ctx):
+    """The store is empty on a fresh install. Routing must degrade to "keep the
+    current model" rather than raise into the middle of the first message."""
+    ctx.set_project_routing_mode("cowork", "auto")
+
+    decision = ctx.routing_application().route_turn(
+        "cowork", "hello there", "openai_compat", "gpt-4o-mini")
+
+    assert decision.switched is False
+    assert decision.target() == ("openai_compat", "gpt-4o-mini")
diff --git a/tests/unit/test_usage_sink.py b/tests/unit/test_usage_sink.py
new file mode 100644
index 0000000..dc200ff
--- /dev/null
+++ b/tests/unit/test_usage_sink.py
@@ -0,0 +1,249 @@
+"""Unit tests for :mod:`infrastructure.telemetry.usage_sink` (R03-T06).
+
+Two things are being protected here:
+
+1. The **numbers do not change**. Extracting usage recording out of the two
+   providers is only safe if the events built from each wire format carry
+   exactly what ``core.usage_tracker.record`` used to receive - a silent change
+   would corrupt the Dashboard's cost history.
+2. The **sink can never break a turn**. Telemetry is observability; a broken
+   store must be swallowed (and logged), never raised into a chat turn.
+"""
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List
+
+import pytest
+
+from cowork_local.infrastructure.telemetry import usage_sink as telemetry
+from cowork_local.providers.anthropic import AnthropicProvider
+from cowork_local.providers.base import Provider
+from cowork_local.providers.openai_compat import OpenAICompatProvider
+
+
+# --------------------------------------------------------------------------- #
+# Event construction - one per wire format
+# --------------------------------------------------------------------------- #
+def test_openai_usage_block_maps_onto_the_canonical_event():
+    event = telemetry.openai_usage_event("openai_compat", "gpt-4o-mini", {
+        "prompt_tokens": 120,
+        "completion_tokens": 45,
+        "prompt_tokens_details": {"cached_tokens": 100},
+    })
+
+    assert (event.input_tokens, event.output_tokens, event.cached_tokens) == (120, 45, 100)
+    assert event.estimated is False
+    # Cached tokens are a SUBSET of input, so adding them would double-count.
+    assert event.total_tokens == 165
+
+
+def test_anthropic_usage_accumulator_maps_onto_the_canonical_event():
+    """Anthropic reports input on message_start and output on message_delta, so
+    providers/anthropic.py accumulates them into in/out/cache keys."""
+    event = telemetry.anthropic_usage_event("anthropic", "claude", {
+        "in": 200, "out": 80, "cache": 150,
+    })
+
+    assert (event.input_tokens, event.output_tokens, event.cached_tokens) == (200, 80, 150)
+    assert event.estimated is False
+
+
+def test_a_missing_usage_block_produces_an_estimated_event():
+    event = telemetry.estimated_event("ollama", "llama3.1", "x" * 400, "y" * 40)
+
+    assert event.estimated is True
+    assert event.input_tokens == 100      # ~4 characters per token
+    assert event.output_tokens == 10
+    assert event.cached_tokens == 0
+
+
+def test_estimation_matches_the_legacy_tracker_formula():
+    """The extraction must not shift a single recorded number, so the estimator
+    is pinned against the one it replaced."""
+    from cowork_local.core import usage_tracker
+
+    for text in ("", "short", "x" * 4001, "unicode - tiếng Việt"):
+        assert telemetry.estimate_tokens(text) == usage_tracker.estimate_tokens(text)
+
+
+# --------------------------------------------------------------------------- #
+# Sinks
+# --------------------------------------------------------------------------- #
+def test_recording_sink_collects_events_for_assertions():
+    sink = telemetry.RecordingUsageSink()
+
+    sink.record(telemetry.UsageEvent("p", "m", input_tokens=10, output_tokens=5))
+    sink.record(telemetry.UsageEvent("p", "m", input_tokens=1, output_tokens=1))
+
+    assert len(sink.events) == 2
+    assert sink.total_tokens == 17
+
+
+def test_null_sink_discards_without_error():
+    telemetry.NullUsageSink().record(telemetry.UsageEvent("p", "m"))
+
+
+def test_tracker_sink_forwards_every_field_positionally():
+    """``core.usage_tracker.record`` takes positional counts plus an ``estimated``
+    keyword; the adapter has to preserve that exact call shape."""
+    seen: Dict[str, Any] = {}
+
+    class _Tracker:
+        @staticmethod
+        def record(provider, model, input_tokens, output_tokens, cached_tokens,
+                   estimated=False):
+            # Fields captured explicitly rather than via locals(), which would
+            # also drag in the closed-over `seen` binding itself.
+            seen.update({"provider": provider, "model": model,
+                         "input_tokens": input_tokens, "output_tokens": output_tokens,
+                         "cached_tokens": cached_tokens, "estimated": estimated})
+
+    telemetry.UsageTrackerSink(tracker=_Tracker()).record(
+        telemetry.UsageEvent("anthropic", "claude", 7, 3, 2, estimated=True))
+
+    assert seen == {"provider": "anthropic", "model": "claude", "input_tokens": 7,
+                    "output_tokens": 3, "cached_tokens": 2, "estimated": True}
+
+
+def test_a_failing_tracker_never_raises_into_the_turn():
+    class _Broken:
+        @staticmethod
+        def record(*_args, **_kwargs):
+            raise OSError("usage store is read-only")
+
+    # Must not raise - the turn that produced this event has already succeeded.
+    telemetry.UsageTrackerSink(tracker=_Broken()).record(telemetry.UsageEvent("p", "m"))
+
+
+def test_set_default_sink_returns_the_previous_one_for_restoration():
+    replacement = telemetry.RecordingUsageSink()
+
+    previous = telemetry.set_default_sink(replacement)
+    try:
+        assert telemetry.default_sink is replacement
+    finally:
+        telemetry.set_default_sink(previous)
+    assert telemetry.default_sink is previous
+
+
+# --------------------------------------------------------------------------- #
+# Provider integration - the seam actually being used
+# --------------------------------------------------------------------------- #
+class _StubResponse:
+    """The few members the provider streaming loop touches."""
+
+    def __init__(self, lines: List[str]) -> None:
+        self._lines = lines
+        self.status_code = 200
+        self.headers: Dict[str, str] = {}
+        self.encoding = "utf-8"
+        self.text = ""
+
+    def iter_lines(self, decode_unicode: bool = False):
+        yield from self._lines
+
+    def close(self) -> None:
+        return None
+
+
+@pytest.fixture
+def sink(monkeypatch):
+    """A per-instance recording sink, so nothing touches the real usage store."""
+    return telemetry.RecordingUsageSink()
+
+
+@pytest.fixture
+def canned(monkeypatch):
+    def _install(lines: List[str]):
+        monkeypatch.setattr(Provider, "_request",
+                            lambda self, method, url, **kw: _StubResponse(lines))
+    return _install
+
+
+def test_openai_provider_reports_server_counts_to_its_sink(canned, sink):
+    canned([
+        'data: ' + json.dumps({"choices": [{"delta": {"content": "hi"}}],
+                               "usage": {"prompt_tokens": 11, "completion_tokens": 2,
+                                         "prompt_tokens_details": {"cached_tokens": 4}}}),
+        "data: [DONE]",
+    ])
+    provider = OpenAICompatProvider({"base_url": "https://x.invalid/v1",
+                                     "api_key": "k", "model": "gpt-4o-mini"})
+    provider.usage_sink = sink
+
+    provider.chat([{"role": "user", "content": "hi"}])
+
+    assert len(sink.events) == 1
+    event = sink.events[0]
+    assert (event.input_tokens, event.output_tokens, event.cached_tokens) == (11, 2, 4)
+    assert event.estimated is False
+    assert event.model == "gpt-4o-mini"
+
+
+def test_openai_provider_estimates_when_the_gateway_sends_no_usage(canned, sink):
+    canned(['data: ' + json.dumps({"choices": [{"delta": {"content": "hello"}}]}),
+            "data: [DONE]"])
+    provider = OpenAICompatProvider({"base_url": "https://x.invalid/v1",
+                                     "api_key": "k", "model": "m"})
+    provider.usage_sink = sink
+
+    provider.chat([{"role": "user", "content": "hi"}])
+
+    assert sink.events[0].estimated is True
+    assert sink.events[0].output_tokens >= 1
+
+
+def test_anthropic_provider_reports_stream_counts_to_its_sink(canned, sink):
+    canned(['data: ' + json.dumps(p) for p in (
+        {"type": "message_start", "message": {"usage": {"input_tokens": 30,
+                                                        "cache_read_input_tokens": 10}}},
+        {"type": "content_block_delta", "index": 0,
+         "delta": {"type": "text_delta", "text": "ok"}},
+        {"type": "message_delta", "usage": {"output_tokens": 5}},
+        {"type": "message_stop"},
+    )])
+    provider = AnthropicProvider({"base_url": "https://x.invalid", "api_key": "k",
+                                  "model": "claude"})
+    provider.usage_sink = sink
+
+    provider.chat([{"role": "user", "content": "hi"}])
+
+    event = sink.events[0]
+    assert (event.input_tokens, event.output_tokens, event.cached_tokens) == (30, 5, 10)
+    assert event.estimated is False
+
+
+def test_a_provider_without_an_explicit_sink_uses_the_process_default(canned):
+    """Existing call sites set no sink, so the default has to keep working -
+    that is what makes this extraction a no-op for production behaviour."""
+    canned(['data: ' + json.dumps({"choices": [{"delta": {"content": "x"}}]}),
+            "data: [DONE]"])
+    recorder = telemetry.RecordingUsageSink()
+    previous = telemetry.set_default_sink(recorder)
+    try:
+        OpenAICompatProvider({"base_url": "https://x.invalid/v1", "api_key": "k",
+                              "model": "m"}).chat([{"role": "user", "content": "hi"}])
+    finally:
+        telemetry.set_default_sink(previous)
+
+    assert len(recorder.events) == 1
+
+
+def test_a_sink_that_raises_does_not_fail_the_turn(canned):
+    """The answer has already been produced by the time usage is recorded;
+    losing the telemetry is strictly better than losing the answer."""
+    canned(['data: ' + json.dumps({"choices": [{"delta": {"content": "x"}}]}),
+            "data: [DONE]"])
+
+    class _Exploding:
+        def record(self, _event):
+            raise RuntimeError("sink is down")
+
+    provider = OpenAICompatProvider({"base_url": "https://x.invalid/v1",
+                                     "api_key": "k", "model": "m"})
+    provider.usage_sink = _Exploding()
+
+    result = provider.chat([{"role": "user", "content": "hi"}])
+
+    assert result["content"] == "x"
diff --git a/ui/chat_panel.py b/ui/chat_panel.py
index 9457d13..c242800 100644
--- a/ui/chat_panel.py
+++ b/ui/chat_panel.py
@@ -638,12 +638,16 @@ class ChatPanel(QWidget):
     def _apply_routing(self, text: str, turn: Dict[str, Any]) -> None:
         """Auto Model Routing hook — run once per outgoing message.
 
-        Off → no-op. Auto → silently switch to the best-fit model. Manual → ask
-        the user (modal, with the configured confirm timeout) before switching.
+        The decision itself lives in ``application/model_routing`` (R03-T04):
+        this method is now only the presentation half — supply the current
+        model, open the confirm dialog when the service asks for one, and render
+        the notice. Off/Auto/Manual/Fallback semantics, the never-raise
+        guarantee and the "which model do we end up on" fallback are the
+        service's job, and are shared with Co4E and AI-Edit instead of being
+        re-implemented here.
+
         Sets ``self._routed_provider``/``self._routed_model`` for THIS turn;
-        :meth:`build_provider` honours them. Never raises — a routing failure
-        must never block sending a message; it just falls back to the tab's
-        own model.
+        :meth:`build_provider` honours them.
         """
         # Recompute fresh each message; clear any previous turn's override.
         self._routed_provider = None
@@ -651,37 +655,30 @@ class ChatPanel(QWidget):
         # An explicitly-pinned Admin agent takes precedence over routing.
         if getattr(self, "_admin_agent", None) is not None:
             return
-        if not (text or "").strip():
+        cur_provider = self.ctx.config.active_provider
+        cur_model = self._model or self.ctx.config.provider_conf(cur_provider).get("model", "")
+        decision = self.ctx.routing_application().route_turn(
+            self.kind, text, cur_provider, cur_model, confirm=self._confirm_routing_switch,
+        )
+        if not decision.switched:
             return
-        try:
-            mode = self.ctx.project_routing_mode(self.kind)  # per-workspace mode
-            if mode == "off":
-                return
-            service = self.ctx.routing()
-            cur_provider = self.ctx.config.active_provider
-            cur_model = self._model or self.ctx.config.provider_conf(cur_provider).get("model", "")
-            result = service.route(self.kind, text, cur_provider, cur_model, mode_override=mode)
-            if not result.should_switch:
-                return
-            target = result.target()
-            if target is None:
-                return
-            to_provider, to_model = target
-            if mode == "manual":
-                from .routing_toggle import confirm_switch
-                timeout = float(self.ctx.config.routing.get("confirm_timeout_sec", 60) or 60)
-                if not confirm_switch(self, result.decision, timeout):
-                    return  # declined / timed out → keep current model
-            self._routed_provider = to_provider
-            self._routed_model = to_model
-            notice = self.chat_view.add_status(tr(
-                "routing.switched_notice",
-                model=to_model, task=result.task_type.value,
-                gain=f"{result.decision.score_gain:.2f}"))
-            turn["bubbles"].append(notice)
-        except Exception:  # noqa: BLE001 — routing must never block a chat turn
-            self._routed_provider = None
-            self._routed_model = None
+        self._routed_provider, self._routed_model = decision.target()
+        notice = self.chat_view.add_status(tr(
+            "routing.switched_notice",
+            model=decision.model, task=decision.task_type,
+            gain=f"{decision.score_gain:.2f}"))
+        turn["bubbles"].append(notice)
+
+    def _confirm_routing_switch(self, decision) -> bool:
+        """Manual mode: ask the user before moving this turn to another model.
+
+        Passed to the routing service as a callback so the pure-Python decision
+        layer never has to know a modal dialog exists. Returning False (declined
+        or timed out) keeps the tab's own model."""
+        from .routing_toggle import confirm_switch
+
+        timeout = float(self.ctx.config.routing.get("confirm_timeout_sec", 60) or 60)
+        return bool(confirm_switch(self, decision, timeout))
 
     def _compress_messages(self) -> None:
         """Manual compress: keep the system prompt + the last 2 turns verbatim and
diff --git a/ui/co4e_tab.py b/ui/co4e_tab.py
index b829b89..8c397bf 100644
--- a/ui/co4e_tab.py
+++ b/ui/co4e_tab.py
@@ -1847,41 +1847,41 @@ class Co4ETab(QWidget):
         self._run_chat_turn(system_parts, request, model)
 
     def _apply_co4e_routing(self, request: str) -> str:
-        """Route this Co4E turn to the best-fit model. Returns the model id to
-        use ('' → provider default) and sets ``self._co4e_routed_provider`` when
-        a cross-provider switch is chosen. Off → no-op. Manual → confirm first.
-        Never raises — falls back to the default model on any error."""
+        """Route this Co4E turn to the best-fit model.
+
+        Returns the model id to use ('' -> provider default) and sets
+        ``self._co4e_routed_provider`` when a cross-provider switch is chosen.
+
+        The decision comes from the shared ``RoutingApplicationService``
+        (R03-T05) - Off/Auto/Manual/Fallback handling, the confirm handshake and
+        the never-raise guarantee are no longer duplicated here. What stays is
+        only the Co4E-specific presentation: the surface key, and where the
+        notice is rendered.
+        """
         self._co4e_routed_provider = None
-        if not (request or "").strip():
-            return ""
-        try:
-            mode = self.ctx.project_routing_mode("co4e")  # per-workspace mode
-            if mode == "off":
-                return ""
-            service = self.ctx.routing()
-            cur_provider = self.ctx.config.active_provider
-            cur_model = self.ctx.config.provider_conf(cur_provider).get("model", "")
-            result = service.route("co4e", request, cur_provider, cur_model, mode_override=mode)
-            if not result.should_switch:
-                return ""
-            target = result.target()
-            if target is None:
-                return ""
-            to_provider, to_model = target
-            if mode == "manual":
-                from .routing_toggle import confirm_switch
-                timeout = float(self.ctx.config.routing.get("confirm_timeout_sec", 60) or 60)
-                if not confirm_switch(self, result.decision, timeout):
-                    return ""
-            self._co4e_routed_provider = to_provider
-            self._append_chat("system", tr(
-                "routing.switched_notice",
-                model=to_model, task=result.task_type.value,
-                gain=f"{result.decision.score_gain:.2f}"))
-            return to_model
-        except Exception:  # noqa: BLE001 — routing must never block a Co4E turn
-            self._co4e_routed_provider = None
+        cur_provider = self.ctx.config.active_provider
+        cur_model = self.ctx.config.provider_conf(cur_provider).get("model", "")
+        decision = self.ctx.routing_application().route_turn(
+            "co4e", request, cur_provider, cur_model, confirm=self._confirm_routing_switch,
+        )
+        if not decision.switched:
             return ""
+        self._co4e_routed_provider = decision.provider
+        self._append_chat("system", tr(
+            "routing.switched_notice",
+            model=decision.model, task=decision.task_type,
+            gain=f"{decision.score_gain:.2f}"))
+        return decision.model
+
+    def _confirm_routing_switch(self, decision) -> bool:
+        """Manual mode: ask before moving this Co4E turn to another model.
+
+        Handed to the routing service as a callback so the pure-Python decision
+        layer never needs to know a modal dialog exists."""
+        from .routing_toggle import confirm_switch
+
+        timeout = float(self.ctx.config.routing.get("confirm_timeout_sec", 60) or 60)
+        return bool(confirm_switch(self, decision, timeout))
 
     def _extract_agent_directive(self, text: str):
         m = re.search(r"(?<!\S)/agent:([\w\-.]+)", text)
diff --git a/ui/folder_tab.py b/ui/folder_tab.py
index c5aeebe..467111b 100644
--- a/ui/folder_tab.py
+++ b/ui/folder_tab.py
@@ -923,46 +923,42 @@ class FolderTab(QWidget):
     def _ai_apply_routing(self, instruction: str) -> None:
         """Auto Model Routing for the AI-Edit surface (always a CODING task).
 
-        Off → no-op. Auto → silently pick the best coding model. Manual → ask
-        first. Sets ``self._ai_routed_provider``/``_ai_routed_model`` for this
-        run; :meth:`_ai_provider` honours them. Never raises."""
+        Sets ``self._ai_routed_provider``/``_ai_routed_model`` for this run;
+        :meth:`_ai_provider` honours them.
+
+        The policy itself lives in the shared ``RoutingApplicationService``
+        (R03-T05). What stays here is genuinely AI-Edit-specific: the task type
+        is pinned to CODING (an edit instruction is never a QA question, so
+        classifying it would only add noise), and the current model comes from
+        this screen's own picker rather than the global active model."""
+        from ..core.routing.models import TaskType
+
         self._ai_routed_provider = None
         self._ai_routed_model = None
-        if not (instruction or "").strip():
+        cur_provider = self.ctx.config.active_provider
+        picked = self.ai_model_combo.currentData() if hasattr(self, "ai_model_combo") else None
+        cur_model = picked or self.ctx.config.provider_conf(cur_provider).get("model", "")
+        decision = self.ctx.routing_application().route_turn(
+            "ai_edit", instruction, cur_provider, cur_model,
+            task_type=TaskType.CODING, confirm=self._confirm_routing_switch,
+        )
+        if not decision.switched:
             return
-        try:
-            from ..core.routing.models import TaskType
-            mode = self.ctx.project_routing_mode("ai_edit")  # per-workspace mode
-            if mode == "off":
-                return
-            service = self.ctx.routing()
-            cur_provider = self.ctx.config.active_provider
-            picked = self.ai_model_combo.currentData() if hasattr(self, "ai_model_combo") else None
-            cur_model = picked or self.ctx.config.provider_conf(cur_provider).get("model", "")
-            result = service.route(
-                "ai_edit", instruction, cur_provider, cur_model,
-                mode_override=mode, task_type=TaskType.CODING,
-            )
-            if not result.should_switch:
-                return
-            target = result.target()
-            if target is None:
-                return
-            to_provider, to_model = target
-            if mode == "manual":
-                from .routing_toggle import confirm_switch
-                timeout = float(self.ctx.config.routing.get("confirm_timeout_sec", 60) or 60)
-                if not confirm_switch(self, result.decision, timeout):
-                    return
-            self._ai_routed_provider = to_provider
-            self._ai_routed_model = to_model
-            self.ai_chat.add_status(tr(
-                "routing.switched_notice",
-                model=to_model, task=result.task_type.value,
-                gain=f"{result.decision.score_gain:.2f}"))
-        except Exception:  # noqa: BLE001 — routing must never block an edit
-            self._ai_routed_provider = None
-            self._ai_routed_model = None
+        self._ai_routed_provider, self._ai_routed_model = decision.target()
+        self.ai_chat.add_status(tr(
+            "routing.switched_notice",
+            model=decision.model, task=decision.task_type,
+            gain=f"{decision.score_gain:.2f}"))
+
+    def _confirm_routing_switch(self, decision) -> bool:
+        """Manual mode: ask before moving this AI-Edit run to another model.
+
+        Passed to the routing service as a callback, keeping the pure-Python
+        decision layer free of any Qt dialog knowledge."""
+        from .routing_toggle import confirm_switch
+
+        timeout = float(self.ctx.config.routing.get("confirm_timeout_sec", 60) or 60)
+        return bool(confirm_switch(self, decision, timeout))
 
     def _ai_image_model(self):
         """Resolve the model+endpoint for image generation, searching ALL