feat: add policy-driven event batches

Problem: EventAgent and direct Runtime executed event batches independently with unbounded gather behavior and unconditional LLM continuation, so timeouts, conflicts, terminal ordering, replay, and plugin result policies were inconsistent.

Risk: Batch scheduling and continuation now affect both invocation modes; focused and full tests cover bounded execution, timeout/failure isolation, stable transcripts, policy aggregation, and reusable-session termination.

src/agent_lab/application/contracts.py

@@ -30,6 +30,8 @@ class AgentParams(BaseModel):
 class EventAgentParams(AgentParams):
     enabled_tools: list[str] = Field(default_factory=lambda: ["handoff_note"])
     max_event_loops: int = 1
+    max_parallel_events: int = Field(default=4, ge=1, le=16)
+    batch_timeout_seconds: float = Field(default=15.0, gt=0, le=120)
 
 
 class DebugRunRequest(BaseModel):

src/agent_lab/application/event_agent.py

@@ -1,11 +1,13 @@
 import asyncio
 import json
-from collections.abc import AsyncIterator, Iterable, Sequence
+from collections.abc import AsyncIterator, Hashable, Iterable, Sequence
 from dataclasses import dataclass
 from typing import Any, Protocol
 
-from agent_lab.application.contracts import AgentParams
+from agent_lab.application.contracts import AgentParams, EventAgentParams
 from agent_lab.application.events import (
+    EventBatchExecutor,
+    EventBatchResult,
     EventDefinition,
     EventExecutionContext,
     EventKernel,
@@ -42,6 +44,8 @@ class EventAgentRequest:
     turn_index: int | None = None
     round_index: int | None = None
     turn_started_at: float | None = None
+    scope: Hashable | None = None
+    response: asyncio.Future[EventBatchResult] | None = None
 
 
 class EventAgent:
@@ -55,13 +59,22 @@ class EventAgent:
         self.enabled_tools = set(enabled_tools)
         self.registry = registry or build_default_tool_registry()
         self.chat_client = chat_client
-        self.params = params or AgentParams()
+        self.params = params or EventAgentParams()
         self.kernel = EventKernel(
             self.registry.event_registry,
             argument_fallback=(
                 self._resolve_arguments_with_llm if chat_client is not None else None
             ),
         )
+        self.executor = EventBatchExecutor(
+            self.kernel,
+            max_parallel_events=getattr(self.params, "max_parallel_events", 4),
+            batch_timeout_seconds=getattr(
+                self.params,
+                "batch_timeout_seconds",
+                15.0,
+            ),
+        )
         self._raw_chunks_by_event: dict[str, list[dict[str, Any]]] = {}
 
     async def handle(
@@ -72,19 +85,13 @@ class EventAgent:
         extra_body: dict[str, Any] | None = None,
     ) -> ChatMessage:
         self._raw_chunks_by_event[event.id] = []
-        result = await self.kernel.execute(
-            self.registry.event_request(event),
-            enabled_names=self.enabled_tools,
-            context=ToolExecutionContext(
-                history=history,
-                system_prompt=system_prompt,
-                extra_body=(
-                    self.params.extra_body if extra_body is None else extra_body
-                ),
-            ),
+        batch = await self.handle_many(
+            [event],
+            history=history,
+            system_prompt=system_prompt,
+            extra_body=extra_body,
         )
-        payload = self.registry.tool_payload(result)
-        return self._tool_reply(event, payload)
+        return self.registry.tool_reply(event, batch.results[0])
 
     async def handle_many(
         self,
@@ -92,19 +99,21 @@ class EventAgent:
         history: Sequence[ChatMessage],
         system_prompt: str = "",
         extra_body: dict[str, Any] | None = None,
-    ) -> list[ChatMessage]:
+        scope: Hashable | None = None,
+    ) -> EventBatchResult:
         for event in events:
             self._raw_chunks_by_event[event.id] = []
-        return await asyncio.gather(
-            *[
-                self.handle(
-                    event,
-                    history=history,
-                    system_prompt=system_prompt,
-                    extra_body=extra_body,
-                )
-                for event in events
-            ]
+        return await self.executor.execute(
+            [self.registry.event_request(event) for event in events],
+            enabled_names=self.enabled_tools,
+            context=ToolExecutionContext(
+                history=history,
+                system_prompt=system_prompt,
+                extra_body=(
+                    self.params.extra_body if extra_body is None else extra_body
+                ),
+            ),
+            scope=scope if scope is not None else object(),
         )
 
     def raw_model_chunks(
@@ -218,23 +227,3 @@ class EventAgent:
                 continue
             projected.append(message)
         return projected
-
-    def summarize_replies(self, replies: Sequence[ChatMessage]) -> ChatMessage | None:
-        if not replies:
-            return None
-        return ChatMessage(
-            role="user",
-            content=(
-                "EventAgent results:\n"
-                + "\n".join(reply.content for reply in replies)
-            ),
-            name="event_agent",
-        )
-
-    def _tool_reply(self, event: ToolCallEvent, payload: dict[str, Any]) -> ChatMessage:
-        return ChatMessage(
-            role="tool",
-            content=json.dumps(payload, ensure_ascii=False),
-            name=event.name,
-            tool_call_id=event.id,
-        )

src/agent_lab/application/events/__init__.py

@@ -1,3 +1,4 @@
+from agent_lab.application.events.batch import EventBatchExecutor
 from agent_lab.application.events.kernel import ArgumentFallback, EventKernel
 from agent_lab.application.events.builtin_plugins import (
     CalendarSchedulePort,
@@ -15,11 +16,14 @@ from agent_lab.application.events.models import (
     EventArgumentNormalizer,
     EventArgumentResolution,
     EventArgumentResolver,
+    EventBatchDecision,
+    EventBatchResult,
     EventDefinition,
     EventExecutionContext,
     EventHandler,
     EventRequest,
     EventResult,
+    EventResultMessageFactory,
     EventSource,
     EventStatus,
     ResolvedEventArguments,
@@ -35,6 +39,9 @@ __all__ = [
     "EventArgumentNormalizer",
     "EventArgumentResolution",
     "EventArgumentResolver",
+    "EventBatchExecutor",
+    "EventBatchDecision",
+    "EventBatchResult",
     "EventDefinition",
     "EventExecutionContext",
     "EventHandler",
@@ -42,6 +49,7 @@ __all__ = [
     "EventRegistry",
     "EventRequest",
     "EventResult",
+    "EventResultMessageFactory",
     "EventSource",
     "EventStatus",
     "DeviceVolumePort",

src/agent_lab/application/events/batch.py

@@ -0,0 +1,253 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from collections.abc import Hashable, Iterable, Sequence
+from dataclasses import dataclass
+from typing import Any
+
+from agent_lab.application.events.kernel import EventKernel
+from agent_lab.application.events.models import (
+    EventBatchResult,
+    EventDefinition,
+    EventExecutionContext,
+    EventRequest,
+    EventResult,
+    EventStatus,
+)
+
+
+@dataclass(frozen=True)
+class _BatchItem:
+    key: str
+    request: EventRequest
+
+
+@dataclass(frozen=True)
+class _ExecutionOutcome:
+    result: EventResult
+    batch_timed_out: bool = False
+
+
+class EventBatchExecutor:
+    def __init__(
+        self,
+        kernel: EventKernel,
+        *,
+        max_parallel_events: int = 4,
+        batch_timeout_seconds: float = 15.0,
+    ) -> None:
+        if max_parallel_events < 1:
+            raise ValueError("max_parallel_events must be positive")
+        if batch_timeout_seconds <= 0:
+            raise ValueError("batch_timeout_seconds must be positive")
+        self.kernel = kernel
+        self.max_parallel_events = max_parallel_events
+        self.batch_timeout_seconds = batch_timeout_seconds
+        self._semaphore = asyncio.Semaphore(max_parallel_events)
+        self._conflict_locks: dict[str, asyncio.Lock] = {}
+        self._replay_results: dict[tuple[Hashable, str, str], EventResult] = {}
+
+    async def execute(
+        self,
+        requests: Sequence[EventRequest],
+        *,
+        enabled_names: Iterable[str] | None = None,
+        context: EventExecutionContext | None = None,
+        scope: Hashable,
+    ) -> EventBatchResult:
+        if not requests:
+            return EventBatchResult()
+
+        items: list[_BatchItem] = []
+        indexes_by_key: dict[str, list[int]] = {}
+        coalesced_count = 0
+        for index, request in enumerate(requests):
+            key = self._request_key(request)
+            indexes = indexes_by_key.get(key)
+            if indexes is not None:
+                indexes.append(index)
+                coalesced_count += 1
+                continue
+            indexes_by_key[key] = [index]
+            items.append(_BatchItem(key=key, request=request))
+
+        results_by_key: dict[str, EventResult] = {}
+        replayed_count = 0
+        pending: list[_BatchItem] = []
+        for item in items:
+            replay_key = (scope, item.request.id, item.key)
+            replayed = self._replay_results.get(replay_key)
+            if replayed is None:
+                pending.append(item)
+                continue
+            results_by_key[item.key] = replayed
+            replayed_count += 1
+
+        nonterminal: list[_BatchItem] = []
+        terminal: list[_BatchItem] = []
+        for item in pending:
+            definition = self.kernel.registry.definition(item.request.name)
+            if definition is not None and definition.terminal:
+                terminal.append(item)
+            else:
+                nonterminal.append(item)
+
+        deadline = asyncio.get_running_loop().time() + self.batch_timeout_seconds
+        deadline_exceeded = False
+        for phase in (nonterminal, terminal):
+            outcomes = await asyncio.gather(
+                *[
+                    self._execute_one(
+                        item.request,
+                        enabled_names=enabled_names,
+                        context=context,
+                        deadline=deadline,
+                    )
+                    for item in phase
+                ]
+            )
+            for item, outcome in zip(phase, outcomes, strict=True):
+                results_by_key[item.key] = outcome.result
+                self._replay_results[(scope, item.request.id, item.key)] = (
+                    outcome.result
+                )
+                deadline_exceeded = deadline_exceeded or outcome.batch_timed_out
+
+        ordered: list[EventResult | None] = [None] * len(requests)
+        for key, indexes in indexes_by_key.items():
+            result = results_by_key[key]
+            for index in indexes:
+                ordered[index] = result
+
+        return EventBatchResult(
+            results=tuple(result for result in ordered if result is not None),
+            coalesced_count=coalesced_count,
+            replayed_count=replayed_count,
+            deadline_exceeded=deadline_exceeded,
+        )
+
+    async def _execute_one(
+        self,
+        request: EventRequest,
+        *,
+        enabled_names: Iterable[str] | None,
+        context: EventExecutionContext | None,
+        deadline: float,
+    ) -> _ExecutionOutcome:
+        definition = self.kernel.registry.definition(request.name)
+        remaining = deadline - asyncio.get_running_loop().time()
+        event_timeout = definition.timeout_seconds if definition is not None else None
+        if remaining <= 0:
+            return _ExecutionOutcome(
+                self._failure_result(
+                    request,
+                    definition,
+                    EventStatus.TIMEOUT,
+                    "batch deadline exceeded",
+                ),
+                batch_timed_out=True,
+            )
+        batch_limited = event_timeout is None or remaining <= event_timeout
+        timeout = remaining if event_timeout is None else min(remaining, event_timeout)
+        acquired_locks: list[asyncio.Lock] = []
+        semaphore_acquired = False
+        try:
+            async with asyncio.timeout(timeout):
+                for key in sorted(set(definition.conflict_keys if definition else ())):
+                    lock = self._conflict_locks.setdefault(key, asyncio.Lock())
+                    await lock.acquire()
+                    acquired_locks.append(lock)
+                await self._semaphore.acquire()
+                semaphore_acquired = True
+                result = await self.kernel.execute(
+                    request,
+                    enabled_names=enabled_names,
+                    context=context,
+                )
+                return _ExecutionOutcome(result)
+        except TimeoutError:
+            error = (
+                "batch deadline exceeded"
+                if batch_limited
+                else f"event timed out after {event_timeout:g} seconds"
+            )
+            return _ExecutionOutcome(
+                self._failure_result(
+                    request,
+                    definition,
+                    EventStatus.TIMEOUT,
+                    error,
+                ),
+                batch_timed_out=batch_limited,
+            )
+        except Exception as exc:
+            return _ExecutionOutcome(
+                self._failure_result(
+                    request,
+                    definition,
+                    EventStatus.HANDLER_ERROR,
+                    f"event execution failed: {exc}",
+                )
+            )
+        finally:
+            if semaphore_acquired:
+                self._semaphore.release()
+            for lock in reversed(acquired_locks):
+                lock.release()
+
+    def _failure_result(
+        self,
+        request: EventRequest,
+        definition: EventDefinition | None,
+        status: EventStatus,
+        error: str,
+    ) -> EventResult:
+        metadata: dict[str, Any] = {}
+        if definition is not None:
+            metadata = {
+                "result_policy": definition.result_policy,
+                "confirmation_policy": definition.confirmation_policy,
+                "risk_level": definition.risk_level,
+                "idempotency_key_fields": definition.idempotency_key_fields,
+                "concurrency_class": definition.concurrency_class,
+                "conflict_keys": definition.conflict_keys,
+                "timeout_seconds": definition.timeout_seconds,
+                "terminal": definition.terminal,
+            }
+        return EventResult(
+            event_id=request.id,
+            event_name=request.name,
+            status=status,
+            source=request.source,
+            arguments=self._safe_arguments(request.arguments),
+            raw_arguments=request.raw_arguments,
+            error=error,
+            **metadata,
+        )
+
+    def _request_key(self, request: EventRequest) -> str:
+        try:
+            arguments = json.dumps(
+                request.arguments,
+                ensure_ascii=False,
+                sort_keys=True,
+                separators=(",", ":"),
+                allow_nan=False,
+            )
+        except (TypeError, ValueError):
+            arguments = request.raw_arguments
+        return json.dumps(
+            [request.id, request.name, request.source.value, arguments],
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+
+    def _safe_arguments(self, arguments: dict[str, Any]) -> dict[str, Any]:
+        try:
+            value = json.loads(
+                json.dumps(arguments, ensure_ascii=False, allow_nan=False)
+            )
+        except (TypeError, ValueError):
+            return {}
+        return value if isinstance(value, dict) else {}

src/agent_lab/application/events/builtin_plugins.py

@@ -12,6 +12,7 @@ from agent_lab.application.events.models import (
     EventDefinition,
     EventExecutionContext,
     EventRequest,
+    EventResult,
     ResultPolicy,
     RiskLevel,
 )
@@ -355,6 +356,7 @@ def _calendar_schedule_definition(port: CalendarSchedulePort) -> EventDefinition
         concurrency_class="schedule-write",
         conflict_keys=("calendar.schedule",),
         timeout_seconds=10.0,
+        result_message_factory=_calendar_schedule_message,
     )
 
 
@@ -401,6 +403,13 @@ def _resolve_session_terminate(
     return deepcopy(request.arguments)
 
 
+def _calendar_schedule_message(result: EventResult) -> str:
+    return (
+        f"Scheduled {result.arguments['title']} for "
+        f"{result.arguments['start_at']} ({result.arguments['timezone']})."
+    )
+
+
 def _resolve_device_volume(
     request: EventRequest,
     context: EventExecutionContext,

src/agent_lab/application/events/models.py

@@ -15,6 +15,7 @@ class EventSource(StrEnum):
 
 class EventStatus(StrEnum):
     SUCCESS = "success"
+    TIMEOUT = "timeout"
     UNKNOWN = "unknown"
     DISABLED = "disabled"
     INVALID_ARGUMENTS = "invalid_arguments"
@@ -77,6 +78,7 @@ EventArgumentResolver = Callable[
     dict[str, Any] | EventArgumentResolution,
 ]
 EventArgumentNormalizer = Callable[[dict[str, Any]], dict[str, Any]]
+EventResultMessageFactory = Callable[["EventResult"], str]
 
 
 @dataclass(frozen=True)
@@ -97,6 +99,7 @@ class EventDefinition:
     timeout_seconds: float | None = None
     terminal: bool = False
     normalizer: EventArgumentNormalizer | None = None
+    result_message_factory: EventResultMessageFactory | None = None
 
 
 @dataclass(frozen=True)
@@ -118,3 +121,18 @@ class EventResult:
     conflict_keys: tuple[str, ...] = ()
     timeout_seconds: float | None = None
     terminal: bool = False
+
+
+@dataclass(frozen=True)
+class EventBatchResult:
+    results: tuple[EventResult, ...] = ()
+    coalesced_count: int = 0
+    replayed_count: int = 0
+    deadline_exceeded: bool = False
+
+
+@dataclass(frozen=True)
+class EventBatchDecision:
+    template_messages: tuple[str, ...] = ()
+    llm_follow_up: bool = False
+    terminate: bool = False

src/agent_lab/application/runtime.py

@@ -1,12 +1,22 @@
 import asyncio
-import json
 import logging
 import time
 from collections.abc import AsyncIterator, Callable
 from typing import Any, Protocol
 
-from agent_lab.application.contracts import AgentParams, DebugRunRequest
+from agent_lab.application.contracts import (
+    AgentParams,
+    DebugRunRequest,
+    EventAgentParams,
+)
 from agent_lab.application.event_agent import EventAgent, EventAgentRequest
+from agent_lab.application.events import (
+    EventBatchDecision,
+    EventBatchExecutor,
+    EventBatchResult,
+    EventKernel,
+    EventSource,
+)
 from agent_lab.application.queues import RuntimeQueues
 from agent_lab.application.session_store import SessionStore
 from agent_lab.application.tools import ToolRegistry, build_default_tool_registry
@@ -43,6 +53,7 @@ class DebugRuntime:
         self.session_store = session_store
         self.clock = clock
         self._tasks: list[asyncio.Task[None]] = []
+        self._batch_executors: dict[tuple[int, float], EventBatchExecutor] = {}
 
     def start(self, request: DebugRunRequest) -> RuntimeQueues:
         queues = self.queues or RuntimeQueues()
@@ -161,6 +172,8 @@ class DebugRuntime:
             assistant_content: list[str] = []
             events: list[ToolCallEvent] = []
             tool_replies: list[ChatMessage] = []
+            batch_result: EventBatchResult | None = None
+            decision: EventBatchDecision | None = None
             message_stream_started = False
             message_delta_count = 0
             configured_events = request.event_agent.enabled_tools
@@ -309,13 +322,27 @@ class DebugRuntime:
                 messages.append(assistant_message)
 
             if events:
+                batch_scope = self._event_scope(request, queues, turn_index=1)
+                await self._audit(
+                    queues,
+                    "event_batch_started",
+                    turn_started_at=turn_started_at,
+                    round_index=round_index,
+                    scope=batch_scope,
+                    event_names=[event.name for event in events],
+                    max_parallel_events=request.event_agent.max_parallel_events,
+                    batch_timeout_seconds=request.event_agent.batch_timeout_seconds,
+                )
                 if request.tool_invocation_mode == "chat_agent_tools":
-                    tool_replies = await self._execute_provider_tools(
+                    batch_result, tool_replies = await self._execute_provider_tools(
                         events,
                         enabled_names=request.event_agent.enabled_tools,
+                        params=request.event_agent,
+                        scope=batch_scope,
                     )
                     messages.extend(tool_replies)
                 else:
+                    response = asyncio.get_running_loop().create_future()
                     await queues.events.put(
                         EventAgentRequest(
                             events=events,
@@ -323,8 +350,11 @@ class DebugRuntime:
                             system_prompt=request.event_agent.system_prompt,
                             extra_body=request.event_agent.extra_body,
                             turn_started_at=turn_started_at,
+                            scope=batch_scope,
+                            response=response,
                         )
                     )
+                    batch_result = await response
                     tool_replies = await self._wait_for_tool_replies(
                         queues,
                         [event.id for event in events],
@@ -352,10 +382,28 @@ class DebugRuntime:
                         round_index=round_index,
                         event_names=[event.name for event in events],
                         result_count=len(tool_replies),
-                        result_summary="\n".join(
-                            reply.content for reply in tool_replies
-                        ),
                     )
+                assert batch_result is not None
+                await self._audit_batch_result(
+                    queues,
+                    batch_result,
+                    turn_started_at=turn_started_at,
+                    round_index=round_index,
+                )
+                decision = await self._apply_batch_policy(
+                    queues,
+                    messages,
+                    batch_result,
+                    turn_started_at=turn_started_at,
+                    round_index=round_index,
+                )
+                if (
+                    decision.llm_follow_up
+                    and request.tool_invocation_mode == "dual_agent"
+                ):
+                    summary = self.registry.compact_results_message(batch_result)
+                    if summary is not None:
+                        messages.append(summary)
 
             elapsed_ms = self._elapsed_ms(started_at)
             await queues.output.put(
@@ -385,6 +433,9 @@ class DebugRuntime:
                 break
 
             event_loops += 1
+            assert decision is not None
+            if decision.terminate or not decision.llm_follow_up:
+                break
 
         await self._audit(
             queues,
@@ -425,7 +476,7 @@ class DebugRuntime:
             turn_index += 1
             turn_started_at = next_turn_started_at or self.clock()
             next_turn_started_at = None
-            await self._run_chat_turn(
+            terminated = await self._run_chat_turn(
                 request,
                 queues,
                 messages,
@@ -434,6 +485,17 @@ class DebugRuntime:
                 session_id,
                 turn_started_at,
             )
+            if terminated:
+                await self._audit(
+                    queues,
+                    "session_finished",
+                    turn_started_at=turn_started_at,
+                    session_id=session_id,
+                    turn_index=turn_index,
+                    terminated=True,
+                )
+                await queues.output.put({"type": "done"})
+                return
 
     async def _run_chat_turn(
         self,
@@ -444,7 +506,7 @@ class DebugRuntime:
         turn_index: int,
         session_id: str | None = None,
         turn_started_at: float | None = None,
-    ) -> None:
+    ) -> bool:
         turn_started_at = turn_started_at or self.clock()
         messages.append(user_message)
         self._start_persisted_turn(session_id, turn_index, user_message)
@@ -460,6 +522,7 @@ class DebugRuntime:
         event_loops = 0
         round_index = 0
         deferred_user_messages: list[ChatMessage] = []
+        terminate_session = False
         while True:
             if round_index:
                 await self._drain_input(
@@ -477,6 +540,8 @@ class DebugRuntime:
             saw_event = False
             assistant_content: list[str] = []
             events: list[ToolCallEvent] = []
+            batch_result: EventBatchResult | None = None
+            decision: EventBatchDecision | None = None
             message_stream_started = False
             message_delta_count = 0
             configured_events = request.event_agent.enabled_tools
@@ -642,13 +707,33 @@ class DebugRuntime:
                 )
 
             if events:
+                batch_scope = self._event_scope(
+                    request,
+                    queues,
+                    turn_index=turn_index,
+                )
+                await self._audit(
+                    queues,
+                    "event_batch_started",
+                    turn_started_at=turn_started_at,
+                    session_id=session_id,
+                    turn_index=turn_index,
+                    round_index=round_index,
+                    scope=batch_scope,
+                    event_names=[event.name for event in events],
+                    max_parallel_events=request.event_agent.max_parallel_events,
+                    batch_timeout_seconds=request.event_agent.batch_timeout_seconds,
+                )
                 if request.tool_invocation_mode == "chat_agent_tools":
-                    tool_replies = await self._execute_provider_tools(
+                    batch_result, tool_replies = await self._execute_provider_tools(
                         events,
                         enabled_names=request.event_agent.enabled_tools,
+                        params=request.event_agent,
+                        scope=batch_scope,
                     )
                     messages.extend(tool_replies)
                 else:
+                    response = asyncio.get_running_loop().create_future()
                     await queues.events.put(
                         EventAgentRequest(
                             events=events,
@@ -659,8 +744,11 @@ class DebugRuntime:
                             turn_index=turn_index,
                             round_index=round_index,
                             turn_started_at=turn_started_at,
+                            scope=batch_scope,
+                            response=response,
                         )
                     )
+                    batch_result = await response
                     tool_replies = await self._wait_for_tool_replies(
                         queues,
                         [event.id for event in events],
@@ -692,10 +780,32 @@ class DebugRuntime:
                         round_index=round_index,
                         event_names=[event.name for event in events],
                         result_count=len(tool_replies),
-                        result_summary="\n".join(
-                            reply.content for reply in tool_replies
-                        ),
                     )
+                assert batch_result is not None
+                await self._audit_batch_result(
+                    queues,
+                    batch_result,
+                    turn_started_at=turn_started_at,
+                    session_id=session_id,
+                    turn_index=turn_index,
+                    round_index=round_index,
+                )
+                decision = await self._apply_batch_policy(
+                    queues,
+                    messages,
+                    batch_result,
+                    turn_started_at=turn_started_at,
+                    session_id=session_id,
+                    turn_index=turn_index,
+                    round_index=round_index,
+                )
+                if (
+                    decision.llm_follow_up
+                    and request.tool_invocation_mode == "dual_agent"
+                ):
+                    summary = self.registry.compact_results_message(batch_result)
+                    if summary is not None:
+                        messages.append(summary)
 
             elapsed_ms = self._elapsed_ms(started_at)
             self._append_persisted_usage(
@@ -740,6 +850,12 @@ class DebugRuntime:
                 break
 
             event_loops += 1
+            assert decision is not None
+            if decision.terminate:
+                terminate_session = True
+                break
+            if not decision.llm_follow_up:
+                break
 
         for deferred_message in deferred_user_messages:
             await queues.input.put(deferred_message)
@@ -759,6 +875,7 @@ class DebugRuntime:
                 "round_count": round_index,
             }
         )
+        return terminate_session
 
     async def _consume_events(
         self,
@@ -791,28 +908,35 @@ class DebugRuntime:
                     if tool is not None
                 ],
             )
-            replies = await event_agent.handle_many(
-                request.events,
-                history=request.history,
-                system_prompt=request.system_prompt,
-                extra_body=request.extra_body,
-            )
-            await self._audit(
-                queues,
-                "event_agent_response",
-                turn_started_at=request.turn_started_at,
-                session_id=request.session_id,
-                agent="event_agent",
-                turn_index=request.turn_index,
-                round_index=request.round_index,
-                replies=[reply.model_dump() for reply in replies],
-                raw_model_chunks=event_agent.raw_model_chunks(request.events),
-            )
-            for reply in replies:
-                await queues.input.put(reply)
-            summary = event_agent.summarize_replies(replies)
-            if summary is not None:
-                await queues.input.put(summary)
+            try:
+                batch = await event_agent.handle_many(
+                    request.events,
+                    history=request.history,
+                    system_prompt=request.system_prompt,
+                    extra_body=request.extra_body,
+                    scope=request.scope,
+                )
+                replies = event_agent.registry.tool_replies(request.events, batch)
+                await self._audit(
+                    queues,
+                    "event_agent_response",
+                    turn_started_at=request.turn_started_at,
+                    session_id=request.session_id,
+                    agent="event_agent",
+                    turn_index=request.turn_index,
+                    round_index=request.round_index,
+                    replies=[reply.model_dump() for reply in replies],
+                    raw_model_chunks=event_agent.raw_model_chunks(request.events),
+                )
+                for reply in replies:
+                    await queues.input.put(reply)
+            except BaseException as exc:
+                if request.response is not None and not request.response.done():
+                    request.response.set_exception(exc)
+                raise
+            else:
+                if request.response is not None and not request.response.done():
+                    request.response.set_result(batch)
 
     async def _wait_for_tool_replies(
         self,
@@ -908,25 +1032,124 @@ class DebugRuntime:
         events: list[ToolCallEvent],
         *,
         enabled_names: list[str],
-    ) -> list[ChatMessage]:
-        payloads = await asyncio.gather(
-            *[
-                self.registry.execute_async(
+        params: EventAgentParams,
+        scope: str,
+    ) -> tuple[EventBatchResult, list[ChatMessage]]:
+        executor = self._batch_executor(params)
+        batch = await executor.execute(
+            [
+                self.registry.event_request(
                     event,
-                    enabled_names=enabled_names,
+                    source=EventSource.PROVIDER_RESOLVED,
                 )
                 for event in events
-            ]
+            ],
+            enabled_names=enabled_names,
+            scope=scope,
         )
-        return [
-            ChatMessage(
-                role="tool",
-                content=json.dumps(payload, ensure_ascii=False),
-                name=event.name,
-                tool_call_id=event.id,
+        return batch, self.registry.tool_replies(events, batch)
+
+    def _batch_executor(self, params: EventAgentParams) -> EventBatchExecutor:
+        key = (params.max_parallel_events, params.batch_timeout_seconds)
+        executor = self._batch_executors.get(key)
+        if executor is None:
+            executor = EventBatchExecutor(
+                EventKernel(self.registry.event_registry),
+                max_parallel_events=params.max_parallel_events,
+                batch_timeout_seconds=params.batch_timeout_seconds,
             )
-            for event, payload in zip(events, payloads, strict=True)
-        ]
+            self._batch_executors[key] = executor
+        return executor
+
+    def _event_scope(
+        self,
+        request: DebugRunRequest,
+        queues: RuntimeQueues,
+        *,
+        turn_index: int,
+    ) -> str:
+        session_scope = request.session_id or f"runtime-{id(queues)}"
+        return f"{session_scope}:turn-{turn_index}"
+
+    async def _audit_batch_result(
+        self,
+        queues: RuntimeQueues,
+        batch: EventBatchResult,
+        *,
+        turn_started_at: float | None,
+        session_id: str | None = None,
+        turn_index: int | None = None,
+        round_index: int | None = None,
+    ) -> None:
+        await self._audit(
+            queues,
+            "event_batch_results",
+            turn_started_at=turn_started_at,
+            session_id=session_id,
+            turn_index=turn_index,
+            round_index=round_index,
+            results=[
+                {
+                    "event_id": result.event_id,
+                    "event_name": result.event_name,
+                    "status": result.status.value,
+                    "error": result.error,
+                    "result_policy": result.result_policy.value,
+                    "terminal": result.terminal,
+                }
+                for result in batch.results
+            ],
+            timeout_count=sum(
+                result.status.value == "timeout" for result in batch.results
+            ),
+            coalesced_count=batch.coalesced_count,
+            replayed_count=batch.replayed_count,
+            deadline_exceeded=batch.deadline_exceeded,
+        )
+
+    async def _apply_batch_policy(
+        self,
+        queues: RuntimeQueues,
+        messages: list[ChatMessage],
+        batch: EventBatchResult,
+        *,
+        turn_started_at: float | None,
+        session_id: str | None = None,
+        turn_index: int | None = None,
+        round_index: int | None = None,
+    ) -> EventBatchDecision:
+        decision = self.registry.batch_decision(batch)
+        for content in decision.template_messages:
+            message = ChatMessage(role="assistant", content=content)
+            messages.append(message)
+            self._append_persisted_message(session_id, turn_index or 0, message)
+            await queues.output.put({"type": "message_delta", "content": content})
+        await self._audit(
+            queues,
+            "event_policy_decision",
+            turn_started_at=turn_started_at,
+            session_id=session_id,
+            turn_index=turn_index,
+            round_index=round_index,
+            template_count=len(decision.template_messages),
+            llm_follow_up=decision.llm_follow_up,
+            terminate=decision.terminate,
+        )
+        if decision.terminate:
+            await self._audit(
+                queues,
+                "terminal_completed",
+                turn_started_at=turn_started_at,
+                session_id=session_id,
+                turn_index=turn_index,
+                round_index=round_index,
+                event_ids=[
+                    result.event_id
+                    for result in batch.results
+                    if result.terminal and result.status.value == "success"
+                ],
+            )
+        return decision
 
     def _chat_messages_for_round(
         self,

src/agent_lab/application/tools.py

@@ -1,4 +1,5 @@
 import inspect
+import json
 from collections.abc import Awaitable, Callable, Iterable, Sequence
 from copy import deepcopy
 from dataclasses import dataclass
@@ -9,11 +10,14 @@ from agent_lab.application.events import (
     CalendarSchedulePort,
     DeviceVolumePort,
     EventDefinition,
+    EventBatchDecision,
+    EventBatchResult,
     EventExecutionContext,
     EventKernel,
     EventRegistry,
     EventRequest,
     EventResult,
+    EventResultMessageFactory,
     EventSource,
     EventStatus,
     ResultPolicy,
@@ -58,6 +62,7 @@ class ToolDefinition:
     timeout_seconds: float | None = None
     terminal: bool = False
     normalizer: ToolArgumentNormalizer | None = None
+    result_message_factory: EventResultMessageFactory | None = None
 
 
 class ToolRegistry:
@@ -197,6 +202,79 @@ class ToolRegistry:
             error = error.replace("EventKernel.execute", "execute_async")
         return {"tool": result.event_name, "error": error}
 
+    def tool_reply(
+        self,
+        event: ToolCallEvent,
+        result: EventResult,
+    ) -> ChatMessage:
+        return ChatMessage(
+            role="tool",
+            content=json.dumps(self.tool_payload(result), ensure_ascii=False),
+            name=event.name,
+            tool_call_id=event.id,
+        )
+
+    def tool_replies(
+        self,
+        events: Sequence[ToolCallEvent],
+        batch: EventBatchResult,
+    ) -> list[ChatMessage]:
+        return [
+            self.tool_reply(event, result)
+            for event, result in zip(events, batch.results, strict=True)
+        ]
+
+    def batch_decision(self, batch: EventBatchResult) -> EventBatchDecision:
+        templates: list[str] = []
+        llm_follow_up = False
+        terminate = False
+        for result in batch.results:
+            if result.status is not EventStatus.SUCCESS:
+                llm_follow_up = True
+                continue
+            if result.result_policy is ResultPolicy.TERMINATE:
+                terminate = True
+                continue
+            if result.result_policy is ResultPolicy.LLM_FOLLOW_UP:
+                llm_follow_up = True
+                continue
+            if result.result_policy is not ResultPolicy.TEMPLATE_FOLLOW_UP:
+                continue
+            definition = self.event_registry.definition(result.event_name)
+            factory = definition.result_message_factory if definition else None
+            if factory is None:
+                continue
+            try:
+                message = factory(result).strip()
+            except Exception:
+                llm_follow_up = True
+                continue
+            if message:
+                templates.append(message)
+        return EventBatchDecision(
+            template_messages=tuple(templates),
+            llm_follow_up=llm_follow_up and not terminate,
+            terminate=terminate,
+        )
+
+    def compact_results_message(
+        self,
+        batch: EventBatchResult,
+    ) -> ChatMessage | None:
+        if not batch.results:
+            return None
+        return ChatMessage(
+            role="user",
+            content=(
+                "EventAgent results:\n"
+                + "\n".join(
+                    json.dumps(self.tool_payload(result), ensure_ascii=False)
+                    for result in batch.results
+                )
+            ),
+            name="event_agent",
+        )
+
     def _to_event_definition(self, definition: ToolDefinition) -> EventDefinition:
         def resolver(
             request: EventRequest,
@@ -237,6 +315,7 @@ class ToolRegistry:
             conflict_keys=definition.conflict_keys,
             timeout_seconds=definition.timeout_seconds,
             terminal=definition.terminal,
+            result_message_factory=definition.result_message_factory,
         )
 
     def _tool_event(self, request: EventRequest) -> ToolCallEvent:

tests/test_debug_runtime.py

@@ -9,6 +9,7 @@ import pytest
 
 from agent_lab.application.contracts import AgentParams, DebugRunRequest, EventAgentParams
 from agent_lab.application.event_agent import EventAgentRequest
+from agent_lab.application.events import ResultPolicy
 from agent_lab.application.runtime import DebugRuntime
 from agent_lab.application.tools import ToolDefinition, ToolExecutionContext, ToolRegistry
 from agent_lab.domain.events import ToolCallEvent
@@ -647,9 +648,12 @@ async def test_runtime_emits_audit_events_and_backend_logs(caplog):
         "chat_agent_request",
         "chat_event_detected",
         "chat_agent_response",
+        "event_batch_started",
         "event_agent_request",
         "event_agent_response",
         "event_agent_completed",
+        "event_batch_results",
+        "event_policy_decision",
         "chat_round_finished",
         "chat_round_started",
         "chat_agent_request",
@@ -1427,6 +1431,9 @@ async def test_runtime_emits_round_stats_for_each_chat_call_in_event_handoff():
             2.05,
             2.06,
             2.07,
+            2.08,
+            2.09,
+            2.1,
             2.25,
             2.26,
             3.0,
@@ -2222,3 +2229,266 @@ async def test_dual_mode_session_main_path_ignores_text_event_after_budget_exhau
         if message["type"] == "event"
     ] == ["session-accepted"]
     assert outputs[-1]["type"] == "turn_completed"
+
+
+def _policy_request(
+    *,
+    enabled_tools: list[str],
+    mode: str = "dual_agent",
+    session_id: str | None = None,
+) -> DebugRunRequest:
+    return DebugRunRequest(
+        session_id=session_id,
+        user_message="exercise event result policy",
+        system_prompts=[],
+        pre_messages=[],
+        chat_agent=AgentParams(model="chat-model"),
+        event_agent=EventAgentParams(
+            enabled_tools=enabled_tools,
+            max_event_loops=1,
+        ),
+        tool_invocation_mode=mode,
+    )
+
+
+@pytest.mark.asyncio
+async def test_dual_silent_success_does_not_enqueue_legacy_summary_or_call_chat_again():
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="example.silent",
+                description="Complete silently.",
+                parameters={"type": "object"},
+                handler=lambda event: {"tool": event.name, "status": "applied"},
+                result_policy=ResultPolicy.SILENT_SUCCESS,
+            )
+        ]
+    )
+    client = ScriptedChatClient(
+        [
+            [
+                StreamItem.message_delta("Applying it now."),
+                StreamItem.text_event(_tool_call("silent-1", name="example.silent")),
+            ]
+        ]
+    )
+
+    outputs = await _collect_outputs(
+        DebugRuntime(client, registry=registry).run(
+            _policy_request(enabled_tools=["example.silent"])
+        )
+    )
+
+    assert client.calls == 1
+    assert outputs[-1] == {"type": "done"}
+    assert not any(
+        message.get("details", {}).get("result_summary")
+        for message in outputs
+        if message.get("event") == "event_agent_completed"
+    )
+
+
+@pytest.mark.asyncio
+async def test_direct_template_success_emits_plugin_confirmation_without_model_call():
+    schedule = ToolCallEvent(
+        id="schedule-1",
+        name="calendar.schedule.create",
+        arguments={
+            "title": "Design review",
+            "start_at": "2026-07-14T09:30:00+08:00",
+            "timezone": "Asia/Shanghai",
+        },
+        raw_arguments=(
+            '{"title":"Design review","start_at":"2026-07-14T09:30:00+08:00",'
+            '"timezone":"Asia/Shanghai"}'
+        ),
+    )
+    client = ScriptedChatClient(
+        [
+            [
+                StreamItem.message_delta("I will add it."),
+                StreamItem.provider_tool_call(schedule),
+            ]
+        ]
+    )
+
+    outputs = await _collect_outputs(
+        DebugRuntime(client).run(
+            _policy_request(
+                enabled_tools=["calendar.schedule.create"],
+                mode="chat_agent_tools",
+            )
+        )
+    )
+
+    assert client.calls == 1
+    assert [
+        message["content"]
+        for message in outputs
+        if message["type"] == "message_delta"
+    ] == [
+        "I will add it.",
+        "Scheduled Design review for 2026-07-14T09:30:00+08:00 (Asia/Shanghai).",
+    ]
+    assert outputs[-1] == {"type": "done"}
+
+
+@pytest.mark.asyncio
+async def test_direct_silent_failure_requests_exactly_one_corrective_follow_up():
+    def fail(event: ToolCallEvent) -> dict[str, Any]:
+        raise RuntimeError("device offline")
+
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="example.silent",
+                description="Complete silently on success.",
+                parameters={"type": "object"},
+                handler=fail,
+                result_policy=ResultPolicy.SILENT_SUCCESS,
+            )
+        ]
+    )
+    client = ScriptedChatClient(
+        [
+            [
+                StreamItem.message_delta("Trying it."),
+                StreamItem.provider_tool_call(
+                    _tool_call("silent-failure", name="example.silent")
+                ),
+            ],
+            [StreamItem.message_delta("I could not apply that change.")],
+        ]
+    )
+
+    outputs = await _collect_outputs(
+        DebugRuntime(client, registry=registry).run(
+            _policy_request(
+                enabled_tools=["example.silent"],
+                mode="chat_agent_tools",
+            )
+        )
+    )
+
+    assert client.calls == 2
+    assert [
+        message["content"]
+        for message in outputs
+        if message["type"] == "message_delta"
+    ] == ["Trying it.", "I could not apply that change."]
+
+
+@pytest.mark.asyncio
+async def test_mixed_builtin_batch_emits_template_then_terminates_without_llm():
+    calls = [
+        ToolCallEvent(
+            id="terminate-1",
+            name="session.terminate",
+            arguments={},
+            raw_arguments="{}",
+        ),
+        ToolCallEvent(
+            id="schedule-1",
+            name="calendar.schedule.create",
+            arguments={
+                "title": "Design review",
+                "start_at": "2026-07-14T09:30:00+08:00",
+                "timezone": "Asia/Shanghai",
+            },
+            raw_arguments=(
+                '{"title":"Design review","start_at":"2026-07-14T09:30:00+08:00",'
+                '"timezone":"Asia/Shanghai"}'
+            ),
+        ),
+        ToolCallEvent(
+            id="search-1",
+            name="knowledge.web.search",
+            arguments={"query": "event batch executors"},
+            raw_arguments='{"query":"event batch executors"}',
+        ),
+        ToolCallEvent(
+            id="volume-1",
+            name="device.volume.adjust",
+            arguments={"mode": "absolute", "value": 30},
+            raw_arguments='{"mode":"absolute","value":30}',
+        ),
+    ]
+    client = ScriptedChatClient(
+        [
+            [
+                StreamItem.message_delta("Goodbye, I will finish those first."),
+                *[StreamItem.provider_tool_call(call) for call in calls],
+            ]
+        ]
+    )
+
+    outputs = await _collect_outputs(
+        DebugRuntime(client).run(
+            _policy_request(
+                enabled_tools=[call.name for call in calls],
+                mode="chat_agent_tools",
+            )
+        )
+    )
+
+    assert client.calls == 1
+    assert [
+        message["message"]["tool_call_id"]
+        for message in outputs
+        if message["type"] == "tool_result"
+    ] == [call.id for call in calls]
+    assert [
+        message["content"]
+        for message in outputs
+        if message["type"] == "message_delta"
+    ][-1] == (
+        "Scheduled Design review for 2026-07-14T09:30:00+08:00 "
+        "(Asia/Shanghai)."
+    )
+    decision = next(
+        message
+        for message in outputs
+        if message.get("event") == "event_policy_decision"
+    )
+    assert decision["details"]["terminate"] is True
+    assert decision["details"]["llm_follow_up"] is False
+    assert any(
+        message.get("event") == "terminal_completed" for message in outputs
+    )
+    assert outputs[-1] == {"type": "done"}
+
+
+@pytest.mark.asyncio
+async def test_reusable_session_terminate_completes_turn_and_session_task():
+    client = ScriptedChatClient(
+        [
+            [
+                StreamItem.message_delta("Goodbye."),
+                StreamItem.provider_tool_call(
+                    ToolCallEvent(
+                        id="terminate-session",
+                        name="session.terminate",
+                        arguments={},
+                        raw_arguments="{}",
+                    )
+                ),
+            ]
+        ]
+    )
+    runtime = DebugRuntime(client)
+    queues = runtime.start_session(
+        _policy_request(
+            enabled_tools=["session.terminate"],
+            mode="chat_agent_tools",
+            session_id="reusable-session",
+        )
+    )
+    outputs: list[dict[str, Any]] = []
+
+    while not any(message["type"] == "done" for message in outputs):
+        outputs.append(await asyncio.wait_for(queues.output.get(), timeout=1))
+
+    await asyncio.wait_for(runtime._wait_for_tasks(), timeout=1)
+
+    business_types = [message["type"] for message in _without_audit(outputs)]
+    assert business_types[-2:] == ["turn_completed", "done"]

tests/test_event_agent.py

@@ -4,7 +4,7 @@ from collections.abc import AsyncIterator
 
 import pytest
 
-from agent_lab.application.contracts import AgentParams
+from agent_lab.application.contracts import AgentParams, EventAgentParams
 from agent_lab.application.event_agent import EventAgent
 from agent_lab.application.tools import ToolDefinition, ToolExecutionContext, ToolRegistry
 from agent_lab.domain.events import ToolCallEvent
@@ -291,17 +291,21 @@ async def test_event_agent_handle_many_starts_independent_handlers_concurrently(
         for index, name in enumerate(("event.first", "event.second"), start=1)
     ]
 
-    replies = await asyncio.wait_for(
+    batch = await asyncio.wait_for(
         EventAgent(
             enabled_tools=[event.name for event in events],
             registry=registry,
+            params=EventAgentParams(max_parallel_events=2),
         ).handle_many(events, history=[]),
         timeout=0.5,
     )
 
     assert started == ["event.first", "event.second"]
-    assert [reply.name for reply in replies] == ["event.first", "event.second"]
-    assert [json.loads(reply.content) for reply in replies] == [
+    assert [result.event_name for result in batch.results] == [
+        "event.first",
+        "event.second",
+    ]
+    assert [result.payload for result in batch.results] == [
         {"tool": "event.first"},
         {"tool": "event.second"},
     ]

tests/test_event_batch.py

@@ -0,0 +1,349 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from typing import Any
+
+import pytest
+
+import agent_lab.application.events as events_module
+from agent_lab.application.events import (
+    EventDefinition,
+    EventKernel,
+    EventRegistry,
+    EventRequest,
+    EventStatus,
+)
+
+
+def _executor_type():
+    executor_type = getattr(events_module, "EventBatchExecutor", None)
+    assert executor_type is not None, "EventBatchExecutor is not implemented"
+    return executor_type
+
+
+def _definition(
+    name: str,
+    handler: Callable[[EventRequest], Any],
+    *,
+    conflict_keys: tuple[str, ...] = (),
+    timeout_seconds: float | None = None,
+    terminal: bool = False,
+) -> EventDefinition:
+    return EventDefinition(
+        name=name,
+        description=f"Execute {name}.",
+        parameters={"type": "object"},
+        handler=handler,
+        conflict_keys=conflict_keys,
+        timeout_seconds=timeout_seconds,
+        terminal=terminal,
+    )
+
+
+def _request(event_id: str, name: str, arguments: dict[str, Any] | None = None):
+    return EventRequest(
+        id=event_id,
+        name=name,
+        arguments=arguments or {},
+    )
+
+
+def _executor(
+    definitions: list[EventDefinition],
+    *,
+    max_parallel_events: int = 4,
+    batch_timeout_seconds: float = 1.0,
+):
+    return _executor_type()(
+        EventKernel(EventRegistry(definitions)),
+        max_parallel_events=max_parallel_events,
+        batch_timeout_seconds=batch_timeout_seconds,
+    )
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_bounds_global_parallelism():
+    active = 0
+    max_active = 0
+
+    async def handler(request: EventRequest) -> dict[str, Any]:
+        nonlocal active, max_active
+        active += 1
+        max_active = max(max_active, active)
+        await asyncio.sleep(0.02)
+        active -= 1
+        return {"event_id": request.id}
+
+    executor = _executor(
+        [_definition("example.work", handler)],
+        max_parallel_events=2,
+    )
+
+    batch = await executor.execute(
+        [_request(str(index), "example.work") for index in range(4)],
+        scope="session-1:turn-1",
+    )
+
+    assert max_active == 2
+    assert [result.status for result in batch.results] == [
+        EventStatus.SUCCESS,
+    ] * 4
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_preserves_input_order_when_completion_order_differs():
+    release_first = asyncio.Event()
+    second_finished = asyncio.Event()
+
+    async def handler(request: EventRequest) -> dict[str, Any]:
+        if request.id == "first":
+            await release_first.wait()
+        else:
+            second_finished.set()
+        return {"event_id": request.id}
+
+    executor = _executor([_definition("example.work", handler)])
+    task = asyncio.create_task(
+        executor.execute(
+            [
+                _request("first", "example.work"),
+                _request("second", "example.work"),
+            ],
+            scope="session-1:turn-1",
+        )
+    )
+    await asyncio.wait_for(second_finished.wait(), timeout=0.2)
+    release_first.set()
+
+    batch = await asyncio.wait_for(task, timeout=0.2)
+
+    assert [result.event_id for result in batch.results] == ["first", "second"]
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_serializes_shared_conflict_keys():
+    active = 0
+    max_active = 0
+
+    async def handler(request: EventRequest) -> dict[str, Any]:
+        nonlocal active, max_active
+        active += 1
+        max_active = max(max_active, active)
+        await asyncio.sleep(0.02)
+        active -= 1
+        return {"event_id": request.id}
+
+    executor = _executor(
+        [
+            _definition(
+                "example.write",
+                handler,
+                conflict_keys=("shared-resource",),
+            )
+        ]
+    )
+
+    await executor.execute(
+        [
+            _request("first", "example.write"),
+            _request("second", "example.write"),
+        ],
+        scope="session-1:turn-1",
+    )
+
+    assert max_active == 1
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_isolates_per_event_timeout_from_successful_sibling():
+    async def slow_handler(request: EventRequest) -> dict[str, Any]:
+        await asyncio.sleep(1)
+        return {"event_id": request.id}
+
+    executor = _executor(
+        [
+            _definition(
+                "example.slow",
+                slow_handler,
+                timeout_seconds=0.01,
+            ),
+            _definition(
+                "example.fast",
+                lambda request: {"event_id": request.id},
+            ),
+        ],
+        batch_timeout_seconds=0.2,
+    )
+
+    batch = await executor.execute(
+        [
+            _request("slow", "example.slow"),
+            _request("fast", "example.fast"),
+        ],
+        scope="session-1:turn-1",
+    )
+
+    assert [result.status for result in batch.results] == [
+        EventStatus.TIMEOUT,
+        EventStatus.SUCCESS,
+    ]
+    assert batch.deadline_exceeded is False
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_applies_overall_deadline_to_waiting_work():
+    async def blocked_handler(request: EventRequest) -> dict[str, Any]:
+        await asyncio.Event().wait()
+        return {"event_id": request.id}
+
+    executor = _executor(
+        [_definition("example.blocked", blocked_handler)],
+        max_parallel_events=1,
+        batch_timeout_seconds=0.02,
+    )
+
+    batch = await executor.execute(
+        [
+            _request("first", "example.blocked"),
+            _request("second", "example.blocked"),
+        ],
+        scope="session-1:turn-1",
+    )
+
+    assert [result.status for result in batch.results] == [
+        EventStatus.TIMEOUT,
+        EventStatus.TIMEOUT,
+    ]
+    assert batch.deadline_exceeded is True
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_coalesces_exact_duplicates_inside_one_batch():
+    calls = 0
+
+    def handler(request: EventRequest) -> dict[str, Any]:
+        nonlocal calls
+        calls += 1
+        return {"event_id": request.id, "arguments": request.arguments}
+
+    executor = _executor([_definition("example.write", handler)])
+
+    batch = await executor.execute(
+        [
+            _request("same", "example.write", {"a": 1, "b": 2}),
+            _request("same", "example.write", {"b": 2, "a": 1}),
+        ],
+        scope="session-1:turn-1",
+    )
+
+    assert calls == 1
+    assert batch.coalesced_count == 1
+    assert batch.results[0] == batch.results[1]
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_replay_key_includes_scope_event_id_and_request():
+    calls = 0
+
+    def handler(request: EventRequest) -> dict[str, Any]:
+        nonlocal calls
+        calls += 1
+        return {"call": calls, "arguments": request.arguments}
+
+    executor = _executor([_definition("example.write", handler)])
+    original = _request("same", "example.write", {"a": 1, "b": 2})
+    canonical_replay = _request(
+        "same",
+        "example.write",
+        {"b": 2, "a": 1},
+    )
+
+    first = await executor.execute([original], scope="session-1:turn-1")
+    replay = await executor.execute(
+        [canonical_replay],
+        scope="session-1:turn-1",
+    )
+    changed_request = await executor.execute(
+        [_request("same", "example.write", {"a": 2})],
+        scope="session-1:turn-1",
+    )
+    changed_id = await executor.execute(
+        [_request("other", "example.write", {"a": 1, "b": 2})],
+        scope="session-1:turn-1",
+    )
+    changed_scope = await executor.execute(
+        [original],
+        scope="session-1:turn-2",
+    )
+
+    assert first.results[0] == replay.results[0]
+    assert replay.replayed_count == 1
+    assert changed_request.replayed_count == 0
+    assert changed_id.replayed_count == 0
+    assert changed_scope.replayed_count == 0
+    assert calls == 4
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_runs_terminal_events_after_all_nonterminal_results():
+    observed: list[str] = []
+
+    async def nonterminal(request: EventRequest) -> dict[str, Any]:
+        observed.append(f"start:{request.id}")
+        await asyncio.sleep(0.01)
+        observed.append(f"finish:{request.id}")
+        return {"event_id": request.id}
+
+    def terminal(request: EventRequest) -> dict[str, Any]:
+        observed.append(f"terminal:{request.id}")
+        return {"event_id": request.id}
+
+    executor = _executor(
+        [
+            _definition("example.work", nonterminal),
+            _definition("example.terminate", terminal, terminal=True),
+        ]
+    )
+
+    batch = await executor.execute(
+        [
+            _request("terminate", "example.terminate"),
+            _request("first", "example.work"),
+            _request("second", "example.work"),
+        ],
+        scope="session-1:turn-1",
+    )
+
+    terminal_index = observed.index("terminal:terminate")
+    assert observed.index("finish:first") < terminal_index
+    assert observed.index("finish:second") < terminal_index
+    assert [result.event_id for result in batch.results] == [
+        "terminate",
+        "first",
+        "second",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_does_not_swallow_external_cancellation():
+    started = asyncio.Event()
+
+    async def handler(request: EventRequest) -> dict[str, Any]:
+        started.set()
+        await asyncio.Event().wait()
+        return {"event_id": request.id}
+
+    executor = _executor([_definition("example.blocked", handler)])
+    task = asyncio.create_task(
+        executor.execute(
+            [_request("blocked", "example.blocked")],
+            scope="session-1:turn-1",
+        )
+    )
+    await asyncio.wait_for(started.wait(), timeout=0.2)
+
+    task.cancel()
+
+    with pytest.raises(asyncio.CancelledError):
+        await task

tests/test_websocket_api.py

@@ -160,9 +160,23 @@ def test_agent_params_defaults_include_extra_body_and_event_agent_defaults_to_on
     }
     assert event_params.extra_body == chat_params.extra_body
     assert event_params.max_event_loops == 1
+    assert event_params.max_parallel_events == 4
+    assert event_params.batch_timeout_seconds == 15.0
     assert event_params.system_prompt == ""
 
 
+@pytest.mark.parametrize("value", [0, 17])
+def test_event_agent_params_rejects_out_of_range_parallelism(value: int):
+    with pytest.raises(ValidationError, match="max_parallel_events"):
+        EventAgentParams(max_parallel_events=value)
+
+
+@pytest.mark.parametrize("value", [0, -1, 121])
+def test_event_agent_params_rejects_invalid_batch_timeout(value: float):
+    with pytest.raises(ValidationError, match="batch_timeout_seconds"):
+        EventAgentParams(batch_timeout_seconds=value)
+
+
 def test_debug_run_request_defaults_to_dual_agent_tool_invocation_mode():
     request = DebugRunRequest.model_validate(_request_payload())