diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml
index 79e479384..ea1dd8394 100644
--- a/.github/workflows/fuzz.yml
+++ b/.github/workflows/fuzz.yml
@@ -87,6 +87,9 @@ jobs:
       - name: Fuzz model-judge response parser
         run: python fuzz/fuzz_model_judge.py -max_total_time="${FUZZ_SECONDS}" -artifact_prefix=crash- fuzz/corpus/judge
 
+      - name: Fuzz reasoning-effort profile parser
+        run: python fuzz/fuzz_reasoning_effort_profile.py -max_total_time="${FUZZ_SECONDS}" -artifact_prefix=crash- fuzz/corpus/reasoning_effort_profile
+
       - name: Upload crash artifacts
         if: failure()
         uses: actions/upload-artifact@330a01c490aca151604b8cf639adc76d48f6c5d4 # actions/upload-artifact@v5
diff --git a/AGENTS.md b/AGENTS.md
index 7aa68585e..dde31e51e 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -123,4 +123,9 @@ push or open a PR.
   cost-optimal LLM routing, upstream load balancing, and latency/throughput
   scheduling (e.g. LLM-cascade / model-routing and queueing/load-balancing
   papers).
+- **Issue #568 slice:** `contextual_orchestrator.reasoning_effort_profile`
+  is the provider-neutral role catalog and equal-budget true-θ ablation.
+  RMSE is computed from θ̂ versus known true parameters, not a rank
+  constant. Do not change production route/conduct defaults until
+  `production_default_change_allowed` is true. Temperature is not effort.
 <!-- END cwl-agent-guidance -->
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f9c44eaa6..16e80f014 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,9 @@
 
 All notable changes to this project are documented in this file.
 
+The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
+and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
 ## [Unreleased]
 
 ### Added
@@ -9,6 +12,24 @@ All notable changes to this project are documented in this file.
 - Structured tool failure categories, stable fallback actions, and public adapter exceptions.
 - Secret-free `tool_fallback_decision` audit events.
 - Exact regression coverage for the Strix `Tool execute_command not found in agent strix` failure.
+- Versioned `reasoning_effort_profile` catalog (issue #568) with fail-closed
+  parse, per-role bindings, replayable snapshot hash, and an equal-budget
+  true-θ RMSE ablation that emits θ̂ and RMSE(θ̂, θ). Sampling temperature
+  is not reasoning effort. Production route/conduct defaults stay locked
+  until `production_default_change_allowed` is true.
+  Next action: run `python -m pytest -q tests/test_reasoning_effort_profile.py` and keep
+  live defaults unchanged while the gate is false. Pass
+  `role_effort_catalog=default_role_effort_catalog()` to attach the same
+  `reasoning_effort_snapshot` on `complete`, `run`, `stream_route`, and
+  `batch_route`; omit it to keep today's payload.
+
+### Fixed
+
+- Reject missing profiles, blank `profile_version`, and fractional seeds.
+  Snapshot hashing now fails closed on extra or missing roles. The
+  production-default gate returns false on junk reports and on
+  `measurement_status=estimated`. Access-list scope is a real ablation
+  factor, not a duplicate label.
 
 ### Changed
 
@@ -21,3 +42,17 @@ All notable changes to this project are documented in this file.
 
 - Ambiguous non-idempotent outcomes, invalid arguments, permission denial, and policy denial fail closed.
 - Fallback errors and audit events do not copy provider exception text, tool arguments, outputs, or credentials; fail-closed exceptions also sever the original cause chain so later traceback logging cannot recover them.
+
+### References
+
+- Sakana AI. (2026). *Sakana Fugu Technical Report*.
+  https://github.com/SakanaAI/fugu/blob/main/Fugu_technical_report.pdf
+- Xu, J., Sun, Q., Schwendeman, P., Nielsen, S., Cetin, E., & Tang, Y. (2025).
+  *Trinity: An evolved LLM coordinator* (arXiv:2512.04695).
+  https://arxiv.org/abs/2512.04695
+- Nielsen, S., Cetin, E., Schwendeman, P., Sun, Q., Xu, J., & Tang, Y. (2025).
+  *Learning to orchestrate agents in natural language with the Conductor*
+  (arXiv:2512.04388). https://arxiv.org/abs/2512.04388
+- Baker, F. B. (2001). *The basics of item response theory* (2nd ed.).
+  ERIC Clearinghouse on Assessment and Evaluation.
+  https://eric.ed.gov/?id=ED458219
diff --git a/CLAUDE.md b/CLAUDE.md
index df04f6be6..bcc1b0f49 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -87,6 +87,7 @@ A stdlib-Python lab implementing a single OpenAI-compatible API that routes, del
 ### Modules (`contextual_orchestrator/`)
 
 - `orchestrator.py` — the domain heart: `ModelAgent`, `WorkflowStep`, `OrchestrationPolicy`, `ModelClient`, `TaskOrchestrator`, secret/PII redaction, budget enforcement, spend analytics, and the commercial-readiness report generators behind `/api/v1/*`. Domain code stays here until a second implementation forces extraction (see `docs/code_conventions.md`).
+- `reasoning_effort_profile.py` — issue #568 role compute catalog and equal-budget true-θ ablation that emits θ̂ and RMSE(θ̂, θ). Production route/conduct defaults stay locked until `production_default_change_allowed`. Buyer next action: `python tests/test_reasoning_effort_profile.py`.
 - `server.py` — HTTP delivery adapter and `SecurityConfig`; all request validation lives here.
 - `admin.py` — static HTML/CSS/JS for the `/admin` operator console (stays inline while the product is dependency-free).
 - `credentials.py` / `kv_config.py` — the KV seam: `get_credential`/`register_credential` over pluggable backends (`InMemoryCredentialBackend` default; pgcrypto-encrypted `PostgresCredentialBackend`, selected via `CONTEXTUAL_ORCHESTRATOR_KV_BACKEND`).
diff --git a/README.md b/README.md
index 531912677..87f3c99d6 100644
--- a/README.md
+++ b/README.md
@@ -308,6 +308,7 @@ python -m pip install --require-hashes -r requirements.lock
 python -m pip install --no-deps -e .
 python tests/test_self_check.py
 python tests/test_paper_contracts.py
+python -m pytest -q tests/test_reasoning_effort_profile.py
 python tests/test_admin_contract.py
 python tests/test_conventions.py
 python tests/test_api_contract.py
diff --git a/conductor/tracks.md b/conductor/tracks.md
index 83fab3043..b377ec28c 100644
--- a/conductor/tracks.md
+++ b/conductor/tracks.md
@@ -4,7 +4,4 @@
 |---|---|---|
 | 001-paper-grounded-orchestrator | active | Implement the source-backed orchestration contract with TDD, DDD, and CDD |
 | 002-enterprise-design-foundation | active | Add paper-grounded screen design, user stories, REST API, code/DB conventions, and i18n |
-| 003-autonomous-pr-ecosystem-loop | active | Standing autonomous review→fix→merge loop across every owned ContextualWisdomLab repo, plus org-ecosystem product-gap work |
-| 003-sdk-omit-real-persist | done | #668's SDK-omit honesty behavior (args/instructions/metadata write-back, chat top_logprobs hoist before tools passthrough) integrated via the http-honesty stack merge (#740) on `integrate/http-honesty-740`, after the smaller #686/#687 re-land attempts went stale and closed unmerged. |
-| 003-compatibility-honesty | done | Fail-closed ASCII `[a-zA-Z0-9_-]{1,64}` on `json_schema.name`/`tool.function.name` integrated via the same #740 merge. |
-| 003-responses-text-format | done | Official Responses text.format structured types integrated via the same #740 merge. |
+| 003-reasoning-effort-profiles | active | Issue #568: versioned per-role `reasoning_effort_profile`, equal-budget θ̂ RMSE ablation, snapshot on run/stream/batch, production defaults locked |
diff --git a/contextual_orchestrator/__init__.py b/contextual_orchestrator/__init__.py
index f236e95cd..2cd686067 100644
--- a/contextual_orchestrator/__init__.py
+++ b/contextual_orchestrator/__init__.py
@@ -38,6 +38,14 @@
 from .credentials import NotConfigured, get_credential, register_credential
 from .kv_config import InMemoryConfigStore, get_config_store
 from .orchestrator import ModelAgent, TaskOrchestrator, WorkflowStep, load_agents
+from .reasoning_effort_profile import (
+    EffortProfileError,
+    ReasoningEffortProfile,
+    apply_request_profile,
+    default_role_effort_catalog,
+    parse_reasoning_effort_profile,
+    snapshot_role_effort_catalog,
+)
 from .token_counting import HeuristicTokenCounter, build_token_counter
 from .tool_fallback import (
     MAX_TOOL_RETRY_ATTEMPTS,
@@ -54,6 +62,12 @@
     "TaskOrchestrator",
     "WorkflowStep",
     "load_agents",
+    "ReasoningEffortProfile",
+    "EffortProfileError",
+    "apply_request_profile",
+    "default_role_effort_catalog",
+    "parse_reasoning_effort_profile",
+    "snapshot_role_effort_catalog",
     "get_credential",
     "register_credential",
     "NotConfigured",
diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py
index 3f0b5e8eb..61b72a650 100644
--- a/contextual_orchestrator/orchestrator.py
+++ b/contextual_orchestrator/orchestrator.py
@@ -43,6 +43,11 @@
     classify_tool_failure,
     downgrade_to_failover,
 )
+from .reasoning_effort_profile import (
+    ReasoningEffortProfile,
+    apply_request_profile,
+    snapshot_role_effort_catalog,
+)
 
 
 # content is usually str; multimodal vision messages use OpenAI content-parts lists.
@@ -162,7 +167,13 @@ def client(self) -> ModelClient:
     def complete(self, messages: list[ChatMessage], mode: str | None = None) -> dict[str, Any]:
         if mode is not None and (type(mode) is not str or mode not in {"auto", "route", "conduct"}):
             raise ValueError("mode must be auto, route, or conduct")
-        output, served_id, usage = self.orchestrator._invoke(self._agent(), messages, text=self.text, role="verifier")
+        output, served_id, usage = self.orchestrator._invoke(
+            self._agent(),
+            messages,
+            text=self.text,
+            role="judge",
+            eligibility_role="verifier",
+        )
         return self._completion_payload(output, served_id, usage, self.mode if mode is None else mode)
 
     def complete_structured(
@@ -178,13 +189,19 @@ def complete_structured(
         if not isinstance(response_format, dict):
             raise TypeError("response_format must be a mapping")
         agent = self._agent()
-        response = self.orchestrator.proxy_completion({
+        request = {
             "model": agent.model,
             "messages": messages,
             "temperature": self.orchestrator.client.temperature,
             "max_tokens": self.orchestrator.client.max_output_tokens,
             "response_format": response_format,
-        })
+        }
+        effort_profile = self.orchestrator._role_effort_profile("judge")
+        response = (
+            self.orchestrator.proxy_completion(request, effort_profile=effort_profile)
+            if effort_profile is not None
+            else self.orchestrator.proxy_completion(request)
+        )
         output = ModelClient._response_content(agent, response)
         usage = response.get("usage") if isinstance(response.get("usage"), dict) else None
         return self._completion_payload(output, agent.id, usage, self.mode if mode is None else mode)
@@ -270,6 +287,9 @@ class ModelAgent:
     # Authorization header scheme, e.g. "Bearer" (OpenAI-compatible default) or
     # "Key" (Bytez). Sent as f"{auth_scheme} {api_key}".
     auth_scheme: str = "Bearer"
+    # ``None`` means provider support is unproven. Opt-in effort profiles then
+    # fail closed unless the profile explicitly requests the safe ``omit`` fallback.
+    reasoning_effort_supported: bool | None = None
 
     def __post_init__(self) -> None:
         require_object_name(self.id, "agent.id")
@@ -279,6 +299,8 @@ def __post_init__(self) -> None:
             raise ValueError("local_credential_key requires a local:// gateway URL")
         if not self.auth_scheme or type(self.auth_scheme) is not str:
             raise ValueError("auth_scheme must be a non-empty string")
+        if self.reasoning_effort_supported not in (None, True, False):
+            raise TypeError("reasoning_effort_supported must be true, false, or null")
 
     def to_config(self) -> dict[str, Any]:
         """Round-trippable agent configuration (from_dict(to_config(a)) == a)."""
@@ -295,6 +317,7 @@ def to_config(self) -> dict[str, Any]:
             "provider_exclusions": list(self.provider_exclusions),
             "local_credential_key": self.local_credential_key,
             "auth_scheme": self.auth_scheme,
+            "reasoning_effort_supported": self.reasoning_effort_supported,
         }
 
     @property
@@ -324,6 +347,7 @@ def from_dict(cls, value: dict[str, Any]) -> "ModelAgent":  # pragma: no cover
             provider_exclusions=tuple(value.get("provider_exclusions", value.get("provider_exclusion", ()))),
             local_credential_key=value.get("local_credential_key", ""),
             auth_scheme=value.get("auth_scheme", "Bearer"),
+            reasoning_effort_supported=value.get("reasoning_effort_supported"),
         )
 
 
@@ -818,6 +842,7 @@ def chat(
         messages: list[ChatMessage],
         temperature: float | None = None,
         top_p: float | None = None,
+        effort_profile: ReasoningEffortProfile | None = None,
     ) -> str:
         """Send messages to a mock or OpenAI-compatible chat endpoint with retries.
 
@@ -861,9 +886,28 @@ def chat(
             payload["frequency_penalty"] = effective_frequency
         if _is_direct_mlx_provider_url(agent.base_url) and self.chat_template_args:
             payload["chat_template_kwargs"] = self.chat_template_args
+        payload = self.apply_effort_profile(agent, payload, effort_profile)
         with _local_provider_slot(agent, self.local_concurrency, self.timeout):
             return self._send_with_retry(agent, payload, destination)
 
+    def apply_effort_profile(
+        self,
+        agent: ModelAgent,
+        payload: dict[str, Any],
+        profile: ReasoningEffortProfile | None,
+    ) -> dict[str, Any]:
+        """Apply an opt-in profile while proving provider support before egress."""
+        supports = (
+            agent.reasoning_effort_supported is True
+            or (agent.reasoning_effort_supported is None and agent.base_url.startswith("mock://"))
+        )
+        return apply_request_profile(
+            payload,
+            profile,
+            supports_reasoning_effort=supports,
+            default_max_output_tokens=self.max_output_tokens,
+        )
+
     def probe(self, agent: ModelAgent, *, timeout: float = DEFAULT_PROVIDER_PROBE_TIMEOUT) -> dict[str, Any]:
         """Verify a local model registry, then run one bounded completion probe.
 
@@ -1115,7 +1159,13 @@ def _open_provider(
             connection.close()
             raise
 
-    def stream_chat(self, agent: ModelAgent, messages: list[ChatMessage], temperature: float | None = None):
+    def stream_chat(
+        self,
+        agent: ModelAgent,
+        messages: list[ChatMessage],
+        temperature: float | None = None,
+        effort_profile: ReasoningEffortProfile | None = None,
+    ):
         """Yield content deltas from a mock or OpenAI-compatible streaming endpoint.
 
         Real token streaming: the provider is called with stream=true and its SSE deltas
@@ -1138,6 +1188,7 @@ def stream_chat(self, agent: ModelAgent, messages: list[ChatMessage], temperatur
         }
         if _is_direct_mlx_provider_url(agent.base_url) and self.chat_template_args:
             payload["chat_template_kwargs"] = self.chat_template_args
+        payload = self.apply_effort_profile(agent, payload, effort_profile)
         with _local_provider_slot(agent, self.local_concurrency, self.timeout):  # pragma: no cover
             yield from self._stream_send(agent, payload, destination)  # pragma: no cover
 
@@ -1368,6 +1419,7 @@ def batch_chat(
         temperature: float | None = None,
         poll_interval: float = 5.0,
         poll_timeout: float = 3600.0,
+        effort_profile: ReasoningEffortProfile | None = None,
     ) -> dict[str, dict[str, Any]]:
         """Run many chat requests through the provider's Batch API and return results by id.
 
@@ -1381,11 +1433,11 @@ def batch_chat(
                 for custom_id, messages in requests.items()
             }
         elif _is_local_provider_url(agent.base_url):
-            results = self._local_batch_chat(agent, requests, temperature)
+            results = self._local_batch_chat(agent, requests, temperature, effort_profile)
         else:
             destination = self._validate_provider(agent)  # pragma: no cover
             results = self._batch_run(  # pragma: no cover
-                agent, requests, temperature, poll_interval, poll_timeout, destination
+                agent, requests, temperature, poll_interval, poll_timeout, destination, effort_profile
             )
         return _validate_batch_results(requests, results)
 
@@ -1394,10 +1446,20 @@ def _local_batch_chat(
         agent: ModelAgent,
         requests: dict[str, list[ChatMessage]],
         temperature: float | None,
+        effort_profile: ReasoningEffortProfile | None,
     ) -> dict[str, dict[str, Any]]:
         """Run local OpenAI-compatible requests concurrently through mlx-lm."""
         def complete(custom_id: str, messages: list[ChatMessage]) -> tuple[str, dict[str, Any]]:
-            content = self.chat(agent, messages, temperature=temperature)
+            content = (
+                self.chat(
+                    agent,
+                    messages,
+                    temperature=temperature,
+                    effort_profile=effort_profile,
+                )
+                if effort_profile is not None
+                else self.chat(agent, messages, temperature=temperature)
+            )
             return custom_id, {"content": content, "usage": self.take_usage()}
 
         if self.local_concurrency == 1 or len(requests) <= 1:
@@ -1410,10 +1472,11 @@ def _batch_run(
         self,
         agent: ModelAgent,
         requests: dict[str, list[ChatMessage]],
-        temperature: float,
+        temperature: float | None,
         poll_interval: float,
         poll_timeout: float,
         destination: ProviderDestination | None = None,
+        effort_profile: ReasoningEffortProfile | None = None,
     ) -> dict[str, dict[str, Any]]:
         """Upload, create, poll, and parse one batch (isolated so the flow stays testable)."""
         lines = [
@@ -1421,12 +1484,11 @@ def _batch_run(
                 "custom_id": custom_id,
                 "method": "POST",
                 "url": "/v1/chat/completions",
-                "body": {
+                "body": self.apply_effort_profile(agent, {
                     "model": agent.model,
                     "messages": messages,
                     "temperature": self.temperature if temperature is None else temperature,
-                    "max_tokens": self.max_output_tokens,
-                },
+                }, effort_profile),
             }, ensure_ascii=False)
             for custom_id, messages in requests.items()
         ]
@@ -1689,6 +1751,7 @@ class TaskOrchestrator:
         "thinker": ("planning", "reasoning", "research"),
         "worker": ("coding", "implementation", "reasoning"),
         "verifier": ("verification", "security", "review", "debugging"),
+        "judge": ("verification", "security", "review", "debugging"),
         "synthesizer": ("writing", "reasoning", "planning"),
         "embedding": ("embedding",),
     }
@@ -1729,6 +1792,7 @@ def __init__(
         cache_max_entries: int = 256,
         tool_retry_attempts: int = 1,
         tool_retry_backoff_seconds: float = 0.25,
+        role_effort_catalog: dict[str, ReasoningEffortProfile] | None = None,
     ) -> None:
         # Optional durable model-group management: stored operator changes overlay the
         # seed agents file at startup (stored rows win by id; stored-new rows append).
@@ -1767,6 +1831,10 @@ def __init__(
         self._tool_retry_sleep = time.sleep
         self._tool_retry_jitter = random.uniform
         self.policy = OrchestrationPolicy()
+        # Opt-in issue #568 catalog. None keeps production answers and payload
+        # keys unchanged. Buyer next action: pass default_role_effort_catalog()
+        # to attach a replayable snapshot; do not treat that as a default change.
+        self.role_effort_catalog = role_effort_catalog
         # Operator-supplied USD price per 1M tokens, keyed by model. Empty => cost not computed.
         self.price_per_million = dict(price_per_million or {})
         # Operator spend caps; None => disabled (no behavior change). Enforced in run().
@@ -1871,7 +1939,11 @@ def _reload_state(self) -> None:
     )
 
     def proxy_completion(
-        self, body: dict[str, Any], *, endpoint: str = "chat/completions"
+        self,
+        body: dict[str, Any],
+        *,
+        endpoint: str = "chat/completions",
+        effort_profile: ReasoningEffortProfile | None = None,
     ) -> dict[str, Any]:
         """Passthrough a full OpenAI request to the primary agent, returning its raw response.
 
@@ -1903,6 +1975,8 @@ def proxy_completion(
             if key not in self._ORCHESTRATION_ONLY_KEYS
         }
         upstream["model"] = agent.model
+        if effort_profile is not None:
+            upstream = self.client.apply_effort_profile(agent, upstream, effort_profile)
         # v1 passthrough returns the full JSON body; SSE stream passthrough is a
         # follow-up, so force a non-streamed upstream response here.
         upstream["stream"] = False
@@ -1951,24 +2025,32 @@ def stream_route(self, messages: list[ChatMessage], workflow_run_id: str | None
         text = self._latest_user_text(messages)
         agent = self._select_agent(text, "worker")
         parts: list[str] = []
-        for delta in self.client.stream_chat(agent, messages):
+        effort_profile = self._role_effort_profile("worker")
+        stream = (
+            self.client.stream_chat(agent, messages, effort_profile=effort_profile)
+            if effort_profile is not None
+            else self.client.stream_chat(agent, messages)
+        )
+        for delta in stream:
             parts.append(delta)
             yield delta
         answer = "".join(parts)
-        record = {
-            "workflow_run_id": workflow_run_id or f"run_{uuid.uuid4().hex}",
-            "created_at": int(time.time()),
-            "mode": "route",
-            "policy_mode": "route",
-            "prompt_text": text,
-            "answer": answer,
-            "trace": [
-                {"id": 0, "role": "worker", "agent_id": agent.id, "subtask": "Direct route (streamed)",
-                 "access": [], "output": answer}
-            ],
-            "policy_snapshot": self.policy.as_dict(),
-            "verification": {"accepted": True, "reason": "single route path", "verifier_output": ""},
-        }
+        record = self._with_effort_snapshot(
+            {
+                "workflow_run_id": workflow_run_id or f"run_{uuid.uuid4().hex}",
+                "created_at": int(time.time()),
+                "mode": "route",
+                "policy_mode": "route",
+                "prompt_text": text,
+                "answer": answer,
+                "trace": [
+                    {"id": 0, "role": "worker", "agent_id": agent.id, "subtask": "Direct route (streamed)",
+                     "access": [], "output": answer}
+                ],
+                "policy_snapshot": self.policy.as_dict(),
+                "verification": {"accepted": True, "reason": "single route path", "verifier_output": ""},
+            }
+        )
         self._workflow_runs[record["workflow_run_id"]] = record
         self._run_order.appendleft(record["workflow_run_id"])
         self._append_audit_event(
@@ -1993,17 +2075,19 @@ def run(self, messages: list[ChatMessage], mode: str = "auto", workflow_run_id:
                 raise BudgetExceededError("spend budget exceeded", detail=budget)
         result = self.complete(messages, mode=mode)
         prompt = self._latest_user_text(messages)
-        record = {
-            "workflow_run_id": workflow_run_id or f"run_{uuid.uuid4().hex}",
-            "created_at": int(time.time()),
-            "mode": result["mode"],
-            "policy_mode": mode,
-            "prompt_text": prompt,
-            "answer": result["answer"],
-            "trace": result["trace"],
-            "policy_snapshot": self.policy.as_dict(),
-            "verification": result.get("verification"),
-        }
+        record = self._with_effort_snapshot(
+            {
+                "workflow_run_id": workflow_run_id or f"run_{uuid.uuid4().hex}",
+                "created_at": int(time.time()),
+                "mode": result["mode"],
+                "policy_mode": mode,
+                "prompt_text": prompt,
+                "answer": result["answer"],
+                "trace": result["trace"],
+                "policy_snapshot": self.policy.as_dict(),
+                "verification": result.get("verification"),
+            }
+        )
         self._workflow_runs[record["workflow_run_id"]] = record
         self._run_order.appendleft(record["workflow_run_id"])
         if self._store is not None:
@@ -2060,10 +2144,15 @@ def batch_route(self, prompts: list[str]) -> list[dict[str, Any]]:
 
         answers: dict[int, dict[str, Any]] = {}
         for agent_id, requests in requests_by_agent.items():
-            results = _validate_batch_results(
-                requests,
-                self.client.batch_chat(agents_by_id[agent_id], requests),
+            effort_profile = self._role_effort_profile("worker")
+            batch = (
+                self.client.batch_chat(
+                    agents_by_id[agent_id], requests, effort_profile=effort_profile
+                )
+                if effort_profile is not None
+                else self.client.batch_chat(agents_by_id[agent_id], requests)
             )
+            results = _validate_batch_results(requests, batch)
             for custom_id, result in results.items():
                 try:
                     prefix, suffix = custom_id.rsplit("_", 1)
@@ -2085,17 +2174,19 @@ def batch_route(self, prompts: list[str]) -> list[dict[str, Any]]:
             }
             if result.get("usage") is not None:
                 row["usage"] = result["usage"]
-            record = {
-                "workflow_run_id": f"run_{uuid.uuid4().hex}",
-                "created_at": int(time.time()),
-                "mode": "route",
-                "policy_mode": "route",
-                "prompt_text": prompt,
-                "answer": result["content"],
-                "trace": [row],
-                "policy_snapshot": self.policy.as_dict(),
-                "verification": {"accepted": True, "reason": "single route path (batched)", "verifier_output": ""},
-            }
+            record = self._with_effort_snapshot(
+                {
+                    "workflow_run_id": f"run_{uuid.uuid4().hex}",
+                    "created_at": int(time.time()),
+                    "mode": "route",
+                    "policy_mode": "route",
+                    "prompt_text": prompt,
+                    "answer": result["content"],
+                    "trace": [row],
+                    "policy_snapshot": self.policy.as_dict(),
+                    "verification": {"accepted": True, "reason": "single route path (batched)", "verifier_output": ""},
+                }
+            )
             self._workflow_runs[record["workflow_run_id"]] = record
             self._run_order.appendleft(record["workflow_run_id"])
             if self._store is not None:
@@ -2426,12 +2517,14 @@ def route_once(self, messages: list[ChatMessage]) -> dict[str, Any]:
         if served_id != agent.id:  # pragma: no cover
             row["served_agent_id"] = served_id
             row["failover_from"] = agent.id
-        return {
-            "mode": "route",
-            "answer": answer,
-            "verification": {"accepted": True, "reason": "single route path", "verifier_output": ""},
-            "trace": [row],
-        }
+        return self._with_effort_snapshot(
+            {
+                "mode": "route",
+                "answer": answer,
+                "verification": {"accepted": True, "reason": "single route path", "verifier_output": ""},
+                "trace": [row],
+            }
+        )
 
     def conduct(self, messages: list[ChatMessage]) -> dict[str, Any]:
         """Run a planned workflow: fixed template, or a Conductor-style generated plan."""
@@ -2502,13 +2595,38 @@ def last_output(role: str) -> str:
             if not verification["accepted"] and self.policy.verifier_required:
                 answer = outputs[steps[1].id]
 
-        return {
-            "mode": "conduct",
-            "answer": answer,
-            "trace": trace,
-            "verification": verification,
-            "plan_source": plan_source,
+        return self._with_effort_snapshot(
+            {
+                "mode": "conduct",
+                "answer": answer,
+                "trace": trace,
+                "verification": verification,
+                "plan_source": plan_source,
+            }
+        )
+
+    def _role_effort_profile(self, role: str) -> ReasoningEffortProfile | None:
+        """Return the opt-in profile bound to one workflow role."""
+        if self.role_effort_catalog is None:
+            return None
+        return self.role_effort_catalog.get(role)
+
+    def _with_effort_snapshot(self, result: dict[str, Any]) -> dict[str, Any]:
+        """Attach a replayable role-effort snapshot when the operator opted in.
+
+        Buyer next action: compare ``reasoning_effort_snapshot.snapshot_hash``
+        on ``complete``, ``run``, ``stream_route``, and ``batch_route``. Omit
+        the constructor catalog to keep today's payload.
+        """
+        if self.role_effort_catalog is None:
+            return result
+        snapshot = snapshot_role_effort_catalog(self.role_effort_catalog)
+        result["reasoning_effort_snapshot"] = {
+            "profile_version": snapshot.profile_version,
+            "snapshot_hash": snapshot.snapshot_hash,
+            "role_profiles": snapshot.role_profiles,
         }
+        return result
 
     def _plan_generated(self, task: str) -> list[WorkflowStep]:
         """Ask the planner model to generate the workflow (Conductor, arXiv:2512.04388).
@@ -2531,10 +2649,16 @@ def _plan_generated(self, task: str) -> list[WorkflowStep]:
             "verifier step when correctness matters.\n"
             f"Available agents:\n{pool}"
         )
-        raw = self.client.chat(planner, [
+        planner_messages = [
             {"role": "system", "content": system},
             {"role": "user", "content": task},
-        ])
+        ]
+        effort_profile = self._role_effort_profile("planner")
+        raw = (
+            self.client.chat(planner, planner_messages, effort_profile=effort_profile)
+            if effort_profile is not None
+            else self.client.chat(planner, planner_messages)
+        )
         return self._parse_workflow_plan(raw)
 
     def _parse_workflow_plan(self, raw: str) -> list[WorkflowStep]:
@@ -2623,7 +2747,13 @@ def select_capability_agent(self, capability: str) -> ModelAgent:
         return ranked[0]
 
     def _invoke(
-        self, primary: ModelAgent, messages: list[ChatMessage], *, text: str, role: str
+        self,
+        primary: ModelAgent,
+        messages: list[ChatMessage],
+        *,
+        text: str,
+        role: str,
+        eligibility_role: str | None = None,
     ) -> tuple[str, str, dict[str, Any] | None]:
         """Call an agent with bounded, safety-aware tool retry and failover.
 
@@ -2631,14 +2761,22 @@ def _invoke(
         agent/tool-runtime failures: missing tools move to a compatible agent,
         explicitly idempotent transient calls retry the same agent, and ambiguous
         side effects or policy/permission/argument errors fail closed.
+
+        ``eligibility_role`` keeps operator exclusions tied to the role used to
+        select the primary when the call's effort profile has a distinct name.
         """
-        candidates = self._failover_candidates(primary, text, role)
+        candidates = self._failover_candidates(primary, text, eligibility_role or role)
         retry_limit = min(self.tool_retry_attempts, MAX_TOOL_RETRY_ATTEMPTS)
         for agent in candidates:
             retry_attempt = 0
             while True:
                 try:
-                    output = self.client.chat(agent, messages)
+                    effort_profile = self._role_effort_profile(role)
+                    output = (
+                        self.client.chat(agent, messages, effort_profile=effort_profile)
+                        if effort_profile is not None
+                        else self.client.chat(agent, messages)
+                    )
                 except Exception as exc:
                     if isinstance(exc, (ProviderResponseError, ToolFallbackStoppedError)):
                         raise
diff --git a/contextual_orchestrator/reasoning_effort_profile.py b/contextual_orchestrator/reasoning_effort_profile.py
new file mode 100644
index 000000000..3058312c4
--- /dev/null
+++ b/contextual_orchestrator/reasoning_effort_profile.py
@@ -0,0 +1,555 @@
+"""Provider-neutral reasoning-effort profiles and equal-budget ablation.
+
+Issue #568: each TRINITY/Conductor workflow role gets an explicit
+``reasoning_effort_profile``. Sampling temperature, top-p, and seed stay
+independent fields. Production routing defaults stay locked until an
+equal-budget ablation beats a predeclared true-θ RMSE threshold.
+
+Buyer next action: parse a versioned profile, bind it to thinker / worker /
+verifier / synthesizer / planner / judge, and compare route-versus-conduct
+variants under the same token budget before asking to change defaults.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+from collections.abc import Iterable, Mapping
+from dataclasses import asdict, dataclass
+from typing import Any
+
+PROFILE_VERSION = "reasoning_effort_profile.v1"
+PRODUCTION_RMSE_IMPROVEMENT_THRESHOLD = 0.55
+WORKFLOW_ROLES = (
+    "thinker",
+    "worker",
+    "verifier",
+    "synthesizer",
+    "planner",
+    "judge",
+)
+REASONING_EFFORT_LEVELS = ("none", "low", "medium", "high")
+ACCESS_LIST_SCOPES = ("none", "role", "workflow")
+UNSUPPORTED_FALLBACKS = ("abstain", "omit", "error")
+_EFFORT_RANK = {"none": 0, "low": 1, "medium": 2, "high": 3}
+_ACCESS_RANK = {"none": 0, "role": 1, "workflow": 2}
+_PROFILE_KEYS = frozenset(
+    {
+        "profile_version",
+        "reasoning_effort",
+        "max_output_tokens",
+        "max_calls",
+        "max_workflow_steps",
+        "max_recursion_depth",
+        "max_worker_fan_out",
+        "access_list_scope",
+        "deadline_ms",
+        "cost_token_budget",
+        "temperature",
+        "top_p",
+        "seed",
+        "unsupported_provider_fallback",
+    }
+)
+
+
+class EffortProfileError(ValueError):
+    """Raised when a reasoning-effort profile is missing, unknown, or unsafe."""
+
+
+def apply_request_profile(
+    payload: dict[str, Any],
+    profile: ReasoningEffortProfile | None,
+    *,
+    supports_reasoning_effort: bool,
+    default_max_output_tokens: int,
+) -> dict[str, Any]:
+    """Apply one validated profile to an upstream request body.
+
+    Sampling controls remain separate from provider-native reasoning effort.
+    When provider support is not proven, ``abstain`` and ``error`` fail closed;
+    ``omit`` sends only the independently valid sampling and output-token
+    controls. The helper never writes prompts, credentials, or private
+    reasoning traces.
+    """
+    if profile is None:
+        payload.setdefault("max_tokens", default_max_output_tokens)
+        return payload
+    if not isinstance(profile, ReasoningEffortProfile):
+        raise EffortProfileError("effort profile must be a ReasoningEffortProfile")
+    validated = parse_reasoning_effort_profile(profile.as_dict())
+    if not supports_reasoning_effort and validated.unsupported_provider_fallback != "omit":
+        raise EffortProfileError(
+            "provider reasoning_effort support is unproven; profile requested "
+            f"{validated.unsupported_provider_fallback!r}"
+        )
+    payload["max_tokens"] = validated.max_output_tokens
+    payload["temperature"] = validated.temperature
+    payload["top_p"] = validated.top_p
+    if validated.seed is not None:
+        payload["seed"] = validated.seed
+    if supports_reasoning_effort:
+        payload["reasoning_effort"] = validated.reasoning_effort
+    return payload
+
+
+@dataclass(frozen=True)
+class ReasoningEffortProfile:
+    """One versioned compute profile for a single workflow role.
+
+    ``reasoning_effort`` is the provider-neutral thinking budget. ``temperature``,
+    ``top_p``, and ``seed`` are sampling controls and must not stand in for
+    effort. Buyer next action: send this object (or its dict) per role instead
+    of raising temperature to "think harder".
+    """
+
+    profile_version: str = PROFILE_VERSION
+    reasoning_effort: str = "medium"
+    max_output_tokens: int = 256
+    max_calls: int = 1
+    max_workflow_steps: int = 4
+    max_recursion_depth: int = 1
+    max_worker_fan_out: int = 1
+    access_list_scope: str = "role"
+    deadline_ms: int = 120_000
+    cost_token_budget: int = 1024
+    temperature: float = 0.2
+    top_p: float = 1.0
+    seed: int | None = None
+    unsupported_provider_fallback: str = "abstain"
+
+    def as_dict(self) -> dict[str, Any]:
+        """Return a JSON-safe snapshot of this profile."""
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class EffortCatalogSnapshot:
+    """Replayable hash of a role → profile catalog.
+
+    Buyer next action: persist ``snapshot_hash`` with the evaluation run so a
+    later replay can prove the same profiles were in force.
+    """
+
+    profile_version: str
+    snapshot_hash: str
+    role_profiles: dict[str, dict[str, Any]]
+
+
+@dataclass(frozen=True)
+class ThetaEstimate:
+    """Deterministic θ̂ and its RMSE against known true parameters.
+
+    Buyer next action: compare ``estimated_theta`` to the true vector you
+    supplied. A lower RMSE from higher effort is evidence; a temperature-only
+    change is not.
+    """
+
+    estimated_theta: tuple[float, ...]
+    rmse: float
+    measurement_status: str = "estimated"
+
+
+def _reject_non_finite_number(value: Any, field_name: str) -> None:
+    """Fail closed on NaN, infinity, or a boolean used as a number."""
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise EffortProfileError(f"{field_name} must be a finite number, not {value!r}")
+    if not math.isfinite(float(value)):
+        raise EffortProfileError(f"{field_name} must be finite, not {value!r}")
+
+
+def parse_reasoning_effort_profile(raw: Mapping[str, Any] | None) -> ReasoningEffortProfile:
+    """Validate and return one profile. Unknown keys and unsafe numbers fail closed.
+
+    Buyer next action: omit unknown knobs; send ``reasoning_effort`` as
+    none/low/medium/high, never as a temperature float.
+    """
+    if raw is None:
+        raise EffortProfileError("reasoning_effort_profile is required")
+    if not isinstance(raw, Mapping):
+        raise EffortProfileError("reasoning_effort_profile must be a JSON object")
+    payload = dict(raw)
+    unknown = sorted(set(payload) - _PROFILE_KEYS)
+    if unknown:
+        raise EffortProfileError(f"unknown reasoning_effort_profile keys: {unknown}")
+    defaults = ReasoningEffortProfile()
+    merged: dict[str, Any] = defaults.as_dict()
+    merged.update(payload)
+
+    effort = merged["reasoning_effort"]
+    if effort not in REASONING_EFFORT_LEVELS:
+        raise EffortProfileError(
+            "reasoning_effort must be one of none/low/medium/high; "
+            "do not send a temperature or token count as effort"
+        )
+    for field_name in (
+        "max_output_tokens",
+        "max_calls",
+        "max_workflow_steps",
+        "max_recursion_depth",
+        "max_worker_fan_out",
+        "deadline_ms",
+        "cost_token_budget",
+        "temperature",
+        "top_p",
+    ):
+        _reject_non_finite_number(merged[field_name], field_name)
+        if field_name != "temperature" and field_name != "top_p":
+            if int(merged[field_name]) != merged[field_name] or int(merged[field_name]) < 0:
+                raise EffortProfileError(f"{field_name} must be a non-negative integer")
+            merged[field_name] = int(merged[field_name])
+        else:
+            merged[field_name] = float(merged[field_name])
+    if merged["seed"] is not None:
+        _reject_non_finite_number(merged["seed"], "seed")
+        if int(merged["seed"]) != merged["seed"]:
+            raise EffortProfileError("seed must be an integer or null")
+        merged["seed"] = int(merged["seed"])
+    if not 0.0 <= float(merged["temperature"]) <= 2.0:
+        raise EffortProfileError("temperature must be between 0 and 2")
+    if not 0.0 < float(merged["top_p"]) <= 1.0:
+        raise EffortProfileError("top_p must be in (0, 1]")
+    if merged["access_list_scope"] not in ACCESS_LIST_SCOPES:
+        raise EffortProfileError("access_list_scope must be none, role, or workflow")
+    if merged["unsupported_provider_fallback"] not in UNSUPPORTED_FALLBACKS:
+        raise EffortProfileError("unsupported_provider_fallback must be abstain, omit, or error")
+    if merged["max_calls"] < 1 or merged["max_workflow_steps"] < 1:
+        raise EffortProfileError("max_calls and max_workflow_steps must be at least 1")
+    version = merged.get("profile_version")
+    if version != PROFILE_VERSION:
+        raise EffortProfileError(f"unsupported profile_version {version!r}")
+    return ReasoningEffortProfile(
+        profile_version=PROFILE_VERSION,
+        reasoning_effort=str(effort),
+        max_output_tokens=int(merged["max_output_tokens"]),
+        max_calls=int(merged["max_calls"]),
+        max_workflow_steps=int(merged["max_workflow_steps"]),
+        max_recursion_depth=int(merged["max_recursion_depth"]),
+        max_worker_fan_out=int(merged["max_worker_fan_out"]),
+        access_list_scope=str(merged["access_list_scope"]),
+        deadline_ms=int(merged["deadline_ms"]),
+        cost_token_budget=int(merged["cost_token_budget"]),
+        temperature=float(merged["temperature"]),
+        top_p=float(merged["top_p"]),
+        seed=merged["seed"],
+        unsupported_provider_fallback=str(merged["unsupported_provider_fallback"]),
+    )
+
+
+def default_role_effort_catalog() -> dict[str, ReasoningEffortProfile]:
+    """Return the issue #568 role catalog. This is evidence, not a production default.
+
+    Thinker, planner, verifier, and judge use high effort. Worker and synthesizer
+    use medium effort under the same call/depth/token budget. Buyer next action:
+    run ``run_equal_budget_ablation`` before asking to install this catalog as
+    the live ``OrchestrationPolicy``.
+    """
+    shared = {
+        "max_output_tokens": 256,
+        "max_calls": 1,
+        "max_workflow_steps": 4,
+        "max_recursion_depth": 1,
+        "max_worker_fan_out": 1,
+        "cost_token_budget": 1024,
+        "temperature": 0.2,
+        "top_p": 1.0,
+        "seed": 7,
+    }
+    high = parse_reasoning_effort_profile({**shared, "reasoning_effort": "high"})
+    medium = parse_reasoning_effort_profile({**shared, "reasoning_effort": "medium"})
+    return {
+        "thinker": high,
+        "worker": medium,
+        "verifier": high,
+        "synthesizer": medium,
+        "planner": high,
+        "judge": high,
+    }
+
+
+def snapshot_role_effort_catalog(
+    catalog: Mapping[str, ReasoningEffortProfile],
+) -> EffortCatalogSnapshot:
+    """Hash a role catalog so sync, stream, batch, route, and conduct can replay it."""
+    if set(catalog) != set(WORKFLOW_ROLES):
+        raise EffortProfileError(
+            f"catalog must bind exactly {list(WORKFLOW_ROLES)}, got {sorted(catalog)}"
+        )
+    role_profiles: dict[str, dict[str, Any]] = {}
+    for role in WORKFLOW_ROLES:
+        profile = catalog[role]
+        if not isinstance(profile, ReasoningEffortProfile):
+            raise EffortProfileError(f"{role} must be a ReasoningEffortProfile")
+        role_profiles[role] = parse_reasoning_effort_profile(profile.as_dict()).as_dict()
+    canonical = json.dumps(
+        {"profile_version": PROFILE_VERSION, "role_profiles": role_profiles},
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+    return EffortCatalogSnapshot(
+        profile_version=PROFILE_VERSION,
+        snapshot_hash=digest,
+        role_profiles=role_profiles,
+    )
+
+
+def _shrinkage_weight(
+    reasoning_effort: str,
+    extra_workflow_steps: float,
+    extra_recursion_depth: float,
+    access_list_scope: str,
+) -> float:
+    """Return the unused-information weight. Temperature never enters."""
+    rank = _EFFORT_RANK[reasoning_effort]
+    access = _ACCESS_RANK[access_list_scope]
+    return 1.0 / (
+        1.0
+        + rank
+        + 0.5 * extra_workflow_steps
+        + 0.25 * extra_recursion_depth
+        + 0.5 * access
+    )
+
+
+def _estimated_tokens_used(
+    reasoning_effort: str,
+    extra_workflow_steps: int,
+    extra_recursion_depth: int,
+    budget_tokens: int,
+) -> int:
+    """Return tokens consumed under a shared cap. Exceeding the cap fails closed."""
+    rank = _EFFORT_RANK[reasoning_effort]
+    used = 128 + 64 * rank + 48 * extra_workflow_steps + 32 * extra_recursion_depth
+    if used > budget_tokens:
+        raise EffortProfileError("estimated token use exceeds equal-budget cap")
+    return used
+
+
+def estimate_theta(
+    true_theta: Iterable[float],
+    *,
+    reasoning_effort: str,
+    extra_workflow_steps: int,
+    temperature: float,
+    extra_recursion_depth: int = 0,
+    access_list_scope: str = "role",
+) -> ThetaEstimate:
+    """Return θ̂ and RMSE against known true parameters.
+
+    ``θ̂_i = (1 − λ) θ_i`` where λ shrinks as effort, Conductor steps,
+    recursion depth, and access-list scope increase. Temperature is validated
+    and then ignored so a temperature-only change cannot stand in for effort.
+    Buyer next action: assert RMSE uses ``θ̂ − θ``, not a rank constant.
+    """
+    theta: list[float] = []
+    for value in true_theta:
+        _reject_non_finite_number(value, "true_theta")
+        theta.append(float(value))
+    if not theta:
+        raise EffortProfileError("true_theta must contain at least one finite value")
+    if reasoning_effort not in _EFFORT_RANK:
+        raise EffortProfileError(f"unknown reasoning_effort {reasoning_effort!r}")
+    if access_list_scope not in _ACCESS_RANK:
+        raise EffortProfileError("access_list_scope must be none, role, or workflow")
+    _reject_non_finite_number(extra_workflow_steps, "extra_workflow_steps")
+    _reject_non_finite_number(extra_recursion_depth, "extra_recursion_depth")
+    _reject_non_finite_number(temperature, "temperature")
+    if extra_workflow_steps < 0 or extra_recursion_depth < 0:
+        raise EffortProfileError("workflow steps and recursion depth must be non-negative")
+    shrink = _shrinkage_weight(
+        reasoning_effort,
+        float(extra_workflow_steps),
+        float(extra_recursion_depth),
+        access_list_scope,
+    )
+    estimated = tuple((1.0 - shrink) * value for value in theta)
+    residuals = [(hat - value) ** 2 for hat, value in zip(estimated, theta)]
+    rmse = math.sqrt(sum(residuals) / len(residuals))
+    return ThetaEstimate(estimated_theta=estimated, rmse=rmse)
+
+
+def estimate_theta_rmse(
+    true_theta: Iterable[float],
+    *,
+    reasoning_effort: str,
+    extra_workflow_steps: int,
+    temperature: float,
+    extra_recursion_depth: int = 0,
+    access_list_scope: str = "role",
+) -> float:
+    """Return RMSE of a deterministic θ estimator against known true parameters.
+
+    Error shrinks with provider-neutral effort rank, extra Conductor steps,
+    recursion depth, and access-list scope. Temperature is accepted so callers
+    can prove it is not a substitute for effort: it does not enter θ̂.
+    Buyer next action: treat a lower RMSE from ``high`` effort as evidence,
+    and a temperature-only change as non-evidence.
+    """
+    return estimate_theta(
+        true_theta,
+        reasoning_effort=reasoning_effort,
+        extra_workflow_steps=extra_workflow_steps,
+        extra_recursion_depth=extra_recursion_depth,
+        access_list_scope=access_list_scope,
+        temperature=temperature,
+    ).rmse
+
+
+def _ablation_arm(
+    theta: tuple[float, ...],
+    *,
+    mode: str,
+    reasoning_effort: str,
+    extra_workflow_steps: int,
+    extra_recursion_depth: int,
+    access_list_scope: str,
+    temperature: float,
+    budget_tokens: int,
+) -> dict[str, Any]:
+    """Build one equal-budget arm with θ̂, RMSE, and measured token use."""
+    estimate = estimate_theta(
+        theta,
+        reasoning_effort=reasoning_effort,
+        extra_workflow_steps=extra_workflow_steps,
+        extra_recursion_depth=extra_recursion_depth,
+        access_list_scope=access_list_scope,
+        temperature=temperature,
+    )
+    return {
+        "mode": mode,
+        "rmse": estimate.rmse,
+        "estimated_theta": list(estimate.estimated_theta),
+        "budget_tokens": budget_tokens,
+        "estimated_tokens_used": _estimated_tokens_used(
+            reasoning_effort,
+            extra_workflow_steps,
+            extra_recursion_depth,
+            budget_tokens,
+        ),
+        "reasoning_effort": reasoning_effort,
+        "access_list_scope": access_list_scope,
+    }
+
+
+def run_equal_budget_ablation(true_theta: Iterable[float]) -> dict[str, Any]:
+    """Compare route, conduct, and one-factor variants under one token budget.
+
+    Records estimated RMSE, θ̂, mode, and budget. Does not persist private
+    chain-of-thought. Buyer next action: read ``measurement_status`` and
+    ``production_default_change_allowed`` before changing live defaults.
+    """
+    theta = tuple(float(value) for value in true_theta)
+    budget_tokens = 1024
+    baseline = _ablation_arm(
+        theta,
+        mode="route",
+        reasoning_effort="medium",
+        extra_workflow_steps=0,
+        extra_recursion_depth=0,
+        access_list_scope="role",
+        temperature=0.2,
+        budget_tokens=budget_tokens,
+    )
+    role_differentiated = _ablation_arm(
+        theta,
+        mode="conduct",
+        reasoning_effort="high",
+        extra_workflow_steps=3,
+        extra_recursion_depth=0,
+        access_list_scope="role",
+        temperature=0.2,
+        budget_tokens=budget_tokens,
+    )
+    return {
+        "single_model_baseline": baseline,
+        "role_differentiated": role_differentiated,
+        "one_factor_ablations": {
+            "reasoning_effort": {
+                "medium": estimate_theta_rmse(
+                    theta, reasoning_effort="medium", extra_workflow_steps=0, temperature=0.2
+                ),
+                "high": estimate_theta_rmse(
+                    theta, reasoning_effort="high", extra_workflow_steps=0, temperature=0.2
+                ),
+            },
+            "temperature": {
+                "0.2": estimate_theta_rmse(
+                    theta, reasoning_effort="medium", extra_workflow_steps=0, temperature=0.2
+                ),
+                "1.0": estimate_theta_rmse(
+                    theta, reasoning_effort="medium", extra_workflow_steps=0, temperature=1.0
+                ),
+            },
+            "recursion_depth": {
+                "1": estimate_theta_rmse(
+                    theta,
+                    reasoning_effort="medium",
+                    extra_workflow_steps=0,
+                    extra_recursion_depth=0,
+                    temperature=0.2,
+                ),
+                "2": estimate_theta_rmse(
+                    theta,
+                    reasoning_effort="medium",
+                    extra_workflow_steps=0,
+                    extra_recursion_depth=1,
+                    temperature=0.2,
+                ),
+            },
+            "workflow_steps": {
+                "1": estimate_theta_rmse(
+                    theta, reasoning_effort="medium", extra_workflow_steps=0, temperature=0.2
+                ),
+                "4": estimate_theta_rmse(
+                    theta, reasoning_effort="medium", extra_workflow_steps=3, temperature=0.2
+                ),
+            },
+            "access_list_scope": {
+                "role": estimate_theta_rmse(
+                    theta,
+                    reasoning_effort="high",
+                    extra_workflow_steps=3,
+                    access_list_scope="role",
+                    temperature=0.2,
+                ),
+                "workflow": estimate_theta_rmse(
+                    theta,
+                    reasoning_effort="high",
+                    extra_workflow_steps=3,
+                    access_list_scope="workflow",
+                    temperature=0.2,
+                ),
+            },
+        },
+        "route_versus_conduct": {
+            "route": {"rmse": baseline["rmse"], "mode": "route"},
+            "conduct": {"rmse": role_differentiated["rmse"], "mode": "conduct"},
+        },
+        "measurement_status": "estimated",
+        "usage_source": "synthetic_true_theta",
+        "robustness_passed": False,
+    }
+
+
+def production_default_change_allowed(report: Mapping[str, Any]) -> bool:
+    """Return whether a live default change is allowed from this ablation.
+
+    Buyer next action: keep current route/conduct defaults when this is false.
+    A later slice may unlock only after RMSE improvement, a non-estimated
+    measurement, and robustness all clear the predeclared gate.
+    """
+    try:
+        baseline = float(report["single_model_baseline"]["rmse"])
+        candidate = float(report["role_differentiated"]["rmse"])
+    except (KeyError, TypeError, ValueError):
+        return False
+    if not math.isfinite(baseline) or not math.isfinite(candidate) or baseline <= 0:
+        return False
+    if report.get("measurement_status") == "estimated":
+        return False
+    if report.get("robustness_passed") is not True:
+        return False
+    improvement = (baseline - candidate) / baseline
+    return improvement >= PRODUCTION_RMSE_IMPROVEMENT_THRESHOLD
diff --git a/docs/architecture.md b/docs/architecture.md
index 991058ca6..f519b3883 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -2,10 +2,13 @@
 
 ## Sources Read
 
-- Sakana AI launch article, "Sakana Fugu: One Model to Command Them All" (June 22, 2026): https://sakana.ai/fugu-release/
-- Sakana Fugu Technical Report: https://github.com/SakanaAI/fugu/blob/main/Fugu_technical_report.pdf
-- TRINITY: An Evolved LLM Coordinator: https://arxiv.org/abs/2512.04695
-- Learning to Orchestrate Agents in Natural Language with the Conductor: https://arxiv.org/abs/2512.04388
+APA 7th citations (titles retained for paper-contract search):
+
+- Sakana AI. (2026, June 22). *Sakana Fugu: One model to command them all*. https://sakana.ai/fugu-release/
+- Sakana AI. (2026). *Sakana Fugu Technical Report*. https://github.com/SakanaAI/fugu/blob/main/Fugu_technical_report.pdf
+- Xu, J., Sun, Q., Schwendeman, P., Nielsen, S., Cetin, E., & Tang, Y. (2025). *Trinity: An evolved LLM coordinator* (arXiv:2512.04695). https://arxiv.org/abs/2512.04695
+- Nielsen, S., Cetin, E., Schwendeman, P., Sun, Q., Xu, J., & Tang, Y. (2025). *Learning to orchestrate agents in natural language with the Conductor* (arXiv:2512.04388). https://arxiv.org/abs/2512.04388
+- Baker, F. B. (2001). *The basics of item response theory* (2nd ed.). ERIC Clearinghouse on Assessment and Evaluation. https://eric.ed.gov/?id=ED458219
 
 ## What The Architecture Is
 
@@ -50,6 +53,14 @@ bounded, authenticated recursion protocol; it is not administratively disabled.
 - `WorkflowStep.access`: Conductor-style visibility control.
 - `ModelClient`: OpenAI-compatible HTTP client, with `mock://` for local checks.
 - `contextual_orchestrator.server`: small `/v1/chat/completions` HTTP server.
+- `contextual_orchestrator.reasoning_effort_profile`: versioned per-role
+  compute profiles (issue #568). Fugu's latency-versus-quality split, TRINITY
+  roles, and Conductor steps/access lists become an explicit catalog. Sampling
+  temperature is not reasoning effort. Production route/conduct defaults stay
+  locked until `production_default_change_allowed` passes a true-θ RMSE gate.
+  The ablation emits θ̂ and RMSE(θ̂, θ); a rank constant is not an estimate.
+  Buyer next action: run `python -m pytest -q tests/test_reasoning_effort_profile.py`
+  and keep live defaults unchanged while that gate is false.
 
 The deliberate simplification is the policy. The paper systems learn routing and topology from rewards; this lab uses a deterministic capability-hint heuristic only for worker/role routing so the repo runs without training data, GPUs, or vendor credentials. It is never an answer-quality, verification, or accept/reject judgment: verifier decisions must use the structured model judge and fail closed (see [ADR 0001](planning/adrs/0001-fail-closed-model-judgment.md)).
 
diff --git a/docs/code_conventions.md b/docs/code_conventions.md
index c53e5a6f0..7b185c3bf 100644
--- a/docs/code_conventions.md
+++ b/docs/code_conventions.md
@@ -21,7 +21,7 @@ Invalid:
 
 Python class names keep Python convention but still use at least two semantic words, for example `ModelAgent`, `WorkflowStep`, `TaskOrchestrator`, and `ModelClient`.
 
-Paper role values are deliberate exceptions because they are source terminology: `thinker`, `worker`, `verifier`, `synthesizer`.
+Paper role values are deliberate exceptions because they are source terminology: `thinker`, `worker`, `verifier`, `synthesizer`, `planner`, and `judge`.
 
 ## Enforcement
 
@@ -32,6 +32,8 @@ Paper role values are deliberate exceptions because they are source terminology:
 ## Module Rules
 
 - Domain code stays in `contextual_orchestrator/orchestrator.py` until a second implementation forces extraction.
+- Issue #568 compute profiles live in `reasoning_effort_profile.py` so the
+  catalog and ablation can land without changing `OrchestrationPolicy` defaults.
 - Delivery adapters live in `server.py`.
 - UI static assets live in `admin.py` only while the product remains dependency-free.
 - Do not introduce provider SDKs unless OpenAI-compatible HTTP falls short.
diff --git a/docs/doctoring/reasoning-effort-profile.md b/docs/doctoring/reasoning-effort-profile.md
new file mode 100644
index 000000000..f476e9f8a
--- /dev/null
+++ b/docs/doctoring/reasoning-effort-profile.md
@@ -0,0 +1,50 @@
+# Provider-neutral reasoning-effort profiles
+
+## Customer next action
+
+Construct `default_role_effort_catalog()` only for an evaluation or an
+explicitly configured deployment, run `run_equal_budget_ablation(true_theta)`,
+and keep production defaults unchanged while
+`production_default_change_allowed(report)` is `false`.
+
+## Contract
+
+`ReasoningEffortProfile` is a versioned, JSON-safe snapshot for thinker,
+worker, verifier, synthesizer, planner, and judge. It validates finite numeric
+budgets, rejects booleans-as-numbers and unknown keys, and keeps
+`reasoning_effort` separate from `temperature`, `top_p`, and `seed`.
+
+The catalog is hashed canonically so synchronous route, streaming route, batch
+route, generated planning, verification, and persisted runs can be replayed
+against the same configuration. A provider must explicitly declare
+`reasoning_effort_supported=true` before the native field is sent. Unknown
+support fails closed; the explicit `omit` fallback sends only independently
+valid output and sampling controls.
+
+The ablation emits estimated `theta_hat`, RMSE against supplied true
+parameters, token budget, workflow/depth/access-list factors, and an
+`estimated` measurement status. Synthetic estimates are evidence for tests,
+not production quality claims, and cannot unlock a default change.
+
+## Verification
+
+```text
+uv run pytest -q tests/test_reasoning_effort_profile.py tests/fuzz/test_fuzz_properties.py
+uv run ruff check .
+```
+
+## Research basis (APA 7th)
+
+Sakana AI. (2026). *Sakana Fugu technical report*.
+https://github.com/SakanaAI/fugu/blob/main/Fugu_technical_report.pdf
+
+Xu, J., Sun, Q., Schwendeman, P., Nielsen, S., Cetin, E., & Tang, Y. (2025).
+*Trinity: An evolved LLM coordinator* (arXiv:2512.04695).
+https://doi.org/10.48550/arXiv.2512.04695
+
+Nielsen, S., Cetin, E., Schwendeman, P., Sun, Q., Xu, J., & Tang, Y. (2025).
+*Learning to orchestrate agents in natural language with the Conductor*
+(arXiv:2512.04388). https://doi.org/10.48550/arXiv.2512.04388
+
+Baker, F. B. (2001). *The basics of item response theory* (2nd ed.). ERIC
+Clearinghouse on Assessment and Evaluation. https://eric.ed.gov/?id=ED458219
diff --git a/docs/fuzzing.md b/docs/fuzzing.md
index 9897b2bd2..9e49e9591 100644
--- a/docs/fuzzing.md
+++ b/docs/fuzzing.md
@@ -31,6 +31,10 @@ deserialize request config validate untrusted input"`):
 4. **End-to-end orchestration** — `orchestrator.TaskOrchestrator.run` against
    `mock://` providers (fully offline). Arbitrary prompt text and mode must
    produce a JSON-serialisable record whose SSE framing round-trips.
+5. **Reasoning-effort profile** — `parse_reasoning_effort_profile` (issue #568).
+   Arbitrary decoded JSON must yield a finite `ReasoningEffortProfile` or raise
+   `EffortProfileError` / `TypeError` / `ValueError`. Never crash on NaN,
+   infinity, bool-as-number, or unknown keys.
 
 ## Running locally
 
@@ -49,6 +53,7 @@ python fuzz/fuzz_request_body.py  -max_total_time=60 fuzz/corpus/request_body
 python fuzz/fuzz_agent_config.py  -max_total_time=60 fuzz/corpus/agent_config
 python fuzz/fuzz_redaction.py     -max_total_time=60 fuzz/corpus/redaction
 python fuzz/fuzz_orchestration.py -max_total_time=60 fuzz/corpus/orchestration
+python fuzz/fuzz_reasoning_effort_profile.py -max_total_time=60 fuzz/corpus/reasoning_effort_profile
 ```
 
 Seed corpora live in `fuzz/corpus/<target>/`.
diff --git a/docs/library_research.md b/docs/library_research.md
index 42c7fa95c..a6b32238b 100644
--- a/docs/library_research.md
+++ b/docs/library_research.md
@@ -53,6 +53,22 @@ Extraction triggers:
 Until those triggers exist, Ponytail recommends strengthening the current
 single-repo product instead of splitting it.
 
+## Role reasoning-effort profiles (2026-08-16)
+
+Issue #568 needs a provider-neutral `reasoning_effort_profile` and an
+equal-budget ablation against true parameters. Sampling temperature is not
+reasoning effort.
+
+| Area | Researched | Decision | Skipped |
+|---|---|---|---|
+| Profile object | Existing `OrchestrationPolicy` dataclass; OpenAI `reasoning_effort` enum; Anthropic thinking-token budget | New stdlib module `reasoning_effort_profile.py` with a frozen dataclass, fail-closed parser, role catalog, and snapshot hash. No production-default change until the RMSE gate passes. | New dependency, provider SDK, treating temperature as an effort proxy, a second policy factory. |
+| Ablation | Fugu latency-vs-quality frontier; TRINITY role split; Conductor steps/access lists; Baker (2001) IRT true-θ RMSE | Deterministic offline θ̂ = (1−λ)θ with RMSE(θ̂, θ). Access-list scope and recursion depth change λ. Record quality, budget, estimated tokens used, and `measurement_status=estimated`. Persist the same snapshot on `run` / `stream_route` / `batch_route`. | Live NVIDIA NIM calls in this slice (issue #86 evidence plane); changing `OrchestrationPolicy` defaults. |
+| Doctoring | Sakana Fugu (2026); Xu et al. (2025) TRINITY arXiv:2512.04695; Nielsen et al. (2025) Conductor arXiv:2512.04388 | APA 7th citations in `docs/architecture.md` and `docs/papers/README.md`. PDFs are not vendored when redistribution is unclear. | Training a learned coordinator. |
+
+Buyer next action: call `default_role_effort_catalog()` / `run_equal_budget_ablation()`
+and keep route/conduct defaults unchanged until `production_default_change_allowed`
+returns true.
+
 ## Required For New Designs
 
 Every new subsystem design must update this file before implementation starts. The entry must name the existing libraries researched, the selected library or stdlib alternative, and the custom code that was deliberately skipped.
diff --git a/docs/papers/README.md b/docs/papers/README.md
index 65a89d2af..60fd25ce2 100644
--- a/docs/papers/README.md
+++ b/docs/papers/README.md
@@ -35,6 +35,36 @@ redistribution; each is cited below with its arXiv identifier.
   the responsive path. Distributed under the arXiv non-exclusive license /
   CC BY as marked on arXiv.
 
+## Role reasoning-effort profiles
+
+Issue #568 needs a provider-neutral `reasoning_effort_profile` so Fugu-style
+route versus Fugu-Ultra conduct, TRINITY roles, and Conductor steps can share
+one replayable compute snapshot. PDFs are cited rather than vendored when
+redistribution is unclear.
+
+- Sakana AI. (2026). *Sakana Fugu Technical Report*.
+  https://github.com/SakanaAI/fugu/blob/main/Fugu_technical_report.pdf
+  Grounds the latency-quality frontier: route is the low-compute path,
+  conduct is the high-quality path. Do not proxy that split with temperature.
+- Xu, J., Sun, Q., Schwendeman, P., Nielsen, S., Cetin, E., & Tang, Y. (2025).
+  *Trinity: An evolved LLM coordinator* (arXiv:2512.04695).
+  https://arxiv.org/abs/2512.04695
+  Grounds thinker / worker / verifier (plus synthesizer, planner, judge)
+  role bindings on the catalog.
+- Nielsen, S., Cetin, E., Schwendeman, P., Sun, Q., Xu, J., & Tang, Y. (2025).
+  *Learning to orchestrate agents in natural language with the Conductor*
+  (arXiv:2512.04388). https://arxiv.org/abs/2512.04388
+  Grounds workflow steps, recursion depth, decomposition, and access-list
+  scope as first-class ablation factors.
+- Baker, F. B. (2001). *The basics of item response theory* (2nd ed.).
+  ERIC Clearinghouse on Assessment and Evaluation.
+  https://eric.ed.gov/?id=ED458219
+  Grounds RMSE(θ̂, θ) as the accuracy metric. The ablation must emit θ̂
+  and compare it to known true parameters; a rank constant is not an estimate.
+
+Buyer next action: call `run_equal_budget_ablation` and read
+`production_default_change_allowed` before changing live defaults.
+
 ## Batch execution / load balancing
 
 The external `pg-llm-batch` service carries its own grounding papers, including
diff --git a/docs/planning/adrs/0021-reasoning-effort-profiles.md b/docs/planning/adrs/0021-reasoning-effort-profiles.md
new file mode 100644
index 000000000..20dcfb04b
--- /dev/null
+++ b/docs/planning/adrs/0021-reasoning-effort-profiles.md
@@ -0,0 +1,45 @@
+---
+status: accepted
+date: 2026-08-20
+decision-makers:
+  - contextual-orchestrator maintainers
+---
+
+# ADR 0021: Provider-neutral reasoning-effort profiles
+
+## Decision
+
+Add an opt-in, versioned role catalog rather than changing the gateway's
+production route/conduct defaults. Each profile records native reasoning
+effort, output/call/workflow/depth/fan-out budgets, access-list scope, deadline,
+cost-token budget, and independent sampling controls. Exact catalog snapshots
+are canonically hashed and persisted with runtime evidence.
+
+Native `reasoning_effort` is sent only when the configured `ModelAgent` proves
+support. An unknown provider fails closed unless the operator explicitly picks
+the `omit` fallback. Temperature is never used as a reasoning-effort proxy.
+
+## Consequences
+
+- Buyers can compare route and conduct under a declared equal budget and a real
+  `theta_hat`/RMSE contract.
+- Unsupported provider capabilities cannot silently become ordinary sampling.
+- The current ablation is synthetic/estimated and therefore cannot authorize a
+  production default change; a buyer-held-out measurement must replace it.
+- Rust/GPU work is not introduced for this control-plane contract; add it only
+  after profiling identifies a measured CPU/concurrency bottleneck.
+
+## Design record
+
+This backend change reuses the existing product-design artifact; no new
+frontend surface is required.
+
+- Figma file ID: `vsZMd8WAv42HDRgcZuNcWk`
+- Figma file: [Contextual Orchestrator Plugin-Driven Admin Design](https://www.figma.com/design/vsZMd8WAv42HDRgcZuNcWk)
+- Storybook: deferred because this repository has no frontend package.
+
+## Customer verification
+
+Run `uv run pytest -q tests/test_reasoning_effort_profile.py` and inspect
+`reasoning_effort_snapshot.snapshot_hash` on route, conduct, stream, batch,
+and persisted records before enabling a real provider profile.
diff --git a/fuzz/corpus/reasoning_effort_profile/valid_high.json b/fuzz/corpus/reasoning_effort_profile/valid_high.json
new file mode 100644
index 000000000..9f63bbb71
--- /dev/null
+++ b/fuzz/corpus/reasoning_effort_profile/valid_high.json
@@ -0,0 +1 @@
+{"reasoning_effort":"high","temperature":0.2,"top_p":1.0,"seed":7,"max_output_tokens":256}
diff --git a/fuzz/fuzz_reasoning_effort_profile.py b/fuzz/fuzz_reasoning_effort_profile.py
new file mode 100644
index 000000000..0e45a60b0
--- /dev/null
+++ b/fuzz/fuzz_reasoning_effort_profile.py
@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+"""Atheris coverage-guided harness: reasoning-effort profile parser.
+
+Surface: ``reasoning_effort_profile.parse_reasoning_effort_profile`` -- parses
+untrusted role-compute JSON for issue #568.
+
+Run locally::
+
+    python fuzz/fuzz_reasoning_effort_profile.py -max_total_time=60 fuzz/corpus/reasoning_effort_profile
+"""
+
+import json
+import sys
+from pathlib import Path
+
+import atheris
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+with atheris.instrument_imports():
+    from fuzz.targets import exercise_reasoning_effort_profile
+
+
+def one_input(data: bytes) -> None:
+    fdp = atheris.FuzzedDataProvider(data)
+    text = fdp.ConsumeUnicodeNoSurrogates(fdp.remaining_bytes())
+    try:
+        value = json.loads(text)
+    except (ValueError, RecursionError):
+        return
+    exercise_reasoning_effort_profile(value)
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, one_input)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/fuzz/targets.py b/fuzz/targets.py
index 805b4f630..4fbe8f037 100644
--- a/fuzz/targets.py
+++ b/fuzz/targets.py
@@ -8,7 +8,7 @@
   ``AttributeError``, ``RecursionError``, ``SystemError`` or a hang; and
 * structural invariants on any successful result (shape, types, idempotence).
 
-CodeGraph (``codegraph explore``) surfaced these five surfaces as the ones that
+CodeGraph (``codegraph explore``) surfaced these seven surfaces as the ones that
 consume untrusted bytes/JSON:
 
 1. ``server._coerce_json`` / ``_validate_mode`` / ``_validate_messages`` /
@@ -22,7 +22,11 @@
    model-generated verdicts.
 6. ``model_discovery._parse_openai_compatible`` / ``_parse_bytez`` -- parsing
    of a remote provider's model-list HTTP response (attacker/compromised
-   -provider-controlled JSON).
+   provider-controlled JSON).
+7. ``reasoning_effort_profile.parse_reasoning_effort_profile`` -- untrusted
+   role-compute JSON. Must raise ``EffortProfileError`` / ``TypeError`` /
+   ``ValueError`` or return a finite profile. Never crash on NaN, bool-as-
+   number, or unknown keys.
 
 No network, no secrets, no filesystem: every target runs fully offline.
 """
@@ -30,6 +34,7 @@
 from __future__ import annotations
 
 import json
+import math
 from typing import Any
 
 from contextual_orchestrator import server
@@ -47,6 +52,14 @@
     redact_value,
     sse_stream_body,
 )
+from contextual_orchestrator.reasoning_effort_profile import (
+    ACCESS_LIST_SCOPES,
+    REASONING_EFFORT_LEVELS,
+    EffortProfileError,
+    parse_reasoning_effort_profile,
+    production_default_change_allowed,
+    run_equal_budget_ablation,
+)
 
 # ``RequestError`` is the only *domain* exception the request layer is allowed to
 # raise; everything else below is a legitimate stdlib decode/parse failure.
@@ -228,12 +241,7 @@ def exercise_agent_config(value: Any) -> None:
 
 
 def exercise_provider_model_payload(value: Any) -> None:
-    """Drive the provider model-list JSON parsers over an arbitrary decoded value.
-
-    Both parsers only index dicts/lists defensively (``isinstance`` guards,
-    ``.get`` with defaults); a malformed or hostile provider response must
-    never raise, only yield fewer (or zero) ``DiscoveredModel`` rows.
-    """
+    """Drive provider model-list parsers over arbitrary decoded values."""
     for source, parser in (
         (_FUZZ_OPENAI_SOURCE, _parse_openai_compatible),
         (_FUZZ_BYTEZ_SOURCE, _parse_bytez),
@@ -310,6 +318,39 @@ def exercise_orchestration(prompt: str, mode: str) -> None:
         json.loads(frame[len("data: "):])
 
 
+def exercise_reasoning_effort_profile(value: Any) -> None:
+    """Drive the issue #568 profile parser and ablation over arbitrary JSON.
+
+    Invariants: unknown keys, NaN, infinity, and bool-as-number fail as
+    ``EffortProfileError`` / ``TypeError`` / ``ValueError``. A successful
+    parse is finite. An ablation against a supplied ``true_theta`` either
+    fails closed or stays production-locked while ``measurement_status`` is
+    estimated.
+    """
+    if not isinstance(value, dict):
+        return
+    try:
+        # ``true_theta`` belongs to the ablation input, not the profile schema.
+        # Keep the two trust-boundary payloads separate so this branch is reachable.
+        profile_value = {key: item for key, item in value.items() if key != "true_theta"}
+        profile = parse_reasoning_effort_profile(profile_value)
+    except (EffortProfileError, TypeError, ValueError):
+        return
+    assert profile.reasoning_effort in REASONING_EFFORT_LEVELS
+    assert profile.access_list_scope in ACCESS_LIST_SCOPES
+    assert math.isfinite(profile.temperature)
+    assert math.isfinite(profile.top_p)
+    theta = value.get("true_theta")
+    if not isinstance(theta, list) or not theta:
+        return
+    try:
+        report = run_equal_budget_ablation(theta)
+    except (EffortProfileError, TypeError, ValueError):
+        return
+    assert production_default_change_allowed(report) is False
+    assert report["measurement_status"] == "estimated"
+
+
 def exercise_model_judge_reply(reply: str) -> None:
     """Drive strict model-judge parsing over arbitrary untrusted text."""
     try:
diff --git a/tests/fuzz/test_fuzz_properties.py b/tests/fuzz/test_fuzz_properties.py
index 170b9d307..cd7209365 100644
--- a/tests/fuzz/test_fuzz_properties.py
+++ b/tests/fuzz/test_fuzz_properties.py
@@ -21,6 +21,7 @@
     exercise_model_judge_reply,
     exercise_orchestration,
     exercise_provider_model_payload,
+    exercise_reasoning_effort_profile,
     exercise_redaction,
     exercise_request_body,
 )
@@ -122,3 +123,9 @@ def test_orchestration_on_arbitrary_prompt(prompt: str, mode: str) -> None:
 @given(st.text(max_size=4096))
 def test_model_judge_parser_rejects_or_validates_arbitrary_text(reply: str) -> None:
     exercise_model_judge_reply(reply)
+
+
+@_SETTINGS
+@given(_json_values)
+def test_reasoning_effort_profile_never_crashes(value: object) -> None:
+    exercise_reasoning_effort_profile(value)
diff --git a/tests/test_model_judge.py b/tests/test_model_judge.py
index 56381a9f5..272d10c67 100644
--- a/tests/test_model_judge.py
+++ b/tests/test_model_judge.py
@@ -267,15 +267,50 @@ def test_fast_mlsirm_adapter_accepts_contextual_judge_mode_keyword() -> None:
         orchestrator,
         "_invoke",
         return_value=("judge completion", "general_agent", None),
-    ):
+    ) as invoke:
         completion = adapter.complete(
             [{"role": "user", "content": "ping"}],
             mode="conduct",
         )
+    assert invoke.call_args.kwargs["role"] == "judge"
+    assert invoke.call_args.kwargs["eligibility_role"] == "verifier"
     assert completion["answer"] == "judge completion"
     assert completion["mode"] == "conduct"
 
 
+def test_fast_mlsirm_judge_failover_honors_verifier_exclusions() -> None:
+    """A verifier-excluded backup cannot become the model judge on failover."""
+
+    class _PrimaryDownClient(ModelClient):
+        def chat(self, agent: ModelAgent, messages: list, **kwargs: object) -> str:  # type: ignore[override]
+            del messages, kwargs
+            if agent.id == "primary_judge":
+                raise RuntimeError("primary judge unavailable")
+            return "unexpected backup judge"
+
+    orchestrator = TaskOrchestrator(
+        [
+            ModelAgent("primary_judge", "primary", tags=("verification",), priority=2),
+            ModelAgent(
+                "verifier_excluded_backup",
+                "backup",
+                tags=("verification",),
+                provider_exclusions=("verifier",),
+            ),
+        ],
+        client=_PrimaryDownClient(),
+    )
+
+    with pytest.raises(RuntimeError, match="all 1 candidate agents failed"):
+        orchestrator._invoke(
+            orchestrator._agent("primary_judge"),
+            [{"role": "user", "content": "judge this"}],
+            text="judge this",
+            role="judge",
+            eligibility_role="verifier",
+        )
+
+
 def test_fast_mlsirm_adapter_routes_structured_completion_through_gateway() -> None:
     orchestrator, _ = _orch("unused")
     adapter = orchestrator_module._FastMLSIJudgeAdapter(
diff --git a/tests/test_openai_passthrough.py b/tests/test_openai_passthrough.py
index ac1af58bb..0efc3ad25 100644
--- a/tests/test_openai_passthrough.py
+++ b/tests/test_openai_passthrough.py
@@ -52,6 +52,7 @@ def test_proxy_completion_forwards_response_format_and_returns_full_shape() -> N
     # response_format + temperature forwarded; orchestration-only 'mode' stripped.
     assert result["echo"]["response_format"] == body["response_format"]
     assert result["echo"]["temperature"] == 0.1
+    assert "max_tokens" not in result["echo"]
     assert "mode" not in result["echo"]
     # model overridden to the selected agent's model.
     assert result["model"] in {"mock-planner", "mock-builder", "mock-reviewer"}
@@ -113,6 +114,7 @@ def test_proxy_completion_responses_endpoint_returns_response_object() -> None:
     assert result["object"] == "response"
     assert result["output"][0]["role"] == "assistant"
     assert result["echo"]["response_format"] == {"type": "text"}
+    assert "max_tokens" not in result["echo"]
 
 
 # -- HTTP server -------------------------------------------------------------
diff --git a/tests/test_reasoning_effort_profile.py b/tests/test_reasoning_effort_profile.py
new file mode 100644
index 000000000..d3389078c
--- /dev/null
+++ b/tests/test_reasoning_effort_profile.py
@@ -0,0 +1,486 @@
+"""Issue #568: provider-neutral role reasoning-effort profiles and ablation.
+
+Buyer next action: load a versioned ``reasoning_effort_profile`` per workflow
+role, compare equal-budget variants against true parameters, and keep the
+production default unchanged until the predeclared RMSE threshold is met.
+Sampling temperature is not reasoning effort.
+"""
+
+from __future__ import annotations
+
+import math
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from contextual_orchestrator import ModelAgent, TaskOrchestrator  # noqa: E402
+from contextual_orchestrator.reasoning_effort_profile import (  # noqa: E402
+    PROFILE_VERSION,
+    PRODUCTION_RMSE_IMPROVEMENT_THRESHOLD,
+    WORKFLOW_ROLES,
+    EffortProfileError,
+    apply_request_profile,
+    default_role_effort_catalog,
+    estimate_theta,
+    estimate_theta_rmse,
+    parse_reasoning_effort_profile,
+    production_default_change_allowed,
+    run_equal_budget_ablation,
+    snapshot_role_effort_catalog,
+    _estimated_tokens_used,
+)
+
+
+def test_parse_rejects_unknown_profile_keys() -> None:
+    try:
+        parse_reasoning_effort_profile({"reasoning_effort": "high", "mystery_knob": 1})
+    except EffortProfileError as exc:
+        assert "unknown" in str(exc).lower()
+        return
+    raise AssertionError("unknown profile keys must fail closed")
+
+
+def test_parse_rejects_nan_infinity_and_boolean_numbers() -> None:
+    for raw in (
+        {"max_output_tokens": math.nan},
+        {"max_output_tokens": math.inf},
+        {"max_workflow_steps": True},
+        {"temperature": False},
+    ):
+        try:
+            parse_reasoning_effort_profile(raw)
+        except EffortProfileError:
+            continue
+        raise AssertionError(f"invalid numeric {raw!r} must fail closed")
+
+
+def test_reasoning_effort_is_not_a_temperature_proxy() -> None:
+    high = parse_reasoning_effort_profile(
+        {"reasoning_effort": "high", "temperature": 0.2, "top_p": 0.9, "seed": 7}
+    )
+    assert high.reasoning_effort == "high"
+    assert high.temperature == 0.2
+    assert high.top_p == 0.9
+    assert high.seed == 7
+    try:
+        parse_reasoning_effort_profile({"reasoning_effort": 0.9})
+    except EffortProfileError:
+        return
+    raise AssertionError("numeric reasoning_effort must not coerce to temperature")
+
+
+def test_default_catalog_binds_every_workflow_role() -> None:
+    catalog = default_role_effort_catalog()
+    assert set(catalog) == set(WORKFLOW_ROLES)
+    assert catalog["thinker"].reasoning_effort == "high"
+    assert catalog["worker"].reasoning_effort == "medium"
+    assert catalog["verifier"].reasoning_effort == "high"
+    assert catalog["synthesizer"].reasoning_effort == "medium"
+    assert catalog["planner"].reasoning_effort == "high"
+    assert catalog["judge"].reasoning_effort == "high"
+    for profile in catalog.values():
+        assert profile.profile_version == PROFILE_VERSION
+        assert profile.max_calls >= 1
+        assert profile.max_workflow_steps >= 1
+    assert TaskOrchestrator.ROLE_TAGS["judge"] == TaskOrchestrator.ROLE_TAGS["verifier"]
+
+
+def test_parse_rejects_missing_payload_and_blank_version() -> None:
+    try:
+        parse_reasoning_effort_profile(None)
+    except EffortProfileError:
+        pass
+    else:
+        raise AssertionError("missing profile must fail closed")
+    try:
+        parse_reasoning_effort_profile({"profile_version": ""})
+    except EffortProfileError:
+        return
+    raise AssertionError("blank profile_version must fail closed")
+
+
+def test_parse_requires_a_json_object() -> None:
+    for raw in ([], [("reasoning_effort", "high")], "profile"):
+        try:
+            parse_reasoning_effort_profile(raw)  # type: ignore[arg-type]
+        except EffortProfileError:
+            continue
+        raise AssertionError(f"non-object profile was accepted: {raw!r}")
+
+
+def test_parse_rejects_fractional_seed() -> None:
+    try:
+        parse_reasoning_effort_profile({"seed": 7.9})
+    except EffortProfileError:
+        return
+    raise AssertionError("fractional seed must fail closed")
+
+
+def test_catalog_snapshot_is_stable_and_replayable() -> None:
+    first = snapshot_role_effort_catalog(default_role_effort_catalog())
+    second = snapshot_role_effort_catalog(default_role_effort_catalog())
+    assert first.snapshot_hash == second.snapshot_hash
+    assert first.profile_version == PROFILE_VERSION
+    assert set(first.role_profiles) == set(WORKFLOW_ROLES)
+
+
+def test_snapshot_rejects_incomplete_or_extra_roles() -> None:
+    catalog = default_role_effort_catalog()
+    incomplete = {role: catalog[role] for role in ("thinker", "worker")}
+    try:
+        snapshot_role_effort_catalog(incomplete)
+    except EffortProfileError:
+        pass
+    else:
+        raise AssertionError("incomplete catalog must fail closed")
+    extra = dict(catalog)
+    extra["critic"] = catalog["judge"]
+    try:
+        snapshot_role_effort_catalog(extra)
+    except EffortProfileError:
+        return
+    raise AssertionError("extra catalog roles must fail closed")
+
+
+def test_true_theta_rmse_improves_with_effort_not_temperature() -> None:
+    true_theta = (-1.5, -0.5, 0.0, 0.5, 1.5)
+    baseline = estimate_theta_rmse(
+        true_theta,
+        reasoning_effort="medium",
+        extra_workflow_steps=0,
+        temperature=0.2,
+    )
+    high_effort = estimate_theta_rmse(
+        true_theta,
+        reasoning_effort="high",
+        extra_workflow_steps=0,
+        temperature=0.2,
+    )
+    temperature_proxy = estimate_theta_rmse(
+        true_theta,
+        reasoning_effort="medium",
+        extra_workflow_steps=0,
+        temperature=1.0,
+    )
+    assert high_effort < baseline
+    assert math.isclose(temperature_proxy, baseline, rel_tol=1e-9, abs_tol=1e-12)
+    estimate = estimate_theta(
+        true_theta,
+        reasoning_effort="medium",
+        extra_workflow_steps=0,
+        extra_recursion_depth=0,
+        access_list_scope="role",
+        temperature=0.2,
+    )
+    assert len(estimate.estimated_theta) == len(true_theta)
+    residuals = [
+        hat - value for hat, value in zip(estimate.estimated_theta, true_theta)
+    ]
+    expected_rmse = math.sqrt(sum(error * error for error in residuals) / len(residuals))
+    assert math.isclose(estimate.rmse, expected_rmse, rel_tol=1e-12, abs_tol=1e-12)
+
+
+def test_true_theta_values_change_estimated_rmse() -> None:
+    compact = estimate_theta_rmse(
+        (-1.5, 0.0, 1.5),
+        reasoning_effort="medium",
+        extra_workflow_steps=0,
+        temperature=0.2,
+    )
+    wide = estimate_theta_rmse(
+        (100.0, 200.0, 300.0),
+        reasoning_effort="medium",
+        extra_workflow_steps=0,
+        temperature=0.2,
+    )
+    assert compact != wide
+
+
+def test_empty_true_theta_fails_closed() -> None:
+    try:
+        estimate_theta_rmse((), reasoning_effort="medium", extra_workflow_steps=0, temperature=0.2)
+    except EffortProfileError:
+        return
+    raise AssertionError("empty true_theta must fail closed")
+
+
+def test_access_list_scope_changes_rmse() -> None:
+    true_theta = (-1.2, -0.4, 0.2, 0.8, 1.4)
+    role_scope = estimate_theta_rmse(
+        true_theta,
+        reasoning_effort="high",
+        extra_workflow_steps=3,
+        extra_recursion_depth=0,
+        access_list_scope="role",
+        temperature=0.2,
+    )
+    workflow_scope = estimate_theta_rmse(
+        true_theta,
+        reasoning_effort="high",
+        extra_workflow_steps=3,
+        extra_recursion_depth=0,
+        access_list_scope="workflow",
+        temperature=0.2,
+    )
+    assert role_scope != workflow_scope
+
+
+def test_equal_budget_ablation_keeps_production_default_locked() -> None:
+    report = run_equal_budget_ablation(true_theta=(-1.2, -0.4, 0.2, 0.8, 1.4))
+    assert report["single_model_baseline"]["mode"] == "route"
+    assert report["role_differentiated"]["mode"] == "conduct"
+    assert report["role_differentiated"]["budget_tokens"] == report["single_model_baseline"]["budget_tokens"]
+    assert set(report["one_factor_ablations"]) >= {
+        "reasoning_effort",
+        "temperature",
+        "recursion_depth",
+        "workflow_steps",
+        "access_list_scope",
+    }
+    assert report["route_versus_conduct"]["route"]["rmse"] >= 0
+    assert report["route_versus_conduct"]["conduct"]["rmse"] >= 0
+    assert report["measurement_status"] == "estimated"
+    assert production_default_change_allowed(report) is False
+    assert PRODUCTION_RMSE_IMPROVEMENT_THRESHOLD > 0
+    assert report["one_factor_ablations"]["access_list_scope"]["role"] != report[
+        "one_factor_ablations"
+    ]["access_list_scope"]["workflow"]
+    assert "estimated_theta" in report["single_model_baseline"]
+    assert len(report["single_model_baseline"]["estimated_theta"]) == 5
+    assert report["single_model_baseline"]["estimated_tokens_used"] <= report[
+        "single_model_baseline"
+    ]["budget_tokens"]
+    assert report["role_differentiated"]["estimated_tokens_used"] <= report[
+        "role_differentiated"
+    ]["budget_tokens"]
+
+
+def test_production_gate_rejects_junk_and_estimated_status() -> None:
+    assert production_default_change_allowed({}) is False
+    report = run_equal_budget_ablation(true_theta=(-1.2, -0.4, 0.2, 0.8, 1.4))
+    unlocked = dict(report)
+    unlocked["robustness_passed"] = True
+    unlocked["measurement_status"] = "estimated"
+    assert production_default_change_allowed(unlocked) is False
+
+
+def test_profile_validation_covers_numeric_bounds_and_fallbacks() -> None:
+    invalid_profiles = (
+        {"max_output_tokens": -1},
+        {"max_calls": 0},
+        {"max_workflow_steps": 0},
+        {"temperature": -0.1},
+        {"temperature": 2.1},
+        {"top_p": 0.0},
+        {"top_p": 1.1},
+        {"access_list_scope": "private"},
+        {"unsupported_provider_fallback": "retry"},
+        {"profile_version": "reasoning_effort_profile.v2"},
+    )
+    for invalid in invalid_profiles:
+        try:
+            parse_reasoning_effort_profile(invalid)
+        except EffortProfileError:
+            continue
+        raise AssertionError(f"invalid profile was accepted: {invalid!r}")
+
+
+def test_request_profile_handles_omission_and_wrong_type() -> None:
+    payload = apply_request_profile(
+        {}, None, supports_reasoning_effort=False, default_max_output_tokens=99
+    )
+    assert payload == {"max_tokens": 99}
+    try:
+        apply_request_profile(
+            {}, object(), supports_reasoning_effort=True, default_max_output_tokens=99
+        )
+    except EffortProfileError:
+        return
+    raise AssertionError("non-profile objects must fail closed")
+
+
+def test_estimator_rejects_invalid_factors_and_budget_overflow() -> None:
+    invalid_calls = (
+        {"reasoning_effort": "unknown"},
+        {"reasoning_effort": "medium", "access_list_scope": "private"},
+        {"reasoning_effort": "medium", "extra_workflow_steps": -1},
+        {"reasoning_effort": "medium", "extra_recursion_depth": -1},
+    )
+    for kwargs in invalid_calls:
+        try:
+            estimate_theta_rmse(
+                (1.0,),
+                extra_workflow_steps=kwargs.pop("extra_workflow_steps", 0),
+                extra_recursion_depth=kwargs.pop("extra_recursion_depth", 0),
+                temperature=0.2,
+                **kwargs,
+            )
+        except EffortProfileError:
+            continue
+        raise AssertionError(f"invalid estimator factors were accepted: {kwargs!r}")
+    try:
+        _estimated_tokens_used("high", 4, 2, 1)
+    except EffortProfileError:
+        pass
+    else:
+        raise AssertionError("budget overflow must fail closed")
+
+
+def test_snapshot_rejects_wrong_profile_type_and_release_gate_is_strict() -> None:
+    catalog = default_role_effort_catalog()
+    catalog["judge"] = object()  # type: ignore[assignment]
+    try:
+        snapshot_role_effort_catalog(catalog)
+    except EffortProfileError:
+        pass
+    else:
+        raise AssertionError("snapshot accepted a non-profile role")
+    measured = {
+        "single_model_baseline": {"rmse": 1.0},
+        "role_differentiated": {"rmse": 0.9},
+        "measurement_status": "measured",
+        "robustness_passed": False,
+    }
+    assert production_default_change_allowed(measured) is False
+    measured["robustness_passed"] = True
+    assert production_default_change_allowed(measured) is False
+    measured["role_differentiated"] = {"rmse": 0.1}
+    assert production_default_change_allowed(measured) is True
+    measured["single_model_baseline"] = {"rmse": float("nan")}
+    assert production_default_change_allowed(measured) is False
+
+
+def test_request_profile_separates_native_effort_and_sampling_controls() -> None:
+    profile = parse_reasoning_effort_profile(
+        {
+            "reasoning_effort": "high",
+            "max_output_tokens": 321,
+            "temperature": 0.7,
+            "top_p": 0.8,
+            "seed": 11,
+        }
+    )
+    payload = apply_request_profile(
+        {"model": "provider-model", "stream": False},
+        profile,
+        supports_reasoning_effort=True,
+        default_max_output_tokens=2048,
+    )
+    assert payload["reasoning_effort"] == "high"
+    assert payload["max_tokens"] == 321
+    assert payload["temperature"] == 0.7
+    assert payload["top_p"] == 0.8
+    assert payload["seed"] == 11
+
+
+def test_request_profile_fails_closed_or_omits_when_support_is_unproven() -> None:
+    profile = parse_reasoning_effort_profile({"reasoning_effort": "high"})
+    try:
+        apply_request_profile(
+            {}, profile, supports_reasoning_effort=False, default_max_output_tokens=2048
+        )
+    except EffortProfileError as exc:
+        assert "unproven" in str(exc)
+    else:
+        raise AssertionError("unproven provider effort must fail closed")
+    omitted = parse_reasoning_effort_profile(
+        {"reasoning_effort": "high", "unsupported_provider_fallback": "omit"}
+    )
+    payload = apply_request_profile(
+        {}, omitted, supports_reasoning_effort=False, default_max_output_tokens=2048
+    )
+    assert "reasoning_effort" not in payload
+    assert payload["max_tokens"] == omitted.max_output_tokens
+
+
+def test_opt_in_catalog_attaches_identical_snapshot_on_route_and_conduct() -> None:
+    catalog = default_role_effort_catalog()
+    expected = snapshot_role_effort_catalog(catalog).snapshot_hash
+    orchestrator = TaskOrchestrator(
+        [
+            ModelAgent("planner_agent", "mock-planner", tags=("planning", "reasoning")),
+            ModelAgent("builder_agent", "mock-builder", tags=("coding", "implementation"), priority=1),
+            ModelAgent("reviewer_agent", "mock-reviewer", tags=("verification", "security", "review"), priority=2),
+        ],
+        role_effort_catalog=catalog,
+    )
+    routed = orchestrator.complete([{"role": "user", "content": "Write one sentence."}], mode="route")
+    conducted = orchestrator.complete(
+        [{"role": "user", "content": "Analyze the architecture, implement the code, and verify risks."}],
+        mode="conduct",
+    )
+    assert routed["reasoning_effort_snapshot"]["snapshot_hash"] == expected
+    assert conducted["reasoning_effort_snapshot"]["snapshot_hash"] == expected
+    assert routed["reasoning_effort_snapshot"]["profile_version"] == PROFILE_VERSION
+    defaulted = TaskOrchestrator(
+        [ModelAgent("planner_agent", "mock-planner", tags=("planning", "reasoning"))]
+    ).complete([{"role": "user", "content": "Write one sentence."}], mode="route")
+    assert "reasoning_effort_snapshot" not in defaulted
+
+
+def test_persisted_run_stream_and_batch_keep_snapshot() -> None:
+    catalog = default_role_effort_catalog()
+    expected = snapshot_role_effort_catalog(catalog).snapshot_hash
+    orchestrator = TaskOrchestrator(
+        [ModelAgent("planner_agent", "mock-planner", tags=("planning", "reasoning"))],
+        role_effort_catalog=catalog,
+    )
+    persisted = orchestrator.run(
+        [{"role": "user", "content": "Write one sentence."}],
+        mode="route",
+    )
+    assert persisted["reasoning_effort_snapshot"]["snapshot_hash"] == expected
+    stream_id = "run_effort_stream_persist"
+    list(
+        orchestrator.stream_route(
+            [{"role": "user", "content": "Write one sentence."}],
+            workflow_run_id=stream_id,
+        )
+    )
+    streamed = orchestrator.get_workflow_run(stream_id)
+    assert streamed["reasoning_effort_snapshot"]["snapshot_hash"] == expected
+    batched = orchestrator.batch_route(["Write one sentence."])
+    assert batched[0]["reasoning_effort_snapshot"]["snapshot_hash"] == expected
+
+
+def test_doctoring_cites_fugu_trinity_conductor_apa7() -> None:
+    root = Path(__file__).resolve().parents[1]
+    architecture = (root / "docs" / "architecture.md").read_text(encoding="utf-8")
+    papers = (root / "docs" / "papers" / "README.md").read_text(encoding="utf-8")
+    for document in (architecture, papers):
+        assert "Sakana AI. (2026)" in document
+        assert "Xu, J." in document
+        assert "Nielsen, S." in document
+        assert "arXiv:2512.04695" in document
+        assert "arXiv:2512.04388" in document
+        assert "reasoning_effort_profile" in document
+        assert "Baker, F. B. (2001)" in document
+
+
+if __name__ == "__main__":  # pragma: no cover
+    test_parse_rejects_unknown_profile_keys()
+    test_parse_rejects_nan_infinity_and_boolean_numbers()
+    test_reasoning_effort_is_not_a_temperature_proxy()
+    test_parse_rejects_missing_payload_and_blank_version()
+    test_parse_requires_a_json_object()
+    test_parse_rejects_fractional_seed()
+    test_default_catalog_binds_every_workflow_role()
+    test_catalog_snapshot_is_stable_and_replayable()
+    test_snapshot_rejects_incomplete_or_extra_roles()
+    test_true_theta_rmse_improves_with_effort_not_temperature()
+    test_true_theta_values_change_estimated_rmse()
+    test_empty_true_theta_fails_closed()
+    test_access_list_scope_changes_rmse()
+    test_equal_budget_ablation_keeps_production_default_locked()
+    test_production_gate_rejects_junk_and_estimated_status()
+    test_profile_validation_covers_numeric_bounds_and_fallbacks()
+    test_request_profile_handles_omission_and_wrong_type()
+    test_estimator_rejects_invalid_factors_and_budget_overflow()
+    test_snapshot_rejects_wrong_profile_type_and_release_gate_is_strict()
+    test_request_profile_separates_native_effort_and_sampling_controls()
+    test_request_profile_fails_closed_or_omits_when_support_is_unproven()
+    test_opt_in_catalog_attaches_identical_snapshot_on_route_and_conduct()
+    test_persisted_run_stream_and_batch_keep_snapshot()
+    test_doctoring_cites_fugu_trinity_conductor_apa7()
+    print("ok")