From 0f856a98409a444fa3670947a0b51bf0278dc219 Mon Sep 17 00:00:00 2001
From: Brent Rusinow <brrusino@microsoft.com>
Date: Tue, 16 Jun 2026 09:27:03 -0700
Subject: [PATCH 1/3] feat(context): add per-agent context_tier (long_context /
 1M) for Copilot

Adds a unified `context_tier` knob (default | long_context) so workflows can
pin heavy-reasoning agents to a model's long-context window via the Copilot
SDK's create_session(context_tier=...) param.

Mirrors the existing two-level reasoning pattern:
- AgentDef.context_tier (per-agent override, sibling to model)
- RuntimeConfig.default_context_tier (workflow-wide default)
- resolve_context_tier() helper (per-agent wins, else runtime default)

The Copilot provider forwards the resolved value into agent and dialog
sessions; other providers ignore it. Composes independently with
reasoning.effort. Rejected on script/human_gate/workflow agents. Per-agent
context_tier is surfaced in system metadata for the dashboard/logs.

Includes schema, provider, and factory tests, an example workflow, and docs.

Closes #251

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 CHANGELOG.md                            |  14 +++
 docs/configuration.md                   |  52 +++++++++++
 docs/workflow-syntax.md                 |  14 +++
 examples/README.md                      |  17 ++++
 examples/context-tier.yaml              | 105 ++++++++++++++++++++++
 src/conductor/config/schema.py          |  38 ++++++++
 src/conductor/engine/workflow.py        |   4 +
 src/conductor/providers/context_tier.py |  43 +++++++++
 src/conductor/providers/copilot.py      |  29 ++++++
 src/conductor/providers/factory.py      |   9 ++
 tests/test_config/test_schema.py        |  99 ++++++++++++++++++++
 tests/test_providers/test_copilot.py    | 114 ++++++++++++++++++++++++
 tests/test_providers/test_factory.py    |  20 +++++
 13 files changed, 558 insertions(+)
 create mode 100644 examples/context-tier.yaml
 create mode 100644 src/conductor/providers/context_tier.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3c98f0fb..ad544406 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased](https://github.com/microsoft/conductor/compare/v0.1.18...HEAD)
 
+### Added
+
+- **Context tier** — new `context_tier` knob (`default` | `long_context`) to
+  select a model's long-context (e.g. 1M-token) window on the Copilot provider.
+  Set per agent via `context_tier:` (sibling to `model`) or workflow-wide via
+  `runtime.default_context_tier`; the per-agent value wins. It composes
+  independently with `reasoning.effort` (the two map to separate
+  `create_session` kwargs). The Copilot provider forwards the resolved value as
+  `context_tier` to `create_session`; other providers ignore it. Only valid on
+  standard `agent`-type agents (rejected on `script`, `human_gate`, and
+  `workflow` agents). See [`examples/context-tier.yaml`](examples/context-tier.yaml)
+  and [Context Tier](docs/configuration.md#context-tier).
+  ([#251](https://github.com/microsoft/conductor/issues/251))
+
 ### Fixed
 
 - `human_gate` agents: the dict returned by `prompt_for` text-collection fields
diff --git a/docs/configuration.md b/docs/configuration.md
index 6a3919d2..67224341 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -16,6 +16,7 @@ workflow:
     temperature: 0.7
     max_tokens: 4096
     default_reasoning_effort: medium  # low | medium | high | xhigh (optional)
+    default_context_tier: default  # default | long_context (optional, Copilot only)
     # Provider-specific settings...
 ```
 
@@ -25,6 +26,11 @@ unless it declares its own `reasoning.effort` override. See
 [Reasoning Effort](#reasoning-effort) for the per-provider translation and
 constraints.
 
+The `default_context_tier` field sets a workflow-wide default for the model's
+context-window tier that every provider-backed agent inherits unless it
+declares its own `context_tier` override. See [Context Tier](#context-tier)
+for details. This is a Copilot-only capability.
+
 ## Provider Selection
 
 ### Copilot Provider
@@ -323,6 +329,52 @@ Reasoning / thinking content emitted by the model is surfaced via
 `agent_reasoning` events and rendered in the dashboard, JSONL logs, and
 `-vv` console output for both providers.
 
+## Context Tier
+
+Some models expose a larger context window (e.g. a 1M-token tier) selected via
+a separate session parameter rather than the model name. Conductor surfaces
+this as a unified `context_tier` knob. Allowed values: `default`,
+`long_context`.
+
+Use `long_context` for heavy-reasoning agents that ingest large evidence
+(multi-MB logs, many candidate source files) and would otherwise truncate at
+the default (~200K) tier.
+
+`context_tier` composes independently with `reasoning.effort` — they map to two
+separate `create_session` kwargs, so an agent may set both.
+
+Set a workflow-wide default and/or override per agent:
+
+```yaml
+workflow:
+  runtime:
+    provider: copilot
+    default_context_tier: default       # workflow-wide default
+
+agents:
+  - name: triage
+    # No context_tier — inherits `default` from the runtime default.
+    prompt: "Triage {{ workflow.input.topic }}"
+
+  - name: analyze
+    context_tier: long_context          # per-agent override wins
+    reasoning:
+      effort: high                      # composes with context_tier
+    prompt: "Deeply analyze {{ workflow.input.topic }}"
+```
+
+Per-agent overrides always win over the workflow-wide default. The
+`context_tier` field is **only** valid on standard `agent`-type agents; it is
+rejected on `script`, `human_gate`, and `workflow` agents (none of which call a
+model).
+
+### Per-provider translation
+
+- **Copilot** — Forwards the chosen tier as `context_tier` to
+  `CopilotClient.create_session`. No static capability validation is performed;
+  the SDK accepts or rejects the value at session creation.
+- **Other providers** — The value is ignored; there is no equivalent knob.
+
 ## MCP Servers
 
 Configure [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) servers for tool access. Both the Copilot and Claude providers support MCP tools.
diff --git a/docs/workflow-syntax.md b/docs/workflow-syntax.md
index a9bfaf26..bc29d4e7 100644
--- a/docs/workflow-syntax.md
+++ b/docs/workflow-syntax.md
@@ -55,6 +55,12 @@ workflow:
                                       # every provider-backed agent unless it
                                       # declares its own `reasoning.effort`.
                                       # See docs/configuration.md#reasoning-effort.
+    default_context_tier: default     # Optional: default | long_context (Copilot only)
+                                      # Workflow-wide default for the model's
+                                      # context-window tier. Inherited by every
+                                      # provider-backed agent unless it declares
+                                      # its own `context_tier`.
+                                      # See docs/configuration.md#context-tier.
 ```
 
 **Workflow metadata** is included verbatim in the `workflow_started` event and lets downstream consumers (dashboards, queue runners, observability tools) adapt without parsing the YAML. CLI `--metadata key=value` flags merge on top of YAML metadata (CLI wins on conflicts).
@@ -104,6 +110,14 @@ agents:
                                     # script, human_gate, workflow).
                                     # See docs/configuration.md#reasoning-effort.
 
+    context_tier: long_context      # Optional: per-agent context-tier override
+                                    # default | long_context (Copilot only)
+                                    # Overrides runtime.default_context_tier.
+                                    # Composes with reasoning. Only valid on
+                                    # type=agent (rejected on script,
+                                    # human_gate, workflow).
+                                    # See docs/configuration.md#context-tier.
+
     routes:                         # Optional: Routing logic
       - to: next_agent              # Agent name or $end
         when: "{{ condition }}"     # Optional: Route condition
diff --git a/examples/README.md b/examples/README.md
index 0c7a7ba9..166cf4b5 100644
--- a/examples/README.md
+++ b/examples/README.md
@@ -135,6 +135,23 @@ conductor run examples/reasoning-effort.yaml \
 
 See [Reasoning Effort](../docs/configuration.md#reasoning-effort) for the per-provider translation, supported models, and validation rules.
 
+## Context Tier
+
+### context-tier.yaml
+
+A two-stage workflow that demonstrates selecting a model's long-context (e.g. 1M-token) window. Demonstrates:
+- Workflow-wide default via `runtime.default_context_tier`
+- Per-agent override via `context_tier` (wins over the default)
+- Composition with `reasoning.effort` — the two map to separate `create_session` kwargs
+- Conditional routing on a structured boolean output
+
+```bash
+conductor run examples/context-tier.yaml \
+  --input topic="root-causing a multi-service latency regression"
+```
+
+See [Context Tier](../docs/configuration.md#context-tier) for details. This is a Copilot-only capability.
+
 ## Multi-Agent Workflows
 
 ### research-assistant.yaml
diff --git a/examples/context-tier.yaml b/examples/context-tier.yaml
new file mode 100644
index 00000000..9f56c5d3
--- /dev/null
+++ b/examples/context-tier.yaml
@@ -0,0 +1,105 @@
+# Context Tier Workflow
+#
+# This example demonstrates selecting a model's long-context (e.g. 1M-token)
+# window for heavy-reasoning agents. It shows:
+# - A workflow-wide default via `runtime.default_context_tier`
+# - A per-agent override via `context_tier` (wins over the default)
+# - Composition with `reasoning.effort` — the two are independent and
+#   map to two separate `create_session` kwargs
+#
+# Context tier is a Copilot-provider capability:
+# - Copilot SDK: passes `context_tier` ("default" | "long_context") to
+#   `create_session` to select the model's context-window tier.
+# - Other providers ignore the value.
+#
+# Use `long_context` for agents that ingest large evidence (multi-MB logs,
+# many candidate source files) and would otherwise truncate at the default
+# (~200K) tier.
+#
+# Usage:
+#   conductor run examples/context-tier.yaml \
+#     --input topic="root-causing a multi-service latency regression"
+
+workflow:
+  name: context-tier
+  description: Demonstrates workflow-wide and per-agent context-tier configuration
+  version: "1.0.0"
+  entry_point: triage
+
+  runtime:
+    provider: copilot
+    # Workflow-wide default applied to every provider-backed agent
+    # unless the agent declares its own `context_tier`.
+    default_context_tier: default
+
+  input:
+    topic:
+      type: string
+      required: true
+      description: An investigation topic that may require ingesting large evidence
+
+agents:
+  - name: triage
+    description: Lightweight triage that inherits the runtime default tier
+    model: claude-opus-4.8
+    # No `context_tier` here — inherits `runtime.default_context_tier: default`.
+    prompt: |
+      Briefly triage the following investigation. Decide whether a deep,
+      evidence-heavy analysis pass is warranted (true when it implies sifting
+      large logs or many source files).
+
+      Topic: {{ workflow.input.topic }}
+    output:
+      summary:
+        type: string
+        description: A short triage summary
+      needs_deep_analysis:
+        type: boolean
+        description: Whether a long-context analysis pass is warranted
+    routes:
+      - to: analyze
+        when: "{{ output.needs_deep_analysis }}"
+      - to: $end
+
+  - name: analyze
+    description: Deep analysis over large evidence (pinned to the long-context tier)
+    model: claude-opus-4.8
+    # Per-agent override: this agent always runs on the long-context tier,
+    # regardless of the workflow-wide default. Composes with reasoning.
+    context_tier: long_context
+    reasoning:
+      effort: high
+    input:
+      - workflow.input.topic
+      - triage.output.summary
+    prompt: |
+      Perform a thorough analysis of the investigation below. Assume you may
+      need to reason over a large body of evidence.
+
+      **Topic:** {{ workflow.input.topic }}
+
+      **Triage summary:**
+      {{ triage.output.summary }}
+
+      Produce a root-cause hypothesis, the evidence that supports it, and the
+      next concrete step to confirm it.
+    output:
+      hypothesis:
+        type: string
+        description: Root-cause hypothesis
+      evidence:
+        type: string
+        description: Evidence supporting the hypothesis
+      next_step:
+        type: string
+        description: The next concrete step to confirm the hypothesis
+    routes:
+      - to: $end
+
+output:
+  topic: "{{ workflow.input.topic }}"
+  summary: "{{ triage.output.summary }}"
+  needs_deep_analysis: "{{ triage.output.needs_deep_analysis }}"
+  hypothesis: "{% if analyze is defined %}{{ analyze.output.hypothesis }}{% else %}(skipped — no deep analysis needed){% endif %}"
+  evidence: "{% if analyze is defined %}{{ analyze.output.evidence }}{% endif %}"
+  next_step: "{% if analyze is defined %}{{ analyze.output.next_step }}{% endif %}"
diff --git a/src/conductor/config/schema.py b/src/conductor/config/schema.py
index a2ea6300..4beb6948 100644
--- a/src/conductor/config/schema.py
+++ b/src/conductor/config/schema.py
@@ -19,6 +19,7 @@
 )
 
 from conductor.duration import parse_duration
+from conductor.providers.context_tier import ContextTier
 from conductor.providers.reasoning import ReasoningEffort
 
 # Maximum allowed wait-step duration (24 hours). Anything longer almost
@@ -533,6 +534,26 @@ class AgentDef(BaseModel):
     Supports Jinja2 templates: {{ workflow.input.model_name }}
     """
 
+    context_tier: ContextTier | 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
+    model's long-context (e.g. 1M-token) window. ``default`` selects the
+    standard tier; ``None`` sends no value (provider default).
+
+    Falls back to ``runtime.default_context_tier`` when unset. Composes
+    independently with ``reasoning`` — an agent may set both.
+
+    Only the Copilot provider forwards this today (maps to the SDK's
+    ``create_session`` ``context_tier`` param). Other providers ignore it.
+
+    Only applies to provider-backed agents (type='agent' or None).
+
+    Example YAML::
+
+        context_tier: long_context
+    """
+
     input: list[str] = Field(default_factory=list)
     """Context dependencies. Format: 'agent_name.output' or 'workflow.input.param'.
     Suffix with '?' for optional dependencies."""
@@ -892,6 +913,8 @@ def validate_agent_type(self) -> AgentDef:
                 raise ValueError("human_gate agents cannot have 'max_depth'")
             if self.reasoning is not None:
                 raise ValueError("human_gate agents cannot have 'reasoning'")
+            if self.context_tier is not None:
+                raise ValueError("human_gate agents cannot have 'context_tier'")
             if self.timeout_seconds is not None:
                 raise ValueError("human_gate agents cannot have 'timeout_seconds'")
             if self.value is not None:
@@ -931,6 +954,8 @@ def validate_agent_type(self) -> AgentDef:
                 raise ValueError("script agents cannot have 'max_depth'")
             if self.reasoning is not None:
                 raise ValueError("script agents cannot have 'reasoning'")
+            if self.context_tier is not None:
+                raise ValueError("script agents cannot have 'context_tier'")
             if self.timeout_seconds is not None:
                 raise ValueError(
                     "script agents cannot have 'timeout_seconds' "
@@ -1183,6 +1208,8 @@ def validate_agent_type(self) -> AgentDef:
                 )
         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:
+            raise ValueError("workflow agents cannot have 'context_tier'")
 
         # Wait-only fields are forbidden on every other type. ``reason`` is
         # shared with ``type: terminate`` (which has its own required-non-
@@ -1565,6 +1592,17 @@ def _coerce_provider(cls, value: Any) -> Any:
     the request through to the SDK.
     """
 
+    default_context_tier: ContextTier | None = None
+    """Workflow-wide default context-window tier (Copilot provider only).
+
+    Each agent may override with its own ``context_tier``. ``long_context``
+    selects a model's long-context (e.g. 1M-token) window; ``default`` selects
+    the standard tier; ``None`` sends no value.
+
+    Only the Copilot provider forwards this (maps to the SDK's
+    ``create_session`` ``context_tier`` param). Other providers ignore it.
+    """
+
 
 class WorkflowDef(BaseModel):
     """Top-level workflow configuration."""
diff --git a/src/conductor/engine/workflow.py b/src/conductor/engine/workflow.py
index bd7b85b7..fe6aaf5a 100644
--- a/src/conductor/engine/workflow.py
+++ b/src/conductor/engine/workflow.py
@@ -584,6 +584,7 @@ def build_workflow_started_data(self) -> dict[str, Any]:
             self._system_metadata = self._build_system_metadata()
 
         default_effort = self.config.workflow.runtime.default_reasoning_effort
+        default_tier = self.config.workflow.runtime.default_context_tier
         default_provider_name = self.config.workflow.runtime.provider.name
 
         # Resolve the provider per agent (honoring per-agent overrides).
@@ -674,6 +675,9 @@ def _record_provider(name: str) -> None:
                     "reasoning_effort": (
                         a.reasoning.effort if a.reasoning is not None else default_effort
                     ),
+                    "context_tier": (
+                        a.context_tier if a.context_tier is not None else default_tier
+                    ),
                 }
                 for a in self.config.agents
             ],
diff --git a/src/conductor/providers/context_tier.py b/src/conductor/providers/context_tier.py
new file mode 100644
index 00000000..8cdae244
--- /dev/null
+++ b/src/conductor/providers/context_tier.py
@@ -0,0 +1,43 @@
+"""Shared context-tier helpers for providers.
+
+This module centralizes Conductor's provider-agnostic notion of a model's
+context-window tier and how it resolves between the workflow-wide default
+and a per-agent override.
+
+- The **Copilot SDK** accepts a ``context_tier`` literal on ``create_session``
+  (``ContextTier = Literal["default", "long_context"]``) to select a model's
+  long-context (e.g. 1M-token) window.
+- Other providers do not currently expose an equivalent knob, so they ignore
+  the resolved value.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Literal
+
+if TYPE_CHECKING:
+    from conductor.config.schema import AgentDef
+
+ContextTier = Literal["default", "long_context"]
+
+
+def resolve_context_tier(
+    agent: AgentDef,
+    runtime_default: ContextTier | None,
+) -> ContextTier | None:
+    """Resolve the effective context tier for an agent.
+
+    Per-agent ``context_tier`` wins over the workflow-wide
+    ``runtime.default_context_tier``. Returns ``None`` when neither is set,
+    signalling that no context-tier parameter should be sent to the SDK.
+
+    Args:
+        agent: The agent whose effective context tier is being resolved.
+        runtime_default: Workflow-wide default, or ``None``.
+
+    Returns:
+        The resolved ``ContextTier``, or ``None`` to send no value.
+    """
+    if agent.context_tier is not None:
+        return agent.context_tier
+    return runtime_default
diff --git a/src/conductor/providers/copilot.py b/src/conductor/providers/copilot.py
index a51825ff..f1cf5c66 100644
--- a/src/conductor/providers/copilot.py
+++ b/src/conductor/providers/copilot.py
@@ -24,6 +24,7 @@
 )
 from conductor.providers.base import AgentOutput, AgentProvider, EventCallback, match_model_id
 from conductor.providers.capabilities import ProviderCapabilities
+from conductor.providers.context_tier import ContextTier, resolve_context_tier
 from conductor.providers.reasoning import ReasoningEffort, resolve_reasoning_effort
 
 if TYPE_CHECKING:
@@ -204,6 +205,7 @@ def __init__(
         temperature: float | None = None,
         max_agent_iterations: int | None = None,
         default_reasoning_effort: ReasoningEffort | None = None,
+        default_context_tier: ContextTier | None = None,
         provider_settings: ProviderSettings | None = None,
     ) -> None:
         """Initialize the Copilot provider.
@@ -226,6 +228,10 @@ def __init__(
                 applied to ``create_session`` when an agent does not specify
                 its own ``reasoning.effort``. One of ``low``, ``medium``,
                 ``high``, ``xhigh``, or ``None`` to send no value.
+            default_context_tier: Workflow-wide default ``context_tier`` applied
+                to ``create_session`` when an agent does not specify its own
+                ``context_tier``. One of ``default``, ``long_context``, or
+                ``None`` to send no value.
             provider_settings: Optional structured provider settings from
                 ``runtime.provider``. When ``has_custom_routing()`` is True,
                 the resolved SDK ``ProviderConfig`` is attached to every
@@ -255,6 +261,7 @@ def __init__(
         self._temperature = temperature
         self._default_max_agent_iterations = max_agent_iterations
         self._default_reasoning_effort = default_reasoning_effort
+        self._default_context_tier = default_context_tier
         self._max_schema_depth = 10  # Max nesting depth for recursive schema building
         self._session_ids: dict[str, str] = {}
         self._resume_session_ids: dict[str, str] = {}
@@ -781,6 +788,17 @@ async def _execute_sdk_call(
                     model,
                 )
 
+            # Resolve context tier: per-agent override wins over runtime default.
+            tier = resolve_context_tier(agent, self._default_context_tier)
+            if tier is not None:
+                session_kwargs["context_tier"] = tier
+                logger.debug(
+                    "Setting context_tier=%s for agent %r (model=%s)",
+                    tier,
+                    agent.name,
+                    model,
+                )
+
             # Attempt to resume a previous session if one exists for this agent
             session: Any = None
             resume_sid = self._resume_session_ids.get(agent.name)
@@ -2152,6 +2170,17 @@ async def execute_dialog_turn(
                     dialog_kwargs["model"],
                 )
 
+            # Dialog turns likewise honor only the workflow-wide default
+            # context tier (no agent-scoped override at this layer).
+            tier = self._default_context_tier
+            if tier is not None:
+                dialog_kwargs["context_tier"] = tier
+                logger.debug(
+                    "Setting context_tier=%s for dialog turn (model=%s)",
+                    tier,
+                    dialog_kwargs["model"],
+                )
+
             session = await self._client.create_session(**dialog_kwargs)
 
             response_content = ""
diff --git a/src/conductor/providers/factory.py b/src/conductor/providers/factory.py
index bd913050..b36c34c3 100644
--- a/src/conductor/providers/factory.py
+++ b/src/conductor/providers/factory.py
@@ -15,6 +15,7 @@
     CLAUDE_AGENT_SDK_AVAILABLE,
     ClaudeAgentSdkProvider,
 )
+from conductor.providers.context_tier import ContextTier
 from conductor.providers.copilot import CopilotProvider, IdleRecoveryConfig
 from conductor.providers.reasoning import ReasoningEffort
 
@@ -33,6 +34,7 @@ async def create_provider(
     max_session_seconds: float | None = None,
     max_agent_iterations: int | None = None,
     default_reasoning_effort: ReasoningEffort | None = None,
+    default_context_tier: ContextTier | None = None,
     provider_settings: ProviderSettings | None = None,
 ) -> AgentProvider:
     """Factory function to create the appropriate provider.
@@ -57,6 +59,10 @@ async def create_provider(
         default_reasoning_effort: Workflow-wide default reasoning effort
             (``low`` / ``medium`` / ``high`` / ``xhigh``) applied when an agent
             does not specify its own ``reasoning.effort``.
+        default_context_tier: Workflow-wide default context-window tier
+            (``default`` / ``long_context``) applied when an agent does not
+            specify its own ``context_tier``. Only the Copilot provider
+            forwards this; ignored for all other providers.
         provider_settings: Structured ``runtime.provider`` settings. Only
             applied when ``provider_type == "copilot"`` and the settings
             opted into custom routing; ignored for all other providers
@@ -87,6 +93,7 @@ async def create_provider(
                 idle_recovery_config=idle_recovery_config,
                 max_agent_iterations=max_agent_iterations,
                 default_reasoning_effort=default_reasoning_effort,
+                default_context_tier=default_context_tier,
                 provider_settings=provider_settings,
             )
         case "openai-agents":
@@ -214,6 +221,7 @@ async def create_provider(
         max_session_seconds = getattr(runtime_config, "max_session_seconds", None)
         max_agent_iterations = getattr(runtime_config, "max_agent_iterations", None)
         default_reasoning_effort = getattr(runtime_config, "default_reasoning_effort", None)
+        default_context_tier = getattr(runtime_config, "default_context_tier", None)
 
         return await create_provider(
             provider_type=provider_type,
@@ -225,5 +233,6 @@ async def create_provider(
             max_session_seconds=max_session_seconds,
             max_agent_iterations=max_agent_iterations,
             default_reasoning_effort=default_reasoning_effort,
+            default_context_tier=default_context_tier,
             provider_settings=provider_settings,
         )
diff --git a/tests/test_config/test_schema.py b/tests/test_config/test_schema.py
index e16be32a..d0129d1e 100644
--- a/tests/test_config/test_schema.py
+++ b/tests/test_config/test_schema.py
@@ -1399,6 +1399,105 @@ def test_rejects_invalid_effort(self, effort: object) -> None:
             RuntimeConfig(default_reasoning_effort=effort)  # type: ignore[arg-type]
 
 
+class TestAgentDefContextTier:
+    """Tests for the context_tier field on AgentDef."""
+
+    @pytest.mark.parametrize("tier", ["default", "long_context"])
+    def test_accepts_valid_tier(self, tier: str) -> None:
+        """Test that AgentDef accepts each valid context tier."""
+        agent = AgentDef(name="a", model="gpt-4", prompt="test", context_tier=tier)  # type: ignore[arg-type]
+        assert agent.context_tier == tier
+
+    @pytest.mark.parametrize("tier", ["1m", "huge", 42, ""])
+    def test_rejects_invalid_tier(self, tier: object) -> None:
+        """Test that invalid context_tier values raise ValidationError."""
+        with pytest.raises(ValidationError):
+            AgentDef(
+                name="a",
+                model="gpt-4",
+                prompt="test",
+                context_tier=tier,  # type: ignore[arg-type]
+            )
+
+    def test_context_tier_defaults_to_none(self) -> None:
+        """Test that context_tier defaults to None when omitted."""
+        agent = AgentDef(name="x", model="gpt-4", prompt="test")
+        assert agent.context_tier is None
+
+    def test_context_tier_composes_with_reasoning(self) -> None:
+        """Test that context_tier and reasoning can be set together."""
+        agent = AgentDef(
+            name="a",
+            model="claude-opus-4.8",
+            prompt="test",
+            context_tier="long_context",
+            reasoning={"effort": "high"},
+        )
+        assert agent.context_tier == "long_context"
+        assert agent.reasoning is not None
+        assert agent.reasoning.effort == "high"
+
+    def test_human_gate_with_context_tier_raises(self) -> None:
+        """Test that human_gate agents cannot have context_tier."""
+        with pytest.raises(ValidationError) as exc_info:
+            AgentDef(
+                name="g",
+                type="human_gate",
+                prompt="Approve?",
+                options=[GateOption(label="Ok", value="ok", route="next")],
+                context_tier="long_context",
+            )
+        assert "human_gate agents cannot have 'context_tier'" in str(exc_info.value)
+
+    def test_script_with_context_tier_raises(self) -> None:
+        """Test that script agents cannot have context_tier."""
+        with pytest.raises(ValidationError) as exc_info:
+            AgentDef(
+                name="s",
+                type="script",
+                command="echo hi",
+                context_tier="long_context",
+            )
+        assert "script agents cannot have 'context_tier'" in str(exc_info.value)
+
+    def test_workflow_with_context_tier_raises(self) -> None:
+        """Test that workflow agents cannot have context_tier."""
+        with pytest.raises(ValidationError) as exc_info:
+            AgentDef(
+                name="w",
+                type="workflow",
+                workflow="./sub.yaml",
+                context_tier="long_context",
+            )
+        assert "workflow agents cannot have 'context_tier'" in str(exc_info.value)
+
+
+class TestRuntimeConfigDefaultContextTier:
+    """Tests for default_context_tier on RuntimeConfig."""
+
+    def test_default_is_none(self) -> None:
+        """Test that default_context_tier defaults to None."""
+        config = RuntimeConfig()
+        assert config.default_context_tier is None
+
+    def test_explicit_none_is_valid(self) -> None:
+        """Test that explicitly passing None is valid."""
+        config = RuntimeConfig(default_context_tier=None)
+        assert config.default_context_tier is None
+
+    @pytest.mark.parametrize("tier", ["default", "long_context"])
+    def test_accepts_valid_tier(self, tier: str) -> None:
+        """Test that each valid context tier is accepted."""
+        config = RuntimeConfig(default_context_tier=tier)  # type: ignore[arg-type]
+        assert config.default_context_tier == tier
+
+    @pytest.mark.parametrize("tier", ["1m", "huge", 42, ""])
+    def test_rejects_invalid_tier(self, tier: object) -> None:
+        """Test that invalid context tier values raise ValidationError."""
+        with pytest.raises(ValidationError):
+            RuntimeConfig(default_context_tier=tier)  # type: ignore[arg-type]
+
+
 class TestExtraFieldsForbidden:
     """Tests that workflow models reject unknown fields.
 
diff --git a/tests/test_providers/test_copilot.py b/tests/test_providers/test_copilot.py
index f0db9ebc..a45e9540 100644
--- a/tests/test_providers/test_copilot.py
+++ b/tests/test_providers/test_copilot.py
@@ -1606,3 +1606,117 @@ async def fake_sleep(delay: float) -> None:
         assert len(provider.get_retry_history()) == 3
         # Two backoff sleeps between three attempts.
         assert len(sleep_calls) == 2
+
+
+class TestContextTier:
+    """Tests for context_tier plumbing into create_session."""
+
+    @staticmethod
+    async def _build_provider(
+        captured: dict[str, Any],
+        monkeypatch: pytest.MonkeyPatch,
+        *,
+        default_context_tier: str | None = None,
+    ) -> CopilotProvider:
+        """Build a real-SDK-mode provider that captures create_session kwargs."""
+
+        class _FakeSession:
+            session_id = "session-xyz"
+
+            async def disconnect(self) -> None:
+                return None
+
+        class _FakeClient:
+            async def create_session(self, **kwargs: Any) -> _FakeSession:
+                captured["create_session_kwargs"] = kwargs
+                return _FakeSession()
+
+            async def list_models(self) -> list[Any]:
+                return []
+
+        provider = CopilotProvider(
+            mock_handler=stub_handler,
+            default_context_tier=default_context_tier,  # type: ignore[arg-type]
+        )
+        provider._mock_handler = None
+        provider._client = _FakeClient()
+        provider._started = True
+
+        async def _noop() -> None:
+            return None
+
+        async def _fake_send_and_wait(*args: Any, **kwargs: Any) -> SDKResponse:
+            return SDKResponse(content='{"ok":true}')
+
+        monkeypatch.setattr(provider, "_ensure_client_started", _noop)
+        monkeypatch.setattr(provider, "_send_and_wait", _fake_send_and_wait)
+        return provider
+
+    @pytest.mark.asyncio
+    async def test_per_agent_tier_forwarded_to_create_session(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        captured: dict[str, Any] = {}
+        provider = await self._build_provider(captured, monkeypatch)
+        agent = AgentDef(
+            name="analyze",
+            model="claude-opus-4.8",
+            prompt="Analyze",
+            context_tier="long_context",
+        )
+        await provider.execute(agent=agent, context={}, rendered_prompt="Analyze")
+        assert captured["create_session_kwargs"]["context_tier"] == "long_context"
+
+    @pytest.mark.asyncio
+    async def test_runtime_default_used_when_agent_has_none(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        captured: dict[str, Any] = {}
+        provider = await self._build_provider(
+            captured, monkeypatch, default_context_tier="long_context"
+        )
+        agent = AgentDef(name="analyze", model="claude-opus-4.8", prompt="Analyze")
+        await provider.execute(agent=agent, context={}, rendered_prompt="Analyze")
+        assert captured["create_session_kwargs"]["context_tier"] == "long_context"
+
+    @pytest.mark.asyncio
+    async def test_per_agent_tier_overrides_runtime_default(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        captured: dict[str, Any] = {}
+        provider = await self._build_provider(
+            captured, monkeypatch, default_context_tier="long_context"
+        )
+        agent = AgentDef(
+            name="cheap",
+            model="claude-opus-4.8",
+            prompt="Cheap",
+            context_tier="default",
+        )
+        await provider.execute(agent=agent, context={}, rendered_prompt="Cheap")
+        assert captured["create_session_kwargs"]["context_tier"] == "default"
+
+    @pytest.mark.asyncio
+    async def test_no_tier_set_means_key_absent(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        captured: dict[str, Any] = {}
+        provider = await self._build_provider(captured, monkeypatch)
+        agent = AgentDef(name="analyze", model="claude-opus-4.8", prompt="Analyze")
+        await provider.execute(agent=agent, context={}, rendered_prompt="Analyze")
+        assert "context_tier" not in captured["create_session_kwargs"]
+
+    @pytest.mark.asyncio
+    async def test_tier_and_reasoning_compose(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        from conductor.config.schema import ReasoningConfig
+
+        captured: dict[str, Any] = {}
+        provider = await self._build_provider(captured, monkeypatch)
+        agent = AgentDef(
+            name="analyze",
+            model="claude-opus-4.8",
+            prompt="Analyze",
+            context_tier="long_context",
+            reasoning=ReasoningConfig(effort="high"),
+        )
+        await provider.execute(agent=agent, context={}, rendered_prompt="Analyze")
+        assert captured["create_session_kwargs"]["context_tier"] == "long_context"
+        assert captured["create_session_kwargs"]["reasoning_effort"] == "high"
diff --git a/tests/test_providers/test_factory.py b/tests/test_providers/test_factory.py
index edab3c83..c078d676 100644
--- a/tests/test_providers/test_factory.py
+++ b/tests/test_providers/test_factory.py
@@ -45,6 +45,26 @@ async def test_create_openai_provider_raises(self) -> None:
         assert exc_info.value.suggestion is not None
         assert "copilot" in exc_info.value.suggestion
 
+    @pytest.mark.asyncio
+    async def test_copilot_provider_receives_default_context_tier(self) -> None:
+        """default_context_tier is threaded into the Copilot provider."""
+        provider = await create_provider(
+            "copilot",
+            validate=False,
+            default_context_tier="long_context",
+        )
+        assert isinstance(provider, CopilotProvider)
+        assert provider._default_context_tier == "long_context"
+        await provider.close()
+
+    @pytest.mark.asyncio
+    async def test_copilot_provider_default_context_tier_none(self) -> None:
+        """default_context_tier defaults to None on the Copilot provider."""
+        provider = await create_provider("copilot", validate=False)
+        assert isinstance(provider, CopilotProvider)
+        assert provider._default_context_tier is None
+        await provider.close()
+
     @patch("conductor.providers.factory.ANTHROPIC_SDK_AVAILABLE", False)
     @pytest.mark.asyncio
     async def test_create_claude_provider_raises_when_sdk_not_available(self) -> None:

From a99c8a53212a5d912558b336f7e16543b895a8b5 Mon Sep 17 00:00:00 2001
From: Brent Rusinow <brrusino@microsoft.com>
Date: Tue, 16 Jun 2026 09:46:01 -0700
Subject: [PATCH 2/3] test(copilot): cover context_tier dialog-turn forwarding

Codecov flagged 2 uncovered lines on PR #252: the dialog-turn path
that forwards default_context_tier into create_session was never
exercised. Add two dialog-turn tests (default set and absent) so the
patch coverage gap is closed.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 tests/test_providers/test_copilot.py | 27 +++++++++++++++++++++++++++
 1 file changed, 27 insertions(+)

diff --git a/tests/test_providers/test_copilot.py b/tests/test_providers/test_copilot.py
index a45e9540..6cd1daca 100644
--- a/tests/test_providers/test_copilot.py
+++ b/tests/test_providers/test_copilot.py
@@ -1139,6 +1139,33 @@ async def test_dialog_turn_model_override_used(self) -> None:
 
         assert captured["create_session_kwargs"]["model"] == "claude-sonnet-4.5"
 
+    @pytest.mark.asyncio
+    async def test_dialog_turn_default_context_tier_forwarded(self) -> None:
+        captured: dict[str, Any] = {}
+        provider = await self._make_provider_with_session(captured)
+        provider._default_context_tier = "long_context"
+
+        await provider.execute_dialog_turn(
+            system_prompt="sys",
+            user_message="hi",
+            history=None,
+        )
+
+        assert captured["create_session_kwargs"]["context_tier"] == "long_context"
+
+    @pytest.mark.asyncio
+    async def test_dialog_turn_no_context_tier_means_key_absent(self) -> None:
+        captured: dict[str, Any] = {}
+        provider = await self._make_provider_with_session(captured)
+
+        await provider.execute_dialog_turn(
+            system_prompt="sys",
+            user_message="hi",
+            history=None,
+        )
+
+        assert "context_tier" not in captured["create_session_kwargs"]
+
     @pytest.mark.asyncio
     async def test_dialog_turn_session_error_wrapped_as_provider_error(self) -> None:
         from unittest.mock import AsyncMock as _AsyncMock

From 2c5e2bb8c94549048c5e684db8e7f9e4e9629ade Mon Sep 17 00:00:00 2001
From: Brent Rusinow <brrusino@microsoft.com>
Date: Tue, 16 Jun 2026 12:32:57 -0700
Subject: [PATCH 3/3] fix(context): reject context_tier on wait/set/terminate
 agents

Address review from Jason Robert on PR #252. context_tier was only
guarded on human_gate/script/workflow agents, so it was silently
accepted on wait/set/terminate - dropping the parity with reasoning,
which is rejected on all six non-model types. The value would then
surface in workflow_started metadata as if active.

- Add the three missing schema guards mirroring the reasoning checks.
- Add wait/set/terminate rejection tests.
- Add workflow_started event tests for per-agent context_tier
  (override, runtime-default, None-when-unset), cloning the existing
  reasoning_effort coverage.
- Note the validation asymmetry vs reasoning effort with a comment in
  the copilot provider (no advertised supported_context_tiers, so the
  SDK is the sole authority).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 src/conductor/config/schema.py           |  6 ++
 src/conductor/providers/copilot.py       |  4 ++
 tests/test_config/test_schema.py         | 24 ++++++++
 tests/test_engine/test_event_emission.py | 77 ++++++++++++++++++++++++
 4 files changed, 111 insertions(+)

diff --git a/src/conductor/config/schema.py b/src/conductor/config/schema.py
index 4beb6948..24cd1700 100644
--- a/src/conductor/config/schema.py
+++ b/src/conductor/config/schema.py
@@ -1041,6 +1041,8 @@ def validate_agent_type(self) -> AgentDef:
                 raise ValueError("wait agents cannot have 'dialog'")
             if self.reasoning is not None:
                 raise ValueError("wait agents cannot have 'reasoning'")
+            if self.context_tier is not None:
+                raise ValueError("wait agents cannot have 'context_tier'")
             if self.timeout_seconds is not None:
                 raise ValueError("wait agents cannot have 'timeout_seconds'")
             if self.output is not None:
@@ -1100,6 +1102,8 @@ def validate_agent_type(self) -> AgentDef:
                 raise ValueError("set agents cannot have 'dialog'")
             if self.reasoning is not None:
                 raise ValueError("set agents cannot have 'reasoning'")
+            if self.context_tier is not None:
+                raise ValueError("set agents cannot have 'context_tier'")
             if self.timeout_seconds is not None:
                 raise ValueError("set agents cannot have 'timeout_seconds'")
             if self.duration is not None:
@@ -1158,6 +1162,8 @@ def validate_agent_type(self) -> AgentDef:
                 raise ValueError("terminate agents cannot have 'dialog'")
             if self.reasoning is not None:
                 raise ValueError("terminate agents cannot have 'reasoning'")
+            if self.context_tier is not None:
+                raise ValueError("terminate agents cannot have 'context_tier'")
             if self.workflow:
                 raise ValueError("terminate agents cannot have 'workflow'")
             if self.input_mapping is not None:
diff --git a/src/conductor/providers/copilot.py b/src/conductor/providers/copilot.py
index f1cf5c66..0edb1416 100644
--- a/src/conductor/providers/copilot.py
+++ b/src/conductor/providers/copilot.py
@@ -789,6 +789,10 @@ async def _execute_sdk_call(
                 )
 
             # Resolve context tier: per-agent override wins over runtime default.
+            # Unlike reasoning effort (validated against the model's advertised
+            # supported_reasoning_efforts first), the tier is forwarded as-is:
+            # there is no advertised supported_context_tiers, so the SDK is the
+            # sole authority and validates it at session creation.
             tier = resolve_context_tier(agent, self._default_context_tier)
             if tier is not None:
                 session_kwargs["context_tier"] = tier
diff --git a/tests/test_config/test_schema.py b/tests/test_config/test_schema.py
index d0129d1e..c15cba9c 100644
--- a/tests/test_config/test_schema.py
+++ b/tests/test_config/test_schema.py
@@ -1471,6 +1471,30 @@ def test_workflow_with_context_tier_raises(self) -> None:
             )
         assert "workflow agents cannot have 'context_tier'" in str(exc_info.value)
 
+    def test_wait_with_context_tier_raises(self) -> None:
+        """Test that wait agents cannot have context_tier."""
+        with pytest.raises(ValidationError) as exc_info:
+            AgentDef(name="w", type="wait", duration="1s", context_tier="long_context")
+        assert "wait agents cannot have 'context_tier'" in str(exc_info.value)
+
+    def test_set_with_context_tier_raises(self) -> None:
+        """Test that set agents cannot have context_tier."""
+        with pytest.raises(ValidationError) as exc_info:
+            AgentDef(name="s", type="set", value="42", context_tier="long_context")
+        assert "set agents cannot have 'context_tier'" in str(exc_info.value)
+
+    def test_terminate_with_context_tier_raises(self) -> None:
+        """Test that terminate agents cannot have context_tier."""
+        with pytest.raises(ValidationError) as exc_info:
+            AgentDef(
+                name="t",
+                type="terminate",
+                status="success",
+                reason="done",
+                context_tier="long_context",
+            )
+        assert "terminate agents cannot have 'context_tier'" in str(exc_info.value)
+
 
 class TestRuntimeConfigDefaultContextTier:
     """Tests for default_context_tier on RuntimeConfig."""
diff --git a/tests/test_engine/test_event_emission.py b/tests/test_engine/test_event_emission.py
index de1a4d02..7c9927a8 100644
--- a/tests/test_engine/test_event_emission.py
+++ b/tests/test_engine/test_event_emission.py
@@ -279,6 +279,83 @@ async def test_workflow_started_reasoning_effort_unset_is_none(self) -> None:
         event = collector.first("workflow_started")
         assert event.data["agents"][0]["reasoning_effort"] is None
 
+    @pytest.mark.asyncio
+    async def test_workflow_started_includes_context_tier(self) -> None:
+        """workflow_started includes per-agent context_tier.
+
+        - Agent with explicit context_tier wins.
+        - Agent without explicit context_tier falls back to
+          runtime.default_context_tier.
+        """
+        emitter, collector = _make_emitter_and_collector()
+        config = WorkflowConfig(
+            workflow=WorkflowDef(
+                name="context-tier-test",
+                entry_point="explicit",
+                runtime=RuntimeConfig(provider="copilot", default_context_tier="default"),
+                context=ContextConfig(mode="accumulate"),
+                limits=LimitsConfig(max_iterations=10),
+            ),
+            agents=[
+                AgentDef(
+                    name="explicit",
+                    model="gpt-4",
+                    prompt="explicit override",
+                    context_tier="long_context",
+                    output={"x": OutputField(type="string")},
+                    routes=[RouteDef(to="default")],
+                ),
+                AgentDef(
+                    name="default",
+                    model="gpt-4",
+                    prompt="uses workflow default",
+                    output={"y": OutputField(type="string")},
+                    routes=[RouteDef(to="$end")],
+                ),
+            ],
+            output={"y": "{{ default.output.y }}"},
+        )
+        provider = CopilotProvider(
+            mock_handler=lambda a, p, c: {"x": "1"} if a.name == "explicit" else {"y": "2"}
+        )
+        engine = WorkflowEngine(config, provider, event_emitter=emitter)
+        await engine.run({})
+
+        event = collector.first("workflow_started")
+        agents = {a["name"]: a for a in event.data["agents"]}
+        assert agents["explicit"]["context_tier"] == "long_context"
+        assert agents["default"]["context_tier"] == "default"
+
+    @pytest.mark.asyncio
+    async def test_workflow_started_context_tier_unset_is_none(self) -> None:
+        """When neither agent nor workflow default sets context_tier, field is None."""
+        emitter, collector = _make_emitter_and_collector()
+        config = WorkflowConfig(
+            workflow=WorkflowDef(
+                name="no-context-tier",
+                entry_point="agent1",
+                runtime=RuntimeConfig(provider="copilot"),
+                context=ContextConfig(mode="accumulate"),
+                limits=LimitsConfig(max_iterations=10),
+            ),
+            agents=[
+                AgentDef(
+                    name="agent1",
+                    model="gpt-4",
+                    prompt="no context tier anywhere",
+                    output={"result": OutputField(type="string")},
+                    routes=[RouteDef(to="$end")],
+                ),
+            ],
+            output={"result": "{{ agent1.output.result }}"},
+        )
+        provider = CopilotProvider(mock_handler=lambda a, p, c: {"result": "done"})
+        engine = WorkflowEngine(config, provider, event_emitter=emitter)
+        await engine.run({})
+
+        event = collector.first("workflow_started")
+        assert event.data["agents"][0]["context_tier"] is None
+
 
 class TestAgentEvents:
     """Tests for agent_started, agent_completed, and agent_failed events."""