fix: harden composed schema fallback

Problem: composed-schema validation treated required errors from unrelated branches as fallback-eligible, while argument copying could leak exceptions and validation errors exposed registry internals.

Risk: conservative branch classification may skip fallback for ambiguous composed schemas, and strict JSON snapshots reject non-JSON or type-changing argument values.

src/agent_lab/application/events/kernel.py

@@ -3,12 +3,9 @@ from __future__ import annotations
 import inspect
 import json
 from collections.abc import Awaitable, Callable, Iterable
-from copy import deepcopy
 from dataclasses import dataclass, replace
 from typing import Any
 
-from jsonschema.exceptions import ValidationError
-
 from agent_lab.application.events.models import (
     EventDefinition,
     EventArgumentResolution,
@@ -19,7 +16,7 @@ from agent_lab.application.events.models import (
     EventStatus,
     ResolvedEventArguments,
 )
-from agent_lab.application.events.registry import EventRegistry
+from agent_lab.application.events.registry import EventRegistry, ValidationIssue
 
 
 ArgumentFallbackResult = ResolvedEventArguments | None
@@ -86,7 +83,7 @@ class EventKernel:
             used_fallback = True
             fallback_request = replace(
                 request,
-                arguments=resolved.arguments,
+                arguments=self._strict_json_copy(resolved.arguments),
                 raw_arguments=resolved.raw_arguments,
             )
             try:
@@ -131,7 +128,7 @@ class EventKernel:
 
         resolved_request = replace(
             request,
-            arguments=deepcopy(resolved.arguments),
+            arguments=self._strict_json_copy(resolved.arguments),
             raw_arguments=resolved.raw_arguments,
         )
         try:
@@ -219,7 +216,7 @@ class EventKernel:
             )
         resolved_request = replace(
             request,
-            arguments=deepcopy(resolved.arguments),
+            arguments=self._strict_json_copy(resolved.arguments),
             raw_arguments=resolved.raw_arguments,
         )
         try:
@@ -280,7 +277,7 @@ class EventKernel:
                 event_name=request.name,
                 status=EventStatus.UNKNOWN,
                 source=request.source,
-                arguments=deepcopy(request.arguments),
+                arguments=self._safe_json_object_snapshot(request.arguments),
                 raw_arguments=request.raw_arguments,
                 error="unknown event",
             )
@@ -291,7 +288,7 @@ class EventKernel:
                 EventStatus.DISABLED,
                 resolved=ResolvedEventArguments(
                     event_name=request.name,
-                    arguments=deepcopy(request.arguments),
+                    arguments=self._safe_json_object_snapshot(request.arguments),
                     raw_arguments=request.raw_arguments,
                 ),
                 error="event disabled",
@@ -305,10 +302,24 @@ class EventKernel:
         context: EventExecutionContext,
     ) -> tuple[ResolvedEventArguments | None, bool, EventResult | None]:
         if request.source is EventSource.PROVIDER_RESOLVED:
+            try:
+                arguments = self._strict_json_copy(request.arguments)
+            except Exception:
+                return None, False, self._result(
+                    definition,
+                    request,
+                    EventStatus.INVALID_ARGUMENTS,
+                    resolved=ResolvedEventArguments(
+                        event_name=request.name,
+                        arguments={},
+                        raw_arguments=request.raw_arguments,
+                    ),
+                    error="provider-resolved event arguments are not valid JSON",
+                )
             return (
                 ResolvedEventArguments(
                     event_name=request.name,
-                    arguments=deepcopy(request.arguments),
+                    arguments=arguments,
                     raw_arguments=request.raw_arguments,
                 ),
                 True,
@@ -338,10 +349,10 @@ class EventKernel:
                 request,
                 "event argument resolver returned invalid payload",
             )
-        copied_arguments = deepcopy(arguments)
         try:
+            copied_arguments = self._strict_json_copy(arguments)
             raw_arguments = self._json_arguments(copied_arguments)
-        except (TypeError, ValueError) as exc:
+        except Exception as exc:
             return None, False, self._resolution_error(
                 definition,
                 request,
@@ -399,8 +410,8 @@ class EventKernel:
                 used_fallback=True,
             )
         try:
-            self._json_arguments(value.arguments)
-        except (TypeError, ValueError):
+            copied_arguments = self._strict_json_copy(value.arguments)
+        except Exception:
             return None, self._resolution_error(
                 definition,
                 request,
@@ -419,7 +430,7 @@ class EventKernel:
         return (
             ResolvedEventArguments(
                 event_name=value.event_name,
-                arguments=deepcopy(value.arguments),
+                arguments=copied_arguments,
                 raw_arguments=value.raw_arguments,
             ),
             None,
@@ -444,30 +455,32 @@ class EventKernel:
             )
         if not errors:
             return _ValidationResult()
-        missing_errors = [
-            required_error
-            for error in errors
-            for required_error in _required_errors(error)
-        ]
-        error = missing_errors[0] if missing_errors else errors[0]
+        missing_error = next(
+            (
+                required_error
+                for error in errors
+                if (required_error := _fallback_required_error(error))
+                is not None
+            ),
+            None,
+        )
+        error = missing_error or errors[0]
         return _ValidationResult(
             status=EventStatus.INVALID_ARGUMENTS,
             error=self._format_validation_error(error),
-            missing_required=bool(missing_errors),
+            missing_required=missing_error is not None,
         )
 
-    def _format_validation_error(self, error: ValidationError) -> str:
+    def _format_validation_error(self, error: ValidationIssue) -> str:
         if error.validator == "required":
-            missing = [
-                name
-                for name in error.validator_value
-                if name not in error.instance
-            ]
-            return f"missing required arguments: {', '.join(missing)}"
+            return (
+                "missing required arguments: "
+                f"{', '.join(error.missing_required)}"
+            )
         if error.validator == "type" and error.path:
             return (
                 f"invalid argument type for {error.path[-1]}: "
-                f"expected {error.validator_value}"
+                f"expected {error.expected}"
             )
         return f"invalid event arguments: {error.message}"
 
@@ -487,7 +500,7 @@ class EventKernel:
             resolved=resolved
             or ResolvedEventArguments(
                 event_name=request.name,
-                arguments=deepcopy(request.arguments),
+                arguments=self._safe_json_object_snapshot(request.arguments),
                 raw_arguments=request.raw_arguments,
             ),
             error=error,
@@ -510,9 +523,9 @@ class EventKernel:
             event_name=request.name,
             status=status,
             source=request.source,
-            arguments=deepcopy(resolved.arguments),
+            arguments=self._safe_json_object_snapshot(resolved.arguments),
             raw_arguments=resolved.raw_arguments,
-            payload=deepcopy(payload or {}),
+            payload=self._safe_json_object_snapshot(payload or {}),
             error=error,
             used_fallback=used_fallback,
             result_policy=definition.result_policy,
@@ -542,16 +555,61 @@ class EventKernel:
             raise ValueError("JSON round-trip changed payload")
         return copied
 
+    def _safe_json_object_snapshot(self, value: Any) -> dict[str, Any]:
+        if not isinstance(value, dict):
+            return {}
+        try:
+            return self._strict_json_copy(value)
+        except Exception:
+            return {}
+
 
 def _reject_json_constant(value: str) -> None:
     raise ValueError(f"non-standard JSON constant: {value}")
 
 
-def _required_errors(error: ValidationError) -> Iterable[ValidationError]:
+def _fallback_required_error(error: ValidationIssue) -> ValidationIssue | None:
     if error.validator == "required":
-        yield error
+        return error
+    if error.validator not in {"anyOf", "oneOf"}:
+        return None
+
+    branches: dict[int, list[ValidationIssue]] = {}
     for nested_error in error.context:
-        yield from _required_errors(nested_error)
+        branch = _branch_index(nested_error)
+        if branch is None:
+            return None
+        branches.setdefault(branch, []).append(nested_error)
+
+    viable: list[ValidationIssue] = []
+    for branch_errors in branches.values():
+        required_errors: list[ValidationIssue] = []
+        substantive = False
+        for branch_error in branch_errors:
+            required_error = _missing_only_required_error(branch_error)
+            if required_error is None:
+                substantive = True
+                break
+            required_errors.append(required_error)
+        if required_errors and not substantive:
+            viable.append(required_errors[0])
+    return viable[0] if len(viable) == 1 else None
+
+
+def _missing_only_required_error(
+    error: ValidationIssue,
+) -> ValidationIssue | None:
+    if error.validator == "required":
+        return error
+    if error.validator in {"anyOf", "oneOf"}:
+        return _fallback_required_error(error)
+    return None
+
+
+def _branch_index(error: ValidationIssue) -> int | None:
+    if error.schema_path and isinstance(error.schema_path[0], int):
+        return error.schema_path[0]
+    return None
 
 
 def _json_values_equal(left: Any, right: Any) -> bool:

src/agent_lab/application/events/registry.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Iterable
 from copy import deepcopy
-from dataclasses import replace
+from dataclasses import dataclass, replace
 from types import MappingProxyType
 from typing import Any
 
@@ -12,6 +12,17 @@ from jsonschema.exceptions import SchemaError, ValidationError
 from agent_lab.application.events.models import EventDefinition
 
 
+@dataclass(frozen=True)
+class ValidationIssue:
+    validator: str | None
+    message: str
+    path: tuple[str | int, ...]
+    schema_path: tuple[str | int, ...]
+    expected: str | None = None
+    missing_required: tuple[str, ...] = ()
+    context: tuple[ValidationIssue, ...] = ()
+
+
 class EventRegistry:
     def __init__(self, definitions: Iterable[EventDefinition] = ()) -> None:
         self._definitions: dict[str, EventDefinition] = {}
@@ -44,12 +55,13 @@ class EventRegistry:
         self,
         name: str,
         arguments: dict[str, Any],
-    ) -> tuple[ValidationError, ...]:
+    ) -> tuple[ValidationIssue, ...]:
         validator = self._validators.get(name)
         if validator is None:
             raise LookupError(f"missing validator for event: {name}")
         return tuple(
-            deepcopy(error) for error in validator.iter_errors(arguments)
+            _summarize_validation_error(error)
+            for error in validator.iter_errors(arguments)
         )
 
     def catalog(self, enabled_names: Iterable[str] | None = None) -> list[dict]:
@@ -94,3 +106,39 @@ def _thaw_json(value: Any) -> Any:
     if isinstance(value, tuple):
         return [_thaw_json(item) for item in value]
     return deepcopy(value)
+
+
+def _summarize_validation_error(error: ValidationError) -> ValidationIssue:
+    missing_required: tuple[str, ...] = ()
+    if (
+        error.validator == "required"
+        and isinstance(error.instance, dict)
+        and isinstance(error.validator_value, list)
+    ):
+        missing_required = tuple(
+            name
+            for name in error.validator_value
+            if isinstance(name, str) and name not in error.instance
+        )
+    return ValidationIssue(
+        validator=error.validator if isinstance(error.validator, str) else None,
+        message=str(error.message),
+        path=tuple(_path_component(item) for item in error.path),
+        schema_path=tuple(_path_component(item) for item in error.schema_path),
+        expected=(
+            str(error.validator_value)
+            if error.validator == "type"
+            else None
+        ),
+        missing_required=missing_required,
+        context=tuple(
+            _summarize_validation_error(nested_error)
+            for nested_error in error.context
+        ),
+    )
+
+
+def _path_component(value: Any) -> str | int:
+    if isinstance(value, int | str):
+        return value
+    return repr(value)

tests/test_event_kernel.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import gc
 import json
 import warnings
+from dataclasses import FrozenInstanceError
 from typing import Any
 
 import pytest
@@ -101,7 +102,10 @@ def test_registry_does_not_expose_mutable_validator_instances():
     )
     assert errors
 
-    errors[0].schema["type"] = "integer"
+    assert not hasattr(errors[0], "schema")
+    assert not hasattr(errors[0], "instance")
+    with pytest.raises(FrozenInstanceError):
+        errors[0].message = "mutated"  # type: ignore[misc]
 
     assert list(
         registry.iter_validation_errors(
@@ -261,9 +265,38 @@ async def test_kernel_does_not_fallback_for_complete_invalid_arguments(
     assert fallback_calls == 0
 
 
+def _discriminated_composed_parameters(composition: str) -> dict[str, Any]:
+    return {
+        "type": "object",
+        "properties": {
+            "choice": {
+                composition: [
+                    {
+                        "type": "object",
+                        "properties": {
+                            "kind": {"const": "a"},
+                            "value": {"type": "string"},
+                        },
+                        "required": ["kind", "value"],
+                    },
+                    {
+                        "type": "object",
+                        "properties": {
+                            "kind": {"const": "b"},
+                            "count": {"type": "integer"},
+                        },
+                        "required": ["kind", "count"],
+                    },
+                ]
+            }
+        },
+        "required": ["choice"],
+    }
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("composition", ["anyOf", "oneOf"])
-async def test_kernel_falls_back_for_required_missing_inside_composed_schema(
+async def test_kernel_falls_back_for_matching_composed_branch_missing_required(
     composition: str,
 ):
     fallback_calls = 0
@@ -273,32 +306,13 @@ async def test_kernel_falls_back_for_required_missing_inside_composed_schema(
         fallback_calls += 1
         return ResolvedEventArguments(
             event_name="example.lookup",
-            arguments={"choice": {"mode": "alpha"}},
-            raw_arguments='{"choice":{"mode":"alpha"}}',
+            arguments={"choice": {"kind": "a", "value": "resolved"}},
+            raw_arguments='{"choice":{"kind":"a","value":"resolved"}}',
         )
 
     definition = _definition(
-        parameters={
-            "type": "object",
-            "properties": {
-                "choice": {
-                    composition: [
-                        {
-                            "type": "object",
-                            "properties": {"mode": {"const": "alpha"}},
-                            "required": ["mode"],
-                        },
-                        {
-                            "type": "object",
-                            "properties": {"mode": {"const": "beta"}},
-                            "required": ["mode"],
-                        },
-                    ]
-                }
-            },
-            "required": ["choice"],
-        },
-        resolver=lambda request, context: {"choice": {}},
+        parameters=_discriminated_composed_parameters(composition),
+        resolver=lambda request, context: {"choice": {"kind": "a"}},
         handler=lambda request: {"ok": True},
     )
 
@@ -316,7 +330,7 @@ async def test_kernel_falls_back_for_required_missing_inside_composed_schema(
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("composition", ["anyOf", "oneOf"])
-async def test_kernel_does_not_fallback_for_complete_invalid_composed_schema(
+async def test_kernel_does_not_fallback_when_matching_composed_branch_is_invalid(
     composition: str,
 ):
     fallback_calls = 0
@@ -327,27 +341,42 @@ async def test_kernel_does_not_fallback_for_complete_invalid_composed_schema(
         raise AssertionError("fallback should not run")
 
     definition = _definition(
-        parameters={
-            "type": "object",
-            "properties": {
-                "choice": {
-                    composition: [
-                        {
-                            "type": "object",
-                            "properties": {"mode": {"const": "alpha"}},
-                            "required": ["mode"],
-                        },
-                        {
-                            "type": "object",
-                            "properties": {"mode": {"const": "beta"}},
-                            "required": ["mode"],
-                        },
-                    ]
-                }
-            },
-            "required": ["choice"],
+        parameters=_discriminated_composed_parameters(composition),
+        resolver=lambda request, context: {
+            "choice": {"kind": "a", "value": 42}
         },
-        resolver=lambda request, context: {"choice": {"mode": "other"}},
+        handler=lambda request: {"ok": True},
+    )
+
+    result = await EventKernel(
+        EventRegistry([definition]), argument_fallback=fallback
+    ).execute(
+        EventRequest(id="event-1", name="example.lookup"),
+        enabled_names=["example.lookup"],
+    )
+
+    assert result.status is EventStatus.INVALID_ARGUMENTS
+    assert result.used_fallback is False
+    assert fallback_calls == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("composition", ["anyOf", "oneOf"])
+@pytest.mark.parametrize("choice", [{}, {"kind": "other"}])
+async def test_kernel_does_not_fallback_when_composed_branch_is_ambiguous(
+    composition: str,
+    choice: dict[str, Any],
+):
+    fallback_calls = 0
+
+    async def fallback(*args: Any) -> ResolvedEventArguments:
+        nonlocal fallback_calls
+        fallback_calls += 1
+        raise AssertionError("fallback should not run")
+
+    definition = _definition(
+        parameters=_discriminated_composed_parameters(composition),
+        resolver=lambda request, context: {"choice": choice},
         handler=lambda request: {"ok": True},
     )
 
@@ -853,6 +882,73 @@ async def test_kernel_normalizes_non_json_resolver_arguments():
     assert result.error.startswith("event argument resolver failed to serialize:")
 
 
+class _ExplodingDeepcopyDict(dict[str, Any]):
+    def __deepcopy__(self, memo: dict[int, Any]) -> dict[str, Any]:
+        raise RuntimeError("deepcopy must not be used")
+
+
+@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 resolver failed to serialize: "
+            "JSON round-trip changed payload",
+        ),
+        (
+            EventSource.PROVIDER_RESOLVED,
+            EventStatus.INVALID_ARGUMENTS,
+            "provider-resolved event arguments are not valid JSON",
+        ),
+    ],
+)
+async def test_kernel_normalizes_argument_snapshot_failures_without_deepcopy(
+    execution: str,
+    source: EventSource,
+    expected_status: EventStatus,
+    expected_error: str,
+):
+    fallback_calls = 0
+    arguments = {"query": _ExplodingDeepcopyDict({"nested": "value"})}
+
+    async def fallback(*args: Any) -> ResolvedEventArguments:
+        nonlocal fallback_calls
+        fallback_calls += 1
+        raise AssertionError("fallback should not run")
+
+    definition = _definition(
+        parameters={
+            "type": "object",
+            "properties": {"query": {"type": "object"}},
+            "required": ["query"],
+        },
+        resolver=lambda request, context: arguments,
+        handler=lambda request: {"ok": True},
+    )
+    request = EventRequest(
+        id="event-1",
+        name="example.lookup",
+        arguments=arguments if source is EventSource.PROVIDER_RESOLVED else {},
+        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.status is not EventStatus.DEFINITION_ERROR
+    assert result.error == expected_error
+    assert result.used_fallback is False
+    assert fallback_calls == 0
+
+
 @pytest.mark.asyncio
 async def test_kernel_normalizes_non_object_fallback_arguments():
     async def fallback(*args: Any) -> ResolvedEventArguments: