diff --git a/docs/source_en/Usage Guide/Agentic-Evaluator.md b/docs/source_en/Usage Guide/Agentic-Evaluator.md
new file mode 100644
index 000000000..4d8c36f98
--- /dev/null
+++ b/docs/source_en/Usage Guide/Agentic-Evaluator.md	
@@ -0,0 +1,44 @@
+# Agentic Evaluator
+
+Install evaluation support separately:
+
+```bash
+pip install 'twinkle-kit[eval]'
+```
+
+`twinkle_agentic.evaluator.Evaluator` is a single-use facade for EvalScope's native runner. EvalScope owns datasets, agent loops, tools, judges, caches, metrics, and reports; Twinkle only adapts the candidate model boundary.
+
+## Protocol API
+
+```python
+from twinkle_agentic.evaluator import Evaluator
+from twinkle_agentic.protocol.openai import OpenAI
+
+reports = Evaluator(
+    api=OpenAI(model='qwen-plus', api_key='...', base_url='https://example.com/v1'),
+    datasets=['gsm8k', 'bfcl_v4'],
+    task_config={'generation_config': {'temperature': 0.0}},
+).run()
+```
+
+## Sampler
+
+```python
+from twinkle_agentic.evaluator import Evaluator
+
+reports = Evaluator(
+    sampler=sampler,
+    datasets=['gsm8k'],
+    template=template,
+    sampler_kwargs={'adapter_uri': 'twinkle://my-adapter'},
+    task_config={'limit': 100, 'generation_config': {'max_tokens': 2048}},
+).run()
+```
+
+The sampler path micro-batches compatible EvalScope requests. `template.parse_tool_call()` is required for tool benchmarks unless the sampler returns structured assistant tool calls. The HTTP sampler accepts either Twinkle `SamplingParams` or a mapping.
+
+## Capability boundary
+
+Text/chat, function-calling, and EvalScope native AgentLoop tasks are supported. Image, audio, video, streaming, non-native EvalScope backends, and nested Twinkle rollout loops are not. Explicit generation options are mapped exactly or rejected before evaluation begins; Twinkle does not approximate unsupported parameters.
+
+`run()` returns EvalScope's `dict[str, Report]` unchanged. `resolved_task_config` exposes the constructed EvalScope configuration, and `output_dir` becomes available after EvalScope resolves the run directory. See EvalScope `TaskConfig` for pass-through fields and install its benchmark-specific extras separately when needed.
diff --git a/docs/source_en/index.rst b/docs/source_en/index.rst
index e1c4ba4b6..0a00efdd1 100644
--- a/docs/source_en/index.rst
+++ b/docs/source_en/index.rst
@@ -15,6 +15,7 @@ Twinkle DOCUMENTATION
    Usage Guide/NPU-Support.md
    Usage Guide/Train-as-a-Service.md
    Usage Guide/Agentic-RL-Deployment-and-Training.md
+   Usage Guide/Agentic-Evaluator.md
    Usage Guide/Introduction-with-Qwen3.5.md
    Usage Guide/Embedding-Training.md
 
diff --git a/docs/source_zh/index.rst b/docs/source_zh/index.rst
index 3a3151549..96ac5a05a 100644
--- a/docs/source_zh/index.rst
+++ b/docs/source_zh/index.rst
@@ -15,6 +15,7 @@ Twinkle DOCUMENTATION
    使用指引/NPU的支持.md
    使用指引/训练服务.md
    使用指引/Agentic RL部署与训练.md
+   使用指引/Agentic评测.md
    使用指引/Qwen3.5最佳实践.md
    使用指引/Embedding训练.md
 
diff --git "a/docs/source_zh/\344\275\277\347\224\250\346\214\207\345\274\225/Agentic\350\257\204\346\265\213.md" "b/docs/source_zh/\344\275\277\347\224\250\346\214\207\345\274\225/Agentic\350\257\204\346\265\213.md"
new file mode 100644
index 000000000..b43d414c6
--- /dev/null
+++ "b/docs/source_zh/\344\275\277\347\224\250\346\214\207\345\274\225/Agentic\350\257\204\346\265\213.md"
@@ -0,0 +1,42 @@
+# Agentic 评测
+
+评测能力是可选依赖：
+
+```bash
+pip install 'twinkle-kit[eval]'
+```
+
+`twinkle_agentic.evaluator.Evaluator` 是 EvalScope Native runner 的一次性轻量封装。数据集、AgentLoop、工具、Judge、缓存、指标和报告仍由 EvalScope 负责，Twinkle 只适配被评测模型。
+
+## Protocol API
+
+```python
+from twinkle_agentic.evaluator import Evaluator
+from twinkle_agentic.protocol.openai import OpenAI
+
+reports = Evaluator(
+    api=OpenAI(model='qwen-plus', api_key='...', base_url='https://example.com/v1'),
+    datasets=['gsm8k', 'bfcl_v4'],
+    task_config={'generation_config': {'temperature': 0.0}},
+).run()
+```
+
+## Sampler
+
+```python
+reports = Evaluator(
+    sampler=sampler,
+    datasets=['gsm8k'],
+    template=template,
+    sampler_kwargs={'adapter_uri': 'twinkle://my-adapter'},
+    task_config={'limit': 100, 'generation_config': {'max_tokens': 2048}},
+).run()
+```
+
+Sampler 路径会将兼容的 EvalScope 请求微批处理。工具评测要求 sampler 返回结构化 tool calls，或提供 `template.parse_tool_call()`。HTTP sampler 同时接受 Twinkle `SamplingParams` 和字典。
+
+## 能力边界
+
+支持文本/对话、函数调用和 EvalScope Native AgentLoop；不支持多模态、流式、EvalScope 非 Native backend，以及在 adapter 内嵌套 Twinkle rollout。显式 generation 参数必须精确映射，否则会在评测前报错，不会静默近似或忽略。
+
+`run()` 原样返回 EvalScope 的 `dict[str, Report]`。`resolved_task_config` 可获取实际构建的配置，EvalScope 创建输出目录后可通过 `output_dir` 获取。其余透传字段请参考 EvalScope `TaskConfig`；benchmark 的额外依赖仍应按 EvalScope 文档单独安装。
diff --git a/pyproject.toml b/pyproject.toml
index 27a720456..3110a7113 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -25,6 +25,7 @@ twinkle-server = "twinkle.server.cli:main"
 twinkle-auto = "twinkle_client.auto:main"
 
 [project.optional-dependencies]
+eval = ["evalscope>=1.11,<1.12"]
 megatron = ["megatron-core>=0.12.0", "transformer-engine[pytorch]", "mcore_bridge"]
 data = ["py-data-juicer"]
 rl = [
diff --git a/src/twinkle_agentic/evaluator/__init__.py b/src/twinkle_agentic/evaluator/__init__.py
new file mode 100644
index 000000000..099110a29
--- /dev/null
+++ b/src/twinkle_agentic/evaluator/__init__.py
@@ -0,0 +1,12 @@
+"""EvalScope-backed evaluation for Twinkle Agentic backends."""
+
+from ._contracts import BackendContractError, EvaluatorConfigError, SamplerBatchError, UnsupportedCapabilityError
+from .evaluator import Evaluator
+
+__all__ = [
+    'BackendContractError',
+    'Evaluator',
+    'EvaluatorConfigError',
+    'SamplerBatchError',
+    'UnsupportedCapabilityError',
+]
diff --git a/src/twinkle_agentic/evaluator/_batcher.py b/src/twinkle_agentic/evaluator/_batcher.py
new file mode 100644
index 000000000..d960961a2
--- /dev/null
+++ b/src/twinkle_agentic/evaluator/_batcher.py
@@ -0,0 +1,157 @@
+"""A single-worker micro-batcher for structurally compatible samplers."""
+
+from collections import deque
+from concurrent.futures import Future
+from dataclasses import dataclass
+from threading import Condition, Thread
+from time import monotonic
+from typing import Any, Hashable, Mapping
+
+from twinkle.data_format import SamplingParams, Trajectory
+
+from ._contracts import BackendContractError, SamplerBatchError
+
+
+def _freeze(value: Any) -> Hashable:
+    if isinstance(value, Mapping):
+        return tuple(sorted((key, _freeze(item)) for key, item in value.items()))
+    if isinstance(value, (list, tuple)):
+        return tuple(_freeze(item) for item in value)
+    if isinstance(value, (str, int, float, bool, type(None))):
+        return value
+    try:
+        hash(value)
+    except TypeError as exc:
+        raise ValueError(f'Cannot safely batch unhashable value of type {type(value).__name__}') from exc
+    return value
+
+
+def _params_key(params: SamplingParams) -> Hashable:
+    return _freeze(vars(params))
+
+
+@dataclass
+class _BatchRequest:
+    trajectory: Trajectory
+    sampling_params: SamplingParams
+    sampler_kwargs: Mapping[str, Any]
+    compatibility_key: Hashable
+    future: Future
+    request_id: int
+
+
+class SamplerBatcher:
+    """Serialize sampler calls while coalescing equal requests into batches."""
+
+    def __init__(self, sampler: Any, *, batch_size: int, batch_wait_ms: float, sampler_kwargs: Mapping[str, Any]):
+        self._sampler = sampler
+        self._batch_size = batch_size
+        self._batch_wait_seconds = batch_wait_ms / 1000
+        self._sampler_kwargs = dict(sampler_kwargs)
+        self._queue: deque[_BatchRequest] = deque()
+        self._condition = Condition()
+        self._closed = False
+        self._request_id = 0
+        self._worker = Thread(target=self._run, name='twinkle-evaluator-sampler-batcher', daemon=False)
+        self._worker.start()
+
+    def submit(self, trajectory: Trajectory, sampling_params: SamplingParams) -> Any:
+        future: Future = Future()
+        key = (_params_key(sampling_params), _freeze(self._sampler_kwargs))
+        with self._condition:
+            if self._closed:
+                raise RuntimeError('Sampler batcher is closed')
+            request = _BatchRequest(
+                trajectory=trajectory,
+                sampling_params=sampling_params,
+                sampler_kwargs=self._sampler_kwargs,
+                compatibility_key=key,
+                future=future,
+                request_id=self._request_id,
+            )
+            self._request_id += 1
+            self._queue.append(request)
+            self._condition.notify()
+        return future.result()
+
+    def _pop_first(self) -> _BatchRequest | None:
+        return self._queue.popleft() if self._queue else None
+
+    def _take_compatible(self, key: Hashable, capacity: int) -> list[_BatchRequest]:
+        selected: list[_BatchRequest] = []
+        retained: deque[_BatchRequest] = deque()
+        while self._queue:
+            request = self._queue.popleft()
+            if request.compatibility_key == key and len(selected) < capacity:
+                selected.append(request)
+            else:
+                retained.append(request)
+        self._queue = retained
+        return selected
+
+    def _minimum_physical_batch_size(self) -> int:
+        mesh = getattr(self._sampler, 'device_mesh', None)
+        for name in ('data_world_size', 'dp_world_size'):
+            value = getattr(mesh, name, None)
+            if isinstance(value, int) and value > 0:
+                return value
+        return 1
+
+    def _complete_batch(self, requests: list[_BatchRequest]) -> None:
+        inputs = [request.trajectory for request in requests]
+        physical_size = max(len(inputs), self._minimum_physical_batch_size())
+        physical_inputs = inputs + [inputs[-1]] * (physical_size - len(inputs))
+        try:
+            responses = list(self._sampler.sample(
+                physical_inputs,
+                sampling_params=requests[0].sampling_params,
+                **requests[0].sampler_kwargs,
+            ))
+            if len(responses) != physical_size:
+                raise BackendContractError(
+                    f'Sampler returned {len(responses)} responses for physical batch size {physical_size}')
+        except Exception as exc:
+            error = SamplerBatchError(f'Sampler batch for {len(requests)} request(s) failed: {exc}')
+            for request in requests:
+                if not request.future.done():
+                    request.future.set_exception(error)
+            return
+        for request, response in zip(requests, responses):
+            if not request.future.done():
+                request.future.set_result(response)
+
+    def _run(self) -> None:
+        while True:
+            with self._condition:
+                while not self._queue and not self._closed:
+                    self._condition.wait()
+                if self._closed and not self._queue:
+                    return
+                first = self._pop_first()
+                assert first is not None
+                deadline = monotonic() + self._batch_wait_seconds
+                selected = [first]
+                selected.extend(self._take_compatible(first.compatibility_key, self._batch_size - len(selected)))
+                while len(selected) < self._batch_size and not self._closed:
+                    remaining = deadline - monotonic()
+                    if remaining <= 0:
+                        break
+                    self._condition.wait(remaining)
+                    selected.extend(self._take_compatible(first.compatibility_key, self._batch_size - len(selected)))
+                    if len(selected) > self._batch_size:
+                        overflow = selected[self._batch_size:]
+                        selected = selected[:self._batch_size]
+                        self._queue.extendleft(reversed(overflow))
+            self._complete_batch(selected)
+
+    def close(self) -> None:
+        with self._condition:
+            if self._closed:
+                return
+            self._closed = True
+            while self._queue:
+                request = self._queue.popleft()
+                if not request.future.done():
+                    request.future.set_exception(RuntimeError('Sampler batcher was closed'))
+            self._condition.notify_all()
+        self._worker.join()
diff --git a/src/twinkle_agentic/evaluator/_contracts.py b/src/twinkle_agentic/evaluator/_contracts.py
new file mode 100644
index 000000000..7839c7ad4
--- /dev/null
+++ b/src/twinkle_agentic/evaluator/_contracts.py
@@ -0,0 +1,38 @@
+"""Small public-facing contracts shared by evaluator internals."""
+
+from typing import Any, Mapping, Protocol, Sequence, runtime_checkable
+
+from twinkle.data_format import SamplingParams, Trajectory
+
+
+class EvaluatorConfigError(ValueError):
+    """The evaluator constructor or its owned configuration is invalid."""
+
+
+class UnsupportedCapabilityError(ValueError):
+    """The selected backend cannot represent an explicit request exactly."""
+
+
+class BackendContractError(RuntimeError):
+    """A caller-owned API or sampler returned an invalid value."""
+
+
+class SamplerBatchError(RuntimeError):
+    """A single physical sampler batch failed."""
+
+
+@runtime_checkable
+class SamplerLike(Protocol):
+    def sample(
+        self,
+        inputs: list[Trajectory],
+        sampling_params: SamplingParams | Mapping[str, Any],
+        **kwargs: Any,
+    ) -> Sequence[Any]: ...
+
+
+def read_value(value: Any, name: str, default: Any = None) -> Any:
+    """Read an attribute or mapping key without treating falsy values as absent."""
+    if isinstance(value, Mapping):
+        return value.get(name, default)
+    return getattr(value, name, default)
diff --git a/src/twinkle_agentic/evaluator/_evalscope_adapter.py b/src/twinkle_agentic/evaluator/_evalscope_adapter.py
new file mode 100644
index 000000000..b7fe86e61
--- /dev/null
+++ b/src/twinkle_agentic/evaluator/_evalscope_adapter.py
@@ -0,0 +1,320 @@
+"""EvalScope ``ModelAPI`` adapters and deliberately narrow converters."""
+
+import json
+from copy import deepcopy
+from threading import Lock
+from time import monotonic
+from typing import Any, Mapping, Sequence
+
+from evalscope.api.messages import (ChatMessageAssistant, ContentReasoning, ContentText)
+from evalscope.api.model import (ChatCompletionChoice, GenerateConfig, ModelAPI, ModelOutput)
+from evalscope.api.model.model_output import Logprob, Logprobs, ModelUsage, TopLogprob, as_stop_reason
+from evalscope.api.tool import ToolCall, ToolFunction
+
+from twinkle.data_format import SamplingParams
+from twinkle_agentic.protocol.openai import OpenAI
+
+from ._batcher import SamplerBatcher
+from ._contracts import BackendContractError, UnsupportedCapabilityError, read_value
+
+
+_COMMON_FIELDS = {
+    'max_tokens', 'seed', 'stop_seqs', 'temperature', 'top_k', 'top_p', 'repetition_penalty', 'n', 'logprobs',
+    'top_logprobs',
+}
+_OPENAI_FIELDS = {
+    'timeout', 'frequency_penalty', 'presence_penalty', 'logit_bias',
+    'parallel_tool_calls', 'reasoning_effort', 'reasoning_summary', 'extra_body', 'extra_query', 'extra_headers',
+}
+
+
+def _content_parts(content: Any) -> tuple[str, str | None]:
+    if isinstance(content, str):
+        return content, None
+    if not isinstance(content, Sequence):
+        raise UnsupportedCapabilityError(f'Unsupported message content type {type(content).__name__}')
+    texts: list[str] = []
+    reasoning: list[str] = []
+    for part in content:
+        kind = read_value(part, 'type')
+        if kind == 'text':
+            texts.append(read_value(part, 'text', ''))
+        elif kind == 'reasoning':
+            reasoning.append(read_value(part, 'reasoning', ''))
+        else:
+            raise UnsupportedCapabilityError(f'Multimodal content type {kind!r} is unsupported by Twinkle Evaluator')
+    return '\n'.join(texts), '\n'.join(reasoning) if reasoning else None
+
+
+def _tool_call_to_twinkle(tool_call: Any) -> dict[str, Any]:
+    function = read_value(tool_call, 'function', {})
+    arguments = read_value(function, 'arguments', {})
+    return {
+        'id': read_value(tool_call, 'id'),
+        'type': read_value(tool_call, 'type', 'function') or 'function',
+        'function': {'name': read_value(function, 'name'), 'arguments': deepcopy(arguments)},
+    }
+
+
+def to_twinkle_trajectory(input: Sequence[Any], tools: Sequence[Any], *, include_tools: bool = True) -> dict[str, Any]:
+    messages: list[dict[str, Any]] = []
+    for source in input:
+        role = read_value(source, 'role')
+        content, reasoning = _content_parts(read_value(source, 'content'))
+        if role not in {'system', 'user', 'assistant', 'tool'}:
+            raise BackendContractError(f'Unsupported EvalScope message role {role!r}')
+        message: dict[str, Any] = {'role': role, 'content': content}
+        if reasoning:
+            message['reasoning_content'] = reasoning
+        if role == 'assistant':
+            calls = read_value(source, 'tool_calls')
+            if calls:
+                message['tool_calls'] = [_tool_call_to_twinkle(call) for call in calls]
+        if role == 'tool':
+            call_id = read_value(source, 'tool_call_id')
+            if call_id:
+                message['tool_call_id'] = call_id
+        messages.append(message)
+    trajectory: dict[str, Any] = {'messages': messages}
+    if include_tools and tools:
+        trajectory['tools'] = [{
+            'type': 'function',
+            'function': {
+                'name': tool.name,
+                'description': tool.description,
+                'parameters': tool.parameters.model_dump(exclude_none=True),
+            },
+        } for tool in tools]
+    return trajectory
+
+
+def _normalize_tool_calls(raw_calls: Any, request_id: int, choice_index: int) -> list[ToolCall]:
+    if not isinstance(raw_calls, Sequence) or isinstance(raw_calls, (str, bytes)):
+        raise BackendContractError('tool_calls must be a sequence')
+    calls: list[ToolCall] = []
+    for call_index, raw in enumerate(raw_calls):
+        function = read_value(raw, 'function')
+        name = read_value(function, 'name')
+        arguments = read_value(function, 'arguments')
+        if not isinstance(name, str) or not name:
+            raise BackendContractError(f'Tool call {call_index} has no function name')
+        if isinstance(arguments, str):
+            try:
+                arguments = json.loads(arguments)
+            except json.JSONDecodeError as exc:
+                raise BackendContractError(f'Tool call {call_index} has invalid JSON arguments') from exc
+        if not isinstance(arguments, Mapping):
+            raise BackendContractError(f'Tool call {call_index} arguments must be an object')
+        call_id = read_value(raw, 'id') or f'call_{request_id}_{choice_index}_{call_index}'
+        calls.append(ToolCall(id=call_id, function=ToolFunction(name=name, arguments=dict(arguments))))
+    return calls
+
+
+def _assistant_choice(raw: Any, *, model: str, request_id: int, choice_index: int, decoded: str | None = None,
+                      stop_reason: str | None = None, logprobs: Logprobs | None = None) -> ChatCompletionChoice:
+    content = read_value(raw, 'content', decoded if decoded is not None else '')
+    reasoning = read_value(raw, 'reasoning_content')
+    calls = read_value(raw, 'tool_calls')
+    content_parts: list[Any] = []
+    if reasoning:
+        content_parts.append(ContentReasoning(reasoning=reasoning))
+    if content is not None:
+        content_parts.append(ContentText(text=content))
+    message_content: str | list[Any] = content_parts if reasoning else (content if content is not None else '')
+    tool_calls = _normalize_tool_calls(calls, request_id, choice_index) if calls else None
+    return ChatCompletionChoice(
+        message=ChatMessageAssistant(content=message_content, tool_calls=tool_calls, model=model),
+        stop_reason='tool_calls' if tool_calls else as_stop_reason(stop_reason),
+        logprobs=logprobs,
+    )
+
+
+def _sequence_logprobs(sequence: Any, template: Any) -> Logprobs | None:
+    values = read_value(sequence, 'logprobs')
+    if values is None:
+        return None
+    tokens = read_value(sequence, 'tokens')
+    if not isinstance(tokens, Sequence) or len(values) != len(tokens) or template is None:
+        raise UnsupportedCapabilityError('Exact token logprobs require token ids, aligned logprobs, and a template decoder')
+    result: list[Logprob] = []
+    for token_id, position in zip(tokens, values):
+        if not position:
+            raise BackendContractError('Sampler returned an empty logprob position')
+        token = template.decode([token_id])
+        top = [TopLogprob(token=template.decode([candidate]), logprob=score) for candidate, score in position]
+        own = next((score for candidate, score in position if candidate == token_id), position[0][1])
+        result.append(Logprob(token=token, logprob=own, top_logprobs=top))
+    return Logprobs(content=result)
+
+
+class _BaseModelAPI(ModelAPI):
+    def __init__(self, model_id: str, explicit_generation_keys: set[str]) -> None:
+        super().__init__(model_name=model_id)
+        self._explicit_generation_keys = set(explicit_generation_keys)
+        self._request_id = 0
+        self._request_lock = Lock()
+
+    def _next_request_id(self) -> int:
+        with self._request_lock:
+            value = self._request_id
+            self._request_id += 1
+            return value
+
+    def _unsupported(self, supported: set[str]) -> list[str]:
+        ignored = {'batch_size'}
+        unsupported = self._explicit_generation_keys - supported - ignored
+        if 'stream' in unsupported:
+            unsupported.remove('stream')
+        return sorted(unsupported)
+
+    def _sampling_params(self, config: GenerateConfig, *, is_openai: bool = False) -> SamplingParams:
+        repetition = config.repetition_penalty if config.repetition_penalty is not None else 1.0
+        if is_openai and 'repetition_penalty' in self._explicit_generation_keys:
+            extra = config.extra_body or {}
+            if extra.get('repetition_penalty') != config.repetition_penalty:
+                raise UnsupportedCapabilityError(
+                    'protocol.OpenAI cannot map repetition_penalty exactly; pass the same value in extra_body or omit it')
+            repetition = 1.0
+        return SamplingParams(
+            max_tokens=config.max_tokens,
+            seed=config.seed,
+            stop=config.stop_seqs,
+            temperature=config.temperature if config.temperature is not None else 1.0,
+            top_k=config.top_k if config.top_k is not None else -1,
+            top_p=config.top_p if config.top_p is not None else 1.0,
+            repetition_penalty=repetition,
+            logprobs=config.top_logprobs if config.logprobs else None,
+            num_samples=config.n if config.n is not None else 1,
+        )
+
+    def _validate_common(self, config: GenerateConfig, supported: set[str]) -> None:
+        if 'stream' in self._explicit_generation_keys and config.stream:
+            raise UnsupportedCapabilityError('Streaming is unsupported by Twinkle Evaluator')
+        unsupported = self._unsupported(supported)
+        if unsupported:
+            raise UnsupportedCapabilityError(
+                f'{type(self).__name__} for {self.model_name} cannot represent explicit generation fields: '
+                f"{', '.join(unsupported)}")
+        if 'top_logprobs' in self._explicit_generation_keys and not config.logprobs:
+            raise UnsupportedCapabilityError('top_logprobs requires logprobs=True')
+
+
+class ProtocolModelAPI(_BaseModelAPI):
+    def __init__(self, api: Any, model_id: str, explicit_generation_keys: set[str]) -> None:
+        super().__init__(model_id, explicit_generation_keys)
+        self.api = api
+
+    def validate_generation_config(self, config: GenerateConfig) -> None:
+        supported = _COMMON_FIELDS | (_OPENAI_FIELDS if isinstance(self.api, OpenAI) else set())
+        self._validate_common(config, supported)
+        self._sampling_params(config, is_openai=isinstance(self.api, OpenAI))
+
+    def _api_overrides(self, config: GenerateConfig, tool_choice: Any) -> dict[str, Any]:
+        overrides: dict[str, Any] = {}
+        if isinstance(self.api, OpenAI):
+            for name in _OPENAI_FIELDS:
+                if name in self._explicit_generation_keys:
+                    overrides[name] = getattr(config, name)
+            if tool_choice == 'any':
+                overrides['tool_choice'] = 'required'
+            elif tool_choice == 'none':
+                overrides['tool_choice'] = 'none'
+            elif tool_choice != 'auto':
+                overrides['tool_choice'] = {'type': 'function', 'function': {'name': tool_choice.name}}
+        elif tool_choice not in ('auto', None):
+            overrides['tool_choice'] = tool_choice
+        return overrides
+
+    def generate(self, input: list[Any], tools: list[Any], tool_choice: Any, config: GenerateConfig) -> ModelOutput:
+        self.validate_generation_config(config)
+        if tools and tool_choice != 'none' and config.n not in (None, 1):
+            raise UnsupportedCapabilityError('Agent/tool evaluation requires generation_config.n == 1')
+        request_id = self._next_request_id()
+        trajectory = to_twinkle_trajectory(input, tools, include_tools=tool_choice != 'none')
+        started = monotonic()
+        response = self.api(
+            trajectory,
+            self._sampling_params(config, is_openai=isinstance(self.api, OpenAI)),
+            **self._api_overrides(config, tool_choice),
+        )
+        elapsed = monotonic() - started
+        choices_raw = response if isinstance(response, list) else [response]
+        if not choices_raw:
+            raise BackendContractError(f'API {type(self.api).__name__} returned no choices for {self.model_name}')
+        if any(not isinstance(item, Mapping) for item in choices_raw):
+            raise BackendContractError(f'API {type(self.api).__name__} returned a non-message choice for {self.model_name}')
+        choices = [_assistant_choice(
+            item, model=self.model_name, request_id=request_id, choice_index=index,
+            stop_reason=read_value(item, 'finish_reason'),
+        ) for index, item in enumerate(choices_raw)]
+        return ModelOutput(model=self.model_name, choices=choices, time=elapsed)
+
+
+class SamplerModelAPI(_BaseModelAPI):
+    def __init__(self, sampler: Any, model_id: str, template: Any, explicit_generation_keys: set[str], *, batch_size: int,
+                 batch_wait_ms: float, sampler_kwargs: Mapping[str, Any]) -> None:
+        super().__init__(model_id, explicit_generation_keys)
+        self.sampler = sampler
+        self.template = template
+        self.batcher = SamplerBatcher(
+            sampler, batch_size=batch_size, batch_wait_ms=batch_wait_ms, sampler_kwargs=sampler_kwargs)
+
+    def validate_generation_config(self, config: GenerateConfig) -> None:
+        self._validate_common(config, _COMMON_FIELDS)
+        self._sampling_params(config)
+
+    def generate(self, input: list[Any], tools: list[Any], tool_choice: Any, config: GenerateConfig) -> ModelOutput:
+        self.validate_generation_config(config)
+        if tool_choice not in ('auto', 'none', None):
+            raise UnsupportedCapabilityError('Sampler backends support only tool_choice="auto" or "none"')
+        if tools and tool_choice != 'none' and config.n not in (None, 1):
+            raise UnsupportedCapabilityError('Agent/tool evaluation requires generation_config.n == 1')
+        request_id = self._next_request_id()
+        trajectory = to_twinkle_trajectory(input, tools, include_tools=tool_choice != 'none')
+        started = monotonic()
+        response = self.batcher.submit(trajectory, self._sampling_params(config))
+        elapsed = monotonic() - started
+        sequences = read_value(response, 'sequences')
+        if not isinstance(sequences, Sequence) or not sequences:
+            raise BackendContractError(f'Sampler response for {self.model_name} has no sequences')
+        choices: list[ChatCompletionChoice] = []
+        for index, sequence in enumerate(sequences):
+            stop_reason = read_value(sequence, 'stop_reason')
+            if stop_reason in ('abort', 'error'):
+                raise BackendContractError(f'Sampler response sequence {index} ended with {stop_reason}')
+            decoded = read_value(sequence, 'decoded')
+            feature = read_value(sequence, 'new_input_feature')
+            messages = read_value(feature, 'messages', []) if feature is not None else []
+            structured = messages[-1] if messages and read_value(messages[-1], 'role') == 'assistant' else None
+            if decoded is None and structured is None:
+                raise BackendContractError(f'Sampler response sequence {index} has neither decoded text nor assistant message')
+            raw = dict(structured) if isinstance(structured, Mapping) else {}
+            if not raw:
+                raw['content'] = decoded
+            elif raw.get('content') is None:
+                raw['content'] = decoded or ''
+            if tools and tool_choice != 'none' and not raw.get('tool_calls'):
+                if self.template is None or not hasattr(self.template, 'parse_tool_call'):
+                    raise UnsupportedCapabilityError('Sampler tool calls require structured output or template.parse_tool_call()')
+                parsed = self.template.parse_tool_call(decoded or '')
+                if parsed:
+                    raw['tool_calls'] = parsed
+            choices.append(_assistant_choice(
+                raw,
+                model=self.model_name,
+                request_id=request_id,
+                choice_index=index,
+                decoded=decoded,
+                stop_reason=stop_reason,
+                logprobs=_sequence_logprobs(sequence, self.template)
+                if 'top_logprobs' in self._explicit_generation_keys else None,
+            ))
+        prompt_ids = read_value(response, 'prompt_token_ids')
+        output_tokens = sum(len(read_value(sequence, 'tokens', []) or []) for sequence in sequences)
+        input_tokens = len(prompt_ids) if prompt_ids is not None else 0
+        return ModelOutput(
+            model=self.model_name,
+            choices=choices,
+            usage=ModelUsage(input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=input_tokens + output_tokens),
+            time=elapsed,
+        )
diff --git a/src/twinkle_agentic/evaluator/evaluator.py b/src/twinkle_agentic/evaluator/evaluator.py
new file mode 100644
index 000000000..40fbfc305
--- /dev/null
+++ b/src/twinkle_agentic/evaluator/evaluator.py
@@ -0,0 +1,155 @@
+"""The lightweight, single-use EvalScope facade."""
+
+from copy import deepcopy
+from enum import Enum
+from threading import Lock
+from typing import Any, Mapping, Sequence
+
+from twinkle_agentic.protocol.base import API
+
+from ._contracts import EvaluatorConfigError
+
+
+_OWNED_TASK_KEYS = {
+    'model', 'model_id', 'datasets', 'eval_type', 'eval_backend', 'model_task', 'api_url', 'api_key', 'model_args',
+}
+
+
+class _State(Enum):
+    NEW = 'new'
+    RUNNING = 'running'
+    SUCCEEDED = 'succeeded'
+    FAILED = 'failed'
+
+
+def _copy_mapping(value: Mapping[str, Any] | None, name: str) -> dict[str, Any]:
+    if value is None:
+        return {}
+    if not isinstance(value, Mapping):
+        raise EvaluatorConfigError(f'{name} must be a mapping')
+    return deepcopy(dict(value))
+
+
+def _explicit_generation_keys(value: Any) -> set[str]:
+    if isinstance(value, Mapping):
+        return set(value)
+    return set(getattr(value, 'model_fields_set', set()))
+
+
+class Evaluator:
+    """Evaluate one Twinkle API or sampler through EvalScope's native runner."""
+
+    def __init__(
+        self,
+        *,
+        datasets: Sequence[str],
+        sampler: object | None = None,
+        api: API | None = None,
+        model_id: str | None = None,
+        template: object | None = None,
+        sampler_kwargs: Mapping[str, Any] | None = None,
+        task_config: Mapping[str, Any] | None = None,
+        sampler_batch_size: int | None = None,
+        sampler_batch_wait_ms: float = 5.0,
+    ) -> None:
+        if isinstance(datasets, (str, bytes)) or not isinstance(datasets, Sequence) or not datasets:
+            raise EvaluatorConfigError('datasets must be a non-empty sequence of non-empty strings')
+        self._datasets = list(datasets)
+        if any(not isinstance(item, str) or not item.strip() for item in self._datasets):
+            raise EvaluatorConfigError('datasets must contain only non-empty strings')
+        if (sampler is None) == (api is None):
+            raise EvaluatorConfigError('provide exactly one of sampler or api')
+        if api is not None and not isinstance(api, API):
+            raise EvaluatorConfigError('api must implement twinkle_agentic.protocol.base.API')
+        self._sampler = sampler
+        self._api = api
+        self._task_config = _copy_mapping(task_config, 'task_config')
+        conflicts = sorted(_OWNED_TASK_KEYS.intersection(self._task_config))
+        if conflicts:
+            names = ', '.join(conflicts)
+            raise EvaluatorConfigError(f'task_config cannot set Twinkle-owned field(s): {names}')
+        self._sampler_kwargs = _copy_mapping(sampler_kwargs, 'sampler_kwargs')
+        if not isinstance(sampler_batch_wait_ms, (int, float)) or sampler_batch_wait_ms < 0:
+            raise EvaluatorConfigError('sampler_batch_wait_ms must be >= 0')
+        if sampler_batch_size is not None and (not isinstance(sampler_batch_size, int) or sampler_batch_size < 1):
+            raise EvaluatorConfigError('sampler_batch_size must be an integer >= 1')
+        if api is not None and (template is not None or self._sampler_kwargs or sampler_batch_size is not None
+                                or sampler_batch_wait_ms != 5.0):
+            raise EvaluatorConfigError('template and sampler batching options are only valid with sampler')
+        self._template = template if sampler is not None else None
+        if self._template is None and sampler is not None:
+            self._template = getattr(sampler, 'template', None)
+        inferred = model_id or getattr(sampler, 'model_id', None) or getattr(api, 'model', None) or getattr(api, 'model_name', None)
+        if not isinstance(inferred, str) or not inferred.strip():
+            raise EvaluatorConfigError('model_id is required when it cannot be inferred from sampler.model_id or api.model')
+        self._model_id = inferred
+        self._sampler_batch_size = sampler_batch_size or self._task_config.get('eval_batch_size', 8)
+        if not isinstance(self._sampler_batch_size, int) or self._sampler_batch_size < 1:
+            raise EvaluatorConfigError('task_config.eval_batch_size must be an integer >= 1')
+        self._sampler_batch_wait_ms = float(sampler_batch_wait_ms)
+        self._generation_keys = _explicit_generation_keys(self._task_config.get('generation_config', {}))
+        self._state = _State.NEW
+        self._state_lock = Lock()
+        self._resolved_task_config: Any = None
+        self._output_dir: str | None = None
+
+    @property
+    def resolved_task_config(self) -> Any:
+        return self._resolved_task_config
+
+    @property
+    def output_dir(self) -> str | None:
+        return self._output_dir
+
+    def run(self) -> Any:
+        with self._state_lock:
+            if self._state is not _State.NEW:
+                raise RuntimeError('Evaluator instances are single-use; create a new Evaluator to run again')
+            self._state = _State.RUNNING
+        batcher = None
+        try:
+            try:
+                from ._evalscope_adapter import ProtocolModelAPI, SamplerModelAPI
+                from evalscope.config import TaskConfig
+                from evalscope.constants import EvalBackend, EvalType
+                from evalscope.run import run_task
+            except ImportError as exc:
+                raise ImportError("Evaluator requires EvalScope. Install it with:\n  pip install 'twinkle-kit[eval]'") from exc
+            if self._api is not None:
+                adapter = ProtocolModelAPI(self._api, self._model_id, self._generation_keys)
+            else:
+                adapter = SamplerModelAPI(
+                    self._sampler,
+                    self._model_id,
+                    self._template,
+                    self._generation_keys,
+                    batch_size=self._sampler_batch_size,
+                    batch_wait_ms=self._sampler_batch_wait_ms,
+                    sampler_kwargs=self._sampler_kwargs,
+                )
+                batcher = adapter.batcher
+            config = dict(self._task_config)
+            config.setdefault('eval_batch_size', 8)
+            config.update({
+                'model': adapter,
+                'model_id': self._model_id,
+                'datasets': list(self._datasets),
+                'eval_type': EvalType.CUSTOM,
+                'eval_backend': EvalBackend.NATIVE,
+                'model_task': 'text_generation',
+            })
+            self._resolved_task_config = TaskConfig(**config)
+            adapter.validate_generation_config(self._resolved_task_config.generation_config)
+            result = run_task(self._resolved_task_config)
+            self._output_dir = self._resolved_task_config.work_dir
+        except Exception:
+            with self._state_lock:
+                self._state = _State.FAILED
+            raise
+        else:
+            with self._state_lock:
+                self._state = _State.SUCCEEDED
+            return result
+        finally:
+            if batcher is not None:
+                batcher.close()
diff --git a/src/twinkle_client/sampler/vllm_sampler.py b/src/twinkle_client/sampler/vllm_sampler.py
index 0d553bb32..0e8d8cf3d 100644
--- a/src/twinkle_client/sampler/vllm_sampler.py
+++ b/src/twinkle_client/sampler/vllm_sampler.py
@@ -1,8 +1,9 @@
+from dataclasses import asdict
 from typing import Any, Dict, List, Optional, Union
 from twinkle_client.http import http_post
 from twinkle_client.types.sampler import AddAdapterResponse, SampleResponseModel, SetTemplateResponse
 from peft import PeftConfig
-from twinkle.data_format import Trajectory, InputFeature
+from twinkle.data_format import Trajectory, InputFeature, SamplingParams
 
 
 # Intentionally does NOT subclass ``twinkle.sampler.base.Sampler``: importing
@@ -68,7 +69,7 @@ def add_adapter_to_sampler(self, adapter_name: str, config: PeftConfig, **kwargs
     def sample(
         self,
         inputs: Union[List[Trajectory], List[InputFeature]],
-        sampling_params: Optional[Dict[str, Any]] = None,
+        sampling_params: Optional[Union[SamplingParams, Dict[str, Any]]] = None,
         adapter_name: str = '',
         adapter_uri: Optional[str] = None,
         num_samples: int = 1,
@@ -77,7 +78,7 @@ def sample(
 
         Args:
             inputs: List of Trajectory or InputFeature to sample from.
-            sampling_params: Sampling parameters dict.
+            sampling_params: Sampling parameters mapping or Twinkle ``SamplingParams``.
             adapter_name: Adapter name for LoRA inference.
             adapter_uri: Adapter URI (twinkle:// path or local path) for LoRA inference.
             num_samples: Number of completions to generate per prompt.
@@ -85,11 +86,21 @@ def sample(
         Returns:
             SampleResponseModel with 'sequences' list, each containing tokens, logprobs, stop_reason.
         """
+        if isinstance(sampling_params, SamplingParams):
+            sampling_params = asdict(sampling_params)
+        elif sampling_params is not None:
+            sampling_params = dict(sampling_params)
+        if num_samples != 1:
+            if sampling_params is None:
+                sampling_params = {'num_samples': num_samples}
+            elif sampling_params.get('num_samples', num_samples) != num_samples:
+                raise ValueError('num_samples conflicts with sampling_params.num_samples')
+            else:
+                sampling_params['num_samples'] = num_samples
         json_data = {
             'inputs': _json_safe(inputs),
-            'sampling_params': sampling_params,
+            'sampling_params': _json_safe(sampling_params),
             'adapter_name': adapter_name,
-            'num_samples': num_samples,
         }
         if adapter_uri is not None:
             json_data['adapter_uri'] = adapter_uri
diff --git a/tests/twinkle_agentic/evaluator/__init__.py b/tests/twinkle_agentic/evaluator/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/twinkle_agentic/evaluator/conftest.py b/tests/twinkle_agentic/evaluator/conftest.py
new file mode 100644
index 000000000..f5a841887
--- /dev/null
+++ b/tests/twinkle_agentic/evaluator/conftest.py
@@ -0,0 +1,51 @@
+from dataclasses import dataclass
+from typing import Any
+
+import pytest
+
+from twinkle.data_format import SampleResponse, SampledSequence
+from twinkle_agentic.protocol.base import API
+
+
+class RecordingAPI(API):
+    model = 'recording-api'
+
+    def __init__(self, response: Any | None = None):
+        self.calls = []
+        self.response = response or {'role': 'assistant', 'content': 'ok', 'finish_reason': 'stop'}
+
+    def __call__(self, trajectory, sampling_params, **kwargs):
+        self.calls.append((trajectory, sampling_params, kwargs))
+        return self.response
+
+
+class RecordingSampler:
+    model_id = 'recording-sampler'
+
+    def __init__(self, dp_world_size: int = 1):
+        self.calls = []
+        self.device_mesh = type('Mesh', (), {'dp_world_size': dp_world_size})()
+
+    def sample(self, inputs, sampling_params, **kwargs):
+        self.calls.append((list(inputs), sampling_params, kwargs))
+        return [SampleResponse(sequences=[SampledSequence(stop_reason='stop', tokens=[1], decoded='ok')]) for _ in inputs]
+
+
+class ToolTemplate:
+    def decode(self, tokens):
+        return ''.join(str(token) for token in tokens)
+
+    def parse_tool_call(self, decoded):
+        if decoded == 'tool':
+            return [{'id': 'parsed', 'type': 'function', 'function': {'name': 'lookup', 'arguments': {'q': 'x'}}}]
+        return []
+
+
+@pytest.fixture
+def recording_api():
+    return RecordingAPI()
+
+
+@pytest.fixture
+def recording_sampler():
+    return RecordingSampler()
diff --git a/tests/twinkle_agentic/evaluator/test_batcher.py b/tests/twinkle_agentic/evaluator/test_batcher.py
new file mode 100644
index 000000000..b12c00184
--- /dev/null
+++ b/tests/twinkle_agentic/evaluator/test_batcher.py
@@ -0,0 +1,51 @@
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+from twinkle.data_format import SamplingParams
+from twinkle_agentic.evaluator._batcher import SamplerBatcher
+
+from .conftest import RecordingSampler
+
+
+def _trajectory(value):
+    return {'messages': [{'role': 'user', 'content': value}]}
+
+
+def test_compatible_requests_are_batched_and_tail_is_padded():
+    sampler = RecordingSampler(dp_world_size=4)
+    batcher = SamplerBatcher(sampler, batch_size=4, batch_wait_ms=20, sampler_kwargs={})
+    try:
+        with ThreadPoolExecutor(max_workers=3) as pool:
+            result = list(pool.map(lambda i: batcher.submit(_trajectory(str(i)), SamplingParams(max_tokens=3)), range(3)))
+        assert len(result) == 3
+        assert len(sampler.calls) == 1
+        assert len(sampler.calls[0][0]) == 4
+    finally:
+        batcher.close()
+    assert not batcher._worker.is_alive()
+
+
+def test_incompatible_requests_do_not_share_a_sampler_call():
+    sampler = RecordingSampler()
+    batcher = SamplerBatcher(sampler, batch_size=2, batch_wait_ms=0, sampler_kwargs={})
+    try:
+        batcher.submit(_trajectory('a'), SamplingParams(max_tokens=1))
+        batcher.submit(_trajectory('b'), SamplingParams(max_tokens=2))
+        assert [call[1].max_tokens for call in sampler.calls] == [1, 2]
+    finally:
+        batcher.close()
+
+
+def test_sampler_error_is_delivered_to_all_requests():
+    sampler = RecordingSampler()
+    sampler.sample = lambda *args, **kwargs: (_ for _ in ()).throw(ValueError('boom'))
+    batcher = SamplerBatcher(sampler, batch_size=2, batch_wait_ms=20, sampler_kwargs={})
+    try:
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            futures = [pool.submit(batcher.submit, _trajectory(str(i)), SamplingParams()) for i in range(2)]
+            for future in futures:
+                with pytest.raises(RuntimeError, match='boom'):
+                    future.result()
+    finally:
+        batcher.close()
diff --git a/tests/twinkle_agentic/evaluator/test_client_sampler.py b/tests/twinkle_agentic/evaluator/test_client_sampler.py
new file mode 100644
index 000000000..a542f0182
--- /dev/null
+++ b/tests/twinkle_agentic/evaluator/test_client_sampler.py
@@ -0,0 +1,26 @@
+from twinkle.data_format import SamplingParams
+from twinkle_client.sampler.vllm_sampler import vLLMSampler
+
+
+def test_http_sampler_serializes_sampling_params_dataclass_once(monkeypatch):
+    request = {}
+
+    class Response:
+        def raise_for_status(self):
+            pass
+
+        def json(self):
+            return {'samples': []}
+
+    def post(*, url, json_data):
+        request['url'] = url
+        request['body'] = json_data
+        return Response()
+
+    import twinkle_client.sampler.vllm_sampler as module
+    monkeypatch.setattr(module, 'http_post', post)
+    sampler = object.__new__(vLLMSampler)
+    sampler.server_url = 'http://example/sampler/model/twinkle'
+    sampler.sample([{'messages': []}], SamplingParams(max_tokens=4, num_samples=2))
+    assert request['body']['sampling_params']['num_samples'] == 2
+    assert 'num_samples' not in request['body']
diff --git a/tests/twinkle_agentic/evaluator/test_conversion.py b/tests/twinkle_agentic/evaluator/test_conversion.py
new file mode 100644
index 000000000..955732162
--- /dev/null
+++ b/tests/twinkle_agentic/evaluator/test_conversion.py
@@ -0,0 +1,72 @@
+import pytest
+
+from evalscope.api.messages import ChatMessageAssistant, ChatMessageTool, ChatMessageUser, ContentImage, ContentReasoning, ContentText
+from evalscope.api.model import GenerateConfig, Model
+from evalscope.api.tool import ToolInfo, ToolParams
+
+from twinkle_agentic.evaluator._contracts import BackendContractError, UnsupportedCapabilityError
+from twinkle_agentic.evaluator._evalscope_adapter import ProtocolModelAPI, SamplerModelAPI, to_twinkle_trajectory
+
+from .conftest import RecordingAPI, RecordingSampler, ToolTemplate
+
+
+def test_input_conversion_preserves_roles_reasoning_tools_and_does_not_mutate():
+    assistant = ChatMessageAssistant(
+        content=[ContentReasoning(reasoning='think'), ContentText(text='answer')],
+        tool_calls=[{'id': 'call-1', 'function': {'name': 'lookup', 'arguments': {'q': 'x'}}}],
+    )
+    tool = ChatMessageTool(content='result', tool_call_id='call-1')
+    source = [ChatMessageUser(content='question'), assistant, tool]
+    tool_info = ToolInfo(name='lookup', description='find', parameters=ToolParams())
+    trajectory = to_twinkle_trajectory(source, [tool_info])
+    assert trajectory['messages'][1]['reasoning_content'] == 'think'
+    assert trajectory['messages'][1]['tool_calls'][0]['function']['arguments'] == {'q': 'x'}
+    assert trajectory['tools'][0]['function']['name'] == 'lookup'
+    assert assistant.tool_calls[0].function.arguments == {'q': 'x'}
+
+
+def test_multimodal_input_is_rejected():
+    with pytest.raises(UnsupportedCapabilityError, match='Multimodal'):
+        to_twinkle_trajectory([ChatMessageUser(content=[ContentImage(image='x')])], [])
+
+
+def test_protocol_output_preserves_reasoning_multiple_choices_and_tools():
+    api = RecordingAPI([
+        {'role': 'assistant', 'content': 'one', 'reasoning_content': 'r', 'tool_calls': [
+            {'function': {'name': 'f', 'arguments': '{"a": 1}'}}]},
+        {'role': 'assistant', 'content': 'two', 'finish_reason': 'length'},
+    ])
+    output = Model(ProtocolModelAPI(api, 'fake', set()), GenerateConfig()).generate([ChatMessageUser(content='x')])
+    assert len(output.choices) == 2
+    assert output.choices[0].stop_reason == 'tool_calls'
+    assert output.choices[0].message.tool_calls[0].id == 'call_0_0_0'
+    assert output.choices[1].stop_reason == 'max_tokens'
+
+
+def test_invalid_tool_arguments_fail():
+    api = RecordingAPI({'role': 'assistant', 'tool_calls': [{'function': {'name': 'f', 'arguments': 'bad'}}]})
+    with pytest.raises(BackendContractError, match='invalid JSON'):
+        Model(ProtocolModelAPI(api, 'fake', set()), GenerateConfig()).generate([ChatMessageUser(content='x')])
+
+
+def test_explicit_unsupported_config_fails_before_api_call():
+    api = RecordingAPI()
+    adapter = ProtocolModelAPI(api, 'fake', {'response_schema', 'temperature'})
+    with pytest.raises(UnsupportedCapabilityError, match='response_schema'):
+        Model(adapter, GenerateConfig(response_schema={'name': 'x', 'json_schema': {'type': 'object'}}, temperature=0)).generate(
+            [ChatMessageUser(content='x')])
+    assert not api.calls
+
+
+def test_sampler_parses_tools_and_rejects_forcing():
+    sampler = RecordingSampler()
+    sampler.sample = lambda inputs, sampling_params, **kwargs: [
+        {'sequences': [{'stop_reason': 'stop', 'tokens': [1], 'decoded': 'tool'}]} for _ in inputs]
+    adapter = SamplerModelAPI(sampler, 's', ToolTemplate(), set(), batch_size=1, batch_wait_ms=0, sampler_kwargs={})
+    try:
+        output = Model(adapter, GenerateConfig()).generate([ChatMessageUser(content='x')], tools=[ToolInfo(name='lookup', description='x')])
+        assert output.stop_reason == 'tool_calls'
+        with pytest.raises(UnsupportedCapabilityError, match='tool_choice'):
+            adapter.generate([ChatMessageUser(content='x')], [], 'any', GenerateConfig())
+    finally:
+        adapter.batcher.close()
diff --git a/tests/twinkle_agentic/evaluator/test_evaluator.py b/tests/twinkle_agentic/evaluator/test_evaluator.py
new file mode 100644
index 000000000..1b4365266
--- /dev/null
+++ b/tests/twinkle_agentic/evaluator/test_evaluator.py
@@ -0,0 +1,89 @@
+import pytest
+
+from twinkle_agentic.evaluator import Evaluator
+from twinkle_agentic.evaluator._contracts import EvaluatorConfigError
+
+from .conftest import RecordingAPI, RecordingSampler
+
+
+def test_constructor_validates_and_copies_inputs():
+    sampler = RecordingSampler()
+    config = {'limit': 1}
+    kwargs = {'adapter_name': 'a'}
+    evaluator = Evaluator(datasets=['gsm8k'], sampler=sampler, task_config=config, sampler_kwargs=kwargs)
+    config['limit'] = 2
+    kwargs['adapter_name'] = 'changed'
+    assert evaluator._task_config['limit'] == 1
+    assert evaluator._sampler_kwargs['adapter_name'] == 'a'
+
+
+@pytest.mark.parametrize('sampler,api', [(None, None), (RecordingSampler(), RecordingAPI())])
+def test_requires_exactly_one_backend(sampler, api):
+    with pytest.raises(EvaluatorConfigError, match='exactly one'):
+        Evaluator(datasets=['x'], sampler=sampler, api=api)
+
+
+@pytest.mark.parametrize('datasets', [[], [''], 'gsm8k'])
+def test_datasets_are_validated(datasets):
+    with pytest.raises(EvaluatorConfigError, match='datasets'):
+        Evaluator(datasets=datasets, sampler=RecordingSampler())
+
+
+@pytest.mark.parametrize('key', ['model', 'model_id', 'datasets', 'eval_type', 'eval_backend', 'model_task', 'api_url', 'api_key', 'model_args'])
+def test_managed_config_keys_are_rejected(key):
+    with pytest.raises(EvaluatorConfigError, match=key):
+        Evaluator(datasets=['x'], sampler=RecordingSampler(), task_config={key: 'value'})
+
+
+def test_api_mode_rejects_sampler_only_options():
+    with pytest.raises(EvaluatorConfigError, match='sampler'):
+        Evaluator(datasets=['x'], api=RecordingAPI(), sampler_batch_size=2)
+
+
+def test_single_use_after_success(monkeypatch, recording_api):
+    sentinel = {'x': object()}
+    import evalscope.run
+    def run_task(config):
+        config.work_dir = 'outputs/resolved'
+        return sentinel
+    monkeypatch.setattr(evalscope.run, 'run_task', run_task)
+    evaluator = Evaluator(datasets=['x'], api=recording_api)
+    assert evaluator.output_dir is None
+    assert evaluator.run() is sentinel
+    assert evaluator.resolved_task_config.model is not None
+    assert evaluator.output_dir == 'outputs/resolved'
+    with pytest.raises(RuntimeError, match='single-use'):
+        evaluator.run()
+
+
+def test_single_use_after_failure(monkeypatch, recording_api):
+    import evalscope.run
+    monkeypatch.setattr(evalscope.run, 'run_task', lambda config: (_ for _ in ()).throw(ValueError('boom')))
+    evaluator = Evaluator(datasets=['x'], api=recording_api)
+    with pytest.raises(ValueError, match='boom'):
+        evaluator.run()
+    with pytest.raises(RuntimeError, match='single-use'):
+        evaluator.run()
+
+
+@pytest.mark.parametrize('backend', ['api', 'sampler'])
+def test_offline_native_evalscope_run(tmp_path, recording_api, recording_sampler, backend):
+    dataset = tmp_path / 'questions.jsonl'
+    dataset.write_text('{"question": "Say ok", "answer": "ok"}\n', encoding='utf-8')
+    kwargs = {'api': recording_api} if backend == 'api' else {'sampler': recording_sampler}
+    evaluator = Evaluator(
+        datasets=['general_qa'],
+        task_config={
+            'dataset_args': {'general_qa': {'local_path': str(dataset)}},
+            'dataset_hub': 'Local',
+            'work_dir': str(tmp_path / 'outputs'),
+            'no_timestamp': True,
+            'generation_config': {'temperature': 0.0},
+        },
+        **kwargs,
+    )
+    reports = evaluator.run()
+    assert reports
+    assert evaluator.output_dir == str(tmp_path / 'outputs')
+    if backend == 'sampler':
+        assert recording_sampler.calls