feat: add adaptive provider reasoning control

.github/workflows/reasoning-workload-verify.yml

@@ -0,0 +1,68 @@
+name: Reasoning control quality
+
+on:
+  pull_request:
+    paths:
+      - contextual_orchestrator/_reasoning_*.py
+      - contextual_orchestrator/reasoning_*.py
+      - tests/reasoning_fakes.py
+      - tests/test_reasoning_*.py
+      - .github/workflows/reasoning-workload-verify.yml
+  push:
+    branches: [main]
+    paths:
+      - contextual_orchestrator/_reasoning_*.py
+      - contextual_orchestrator/reasoning_*.py
+      - tests/reasoning_fakes.py
+      - tests/test_reasoning_*.py
+      - .github/workflows/reasoning-workload-verify.yml
+
+permissions:
+  contents: read
+
+concurrency:
+  group: reasoning-control-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  verify:
+    name: Reasoning control statement, branch, and docstring coverage
+    runs-on: ubuntu-24.04
+    timeout-minutes: 30
+    steps:
+      - name: Checkout exact source revision without credentials
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # actions/checkout@v7
+        with:
+          ref: ${{ github.event.pull_request.head.sha || github.sha }}
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # actions/setup-python@v6
+        with:
+          python-version: "3.12"
+
+      - name: Install hash-locked verification dependencies
+        run: |
+          python -m pip install --require-hashes -r fuzz/requirements-property.txt
+          python -m pip install --require-hashes -r requirements-opencode-review-ci.txt
+
+      - name: Run full tests and changed-module quality gates
+        run: |
+          python -m compileall -q contextual_orchestrator tests
+          python -m pytest -q
+          python -m coverage erase
+          python -m coverage run --branch --source=contextual_orchestrator -m pytest -q
+          python -m coverage report \
+            --include='contextual_orchestrator/_reasoning_*.py,contextual_orchestrator/reasoning_*.py' \
+            --show-missing \
+            --fail-under=100
+          python -m interrogate -f 100 \
+            contextual_orchestrator/_reasoning_workload.py \
+            contextual_orchestrator/_reasoning_policy.py \
+            contextual_orchestrator/_reasoning_state.py \
+            contextual_orchestrator/_reasoning_orchestrator_hooks.py \
+            contextual_orchestrator/_reasoning_workflow.py \
+            contextual_orchestrator/_reasoning_client_hooks.py \
+            contextual_orchestrator/reasoning_control.py \
+            contextual_orchestrator/reasoning_runtime.py
+          git diff --check

CHANGELOG.md

@@ -11,6 +11,10 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and
 - Add a transport-neutral, versioned model fallback policy that validates explicit cost tiers and deterministically exhausts eligible free candidates before any paid fallback.
 - Filter fallback candidates by repository visibility, required capability, and configured credential name without retaining or serializing secret values.
 - Add a standard-library CLI for immutable cross-repository workflow integration, with complete statement, branch, and public-docstring coverage for the fallback policy.
+- Add explicit per-agent reasoning capability profiles and provider-neutral canonical levels from `none` through `max` without inferring capability from model names.
+- Add role-aware adaptive and fixed-effort policies across routing, conducted workflows, generated planning, model verification, streaming, Responses passthrough, and Batch requests.
+- Add bounded verifier-triggered worker escalation and fixed-effort ablation evidence with provider-reported reasoning-token accounting.
+- Expose reasoning profiles through agent configuration and admin projections while preserving them across runtime agent replacement and durable re-save.
 
 ### Security
 
@@ -28,15 +32,23 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and
 - Remove fallback-policy environment-value inspection; trusted callers now declare only validated available credential names, and the policy CLI rejects the former secret-bearing environment selector.
 - Reject boolean or floating-point schema versions, unsafe programmatic agent selectors, non-string candidate identifiers, and mutable credential-control collections as controlled validation errors instead of leaking Python type exceptions or permitting post-validation policy changes.
 - Document narrowly scoped Semgrep suppressions for parameter-bound database queries, the explicit development-only TLS verification opt-out, and provider URLs that pass the egress guard.
+- Validate custom reasoning payload paths and scalar templates, preserve caller-owned reasoning fields, and record only bounded decision evidence rather than private intermediate reasoning text.
+- Reject mutable direct-constructor reasoning controls, omitted payload-rule values, and NaN or infinite provider mappings before any reasoning payload is projected.
 
 ### Changed
 
 - Pin Atheris by Python interpreter so the Python 3.11 fuzz job and the newer central coverage-evidence image both install a published, hash-locked wheel.
+- Treat reasoning effort as a third test-time-compute axis alongside model routing and workflow topology, with failover models projecting one canonical decision onto their declared capabilities.
+- Keep synthesizer effort below analysis roles by default and require multiple high-impact signals before adaptive policy reaches the model-specific maximum.
+- Make adaptive-reasoning activation explicit and idempotent: package import no longer installs reasoning hooks, while the product CLI activates them before loading agent configuration.
 - Run repository Tests, Fuzz, and Security workflows for stacked pull requests targeting any branch, bind every checkout to the literal contributor-head SHA, and keep checkout credentials non-persistent so local evidence cannot silently become absent or synthetic-merge-only evidence.
 
 ### Documentation
 
 - Add APA 7 doctoring for Python environment-marker semantics, Atheris artifact availability and hashes, and the supported-platform uncertainty boundary.
+- Add architecture, operations, test strategy, provider mapping, governance, and APA 7 research evidence for adaptive reasoning control.
+- Record the standalone/MSA activation boundary and import-purity regression contract for optional reasoning control.
+- Define the immutable direct-constructor and strict interoperable-JSON boundary for reasoning profile control data.
 - Add provider-response resource-bound doctoring covering the 8 MiB fail-closed limit, HTTP framing preflight, `text/event-stream` media-type enforcement, bounded SSE reads, OpenAI-compatible `[DONE]` completion evidence, malformed-event and premature-EOF handling, batch-output partitioning, incident handling, and operational rollback.
 - Add provider-stream UTF-8 doctoring grounding strict SSE/JSON decoding and redacted malformed-input handling in the WHATWG HTML Standard and RFC 8259, with verification, failure, rollback, and authority boundaries.
 - Add provider-JSON trust-boundary doctoring grounding strict UTF-8 object decoding, duplicate-name and non-finite-number rejection, finite-runtime numeric enforcement for extreme exponents, Batch JSONL validation, redacted parser failures, request-path authority, operator recovery, and rollback in RFC 8259, current Python documentation, and the OpenAI Batch API contract.

contextual_orchestrator/__init__.py

@@ -54,8 +54,31 @@
     load_fallback_manifest,
 )
 from .orchestrator import ModelAgent, TaskOrchestrator, WorkflowStep, load_agents
+from .reasoning_control import (
+    CANONICAL_REASONING_LEVELS,
+    PayloadRule,
+    ReasoningAblationCell,
+    ReasoningDecision,
+    ReasoningPolicy,
+    ReasoningProfile,
+    ReasoningWorkload,
+    adapt_reasoning_decision,
+    apply_reasoning_payload,
+    escalate_reasoning_decision,
+    extract_reasoning_tokens,
+    select_reasoning_decision,
+    sum_usage_tokens,
+)
+from .reasoning_runtime import (
+    agent_reasoning_profile,
+    configure_agent_reasoning,
+    configure_orchestrator_reasoning,
+    current_reasoning_decision,
+    enable_reasoning_control,
+    orchestrator_reasoning_policy,
+    reasoning_override,
+)
 from .token_counting import HeuristicTokenCounter, build_token_counter
-
 __all__ = [
     "ModelAgent",
     "TaskOrchestrator",
@@ -114,4 +137,25 @@
     "SkippedCandidate",
     "build_fallback_plan",
     "load_fallback_manifest",
+    # adaptive provider reasoning control
+    "CANONICAL_REASONING_LEVELS",
+    "PayloadRule",
+    "ReasoningAblationCell",
+    "ReasoningDecision",
+    "ReasoningPolicy",
+    "ReasoningProfile",
+    "ReasoningWorkload",
+    "adapt_reasoning_decision",
+    "apply_reasoning_payload",
+    "escalate_reasoning_decision",
+    "extract_reasoning_tokens",
+    "select_reasoning_decision",
+    "sum_usage_tokens",
+    "agent_reasoning_profile",
+    "configure_agent_reasoning",
+    "configure_orchestrator_reasoning",
+    "current_reasoning_decision",
+    "enable_reasoning_control",
+    "orchestrator_reasoning_policy",
+    "reasoning_override",
 ]

contextual_orchestrator/__main__.py

@@ -9,6 +9,7 @@
 
 from .credentials import register_credential
 from .orchestrator import ModelClient, TaskOrchestrator, load_agents
+from .reasoning_runtime import enable_reasoning_control
 from .server import SecurityConfig, serve
 
 
@@ -55,12 +56,21 @@ def _register_credential_command(argv: list[str]) -> None:
     print(json.dumps({"registered": args.name, "backend": "kv"}, ensure_ascii=False))
 
 
+def _enable_reasoning_runtime() -> None:
+    """Activate optional reasoning hooks for the executable product runtime."""
+    enable_reasoning_control()
+
+
 def main() -> None:
     """Parse CLI options and run bootstrap, prompt completion, or the HTTP server."""
     if len(sys.argv) > 1 and sys.argv[1] == "register-credential":
         _register_credential_command(sys.argv[2:])
         return
 
+    # Product execution opts in explicitly. Merely importing the package remains
+    # side-effect free, while configuration loading below understands profiles.
+    _enable_reasoning_runtime()
+
     parser = argparse.ArgumentParser(description="Route or conduct chat requests across model agents.")
     parser.add_argument("prompt", nargs="?", help="User prompt for CLI mode.")
     parser.add_argument("--agents", default="examples/agents.mock.json", help="Agent config JSON.")

contextual_orchestrator/_reasoning_client_hooks.py

@@ -0,0 +1,179 @@
+"""Provider-client hooks for reasoning payload and batch projection."""
+
+from __future__ import annotations
+
+from typing import Any, Iterator
+
+from .reasoning_control import (
+    ReasoningPolicy,
+    ReasoningWorkload,
+    adapt_reasoning_decision,
+    apply_reasoning_payload,
+    select_reasoning_decision,
+)
+from ._reasoning_state import (
+    _ACTIVE_DECISION,
+    _ACTIVE_POLICY,
+    _BATCH_DECISIONS,
+    _append_event,
+    _decision_scope,
+    _infer_role,
+    _input_text,
+    _message_text,
+    _resolve_decision,
+    agent_reasoning_profile,
+)
+from ._reasoning_workflow import _rewrite_batch_payload
+
+
+def install_client_hooks(model_client_type: type[Any]) -> None:
+    """Install chat, stream, passthrough, and Batch reasoning hooks."""
+    original_client_chat = model_client_type.chat
+    original_client_stream_chat = model_client_type.stream_chat
+    original_client_proxy_send = model_client_type.proxy_send
+    original_client_batch_chat = model_client_type.batch_chat
+    original_client_send = model_client_type._send
+    original_client_stream_send = model_client_type._stream_send
+    original_client_send_raw = model_client_type._send_raw
+    original_client_batch_upload = model_client_type._batch_upload
+
+    def client_chat(
+        self: Any,
+        agent: Any,
+        messages: list[dict[str, str]],
+        temperature: float = 0.2,
+    ) -> str:
+        """Keep one role-aware decision active through chat payload construction."""
+        role = _infer_role(messages)
+        decision = _resolve_decision(agent, _message_text(messages), role)
+        with _decision_scope(decision):
+            output = original_client_chat(self, agent, messages, temperature)
+        _append_event(agent, role, decision)
+        return output
+
+    def client_stream_chat(
+        self: Any,
+        agent: Any,
+        messages: list[dict[str, str]],
+        temperature: float = 0.2,
+    ) -> Iterator[str]:
+        """Keep one decision active until a streaming provider response completes."""
+        role = _infer_role(messages)
+        decision = _resolve_decision(agent, _message_text(messages), role)
+        with _decision_scope(decision):
+            yield from original_client_stream_chat(self, agent, messages, temperature)
+        _append_event(agent, role, decision)
+
+    def client_proxy_send(
+        self: Any,
+        agent: Any,
+        endpoint: str,
+        payload: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Apply a worker decision to full-shape chat or Responses passthrough."""
+        decision = _resolve_decision(agent, _input_text(payload), "worker")
+        with _decision_scope(decision):
+            output = original_client_proxy_send(self, agent, endpoint, payload)
+        _append_event(agent, "worker", decision)
+        return output
+
+    def client_send(self: Any, agent: Any, payload: dict[str, Any]) -> str:
+        """Project the active decision into a chat-completions payload."""
+        profile = agent_reasoning_profile(agent)
+        decision = adapt_reasoning_decision(profile, _ACTIVE_DECISION.get())
+        return original_client_send(
+            self,
+            agent,
+            apply_reasoning_payload(payload, profile, decision, "chat/completions"),
+        )
+
+    def client_stream_send(
+        self: Any,
+        agent: Any,
+        payload: dict[str, Any],
+    ) -> Iterator[str]:
+        """Project the active decision into a streaming chat payload."""
+        profile = agent_reasoning_profile(agent)
+        decision = adapt_reasoning_decision(profile, _ACTIVE_DECISION.get())
+        yield from original_client_stream_send(
+            self,
+            agent,
+            apply_reasoning_payload(payload, profile, decision, "chat/completions"),
+        )
+
+    def client_send_raw(
+        self: Any,
+        agent: Any,
+        endpoint: str,
+        payload: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Project the active decision into chat or Responses passthrough payloads."""
+        profile = agent_reasoning_profile(agent)
+        decision = adapt_reasoning_decision(profile, _ACTIVE_DECISION.get())
+        return original_client_send_raw(
+            self,
+            agent,
+            endpoint,
+            apply_reasoning_payload(payload, profile, decision, endpoint),
+        )
+
+    def client_batch_chat(
+        self: Any,
+        agent: Any,
+        requests: dict[str, list[dict[str, str]]],
+        temperature: float = 0.2,
+        poll_interval: float = 5.0,
+        poll_timeout: float = 3600.0,
+    ) -> dict[str, dict[str, Any]]:
+        """Select and retain one direct-route decision for each batch item."""
+        policy = _ACTIVE_POLICY.get() or ReasoningPolicy()
+        profile = agent_reasoning_profile(agent)
+        decisions = {
+            custom_id: decision
+            for custom_id, messages in requests.items()
+            if (
+                decision := select_reasoning_decision(
+                    profile,
+                    policy,
+                    _message_text(messages),
+                    "worker",
+                    workload=ReasoningWorkload(),
+                )
+            )
+            is not None
+        }
+        token = _BATCH_DECISIONS.set(decisions)
+        try:
+            results = original_client_batch_chat(
+                self,
+                agent,
+                requests,
+                temperature,
+                poll_interval,
+                poll_timeout,
+            )
+        finally:
+            _BATCH_DECISIONS.reset(token)
+        for custom_id in results:
+            _append_event(agent, "worker", decisions.get(custom_id))
+        return results
+
+    def client_batch_upload(self: Any, agent: Any, payload: bytes) -> str:
+        """Rewrite provider Batch JSONL immediately before the secured upload."""
+        profile = agent_reasoning_profile(agent)
+        decisions = _BATCH_DECISIONS.get()
+        if profile is not None and decisions:
+            payload = _rewrite_batch_payload(payload, decisions, profile)
+        return original_client_batch_upload(self, agent, payload)
+
+    model_client_type.chat = client_chat
+    model_client_type.stream_chat = client_stream_chat
+    model_client_type.proxy_send = client_proxy_send
+    model_client_type.batch_chat = client_batch_chat
+    model_client_type._send = client_send
+    model_client_type._stream_send = client_stream_send
+    model_client_type._send_raw = client_send_raw
+    model_client_type._batch_upload = client_batch_upload
+
+
+__all__ = ["install_client_hooks"]