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"}), ("volume to 40.5", {"mode": "absolute", "value": "40.5"}), ("volume at 1,000", {"mode": "absolute", "value": "1,000"}), ("volume=40.5", {"mode": "absolute", "value": "40.5"}), ], ) 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", ["volume to 40", "volume at 40", "volume=40"]) def test_bare_volume_integer_forms_remain_valid(content): volume = RecordingVolumePort() registry = build_default_tool_registry(device_volume_port=volume) payload = registry.handle( _event("bare-integer", "device.volume.adjust", {}), EventExecutionContext(history=(ChatMessage(role="user", content=content),)), ) assert payload["port"] == "volume" assert volume.calls == [("bare-integer", "absolute", 40, None)] @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 == []