feat: add benchmark case catalog

Problem:
Benchmark comparisons duplicated scenario setup and lacked a strict, secret-safe configuration and reusable built-in case catalog.

Risk:
Catalog drift or mode-specific request and mock differences could invalidate benchmark parity; focused tests lock validation, immutable case semantics, and the existing 12-run assertions.

src/agent_lab/application/benchmark.py

@@ -0,0 +1,409 @@
+import json
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import Literal, TypeAlias
+from urllib.parse import urlsplit
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from agent_lab.application.contracts import (
+    AgentParams,
+    DebugRunRequest,
+    EventAgentParams,
+)
+from agent_lab.domain.events import ToolCallEvent
+from agent_lab.domain.messages import StreamItem
+
+
+BenchmarkMode: TypeAlias = Literal["dual_agent", "chat_agent_tools"]
+BenchmarkCaseId: TypeAlias = Literal[
+    "ordinary_chat",
+    "device_volume_silent",
+    "calendar_schedule_template",
+    "web_search_two_answers",
+    "session_terminate",
+    "parallel_volume_schedule",
+]
+JsonScalar: TypeAlias = str | int | float | bool | None
+
+BENCHMARK_MODES: tuple[BenchmarkMode, ...] = (
+    "dual_agent",
+    "chat_agent_tools",
+)
+BENCHMARK_CASE_IDS: tuple[BenchmarkCaseId, ...] = (
+    "ordinary_chat",
+    "device_volume_silent",
+    "calendar_schedule_template",
+    "web_search_two_answers",
+    "session_terminate",
+    "parallel_volume_schedule",
+)
+
+BENCHMARK_SYSTEM_PROMPT = "Exercise the requested business scenario."
+BENCHMARK_MAX_TOKENS = 128
+BENCHMARK_MAX_EVENT_LOOPS = 1
+BENCHMARK_MAX_PARALLEL_EVENTS = 2
+BENCHMARK_BATCH_TIMEOUT_SECONDS = 1.0
+
+
+class _StrictBenchmarkModel(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+        strict=True,
+        hide_input_in_errors=True,
+    )
+
+
+class _FrozenStrictBenchmarkModel(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+        strict=True,
+        hide_input_in_errors=True,
+        frozen=True,
+    )
+
+
+def _validate_base_url(value: str) -> str:
+    if not value or value != value.strip() or any(char.isspace() for char in value):
+        raise ValueError("base_url must be a non-empty HTTP(S) URL")
+    try:
+        parsed = urlsplit(value)
+        hostname = parsed.hostname
+        parsed.port
+    except ValueError as exc:
+        raise ValueError("base_url must be a valid HTTP(S) URL") from exc
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc or not hostname:
+        raise ValueError("base_url must use http or https and include a host")
+    if parsed.username is not None or parsed.password is not None or "@" in parsed.netloc:
+        raise ValueError("base_url must not contain userinfo")
+    if "?" in value:
+        raise ValueError("base_url must not contain a query")
+    if "#" in value:
+        raise ValueError("base_url must not contain a fragment")
+    return value
+
+
+def _validate_model(value: str) -> str:
+    stripped = value.strip()
+    if not stripped:
+        raise ValueError("model must not be empty")
+    return stripped
+
+
+class BenchmarkTarget(_StrictBenchmarkModel):
+    base_url: str
+    model: str
+
+    _base_url_validator = field_validator("base_url")(_validate_base_url)
+    _model_validator = field_validator("model")(_validate_model)
+
+
+class BenchmarkConfig(BenchmarkTarget):
+    schema_version: Literal[1]
+    runs_per_case: int = Field(default=1, ge=1)
+    modes: list[BenchmarkMode] = Field(
+        default_factory=lambda: list(BENCHMARK_MODES),
+        min_length=1,
+    )
+    cases: list[BenchmarkCaseId] = Field(
+        default_factory=lambda: list(BENCHMARK_CASE_IDS),
+        min_length=1,
+    )
+
+    @field_validator("schema_version", mode="before")
+    @classmethod
+    def require_integer_schema_version(cls, value: object) -> object:
+        if type(value) is not int or value != 1:
+            raise ValueError("schema_version must be the integer 1")
+        return value
+
+    @field_validator("modes", "cases")
+    @classmethod
+    def reject_duplicates(cls, values: list[str]) -> list[str]:
+        if len(values) != len(set(values)):
+            raise ValueError("values must not contain duplicates")
+        return values
+
+
+class BenchmarkExpectation(_FrozenStrictBenchmarkModel):
+    visible_messages: tuple[str, ...]
+    event_names: tuple[str, ...]
+    answer_count: int = Field(ge=1)
+    terminal: bool
+    first_reply_before_tool: bool
+
+    @model_validator(mode="after")
+    def match_answer_count(self) -> "BenchmarkExpectation":
+        if self.answer_count != len(self.visible_messages):
+            raise ValueError("answer_count must match visible_messages")
+        return self
+
+
+class BenchmarkMockEvent(_FrozenStrictBenchmarkModel):
+    id: str
+    name: str
+    arguments: tuple[tuple[str, JsonScalar], ...]
+
+    @model_validator(mode="after")
+    def validate_event(self) -> "BenchmarkMockEvent":
+        if not self.id.strip():
+            raise ValueError("mock event id must not be empty")
+        if not self.name.strip():
+            raise ValueError("mock event name must not be empty")
+        keys = tuple(key for key, _ in self.arguments)
+        if any(not key.strip() for key in keys):
+            raise ValueError("mock event argument names must not be empty")
+        if len(keys) != len(set(keys)):
+            raise ValueError("mock event arguments must not contain duplicate names")
+        return self
+
+
+class BenchmarkCase(_FrozenStrictBenchmarkModel):
+    case_id: BenchmarkCaseId
+    user_message: str
+    enabled_tools: tuple[str, ...]
+    expectation: BenchmarkExpectation
+    mock_visible_messages: tuple[str, ...]
+    mock_events: tuple[BenchmarkMockEvent, ...]
+
+    @model_validator(mode="after")
+    def validate_case_semantics(self) -> "BenchmarkCase":
+        if not self.user_message.strip():
+            raise ValueError("user_message must not be empty")
+        if len(self.enabled_tools) != len(set(self.enabled_tools)):
+            raise ValueError("enabled_tools must not contain duplicates")
+        mock_event_names = tuple(event.name for event in self.mock_events)
+        if mock_event_names != self.expectation.event_names:
+            raise ValueError("mock event order must match expected event order")
+        if self.expectation.first_reply_before_tool:
+            if not self.mock_events or not self.mock_visible_messages:
+                raise ValueError(
+                    "first_reply_before_tool requires mock text and at least one event"
+                )
+        return self
+
+
+def _mock_event(
+    event_id: str,
+    name: str,
+    arguments: tuple[tuple[str, JsonScalar], ...] = (),
+) -> BenchmarkMockEvent:
+    return BenchmarkMockEvent(id=event_id, name=name, arguments=arguments)
+
+
+_SCHEDULE_ARGUMENTS: tuple[tuple[str, JsonScalar], ...] = (
+    ("title", "Design review"),
+    ("start_at", "2026-07-14T09:30:00+08:00"),
+    ("timezone", "Asia/Shanghai"),
+)
+_SCHEDULE_CONFIRMATION = (
+    "Scheduled Design review for 2026-07-14T09:30:00+08:00 "
+    "(Asia/Shanghai)."
+)
+
+BENCHMARK_CASE_CATALOG: Mapping[BenchmarkCaseId, BenchmarkCase] = MappingProxyType(
+    {
+        "ordinary_chat": BenchmarkCase(
+            case_id="ordinary_chat",
+            user_message="Hello.",
+            enabled_tools=(
+                "session.terminate",
+                "device.volume.adjust",
+                "calendar.schedule.create",
+                "knowledge.web.search",
+            ),
+            expectation=BenchmarkExpectation(
+                visible_messages=("Ordinary answer.",),
+                event_names=(),
+                answer_count=1,
+                terminal=False,
+                first_reply_before_tool=False,
+            ),
+            mock_visible_messages=("Ordinary answer.",),
+            mock_events=(),
+        ),
+        "device_volume_silent": BenchmarkCase(
+            case_id="device_volume_silent",
+            user_message="set volume to 30",
+            enabled_tools=("device.volume.adjust",),
+            expectation=BenchmarkExpectation(
+                visible_messages=("I will set the volume to 30.",),
+                event_names=("device.volume.adjust",),
+                answer_count=1,
+                terminal=False,
+                first_reply_before_tool=True,
+            ),
+            mock_visible_messages=("I will set the volume to 30.",),
+            mock_events=(
+                _mock_event(
+                    "volume-1",
+                    "device.volume.adjust",
+                    (("mode", "absolute"), ("value", 30)),
+                ),
+            ),
+        ),
+        "calendar_schedule_template": BenchmarkCase(
+            case_id="calendar_schedule_template",
+            user_message=(
+                'schedule "Design review" at 2026-07-14T09:30:00+08:00 '
+                "timezone Asia/Shanghai"
+            ),
+            enabled_tools=("calendar.schedule.create",),
+            expectation=BenchmarkExpectation(
+                visible_messages=("I will schedule that.", _SCHEDULE_CONFIRMATION),
+                event_names=("calendar.schedule.create",),
+                answer_count=2,
+                terminal=False,
+                first_reply_before_tool=True,
+            ),
+            mock_visible_messages=("I will schedule that.",),
+            mock_events=(
+                _mock_event(
+                    "schedule-1",
+                    "calendar.schedule.create",
+                    _SCHEDULE_ARGUMENTS,
+                ),
+            ),
+        ),
+        "web_search_two_answers": BenchmarkCase(
+            case_id="web_search_two_answers",
+            user_message="event batch safety",
+            enabled_tools=("knowledge.web.search",),
+            expectation=BenchmarkExpectation(
+                visible_messages=("I will check.", "Grounded search update."),
+                event_names=("knowledge.web.search",),
+                answer_count=2,
+                terminal=False,
+                first_reply_before_tool=True,
+            ),
+            mock_visible_messages=("I will check.", "Grounded search update."),
+            mock_events=(
+                _mock_event(
+                    "search-1",
+                    "knowledge.web.search",
+                    (("query", "event batch safety"),),
+                ),
+            ),
+        ),
+        "session_terminate": BenchmarkCase(
+            case_id="session_terminate",
+            user_message="end this session",
+            enabled_tools=("session.terminate",),
+            expectation=BenchmarkExpectation(
+                visible_messages=("Goodbye.",),
+                event_names=("session.terminate",),
+                answer_count=1,
+                terminal=True,
+                first_reply_before_tool=True,
+            ),
+            mock_visible_messages=("Goodbye.",),
+            mock_events=(_mock_event("terminate-1", "session.terminate"),),
+        ),
+        "parallel_volume_schedule": BenchmarkCase(
+            case_id="parallel_volume_schedule",
+            user_message=(
+                'set volume to 30 and schedule "Design review" at '
+                "2026-07-14T09:30:00+08:00 timezone Asia/Shanghai"
+            ),
+            enabled_tools=(
+                "device.volume.adjust",
+                "calendar.schedule.create",
+            ),
+            expectation=BenchmarkExpectation(
+                visible_messages=(
+                    "I will set the volume to 30 and schedule that.",
+                    _SCHEDULE_CONFIRMATION,
+                ),
+                event_names=(
+                    "device.volume.adjust",
+                    "calendar.schedule.create",
+                ),
+                answer_count=2,
+                terminal=False,
+                first_reply_before_tool=True,
+            ),
+            mock_visible_messages=(
+                "I will set the volume to 30 and schedule that.",
+            ),
+            mock_events=(
+                _mock_event(
+                    "parallel-volume",
+                    "device.volume.adjust",
+                    (("mode", "absolute"), ("value", 30)),
+                ),
+                _mock_event(
+                    "parallel-schedule",
+                    "calendar.schedule.create",
+                    _SCHEDULE_ARGUMENTS,
+                ),
+            ),
+        ),
+    }
+)
+
+
+def build_benchmark_request(
+    case: BenchmarkCase,
+    mode: BenchmarkMode,
+    model: str,
+) -> DebugRunRequest:
+    if mode not in BENCHMARK_MODES:
+        raise ValueError(f"unsupported benchmark mode: {mode}")
+    validated_model = _validate_model(model)
+    return DebugRunRequest(
+        user_message=case.user_message,
+        system_prompts=[BENCHMARK_SYSTEM_PROMPT],
+        pre_messages=[],
+        chat_agent=AgentParams(
+            model=validated_model,
+            temperature=0.0,
+            max_tokens=BENCHMARK_MAX_TOKENS,
+        ),
+        event_agent=EventAgentParams(
+            model=validated_model,
+            temperature=0.0,
+            max_tokens=BENCHMARK_MAX_TOKENS,
+            enabled_tools=list(case.enabled_tools),
+            max_event_loops=BENCHMARK_MAX_EVENT_LOOPS,
+            max_parallel_events=BENCHMARK_MAX_PARALLEL_EVENTS,
+            batch_timeout_seconds=BENCHMARK_BATCH_TIMEOUT_SECONDS,
+        ),
+        tool_invocation_mode=mode,
+    )
+
+
+def build_mock_rounds(
+    case: BenchmarkCase,
+    mode: BenchmarkMode,
+) -> list[list[StreamItem]]:
+    if mode not in BENCHMARK_MODES:
+        raise ValueError(f"unsupported benchmark mode: {mode}")
+
+    first_round: list[StreamItem] = []
+    if case.mock_visible_messages:
+        first_round.append(StreamItem.message_delta(case.mock_visible_messages[0]))
+    for mock_event in case.mock_events:
+        arguments = dict(mock_event.arguments)
+        event = ToolCallEvent(
+            id=mock_event.id,
+            name=mock_event.name,
+            arguments=arguments,
+            raw_arguments=json.dumps(
+                arguments,
+                sort_keys=True,
+                separators=(",", ":"),
+            ),
+        )
+        event_item = (
+            StreamItem.text_event(event)
+            if mode == "dual_agent"
+            else StreamItem.provider_tool_call(event)
+        )
+        first_round.append(event_item)
+
+    rounds = [first_round]
+    rounds.extend(
+        [StreamItem.message_delta(message)]
+        for message in case.mock_visible_messages[1:]
+    )
+    return rounds

tests/test_benchmark_config.py

@@ -0,0 +1,366 @@
+import json
+
+import pytest
+from pydantic import ValidationError
+
+from agent_lab.application.benchmark import (
+    BENCHMARK_CASE_CATALOG,
+    BENCHMARK_CASE_IDS,
+    BENCHMARK_MODES,
+    BenchmarkConfig,
+    BenchmarkTarget,
+    build_benchmark_request,
+    build_mock_rounds,
+)
+
+
+VALID_CONFIG = {
+    "schema_version": 1,
+    "base_url": "https://provider.example/v1",
+    "model": "benchmark-model",
+    "runs_per_case": 1,
+    "modes": ["dual_agent", "chat_agent_tools"],
+    "cases": [
+        "ordinary_chat",
+        "device_volume_silent",
+        "calendar_schedule_template",
+        "web_search_two_answers",
+        "session_terminate",
+        "parallel_volume_schedule",
+    ],
+}
+
+
+def test_benchmark_config_accepts_the_versioned_single_target_schema():
+    config = BenchmarkConfig.model_validate_json(json.dumps(VALID_CONFIG))
+
+    assert config.schema_version == 1
+    assert config.base_url == "https://provider.example/v1"
+    assert config.model == "benchmark-model"
+    assert config.runs_per_case == 1
+    assert tuple(config.modes) == BENCHMARK_MODES
+    assert tuple(config.cases) == BENCHMARK_CASE_IDS
+
+
+def test_benchmark_target_uses_the_same_strict_url_and_model_validation():
+    target = BenchmarkTarget(
+        base_url="http://localhost:8000/v1",
+        model="local-model",
+    )
+
+    assert target.model_dump() == {
+        "base_url": "http://localhost:8000/v1",
+        "model": "local-model",
+    }
+
+
+@pytest.mark.parametrize("schema_version", [0, 2, "1", True, 1.0])
+def test_benchmark_config_rejects_unknown_or_coerced_schema_versions(
+    schema_version: object,
+):
+    payload = {**VALID_CONFIG, "schema_version": schema_version}
+
+    with pytest.raises(ValidationError, match="schema_version"):
+        BenchmarkConfig.model_validate(payload)
+
+
+@pytest.mark.parametrize(
+    "base_url",
+    [
+        "ftp://provider.example/v1",
+        "provider.example/v1",
+        "https:///v1",
+        "https://user:pass@provider.example/v1",
+        "https://provider.example/v1?debug=true",
+        "https://provider.example/v1#fragment",
+    ],
+)
+def test_benchmark_config_rejects_unsafe_or_non_http_base_urls(base_url: str):
+    payload = {**VALID_CONFIG, "base_url": base_url}
+
+    with pytest.raises(ValidationError, match="base_url"):
+        BenchmarkConfig.model_validate(payload)
+
+
+@pytest.mark.parametrize("model", ["", "   "])
+def test_benchmark_config_rejects_empty_models(model: str):
+    payload = {**VALID_CONFIG, "model": model}
+
+    with pytest.raises(ValidationError, match="model"):
+        BenchmarkConfig.model_validate(payload)
+
+
+def test_benchmark_config_rejects_non_positive_runs_per_case():
+    payload = {**VALID_CONFIG, "runs_per_case": 0}
+
+    with pytest.raises(ValidationError, match="runs_per_case"):
+        BenchmarkConfig.model_validate(payload)
+
+
+@pytest.mark.parametrize(
+    "modes",
+    [[], ["dual_agent", "dual_agent"], ["unsupported"]],
+)
+def test_benchmark_config_rejects_empty_duplicate_or_unknown_modes(
+    modes: list[str],
+):
+    payload = {**VALID_CONFIG, "modes": modes}
+
+    with pytest.raises(ValidationError, match="modes"):
+        BenchmarkConfig.model_validate(payload)
+
+
+@pytest.mark.parametrize(
+    "cases",
+    [[], ["ordinary_chat", "ordinary_chat"], ["custom_case"]],
+)
+def test_benchmark_config_rejects_empty_duplicate_or_unknown_cases(
+    cases: list[str],
+):
+    payload = {**VALID_CONFIG, "cases": cases}
+
+    with pytest.raises(ValidationError, match="cases"):
+        BenchmarkConfig.model_validate(payload)
+
+
+@pytest.mark.parametrize("field", ["api_key", "token", "unknown_field"])
+def test_benchmark_config_rejects_secrets_and_unknown_json_fields(field: str):
+    secret_value = "must-not-appear-in-validation-errors"
+    payload = {**VALID_CONFIG, field: secret_value}
+
+    with pytest.raises(ValidationError) as exc_info:
+        BenchmarkConfig.model_validate_json(json.dumps(payload))
+
+    assert field in str(exc_info.value)
+    assert secret_value not in str(exc_info.value)
+
+
+def test_builtin_case_catalog_is_complete_and_immutable():
+    assert tuple(BENCHMARK_CASE_CATALOG) == BENCHMARK_CASE_IDS
+
+    with pytest.raises(TypeError):
+        BENCHMARK_CASE_CATALOG["ordinary_chat"] = BENCHMARK_CASE_CATALOG[
+            "ordinary_chat"
+        ]
+
+    with pytest.raises(ValidationError, match="frozen"):
+        BENCHMARK_CASE_CATALOG["ordinary_chat"].user_message = "changed"
+
+    first_event = BENCHMARK_CASE_CATALOG["device_volume_silent"].mock_events[0]
+    with pytest.raises(TypeError):
+        first_event.arguments[0] = ("mode", "relative")
+
+
+def test_builtin_cases_own_semantics_and_deterministic_mock_data():
+    actual = {
+        case_id: {
+            "user_message": case.user_message,
+            "enabled_tools": case.enabled_tools,
+            "events": case.expectation.event_names,
+            "answer_count": case.expectation.answer_count,
+            "terminal": case.expectation.terminal,
+            "first_reply_before_tool": case.expectation.first_reply_before_tool,
+            "expected_visible": case.expectation.visible_messages,
+            "mock_visible": case.mock_visible_messages,
+            "mock_events": tuple(
+                (event.id, event.name, dict(event.arguments))
+                for event in case.mock_events
+            ),
+        }
+        for case_id, case in BENCHMARK_CASE_CATALOG.items()
+    }
+
+    assert actual == {
+        "ordinary_chat": {
+            "user_message": "Hello.",
+            "enabled_tools": (
+                "session.terminate",
+                "device.volume.adjust",
+                "calendar.schedule.create",
+                "knowledge.web.search",
+            ),
+            "events": (),
+            "answer_count": 1,
+            "terminal": False,
+            "first_reply_before_tool": False,
+            "expected_visible": ("Ordinary answer.",),
+            "mock_visible": ("Ordinary answer.",),
+            "mock_events": (),
+        },
+        "device_volume_silent": {
+            "user_message": "set volume to 30",
+            "enabled_tools": ("device.volume.adjust",),
+            "events": ("device.volume.adjust",),
+            "answer_count": 1,
+            "terminal": False,
+            "first_reply_before_tool": True,
+            "expected_visible": ("I will set the volume to 30.",),
+            "mock_visible": ("I will set the volume to 30.",),
+            "mock_events": (
+                (
+                    "volume-1",
+                    "device.volume.adjust",
+                    {"mode": "absolute", "value": 30},
+                ),
+            ),
+        },
+        "calendar_schedule_template": {
+            "user_message": (
+                'schedule "Design review" at 2026-07-14T09:30:00+08:00 '
+                "timezone Asia/Shanghai"
+            ),
+            "enabled_tools": ("calendar.schedule.create",),
+            "events": ("calendar.schedule.create",),
+            "answer_count": 2,
+            "terminal": False,
+            "first_reply_before_tool": True,
+            "expected_visible": (
+                "I will schedule that.",
+                "Scheduled Design review for 2026-07-14T09:30:00+08:00 "
+                "(Asia/Shanghai).",
+            ),
+            "mock_visible": ("I will schedule that.",),
+            "mock_events": (
+                (
+                    "schedule-1",
+                    "calendar.schedule.create",
+                    {
+                        "title": "Design review",
+                        "start_at": "2026-07-14T09:30:00+08:00",
+                        "timezone": "Asia/Shanghai",
+                    },
+                ),
+            ),
+        },
+        "web_search_two_answers": {
+            "user_message": "event batch safety",
+            "enabled_tools": ("knowledge.web.search",),
+            "events": ("knowledge.web.search",),
+            "answer_count": 2,
+            "terminal": False,
+            "first_reply_before_tool": True,
+            "expected_visible": ("I will check.", "Grounded search update."),
+            "mock_visible": ("I will check.", "Grounded search update."),
+            "mock_events": (
+                (
+                    "search-1",
+                    "knowledge.web.search",
+                    {"query": "event batch safety"},
+                ),
+            ),
+        },
+        "session_terminate": {
+            "user_message": "end this session",
+            "enabled_tools": ("session.terminate",),
+            "events": ("session.terminate",),
+            "answer_count": 1,
+            "terminal": True,
+            "first_reply_before_tool": True,
+            "expected_visible": ("Goodbye.",),
+            "mock_visible": ("Goodbye.",),
+            "mock_events": (("terminate-1", "session.terminate", {}),),
+        },
+        "parallel_volume_schedule": {
+            "user_message": (
+                'set volume to 30 and schedule "Design review" at '
+                "2026-07-14T09:30:00+08:00 timezone Asia/Shanghai"
+            ),
+            "enabled_tools": (
+                "device.volume.adjust",
+                "calendar.schedule.create",
+            ),
+            "events": (
+                "device.volume.adjust",
+                "calendar.schedule.create",
+            ),
+            "answer_count": 2,
+            "terminal": False,
+            "first_reply_before_tool": True,
+            "expected_visible": (
+                "I will set the volume to 30 and schedule that.",
+                "Scheduled Design review for 2026-07-14T09:30:00+08:00 "
+                "(Asia/Shanghai).",
+            ),
+            "mock_visible": (
+                "I will set the volume to 30 and schedule that.",
+            ),
+            "mock_events": (
+                (
+                    "parallel-volume",
+                    "device.volume.adjust",
+                    {"mode": "absolute", "value": 30},
+                ),
+                (
+                    "parallel-schedule",
+                    "calendar.schedule.create",
+                    {
+                        "title": "Design review",
+                        "start_at": "2026-07-14T09:30:00+08:00",
+                        "timezone": "Asia/Shanghai",
+                    },
+                ),
+            ),
+        },
+    }
+
+
+@pytest.mark.parametrize("case_id", BENCHMARK_CASE_IDS)
+def test_build_benchmark_request_uses_fixed_fairness_controls(case_id: str):
+    case = BENCHMARK_CASE_CATALOG[case_id]
+    requests = {
+        mode: build_benchmark_request(case, mode, "comparison-model")
+        for mode in BENCHMARK_MODES
+    }
+
+    dual = requests["dual_agent"]
+    tools = requests["chat_agent_tools"]
+    assert dual.model_dump(exclude={"tool_invocation_mode"}) == tools.model_dump(
+        exclude={"tool_invocation_mode"}
+    )
+    assert dual.user_message == case.user_message
+    assert dual.system_prompts == ["Exercise the requested business scenario."]
+    assert dual.pre_messages == []
+    assert dual.chat_agent.model == "comparison-model"
+    assert dual.chat_agent.temperature == 0.0
+    assert dual.chat_agent.max_tokens == 128
+    assert dual.event_agent.model == "comparison-model"
+    assert dual.event_agent.temperature == 0.0
+    assert dual.event_agent.max_tokens == 128
+    assert tuple(dual.event_agent.enabled_tools) == case.enabled_tools
+    assert dual.event_agent.max_event_loops == 1
+    assert dual.event_agent.max_parallel_events == 2
+    assert dual.event_agent.batch_timeout_seconds == 1.0
+    assert dual.tool_invocation_mode == "dual_agent"
+    assert tools.tool_invocation_mode == "chat_agent_tools"
+
+
+def test_build_benchmark_request_rejects_an_empty_model():
+    case = BENCHMARK_CASE_CATALOG["ordinary_chat"]
+
+    with pytest.raises(ValueError, match="model"):
+        build_benchmark_request(case, "dual_agent", "   ")
+
+
+@pytest.mark.parametrize("mode", BENCHMARK_MODES)
+@pytest.mark.parametrize("case_id", BENCHMARK_CASE_IDS)
+def test_mock_rounds_are_mode_correct_and_derived_from_catalog(
+    mode: str,
+    case_id: str,
+):
+    case = BENCHMARK_CASE_CATALOG[case_id]
+    rounds = build_mock_rounds(case, mode)
+    items = tuple(item for round_items in rounds for item in round_items)
+    message_items = tuple(item for item in items if item.kind == "message_delta")
+    event_items = tuple(item for item in items if item.event is not None)
+
+    assert tuple(item.content for item in message_items) == case.mock_visible_messages
+    assert tuple(item.event.name for item in event_items) == case.expectation.event_names
+    assert tuple(item.event.id for item in event_items) == tuple(
+        event.id for event in case.mock_events
+    )
+    assert tuple(item.event.arguments for item in event_items) == tuple(
+        dict(event.arguments) for event in case.mock_events
+    )
+    expected_kind = "text_event" if mode == "dual_agent" else "provider_tool_call"
+    assert {item.kind for item in event_items} <= {expected_kind}
+    assert len(rounds) == max(1, len(case.mock_visible_messages))

tests/test_tool_invocation_comparison.py

@@ -6,47 +6,19 @@ from typing import Any
 
 import pytest
 
-from agent_lab.application.contracts import AgentParams, DebugRunRequest, EventAgentParams
+from agent_lab.application.benchmark import (
+    BENCHMARK_CASE_CATALOG,
+    BENCHMARK_CASE_IDS,
+    BENCHMARK_MODES,
+    build_benchmark_request,
+    build_mock_rounds,
+)
+from agent_lab.application.contracts import AgentParams
 from agent_lab.application.runtime import DebugRuntime
 from agent_lab.application.tools import build_default_tool_registry
-from agent_lab.domain.events import ToolCallEvent
 from agent_lab.domain.messages import ChatMessage, StreamItem
 
 
-MODES = ("dual_agent", "chat_agent_tools")
-SCENARIOS = (
-    "ordinary_chat",
-    "device_volume_silent",
-    "calendar_schedule_template",
-    "web_search_two_answers",
-    "session_terminate",
-    "parallel_volume_schedule",
-)
-BUILTIN_EVENT_NAMES = (
-    "session.terminate",
-    "device.volume.adjust",
-    "calendar.schedule.create",
-    "knowledge.web.search",
-)
-SCHEDULE_ARGUMENTS = {
-    "title": "Design review",
-    "start_at": "2026-07-14T09:30:00+08:00",
-    "timezone": "Asia/Shanghai",
-}
-SCHEDULE_CONFIRMATION = (
-    "Scheduled Design review for 2026-07-14T09:30:00+08:00 "
-    "(Asia/Shanghai)."
-)
-
-
-@dataclass(frozen=True)
-class ExpectedSemantics:
-    visible_messages: tuple[str, ...]
-    tool_names: tuple[str, ...]
-    model_calls: int
-    terminal_count: int = 0
-
-
 @dataclass(frozen=True)
 class RecordedChatRequest:
     messages: tuple[ChatMessage, ...]
@@ -54,40 +26,6 @@ class RecordedChatRequest:
     tool_choice: dict[str, Any] | None
 
 
-EXPECTED = {
-    "ordinary_chat": ExpectedSemantics(("Ordinary answer.",), (), 1),
-    "device_volume_silent": ExpectedSemantics(
-        ("I will set the volume to 30.",),
-        ("device.volume.adjust",),
-        1,
-    ),
-    "calendar_schedule_template": ExpectedSemantics(
-        ("I will schedule that.", SCHEDULE_CONFIRMATION),
-        ("calendar.schedule.create",),
-        1,
-    ),
-    "web_search_two_answers": ExpectedSemantics(
-        ("I will check.", "Grounded search update."),
-        ("knowledge.web.search",),
-        2,
-    ),
-    "session_terminate": ExpectedSemantics(
-        ("Goodbye.",),
-        ("session.terminate",),
-        1,
-        terminal_count=1,
-    ),
-    "parallel_volume_schedule": ExpectedSemantics(
-        (
-            "I will set the volume to 30 and schedule that.",
-            SCHEDULE_CONFIRMATION,
-        ),
-        ("device.volume.adjust", "calendar.schedule.create"),
-        1,
-    ),
-}
-
-
 class ScriptedChatClient:
     def __init__(self, rounds: list[list[StreamItem]]) -> None:
         self.rounds = rounds
@@ -236,148 +174,6 @@ class RecordingPorts:
         await release.wait()
 
 
-def _event(call_id: str, name: str, arguments: dict[str, Any]) -> ToolCallEvent:
-    return ToolCallEvent(
-        id=call_id,
-        name=name,
-        arguments=arguments,
-        raw_arguments=json.dumps(arguments, sort_keys=True, separators=(",", ":")),
-    )
-
-
-def _event_item(mode: str, event: ToolCallEvent) -> StreamItem:
-    if mode == "dual_agent":
-        return StreamItem.text_event(event)
-    return StreamItem.provider_tool_call(event)
-
-
-def _scenario_setup(
-    mode: str,
-    scenario: str,
-) -> tuple[DebugRunRequest, list[list[StreamItem]]]:
-    if scenario == "ordinary_chat":
-        user_message = "Hello."
-        enabled_tools = list(BUILTIN_EVENT_NAMES)
-        rounds = [[StreamItem.message_delta("Ordinary answer.")]]
-    elif scenario == "device_volume_silent":
-        user_message = "set volume to 30"
-        enabled_tools = ["device.volume.adjust"]
-        rounds = [
-            [
-                StreamItem.message_delta("I will set the volume to 30."),
-                _event_item(
-                    mode,
-                    _event(
-                        "volume-1",
-                        "device.volume.adjust",
-                        {"mode": "absolute", "value": 30},
-                    ),
-                )
-            ]
-        ]
-    elif scenario == "calendar_schedule_template":
-        user_message = (
-            'schedule "Design review" at 2026-07-14T09:30:00+08:00 '
-            "timezone Asia/Shanghai"
-        )
-        enabled_tools = ["calendar.schedule.create"]
-        rounds = [
-            [
-                StreamItem.message_delta("I will schedule that."),
-                _event_item(
-                    mode,
-                    _event(
-                        "schedule-1",
-                        "calendar.schedule.create",
-                        SCHEDULE_ARGUMENTS,
-                    ),
-                )
-            ]
-        ]
-    elif scenario == "web_search_two_answers":
-        user_message = "event batch safety"
-        enabled_tools = ["knowledge.web.search"]
-        rounds = [
-            [
-                StreamItem.message_delta("I will check."),
-                _event_item(
-                    mode,
-                    _event(
-                        "search-1",
-                        "knowledge.web.search",
-                        {"query": user_message},
-                    ),
-                ),
-            ],
-            [StreamItem.message_delta("Grounded search update.")],
-        ]
-    elif scenario == "session_terminate":
-        user_message = "end this session"
-        enabled_tools = ["session.terminate"]
-        rounds = [
-            [
-                StreamItem.message_delta("Goodbye."),
-                _event_item(
-                    mode,
-                    _event("terminate-1", "session.terminate", {}),
-                )
-            ]
-        ]
-    elif scenario == "parallel_volume_schedule":
-        user_message = (
-            'set volume to 30 and schedule "Design review" at '
-            "2026-07-14T09:30:00+08:00 timezone Asia/Shanghai"
-        )
-        enabled_tools = ["device.volume.adjust", "calendar.schedule.create"]
-        rounds = [
-            [
-                StreamItem.message_delta(
-                    "I will set the volume to 30 and schedule that."
-                ),
-                _event_item(
-                    mode,
-                    _event(
-                        "parallel-volume",
-                        "device.volume.adjust",
-                        {"mode": "absolute", "value": 30},
-                    ),
-                ),
-                _event_item(
-                    mode,
-                    _event(
-                        "parallel-schedule",
-                        "calendar.schedule.create",
-                        SCHEDULE_ARGUMENTS,
-                    ),
-                ),
-            ]
-        ]
-    else:
-        raise AssertionError(f"unknown comparison scenario: {scenario}")
-
-    request = DebugRunRequest(
-        user_message=user_message,
-        system_prompts=["Exercise the requested business scenario."],
-        pre_messages=[],
-        chat_agent=AgentParams(
-            model="comparison-model",
-            temperature=0.0,
-            max_tokens=128,
-        ),
-        event_agent=EventAgentParams(
-            model="comparison-model",
-            temperature=0.0,
-            max_tokens=128,
-            enabled_tools=enabled_tools,
-            max_event_loops=1,
-            max_parallel_events=2,
-            batch_timeout_seconds=1.0,
-        ),
-        tool_invocation_mode=mode,
-    )
-    return request, rounds
-
-
 async def _collect_outputs(
     stream: AsyncIterator[dict[str, Any]],
 ) -> list[dict[str, Any]]:
@@ -463,13 +259,15 @@ def _assert_port_calls(scenario: str, ports: RecordingPorts) -> None:
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("mode", MODES)
-@pytest.mark.parametrize("scenario", SCENARIOS)
+@pytest.mark.parametrize("mode", BENCHMARK_MODES)
+@pytest.mark.parametrize("scenario", BENCHMARK_CASE_IDS)
 async def test_invocation_modes_share_business_semantics(
     mode: str,
     scenario: str,
 ) -> None:
-    request, rounds = _scenario_setup(mode, scenario)
+    case = BENCHMARK_CASE_CATALOG[scenario]
+    request = build_benchmark_request(case, mode, "comparison-model")
+    rounds = build_mock_rounds(case, mode)
     ports = RecordingPorts(parallel_barrier=scenario == "parallel_volume_schedule")
     registry = build_default_tool_registry(
         session_termination_port=ports,
@@ -498,7 +296,6 @@ async def test_invocation_modes_share_business_semantics(
             run_task.cancel()
             await asyncio.gather(run_task, return_exceptions=True)
 
-    expected = EXPECTED[scenario]
     visible_messages = tuple(
         message["content"]
         for message in outputs
@@ -509,11 +306,13 @@ async def test_invocation_modes_share_business_semantics(
         message.get("event") == "terminal_completed" for message in outputs
     )
 
-    assert visible_messages == expected.visible_messages
-    assert tuple(payload["tool"] for payload in tool_payloads) == expected.tool_names
-    assert client.calls == expected.model_calls
-    assert len(client.requests) == expected.model_calls
-    assert terminal_count == expected.terminal_count
+    assert visible_messages == case.expectation.visible_messages
+    assert tuple(payload["tool"] for payload in tool_payloads) == (
+        case.expectation.event_names
+    )
+    assert client.calls == len(rounds)
+    assert len(client.requests) == len(rounds)
+    assert terminal_count == int(case.expectation.terminal)
     assert outputs[-1] == {"type": "done"}
     _assert_port_calls(scenario, ports)
 
@@ -527,7 +326,7 @@ async def test_invocation_modes_share_business_semantics(
         assert _provider_tool_names(initial_request.tools) == enabled_names
         assert _generated_event_catalog_names(initial_request.messages) == ()
 
-    if expected.tool_names:
+    if case.expectation.event_names:
         business = [message for message in outputs if message["type"] != "audit"]
         first_message_index = next(
             index
@@ -550,7 +349,7 @@ async def test_invocation_modes_share_business_semantics(
         )
         assert tuple(
             message["details"]["event_source"] for message in detected_events
-        ) == (expected_source,) * len(expected.tool_names)
+        ) == (expected_source,) * len(case.expectation.event_names)
 
     if scenario == "web_search_two_answers":
         assert tool_payloads[0]["sources"][0]["title"] == "Matrix source"