Skip to content
Closed
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .github/workflows/fuzz.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
5 changes: 5 additions & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
<!-- END cwl-agent-guidance -->
43 changes: 43 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -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
1 change: 1 addition & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`).
Expand Down
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions conductor/tracks.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
132 changes: 84 additions & 48 deletions contextual_orchestrator/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down Expand Up @@ -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).
Expand All @@ -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().
Expand Down Expand Up @@ -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

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

run and batch_route call _store.save after _with_effort_snapshot. stream_route only writes _workflow_runs. With --state-db, a process restart drops the streamed run and its reasoning_effort_snapshot while route/batch snapshots survive.

Persist this record when self._store is not None, same as the other two paths this slice claims to keep.

Landed on #662.

self._run_order.appendleft(record["workflow_run_id"])
self._append_audit_event(
Expand All @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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).
Expand Down
Loading
Loading