fix(routing): fail-closed ablation θ and persist streamed snapshots

.github/workflows/fuzz.yml

@@ -83,6 +83,9 @@ jobs:
       - name: Fuzz orchestration engine
         run: python fuzz/fuzz_orchestration.py -max_total_time=${FUZZ_SECONDS} -artifact_prefix=crash- fuzz/corpus/orchestration
 
+      - 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

AGENTS.md

@@ -108,4 +108,12 @@ 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. `run_equal_budget_ablation` must reject boolean/string θ
+  with the same helper as `estimate_theta`. `stream_route` must persist
+  the snapshot through `--state-db`. Do not change production
+  route/conduct defaults until `production_default_change_allowed` is
+  true. Temperature is not effort.
 <!-- END cwl-agent-guidance -->

CHANGELOG.md

@@ -0,0 +1,50 @@
+# Changelog
+
+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
+
+- 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 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.
+- `run_equal_budget_ablation` now uses the same fail-closed true-θ
+  validator as `estimate_theta`, so boolean or string θ cannot be
+  laundered into an RMSE report. The Hypothesis/Atheris target pops
+  `true_theta` before parse and always exercises the ablation after a
+  valid profile. `stream_route` writes the run (and its
+  `reasoning_effort_snapshot`) through `--state-db` the same way `run`
+  and `batch_route` already do.
+
+### 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

CLAUDE.md

@@ -86,6 +86,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(θ̂, θ). Ablation true-θ uses the same fail-closed numeric helper as `estimate_theta`. `stream_route` persists the snapshot through `--state-db`. 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`).

README.md

@@ -252,6 +252,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 tests/test_reasoning_effort_profile.py
 python tests/test_admin_contract.py
 python tests/test_conventions.py
 python tests/test_api_contract.py

conductor/tracks.md

@@ -4,3 +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-reasoning-effort-profiles | active | Issue #568: versioned per-role `reasoning_effort_profile`, equal-budget θ̂ RMSE ablation, fail-closed true-θ, snapshot on run/stream/batch including `--state-db` restart, production defaults locked |

contextual_orchestrator/orchestrator.py

@@ -27,6 +27,10 @@
 
 from .conventions import require_object_name
 from .credentials import NotConfigured, get_credential
+from .reasoning_effort_profile import (
+    ReasoningEffortProfile,
+    snapshot_role_effort_catalog,
+)
 
 
 ChatMessage = dict[str, str]
@@ -823,6 +827,7 @@ def __init__(
         agents_db: str | None = None,
         cache_ttl: float = 0.0,
         cache_max_entries: int = 256,
+        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).
@@ -835,6 +840,10 @@ def __init__(
             raise ValueError("at least one enabled agent is required")
         self.client = client or ModelClient()
         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().
@@ -935,8 +944,11 @@ def would_route(self, messages: list[ChatMessage], mode: str = "auto") -> bool:
     def stream_route(self, messages: list[ChatMessage], workflow_run_id: str | None = None):
         """Stream a single worker's content deltas as they arrive, then persist the run.
 
-        True streaming for the route path. ponytail: no cross-agent failover here — bytes
-        already sent can't be recalled, so a mid-stream provider failure surfaces to the caller.
+        True streaming for the route path. When ``--state-db`` is set the finished
+        record (including an opt-in ``reasoning_effort_snapshot``) is written
+        through the same store as ``run`` and ``batch_route``. ponytail: no
+        cross-agent failover here — bytes already sent can't be recalled, so a
+        mid-stream provider failure surfaces to the caller.
         """
         text = self._latest_user_text(messages)
         agent = self._select_agent(text, "worker")
@@ -945,22 +957,26 @@ def stream_route(self, messages: list[ChatMessage], workflow_run_id: str | None
             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"])
+        if self._store is not None:
+            self._store.save("workflow_run", record["workflow_run_id"], record)
         self._append_audit_event(
             "workflow_run_created",
             {"workflow_run_id": record["workflow_run_id"], "mode": "route", "agent_count": 1},
@@ -983,17 +999,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:
@@ -1062,17 +1080,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:
@@ -1357,12 +1377,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."""
@@ -1433,13 +1455,32 @@ 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 _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).