feat: allow templating per-agent reasoning.effort and context_tier (#262)

src/conductor/config/schema.py

@@ -6,7 +6,7 @@
 
 from __future__ import annotations
 
-from typing import Any, Literal
+from typing import Any, Literal, get_args
 
 from pydantic import (
     BaseModel,
@@ -21,6 +21,7 @@
 from conductor.duration import parse_duration
 from conductor.providers.context_tier import ContextTier
 from conductor.providers.reasoning import ReasoningEffort
+from conductor.templating import is_jinja_template
 
 BudgetMode = Literal["audit", "enforce"]
 """How the engine responds when a workflow cost budget is exceeded.
@@ -563,10 +564,49 @@ class ReasoningConfig(BaseModel):
 
         reasoning:
           effort: high
+
+    Supports Jinja2 templates::
+
+        reasoning:
+          effort: "{{ workflow.input.effort }}"
+
+    A templated ``effort`` is accepted at load time and resolved + validated
+    at runtime (in :mod:`conductor.executor.agent`), mirroring how ``model``
+    and the ``wait`` step's ``duration`` are handled. A *literal* value must
+    be one of :data:`~conductor.providers.reasoning.ReasoningEffort`.
+    """
+
+    effort: ReasoningEffort | str
+    """Reasoning effort level applied to the agent's model calls.
+
+    Either a literal level (``low`` / ``medium`` / ``high`` / ``xhigh``) or a
+    ``{{ ... }}`` Jinja2 template resolved at runtime.
     """
 
-    effort: ReasoningEffort
-    """Reasoning effort level applied to the agent's model calls."""
+    @model_validator(mode="after")
+    def _validate_effort(self) -> ReasoningConfig:
+        """Accept literal efforts or defer ``{{ }}`` / ``{% %}`` templates.
+
+        A templated value (detected by
+        :func:`~conductor.templating.is_jinja_template`, matching ``{{`` or
+        ``{%``) skips literal validation here and is rendered + validated at
+        execute time (:mod:`conductor.executor.agent`, the same place the
+        ``model`` field is rendered). A non-templated value must be a valid
+        :data:`ReasoningEffort` literal.
+
+        Note: this is a broader check than
+        :meth:`AgentDef._validate_wait_duration`, which intentionally matches
+        only ``{{``.
+        """
+        value = self.effort
+        if is_jinja_template(value):
+            return self
+        if value not in get_args(ReasoningEffort):
+            raise ValueError(
+                f"reasoning.effort must be one of {list(get_args(ReasoningEffort))} "
+                f"or a '{{{{ ... }}}}' template (got {value!r})"
+            )
+        return self
 
 
 class AgentDef(BaseModel):
@@ -642,7 +682,7 @@ class AgentDef(BaseModel):
     Supports Jinja2 templates: {{ workflow.input.model_name }}
     """
 
-    context_tier: ContextTier | None = None
+    context_tier: ContextTier | str | None = None
     """Context-window tier for models that support it (Copilot provider only).
 
     Set ``context_tier: long_context`` to pin a heavy-reasoning agent to the
@@ -657,9 +697,18 @@ class AgentDef(BaseModel):
 
     Only applies to provider-backed agents (type='agent' or None).
 
+    Supports Jinja2 templates: a ``{{ workflow.input.tier }}`` value is
+    accepted at load time and resolved + validated at runtime (mirrors
+    ``model`` and the ``reasoning.effort`` handling). A *literal* value must
+    be one of :data:`~conductor.providers.context_tier.ContextTier`.
+
     Example YAML::
 
         context_tier: long_context
+
+    Templated::
+
+        context_tier: "{{ workflow.input.tier }}"
     """
 
     input: list[str] = Field(default_factory=list)
@@ -1417,6 +1466,10 @@ def validate_agent_type(self) -> AgentDef:
                     f"'{self.type or 'agent'}' agents cannot have 'output_type' "
                     "(only 'set' agents support output_type)"
                 )
+            # #262: regular agents may carry a literal or templated
+            # context_tier; validate the literal here and defer templates to
+            # runtime. (reasoning.effort is validated on ReasoningConfig.)
+            self._validate_context_tier()
         if self.type == "workflow" and self.reasoning is not None:
             raise ValueError("workflow agents cannot have 'reasoning'")
         if self.type == "workflow" and self.context_tier is not None:
@@ -1458,6 +1511,33 @@ def effective_output_schema(self) -> dict[str, OutputField] | None:
             return self.output
         return None
 
+    def _validate_context_tier(self) -> None:
+        """Validate ``context_tier`` for a regular (provider-backed) agent.
+
+        An unset (``None``) or templated value (detected by
+        :func:`~conductor.templating.is_jinja_template`, matching ``{{`` or
+        ``{%``) defers all literal validation to runtime (rendered + validated
+        in :mod:`conductor.executor.agent`, alongside ``model``); a
+        non-templated value must be a valid
+        :data:`~conductor.providers.context_tier.ContextTier` literal.
+
+        This differs from :meth:`_validate_wait_duration` on two counts: that
+        method matches only ``{{``, and it does not defer ``None``.
+
+        Non-agent step types reject ``context_tier`` outright via their own
+        ``is not None`` checks in :meth:`validate_agent_type` (a template
+        string is still "not None"), so this helper is only dispatched from
+        the regular-agent branch.
+        """
+        value = self.context_tier
+        if value is None or is_jinja_template(value):
+            return
+        if value not in get_args(ContextTier):
+            raise ValueError(
+                f"context_tier must be one of {list(get_args(ContextTier))} "
+                f"or a '{{{{ ... }}}}' template (got {value!r})"
+            )
+
     def _validate_wait_duration(self) -> None:
         """Validate ``duration`` for a ``wait`` agent.
 

src/conductor/config/validator.py

@@ -16,6 +16,7 @@
 
 from conductor.exceptions import ConfigurationError
 from conductor.providers.capabilities import ProviderCapabilities, get_capabilities
+from conductor.templating import is_jinja_template
 
 if TYPE_CHECKING:
     from conductor.config.schema import AgentDef, WorkflowConfig
@@ -1705,12 +1706,23 @@ def _caps_for(name: str) -> ProviderCapabilities | None:
                 if (agent.reasoning is not None and agent.reasoning.effort is not None)
                 else "runtime.default_reasoning_effort"
             )
+            # #262: whether the provider supports reasoning effort AT ALL is a
+            # value-INDEPENDENT fact known at validate time, so reject even a
+            # templated effort here — no resolved value could ever be valid on
+            # a provider with reasoning_effort=None (e.g. claude-agent-sdk,
+            # which ignores reasoning entirely). Only the membership check
+            # (does the resolved literal fall in the supported subset?) is
+            # value-dependent and must be deferred for templates; providers
+            # that support reasoning re-validate the resolved value at runtime
+            # (copilot._validate_reasoning_effort_for_model / claude thinking).
             if supported is None:
                 errors.append(
                     f"Agent '{agent.name}' resolves to {source}={requested!r} "
                     f"but provider '{provider_name}' does not support reasoning "
                     f"effort (capabilities.reasoning_effort=None)."
                 )
+            elif is_jinja_template(requested):
+                pass
             elif requested not in supported:
                 errors.append(
                     f"Agent '{agent.name}' resolves to {source}={requested!r} "

src/conductor/executor/agent.py

@@ -8,12 +8,15 @@
 
 import asyncio
 import contextlib
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, get_args
 
 from conductor.exceptions import ValidationError
 from conductor.executor.output import parse_json_output, validate_output
 from conductor.executor.template import TemplateRenderer
 from conductor.providers.base import AgentOutput, EventCallback
+from conductor.providers.context_tier import ContextTier
+from conductor.providers.reasoning import ReasoningEffort
+from conductor.templating import is_jinja_template
 
 
 def _verbose_log(message: str, style: str = "dim") -> None:
@@ -112,6 +115,50 @@ def __init__(
         self.instructions_preamble = instructions_preamble
         self.renderer = TemplateRenderer()
 
+    def _render_enum_field(
+        self,
+        *,
+        value: str,
+        context: dict[str, Any],
+        allowed: tuple[str, ...],
+        field_name: str,
+        agent_name: str,
+    ) -> str:
+        """Render a templated enum field and validate the resolved literal.
+
+        Mirrors the ``model`` rendering above: a ``{{ ... }}`` value is
+        rendered with the full agent context, stripped (the renderer keeps
+        trailing newlines), and checked against ``allowed``. Raises a
+        :class:`~conductor.exceptions.ValidationError` when the resolved
+        value is not one of the permitted literals so the failure is actionable
+        at execute time rather than silently forwarded to the provider/SDK.
+        """
+        resolved = self.renderer.render(value, context).strip()
+        if resolved not in allowed:
+            if not resolved:
+                # An empty resolution is almost always a conditional template
+                # (``{% if ... %}``) with no matching branch. Fail closed — the
+                # same way a non-empty invalid value (below) and the
+                # provider-side resolver guards do — rather than silently
+                # treating empty as "unset": to fall back to the runtime
+                # default, omit the field or add an else-branch emitting the
+                # desired literal.
+                raise ValidationError(
+                    f"Agent '{agent_name}': {field_name} template resolved to an empty value.",
+                    suggestion=(
+                        f"A conditional template with no matching branch "
+                        f"produced nothing. Emit one of {list(allowed)}, add an "
+                        f"else-branch, or omit {field_name} to use the runtime "
+                        f"default."
+                    ),
+                )
+            raise ValidationError(
+                f"Agent '{agent_name}': {field_name} template resolved to "
+                f"{resolved!r}, which is not a valid value.",
+                suggestion=f"Resolved value must be one of {list(allowed)}.",
+            )
+        return resolved
+
     async def execute(
         self,
         agent: AgentDef,
@@ -150,10 +197,47 @@ async def execute(
             ValidationError: If output doesn't match schema or tools are invalid.
         """
         # Render model field if it contains template expressions
-        if agent.model and ("{{" in agent.model or "{%" in agent.model):
+        if is_jinja_template(agent.model):
             rendered_model = self.renderer.render(agent.model, context)
             agent = agent.model_copy(update={"model": rendered_model})
 
+        # #262: resolve templated reasoning.effort / context_tier the same
+        # way model is handled above. These fields are strict ``Literal``
+        # aliases that the schema deliberately accepts as templates (deferring
+        # literal validation to here); render the value with full context, then
+        # validate the resolved literal so the provider sees a concrete value.
+        # ``is_jinja_template`` both detects templates and narrows the widened
+        # ``ReasoningEffort | str`` / ``ContextTier | str | None`` field types
+        # to ``str`` for the type checker before the value reaches
+        # ``_render_enum_field``. (``ReasoningEffort`` and ``ContextTier`` are
+        # ``Literal`` aliases, not ``Enum`` types — hence the ``get_args``
+        # calls below.)
+        effort = agent.reasoning.effort if agent.reasoning is not None else None
+        if is_jinja_template(effort):
+            resolved_effort = self._render_enum_field(
+                value=effort,
+                context=context,
+                allowed=get_args(ReasoningEffort),
+                field_name="reasoning.effort",
+                agent_name=agent.name,
+            )
+            # ``agent.reasoning`` is not None here (effort came from it).
+            assert agent.reasoning is not None
+            agent = agent.model_copy(
+                update={"reasoning": agent.reasoning.model_copy(update={"effort": resolved_effort})}
+            )
+
+        tier = agent.context_tier
+        if is_jinja_template(tier):
+            resolved_tier = self._render_enum_field(
+                value=tier,
+                context=context,
+                allowed=get_args(ContextTier),
+                field_name="context_tier",
+                agent_name=agent.name,
+            )
+            agent = agent.model_copy(update={"context_tier": resolved_tier})
+
         # Render prompt with context
         rendered_prompt = self.renderer.render(agent.prompt, context)
 

src/conductor/providers/context_tier.py

@@ -13,7 +13,9 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Literal, cast
+
+from conductor.templating import is_jinja_template
 
 if TYPE_CHECKING:
     from conductor.config.schema import AgentDef
@@ -39,5 +41,18 @@ def resolve_context_tier(
         The resolved ``ContextTier``, or ``None`` to send no value.
     """
     if agent.context_tier is not None:
-        return agent.context_tier
+        # #262: AgentDef widens ``context_tier`` to ``ContextTier | str`` so a
+        # ``{{ ... }}`` / ``{% ... %}`` template survives schema validation. By
+        # the time this resolver runs (provider execute, after AgentExecutor
+        # renders + validates the field) the value is a concrete, validated
+        # literal. Guard the invariant so an unrendered template raises here
+        # rather than being cast straight to the SDK. This matters more than for
+        # reasoning.effort: Copilot forwards the tier to the SDK unvalidated
+        # (no advertised supported_context_tiers, so the SDK is the sole
+        # authority — see ``CopilotProvider``), so a leaked template would
+        # otherwise reach the SDK raw.
+        tier = agent.context_tier
+        if is_jinja_template(tier):
+            raise ValueError(f"context_tier reached the provider unresolved: {tier!r}")
+        return cast(ContextTier, tier)
     return runtime_default

src/conductor/providers/reasoning.py

@@ -11,7 +11,9 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-from typing import TYPE_CHECKING, Final, Literal
+from typing import TYPE_CHECKING, Final, Literal, cast
+
+from conductor.templating import is_jinja_template
 
 if TYPE_CHECKING:
     from conductor.config.schema import AgentDef
@@ -87,5 +89,14 @@ def resolve_reasoning_effort(
     set, signalling that no reasoning parameter should be sent to the SDK.
     """
     if agent.reasoning is not None:
-        return agent.reasoning.effort
+        # #262: ``ReasoningConfig`` widens ``effort`` to ``ReasoningEffort | str``
+        # so a ``{{ ... }}`` / ``{% ... %}`` template survives schema validation.
+        # By the time this resolver runs (provider execute, after AgentExecutor
+        # renders + validates the field) the value is a concrete, validated
+        # literal. Guard the invariant so an unrendered template raises here
+        # rather than being cast straight to the SDK.
+        effort = agent.reasoning.effort
+        if is_jinja_template(effort):
+            raise ValueError(f"reasoning.effort reached the provider unresolved: {effort!r}")
+        return cast(ReasoningEffort, effort)
     return runtime_default

src/conductor/templating.py

@@ -0,0 +1,26 @@
+"""Jinja template detection, shared across schema validation, the executor's
+render step, and the provider resolvers.
+
+Kept as a leaf module (no intra-``conductor`` imports) so it can be imported
+from any layer without risking an import cycle.
+"""
+
+from __future__ import annotations
+
+from typing import TypeGuard
+
+
+def is_jinja_template(value: object) -> TypeGuard[str]:
+    """Return ``True`` if ``value`` is a string carrying a Jinja expression
+    (``{{ ... }}``) or statement (``{% ... %}``) marker.
+
+    Returns a :data:`~typing.TypeGuard` of ``str`` rather than a plain ``bool``
+    so a caller that branches on it narrows the value to ``str`` — the
+    executor's ``_render_enum_field`` path relies on that narrowing.
+
+    Note: this matches *both* ``{{`` and ``{%``.
+    :meth:`AgentDef._validate_wait_duration` deliberately checks only ``{{``
+    (it defers expression templates but not statement templates) and is
+    intentionally *not* routed through this helper.
+    """
+    return isinstance(value, str) and ("{{" in value or "{%" in value)