agent-lab/tests/test_builtin_event_plugins.py

import json
from datetime import datetime, timezone

import pytest

import agent_lab.application.events.builtin_plugins as builtin_plugins
from agent_lab.application.events import (
    EventArgumentResolution,
    EventExecutionContext,
    EventKernel,
    EventRequest,
    EventStatus,
    ResolvedEventArguments,
    ResultPolicy,
    RiskLevel,
)
from agent_lab.application.tools import build_default_tool_registry
from agent_lab.domain.events import ToolCallEvent
from agent_lab.domain.messages import ChatMessage


BUILTIN_EVENT_NAMES = [
    "session.terminate",
    "device.volume.adjust",
    "calendar.schedule.create",
    "knowledge.web.search",
]


def _event(event_id: str, name: str, arguments: dict):
    return ToolCallEvent(
        id=event_id,
        name=name,
        arguments=arguments,
        raw_arguments=json.dumps(arguments, ensure_ascii=False),
    )


def test_default_registry_preserves_legacy_tools_and_adds_builtin_event_plugins():
    registry = build_default_tool_registry()

    assert [item["name"] for item in registry.available_tools()] == [
        "handoff_note",
        "mock_search",
        "mock_ticket",
        *BUILTIN_EVENT_NAMES,
    ]


@pytest.mark.parametrize(
    (
        "name",
        "result_policy",
        "risk_level",
        "fallback_allowed",
        "idempotency_key_fields",
        "concurrency_class",
        "conflict_keys",
        "timeout_seconds",
        "terminal",
    ),
    [
        (
            "session.terminate",
            ResultPolicy.TERMINATE,
            RiskLevel.HIGH,
            False,
            ("event_id",),
            "session-lifecycle",
            ("session",),
            5.0,
            True,
        ),
        (
            "device.volume.adjust",
            ResultPolicy.SILENT_SUCCESS,
            RiskLevel.MEDIUM,
            True,
            ("event_id",),
            "device-volume",
            ("device.volume",),
            5.0,
            False,
        ),
        (
            "calendar.schedule.create",
            ResultPolicy.TEMPLATE_FOLLOW_UP,
            RiskLevel.MEDIUM,
            True,
            ("event_id",),
            "schedule-write",
            ("calendar.schedule",),
            10.0,
            False,
        ),
        (
            "knowledge.web.search",
            ResultPolicy.LLM_FOLLOW_UP,
            RiskLevel.LOW,
            False,
            (),
            "read-only",
            (),
            10.0,
            False,
        ),
    ],
)
def test_builtin_definition_metadata_is_owned_by_each_flat_plugin(
    name,
    result_policy,
    risk_level,
    fallback_allowed,
    idempotency_key_fields,
    concurrency_class,
    conflict_keys,
    timeout_seconds,
    terminal,
):
    definition = build_default_tool_registry().event_registry.definition(name)

    assert definition is not None
    assert definition.result_policy is result_policy
    assert definition.risk_level is risk_level
    assert definition.fallback_allowed is fallback_allowed
    assert definition.idempotency_key_fields == idempotency_key_fields
    assert definition.concurrency_class == concurrency_class
    assert definition.conflict_keys == conflict_keys
    assert definition.timeout_seconds == timeout_seconds
    assert definition.terminal is terminal


def test_session_terminate_schema_accepts_optional_reason_only():
    registry = build_default_tool_registry().event_registry

    assert registry.iter_validation_errors("session.terminate", {}) == ()
    assert registry.iter_validation_errors(
        "session.terminate", {"reason": "user requested"}
    ) == ()
    assert registry.iter_validation_errors(
        "session.terminate", {"reason": 1}
    )
    assert registry.iter_validation_errors(
        "session.terminate", {"unexpected": True}
    )


@pytest.mark.parametrize(
    "arguments",
    [
        {"mode": "absolute", "value": 0},
        {"mode": "absolute", "value": 100},
        {"mode": "relative", "delta": -100},
        {"mode": "relative", "delta": 100},
        {"mode": "mute"},
        {"mode": "unmute"},
    ],
)
def test_volume_schema_accepts_mode_specific_valid_arguments(arguments):
    registry = build_default_tool_registry().event_registry

    assert registry.iter_validation_errors("device.volume.adjust", arguments) == ()


@pytest.mark.parametrize(
    "arguments",
    [
        {"mode": "absolute"},
        {"mode": "absolute", "value": -1},
        {"mode": "absolute", "value": 101},
        {"mode": "absolute", "value": 20, "delta": 5},
        {"mode": "relative"},
        {"mode": "relative", "delta": 0},
        {"mode": "relative", "delta": 101},
        {"mode": "relative", "delta": 5, "value": 20},
        {"mode": "mute", "value": 0},
        {"mode": "unmute", "delta": 1},
    ],
)
def test_volume_schema_rejects_invalid_or_cross_mode_arguments(arguments):
    registry = build_default_tool_registry().event_registry

    assert registry.iter_validation_errors("device.volume.adjust", arguments)


@pytest.mark.parametrize(
    ("content", "expected"),
    [
        ("把音量调到 35", {"mode": "absolute", "value": 35}),
        ("set volume to 42", {"mode": "absolute", "value": 42}),
        ("音量增加 15", {"mode": "relative", "delta": 15}),
        ("decrease volume by 20", {"mode": "relative", "delta": -20}),
        ("请静音", {"mode": "mute"}),
        ("please unmute", {"mode": "unmute"}),
    ],
)
def test_volume_resolver_recognizes_clear_chinese_and_english(content, expected):
    resolved = _resolve_text_event("device.volume.adjust", content)

    assert resolved == expected


@pytest.mark.parametrize("content", ["把音量调高", "静音还是取消静音"])
def test_volume_resolver_marks_incomplete_or_ambiguous_requests_for_fallback(content):
    resolved = _resolve_text_event("device.volume.adjust", content)

    assert isinstance(resolved, EventArgumentResolution)
    assert resolved.complete is False


def test_schedule_schema_requires_explicit_rfc3339_and_timezone():
    registry = build_default_tool_registry().event_registry
    valid = {
        "title": "Design review",
        "start_at": "2026-07-14T09:30:00+08:00",
        "timezone": "Asia/Shanghai",
        "recurrence": "FREQ=WEEKLY",
        "reminder_minutes": 15,
    }

    assert registry.iter_validation_errors("calendar.schedule.create", valid) == ()
    assert registry.iter_validation_errors(
        "calendar.schedule.create",
        {**valid, "start_at": "tomorrow at nine"},
    )
    assert registry.iter_validation_errors(
        "calendar.schedule.create",
        {**valid, "timezone": "Shanghai"},
    )
    assert registry.iter_validation_errors(
        "calendar.schedule.create",
        {**valid, "reminder_minutes": -1},
    )


@pytest.mark.parametrize(
    ("content", "expected"),
    [
        (
            'Schedule "Design review" at 2026-07-14T09:30:00+08:00 '
            "timezone Asia/Shanghai",
            {
                "title": "Design review",
                "start_at": "2026-07-14T09:30:00+08:00",
                "timezone": "Asia/Shanghai",
            },
        ),
        (
            "标题：项目评审；开始时间：2026-07-14T09:30:00+08:00；"
            "时区：Asia/Shanghai",
            {
                "title": "项目评审",
                "start_at": "2026-07-14T09:30:00+08:00",
                "timezone": "Asia/Shanghai",
            },
        ),
    ],
)
def test_schedule_resolver_accepts_only_explicit_datetime_details(content, expected):
    resolved = _resolve_text_event("calendar.schedule.create", content)

    assert resolved == expected


def test_schedule_resolver_does_not_guess_relative_time():
    resolved = _resolve_text_event(
        "calendar.schedule.create", "Schedule standup tomorrow at nine"
    )

    assert isinstance(resolved, EventArgumentResolution)
    assert resolved.complete is False


def test_web_search_resolver_uses_latest_relevant_user_request():
    registry = build_default_tool_registry()
    definition = registry.event_registry.definition("knowledge.web.search")
    assert definition is not None
    assert definition.resolver is not None

    resolved = definition.resolver(
        EventRequest(id="search-1", name="knowledge.web.search"),
        EventExecutionContext(
            history=(
                ChatMessage(role="user", content="old query"),
                ChatMessage(role="assistant", content="assistant summary"),
                ChatMessage(role="user", content="latest focused query"),
                ChatMessage(role="assistant", content="I will search"),
            )
        ),
    )

    assert resolved == {"query": "latest focused query"}


def test_web_search_schema_rejects_empty_query_and_out_of_range_result_limit():
    registry = build_default_tool_registry().event_registry

    assert registry.iter_validation_errors(
        "knowledge.web.search", {"query": "focused", "max_results": 1}
    ) == ()
    assert registry.iter_validation_errors(
        "knowledge.web.search", {"query": "focused", "max_results": 5}
    ) == ()
    assert registry.iter_validation_errors(
        "knowledge.web.search", {"query": ""}
    )
    assert registry.iter_validation_errors(
        "knowledge.web.search", {"query": "focused", "max_results": 0}
    )
    assert registry.iter_validation_errors(
        "knowledge.web.search", {"query": "focused", "max_results": 6}
    )


def test_catalog_and_direct_provider_modes_share_the_same_builtin_schemas():
    registry = build_default_tool_registry()
    catalog = {item["name"]: item for item in registry.available_tools()}
    direct = {
        item["function"]["name"]: item["function"]
        for item in registry.provider_tool_schemas(BUILTIN_EVENT_NAMES)
    }

    for name in BUILTIN_EVENT_NAMES:
        assert direct[name]["description"] == catalog[name]["description"]
        assert direct[name]["parameters"] == catalog[name]["parameters"]


class RecordingSessionPort:
    def __init__(self):
        self.calls = []

    def terminate(self, event_id, *, reason=None):
        self.calls.append((event_id, reason))
        return {"port": "session", "event_id": event_id, "reason": reason}


class RecordingVolumePort:
    def __init__(self):
        self.calls = []

    def adjust(self, event_id, *, mode, value=None, delta=None):
        self.calls.append((event_id, mode, value, delta))
        return {
            "port": "volume",
            "event_id": event_id,
            "mode": mode,
            "value": value,
            "delta": delta,
        }


class RecordingCalendarPort:
    def __init__(self):
        self.calls = []

    def create(
        self,
        event_id,
        *,
        title,
        start_at,
        timezone,
        recurrence=None,
        reminder_minutes=None,
    ):
        self.calls.append(
            (
                event_id,
                title,
                start_at,
                timezone,
                recurrence,
                reminder_minutes,
            )
        )
        return {"port": "calendar", "event_id": event_id, "title": title}


class RecordingSearchPort:
    def __init__(self):
        self.calls = []

    def search(self, event_id, *, query, max_results=3):
        self.calls.append((event_id, query, max_results))
        return {"port": "search", "query": query, "max_results": max_results}


def test_injected_sync_ports_receive_validated_arguments_and_event_ids():
    session = RecordingSessionPort()
    volume = RecordingVolumePort()
    calendar = RecordingCalendarPort()
    search = RecordingSearchPort()
    registry = build_default_tool_registry(
        session_termination_port=session,
        device_volume_port=volume,
        calendar_schedule_port=calendar,
        web_search_port=search,
    )

    assert registry.execute(
        _event("session-1", "session.terminate", {"reason": "done"})
    ) == {"port": "session", "event_id": "session-1", "reason": "done"}
    assert registry.execute(
        _event("volume-1", "device.volume.adjust", {"mode": "absolute", "value": 30})
    )["port"] == "volume"
    assert registry.execute(
        _event(
            "calendar-1",
            "calendar.schedule.create",
            {
                "title": "Review",
                "start_at": "2026-07-14T09:30:00+08:00",
                "timezone": "Asia/Shanghai",
                "reminder_minutes": 10,
            },
        )
    )["port"] == "calendar"
    assert registry.execute(
        _event(
            "search-1",
            "knowledge.web.search",
            {"query": "agent kernels", "max_results": 2},
        )
    ) == {"port": "search", "query": "agent kernels", "max_results": 2}
    assert session.calls == [("session-1", "done")]
    assert volume.calls == [("volume-1", "absolute", 30, None)]
    assert calendar.calls == [
        (
            "calendar-1",
            "Review",
            "2026-07-14T09:30:00+08:00",
            "Asia/Shanghai",
            None,
            10,
        )
    ]
    assert search.calls == [("search-1", "agent kernels", 2)]


@pytest.mark.asyncio
async def test_injected_async_ports_are_awaited_for_all_builtin_plugins():
    class AsyncSessionPort(RecordingSessionPort):
        async def terminate(self, event_id, *, reason=None):
            return super().terminate(event_id, reason=reason)

    class AsyncVolumePort(RecordingVolumePort):
        async def adjust(self, event_id, *, mode, value=None, delta=None):
            return super().adjust(
                event_id, mode=mode, value=value, delta=delta
            )

    class AsyncCalendarPort(RecordingCalendarPort):
        async def create(self, event_id, **arguments):
            return super().create(event_id, **arguments)

    class AsyncSearchPort(RecordingSearchPort):
        async def search(self, event_id, *, query, max_results=3):
            return super().search(
                event_id, query=query, max_results=max_results
            )

    registry = build_default_tool_registry(
        session_termination_port=AsyncSessionPort(),
        device_volume_port=AsyncVolumePort(),
        calendar_schedule_port=AsyncCalendarPort(),
        web_search_port=AsyncSearchPort(),
    )
    events = [
        _event("session-1", "session.terminate", {}),
        _event(
            "volume-1",
            "device.volume.adjust",
            {"mode": "relative", "delta": -10},
        ),
        _event(
            "calendar-1",
            "calendar.schedule.create",
            {
                "title": "Review",
                "start_at": "2026-07-14T09:30:00+08:00",
                "timezone": "Asia/Shanghai",
            },
        ),
        _event("search-1", "knowledge.web.search", {"query": "agent kernels"}),
    ]

    payloads = [await registry.execute_async(event) for event in events]

    assert [payload["port"] for payload in payloads] == [
        "session",
        "volume",
        "calendar",
        "search",
    ]


@pytest.mark.parametrize(
    "arguments",
    [
        {"mode": "absolute", "value": 101},
        {"mode": "relative", "delta": 0},
        {"mode": "mute", "value": 1},
    ],
)
def test_invalid_volume_arguments_never_call_the_port(arguments):
    volume = RecordingVolumePort()
    registry = build_default_tool_registry(device_volume_port=volume)

    payload = registry.execute(_event("volume-invalid", "device.volume.adjust", arguments))

    assert payload["tool"] == "device.volume.adjust"
    assert "invalid" in payload["error"]
    assert volume.calls == []


def test_ambiguous_volume_request_without_fallback_never_calls_the_port():
    volume = RecordingVolumePort()
    registry = build_default_tool_registry(device_volume_port=volume)

    payload = registry.handle(
        _event("volume-ambiguous", "device.volume.adjust", {}),
        EventExecutionContext(
            history=(ChatMessage(role="user", content="静音还是取消静音"),)
        ),
    )

    assert payload == {
        "tool": "device.volume.adjust",
        "error": "missing required arguments: mode",
    }
    assert volume.calls == []


@pytest.mark.parametrize(
    ("port_name", "event"),
    [
        ("session_termination_port", _event("s", "session.terminate", {})),
        (
            "device_volume_port",
            _event("v", "device.volume.adjust", {"mode": "mute"}),
        ),
        (
            "calendar_schedule_port",
            _event(
                "c",
                "calendar.schedule.create",
                {
                    "title": "Review",
                    "start_at": "2026-07-14T09:30:00+08:00",
                    "timezone": "Asia/Shanghai",
                },
            ),
        ),
        (
            "web_search_port",
            _event("w", "knowledge.web.search", {"query": "agent kernels"}),
        ),
    ],
)
def test_port_failures_are_normalized_by_the_generic_kernel(port_name, event):
    class FailingPort:
        def terminate(self, *args, **kwargs):
            raise RuntimeError("port unavailable")

        adjust = terminate
        create = terminate
        search = terminate

    registry = build_default_tool_registry(**{port_name: FailingPort()})

    payload = registry.execute(event)

    assert payload == {
        "tool": event.name,
        "error": "tool handler failed: port unavailable",
    }


def test_default_stateful_adapters_cache_by_event_id_and_canonical_arguments():
    registry = build_default_tool_registry()

    first_session = registry.execute(
        _event("same-session", "session.terminate", {"reason": "first"})
    )
    replayed_session = registry.execute(
        _event("same-session", "session.terminate", {"reason": "first"})
    )
    changed_session = registry.execute(
        _event("same-session", "session.terminate", {"reason": "second"})
    )
    first_volume = registry.execute(
        _event(
            "same-volume",
            "device.volume.adjust",
            {"mode": "  ReLaTiVe  ", "delta": 20},
        )
    )
    replayed_volume = registry.execute(
        _event(
            "same-volume",
            "device.volume.adjust",
            {"mode": "relative", "delta": 20},
        )
    )
    changed_volume = registry.execute(
        _event(
            "same-volume",
            "device.volume.adjust",
            {"mode": "relative", "delta": -10},
        )
    )
    first_schedule = registry.execute(
        _event(
            "same-schedule",
            "calendar.schedule.create",
            {
                "title": "First",
                "start_at": "2026-07-14T09:30:00+08:00",
                "timezone": "Asia/Shanghai",
            },
        )
    )
    replayed_schedule = registry.execute(
        _event(
            "same-schedule",
            "calendar.schedule.create",
            {
                "title": "First",
                "start_at": "2026-07-14T09:30:00+08:00",
                "timezone": "Asia/Shanghai",
            },
        )
    )
    changed_schedule = registry.execute(
        _event(
            "same-schedule",
            "calendar.schedule.create",
            {
                "title": "Second",
                "start_at": "2026-07-15T09:30:00+08:00",
                "timezone": "Asia/Shanghai",
            },
        )
    )

    assert replayed_session == first_session
    assert first_session["reason"] == "first"
    assert changed_session["reason"] == "second"
    assert replayed_volume == first_volume
    assert first_volume["delta"] == 20
    assert changed_volume["delta"] == -10
    assert replayed_schedule == first_schedule
    assert first_schedule["schedule"]["title"] == "First"
    assert changed_schedule["schedule"]["title"] == "Second"


def test_default_web_search_is_deterministic_compact_and_uses_injected_clock():
    registry = build_default_tool_registry(
        clock=lambda: datetime(2030, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
    )

    payload = registry.execute(
        _event(
            "search-1",
            "knowledge.web.search",
            {"query": "agent kernels", "max_results": 2},
        )
    )

    assert payload["tool"] == "knowledge.web.search"
    assert payload["query"] == "agent kernels"
    assert payload["retrieved_at"] == "2030-01-02T03:04:05Z"
    assert len(payload["sources"]) == 2
    assert all(source["url"].startswith("https://example.invalid/") for source in payload["sources"])
    assert all(set(source) == {"title", "url", "snippet"} for source in payload["sources"])


def test_builtin_port_protocols_and_default_adapter_types_are_public():
    assert builtin_plugins.SessionTerminationPort
    assert builtin_plugins.DeviceVolumePort
    assert builtin_plugins.CalendarSchedulePort
    assert builtin_plugins.WebSearchPort
    assert builtin_plugins.InMemorySessionTerminationAdapter
    assert builtin_plugins.InMemoryDeviceVolumeAdapter
    assert builtin_plugins.InMemoryCalendarScheduleAdapter
    assert builtin_plugins.InMemoryWebSearchAdapter


def _resolve_text_event(name: str, content: str):
    registry = build_default_tool_registry()
    definition = registry.event_registry.definition(name)
    assert definition is not None
    assert definition.resolver is not None
    return definition.resolver(
        EventRequest(id="event-1", name=name),
        EventExecutionContext(history=(ChatMessage(role="user", content=content),)),
    )


def test_volume_structured_arguments_are_canonical_in_text_and_provider_modes():
    volume = RecordingVolumePort()
    registry = build_default_tool_registry(device_volume_port=volume)
    arguments = {"mode": "  AbSoLuTe  ", "value": 40}

    text_payload = registry.handle(
        _event("text-volume", "device.volume.adjust", arguments)
    )
    provider_payload = registry.execute(
        _event("provider-volume", "device.volume.adjust", arguments)
    )

    assert text_payload["port"] == "volume"
    assert provider_payload["port"] == "volume"
    assert volume.calls == [
        ("text-volume", "absolute", 40, None),
        ("provider-volume", "absolute", 40, None),
    ]


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "content",
    [
        "不静音",
        "不用静音",
        "无需静音",
        "no mute",
        "don't mute",
        "do not mute",
    ],
)
async def test_negated_volume_requests_are_substantive_invalid_without_fallback(
    content,
):
    volume = RecordingVolumePort()
    registry = build_default_tool_registry(device_volume_port=volume)
    fallback_calls = 0

    async def fallback(*args):
        nonlocal fallback_calls
        fallback_calls += 1
        return ResolvedEventArguments(
            event_name="device.volume.adjust",
            arguments={"mode": "mute"},
            raw_arguments='{"mode":"mute"}',
        )

    result = await EventKernel(
        registry.event_registry,
        argument_fallback=fallback,
    ).execute(
        registry.event_request(_event("negated-volume", "device.volume.adjust", {})),
        enabled_names=["device.volume.adjust"],
        context=EventExecutionContext(
            history=(ChatMessage(role="user", content=content),)
        ),
    )

    assert result.status is EventStatus.INVALID_ARGUMENTS
    assert result.arguments == {"mode": "unsupported"}
    assert result.used_fallback is False
    assert fallback_calls == 0
    assert volume.calls == []


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("content", "expected_arguments"),
    [
        ("set volume to 40.5", {"mode": "absolute", "value": "40.5"}),
        ("increase volume by 10.5", {"mode": "relative", "delta": "10.5"}),
        ("set volume to 1,000", {"mode": "absolute", "value": "1,000"}),
        ("音量增加 1,000", {"mode": "relative", "delta": "1,000"}),
    ],
)
async def test_unsupported_volume_numeric_tokens_are_not_truncated_or_fallback(
    content,
    expected_arguments,
):
    volume = RecordingVolumePort()
    registry = build_default_tool_registry(device_volume_port=volume)
    fallback_calls = 0

    async def fallback(*args):
        nonlocal fallback_calls
        fallback_calls += 1
        raise AssertionError("unsupported numeric tokens must not fallback")

    result = await EventKernel(
        registry.event_registry,
        argument_fallback=fallback,
    ).execute(
        registry.event_request(_event("invalid-number", "device.volume.adjust", {})),
        enabled_names=["device.volume.adjust"],
        context=EventExecutionContext(
            history=(ChatMessage(role="user", content=content),)
        ),
    )

    assert result.status is EventStatus.INVALID_ARGUMENTS
    assert result.arguments == expected_arguments
    assert result.used_fallback is False
    assert fallback_calls == 0
    assert volume.calls == []


@pytest.mark.parametrize("content", ["set volume to 1000", "音量增加 1000"])
def test_volume_resolver_captures_full_oversized_number_and_rejects_it(content):
    volume = RecordingVolumePort()
    registry = build_default_tool_registry(device_volume_port=volume)

    payload = registry.handle(
        _event("oversized-volume", "device.volume.adjust", {}),
        EventExecutionContext(history=(ChatMessage(role="user", content=content),)),
    )

    assert payload["tool"] == "device.volume.adjust"
    assert "invalid" in payload["error"]
    assert volume.calls == []


def test_provider_search_query_is_trimmed_before_port_execution():
    search = RecordingSearchPort()
    registry = build_default_tool_registry(web_search_port=search)

    payload = registry.execute(
        _event(
            "provider-search",
            "knowledge.web.search",
            {"query": "  focused query  ", "max_results": 2},
        )
    )

    assert payload == {"port": "search", "query": "focused query", "max_results": 2}
    assert search.calls == [("provider-search", "focused query", 2)]


def test_whitespace_only_provider_search_query_never_calls_the_port():
    search = RecordingSearchPort()
    registry = build_default_tool_registry(web_search_port=search)

    payload = registry.execute(
        _event("blank-search", "knowledge.web.search", {"query": "   "})
    )

    assert payload["tool"] == "knowledge.web.search"
    assert "invalid" in payload["error"]
    assert search.calls == []


def test_schedule_catalog_uses_standard_datetime_and_iana_timezone_formats():
    registry = build_default_tool_registry()
    schema = registry.event_registry.definition(
        "calendar.schedule.create"
    ).parameters

    assert schema["properties"]["start_at"]["format"] == "date-time"
    assert schema["properties"]["timezone"]["format"] == "iana-timezone"


def test_schedule_standard_datetime_accepts_lowercase_tz_and_rejects_bad_date():
    registry = build_default_tool_registry().event_registry
    base = {
        "title": "Review",
        "timezone": "Asia/Shanghai",
    }

    assert registry.iter_validation_errors(
        "calendar.schedule.create",
        {**base, "start_at": "2026-07-14t09:30:00z"},
    ) == ()
    assert registry.iter_validation_errors(
        "calendar.schedule.create",
        {**base, "start_at": "2026-02-31T09:30:00Z"},
    )


@pytest.mark.parametrize(
    "arguments",
    [
        {
            "title": "Impossible date",
            "start_at": "2026-02-30T09:30:00+08:00",
            "timezone": "Asia/Shanghai",
        },
        {
            "title": "Unknown zone",
            "start_at": "2026-07-14T09:30:00+08:00",
            "timezone": "Mars/Olympus",
        },
        {
            "title": "Unsafe zone",
            "start_at": "2026-07-14T09:30:00+08:00",
            "timezone": "../etc/passwd",
        },
        {
            "title": "   ",
            "start_at": "2026-07-14T09:30:00+08:00",
            "timezone": "Asia/Shanghai",
        },
    ],
)
def test_invalid_schedule_boundaries_never_call_the_port(arguments):
    calendar = RecordingCalendarPort()
    registry = build_default_tool_registry(calendar_schedule_port=calendar)

    payload = registry.execute(
        _event("invalid-schedule", "calendar.schedule.create", arguments)
    )

    assert payload["tool"] == "calendar.schedule.create"
    assert "invalid" in payload["error"]
    assert calendar.calls == []


def test_integral_float_numeric_arguments_reach_ports_as_ints():
    volume = RecordingVolumePort()
    calendar = RecordingCalendarPort()
    search = RecordingSearchPort()
    registry = build_default_tool_registry(
        device_volume_port=volume,
        calendar_schedule_port=calendar,
        web_search_port=search,
    )

    registry.execute(
        _event(
            "absolute-float",
            "device.volume.adjust",
            {"mode": "absolute", "value": 40.0},
        )
    )
    registry.execute(
        _event(
            "relative-float",
            "device.volume.adjust",
            {"mode": "relative", "delta": -10.0},
        )
    )
    registry.execute(
        _event(
            "schedule-float",
            "calendar.schedule.create",
            {
                "title": "Review",
                "start_at": "2026-07-14T09:30:00+08:00",
                "timezone": "Asia/Shanghai",
                "reminder_minutes": 15.0,
            },
        )
    )
    registry.execute(
        _event(
            "search-float",
            "knowledge.web.search",
            {"query": "focused", "max_results": 2.0},
        )
    )

    assert volume.calls == [
        ("absolute-float", "absolute", 40, None),
        ("relative-float", "relative", None, -10),
    ]
    assert type(volume.calls[0][2]) is int
    assert type(volume.calls[1][3]) is int
    assert calendar.calls[0][-1] == 15
    assert type(calendar.calls[0][-1]) is int
    assert search.calls == [("search-float", "focused", 2)]
    assert type(search.calls[0][-1]) is int


def test_integral_float_and_int_share_the_same_volume_cache_key():
    adapter = builtin_plugins.InMemoryDeviceVolumeAdapter()
    registry = build_default_tool_registry(device_volume_port=adapter)

    first = registry.execute(
        _event(
            "same-number",
            "device.volume.adjust",
            {"mode": "absolute", "value": 40},
        )
    )
    replay = registry.execute(
        _event(
            "same-number",
            "device.volume.adjust",
            {"mode": "absolute", "value": 40.0},
        )
    )

    assert replay == first
    assert len(adapter._results) == 1


@pytest.mark.parametrize(
    "arguments",
    [
        {"mode": "absolute", "value": True},
        {"mode": "relative", "delta": False},
    ],
)
def test_volume_normalizer_does_not_treat_bool_as_integral_number(arguments):
    volume = RecordingVolumePort()
    registry = build_default_tool_registry(device_volume_port=volume)

    payload = registry.execute(_event("bool-volume", "device.volume.adjust", arguments))

    assert payload["tool"] == "device.volume.adjust"
    assert "invalid" in payload["error"]
    assert volume.calls == []