README.md

@@ -197,21 +197,51 @@ curl -s -X POST localhost:8080/v1/chat/completions -H 'content-type: application
   -d '{"model":"auto","messages":[{"role":"user","content":"hello"}]}'
 ```
 
-- Ask for `model: "auto"` and the router picks per request.
-- Ask for any real model id and the router dispatches directly, still logged.
-- Streaming is supported: tokens pass through as they arrive while the router scrapes energy/cost from provider SSE comment lines.
-
-### Route overnight/batch work through flex rows
-
-`-flex` rows are discounted asynchronous rows that may be held during peak.
+Ask for a **virtual router model** and the router picks a candidate, subject
+to the profile you specify. The general form is `auto:<profile>`; asking for
+just `auto` is shorthand for `auto:default`.
 
 ```bash
-curl -s -X POST localhost:8080/route -H 'content-type: application/json' \
-  -d '{"task":"nightly code review","latency_tolerance":"batch"}'
+# Normal interactive routing (same as "auto")
+curl -s -X POST localhost:8080/v1/chat/completions -H 'content-type: application/json' \
+  -d '{"model":"auto","messages":...}'
+
+# Restrict to local dispatch models (ollama-local provider only)
+curl -s -X POST localhost:8080/v1/chat/completions -H 'content-type: application/json' \
+  -d '{"model":"auto:locality","messages":...}'
+
+# Restrict to models priced at ≤ $0.50/1M completion tokens
+curl -s -X POST localhost:8080/v1/chat/completions -H 'content-type: application/json' \
+  -d '{"model":"auto:onlycheaps","messages":...}'
+
+# Restrict to frontier tier (tier 3)
+curl -s -X POST localhost:8080/v1/chat/completions -H 'content-type: application/json' \
+  -d '{"model":"auto:bigboybritches","messages":...}'
+
+# Overnight/async work via flex rows
+curl -s -X POST localhost:8080/v1/chat/completions -H 'content-type: application/json' \
+  -d '{"model":"auto:batch","messages":...}'
 ```
 
-- `auto` excludes `-flex` rows; `auto:batch` admits them.
-- `latency_tolerance: batch` on any request flips the router into the batch serving class.
+| Profile | What it does |
+|---|---|
+| `auto` / `auto:default` | Normal quality-first routing; `-flex` rows excluded |
+| `auto:batch` | Admits `-flex` rows that may be held during peak hours |
+| `auto:locality` | Routes only to the `ollama-local` provider (local dispatch) |
+| `auto:onlycheaps` | Limits to models at or below $0.50 per 1M completion tokens |
+| `auto:bigboybritches` | Routes only to tier-3 (frontier) models |
+
+Profiles are **candidate-set filters** — they narrow which models the router
+may pick from, but do not change the ranking objective (quality-first, cost as
+tiebreak). An unknown profile raises HTTP 422.
+
+Operators can define custom profiles under `profiles:` in
+`config/config.yaml`. Each entry accepts `provider`, `min_tier`/`max_tier`,
+`max_cost_per_1m_completion`, `latency_tolerance`, and `allowed_model_ids`
+as rewrite rules on top of the default profile.
+
+- Ask for any real model id and the router dispatches directly, still logged.
+- Streaming is supported: tokens pass through as they arrive while the router scrapes energy/cost from provider SSE comment lines.
 
 ### Ask an image question
 

admin/frontend/decisions.html

@@ -205,6 +205,9 @@ header.navbar{padding-top:2px!important;padding-bottom:2px!important}
                   <select class="form-select form-select-sm" id="f-category">
                     <option value="">All categories</option>
                   </select>
+                  <select class="form-select form-select-sm" id="f-profile">
+                    <option value="">All profiles</option>
+                  </select>
                   <select class="form-select form-select-sm" id="f-tier">
                     <option value="">All tiers</option>
                     <option value="1">Tier 1</option>
@@ -225,12 +228,12 @@ header.navbar{padding-top:2px!important;padding-bottom:2px!important}
                 <table class="table table-vcenter table-hover card-table">
                   <thead>
                     <tr>
-                      <th>Time</th><th>Kind</th><th>Category</th><th>Tier</th>
+                      <th>Time</th><th>Kind</th><th>Category</th><th>Profile</th><th>Tier</th>
                       <th>Source</th><th>Model</th><th>Flags</th><th>Cost</th><th>Prof</th><th>Rejected</th>
                     </tr>
                   </thead>
                   <tbody id="dec-tbody">
-                    <tr><td colspan="10" class="text-muted">Loading&hellip;</td></tr>
+                    <tr><td colspan="11" class="text-muted">Loading&hellip;</td></tr>
                   </tbody>
                 </table>
               </div>
@@ -267,6 +270,7 @@ const MAX_ROWS = 2000;
 let _allDecisions = [];  // newest first
 let _knownKinds = new Set();
 let _knownCategories = new Set();
+let _knownProfiles = new Set();
 let _pageSize = 100;
 let _currentPage = 1;  // 1-indexed
 
@@ -393,25 +397,31 @@ function addDecision(dec) {
 function refreshFilterOptions(rows) {
   const kindSel = document.getElementById('f-kind');
   const catSel = document.getElementById('f-category');
+  const profSel = document.getElementById('f-profile');
   let changed = false;
   for (const d of rows) {
     if (d.kind && !_knownKinds.has(d.kind)) { _knownKinds.add(d.kind); changed = true; }
     if (d.task_category && !_knownCategories.has(d.task_category)) { _knownCategories.add(d.task_category); changed = true; }
+    if (d.profile != null && !_knownProfiles.has(d.profile)) { _knownProfiles.add(d.profile); changed = true; }
   }
   if (!changed) return;
-  const kindVal = kindSel.value, catVal = catSel.value;
+  const kindVal = kindSel.value, catVal = catSel.value, profVal = profSel.value;
   kindSel.innerHTML = '<option value="">All kinds</option>' +
     [...(_knownKinds)].sort().map(k => `<option value="${escapeHtml(k)}">${escapeHtml(k)}</option>`).join('');
   catSel.innerHTML = '<option value="">All categories</option>' +
     [...(_knownCategories)].sort().map(c => `<option value="${escapeHtml(c)}">${escapeHtml(c)}</option>`).join('');
+  profSel.innerHTML = '<option value="">All profiles</option>' +
+    [...(_knownProfiles)].sort().map(p => `<option value="${escapeHtml(p)}">${escapeHtml(p)}</option>`).join('');
   kindSel.value = kindVal;
   catSel.value = catVal;
+  profSel.value = profVal;
 }
 
 function currentFilters() {
   return {
     kind: document.getElementById('f-kind').value,
     category: document.getElementById('f-category').value,
+    profile: document.getElementById('f-profile').value,
     tier: document.getElementById('f-tier').value,
     search: document.getElementById('f-search').value.trim().toLowerCase(),
   };
@@ -420,11 +430,12 @@ function currentFilters() {
 function matchesFilters(d, f) {
   if (f.kind && d.kind !== f.kind) return false;
   if (f.category && d.task_category !== f.category) return false;
+  if (f.profile && d.profile !== f.profile) return false;
   if (f.tier && String(d.task_tier || '') !== f.tier) return false;
   if (f.search) {
     const haystack = [
       d.selected_model, d.selected_provider, d.session_key,
-      d.rejected_reason, d.task_category, d.kind,
+      d.rejected_reason, d.task_category, d.kind, d.profile,
     ].filter(Boolean).join(' ').toLowerCase();
     if (!haystack.includes(f.search)) return false;
   }
@@ -437,12 +448,13 @@ function resetPageAndRender() {
 }
 
 function setupFilters() {
-  const ids = ['f-kind', 'f-category', 'f-tier'];
+  const ids = ['f-kind', 'f-category', 'f-profile', 'f-tier'];
   ids.forEach(id => document.getElementById(id).addEventListener('change', resetPageAndRender));
   document.getElementById('f-search').addEventListener('input', resetPageAndRender);
   document.getElementById('f-clear').addEventListener('click', () => {
     document.getElementById('f-kind').value = '';
     document.getElementById('f-category').value = '';
+    document.getElementById('f-profile').value = '';
     document.getElementById('f-tier').value = '';
     document.getElementById('f-search').value = '';
     resetPageAndRender();
@@ -477,7 +489,7 @@ function renderTable() {
   const rows = _allDecisions.filter(d => matchesFilters(d, f));
   countEl.textContent = `${rows.length} of ${_allDecisions.length}`;
   if (!rows.length) {
-    tbody.innerHTML = `<tr><td colspan="10" class="text-muted text-center">No decisions match these filters</td></tr>`;
+    tbody.innerHTML = `<tr><td colspan="11" class="text-muted text-center">No decisions match these filters</td></tr>`;
     pageInfoEl.textContent = '';
     prevBtn.disabled = true;
     nextBtn.disabled = true;
@@ -500,6 +512,7 @@ function renderTable() {
       <td class="text-nowrap">${formatTs(d.observed_at || '')}</td>
       <td><span class="badge ${badgeClass}">${escapeHtml(String(d.kind || '?'))}</span></td>
       <td title="${escapeHtml(d.task_category || '')}"><span class="dec-badge" style="${categoryStyle(category)}">${escapeHtml(category)}</span></td>
+      <td>${escapeHtml(d.profile || '')}</td>
       <td class="tier-${tier}">${escapeHtml(tier)}</td>
       <td><span class="dec-badge" style="${sourceStyle(source)}">${escapeHtml(source)}</span></td>
       <td title="${escapeHtml(modelStr)}">${escapeHtml(modelStr)}</td>

config/schema.sql

@@ -245,8 +245,10 @@ CREATE TABLE IF NOT EXISTS route_decisions (
     request_id               TEXT,                   -- provider completion id
     exploration              INTEGER DEFAULT 0,      -- 0/1 whether rollout/exploration
     pinch_original_tokens    INTEGER,                -- estimated tokens before pruning
-    pinch_final_tokens       INTEGER                 -- tokens sent after pruning (may
+    pinch_final_tokens       INTEGER,                -- tokens sent after pruning (may
                                                       -- equal original when no pruning)
+    profile                  TEXT                    -- routing profile name (e.g.,
+                                                      -- 'default', 'locality')
 );
 
 CREATE INDEX IF NOT EXISTS idx_verifications_model ON verifications (model_id, provider);

docs/api.md

@@ -36,9 +36,24 @@ Input to `/route` and `/dispatch` can include `task_category`, `task_tier`,
 and `required_context_tokens` overrides — these skip the classifier, useful
 for testing routing without the classifier in the loop.
 
-Two virtual router models:
-- `auto` — normal routing, `-flex` rows excluded (interactive)
-- `auto:batch` — admits `-flex` rows (overnight/async work)
+Named routing profiles (`auto:<profile>`). The general form is
+`auto:<profile>`; `auto` alone is shorthand for `auto:default`. The router
+resolves the profile name against the built-in set and any custom entries in
+`config/config.yaml` under `profiles:`. Profiles narrow the candidate set but
+do not override the quality-first ranking objective.
+
+| Profile | Effect |
+|---|---|
+| `auto:default` | Normal quality-first routing, `-flex` rows excluded |
+| `auto:batch` | Admits `-flex` rows for overnight and async work |
+| `auto:locality` | Restricts to the `ollama-local` provider only |
+| `auto:onlycheaps` | Limits to models priced at ≤ $0.50 per 1M completion tokens |
+| `auto:bigboybritches` | Restricts to tier-3 (frontier) models only |
+
+An unknown profile name raises HTTP 422 and lists the valid names.
+
+The `profile` field is also accepted on `POST /route` and `POST /dispatch`
+to select a profile inline per request.
 
 Ask for **any real model id** in `/v1/chat/completions` and it dispatches
 directly, still logged — routing is transparent, not opaque.

plans/capability-aware-ceiling-warnings.md

@@ -0,0 +1,136 @@
+# Capability-aware ceiling warnings, and a reactive rejection detector
+
+**Status: FINAL — decision-complete.** Written 2026-09-04 from a live incident.
+
+## The incident
+
+On 2026-09-04 two image requests failed with:
+
+```
+422  tier >= 1; context >= 242486 tokens; interactive;
+     vision-capable model (request carries image(s))
+```
+
+Cause: admin availability overrides had `kimi-k3`, `kimi-k3-fast` and
+`kimi-k3-flex` deprecated. Those are the ONLY vision-capable rows with enough
+context (782,324). Every remaining active vision model tops out at 192,500,
+below the request's 242,486. The intersection of the hard filters was empty.
+
+This is [incident #3](../docs/incidents.md) recurring through a dimension the
+existing detector does not model. Resolved by re-activating `kimi-k3`; verified
+by recomputing the candidate set through `routing.select_candidates`, which now
+returns exactly one row.
+
+## What already exists — do not rebuild it
+
+Establish this before writing code, because a first pass at this analysis got
+it wrong by checking for a top-level `warnings` key:
+
+- **`/metrics` surfaces warnings under `coverage.warnings`**, not at top level.
+  `/admin/api/snapshot` carries the same block. The plumbing is done.
+- **`metrics.context_ceilings` already applies admin deprecations** via
+  `exclude_models`, which was the fix after incident #3, and delegates to
+  `routing.select_candidates` so the ceiling matches live routing.
+- **`metrics.demand_ceiling_warnings` already compares ceiling against observed
+  demand**, which is the *correct* shape. Do not replace it with an absolute
+  tier-ordering check: `ceiling(1) >= ceiling(2) >= ceiling(3)` is a theorem
+  (tier is a capability floor, so the eligible set shrinks monotonically), so
+  such a warning fires always and means nothing. `docs/incidents.md` #3 records
+  this trap; do not re-enter it.
+- **`route_decisions` already records every rejection** — `selected_model IS
+  NULL` with the full `rejected_reason` string.
+
+## Gap 1 — ceilings are blind to capability-gated subsets
+
+`context_ceilings` buckets by `(tier, latency_tolerance)` only. During the
+incident the tier-1 interactive ceiling across all models was still **782,324**
+(deepseek and others are large and active), so every existing check was silent
+while the *vision-capable* ceiling had collapsed to 192,500.
+
+Measured during the incident:
+
+| subset | ceiling |
+|---|---|
+| all active models | 782,324 |
+| **vision-capable only** | **192,500** |
+
+The hard filters that can independently empty the candidate set, from
+`routing.rejection_reason`, are `require_vision` and `require_json_mode` —
+both **fail closed on NULL** (an unconfirmed capability is treated as absent),
+which is what makes them able to shrink the set sharply.
+
+**Add capability sub-ceilings.** Extend the bucket key, or compute a small
+number of additional named buckets: `vision` and `json_mode`. Do NOT add a
+bucket per capability combination — that is a combinatorial explosion for
+dimensions that do not interact in practice. Two extra series, compared against
+the demand actually observed for requests carrying images / requesting JSON
+(`route_decisions.images`, `.json_mode` are already recorded), is enough.
+
+Reuse `demand_ceiling_warnings`' existing demand-relative shape; only the
+bucketing changes.
+
+## Gap 2 — nothing watches actual rejections
+
+This is the more valuable half, and it is simpler.
+
+Every hard-filter rejection is already persisted with its reason. Nothing reads
+them. A detector that says *"N requests were rejected in the last hour"* would
+have caught this incident within minutes, **without modelling any capability
+dimension at all** — and would equally catch the next failure through a
+dimension nobody predicted.
+
+Predictive checks (Gap 1) only catch dimensions you thought of. A reactive
+check catches everything, at the cost of firing after the first failure rather
+than before. Ship both; the reactive one is the safety net.
+
+Add a warning in `metrics.scoring_coverage`'s existing `warnings` list:
+
+- Count `route_decisions` rows with `selected_model IS NULL AND rejected_reason
+  IS NOT NULL` in a recent window (1 hour and 24 hours are both useful; pick
+  one, state which).
+- Group by a **normalized** reason, not the literal string. The raw strings
+  embed the request's token count (`context >= 242486 tokens`), so grouping by
+  the literal yields n=1 per row and hides the pattern. Normalize by replacing
+  digit runs with a placeholder before grouping.
+- Report the count, the normalized reason, and the most recent timestamp, so an
+  operator can tell a live problem from an old one.
+- **Zero rejections must produce no warning.** A rejection is not inherently an
+  error — a genuinely impossible request should 422. The signal is a *rate*.
+
+## Non-goals
+
+- Do not change `routing.py`. Nothing about the hard filters is wrong; the
+  candidate set was correctly empty. This plan is observability only.
+- Do not auto-revert admin overrides or "helpfully" re-activate models. The
+  deprecations were deliberate operator cost decisions. Warn; do not act.
+- Do not add a bucket per capability combination.
+- Do not replace `demand_ceiling_warnings` with an absolute tier-ordering
+  comparison (see above — it is a theorem).
+- Do not touch the `local_vision` fallback. It could not have rescued this
+  request anyway: that path receives the raw unpruned message list, and 242,486
+  tokens against `num_ctx 8192` was never going to fit. Worth a doc note, not a
+  code change.
+
+## Success criteria
+
+- A test reproducing the incident state — `kimi-k3*` excluded via
+  `admin_model_overrides`, a vision request above 192,500 tokens — produces a
+  warning naming the vision subset. The same state produces **no** warning from
+  the existing `(tier, latency_tolerance)` checks, proving the gap was real and
+  is now closed.
+- A test with recent NULL-selection rows produces a rejection-rate warning
+  reporting a normalized reason and a count; a test with zero such rows produces
+  none.
+- Both warnings appear in `coverage.warnings` on `/metrics` and in
+  `/admin/api/snapshot`, and are visible on the admin dashboard's warnings bell.
+- `demand_ceiling_warnings` keeps its demand-relative shape; no absolute
+  tier-ordering check is introduced.
+- Full suite green with `local_energy.enabled` both true and false.
+- The user's `config/config.yaml` tariff lines remain uncommitted and verbatim.
+
+## Related, out of scope but worth recording
+
+`coverage.warnings` currently reports **metered usage at 146% of the 6.25 kWh
+plan allowance** (9.15 kWh, next reset 2026-09-06). That warning is working as
+designed and is almost certainly why provider credits ran out on 2026-09-02/03.
+It needs an operator decision about the plan, not a code change.

plans/named-routing-profiles.md

@@ -0,0 +1,183 @@
+# Named routing profiles
+
+**Status: FINAL — decision-complete. §3 was answered by the user on 2026-09-03
+(filter only).** Written 2026-09-03 against `main` at `c2893b5`.
+
+## What this is
+
+Today the router has exactly one behaviour: classify, hard-filter, rank
+quality-first with cost as a tiebreak. The user wants named alternatives —
+`locality`, `onlycheaps`, `bigboybritches` — alongside a `default` that stays
+the feature-rich automatic router it is now.
+
+**The mechanism already exists and is half-built.** `auto:batch` is a profile:
+a named variant of `auto` that changes one hard-filter parameter
+(`latency_tolerance`), documented in `README` and shipped for months. This plan
+generalises that one hard-coded special case into a named, configurable set.
+It is deliberately NOT a new ranking path.
+
+## 1. Where it slots in, precisely
+
+Three call sites, all already doing most of the work:
+
+- **`src/dispatcher.py:2937`** —
+  `wants_routing = bare in (ROUTER_MODEL, ROUTER_MODEL_BATCH)`. An exact
+  membership test against two constants. This becomes a *parse*: split
+  `auto:<profile>`, look the profile up, reject an unknown one.
+- **`src/dispatcher.py:2955`** —
+  `latency = BATCH if requested == ROUTER_MODEL_BATCH else INTERACTIVE`. This
+  is the existing profile, expressed as a ternary. It must become one case of
+  the general mechanism, not survive alongside it.
+- **`src/routing.py:313` `select_candidates`** — already takes
+  `exclude_models: set[str]`, `task_category`, `latency_tolerance`,
+  `allowed_access_levels`, `min_tool_proficiency`. A profile supplies or
+  overrides these. **All the filtering machinery is already there.**
+
+`routing.py` is a pure module with injected dependencies and
+`rejection_reason` is the single copy of the rules (`is_eligible` is a thin
+wrapper over it). Keep that: a profile must not introduce a second place where
+eligibility is decided.
+
+## 2. The one genuinely missing primitive
+
+`select_candidates` has a **denylist** (`exclude_models`) and no **allowlist**.
+Every profile the user named is naturally an allowlist:
+
+- `locality` → only `provider = 'ollama-local'` rows
+- `onlycheaps` → only rows under a cost bar
+- `bigboybritches` → only frontier rows
+
+Add **one** parameter, `restrict_to: set[str] | None = None`, threaded through
+`rejection_reason` → `is_eligible` → `select_candidates`, with `None` meaning
+"unrestricted" (NOT an empty set, which must mean "nothing eligible"). Give it
+its own `rejection_reason` string so a 422 says *which* profile emptied the
+candidate set — the catalog-staleness incident is the precedent: an empty
+candidate set with an opaque reason cost ~19 hours to diagnose.
+
+Prefer expressing profiles as **predicates over catalog columns** (provider,
+cost, tier) rather than hard-coded model-id lists, so a profile does not go
+stale the moment the poller adds a row. A literal id list stays available for
+the cases where that is genuinely what the user means.
+
+## 3. THE OPEN DECISION — filter only, or objective overrides too?
+
+**DECIDED 2026-09-03 by the user: FILTER ONLY.** A profile restricts the
+candidate set and nothing else. Ranking stays quality-first with cost as the
+tiebreak, one rule for the whole system. Do NOT implement objective overrides,
+and do NOT add a `quality_tolerance` field to a profile definition — if a
+future preference-shaped profile needs one, that is a separate plan with its
+own justification.
+
+The reasoning is preserved below because it explains WHY filter-only is
+sufficient, which is not obvious.
+
+A profile is unambiguously a candidate-set filter. The question is whether it
+may ALSO override `objective.*` (`quality_tolerance`,
+`max_energy_per_request`).
+
+It matters concretely, and we have the measurement:
+`plans/local-dispatch-inert-and-test-coupling.md` established that the local
+model scores 0.767 on `file_summarization` against cloud's 0.95 — a 0.183 gap
+against a `quality_tolerance` of 0.1 — so local **never routes** under
+quality-first ranking.
+
+- A **pure-filter** `locality` still works: restrict the set to local rows and
+  the best local model wins by default, because there is no cloud row left to
+  lose to. This covers the user's stated case.
+- A **mixed** profile that *prefers* local without excluding cloud does NOT
+  work as a pure filter. It reproduces the dormancy exactly.
+
+So: pure filters satisfy the three named profiles. Objective overrides are only
+needed for preference-shaped profiles nobody has asked for yet.
+
+**Recommendation: ship filter-only.** It is the smaller change, it satisfies
+every named use case, and it keeps one ranking rule in the system. Add
+overrides later if a real preference-shaped profile appears. If the user wants
+overrides now, they must be scoped strictly to the profile and never mutate
+global config — and `quality_tolerance` in particular describes measurement
+noise on 2-3 samples, so overriding it means asserting a preference through a
+knob that means something else.
+
+## 4. Naming and selection
+
+Profiles are selected the way `auto:batch` already is: the model field.
+
+```
+auto                  -> default profile
+auto:batch            -> MUST keep working, byte-identical behaviour
+auto:locality         -> named profile
+<real model id>       -> passthrough, unchanged
+```
+
+`auto:batch` is load-bearing: it is in `README`, `docs/`, and possibly user
+clients. **Reimplement it as a profile whose definition sets
+`latency_tolerance: batch`, and pin its behaviour with a test that would fail
+if the general mechanism changed it.** Do not leave the ternary in place beside
+the new path — two mechanisms for one behaviour is how they drift.
+
+An unknown profile (`auto:nonsense`) must fail **loudly** with a 422 naming the
+valid profiles. It must NOT silently fall back to `default`: a user who
+mistypes `auto:locallity` would otherwise get cloud routing and a surprising
+bill, with nothing indicating why.
+
+Config lives under a new top-level `profiles:` key. Config is strict
+(`extra="forbid"`), so every field needs declaring; an unknown key is an error
+by design.
+
+## 5. Observability
+
+`route_decisions` must record which profile served each decision, or the
+proficiency loop cannot tell a `bigboybritches` win from a `default` one and
+will read profile-forced selections as evidence of model quality. This is the
+same reasoning that made the local fallback record
+`kind='local_dispatch_fallback'` rather than `chat`.
+
+Add a `profile` column (code-side migration in `ensure_route_decisions`, as
+every prior column addition did — `config/schema.sql` is
+`CREATE TABLE IF NOT EXISTS` and does not alter a live DB). Surface it in the
+admin decisions table and its filters, which already filter by kind, category
+and tier.
+
+## 6. Interactions to get right
+
+- **`eligible_categories`** (local dispatch) is a per-MODEL filter; a profile
+  is a per-REQUEST filter. They compose as AND. A `locality` profile must not
+  bypass `eligible_categories` — that gate is what keeps a summarization-grade
+  model away from code, and `nemotron-mini:4b` detecting 0/4 bugs is why.
+- **The local-dispatch fallback** (`kind='local_dispatch_fallback'`) is a
+  degraded path, not a profile. Leave it alone. Under `locality` the local
+  model is the primary choice, so the fallback should simply never fire.
+- **Circuit breaker and admin overrides** already feed `exclude_models`.
+  Profiles must AND with them, never replace them: a profile must not resurrect
+  a model an operator deprecated or the breaker has marked down.
+- **`exploration`** (epsilon-greedy) picks from eligible candidates. Confirm it
+  reads the profile-filtered set, or exploration will select models the profile
+  excluded.
+
+## Non-goals
+
+- No new ranking algorithm. Quality-first with cost as tiebreak stays.
+- No change to `objective.quality_tolerance`'s global value.
+- No per-category preference logic (that was §1(a) of the previous plan and is
+  explicitly out of scope here).
+- No changes to the local-dispatch fallback path.
+- Do not touch the classifier: profiles select candidates, not categories.
+
+## Success criteria
+
+- `auto` and `auto:batch` behave **exactly** as today, pinned by tests that
+  would catch a regression in the general mechanism.
+- At least the three named profiles ship, defined as predicates over catalog
+  columns where possible.
+- An unknown profile returns 422 naming the valid ones; it never silently
+  degrades to `default`.
+- A profile that empties the candidate set returns a rejection reason naming
+  the profile.
+- `route_decisions.profile` is populated and visible in the admin decisions
+  table.
+- Profiles AND with circuit-breaker exclusions, admin overrides, and
+  `eligible_categories` — proven by a test for each, not by inspection.
+- `rejection_reason` remains the single copy of the eligibility rules.
+- Full suite green with `local_energy.enabled` both true and false (the Plan 5
+  invariant must not regress).
+- The user's `config/config.yaml` tariff lines remain uncommitted and verbatim.

src/config.py

@@ -194,6 +194,58 @@ class FlexPreference(Enum):
     force_flex = "force-flex"
 
 
+class RoutingProfile(StrictModel):
+    """A named candidate-set filter requested through ``auto:<name>``.
+
+    Profiles restrict the candidate set; they do NOT change ranking. All
+    fields are optional, and any absent field falls back to the request's
+    own value or the global default. This keeps profiles small rewrite
+    rules rather than a parallel config layer.
+    """
+
+    provider: Optional[str] = None
+    min_tier: Optional[int] = None
+    max_tier: Optional[int] = None
+    latency_tolerance: Optional[Literal["interactive", "batch"]] = None
+    max_cost_per_1m_completion: Optional[float] = None
+    allowed_model_ids: Optional[set[str]] = None
+
+    @field_validator("min_tier", "max_tier")
+    @classmethod
+    def tier_in_range(cls, v: Optional[int]) -> Optional[int]:
+        if v is not None and v not in (1, 2, 3):
+            raise ValueError(
+                f"profiles[...].{cls.__name__}.tier must be in {{1, 2, 3}}, got {v}"
+            )
+        return v
+
+    @model_validator(mode="after")
+    def min_not_above_max(self) -> "RoutingProfile":
+        if self.min_tier is not None and self.max_tier is not None:
+            if self.min_tier > self.max_tier:
+                raise ValueError(
+                    f"profiles[...].{self.__class__.__name__}.min_tier ({self.min_tier}) "
+                    f"must be <= max_tier ({self.max_tier})"
+                )
+        return self
+
+    @field_validator("max_cost_per_1m_completion")
+    @classmethod
+    def cost_positive(cls, v: Optional[float]) -> Optional[float]:
+        if v is not None and v <= 0:
+            raise ValueError(
+                "profiles[...].max_cost_per_1m_completion must be > 0, or null"
+            )
+        return v
+
+    @field_validator("allowed_model_ids")
+    @classmethod
+    def allowed_model_ids_non_empty(cls, v: Optional[set[str]]) -> Optional[set[str]]:
+        if v is not None and not v:
+            raise ValueError("profiles[...].allowed_model_ids must be non-empty when set")
+        return v
+
+
 class RoutingConfig(StrictModel):
     allowed_access_levels: list[str]
     default_latency_tolerance: str
@@ -704,6 +756,7 @@ class RouterConfig(StrictModel):
     logging: LoggingConfig
     local_energy: LocalEnergyConfig = LocalEnergyConfig()
     local_dispatch_models: list[LocalDispatchModel] = []
+    profiles: dict[str, RoutingProfile] = {}
 
     @model_validator(mode="after")
     def local_energy_needs_tariff_when_enabled(self) -> "RouterConfig":

src/dispatcher.py

@@ -46,6 +46,7 @@ import time
 from datetime import datetime, timezone
 from pathlib import Path
 from statistics import median
+from collections.abc import Sequence
 from typing import Any, Literal, Optional
 
 import requests
@@ -63,7 +64,7 @@ import local_energy
 import logs
 import session_cache
 from capabilities import detect_capabilities, iter_image_url_values
-from config import FlexPreference, RouterConfig, load_config
+from config import FlexPreference, RouterConfig, RoutingProfile, load_config
 from context_prune import (
     _text_only,
     estimate_tokens,
@@ -84,8 +85,6 @@ from metrics import (
     verdict_mix,
 )
 from routing import (
-    BATCH,
-    INTERACTIVE,
     apply_flex_preference,
     capability_gate_reason,
     rank_candidates,
@@ -104,7 +103,17 @@ from verification import (
 # Virtual model names that mean "you pick". Anything else is taken as a real
 # model id and dispatched as asked.
 ROUTER_MODEL = "auto"
-ROUTER_MODEL_BATCH = "auto:batch"
+
+# Built-in routing profiles. Profiles are selectable as `auto:<name>`; they
+# restrict the candidate set and may override the effective latency_tolerance.
+# Unknown profiles raise 422, so unknown built-ins do not silently fall back.
+BUILTIN_PROFILES: dict[str, RoutingProfile] = {
+    "default": RoutingProfile(),
+    "batch": RoutingProfile(latency_tolerance="batch"),
+    "locality": RoutingProfile(provider="ollama-local"),
+    "bigboybritches": RoutingProfile(min_tier=3),
+    "onlycheaps": RoutingProfile(max_cost_per_1m_completion=0.5),
+}
 
 # Observations from the fixed reference workload in seed_energy.py. Only
 # these steer routing; organic traffic is logged for accounting but varies
@@ -219,6 +228,13 @@ class TaskRequest(BaseModel):
         "requires a JSON-mode-capable model (routing.require_json_mode).",
     )
     # Overrides, mostly for testing the router without the classifier in the loop.
+    profile: Optional[str] = Field(
+        None,
+        description=(
+            "Named routing profile (e.g. 'locality', 'batch') to apply for this "
+            "request. Defaults to the 'default' profile."
+        ),
+    )
     task_category: Optional[str] = None
     task_tier: Optional[int] = Field(None, ge=1, le=3)
     required_context_tokens: Optional[int] = Field(None, ge=0)
@@ -258,6 +274,7 @@ class Candidate(BaseModel):
 class RouteResponse(BaseModel):
     classification: Classification
     latency_tolerance: str
+    profile: str = "default"
     flex_preference: str = "auto"
     flex_swapped: bool = False
     flex_forced: bool = False
@@ -286,6 +303,76 @@ def _ms(started: float) -> int:
     return int((time.perf_counter() - started) * 1000)
 
 
+def _all_profile_names() -> list[str]:
+    names = set(BUILTIN_PROFILES) | set(cfg.profiles)
+    return sorted(names)
+
+
+def _resolve_profile(requested: str) -> tuple[str, RoutingProfile]:
+    if requested == ROUTER_MODEL:
+        name = "default"
+    else:
+        name = requested[len(ROUTER_MODEL) + 1 :]  # after "auto:"
+    return _resolve_profile_name(name)
+
+
+def _resolve_profile_name(name: str) -> tuple[str, RoutingProfile]:
+    """Resolve a bare profile name, returning (name, profile_obj) or 422."""
+    builtin = BUILTIN_PROFILES.get(name)
+    configured = cfg.profiles.get(name)
+    if builtin is None and configured is None:
+        valid = _all_profile_names()
+        raise HTTPException(
+            422,
+            f"Unknown routing profile {name!r}. Valid profiles: {valid}",
+        )
+    if configured is not None and builtin is not None:
+        merged = builtin.model_copy(
+            update={
+                k: v
+                for k, v in configured.model_dump().items()
+                if v is not None
+            }
+        )
+        return name, merged
+    return name, (configured if configured is not None else builtin)
+
+
+def _effective_latency(profile: RoutingProfile, req: Optional[TaskRequest]) -> str:
+    return (
+        profile.latency_tolerance
+        or (req.latency_tolerance if req is not None else None)
+        or cfg.routing.default_latency_tolerance
+    )
+
+
+def _restrict_to_from_profile(profile: RoutingProfile, rows: Sequence[dict]) -> set[str] | None:
+    allowed: set[str] = {r["model_id"] for r in rows}
+    if profile.provider is not None:
+        allowed &= {r["model_id"] for r in rows if r.get("provider") == profile.provider}
+    if profile.min_tier is not None or profile.max_tier is not None:
+        def tier_ok(r: dict) -> bool:
+            tier = r.get("tier")
+            if tier is None:
+                return False
+            if profile.min_tier is not None and tier < profile.min_tier:
+                return False
+            if profile.max_tier is not None and tier > profile.max_tier:
+                return False
+            return True
+        allowed &= {r["model_id"] for r in rows if tier_ok(r)}
+    if profile.max_cost_per_1m_completion is not None:
+        allowed &= {
+            r["model_id"]
+            for r in rows
+            if r.get("cost_per_1m_completion") is not None
+            and r["cost_per_1m_completion"] <= profile.max_cost_per_1m_completion
+        }
+    if profile.allowed_model_ids is not None:
+        allowed &= profile.allowed_model_ids
+    return allowed if allowed != {r["model_id"] for r in rows} else None
+
+
 def _db() -> sqlite3.Connection:
     conn = sqlite3.connect(cfg.database.path)
     conn.row_factory = sqlite3.Row
@@ -338,7 +425,8 @@ def ensure_route_decisions(conn: sqlite3.Connection) -> None:
             request_id               TEXT,
             exploration              INTEGER DEFAULT 0,
             pinch_original_tokens    INTEGER,
-            pinch_final_tokens       INTEGER
+            pinch_final_tokens       INTEGER,
+            profile                  TEXT
         )
         """
     )
@@ -360,6 +448,7 @@ def ensure_route_decisions(conn: sqlite3.Connection) -> None:
         ("exploration", "INTEGER DEFAULT 0"),
         ("pinch_original_tokens", "INTEGER"),
         ("pinch_final_tokens", "INTEGER"),
+        ("profile", "TEXT"),
     ):
         if name not in existing:
             conn.execute(f"ALTER TABLE route_decisions ADD COLUMN {name} {decl}")
@@ -834,8 +923,15 @@ def _to_candidate(row: dict) -> Candidate:
     return Candidate(**fields)
 
 
-def route(req: TaskRequest) -> RouteResponse:
-    latency_tolerance = req.latency_tolerance or cfg.routing.default_latency_tolerance
+def route(
+    req: TaskRequest,
+    *,
+    profile: str = "default",
+    profile_obj: Optional[RoutingProfile] = None,
+) -> RouteResponse:
+    if profile_obj is None:
+        profile_obj = BUILTIN_PROFILES["default"]
+    latency_tolerance = _effective_latency(profile_obj, req)
 
     if req.task_category and req.task_tier and req.required_context_tokens is not None:
         classification = Classification(
@@ -882,7 +978,8 @@ def route(req: TaskRequest) -> RouteResponse:
         ),
         task_category=classification.task_category,
     )
-    eligible = select_candidates(rows, **filters)
+    restrict_to = _restrict_to_from_profile(profile_obj, rows)
+    eligible = select_candidates(rows, restrict_to=restrict_to, **filters)
     if logs.enabled_for_debug():
         # "No model satisfies the hard filters" is otherwise a dead end with no
         # explanation. One line per drop, naming the filter and its numbers, is
@@ -976,6 +1073,7 @@ def route(req: TaskRequest) -> RouteResponse:
     return RouteResponse(
         classification=classification,
         latency_tolerance=latency_tolerance,
+        profile=profile,
         flex_preference=flex_pref.value,
         flex_swapped=flex_swapped,
         flex_forced=flex_forced,
@@ -1056,6 +1154,7 @@ def persist_route_decision(
     exploration=0,
     pinch_original_tokens=None,
     pinch_final_tokens=None,
+    profile: Optional[str] = None,
 ) -> Optional[int]:
     """Record one routing decision to route_decisions, best-effort and gated.
 
@@ -1088,6 +1187,7 @@ def persist_route_decision(
     derived_runners = None
     candidates = None
     derived_latency = latency_tolerance
+    derived_profile = profile
 
     if isinstance(classification, RouteResponse):
         clf = classification.classification
@@ -1095,6 +1195,7 @@ def persist_route_decision(
         derived_runners = classification.runners_up
         candidates = classification.candidates_considered
         derived_latency = classification.latency_tolerance
+        derived_profile = classification.profile
         flex_preference = classification.flex_preference
         flex_swapped = int(classification.flex_swapped)
         flex_forced = int(classification.flex_forced)
@@ -1151,8 +1252,8 @@ def persist_route_decision(
                 est_cost_usd, est_proficiency, rejected_reason, session_key,
                 tools, images, json_mode, streamed,
                 flex_preference, flex_swapped, flex_forced, exploration,
-                request_id, pinch_original_tokens, pinch_final_tokens
-            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                request_id, pinch_original_tokens, pinch_final_tokens, profile
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
             (
                 observed_at,
@@ -1183,6 +1284,7 @@ def persist_route_decision(
                 None,
                 pinch_original_tokens,
                 pinch_final_tokens,
+                derived_profile,
             ),
         )
         decision_id: Optional[int] = int(cursor.lastrowid)
@@ -1222,6 +1324,7 @@ def persist_route_decision(
                 "exploration": exploration,
                 "pinch_original_tokens": pinch_original_tokens,
                 "pinch_final_tokens": pinch_final_tokens,
+                "profile": derived_profile,
             }
         )
         return decision_id
@@ -1898,7 +2001,8 @@ def route_endpoint(req: TaskRequest):
     """Classify and pick a model without calling it."""
     logs.new_trace()
     started = time.perf_counter()
-    decision = route(req)
+    profile_name, profile_obj = _resolve_profile_name(req.profile or "default")
+    decision = route(req, profile=profile_name, profile_obj=profile_obj)
     log_decision(
         decision,
         tools=req.tools_present,
@@ -2859,9 +2963,11 @@ def list_models():
         conn.close()
 
     data = [
-        {"id": ROUTER_MODEL, "object": "model", "owned_by": "router"},
-        {"id": ROUTER_MODEL_BATCH, "object": "model", "owned_by": "router"},
+        {"id": f"{ROUTER_MODEL}:{name}", "object": "model", "owned_by": "router"}
+        for name in _all_profile_names()
     ]
+    # `auto` (default profile) is exposed as a bare alias for convenience.
+    data.insert(0, {"id": ROUTER_MODEL, "object": "model", "owned_by": "router"})
     data += [
         {"id": r["model_id"], "object": "model", "owned_by": r["provider"]}
         for r in rows
@@ -2934,7 +3040,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
     # check, the same unstripped id would have gone upstream and drawn a 400
     # from NeuralWatt, which only knows the bare form.
     bare = requested.rsplit("/", 1)[-1]
-    wants_routing = bare in (ROUTER_MODEL, ROUTER_MODEL_BATCH)
+    wants_routing = bare == ROUTER_MODEL or bare.startswith(f"{ROUTER_MODEL}:")
     if wants_routing or (
         bare != requested
         and (_model_exists(bare) or _local_dispatch_config_for(bare) is not None)
@@ -2952,7 +3058,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
     )
 
     if wants_routing:
-        latency = BATCH if requested == ROUTER_MODEL_BATCH else INTERACTIVE
+        profile_name, profile_obj = _resolve_profile(requested)
         tools_present = caps.tools_present
         # The turn before the last: a short follow-up inherits its complexity,
         # so the classifier sees "Context: <prior>\n---\nMessage: <current>"
@@ -3009,17 +3115,20 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
                     task=_last_user_text(messages),
                     # context omitted: the override branch (task_category +
                     # task_tier + required_context_tokens all provided)
-                    # never reads req.context — it skips classify().
-                    latency_tolerance=latency,
+                    # never reads req.context -- it skips classify().
+                    latency_tolerance=_effective_latency(profile_obj, None),
                     tools_present=tools_present,
                     has_images=caps.has_images,
                     require_json_mode=caps.require_json_mode,
                     task_category=cached.task_category,
                     task_tier=cached.task_tier,
                     required_context_tokens=measured,
-                )
+                ),
+                profile=profile_name,
+                profile_obj=profile_obj,
             )
         else:
+
             # Cache miss: classify as today, then re-route on measured context
             # if the conversation outgrows the classifier's own estimate. After
             # the decision, a successful (non-fallback) classification is
@@ -3029,14 +3138,16 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
                 TaskRequest(
                     task=_last_user_text(messages),
                     context=prev_context,
-                    latency_tolerance=latency,
+                    latency_tolerance=_effective_latency(profile_obj, None),
                     tools_present=tools_present,
                     has_images=caps.has_images,
                     require_json_mode=caps.require_json_mode,
                     # Take whichever is larger: what the classifier thinks it
                     # needs, or what the conversation actually measures.
                     required_context_tokens=None,
-                )
+                ),
+                profile=profile_name,
+                profile_obj=profile_obj,
             )
             # The classifier's own verdict, before the re-route below rewrites
             # the Classification's source to 'override'.
@@ -3050,15 +3161,17 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
                         task=_last_user_text(messages),
                         # context omitted: the override branch (task_category +
                         # task_tier + required_context_tokens all provided)
-                        # never reads req.context — it skips classify().
-                        latency_tolerance=latency,
+                        # never reads req.context -- it skips classify().
+                        latency_tolerance=_effective_latency(profile_obj, None),
                         tools_present=tools_present,
                         has_images=caps.has_images,
                         require_json_mode=caps.require_json_mode,
                         task_category=decision.classification.task_category,
                         task_tier=decision.classification.task_tier,
                         required_context_tokens=measured,
-                    )
+                    ),
+                    profile=profile_name,
+                    profile_obj=profile_obj,
                 )
             if (
                 cfg.session_cache.enabled
@@ -3092,6 +3205,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
                         flex_preference=cfg.routing.default_flex_preference.value,
                         flex_swapped=0,
                         flex_forced=0,
+                        profile=None,
                     )
                     return _local_vision_response(
                         fallback, streaming=bool(body.get("stream"))
@@ -3211,6 +3325,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
             flex_preference=cfg.routing.default_flex_preference.value,
             flex_swapped=0,
             flex_forced=0,
+            profile=None,
             pinch_original_tokens=pinch_stats.get("original_tokens")
             if pinch_stats is not None else None,
             pinch_final_tokens=pinch_stats.get("final_tokens")
@@ -3305,6 +3420,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
             flex_preference=decision.flex_preference,
             flex_swapped=decision.flex_swapped,
             flex_forced=decision.flex_forced,
+            profile=decision.profile,
         )
         _write_request_id(fallback_row, result["request_id"])
         return _local_dispatch_response(
@@ -3686,7 +3802,8 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
 def dispatch_endpoint(req: TaskRequest):
     logs.new_trace()
     started = time.perf_counter()
-    decision = route(req)
+    profile_name, profile_obj = _resolve_profile_name(req.profile or "default")
+    decision = route(req, profile=profile_name, profile_obj=profile_obj)
     log_decision(
         decision,
         tools=req.tools_present,

src/metrics.py

@@ -424,7 +424,7 @@ def recent_decisions(
                    runner_up_models, est_cost_usd, est_proficiency,
                    rejected_reason, session_key, tools, images, json_mode, streamed,
                    flex_preference, flex_swapped, flex_forced,
-                   pinch_original_tokens, pinch_final_tokens
+                   pinch_original_tokens, pinch_final_tokens, profile
               FROM route_decisions
              ORDER BY id DESC
              LIMIT ?

src/routing.py

@@ -73,6 +73,7 @@ def rejection_reason(
     require_vision: bool = False,
     require_json_mode: bool = False,
     task_category: str | None = None,
+    restrict_to: set[str] | None = None,
 ) -> str | None:
     """Why this row is not a candidate, or None if it is one.
 
@@ -159,6 +160,11 @@ def rejection_reason(
     if eligible is not None and (task_category is None or task_category not in eligible):
         return "category_ineligible"
 
+    # Profile allowlist. None means unrestricted; an empty set means nothing is
+    # eligible, and a non-empty set means only those model ids may be selected.
+    if restrict_to is not None and row.get("model_id") not in restrict_to:
+        return "profile_excluded"
+
     return None
 
 
@@ -291,10 +297,9 @@ def apply_flex_preference(
     ) is not None:
         return selected_row, False, False, selected_row.get("cost")
 
-    if pref == "prefer-flex":
+    if pref == "prefer-flex" and latency_tolerance == INTERACTIVE:
         # A flex sibling is eligible under the current hard filters unless the
         # latency filter excludes flex rows — which is exactly interactive.
-        if latency_tolerance == INTERACTIVE:
         return selected_row, False, False, selected_row.get("cost")
 
     # prefer-flex under batch, or force-flex (which bypasses interactive).
@@ -324,6 +329,7 @@ def select_candidates(
     require_vision: bool = False,
     require_json_mode: bool = False,
     task_category: str | None = None,
+    restrict_to: set[str] | None = None,
 ) -> list[dict]:
     """Apply every hard filter, preserving input order."""
     return [
@@ -342,6 +348,7 @@ def select_candidates(
             require_vision=require_vision,
             require_json_mode=require_json_mode,
             task_category=task_category,
+            restrict_to=restrict_to,
         )
     ]
 

tests/test_chat_completions.py

@@ -212,6 +212,36 @@ def test_auto_routes_and_reports_the_model_it_actually_used(router):
     assert resp.json()["model"] == CHEAP
 
 
+def test_auto_locality_profile_routes_to_locality_provider(router):
+    client, _, _ = router
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto:locality", "messages": _messages()},
+    )
+    assert resp.status_code == 422, "no ollama-local rows in fixture"
+
+
+def test_routed_chat_named_profile_persists_profile(router):
+    client, calls, _ = router
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto:batch", "messages": _messages()},
+    )
+    assert resp.status_code == 200
+    assert calls[0]["body"]["model"] == CHEAP
+
+    db_path = dispatcher.cfg.database.path
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    rows = conn.execute(
+        "SELECT * FROM route_decisions ORDER BY id"
+    ).fetchall()
+    conn.close()
+    assert len(rows) == 1
+    assert rows[0]["profile"] == "batch"
+    assert rows[0]["kind"] == "chat"
+
+
 def test_a_provider_prefixed_router_name_still_routes(router):
     """opencode sends `llm-router/auto`; only the virtual names are stripped."""
     client, calls, _ = router
@@ -682,6 +712,46 @@ def test_a_json_object_request_without_a_json_capable_model_422s(router):
     assert not calls
 
 
+def test_auto_batch_sets_latency_tolerance_to_batch(router):
+    """auto:batch must resolve to the built-in batch profile."""
+    client, calls, _ = router
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto:batch", "messages": _messages()},
+    )
+    assert resp.status_code == 200
+    assert resp.headers["X-Router-Model"] == CHEAP
+
+
+def test_unknown_profile_returns_422_naming_valid_profiles(router):
+    """A misspelled profile must not silently fall back to default."""
+    client, calls, _ = router
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto:nosuchprofile", "messages": _messages()},
+    )
+    assert resp.status_code == 422
+    detail = resp.json()["detail"]
+    assert "nosuchprofile" in detail
+    assert "default" in detail
+    assert "batch" in detail
+    assert not calls
+
+
+def test_v1_models_lists_all_profiles(router):
+    """Every valid profile appears as auto:<name> in the models list."""
+    client, _, _ = router
+    resp = client.get("/v1/models")
+    assert resp.status_code == 200
+    ids = {m["id"] for m in resp.json()["data"]}
+    assert "auto" in ids
+    assert "auto:default" in ids
+    assert "auto:batch" in ids
+    assert "auto:locality" in ids
+    assert "auto:bigboybritches" in ids
+    assert "auto:onlycheaps" in ids
+
+
 def test_a_plain_text_request_is_unaffected_by_the_gates(router):
     """No images, no response_format: routing is exactly as before."""
     client, calls, _ = router

tests/test_config.py

@@ -0,0 +1,114 @@
+"""Load-time validation for ``RouterConfig`` and its nested models.
+
+This file covers core config shape tests that are not tied to the service
+endpoints in ``test_config_endpoints.py``.
+"""
+
+from __future__ import annotations
+
+import copy
+from pathlib import Path
+
+import pytest
+import yaml
+
+from config import RouterConfig
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+@pytest.fixture
+def raw() -> dict:
+    with open(ROOT / "config" / "config.yaml") as fh:
+        return yaml.safe_load(fh)
+
+
+# --- routing profiles -------------------------------------------------------
+
+
+def test_profiles_empty_dict_defaults(raw):
+    """A missing ``profiles`` section defaults to an empty dict."""
+    cfg = copy.deepcopy(raw)
+    cfg.pop("profiles", None)
+    loaded = RouterConfig(**cfg)
+    assert loaded.profiles == {}
+
+
+def test_a_valid_locality_profile_loads(raw):
+    cfg = copy.deepcopy(raw)
+    cfg["profiles"] = {"locality": {"provider": "ollama-local"}}
+    loaded = RouterConfig(**cfg)
+    assert loaded.profiles["locality"].provider == "ollama-local"
+
+
+def test_a_profile_with_all_fields_loads(raw):
+    cfg = copy.deepcopy(raw)
+    cfg["profiles"] = {
+        "onlycheaps": {
+            "min_tier": 1,
+            "max_tier": 2,
+            "latency_tolerance": "batch",
+            "max_cost_per_1m_completion": 0.50,
+            "allowed_model_ids": {"foo", "bar"},
+        }
+    }
+    loaded = RouterConfig(**cfg)
+    profile = loaded.profiles["onlycheaps"]
+    assert profile.min_tier == 1
+    assert profile.max_tier == 2
+    assert profile.latency_tolerance == "batch"
+    assert profile.max_cost_per_1m_completion == 0.50
+    assert profile.allowed_model_ids == {"foo", "bar"}
+
+
+def test_profile_with_unknown_key_is_rejected(raw):
+    cfg = copy.deepcopy(raw)
+    cfg["profiles"] = {"locality": {"provider": "ollama-local", "quality_tolerance": 0.2}}
+    with pytest.raises(ValueError, match="quality_tolerance"):
+        RouterConfig(**cfg)
+
+
+@pytest.mark.parametrize(
+    "tier_key,tier_value",
+    [
+        ("min_tier", 0),
+        ("min_tier", 4),
+        ("max_tier", 0),
+        ("max_tier", 4),
+    ],
+)
+def test_profile_tier_out_of_range_is_rejected(raw, tier_key, tier_value):
+    cfg = copy.deepcopy(raw)
+    cfg["profiles"] = {"bigboybritches": {tier_key: tier_value}}
+    with pytest.raises(ValueError, match="tier"):
+        RouterConfig(**cfg)
+
+
+def test_profile_min_tier_above_max_tier_is_rejected(raw):
+    cfg = copy.deepcopy(raw)
+    cfg["profiles"] = {"bad": {"min_tier": 3, "max_tier": 1}}
+    with pytest.raises(ValueError, match="min_tier.*max_tier"):
+        RouterConfig(**cfg)
+
+
+@pytest.mark.parametrize("latency", ["realtime", "", "BATCH"])
+def test_profile_bad_latency_tolerance_is_rejected(raw, latency):
+    cfg = copy.deepcopy(raw)
+    cfg["profiles"] = {"locality": {"latency_tolerance": latency}}
+    with pytest.raises(ValueError, match="latency_tolerance"):
+        RouterConfig(**cfg)
+
+
+@pytest.mark.parametrize("cost", [0, -1.5])
+def test_profile_nonpositive_cost_is_rejected(raw, cost):
+    cfg = copy.deepcopy(raw)
+    cfg["profiles"] = {"locality": {"max_cost_per_1m_completion": cost}}
+    with pytest.raises(ValueError, match="max_cost_per_1m_completion"):
+        RouterConfig(**cfg)
+
+
+def test_profile_empty_allowed_model_ids_is_rejected(raw):
+    cfg = copy.deepcopy(raw)
+    cfg["profiles"] = {"locality": {"allowed_model_ids": []}}
+    with pytest.raises(ValueError, match="allowed_model_ids"):
+        RouterConfig(**cfg)

tests/test_metrics_endpoint.py

@@ -381,3 +381,51 @@ async def test_decision_event_stream_replays_then_streams_live():
         assert _sse_frame(live)["id"] == 2
     finally:
         events.clear()
+
+
+@pytest.mark.anyio
+async def test_persist_route_decision_publishes_profile_in_sse_payload(
+    tmp_path, monkeypatch,
+):
+    import asyncio
+
+    events.clear()
+    db_path = tmp_path / "profile-sse.db"
+    conn = sqlite3.connect(db_path)
+    conn.executescript(SCHEMA_SQL)
+    conn.close()
+
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
+    monkeypatch.setattr(dispatcher.cfg.logging, "log_route_decisions", True)
+
+    sse_queue: asyncio.Queue[dict] = asyncio.Queue()
+    try:
+        events.subscribe_sse(sse_queue, replay=True)
+
+        dispatcher.persist_route_decision(
+            "route",
+            classification=dispatcher.RouteResponse(
+                classification=dispatcher.Classification(
+                    task_category="coding_general", task_tier=2,
+                    required_context_tokens=100, confidence=0.9,
+                ),
+                latency_tolerance="interactive",
+                profile="batch",
+                selected=dispatcher.Candidate(
+                    model_id="cheap", provider="neuralwatt",
+                    tier=2, latency_class="standard", reasoning_mode="default",
+                    context_variant="full", effective_context_window=128000,
+                    composite=0.5, cost_score=0.5, proficiency_score=0.9,
+                ),
+                candidates_considered=1,
+            ),
+            latency_tolerance="interactive",
+            selected_model="cheap",
+            selected_provider="neuralwatt",
+        )
+
+        decision = await asyncio.wait_for(sse_queue.get(), timeout=2.0)
+        assert decision["profile"] == "batch"
+    finally:
+        events.unsubscribe_sse(sse_queue)
+        events.clear()

tests/test_route_decisions.py

@@ -62,6 +62,7 @@ ROUTE_DECISIONS_COLUMNS = [
     "exploration",
     "pinch_original_tokens",
     "pinch_final_tokens",
+    "profile",
 ]
 
 
@@ -257,6 +258,130 @@ def test_persist_writes_pinch_columns(tmp_path, monkeypatch):
     conn.close()
 
 
+def test_persist_route_decision_writes_profile_from_response(tmp_path, monkeypatch):
+    """A RouteResponse with profile='locality' writes that profile to the row."""
+    db_path = tmp_path / "profile.db"
+    conn = sqlite3.connect(db_path)
+    conn.executescript(SCHEMA_SQL)
+    conn.close()
+
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
+    monkeypatch.setattr(dispatcher.cfg.logging, "log_route_decisions", True)
+
+    from dispatcher import Candidate
+
+    dispatcher.persist_route_decision(
+        "route",
+        classification=dispatcher.RouteResponse(
+            classification=Classification(
+                task_category="coding_general", task_tier=2,
+                required_context_tokens=100, confidence=0.9,
+            ),
+            latency_tolerance="interactive",
+            profile="locality",
+            selected=Candidate(
+                model_id="deepseek-v4-flash", provider="neuralwatt",
+                tier=2, latency_class="standard", reasoning_mode="default",
+                context_variant="full", effective_context_window=128000,
+                composite=0.5, cost_score=0.5, proficiency_score=0.9,
+            ),
+            candidates_considered=1,
+        ),
+        latency_tolerance="interactive",
+        selected_model="deepseek-v4-flash",
+        selected_provider="neuralwatt",
+    )
+
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    row = conn.execute("SELECT profile FROM route_decisions").fetchone()
+    assert row is not None
+    assert row["profile"] == "locality"
+
+    # A bare Classification (no RouteResponse) should leave profile at None.
+    dispatcher.persist_route_decision(
+        "route",
+        classification=Classification(
+            task_category="coding_general", task_tier=2,
+            required_context_tokens=100, confidence=0.9,
+        ),
+        latency_tolerance="interactive",
+        selected_model="deepseek-v4-flash",
+        selected_provider="neuralwatt",
+    )
+    rows = conn.execute("SELECT profile FROM route_decisions ORDER BY id").fetchall()
+    assert len(rows) == 2
+    assert rows[1]["profile"] is None
+    conn.close()
+
+
+def _schema_minus_profile_column() -> str:
+    """schema.sql with the `profile` column and its comments stripped from
+    route_decisions, yielding the previous-todo schema for migration tests."""
+    lines = []
+    inside_route_decisions = False
+    skip_profile_block = False
+    for line in SCHEMA_SQL.splitlines():
+        stripped = line.strip()
+        if stripped.startswith("CREATE TABLE IF NOT EXISTS route_decisions"):
+            inside_route_decisions = True
+        if inside_route_decisions:
+            if stripped.startswith("pinch_final_tokens"):
+                lines.append("    pinch_final_tokens       INTEGER")
+                skip_profile_block = True
+                continue
+            if skip_profile_block:
+                if stripped == ");":
+                    skip_profile_block = False
+                    inside_route_decisions = False
+                    lines.append(line)
+                    continue
+                continue
+        lines.append(line)
+    return "\n".join(lines)
+
+
+def test_ensure_route_decisions_migrates_profile_column(tmp_path):
+    """A live table without `profile` gets the column once, idempotently."""
+    conn = sqlite3.connect(tmp_path / "migrate.db")
+    conn.executescript(_schema_minus_profile_column())
+    assert "profile" not in {
+        r[1] for r in conn.execute("PRAGMA table_info(route_decisions)")
+    }
+
+    dispatcher.ensure_route_decisions(conn)
+    assert "profile" in {
+        r[1] for r in conn.execute("PRAGMA table_info(route_decisions)")
+    }
+
+    # Second call must be a no-op and not duplicate the column.
+    dispatcher.ensure_route_decisions(conn)
+    cols = [r[1] for r in conn.execute("PRAGMA table_info(route_decisions)")]
+    assert cols.count("profile") == 1
+
+    # The new column accepts NULL and string values.
+    conn.execute(
+        """
+        INSERT INTO route_decisions (
+            observed_at, kind, profile
+        ) VALUES (?, ?, ?)
+        """,
+        ("2026-01-01T00:00:00+00:00", "route", "locality"),
+    )
+    conn.execute(
+        """
+        INSERT INTO route_decisions (
+            observed_at, kind
+        ) VALUES (?, ?)
+        """,
+        ("2026-01-01T00:00:00+00:00", "route"),
+    )
+    conn.commit()
+    profiles = [r[0] for r in conn.execute("SELECT profile FROM route_decisions ORDER BY id")]
+    assert profiles == ["locality", None]
+    conn.close()
+
+
 def test_persist_ensure_on_write_fixes_live_db_missing_table(tmp_path, monkeypatch):
     """A live router.db without route_decisions gets it on the WRITE path.
 
@@ -549,6 +674,58 @@ def test_route_endpoint_persists_one_row(decision_router):
     assert "session_dir" not in r.keys()
 
 
+def test_route_endpoint_returns_named_profile(decision_router):
+    client, db_path = decision_router
+    resp = client.post(
+        "/route", json={"task": "write me a function", "profile": "locality"}
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["profile"] == "locality"
+
+    rows = _rows(db_path)
+    assert len(rows) == 1
+    assert rows[0]["profile"] == "locality"
+    assert rows[0]["kind"] == "route"
+
+
+def test_route_endpoint_unknown_profile_returns_422(decision_router):
+    client, db_path = decision_router
+    resp = client.post(
+        "/route", json={"task": "write me a function", "profile": "nosuchprofile"}
+    )
+    assert resp.status_code == 422
+    detail = resp.json()["detail"]
+    assert "nosuchprofile" in detail
+    assert "default" in detail
+    assert "batch" in detail
+
+    rows = _rows(db_path)
+    assert len(rows) == 0
+
+
+def test_dispatch_endpoint_returns_named_profile(decision_router):
+    client, db_path = decision_router
+    resp = client.post(
+        "/dispatch",
+        json={
+            "task": "write me a function",
+            "profile": "batch",
+            "task_category": "coding_general",
+            "task_tier": 2,
+            "required_context_tokens": 100,
+        },
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["route"]["profile"] == "batch"
+
+    rows = _rows(db_path)
+    assert len(rows) == 1
+    assert rows[0]["profile"] == "batch"
+    assert rows[0]["kind"] == "dispatch"
+
+
 def test_route_endpoint_override_has_classifier_ms_null(decision_router):
     client, db_path = decision_router
     resp = client.post(
@@ -605,6 +782,7 @@ def test_routed_chat_persists_one_row(decision_router):
     assert r["selected_provider"] == "neuralwatt"
     assert r["classification_source"] == "classifier"
     assert r["session_key"] is not None
+    assert r["profile"] == "default"
     # The session key is a hash — never a directory, never content.
     assert len(r["session_key"]) == 16
     assert "/" not in (r["session_key"] or "")
@@ -640,6 +818,7 @@ def test_streamed_routed_chat_persists_one_row(decision_router):
     assert rows[0]["kind"] == "chat"
     assert rows[0]["selected_model"] == CHEAP
     assert rows[0]["streamed"] == 1
+    assert rows[0]["profile"] == "default"
 
 
 def test_streamed_routed_chat_writes_request_id_back(decision_router):
@@ -674,6 +853,7 @@ def test_passthrough_persists_one_row_with_no_nameerror(decision_router):
     assert r["selected_provider"] == "neuralwatt"
     assert r["classification_source"] is None
     assert r["session_key"] is not None
+    assert r["profile"] is None, "non-routed passthrough must write profile=None"
 
 
 def test_passthrough_records_pinch_columns_when_pruned(decision_router, monkeypatch):
@@ -751,6 +931,7 @@ def test_local_vision_success_persists_one_local_row(decision_router, monkeypatc
     assert r["selected_provider"] == "local"
     assert r["rejected_reason"] is None
     assert r["images"] == 1
+    assert r["profile"] is None, "non-routed local_vision must write profile=None"
 
 
 def test_no_candidate_422_still_persists_a_rejection_row(decision_router):
@@ -770,6 +951,7 @@ def test_no_candidate_422_still_persists_a_rejection_row(decision_router):
     assert r["selected_model"] is None
     assert r["rejected_reason"] is not None
     assert "vision" in r["rejected_reason"]
+    assert r["profile"] == "default"
 
 
 # --- failure modes: best-effort, config-gated -------------------------------
@@ -1098,7 +1280,7 @@ def test_exploration_disabled_keeps_winner_and_zero_flag(decision_router, monkey
 
 
 def test_exploration_max_tier_excludes_tier_three(decision_router, monkeypatch):
-    client, db_path = decision_router
+    _, db_path = decision_router
     _seed_proficiency_outcomes(db_path, {CHEAP: 50, EXPLORABLE_DEAR: 0})
     monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", True)
     monkeypatch.setattr(dispatcher.cfg.exploration, "epsilon", 1.0)

tests/test_routing.py

@@ -890,6 +890,71 @@ def test_category_ineligible_reason_is_single_token():
     assert " " not in reason
 
 
+# --- restrict_to profile allowlist ----------------------------------------
+
+
+def test_restrict_to_none_is_unrestricted():
+    row = _row(model_id="m1")
+    assert _eligible(row, restrict_to=None) is True
+
+
+def test_restrict_to_allows_model_in_set():
+    row = _row(model_id="m1")
+    assert _eligible(row, restrict_to={"m1"}) is True
+
+
+def test_restrict_to_excludes_model_not_in_set():
+    row = _row(model_id="m2")
+    assert _eligible(row, restrict_to={"m1"}) is False
+
+
+def test_restrict_to_empty_set_excludes_every_model():
+    row = _row(model_id="m1")
+    assert _eligible(row, restrict_to=set()) is False
+
+
+def test_restrict_to_reason_is_profile_excluded():
+    row = _row(model_id="m2")
+    reason = _reason(row, restrict_to={"m1"})
+    assert reason == "profile_excluded"
+
+
+def test_restrict_to_is_single_token_reason():
+    row = _row(model_id="m2")
+    reason = _reason(row, restrict_to={"m1"})
+    assert " " not in reason
+
+
+def test_select_candidates_applies_restrict_to_allowlist():
+    rows = [_row(model_id="m1"), _row(model_id="m2"), _row(model_id="m3")]
+    selected = select_candidates(
+        rows, required_context_tokens=10_000, required_tier=2,
+        latency_tolerance="interactive", allowed_access_levels=["public"],
+        exclude_stale=True, exclude_deprecated=True, restrict_to={"m1"})
+    assert [r["model_id"] for r in selected] == ["m1"]
+
+
+def test_select_candidates_empty_restrict_to_yields_no_candidates():
+    rows = [_row(model_id="m1"), _row(model_id="m2")]
+    selected = select_candidates(
+        rows, required_context_tokens=10_000, required_tier=2,
+        latency_tolerance="interactive", allowed_access_levels=["public"],
+        exclude_stale=True, exclude_deprecated=True, restrict_to=set())
+    assert selected == []
+
+
+def test_restrict_to_and_exclude_models_compose_as_and():
+    # An allowlist of {m1, m2} plus a circuit-open exclusion of m1 should
+    # leave only m2.
+    rows = [_row(model_id="m1"), _row(model_id="m2"), _row(model_id="m3")]
+    selected = select_candidates(
+        rows, required_context_tokens=10_000, required_tier=2,
+        latency_tolerance="interactive", allowed_access_levels=["public"],
+        exclude_stale=True, exclude_deprecated=True,
+        exclude_models={"m1"}, restrict_to={"m1", "m2"})
+    assert [r["model_id"] for r in selected] == ["m2"]
+
+
 def test_select_candidates_drops_outside_category():
     # Even a perfect row (tier 1, zero cost) is dropped for the wrong category.
     rows = [

tests/test_routing_profiles_integration.py

@@ -0,0 +1,756 @@
+"""Integration pins for the five preserved behaviors of named routing profiles.
+
+This file covers Todo 7 of the named-routing-profiles plan. These tests cut
+across the dispatcher / routing boundary: `route` is invoked directly with a
+real routing profile, against a monkeypatched SQLite catalog in a temp file.
+No classifier, provider, or local model is ever called.
+
+The five planks:
+
+(a) ``auto`` and ``auto:batch`` behavior is byte-identical to the old ternary:
+    route() with the batch profile must return the same selection and
+    runner-up ordering as the default profile with request
+    ``latency_tolerance=batch``.
+(b) Unknown profiles return 422 naming valid profiles. Coverage for
+    ``/v1/chat/completions`` unknown profiles exists in
+    tests/test_chat_completions.py; we do not duplicate it here.
+(c) Empty profile allowlist fails with a profile-specific reason
+    ("profile_excluded") and the reason is persisted to route_decisions
+    when no candidate survives.
+(d) Profiles AND with circuit-breaker exclusions, admin overrides, and
+    ``eligible_categories``.
+(e) Exploration picks only from profile-filtered candidates.
+"""
+
+from __future__ import annotations
+
+import json
+import random
+import sqlite3
+from pathlib import Path
+from typing import Optional
+
+import pytest
+from starlette.testclient import TestClient
+
+import circuit_breaker
+import dispatcher
+from config import RoutingProfile
+from dispatcher import (
+    BUILTIN_PROFILES,
+    TaskRequest,
+    app,
+    persist_route_decision,
+    route,
+)
+from routing import rejection_reason, select_candidates
+
+ROOT = Path(__file__).resolve().parent.parent
+SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()
+ADMIN_SCHEMA_SQL = (ROOT / "config" / "admin_schema.sql").read_text()
+
+M1 = "m1-cloud"
+M1_FLEX = "m1-cloud-flex"
+M2 = "m2-cloud"
+LOCAL = "m3-local"
+LOCAL2 = "m3b-local"
+
+
+def _model_row(
+    model_id: str,
+    *,
+    provider: str = "neuralwatt",
+    tier: int = 2,
+    latency_class: str = "standard",
+    completion_price: float = 1.0,
+    prompt_price: Optional[float] = None,
+    context_window: int = 262128,
+    effective_context_window: int = 192500,
+    supports_vision: int = 0,
+    supports_json_mode: int = 1,
+    eligible_categories: Optional[str] = None,
+) -> dict:
+    """A models-table row, shaped for the fixture's INSERT."""
+    return {
+        "model_id": model_id,
+        "provider": provider,
+        "base_model_id": model_id,
+        "tier": tier,
+        "context_window": context_window,
+        "effective_context_window": effective_context_window,
+        "max_output_tokens": 16384,
+        "cost_per_1m_prompt": (
+            prompt_price if prompt_price is not None else completion_price / 3
+        ),
+        "cost_per_1m_completion": completion_price,
+        "cost_per_1m_prompt_cached": (
+            prompt_price if prompt_price is not None else completion_price / 3
+        ),
+        "supports_vision": supports_vision,
+        "supports_json_mode": supports_json_mode,
+        "latency_class": latency_class,
+        "reasoning_mode": "default",
+        "context_variant": "full",
+        "access_level": "public",
+        "availability": "active",
+        "eligible_categories": eligible_categories,
+        "last_updated": "2026-08-22T00:00:00+00:00",
+    }
+
+
+def _insert_model(conn: sqlite3.Connection, row: dict) -> None:
+    conn.execute(
+        """
+        INSERT INTO models (
+            model_id, provider, base_model_id, tier, context_window,
+            effective_context_window, max_output_tokens,
+            cost_per_1m_prompt, cost_per_1m_completion, cost_per_1m_prompt_cached,
+            supports_vision, supports_json_mode,
+            latency_class, reasoning_mode, context_variant,
+            access_level, availability, eligible_categories, last_updated
+        ) VALUES (
+            :model_id, :provider, :base_model_id, :tier, :context_window,
+            :effective_context_window, :max_output_tokens,
+            :cost_per_1m_prompt, :cost_per_1m_completion, :cost_per_1m_prompt_cached,
+            :supports_vision, :supports_json_mode,
+            :latency_class, :reasoning_mode, :context_variant,
+            :access_level, :availability, :eligible_categories, :last_updated
+        )
+        """,
+        row,
+    )
+
+
+def _insert_proficiency(
+    conn: sqlite3.Connection,
+    model_id: str,
+    provider: str,
+    *,
+    blended_score: float,
+    outcome_samples: int = 0,
+) -> None:
+    conn.execute(
+        """
+        INSERT INTO proficiency (
+            model_id, provider, category, blended_score, source,
+            outcome_samples, last_updated
+        ) VALUES (?, ?, 'coding_general', ?, 'leaderboard', ?,
+                  '2026-08-22T00:00:00+00:00')
+        """,
+        (model_id, provider, blended_score, outcome_samples),
+    )
+
+
+class FakeResponse:
+    """Just enough of requests.Response for the stubbed provider POST."""
+
+    def __init__(self, payload=None):
+        self.status_code = 200
+        self._payload = payload or {}
+        self.text = json.dumps(self._payload)
+        self.headers = {}
+        self.closed = False
+
+    def json(self):
+        return self._payload
+
+    def iter_lines(self, decode_unicode=False):
+        return iter([])
+
+    def close(self):
+        self.closed = True
+
+
+@pytest.fixture
+def profile_router(tmp_path, monkeypatch):
+    """The dispatcher pointed at a curated throwaway catalog, nothing dialled out.
+
+    Candidate economics are chosen so the routing outcome is fully determined:
+
+    - M1 / M1_FLEX tie on proficiency (0.9); the flex twin is cheaper, so under
+      batch latency it outranks standard, and under interactive it is filtered
+      out entirely by the latency_class hard filter.
+    - In exploration tests M2 has zero outcome samples, so an epsilon=1.0 run
+      with the default profile picks it (cloud), while the locality profile's
+      candidate set contains only the two ollama-local rows.
+    """
+    db_path = tmp_path / "profiles.db"
+    conn = sqlite3.connect(db_path)
+    conn.executescript(SCHEMA_SQL)
+    conn.executescript(ADMIN_SCHEMA_SQL)
+
+    _insert_model(conn, _model_row(M1, completion_price=3.0))
+    _insert_model(
+        conn, _model_row(M1_FLEX, completion_price=2.0, latency_class="flex")
+    )
+    _insert_model(conn, _model_row(M2, completion_price=9.0))
+    _insert_model(
+        conn,
+        _model_row(
+            LOCAL, provider="ollama-local", completion_price=5.0, prompt_price=1.0
+        ),
+    )
+    _insert_model(
+        conn,
+        _model_row(
+            LOCAL2, provider="ollama-local", completion_price=1.0, prompt_price=0.33
+        ),
+    )
+
+    _insert_proficiency(conn, M1, "neuralwatt", blended_score=0.90, outcome_samples=50)
+    _insert_proficiency(conn, M1_FLEX, "neuralwatt", blended_score=0.90, outcome_samples=50)
+    _insert_proficiency(conn, M2, "neuralwatt", blended_score=0.85, outcome_samples=0)
+    _insert_proficiency(conn, LOCAL, "ollama-local", blended_score=0.88, outcome_samples=50)
+    # 0.70 sits a full quality-tolerance band below the 0.88/0.90 rows
+    # (0.9 - 0.8 would land there only by float luck: 0.099999... rounds to
+    # band 0). Band 1 keeps LOCAL2 eligible-but-ranked-last under the default
+    # profile while still being the cheapest local row under locality.
+    _insert_proficiency(conn, LOCAL2, "ollama-local", blended_score=0.70, outcome_samples=50)
+
+    conn.commit()
+    conn.close()
+
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
+    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.session_cache, "enabled", False)
+    # Exploration is pinned off for determinism; the exploration tests
+    # re-enable it explicitly.
+    monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.exploration, "epsilon", 0.0)
+    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
+
+    calls = []
+
+    def fake_post(url, headers=None, json=None, stream=False, timeout=None):
+        calls.append({"url": url, "body": json, "stream": stream})
+        return FakeResponse(
+            {
+                "id": "chatcmpl-test-1",
+                "model": json["model"],
+                "choices": [
+                    {"message": {"role": "assistant", "content": "ok"},
+                     "finish_reason": "stop"}
+                ],
+                "usage": {"prompt_tokens": 31, "completion_tokens": 12},
+                "energy": {"energy_kwh": 5e-05},
+                "cost": {"request_cost_usd": 4e-04},
+            }
+        )
+
+    monkeypatch.setattr(dispatcher.requests, "post", fake_post)
+    monkeypatch.setattr(
+        dispatcher, "classify",
+        lambda task, context: dispatcher.Classification(
+            task_category="coding_general", task_tier=2,
+            required_context_tokens=100, confidence=0.9,
+        ),
+    )
+
+    yield TestClient(app), db_path, calls
+
+    circuit_breaker.clear()
+
+
+@pytest.fixture
+def logbuf():
+    """Capture what would reach the journal, through the real formatter."""
+    import io
+
+    import logs
+
+    buf = io.StringIO()
+    logs.configure("debug", stream=buf, journald=False)
+    yield buf
+    for handler in list(logs.log.handlers):
+        logs.log.removeHandler(handler)
+
+
+def _route_req(
+    *,
+    latency_tolerance: Optional[str] = None,
+    task_category: str = "coding_general",
+):
+    return TaskRequest(
+        task="write me a function",
+        task_category=task_category,
+        task_tier=2,
+        required_context_tokens=100,
+        latency_tolerance=latency_tolerance,
+    )
+
+
+def _hard_filters(exclude_models: set[str] = frozenset()) -> dict:
+    """The hard-filter kwargs every direct rejection_reason/select_candidates
+    call in this file shares, matching what route() itself passes."""
+    return dict(
+        required_context_tokens=100,
+        required_tier=2,
+        latency_tolerance="interactive",
+        allowed_access_levels=["public"],
+        exclude_stale=True,
+        exclude_deprecated=True,
+        exclude_models=exclude_models,
+    )
+
+
+# =============================================================================
+# Plank (a): auto and auto:batch behavior is byte-identical to the old ternary.
+# =============================================================================
+
+
+def test_auto_batch_matches_default_with_batch_latency(profile_router):
+    """route() under the batch profile must equal the default profile plus
+    request latency_tolerance=batch, byte for byte.
+
+    The batch profile's only field is latency_tolerance, and a request-level
+    batch tolerance on the default profile reaches the same effective latency;
+    both must produce the same selection, runner-up ordering, and candidate
+    count. A leftover special-case ternary that treated ``auto:batch``
+    differently from an explicit batch request would break this.
+    """
+    _, _, _ = profile_router
+
+    via_request = route(
+        _route_req(latency_tolerance="batch"),
+        profile="default",
+        profile_obj=BUILTIN_PROFILES["default"],
+    )
+    via_profile = route(
+        _route_req(latency_tolerance="batch"),
+        profile="batch",
+        profile_obj=BUILTIN_PROFILES["batch"],
+    )
+
+    assert via_request.selected is not None
+    assert via_profile.selected is not None
+    assert via_request.selected.model_id == via_profile.selected.model_id
+    assert [r.model_id for r in via_request.runners_up] == [
+        r.model_id for r in via_profile.runners_up
+    ]
+    assert via_request.candidates_considered == via_profile.candidates_considered
+    assert via_request.latency_tolerance == via_profile.latency_tolerance == "batch"
+    # The flex twin is cheaper at equal proficiency, so batch must admit it and
+    # rank it first: this is what makes the batch equivalence observable.
+    assert via_profile.selected.model_id == M1_FLEX
+    assert via_profile.selected.latency_class == "flex"
+
+
+def test_interactive_default_excludes_flex_rows_batch_admits_them(profile_router):
+    """Without batch tolerance, the flex twin is latency-filtered out; the
+    selection therefore differs, proving the flex scenario is live."""
+    _, _, _ = profile_router
+
+    interactive = route(
+        _route_req(), profile="default", profile_obj=BUILTIN_PROFILES["default"]
+    )
+    batch = route(
+        _route_req(latency_tolerance="batch"),
+        profile="batch",
+        profile_obj=BUILTIN_PROFILES["batch"],
+    )
+
+    assert interactive.selected is not None
+    assert batch.selected is not None
+    assert interactive.selected.model_id == M1
+    assert interactive.selected.latency_class == "standard"
+    assert batch.selected.model_id == M1_FLEX
+    assert batch.selected.latency_class == "flex"
+    # The flex twin only enters the candidate set under batch tolerance.
+    assert interactive.candidates_considered < batch.candidates_considered
+
+
+def test_chat_auto_batch_selects_same_model_as_auto_under_batch_default(
+    profile_router, monkeypatch
+):
+    """``POST /v1/chat/completions`` with model=auto:batch returns the same
+    selected model as model=auto when the default profile defaults to batch.
+
+    The default profile is overlaid with latency_tolerance=batch via
+    cfg.profiles (the configured-override merge path), so ``auto`` and
+    ``auto:batch`` must both pick the flex twin. If ``auto:batch`` were still
+    handled by an old leftover ternary ignoring the profile layer, the two
+    requests would disagree here.
+    """
+    client, _, _ = profile_router
+    monkeypatch.setitem(
+        dispatcher.cfg.profiles, "default", RoutingProfile(latency_tolerance="batch")
+    )
+
+    auto = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto", "messages": [{"role": "user", "content": "hi"}]},
+    )
+    auto_batch = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto:batch", "messages": [{"role": "user", "content": "hi"}]},
+    )
+
+    assert auto.status_code == 200
+    assert auto_batch.status_code == 200
+    assert auto.headers["X-Router-Model"] == M1_FLEX
+    assert auto_batch.headers["X-Router-Model"] == M1_FLEX
+    assert auto.json()["model"] == auto_batch.json()["model"] == M1_FLEX
+
+
+# =============================================================================
+# Plank (b): Unknown profile returns 422 naming valid profiles.
+# =============================================================================
+
+
+def test_route_unknown_profile_returns_422_naming_valid_profiles(profile_router):
+    """A misspelled bare profile on /route must name the valid choices."""
+    client, _, _ = profile_router
+    resp = client.post("/route", json={"task": "x", "profile": "nosuchprofile"})
+    assert resp.status_code == 422
+    detail = resp.json()["detail"]
+    assert "nosuchprofile" in detail
+    assert "default" in detail
+    assert "batch" in detail
+    assert "locality" in detail
+
+
+# ``/v1/chat/completions`` unknown-profile coverage is NOT duplicated here.
+# See tests/test_chat_completions.py::test_unknown_profile_returns_422_naming_valid_profiles
+# for the auto:<unknown> 422 naming default/batch, added in Todos 3/4.
+
+
+# =============================================================================
+# Plank (c): Empty profile allowlist fails with a profile-specific reason.
+# =============================================================================
+
+
+def test_ghost_allowlist_selects_nothing_and_persists_profile_excluded(profile_router):
+    """A profile whose allowed_model_ids intersect no catalog row selects
+    nothing, debug-logs ``profile_excluded`` for every candidate, and can
+    persist that reason through the persist_route_decision seam."""
+    _, db_path, _ = profile_router
+
+    profile = RoutingProfile(allowed_model_ids={"nonexistent-model"})
+    decision = route(_route_req(), profile="ghost", profile_obj=profile)
+
+    assert decision.selected is None
+    assert decision.candidates_considered == 0
+    assert decision.profile == "ghost"
+
+    # The allowlist resolves to an empty restrict_to set: nothing intersects.
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    rows = dispatcher.load_candidates(conn, "coding_general")
+    admin_deprecated = dispatcher._admin_deprecated_models(conn)
+    conn.close()
+    restrict = dispatcher._restrict_to_from_profile(profile, rows)
+    assert restrict == set()
+    assert admin_deprecated == set()
+
+    # Every catalog row carries a profile-era rejection reason. The flex twin
+    # is caught one filter earlier (latency precedes the profile check in the
+    # documented rejection_reason ordering); everything else lands on the
+    # profile-specific reason.
+    for row in rows:
+        reason = rejection_reason(
+            row, restrict_to=restrict, **_hard_filters()
+        )
+        if row["model_id"] == M1_FLEX:
+            assert reason == "latency_class(flex)"
+        else:
+            assert reason == "profile_excluded"
+
+    # And that reason, persisted through the production write path, lands on
+    # the route_decisions row for the no-candidate decision.
+    persist_route_decision(
+        "route", classification=decision, rejected_reason="profile_excluded"
+    )
+
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    row = conn.execute(
+        "SELECT rejected_reason, selected_model, profile FROM route_decisions "
+        "ORDER BY id DESC LIMIT 1"
+    ).fetchone()
+    conn.close()
+    assert row is not None
+    assert row["selected_model"] is None
+    assert row["profile"] == "ghost"
+    assert row["rejected_reason"] is not None
+    assert row["rejected_reason"].startswith("profile_excluded")
+
+
+def test_debug_filter_lines_exist_for_profile_exclusions(profile_router, logbuf):
+    """Every catalog row dropped by the allowlist gets a ``filter`` line.
+
+    Known limitation, pinned as-is: route()'s debug loop calls
+    ``rejection_reason(row, **filters)`` WITHOUT ``restrict_to``, so the
+    logged reason for a profile-excluded row is the placeholder ``-`` rather
+    than ``profile_excluded``. The authoritative reason (via
+    ``rejection_reason(..., restrict_to=...)``) is pinned in the ghost
+    test above; this test pins that the dropped rows at least surface in
+    the debug log at all. A fix that threads restrict_to into the loop
+    should update the ``== "-"`` expectations to ``== "profile_excluded"``.
+    """
+    _, _, _ = profile_router
+    route(
+        _route_req(),
+        profile="ghost",
+        profile_obj=RoutingProfile(allowed_model_ids={"nonexistent-model"}),
+    )
+
+    reasons: dict[str, str] = {}
+    for line in logbuf.getvalue().splitlines():
+        if line.startswith("filter "):
+            fields = dict(
+                token.split("=", 1) for token in line.split(" ")[1:] if "=" in token
+            )
+            reasons[fields.get("model", "")] = fields.get("reason", "")
+
+    # Nothing survived, so every catalog row produced a filter line.
+    for model_id in (M1, M1_FLEX, M2, LOCAL, LOCAL2):
+        assert model_id in reasons, model_id
+    # The flex row is dropped by the latency filter (reason visible), the
+    # rest by the allowlist (reason currently '-').
+    assert reasons[M1_FLEX] == "latency_class(flex)"
+    for model_id in (M1, M2, LOCAL, LOCAL2):
+        assert reasons[model_id] == "-"
+
+
+def test_restriction_set_passes_rows_inside_the_allowlist(profile_router):
+    """The same restrict_to set keeps rows it names — the set is precise, not
+    an unconditional kill switch."""
+    _, db_path, _ = profile_router
+
+    profile = RoutingProfile(allowed_model_ids={M1, M2})
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    rows = dispatcher.load_candidates(conn, "coding_general")
+    conn.close()
+
+    restrict = dispatcher._restrict_to_from_profile(profile, rows)
+    assert restrict == {M1, M2}
+
+    survivors = select_candidates(rows, restrict_to=restrict, **_hard_filters())
+    assert {r["model_id"] for r in survivors} == {M1, M2}
+
+
+# =============================================================================
+# Plank (d): Profiles AND with circuit-breaker, admin overrides,
+#            and eligible_categories.
+# =============================================================================
+
+
+def test_profile_excludes_circuit_open_model(profile_router, monkeypatch):
+    """A profile that allows only m1 must still drop it when the circuit
+    breaker has it open, and the rejection reason is the exclusion reason."""
+    _, _, _ = profile_router
+
+    monkeypatch.setattr(dispatcher.cfg.circuit_breaker, "enabled", True)
+    circuit_breaker.record_failure(
+        M1,
+        "neuralwatt",
+        initial_cooldown=dispatcher.cfg.circuit_breaker.initial_cooldown_seconds,
+        max_cooldown=dispatcher.cfg.circuit_breaker.max_cooldown_seconds,
+        backoff_multiplier=dispatcher.cfg.circuit_breaker.backoff_multiplier,
+    )
+
+    profile = RoutingProfile(allowed_model_ids={M1})
+    decision = route(_route_req(), profile="m1-only", profile_obj=profile)
+    assert decision.selected is None
+    assert decision.candidates_considered == 0
+
+    # The composition is visible at the primitive level too: the exclude_models
+    # set (circuit-open rows) drops m1 even though the profile admits it.
+    row = _model_row(M1)
+    assert (
+        rejection_reason(row, restrict_to={M1}, **_hard_filters(exclude_models={M1}))
+        == "circuit_open"
+    )
+
+
+def test_profile_excludes_admin_deprecated_model(profile_router):
+    """A profile that allows only m1 must drop it when an admin override
+    deprecates it; the survivor set is empty."""
+    _, db_path, _ = profile_router
+
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        "INSERT INTO admin_model_overrides "
+        "(model_id, provider, availability, reason, updated_at) "
+        "VALUES (?, 'neuralwatt', 'deprecated', 'test', "
+        "'2026-08-22T00:00:00+00:00')",
+        (M1,),
+    )
+    conn.commit()
+    conn.close()
+
+    profile = RoutingProfile(allowed_model_ids={M1})
+    decision = route(_route_req(), profile="m1-only", profile_obj=profile)
+
+    assert decision.selected is None
+    assert decision.candidates_considered == 0
+
+
+def test_profile_excludes_category_ineligible_model(profile_router):
+    """A profile that allows only m1 must drop it when m1's
+    eligible_categories excludes the task's category."""
+    _, db_path, _ = profile_router
+
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        "UPDATE models SET eligible_categories = 'coding_refactor' "
+        "WHERE model_id = ?",
+        (M1,),
+    )
+    conn.commit()
+    conn.close()
+
+    profile = RoutingProfile(allowed_model_ids={M1})
+    req = TaskRequest(
+        task="summarize this",
+        task_category="summarization",
+        task_tier=2,
+        required_context_tokens=100,
+    )
+    decision = route(req, profile="m1-only", profile_obj=profile)
+
+    assert decision.selected is None
+    assert decision.candidates_considered == 0
+
+    # And the primitive composes: the category gate fires even though the
+    # profile allowlist names the row.
+    row = _model_row(M1, eligible_categories="coding_refactor")
+    assert (
+        rejection_reason(
+            row, task_category="summarization", restrict_to={M1}, **_hard_filters()
+        )
+        == "category_ineligible"
+    )
+
+
+def test_profile_of_two_models_selects_the_survivor(profile_router):
+    """When the profile allows both m1 and m2 and an admin override kills m1,
+    the other candidate still wins — the filters compose without wiping the
+    whole profile set."""
+    _, db_path, _ = profile_router
+
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        "INSERT INTO admin_model_overrides "
+        "(model_id, provider, availability, reason, updated_at) "
+        "VALUES (?, 'neuralwatt', 'deprecated', 'test', "
+        "'2026-08-22T00:00:00+00:00')",
+        (M1,),
+    )
+    conn.commit()
+    conn.close()
+
+    profile = RoutingProfile(allowed_model_ids={M1, M2})
+    decision = route(_route_req(), profile="m1-or-m2", profile_obj=profile)
+
+    assert decision.selected is not None
+    assert decision.selected.model_id == M2
+    assert decision.candidates_considered == 1
+
+
+# =============================================================================
+# Plank (e): Exploration picks only from profile-filtered candidates.
+# =============================================================================
+
+
+def _force_exploration(monkeypatch, *, seed: int = 0) -> None:
+    monkeypatch.setattr(dispatcher.cfg.exploration, "enabled", True)
+    monkeypatch.setattr(dispatcher.cfg.exploration, "epsilon", 1.0)
+    monkeypatch.setattr(dispatcher.cfg.exploration, "max_tier", 3)
+    # The local rows price above the cloud winner by more than the default
+    # cap of 4.0; widen it so the explore path is reachable at all.
+    monkeypatch.setattr(dispatcher.cfg.exploration, "max_cost_ratio", 1_000.0)
+    monkeypatch.setattr(dispatcher, "_router_rng", random.Random(seed))
+
+
+def test_exploration_never_leaves_the_locality_profile(profile_router, monkeypatch):
+    """With exploration forced on, the locality profile restricts candidates
+    to ollama-local rows, so every explored pick is local — exploration
+    cannot reach a cloud row it never sees."""
+    _, _, _ = profile_router
+    _force_exploration(monkeypatch)
+
+    profile = BUILTIN_PROFILES["locality"]
+    seen: set[tuple[str, str]] = set()
+    explored_any = False
+    for _ in range(30):
+        decision = route(_route_req(), profile="locality", profile_obj=profile)
+        assert decision.selected is not None
+        seen.add((decision.selected.model_id, decision.selected.provider))
+        explored_any = explored_any or decision.explored
+
+    assert seen, "the locality profile must route to something"
+    assert explored_any, "with two local candidates and epsilon=1 the explore path runs"
+    assert {model_id for model_id, _ in seen} <= {LOCAL, LOCAL2}
+    assert all(provider == "ollama-local" for _, provider in seen)
+
+
+def test_exploration_default_profile_can_reach_a_cloud_row(
+    profile_router, monkeypatch
+):
+    """Positive control: same rng, same epsilon, default profile — exploration
+    does pick a cloud row (the zero-sample one). This is what makes the
+    locality assertion above meaningful rather than vacuous."""
+    _, _, _ = profile_router
+    _force_exploration(monkeypatch)
+
+    seen: set[tuple[str, str]] = set()
+    for _ in range(30):
+        decision = route(
+            _route_req(), profile="default", profile_obj=BUILTIN_PROFILES["default"]
+        )
+        assert decision.selected is not None
+        seen.add((decision.selected.model_id, decision.selected.provider))
+
+    assert any(provider == "neuralwatt" for _, provider in seen), seen
+    assert any(model_id == M2 for model_id, _ in seen), (
+        "the zero-sample cloud row is exactly what exploration targets"
+    )
+
+
+def test_exploration_stays_inside_an_allowed_model_ids_profile(
+    profile_router, monkeypatch
+):
+    """An allowed_model_ids profile pins exploration to its allowlist."""
+    _, _, _ = profile_router
+    _force_exploration(monkeypatch)
+
+    allowlist = {M1, M2}
+    profile = RoutingProfile(allowed_model_ids=allowlist)
+    for _ in range(30):
+        decision = route(_route_req(), profile="m1-or-m2", profile_obj=profile)
+        assert decision.selected is not None
+        assert decision.selected.model_id in allowlist
+
+
+# =============================================================================
+# Route endpoint round-trip: profiles resolve and persist end to end.
+# =============================================================================
+
+
+def test_route_endpoint_batch_profile_round_trip(profile_router):
+    """/route with profile=batch stores profile and batch latency on the row."""
+    client, db_path, _ = profile_router
+
+    resp = client.post(
+        "/route", json={"task": "write me a function", "profile": "batch"}
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["profile"] == "batch"
+    assert data["latency_tolerance"] == "batch"
+    assert data["selected"]["model_id"] == M1_FLEX
+
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    row = conn.execute(
+        "SELECT profile, latency_tolerance, selected_model "
+        "FROM route_decisions ORDER BY id DESC LIMIT 1"
+    ).fetchone()
+    conn.close()
+    assert row is not None
+    assert row["profile"] == "batch"
+    assert row["latency_tolerance"] == "batch"
+    assert row["selected_model"] == M1_FLEX