diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml index cfa62adad..3fea21866 100644 --- a/.github/workflows/fuzz.yml +++ b/.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 diff --git a/AGENTS.md b/AGENTS.md index 7ced99577..c32bdf58c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -108,4 +108,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. diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 000000000..364825c3f --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,43 @@ +# 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. + +### 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 f893b5f7b..a060610fe 100644 --- a/CLAUDE.md +++ b/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(θ̂, θ). 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 65f57dd4c..979bd3c2b 100644 --- a/README.md +++ b/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 diff --git a/conductor/tracks.md b/conductor/tracks.md index 968c08ef8..b377ec28c 100644 --- a/conductor/tracks.md +++ b/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, snapshot on run/stream/batch, production defaults locked | diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 0097b722e..a6be8aab0 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/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(). @@ -945,20 +954,22 @@ 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"]) self._append_audit_event( @@ -983,17 +994,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 +1075,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 +1372,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 +1450,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). diff --git a/contextual_orchestrator/reasoning_effort_profile.py b/contextual_orchestrator/reasoning_effort_profile.py new file mode 100644 index 000000000..38638f765 --- /dev/null +++ b/contextual_orchestrator/reasoning_effort_profile.py @@ -0,0 +1,516 @@ +"""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 dataclasses import asdict, dataclass +from typing import Any, Iterable, Mapping + +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.""" + + +@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") + 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] = profile.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 c0f63a81e..0c44bebdb 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 @@ -37,6 +40,14 @@ This repository implements the interface and control plane, not the trained coor - `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 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 deterministic keyword scoring so the repo runs without training data, GPUs, or vendor credentials. 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/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//`. 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/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 d0c344462..59a3c8f22 100644 --- a/fuzz/targets.py +++ b/fuzz/targets.py @@ -18,6 +18,10 @@ over arbitrary trace payloads (regex + recursion). 4. ``orchestrator.TaskOrchestrator.run`` (+ ``sse_stream_body``) -- end-to-end prompt processing on a mock (offline) provider. +5. ``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. """ @@ -25,6 +29,7 @@ from __future__ import annotations import json +import math from typing import Any from contextual_orchestrator import server @@ -36,6 +41,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. @@ -188,3 +201,33 @@ def exercise_orchestration(prompt: str, mode: str) -> None: continue assert frame.startswith("data: ") 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: + profile = parse_reasoning_effort_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" diff --git a/tests/fuzz/test_fuzz_properties.py b/tests/fuzz/test_fuzz_properties.py index 7e7b3f347..4a122feb8 100644 --- a/tests/fuzz/test_fuzz_properties.py +++ b/tests/fuzz/test_fuzz_properties.py @@ -19,6 +19,7 @@ from fuzz.targets import ( exercise_agent_config, exercise_orchestration, + exercise_reasoning_effort_profile, exercise_redaction, exercise_request_body, ) @@ -108,3 +109,9 @@ def test_redaction_never_crashes_and_is_idempotent(text: str) -> None: ) def test_orchestration_on_arbitrary_prompt(prompt: str, mode: str) -> None: exercise_orchestration(prompt, mode) + + +@_SETTINGS +@given(_json_values) +def test_reasoning_effort_profile_never_crashes(value: object) -> None: + exercise_reasoning_effort_profile(value) diff --git a/tests/test_reasoning_effort_profile.py b/tests/test_reasoning_effort_profile.py new file mode 100644 index 000000000..a4d2c0663 --- /dev/null +++ b/tests/test_reasoning_effort_profile.py @@ -0,0 +1,338 @@ +"""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, + 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, +) + + +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 + + +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_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_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_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_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")