fix: harden batch replay and terminal safety

Problem:
Canonical batch replay, cross-round scope, terminal grace, and tool-only farewell behavior had correctness and lifecycle gaps under duplicate, concurrent, and timeout paths.

Risk:
Thread work may complete after timeout. Production side-effect ports must remain idempotent and tolerate late completion.

src/agent_lab/application/event_agent.py

@@ -103,18 +103,23 @@ class EventAgent:
     ) -> EventBatchResult:
         for event in events:
             self._raw_chunks_by_event[event.id] = []
-        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
+        resolved_scope = scope if scope is not None else object()
+        try:
+            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(),
-        )
+                scope=resolved_scope,
+            )
+        finally:
+            if scope is not None:
+                self.executor.release_scope(resolved_scope)
 
     def raw_model_chunks(
         self,

src/agent_lab/application/events/batch.py

@@ -2,8 +2,9 @@ from __future__ import annotations
 
 import asyncio
 import json
+from collections import OrderedDict
 from collections.abc import Hashable, Iterable, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from typing import Any
 
 from agent_lab.application.events.kernel import EventKernel
@@ -36,17 +37,31 @@ class EventBatchExecutor:
         *,
         max_parallel_events: int = 4,
         batch_timeout_seconds: float = 15.0,
+        terminal_grace_seconds: float = 2.0,
+        max_replay_entries: int = 100,
     ) -> 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")
+        if terminal_grace_seconds <= 0:
+            raise ValueError("terminal_grace_seconds must be positive")
+        if max_replay_entries < 1:
+            raise ValueError("max_replay_entries must be positive")
         self.kernel = kernel
         self.max_parallel_events = max_parallel_events
         self.batch_timeout_seconds = batch_timeout_seconds
+        self.terminal_grace_seconds = terminal_grace_seconds
+        self.max_replay_entries = max_replay_entries
         self._semaphore = asyncio.Semaphore(max_parallel_events)
         self._conflict_locks: dict[str, asyncio.Lock] = {}
-        self._replay_results: dict[tuple[Hashable, str, str], EventResult] = {}
+        self._replay_results: OrderedDict[
+            tuple[Hashable, str, str], EventResult
+        ] = OrderedDict()
+        self._inflight: dict[
+            tuple[Hashable, str, str], asyncio.Task[_ExecutionOutcome]
+        ] = {}
+        self._state_lock = asyncio.Lock()
 
     async def execute(
         self,
@@ -63,7 +78,8 @@ class EventBatchExecutor:
         indexes_by_key: dict[str, list[int]] = {}
         coalesced_count = 0
         for index, request in enumerate(requests):
-            key = self._request_key(request)
+            definition = self.kernel.registry.definition(request.name)
+            key = self._request_key(request, definition)
             indexes = indexes_by_key.get(key)
             if indexes is not None:
                 indexes.append(index)
@@ -72,53 +88,63 @@ class EventBatchExecutor:
             indexes_by_key[key] = [index]
             items.append(_BatchItem(key=key, request=request))
 
-        results_by_key: dict[str, EventResult] = {}
+        outcomes_by_key: dict[str, _ExecutionOutcome] = {}
         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:
+        for item in items:
             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
+        loop = asyncio.get_running_loop()
+        batch_deadline = loop.time() + self.batch_timeout_seconds
         deadline_exceeded = False
-        for phase in (nonterminal, terminal):
-            outcomes = await asyncio.gather(
+        phases = (
+            (nonterminal, batch_deadline, True),
+            (terminal, None, False),
+        )
+        for phase, deadline, count_deadline_exceeded in phases:
+            if deadline is None:
+                deadline = loop.time() + self.terminal_grace_seconds
+            phase_results = await asyncio.gather(
                 *[
-                    self._execute_one(
-                        item.request,
+                    self._execute_replayable(
+                        item,
+                        scope=scope,
                         enabled_names=enabled_names,
                         context=context,
                         deadline=deadline,
+                        count_deadline_exceeded=count_deadline_exceeded,
                     )
                     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
-                )
+            for item, (outcome, replayed) in zip(
+                phase,
+                phase_results,
+                strict=True,
+            ):
+                outcomes_by_key[item.key] = outcome
+                replayed_count += int(replayed)
                 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]
+            result = outcomes_by_key[key].result
+            primary_id = requests[indexes[0]].id
             for index in indexes:
-                ordered[index] = result
+                ordered[index] = self._clone_result(
+                    result,
+                    requests[index],
+                    deduplicated_from=(
+                        primary_id if requests[index].id != primary_id else None
+                    ),
+                )
+
+        await self._cache_ordered_results(scope, requests, indexes_by_key, ordered)
 
         return EventBatchResult(
             results=tuple(result for result in ordered if result is not None),
@@ -127,6 +153,124 @@ class EventBatchExecutor:
             deadline_exceeded=deadline_exceeded,
         )
 
+    def release_scope(self, scope: Hashable) -> None:
+        for key in tuple(self._replay_results):
+            if key[0] == scope:
+                self._replay_results.pop(key, None)
+
+    async def _execute_replayable(
+        self,
+        item: _BatchItem,
+        *,
+        scope: Hashable,
+        enabled_names: Iterable[str] | None,
+        context: EventExecutionContext | None,
+        deadline: float,
+        count_deadline_exceeded: bool,
+    ) -> tuple[_ExecutionOutcome, bool]:
+        replay_key = (scope, item.request.id, item.key)
+        async with self._state_lock:
+            replayed = self._replay_results.get(replay_key)
+            if replayed is not None:
+                self._replay_results.move_to_end(replay_key)
+                return self._outcome_from_cached(replayed), True
+            task = self._inflight.get(replay_key)
+            reused = task is not None
+            if task is None:
+                task = asyncio.create_task(
+                    self._run_and_store(
+                        replay_key,
+                        item.request,
+                        enabled_names=enabled_names,
+                        context=context,
+                        deadline=deadline,
+                        count_deadline_exceeded=count_deadline_exceeded,
+                    )
+                )
+                self._inflight[replay_key] = task
+        return await asyncio.shield(task), reused
+
+    async def _run_and_store(
+        self,
+        replay_key: tuple[Hashable, str, str],
+        request: EventRequest,
+        *,
+        enabled_names: Iterable[str] | None,
+        context: EventExecutionContext | None,
+        deadline: float,
+        count_deadline_exceeded: bool,
+    ) -> _ExecutionOutcome:
+        outcome: _ExecutionOutcome | None = None
+        try:
+            outcome = await self._execute_one(
+                request,
+                enabled_names=enabled_names,
+                context=context,
+                deadline=deadline,
+                count_deadline_exceeded=count_deadline_exceeded,
+            )
+        finally:
+            async with self._state_lock:
+                if self._inflight.get(replay_key) is asyncio.current_task():
+                    self._inflight.pop(replay_key, None)
+                if outcome is not None:
+                    self._store_replay_locked(replay_key, outcome.result)
+        assert outcome is not None
+        return outcome
+
+    async def _cache_ordered_results(
+        self,
+        scope: Hashable,
+        requests: Sequence[EventRequest],
+        indexes_by_key: dict[str, list[int]],
+        ordered: list[EventResult | None],
+    ) -> None:
+        async with self._state_lock:
+            for key, indexes in indexes_by_key.items():
+                for index in indexes:
+                    result = ordered[index]
+                    assert result is not None
+                    self._store_replay_locked(
+                        (scope, requests[index].id, key),
+                        result,
+                    )
+
+    def _store_replay_locked(
+        self,
+        replay_key: tuple[Hashable, str, str],
+        result: EventResult,
+    ) -> None:
+        self._replay_results[replay_key] = result
+        self._replay_results.move_to_end(replay_key)
+        while len(self._replay_results) > self.max_replay_entries:
+            self._replay_results.popitem(last=False)
+
+    def _outcome_from_cached(self, result: EventResult) -> _ExecutionOutcome:
+        return _ExecutionOutcome(
+            result,
+            batch_timed_out=(
+                result.status is EventStatus.TIMEOUT
+                and result.error == "batch deadline exceeded"
+            ),
+        )
+
+    def _clone_result(
+        self,
+        result: EventResult,
+        request: EventRequest,
+        *,
+        deduplicated_from: str | None,
+    ) -> EventResult:
+        payload = dict(result.payload)
+        if deduplicated_from is not None:
+            payload["deduplicated_from"] = deduplicated_from
+        return replace(
+            result,
+            event_id=request.id,
+            raw_arguments=request.raw_arguments,
+            payload=payload,
+        )
+
     async def _execute_one(
         self,
         request: EventRequest,
@@ -134,6 +278,7 @@ class EventBatchExecutor:
         enabled_names: Iterable[str] | None,
         context: EventExecutionContext | None,
         deadline: float,
+        count_deadline_exceeded: bool,
     ) -> _ExecutionOutcome:
         definition = self.kernel.registry.definition(request.name)
         remaining = deadline - asyncio.get_running_loop().time()
@@ -146,7 +291,7 @@ class EventBatchExecutor:
                     EventStatus.TIMEOUT,
                     "batch deadline exceeded",
                 ),
-                batch_timed_out=True,
+                batch_timed_out=count_deadline_exceeded,
             )
         batch_limited = event_timeout is None or remaining <= event_timeout
         timeout = remaining if event_timeout is None else min(remaining, event_timeout)
@@ -179,7 +324,7 @@ class EventBatchExecutor:
                     EventStatus.TIMEOUT,
                     error,
                 ),
-                batch_timed_out=batch_limited,
+                batch_timed_out=(batch_limited and count_deadline_exceeded),
             )
         except Exception as exc:
             return _ExecutionOutcome(
@@ -226,10 +371,25 @@ class EventBatchExecutor:
             **metadata,
         )
 
-    def _request_key(self, request: EventRequest) -> str:
+    def _request_key(
+        self,
+        request: EventRequest,
+        definition: EventDefinition | None,
+    ) -> str:
+        arguments_value = self._safe_arguments(request.arguments)
+        key_kind = "original"
+        if definition is not None and definition.normalizer is not None:
+            try:
+                normalized = definition.normalizer(arguments_value)
+                if not isinstance(normalized, dict):
+                    raise TypeError("normalizer returned non-object")
+                arguments_value = self._safe_arguments(normalized)
+                key_kind = "normalized"
+            except Exception:
+                arguments_value = self._safe_arguments(request.arguments)
         try:
             arguments = json.dumps(
-                request.arguments,
+                arguments_value,
                 ensure_ascii=False,
                 sort_keys=True,
                 separators=(",", ":"),
@@ -238,7 +398,7 @@ class EventBatchExecutor:
         except (TypeError, ValueError):
             arguments = request.raw_arguments
         return json.dumps(
-            [request.id, request.name, request.source.value, arguments],
+            [request.name, request.source.value, key_kind, arguments],
             ensure_ascii=False,
             separators=(",", ":"),
         )

src/agent_lab/application/events/builtin_plugins.py

@@ -240,6 +240,7 @@ def _session_terminate_definition(port: SessionTerminationPort) -> EventDefiniti
         conflict_keys=("session",),
         timeout_seconds=5.0,
         terminal=True,
+        result_message_factory=_session_terminate_message,
     )
 
 
@@ -410,6 +411,11 @@ def _calendar_schedule_message(result: EventResult) -> str:
     )
 
 
+def _session_terminate_message(result: EventResult) -> str:
+    del result
+    return "Goodbye."
+
+
 def _resolve_device_volume(
     request: EventRequest,
     context: EventExecutionContext,

src/agent_lab/application/events/kernel.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import inspect
 import json
 from collections.abc import Awaitable, Callable, Iterable
@@ -152,7 +153,13 @@ class EventKernel:
             raw_arguments=resolved.raw_arguments,
         )
         try:
-            payload = definition.handler(resolved_request)
+            if inspect.iscoroutinefunction(definition.handler):
+                payload = definition.handler(resolved_request)
+            else:
+                payload = await asyncio.to_thread(
+                    definition.handler,
+                    resolved_request,
+                )
             if inspect.isawaitable(payload):
                 payload = await payload
         except Exception as exc:

src/agent_lab/application/runtime.py

@@ -322,7 +322,12 @@ class DebugRuntime:
                 messages.append(assistant_message)
 
             if events:
-                batch_scope = self._event_scope(request, queues, turn_index=1)
+                batch_scope = self._event_scope(
+                    request,
+                    queues,
+                    turn_index=1,
+                    round_index=round_index,
+                )
                 await self._audit(
                     queues,
                     "event_batch_started",
@@ -396,6 +401,9 @@ class DebugRuntime:
                     batch_result,
                     turn_started_at=turn_started_at,
                     round_index=round_index,
+                    ensure_terminal_message=not bool(
+                        "".join(assistant_content).strip()
+                    ),
                 )
                 if (
                     decision.llm_follow_up
@@ -711,6 +719,7 @@ class DebugRuntime:
                     request,
                     queues,
                     turn_index=turn_index,
+                    round_index=round_index,
                 )
                 await self._audit(
                     queues,
@@ -719,6 +728,9 @@ class DebugRuntime:
                     session_id=session_id,
                     turn_index=turn_index,
                     round_index=round_index,
+                    ensure_terminal_message=not bool(
+                        "".join(assistant_content).strip()
+                    ),
                     scope=batch_scope,
                     event_names=[event.name for event in events],
                     max_parallel_events=request.event_agent.max_parallel_events,
@@ -798,6 +810,9 @@ class DebugRuntime:
                     session_id=session_id,
                     turn_index=turn_index,
                     round_index=round_index,
+                    ensure_terminal_message=not bool(
+                        "".join(assistant_content).strip()
+                    ),
                 )
                 if (
                     decision.llm_follow_up
@@ -1036,18 +1051,21 @@ class DebugRuntime:
         scope: str,
     ) -> tuple[EventBatchResult, list[ChatMessage]]:
         executor = self._batch_executor(params)
-        batch = await executor.execute(
-            [
-                self.registry.event_request(
-                    event,
-                    source=EventSource.PROVIDER_RESOLVED,
-                )
-                for event in events
-            ],
-            enabled_names=enabled_names,
-            scope=scope,
-        )
-        return batch, self.registry.tool_replies(events, batch)
+        try:
+            batch = await executor.execute(
+                [
+                    self.registry.event_request(
+                        event,
+                        source=EventSource.PROVIDER_RESOLVED,
+                    )
+                    for event in events
+                ],
+                enabled_names=enabled_names,
+                scope=scope,
+            )
+            return batch, self.registry.tool_replies(events, batch)
+        finally:
+            executor.release_scope(scope)
 
     def _batch_executor(self, params: EventAgentParams) -> EventBatchExecutor:
         key = (params.max_parallel_events, params.batch_timeout_seconds)
@@ -1067,9 +1085,10 @@ class DebugRuntime:
         queues: RuntimeQueues,
         *,
         turn_index: int,
+        round_index: int,
     ) -> str:
         session_scope = request.session_id or f"runtime-{id(queues)}"
-        return f"{session_scope}:turn-{turn_index}"
+        return f"{session_scope}:turn-{turn_index}:round-{round_index}"
 
     async def _audit_batch_result(
         self,
@@ -1117,8 +1136,12 @@ class DebugRuntime:
         session_id: str | None = None,
         turn_index: int | None = None,
         round_index: int | None = None,
+        ensure_terminal_message: bool = False,
     ) -> EventBatchDecision:
-        decision = self.registry.batch_decision(batch)
+        decision = self.registry.batch_decision(
+            batch,
+            ensure_terminal_message=ensure_terminal_message,
+        )
         for content in decision.template_messages:
             message = ChatMessage(role="assistant", content=content)
             messages.append(message)

src/agent_lab/application/tools.py

@@ -224,7 +224,12 @@ class ToolRegistry:
             for event, result in zip(events, batch.results, strict=True)
         ]
 
-    def batch_decision(self, batch: EventBatchResult) -> EventBatchDecision:
+    def batch_decision(
+        self,
+        batch: EventBatchResult,
+        *,
+        ensure_terminal_message: bool = False,
+    ) -> EventBatchDecision:
         templates: list[str] = []
         llm_follow_up = False
         terminate = False
@@ -234,19 +239,18 @@ class ToolRegistry:
                 continue
             if result.result_policy is ResultPolicy.TERMINATE:
                 terminate = True
+                if ensure_terminal_message:
+                    message = self._result_message(result)
+                    if message:
+                        templates.append(message)
                 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:
+            message = self._result_message(result)
+            if message is None:
                 llm_follow_up = True
                 continue
             if message:
@@ -257,6 +261,16 @@ class ToolRegistry:
             terminate=terminate,
         )
 
+    def _result_message(self, result: EventResult) -> str | None:
+        definition = self.event_registry.definition(result.event_name)
+        factory = definition.result_message_factory if definition else None
+        if factory is None:
+            return None
+        try:
+            return factory(result).strip()
+        except Exception:
+            return None
+
     def compact_results_message(
         self,
         batch: EventBatchResult,

tests/test_debug_runtime.py

@@ -9,7 +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.events import EventBatchExecutor, ResultPolicy
 from agent_lab.application.runtime import DebugRuntime
 from agent_lab.application.tools import ToolDefinition, ToolExecutionContext, ToolRegistry
 from agent_lab.domain.events import ToolCallEvent
@@ -2118,8 +2118,16 @@ async def test_direct_mode_executes_batch_concurrently_but_replies_in_provider_o
     client = ScriptedChatClient(
         [
             [
-                StreamItem.provider_tool_call(_tool_call("first-call")),
-                StreamItem.provider_tool_call(_tool_call("second-call")),
+                StreamItem.provider_tool_call(
+                    _tool_call("first-call").model_copy(
+                        update={"arguments": {"order": 1}, "raw_arguments": '{"order":1}'}
+                    )
+                ),
+                StreamItem.provider_tool_call(
+                    _tool_call("second-call").model_copy(
+                        update={"arguments": {"order": 2}, "raw_arguments": '{"order":2}'}
+                    )
+                ),
             ],
             [StreamItem.message_delta("finished")],
         ]
@@ -2492,3 +2500,228 @@ async def test_reusable_session_terminate_completes_turn_and_session_task():
 
     business_types = [message["type"] for message in _without_audit(outputs)]
     assert business_types[-2:] == ["turn_completed", "done"]
+
+
+@pytest.mark.asyncio
+async def test_dual_text_event_scope_uses_round_and_does_not_replay_stale_history():
+    handled_messages: list[str] = []
+
+    def resolve_from_history(
+        event: ToolCallEvent,
+        context: ToolExecutionContext,
+    ) -> dict[str, Any]:
+        del event
+        return {
+            "message": next(
+                message.content
+                for message in reversed(context.history)
+                if message.role == "assistant" and message.content
+            )
+        }
+
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="example.history",
+                description="Capture the current assistant history.",
+                parameters={
+                    "type": "object",
+                    "properties": {"message": {"type": "string"}},
+                    "required": ["message"],
+                },
+                handler=lambda event: handled_messages.append(
+                    event.arguments["message"]
+                )
+                or {"message": event.arguments["message"]},
+                argument_resolver=resolve_from_history,
+            )
+        ]
+    )
+    repeated = _tool_call("event_1", name="example.history")
+    client = ScriptedChatClient(
+        [
+            [StreamItem.message_delta("first history"), StreamItem.text_event(repeated)],
+            [StreamItem.message_delta("second history"), StreamItem.text_event(repeated)],
+            [StreamItem.message_delta("final answer")],
+        ]
+    )
+    request = DebugRunRequest(
+        user_message="capture each round",
+        system_prompts=[],
+        pre_messages=[],
+        chat_agent=AgentParams(model="chat-model"),
+        event_agent=EventAgentParams(
+            enabled_tools=["example.history"],
+            max_event_loops=2,
+        ),
+    )
+
+    outputs = await _collect_outputs(
+        DebugRuntime(client, registry=registry).run(request)
+    )
+
+    assert handled_messages == ["first history", "second history"]
+    scopes = [
+        message["details"]["scope"]
+        for message in outputs
+        if message.get("event") == "event_batch_started"
+    ]
+    assert scopes[0].endswith(":round-1")
+    assert scopes[1].endswith(":round-2")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["dual_agent", "chat_agent_tools"])
+async def test_runtime_releases_batch_scope_after_results(monkeypatch, mode: str):
+    released: list[Any] = []
+    original_release = EventBatchExecutor.release_scope
+
+    def record_release(self: EventBatchExecutor, scope: Any) -> None:
+        released.append(scope)
+        original_release(self, scope)
+
+    monkeypatch.setattr(EventBatchExecutor, "release_scope", record_release)
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="example.silent",
+                description="Complete without another model call.",
+                parameters={"type": "object"},
+                handler=lambda event: {"event_id": event.id},
+                result_policy=ResultPolicy.SILENT_SUCCESS,
+            )
+        ]
+    )
+    event = _tool_call("release-1", name="example.silent")
+    item = (
+        StreamItem.text_event(event)
+        if mode == "dual_agent"
+        else StreamItem.provider_tool_call(event)
+    )
+    client = ScriptedChatClient([[item]])
+
+    await _collect_outputs(
+        DebugRuntime(client, registry=registry).run(
+            _policy_request(enabled_tools=["example.silent"], mode=mode)
+        )
+    )
+
+    assert len(released) == 1
+    assert str(released[0]).endswith(":round-1")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["dual_agent", "chat_agent_tools"])
+async def test_tool_only_terminate_emits_plugin_farewell_once(mode: str):
+    event = ToolCallEvent(
+        id="terminate-only",
+        name="session.terminate",
+        arguments={},
+        raw_arguments="{}",
+    )
+    item = (
+        StreamItem.text_event(event)
+        if mode == "dual_agent"
+        else StreamItem.provider_tool_call(event)
+    )
+    client = ScriptedChatClient([[item]])
+
+    outputs = await _collect_outputs(
+        DebugRuntime(client).run(
+            _policy_request(enabled_tools=["session.terminate"], mode=mode)
+        )
+    )
+
+    assert client.calls == 1
+    assert [
+        message["content"]
+        for message in outputs
+        if message["type"] == "message_delta"
+    ] == ["Goodbye."]
+    assert outputs[-1] == {"type": "done"}
+
+
+@pytest.mark.asyncio
+async def test_tool_only_terminate_ends_reusable_session_after_plugin_farewell():
+    client = ScriptedChatClient(
+        [
+            [
+                StreamItem.provider_tool_call(
+                    ToolCallEvent(
+                        id="terminate-session-only",
+                        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="terminal-only-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)
+
+    assert [
+        message["content"]
+        for message in outputs
+        if message["type"] == "message_delta"
+    ] == ["Goodbye."]
+    assert [message["type"] for message in _without_audit(outputs)][-2:] == [
+        "turn_completed",
+        "done",
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["dual_agent", "chat_agent_tools"])
+async def test_web_search_has_first_answer_tool_result_and_one_second_answer(mode: str):
+    search = ToolCallEvent(
+        id="search-two-answer",
+        name="knowledge.web.search",
+        arguments={"query": "event batch safety"},
+        raw_arguments='{"query":"event batch safety"}',
+    )
+    item = (
+        StreamItem.text_event(search)
+        if mode == "dual_agent"
+        else StreamItem.provider_tool_call(search)
+    )
+    client = ScriptedChatClient(
+        [
+            [StreamItem.message_delta("I will check."), item],
+            [StreamItem.message_delta("Grounded search update.")],
+        ]
+    )
+
+    outputs = await _collect_outputs(
+        DebugRuntime(client).run(
+            _policy_request(enabled_tools=["knowledge.web.search"], mode=mode)
+        )
+    )
+
+    business = _without_audit(outputs)
+    visible = [
+        (index, message["content"])
+        for index, message in enumerate(business)
+        if message["type"] == "message_delta"
+    ]
+    tool_index = next(
+        index for index, message in enumerate(business) if message["type"] == "tool_result"
+    )
+    assert client.calls == 2
+    assert [content for _, content in visible] == [
+        "I will check.",
+        "Grounded search update.",
+    ]
+    assert visible[0][0] < tool_index < visible[1][0]
+    assert "sources" in business[tool_index]["message"]["content"]

tests/test_event_batch.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import threading
 from collections.abc import Callable
 from typing import Any
 
@@ -12,8 +13,11 @@ from agent_lab.application.events import (
     EventKernel,
     EventRegistry,
     EventRequest,
+    EventSource,
     EventStatus,
 )
+from agent_lab.application.tools import ToolDefinition, ToolRegistry
+from agent_lab.domain.events import ToolCallEvent
 
 
 def _executor_type():
@@ -29,6 +33,7 @@ def _definition(
     conflict_keys: tuple[str, ...] = (),
     timeout_seconds: float | None = None,
     terminal: bool = False,
+    normalizer: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
 ) -> EventDefinition:
     return EventDefinition(
         name=name,
@@ -38,6 +43,7 @@ def _definition(
         conflict_keys=conflict_keys,
         timeout_seconds=timeout_seconds,
         terminal=terminal,
+        normalizer=normalizer,
     )
 
 
@@ -54,11 +60,13 @@ def _executor(
     *,
     max_parallel_events: int = 4,
     batch_timeout_seconds: float = 1.0,
+    **options: Any,
 ):
     return _executor_type()(
         EventKernel(EventRegistry(definitions)),
         max_parallel_events=max_parallel_events,
         batch_timeout_seconds=batch_timeout_seconds,
+        **options,
     )
 
 
@@ -81,7 +89,10 @@ async def test_batch_executor_bounds_global_parallelism():
     )
 
     batch = await executor.execute(
-        [_request(str(index), "example.work") for index in range(4)],
+        [
+            _request(str(index), "example.work", {"index": index})
+            for index in range(4)
+        ],
         scope="session-1:turn-1",
     )
 
@@ -107,8 +118,8 @@ async def test_batch_executor_preserves_input_order_when_completion_order_differ
     task = asyncio.create_task(
         executor.execute(
             [
-                _request("first", "example.work"),
-                _request("second", "example.work"),
+                _request("first", "example.work", {"order": 1}),
+                _request("second", "example.work", {"order": 2}),
             ],
             scope="session-1:turn-1",
         )
@@ -146,8 +157,8 @@ async def test_batch_executor_serializes_shared_conflict_keys():
 
     await executor.execute(
         [
-            _request("first", "example.write"),
-            _request("second", "example.write"),
+            _request("first", "example.write", {"order": 1}),
+            _request("second", "example.write", {"order": 2}),
         ],
         scope="session-1:turn-1",
     )
@@ -309,8 +320,8 @@ async def test_batch_executor_runs_terminal_events_after_all_nonterminal_results
     batch = await executor.execute(
         [
             _request("terminate", "example.terminate"),
-            _request("first", "example.work"),
-            _request("second", "example.work"),
+            _request("first", "example.work", {"order": 1}),
+            _request("second", "example.work", {"order": 2}),
         ],
         scope="session-1:turn-1",
     )
@@ -347,3 +358,305 @@ async def test_batch_executor_does_not_swallow_external_cancellation():
 
     with pytest.raises(asyncio.CancelledError):
         await task
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_coalesces_normalized_provider_arguments_across_ids():
+    calls: list[EventRequest] = []
+
+    def normalize(arguments: dict[str, Any]) -> dict[str, Any]:
+        value = arguments["value"]
+        return {"value": int(value) if isinstance(value, float) else value}
+
+    executor = _executor(
+        [
+            _definition(
+                "example.normalized",
+                lambda request: calls.append(request) or {"value": request.arguments["value"]},
+                normalizer=normalize,
+            )
+        ]
+    )
+    requests = [
+        EventRequest(
+            id="integer-id",
+            name="example.normalized",
+            arguments={"value": 30},
+            source=EventSource.PROVIDER_RESOLVED,
+            raw_arguments='{"value":30}',
+        ),
+        EventRequest(
+            id="float-id",
+            name="example.normalized",
+            arguments={"value": 30.0},
+            source=EventSource.PROVIDER_RESOLVED,
+            raw_arguments='{"value":30.0}',
+        ),
+    ]
+
+    batch = await executor.execute(requests, scope="normalized-scope")
+
+    assert len(calls) == 1
+    assert batch.coalesced_count == 1
+    assert [result.event_id for result in batch.results] == [
+        "integer-id",
+        "float-id",
+    ]
+    assert batch.results[0].payload == {"value": 30}
+    assert batch.results[1].payload == {
+        "value": 30,
+        "deduplicated_from": "integer-id",
+    }
+
+
+@pytest.mark.asyncio
+async def test_batch_executor_tool_replies_keep_each_coalesced_call_id():
+    registry = ToolRegistry(
+        [
+            ToolDefinition(
+                name="example.tool",
+                description="Execute a coalesced tool.",
+                parameters={"type": "object"},
+                handler=lambda event: {"handled_by": event.id},
+            )
+        ]
+    )
+    executor = _executor_type()(
+        EventKernel(registry.event_registry),
+        batch_timeout_seconds=1,
+    )
+    events = [
+        ToolCallEvent(id=event_id, name="example.tool", arguments={}, raw_arguments="{}")
+        for event_id in ("first-id", "second-id")
+    ]
+    batch = await executor.execute(
+        [
+            registry.event_request(event, source=EventSource.PROVIDER_RESOLVED)
+            for event in events
+        ],
+        scope="tool-scope",
+    )
+
+    replies = registry.tool_replies(events, batch)
+
+    assert [reply.tool_call_id for reply in replies] == ["first-id", "second-id"]
+    assert [result.event_id for result in batch.results] == [
+        "first-id",
+        "second-id",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_batch_key_falls_back_to_original_arguments_when_normalizer_fails():
+    calls = 0
+
+    def normalize(arguments: dict[str, Any]) -> dict[str, Any]:
+        if isinstance(arguments.get("value"), str):
+            raise ValueError("unsupported value")
+        return {"value": int(arguments["value"])}
+
+    def handler(request: EventRequest) -> dict[str, Any]:
+        nonlocal calls
+        calls += 1
+        return {"value": request.arguments["value"]}
+
+    executor = _executor(
+        [_definition("example.normalized", handler, normalizer=normalize)]
+    )
+
+    batch = await executor.execute(
+        [
+            EventRequest(
+                id="valid",
+                name="example.normalized",
+                arguments={"value": 30},
+                source=EventSource.PROVIDER_RESOLVED,
+            ),
+            EventRequest(
+                id="invalid",
+                name="example.normalized",
+                arguments={"value": "30"},
+                source=EventSource.PROVIDER_RESOLVED,
+            ),
+        ],
+        scope="normalizer-failure",
+    )
+
+    assert batch.coalesced_count == 0
+    assert [result.status for result in batch.results] == [
+        EventStatus.SUCCESS,
+        EventStatus.INVALID_ARGUMENTS,
+    ]
+    assert calls == 1
+
+
+@pytest.mark.asyncio
+async def test_concurrent_replay_waiters_share_execution_and_cancel_independently():
+    calls = 0
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def handler(request: EventRequest) -> dict[str, Any]:
+        nonlocal calls
+        calls += 1
+        started.set()
+        await release.wait()
+        return {"event_id": request.id}
+
+    executor = _executor([_definition("example.shared", handler)])
+    request = _request("same-id", "example.shared")
+    first = asyncio.create_task(executor.execute([request], scope="shared-scope"))
+    second = asyncio.create_task(executor.execute([request], scope="shared-scope"))
+    await asyncio.wait_for(started.wait(), timeout=0.2)
+
+    first.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await first
+    release.set()
+    second_batch = await asyncio.wait_for(second, timeout=0.2)
+
+    assert calls == 1
+    assert second_batch.results[0].status is EventStatus.SUCCESS
+
+
+@pytest.mark.asyncio
+async def test_release_scope_removes_replay_entries_for_reuse():
+    calls = 0
+
+    def handler(request: EventRequest) -> dict[str, Any]:
+        nonlocal calls
+        calls += 1
+        return {"call": calls}
+
+    executor = _executor([_definition("example.cached", handler)])
+    request = _request("same-id", "example.cached")
+
+    await executor.execute([request], scope="released-scope")
+    executor.release_scope("released-scope")
+    batch = await executor.execute([request], scope="released-scope")
+
+    assert calls == 2
+    assert batch.replayed_count == 0
+
+
+@pytest.mark.asyncio
+async def test_replay_cache_stays_bounded_across_250_scopes():
+    executor = _executor(
+        [_definition("example.cached", lambda request: {"id": request.id})],
+        max_replay_entries=100,
+    )
+    request = _request("same-id", "example.cached")
+
+    for index in range(250):
+        await executor.execute([request], scope=f"scope-{index}")
+
+    assert len(executor._replay_results) <= 100
+
+
+@pytest.mark.asyncio
+async def test_terminal_uses_independent_grace_after_sibling_batch_timeout():
+    terminal_calls: list[str] = []
+
+    async def blocked(request: EventRequest) -> dict[str, Any]:
+        await asyncio.Event().wait()
+        return {"event_id": request.id}
+
+    def terminate(request: EventRequest) -> dict[str, Any]:
+        terminal_calls.append(request.id)
+        return {"terminated": True}
+
+    executor = _executor(
+        [
+            _definition("example.blocked", blocked),
+            _definition(
+                "example.terminate",
+                terminate,
+                terminal=True,
+                timeout_seconds=0.05,
+            ),
+        ],
+        batch_timeout_seconds=0.01,
+        terminal_grace_seconds=0.1,
+    )
+
+    batch = await executor.execute(
+        [
+            _request("sibling", "example.blocked"),
+            _request("terminal", "example.terminate"),
+        ],
+        scope="terminal-grace",
+    )
+
+    assert [result.status for result in batch.results] == [
+        EventStatus.TIMEOUT,
+        EventStatus.SUCCESS,
+    ]
+    assert batch.deadline_exceeded is True
+    assert terminal_calls == ["terminal"]
+
+
+@pytest.mark.asyncio
+async def test_terminal_grace_is_bounded_by_terminal_event_timeout():
+    async def blocked_terminal(request: EventRequest) -> dict[str, Any]:
+        await asyncio.Event().wait()
+        return {"event_id": request.id}
+
+    executor = _executor(
+        [
+            _definition(
+                "example.terminate",
+                blocked_terminal,
+                terminal=True,
+                timeout_seconds=0.01,
+            )
+        ],
+        terminal_grace_seconds=0.2,
+    )
+
+    batch = await executor.execute(
+        [_request("terminal", "example.terminate")],
+        scope="terminal-timeout",
+    )
+
+    assert batch.results[0].status is EventStatus.TIMEOUT
+    assert batch.results[0].error == "event timed out after 0.01 seconds"
+    assert batch.deadline_exceeded is False
+
+
+@pytest.mark.asyncio
+async def test_blocking_sync_handler_timeout_returns_before_thread_finishes():
+    started = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+
+    def blocking_handler(request: EventRequest) -> dict[str, Any]:
+        started.set()
+        release.wait(timeout=1)
+        finished.set()
+        return {"event_id": request.id}
+
+    executor = _executor(
+        [
+            _definition(
+                "example.blocking",
+                blocking_handler,
+                timeout_seconds=0.02,
+            )
+        ],
+        batch_timeout_seconds=0.2,
+    )
+    loop = asyncio.get_running_loop()
+    started_at = loop.time()
+
+    batch = await executor.execute(
+        [_request("blocking", "example.blocking")],
+        scope="blocking-handler",
+    )
+    elapsed = loop.time() - started_at
+    release.set()
+    await asyncio.to_thread(finished.wait, 0.2)
+
+    assert started.is_set()
+    assert batch.results[0].status is EventStatus.TIMEOUT
+    assert elapsed < 0.1
+    assert finished.is_set()

tests/test_event_kernel.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import gc
 import json
+import threading
 import warnings
 from dataclasses import FrozenInstanceError
 from typing import Any
@@ -1497,6 +1498,32 @@ async def test_async_kernel_supports_async_handler():
     assert result.payload == {"query": "async"}
 
 
+@pytest.mark.asyncio
+async def test_async_kernel_offloads_sync_handler_but_execute_sync_stays_inline():
+    caller_thread = threading.get_ident()
+    handler_threads: list[int] = []
+
+    def handler(request: EventRequest) -> dict[str, Any]:
+        handler_threads.append(threading.get_ident())
+        return {"query": request.arguments["query"]}
+
+    kernel = EventKernel(EventRegistry([_definition(handler=handler)]))
+    request = EventRequest(
+        id="threaded",
+        name="example.lookup",
+        arguments={"query": "threaded"},
+        raw_arguments='{"query":"threaded"}',
+    )
+
+    async_result = await kernel.execute(request)
+    sync_result = kernel.execute_sync(request)
+
+    assert async_result.status is EventStatus.SUCCESS
+    assert sync_result.status is EventStatus.SUCCESS
+    assert handler_threads[0] != caller_thread
+    assert handler_threads[1] == caller_thread
+
+
 @pytest.mark.asyncio
 async def test_tool_registry_async_entry_points_support_async_handler():
     async def handler(event: ToolCallEvent) -> dict[str, Any]: