fix: harden built-in plugin boundaries

Problem:
- Provider-resolved arguments bypassed canonicalization, format validation was incomplete, negated or oversized volume requests could reach ports, and reused short event IDs could return stale adapter results.

Risk:
- Normalizer ordering must preserve provider raw arguments and no-fallback behavior; stricter date-time/timezone checks and canonical cache keys may reject inputs previously accepted by the lab.

src/agent_lab/application/events/__init__.py

@@ -12,6 +12,7 @@ from agent_lab.application.events.builtin_plugins import (
 )
 from agent_lab.application.events.models import (
     ConfirmationPolicy,
+    EventArgumentNormalizer,
     EventArgumentResolution,
     EventArgumentResolver,
     EventDefinition,
@@ -31,6 +32,7 @@ __all__ = [
     "ArgumentFallback",
     "CalendarSchedulePort",
     "ConfirmationPolicy",
+    "EventArgumentNormalizer",
     "EventArgumentResolution",
     "EventArgumentResolver",
     "EventDefinition",

src/agent_lab/application/events/builtin_plugins.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import re
 from collections.abc import Awaitable, Callable
 from copy import deepcopy
@@ -78,7 +79,7 @@ class WebSearchPort(Protocol):
 
 class InMemorySessionTerminationAdapter:
     def __init__(self) -> None:
-        self._results: dict[str, dict[str, Any]] = {}
+        self._results: dict[tuple[str, str], dict[str, Any]] = {}
 
     def terminate(
         self,
@@ -86,19 +87,21 @@ class InMemorySessionTerminationAdapter:
         *,
         reason: str | None = None,
     ) -> dict[str, Any]:
-        if event_id not in self._results:
-            self._results[event_id] = {
+        arguments = {} if reason is None else {"reason": reason}
+        cache_key = _adapter_cache_key(event_id, arguments)
+        if cache_key not in self._results:
+            self._results[cache_key] = {
                 "tool": "session.terminate",
                 "status": "terminated",
                 "event_id": event_id,
                 "reason": reason,
             }
-        return deepcopy(self._results[event_id])
+        return deepcopy(self._results[cache_key])
 
 
 class InMemoryDeviceVolumeAdapter:
     def __init__(self) -> None:
-        self._results: dict[str, dict[str, Any]] = {}
+        self._results: dict[tuple[str, str], dict[str, Any]] = {}
 
     def adjust(
         self,
@@ -108,7 +111,13 @@ class InMemoryDeviceVolumeAdapter:
         value: int | None = None,
         delta: int | None = None,
     ) -> dict[str, Any]:
-        if event_id not in self._results:
+        arguments: dict[str, Any] = {"mode": mode}
+        if value is not None:
+            arguments["value"] = value
+        if delta is not None:
+            arguments["delta"] = delta
+        cache_key = _adapter_cache_key(event_id, arguments)
+        if cache_key not in self._results:
             payload: dict[str, Any] = {
                 "tool": "device.volume.adjust",
                 "status": "applied",
@@ -119,13 +128,13 @@ class InMemoryDeviceVolumeAdapter:
                 payload["value"] = value
             if delta is not None:
                 payload["delta"] = delta
-            self._results[event_id] = payload
-        return deepcopy(self._results[event_id])
+            self._results[cache_key] = payload
+        return deepcopy(self._results[cache_key])
 
 
 class InMemoryCalendarScheduleAdapter:
     def __init__(self) -> None:
-        self._results: dict[str, dict[str, Any]] = {}
+        self._results: dict[tuple[str, str], dict[str, Any]] = {}
 
     def create(
         self,
@@ -137,23 +146,24 @@ class InMemoryCalendarScheduleAdapter:
         recurrence: str | None = None,
         reminder_minutes: int | None = None,
     ) -> dict[str, Any]:
-        if event_id not in self._results:
-            schedule: dict[str, Any] = {
-                "title": title,
-                "start_at": start_at,
-                "timezone": timezone,
-            }
-            if recurrence is not None:
-                schedule["recurrence"] = recurrence
-            if reminder_minutes is not None:
-                schedule["reminder_minutes"] = reminder_minutes
-            self._results[event_id] = {
+        schedule: dict[str, Any] = {
+            "title": title,
+            "start_at": start_at,
+            "timezone": timezone,
+        }
+        if recurrence is not None:
+            schedule["recurrence"] = recurrence
+        if reminder_minutes is not None:
+            schedule["reminder_minutes"] = reminder_minutes
+        cache_key = _adapter_cache_key(event_id, schedule)
+        if cache_key not in self._results:
+            self._results[cache_key] = {
                 "tool": "calendar.schedule.create",
                 "status": "created",
                 "event_id": event_id,
                 "schedule": schedule,
             }
-        return deepcopy(self._results[event_id])
+        return deepcopy(self._results[cache_key])
 
 
 class InMemoryWebSearchAdapter:
@@ -292,6 +302,7 @@ def _device_volume_definition(port: DeviceVolumePort) -> EventDefinition:
         },
         handler=handler,
         resolver=_resolve_device_volume,
+        normalizer=_normalize_device_volume,
         fallback_allowed=True,
         result_policy=ResultPolicy.SILENT_SUCCESS,
         risk_level=RiskLevel.MEDIUM,
@@ -319,14 +330,14 @@ def _calendar_schedule_definition(port: CalendarSchedulePort) -> EventDefinition
         parameters={
             "type": "object",
             "properties": {
-                "title": {"type": "string", "minLength": 1},
+                "title": {"type": "string", "pattern": r"\S"},
                 "start_at": {
                     "type": "string",
-                    "pattern": f"^{_RFC3339_PATTERN}$",
+                    "format": "date-time",
                 },
                 "timezone": {
                     "type": "string",
-                    "pattern": f"^{_TIMEZONE_PATTERN}$",
+                    "format": "iana-timezone",
                 },
                 "recurrence": {"type": "string", "minLength": 1},
                 "reminder_minutes": {"type": "integer", "minimum": 0},
@@ -360,7 +371,7 @@ def _knowledge_search_definition(port: WebSearchPort) -> EventDefinition:
         parameters={
             "type": "object",
             "properties": {
-                "query": {"type": "string", "minLength": 1},
+                "query": {"type": "string", "pattern": r"\S"},
                 "max_results": {
                     "type": "integer",
                     "minimum": 1,
@@ -372,6 +383,7 @@ def _knowledge_search_definition(port: WebSearchPort) -> EventDefinition:
         },
         handler=handler,
         resolver=_resolve_knowledge_search,
+        normalizer=_normalize_knowledge_search,
         fallback_allowed=False,
         result_policy=ResultPolicy.LLM_FOLLOW_UP,
         risk_level=RiskLevel.LOW,
@@ -396,8 +408,7 @@ def _resolve_device_volume(
         arguments = deepcopy(request.arguments)
         mode = arguments.get("mode")
         if isinstance(mode, str):
-            arguments["mode"] = mode.strip().lower()
-            mode = arguments["mode"]
+            mode = mode.strip().lower()
         complete = mode in {"mute", "unmute"} or (
             mode == "absolute" and "value" in arguments
         ) or (mode == "relative" and "delta" in arguments)
@@ -410,6 +421,11 @@ def _resolve_device_volume(
         )
 
     content = _latest_user_content(context).lower()
+    if re.search(
+        r"\bdon['’]t\b|\bdo\s+not\b|\bnot\b|\bnever\b|不要|别|不许|禁止",
+        content,
+    ):
+        return EventArgumentResolution(arguments={}, complete=False)
     unmute = bool(re.search(r"\bunmute\b|取消静音|解除静音|恢复声音", content))
     mute = bool(re.search(r"\bmute\b|(?<!取消)(?<!解除)静音", content))
     if mute and unmute:
@@ -423,32 +439,32 @@ def _resolve_device_volume(
 
     absolute_matches = [
         re.search(
-            r"音量.{0,6}?(?:调到|调至|设置为|设为|到)\s*(\d{1,3})",
+            r"音量.{0,6}?(?:调到|调至|设置为|设为|到)\s*(\d+)",
             content,
         ),
         re.search(
-            r"(?:set|change)\s+(?:the\s+)?volume\s+(?:to|at)\s*(\d{1,3})",
+            r"(?:set|change)\s+(?:the\s+)?volume\s+(?:to|at)\s*(\d+)",
             content,
         ),
-        re.search(r"\bvolume\s*(?:to|at|=)\s*(\d{1,3})", content),
+        re.search(r"\bvolume\s*(?:to|at|=)\s*(\d+)", content),
     ]
     positive_matches = [
-        re.search(r"音量.{0,4}?(?:增加|调高|提高|加)\s*(\d{1,3})", content),
+        re.search(r"音量.{0,4}?(?:增加|调高|提高|加)\s*(\d+)", content),
         re.search(
             r"(?:increase|raise|turn\s+up)\s+(?:the\s+)?volume"
-            r"(?:\s+by)?\s*(\d{1,3})",
+            r"(?:\s+by)?\s*(\d+)",
             content,
         ),
     ]
     negative_matches = [
-        re.search(r"音量.{0,4}?(?:降低|调低|减少|减)\s*(\d{1,3})", content),
+        re.search(r"音量.{0,4}?(?:降低|调低|减少|减)\s*(\d+)", content),
         re.search(
             r"(?:decrease|lower|turn\s+down)\s+(?:the\s+)?volume"
-            r"(?:\s+by)?\s*(\d{1,3})",
+            r"(?:\s+by)?\s*(\d+)",
             content,
         ),
     ]
-    signed = re.search(r"(?:音量|\bvolume\b)\s*([+-])\s*(\d{1,3})", content)
+    signed = re.search(r"(?:音量|\bvolume\b)\s*([+-])\s*(\d+)", content)
     absolute = next((match for match in absolute_matches if match), None)
     positive = next((match for match in positive_matches if match), None)
     negative = next((match for match in negative_matches if match), None)
@@ -526,6 +542,14 @@ def _resolve_calendar_schedule(
     return EventArgumentResolution(arguments=arguments, complete=False)
 
 
+def _normalize_device_volume(arguments: dict[str, Any]) -> dict[str, Any]:
+    normalized = deepcopy(arguments)
+    mode = normalized.get("mode")
+    if isinstance(mode, str):
+        normalized["mode"] = mode.strip().lower()
+    return normalized
+
+
 def _resolve_knowledge_search(
     request: EventRequest,
     context: EventExecutionContext,
@@ -540,6 +564,14 @@ def _resolve_knowledge_search(
     return EventArgumentResolution(arguments=arguments, complete=False)
 
 
+def _normalize_knowledge_search(arguments: dict[str, Any]) -> dict[str, Any]:
+    normalized = deepcopy(arguments)
+    query = normalized.get("query")
+    if isinstance(query, str):
+        normalized["query"] = query.strip()
+    return normalized
+
+
 def _latest_user_content(context: EventExecutionContext) -> str:
     for message in reversed(context.history):
         if message.role == "user" and message.content.strip():
@@ -558,3 +590,18 @@ def _format_retrieved_at(value: datetime | str) -> str:
         value = value.replace(tzinfo=datetime_timezone.utc)
     rendered = value.astimezone(datetime_timezone.utc).isoformat(timespec="seconds")
     return rendered.replace("+00:00", "Z")
+
+
+def _adapter_cache_key(
+    event_id: str,
+    arguments: dict[str, Any],
+) -> tuple[str, str]:
+    return (
+        event_id,
+        json.dumps(
+            arguments,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        ),
+    )

src/agent_lab/application/events/kernel.py

@@ -65,6 +65,14 @@ class EventKernel:
         if resolution_error is not None:
             return resolution_error
         assert resolved is not None
+        resolved, normalization_error = self._apply_normalizer(
+            definition,
+            request,
+            resolved,
+        )
+        if normalization_error is not None:
+            return normalization_error
+        assert resolved is not None
 
         validation = self._validate(definition, resolved.arguments)
         fallback_eligible = validation.missing_required or (
@@ -114,6 +122,14 @@ class EventKernel:
                 if fallback_error is not None:
                     return fallback_error
                 assert resolved is not None
+                resolved, normalization_error = self._apply_normalizer(
+                    definition,
+                    request,
+                    resolved,
+                )
+                if normalization_error is not None:
+                    return normalization_error
+                assert resolved is not None
                 validation = self._validate(
                     definition,
                     resolved.arguments,
@@ -195,6 +211,14 @@ class EventKernel:
         if resolution_error is not None:
             return resolution_error
         assert resolved is not None
+        resolved, normalization_error = self._apply_normalizer(
+            definition,
+            request,
+            resolved,
+        )
+        if normalization_error is not None:
+            return normalization_error
+        assert resolved is not None
         validation = self._validate(definition, resolved.arguments)
         if not resolution_complete and validation.status is None:
             validation = _ValidationResult(
@@ -371,6 +395,51 @@ class EventKernel:
             None,
         )
 
+    def _apply_normalizer(
+        self,
+        definition: EventDefinition,
+        request: EventRequest,
+        resolved: ResolvedEventArguments,
+    ) -> tuple[ResolvedEventArguments | None, EventResult | None]:
+        if definition.normalizer is None:
+            return resolved, None
+        try:
+            normalized = definition.normalizer(
+                self._strict_json_copy(resolved.arguments)
+            )
+            if not isinstance(normalized, dict):
+                raise TypeError("normalizer returned non-object")
+            normalized = self._strict_json_copy(normalized)
+        except Exception:
+            if request.source is EventSource.PROVIDER_RESOLVED:
+                return None, self._result(
+                    definition,
+                    request,
+                    EventStatus.INVALID_ARGUMENTS,
+                    resolved=resolved,
+                    error=(
+                        "provider-resolved event arguments could not be normalized"
+                    ),
+                )
+            return None, self._resolution_error(
+                definition,
+                request,
+                "event argument normalizer failed",
+                resolved=resolved,
+            )
+        return (
+            ResolvedEventArguments(
+                event_name=resolved.event_name,
+                arguments=normalized,
+                raw_arguments=(
+                    resolved.raw_arguments
+                    if request.source is EventSource.PROVIDER_RESOLVED
+                    else self._json_arguments(normalized)
+                ),
+            ),
+            None,
+        )
+
     def _normalize_fallback(
         self,
         definition: EventDefinition,

src/agent_lab/application/events/models.py

@@ -76,6 +76,7 @@ EventArgumentResolver = Callable[
     [EventRequest, EventExecutionContext],
     dict[str, Any] | EventArgumentResolution,
 ]
+EventArgumentNormalizer = Callable[[dict[str, Any]], dict[str, Any]]
 
 
 @dataclass(frozen=True)
@@ -85,6 +86,7 @@ class EventDefinition:
     parameters: dict[str, Any]
     handler: EventHandler
     resolver: EventArgumentResolver | None = None
+    normalizer: EventArgumentNormalizer | None = None
     schema_version: str = "1"
     fallback_allowed: bool = True
     result_policy: ResultPolicy = ResultPolicy.LLM_FOLLOW_UP

src/agent_lab/application/events/registry.py

@@ -1,17 +1,48 @@
 from __future__ import annotations
 
+import re
 from collections.abc import Iterable
 from copy import deepcopy
 from dataclasses import dataclass, replace
+from datetime import datetime
 from types import MappingProxyType
 from typing import Any
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
-from jsonschema import Draft202012Validator
+from jsonschema import Draft202012Validator, FormatChecker
 from jsonschema.exceptions import SchemaError, ValidationError
 
 from agent_lab.application.events.models import EventDefinition
 
 
+_FORMAT_CHECKER = FormatChecker()
+_RFC3339_RE = re.compile(
+    r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}"
+    r"(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$"
+)
+
+
+@_FORMAT_CHECKER.checks("date-time", raises=ValueError)
+def _is_rfc3339_datetime(value: object) -> bool:
+    if not isinstance(value, str):
+        return True
+    if _RFC3339_RE.fullmatch(value) is None:
+        return False
+    datetime.fromisoformat(value.replace("Z", "+00:00"))
+    return True
+
+
+@_FORMAT_CHECKER.checks(
+    "iana-timezone",
+    raises=(ZoneInfoNotFoundError, ValueError),
+)
+def _is_iana_timezone(value: object) -> bool:
+    if not isinstance(value, str):
+        return True
+    ZoneInfo(value)
+    return True
+
+
 @dataclass(frozen=True)
 class ValidationIssue:
     validator: str | None
@@ -45,7 +76,8 @@ class EventRegistry:
             parameters=_freeze_json(parameters),
         )
         self._validators[definition.name] = Draft202012Validator(
-            deepcopy(parameters)
+            deepcopy(parameters),
+            format_checker=_FORMAT_CHECKER,
         )
 
     def definition(self, name: str) -> EventDefinition | None:

src/agent_lab/application/tools.py

@@ -37,6 +37,7 @@ ToolArgumentResolver = Callable[
     [ToolCallEvent, ToolExecutionContext],
     dict[str, Any],
 ]
+ToolArgumentNormalizer = Callable[[dict[str, Any]], dict[str, Any]]
 
 
 @dataclass(frozen=True)
@@ -46,6 +47,7 @@ class ToolDefinition:
     parameters: dict[str, Any]
     handler: ToolHandler
     argument_resolver: ToolArgumentResolver | None = None
+    normalizer: ToolArgumentNormalizer | None = None
     schema_version: str = "1"
     fallback_allowed: bool = True
     result_policy: ResultPolicy = ResultPolicy.LLM_FOLLOW_UP
@@ -224,6 +226,7 @@ class ToolRegistry:
             parameters=deepcopy(definition.parameters),
             handler=handler,
             resolver=resolver,
+            normalizer=definition.normalizer,
             schema_version=definition.schema_version,
             fallback_allowed=definition.fallback_allowed,
             result_policy=definition.result_policy,

tests/test_builtin_event_plugins.py

@@ -573,27 +573,37 @@ def test_port_failures_are_normalized_by_the_generic_kernel(port_name, event):
     }
 
 
-def test_default_stateful_adapters_are_idempotent_by_event_id():
+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"})
     )
-    second_session = registry.execute(
+    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": "absolute", "value": 20},
+            {"mode": "  ReLaTiVe  ", "delta": 20},
+        )
+    )
+    replayed_volume = registry.execute(
+        _event(
+            "same-volume",
+            "device.volume.adjust",
+            {"mode": "relative", "delta": 20},
         )
     )
-    second_volume = registry.execute(
+    changed_volume = registry.execute(
         _event(
             "same-volume",
             "device.volume.adjust",
-            {"mode": "absolute", "value": 80},
+            {"mode": "relative", "delta": -10},
         )
     )
     first_schedule = registry.execute(
@@ -607,7 +617,18 @@ def test_default_stateful_adapters_are_idempotent_by_event_id():
             },
         )
     )
-    second_schedule = registry.execute(
+    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",
@@ -619,12 +640,15 @@ def test_default_stateful_adapters_are_idempotent_by_event_id():
         )
     )
 
-    assert second_session == first_session
+    assert replayed_session == first_session
     assert first_session["reason"] == "first"
-    assert second_volume == first_volume
-    assert first_volume["value"] == 20
-    assert second_schedule == first_schedule
+    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():
@@ -668,3 +692,139 @@ def _resolve_text_event(name: str, content: str):
         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.parametrize(
+    "content",
+    [
+        "don't mute",
+        "do not set volume to 40",
+        "不要静音",
+        "别把音量调到 20",
+        "不许音量增加 10",
+    ],
+)
+def test_negated_volume_requests_never_call_the_port(content):
+    volume = RecordingVolumePort()
+    registry = build_default_tool_registry(device_volume_port=volume)
+
+    payload = registry.handle(
+        _event("negated-volume", "device.volume.adjust", {}),
+        EventExecutionContext(history=(ChatMessage(role="user", content=content),)),
+    )
+
+    assert payload["tool"] == "device.volume.adjust"
+    assert "error" in payload
+    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"
+
+
+@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 == []

tests/test_event_kernel.py

@@ -1648,3 +1648,187 @@ async def test_kernel_applies_identical_behavior_to_arbitrary_event_names(name:
 
     assert result.status is EventStatus.SUCCESS
     assert result.payload == {"event": name, "query": "value"}
+
+
+@pytest.mark.asyncio
+async def test_kernel_runs_normalizer_after_text_resolver_before_validation():
+    captured: list[EventRequest] = []
+    definition = _definition(
+        resolver=lambda request, context: {"query": "  TEXT Value  "},
+        normalizer=lambda arguments: {"query": arguments["query"].strip().lower()},
+        handler=lambda request: captured.append(request) or {"ok": True},
+    )
+
+    result = await EventKernel(EventRegistry([definition])).execute(
+        EventRequest(id="event-1", name=definition.name),
+        enabled_names=[definition.name],
+    )
+
+    assert result.status is EventStatus.SUCCESS
+    assert result.arguments == {"query": "text value"}
+    assert result.raw_arguments == '{"query":"text value"}'
+    assert captured[0].arguments == {"query": "text value"}
+    assert captured[0].raw_arguments == '{"query":"text value"}'
+
+
+@pytest.mark.asyncio
+async def test_kernel_normalizes_provider_arguments_and_preserves_original_raw_json():
+    captured: list[EventRequest] = []
+    original_raw = '{ "query": "  PROVIDER Value  " }'
+    definition = _definition(
+        resolver=lambda request, context: (_ for _ in ()).throw(
+            AssertionError("provider must not run resolver")
+        ),
+        normalizer=lambda arguments: {"query": arguments["query"].strip().lower()},
+        handler=lambda request: captured.append(request) or {"ok": True},
+    )
+
+    result = await EventKernel(EventRegistry([definition])).execute(
+        EventRequest(
+            id="event-1",
+            name=definition.name,
+            arguments={"query": "  PROVIDER Value  "},
+            raw_arguments=original_raw,
+            source=EventSource.PROVIDER_RESOLVED,
+        ),
+        enabled_names=[definition.name],
+    )
+
+    assert result.status is EventStatus.SUCCESS
+    assert result.arguments == {"query": "provider value"}
+    assert result.raw_arguments == original_raw
+    assert captured[0].arguments == {"query": "provider value"}
+    assert captured[0].raw_arguments == original_raw
+
+
+@pytest.mark.asyncio
+async def test_provider_normalization_never_enables_argument_fallback():
+    fallback_calls = 0
+
+    async def fallback(*args: Any) -> ResolvedEventArguments:
+        nonlocal fallback_calls
+        fallback_calls += 1
+        return ResolvedEventArguments(
+            event_name="example.lookup",
+            arguments={"query": "fallback"},
+            raw_arguments='{"query":"fallback"}',
+        )
+
+    definition = _definition(normalizer=lambda arguments: {})
+    original_raw = '{"query":"provider"}'
+
+    result = await EventKernel(
+        EventRegistry([definition]), argument_fallback=fallback
+    ).execute(
+        EventRequest(
+            id="event-1",
+            name=definition.name,
+            arguments={"query": "provider"},
+            raw_arguments=original_raw,
+            source=EventSource.PROVIDER_RESOLVED,
+        ),
+        enabled_names=[definition.name],
+    )
+
+    assert result.status is EventStatus.INVALID_ARGUMENTS
+    assert result.error == "missing required arguments: query"
+    assert result.raw_arguments == original_raw
+    assert result.used_fallback is False
+    assert fallback_calls == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("execution", ["async", "sync"])
+@pytest.mark.parametrize(
+    ("source", "expected_status", "expected_error"),
+    [
+        (
+            EventSource.TEXT_EVENT,
+            EventStatus.RESOLUTION_ERROR,
+            "event argument normalizer failed",
+        ),
+        (
+            EventSource.PROVIDER_RESOLVED,
+            EventStatus.INVALID_ARGUMENTS,
+            "provider-resolved event arguments could not be normalized",
+        ),
+    ],
+)
+@pytest.mark.parametrize("normalizer_kind", ["raises", "invalid_return"])
+async def test_kernel_normalizes_normalizer_failures_without_leaking_details(
+    execution: str,
+    source: EventSource,
+    expected_status: EventStatus,
+    expected_error: str,
+    normalizer_kind: str,
+):
+    fallback_calls = 0
+
+    async def fallback(*args: Any) -> ResolvedEventArguments:
+        nonlocal fallback_calls
+        fallback_calls += 1
+        raise AssertionError("fallback must not run")
+
+    def normalizer(arguments: dict[str, Any]):
+        if normalizer_kind == "raises":
+            raise RuntimeError("secret adapter detail")
+        return [arguments]
+
+    definition = _definition(
+        resolver=lambda request, context: {"query": "value"},
+        normalizer=normalizer,
+    )
+    request = EventRequest(
+        id="event-1",
+        name=definition.name,
+        arguments={"query": "value"} if source is EventSource.PROVIDER_RESOLVED else {},
+        raw_arguments='{"query":"value"}',
+        source=source,
+    )
+    kernel = EventKernel(EventRegistry([definition]), argument_fallback=fallback)
+
+    result = (
+        await kernel.execute(request, enabled_names=[definition.name])
+        if execution == "async"
+        else kernel.execute_sync(request, enabled_names=[definition.name])
+    )
+
+    assert result.status is expected_status
+    assert result.error == expected_error
+    assert "secret" not in (result.error or "")
+    assert result.used_fallback is False
+    assert fallback_calls == 0
+
+
+def test_tool_definition_normalizer_remains_compatible_with_provider_execution():
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="compat.normalize",
+                description="Normalize compatibility data.",
+                parameters={
+                    "type": "object",
+                    "properties": {"query": {"type": "string"}},
+                    "required": ["query"],
+                },
+                handler=lambda event: {
+                    "tool": event.name,
+                    "query": event.arguments["query"],
+                },
+                normalizer=lambda arguments: {
+                    "query": arguments["query"].strip().lower()
+                },
+            )
+        ]
+    )
+
+    payload = registry.execute(
+        ToolCallEvent(
+            id="call-1",
+            name="compat.normalize",
+            arguments={"query": "  PROVIDER  "},
+            raw_arguments='{ "query": "  PROVIDER  " }',
+        )
+    )
+
+    assert payload == {"tool": "compat.normalize", "query": "provider"}