feat: enhance tool call lifecycle management with call_id

backend/app/adapters/adapter_tools.py

@@ -23,6 +23,7 @@
 
 from __future__ import annotations
 
+import time
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
@@ -107,23 +108,40 @@ async def execute(
         step_index: int,
         name: str,
         arguments: dict[str, Any],
+        call_id: str | None = None,
     ) -> dict[str, Any]:
-        """Invoke a tool and emit standardized ToolCall lifecycle events."""
+        """Invoke a tool and emit standardized ToolCall lifecycle events.
+
+        ``call_id`` links ``started`` ↔ ``completed`` so parallel (or same-name)
+        invocations do not clobber each other. When omitted a ULID is allocated.
+        """
         tool = self.lookup(name)
-        await ctx.emit_tool_call_started(
-            step_index=step_index, name=tool.name, arguments=arguments
+        cid = await ctx.emit_tool_call_started(
+            step_index=step_index,
+            name=tool.name,
+            arguments=arguments,
+            call_id=call_id,
         )
+        started = time.monotonic()
         try:
             result = await tool.handler(arguments)
             if not isinstance(result, dict):
                 result = {"result": result}
             await ctx.emit_tool_call_completed(
-                step_index=step_index, name=tool.name, result=result
+                step_index=step_index,
+                name=tool.name,
+                call_id=cid,
+                result=result,
+                latency_ms=int((time.monotonic() - started) * 1000),
             )
             return result
         except Exception as exc:
             await ctx.emit_tool_call_completed(
-                step_index=step_index, name=tool.name, error=str(exc)
+                step_index=step_index,
+                name=tool.name,
+                call_id=cid,
+                error=str(exc),
+                latency_ms=int((time.monotonic() - started) * 1000),
             )
             raise
 

backend/app/adapters/base.py

@@ -14,13 +14,20 @@
 from datetime import UTC, datetime
 from typing import Any
 
+from ulid import ULID
+
 from app.models.run import RunStatus
-from app.schemas.run import EventType
 from app.runtime.resume_context import RunResumeContext
+from app.schemas.run import EventType
 
 EmitCallback = Callable[[EventType, dict[str, Any]], Awaitable[None]]
 
 
+def new_tool_call_id() -> str:
+    """Stable id linking ``tool_call.started`` ↔ ``tool_call.completed``."""
+    return str(ULID())
+
+
 @dataclass
 class AdapterContext:
     """Per-run state exposed to adapters.
@@ -104,30 +111,52 @@ async def emit_message(
         )
 
     async def emit_tool_call_started(
-        self, *, step_index: int, name: str, arguments: dict[str, Any]
-    ) -> None:
+        self,
+        *,
+        step_index: int,
+        name: str,
+        arguments: dict[str, Any],
+        call_id: str | None = None,
+    ) -> str:
+        """Emit ``tool_call.started`` and return the association ``call_id``.
+
+        Pass the returned id (or the same ``call_id``) to
+        ``emit_tool_call_completed`` so parallel / same-name tool calls are
+        matched correctly. When omitted, a new ULID is generated.
+        """
+        cid = call_id or new_tool_call_id()
         await self.emit(
             "tool_call.started",
-            {"step_index": step_index, "name": name, "arguments": arguments},
+            {
+                "step_index": step_index,
+                "name": name,
+                "arguments": arguments,
+                "call_id": cid,
+            },
         )
+        return cid
 
     async def emit_tool_call_completed(
         self,
         *,
         step_index: int,
         name: str,
+        call_id: str | None = None,
         result: dict[str, Any] | None = None,
         error: str | None = None,
+        latency_ms: int | None = None,
     ) -> None:
-        await self.emit(
-            "tool_call.completed",
-            {
-                "step_index": step_index,
-                "name": name,
-                "result": result,
-                "error": error,
-            },
-        )
+        payload: dict[str, Any] = {
+            "step_index": step_index,
+            "name": name,
+            "result": result,
+            "error": error,
+        }
+        if call_id is not None:
+            payload["call_id"] = call_id
+        if latency_ms is not None:
+            payload["latency_ms"] = latency_ms
+        await self.emit("tool_call.completed", payload)
 
     async def emit_log(self, message: str, **fields: Any) -> None:
         await self.emit(

backend/app/adapters/echo_adapter.py

@@ -20,6 +20,7 @@
 from app.models.run import RunStatus
 from app.runtime.pricing import estimate_cost_usd
 from app.runtime.tokens import estimate_tokens
+
 _STEPS = ("plan", "tool", "reply")
 
 
@@ -78,12 +79,15 @@ async def run(self, ctx: AdapterContext) -> AdapterResult:
                 )
 
             await ctx.emit_step_started(index=idx, node="tool")
-            await ctx.emit_tool_call_started(
+            call_id = await ctx.emit_tool_call_started(
                 step_index=idx, name="echo", arguments={"text": prompt}
             )
             await asyncio.sleep(delay)
             await ctx.emit_tool_call_completed(
-                step_index=idx, name="echo", result={"text": prompt}
+                step_index=idx,
+                name="echo",
+                call_id=call_id,
+                result={"text": prompt},
             )
             await ctx.emit_step_completed(
                 index=idx, node="tool", output={"echo": prompt}

backend/app/adapters/langgraph_adapter.py

@@ -57,6 +57,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import time
 from collections.abc import Awaitable, Callable
@@ -544,7 +545,8 @@ async def handler(state: dict[str, Any]) -> dict[str, Any]:
                                 ],
                             }
                         )
-                        for tc in response.tool_calls:
+
+                        async def _run_tool(tc: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
                             name = str(tc.get("name") or "")
                             arguments = dict(tc.get("arguments") or {})
                             result = await run_state.tool_surface.execute(
@@ -553,6 +555,15 @@ async def handler(state: dict[str, Any]) -> dict[str, Any]:
                                 name=name,
                                 arguments=arguments,
                             )
+                            return tc, result
+
+                        # Parallel tool calls share one step; call_id keeps
+                        # started/completed pairs associated correctly.
+                        pairs = await asyncio.gather(
+                            *[_run_tool(tc) for tc in response.tool_calls]
+                        )
+                        for tc, result in pairs:
+                            name = str(tc.get("name") or "")
                             tool_results[name] = result
                             messages.append(
                                 {

backend/app/services/run_service.py

@@ -434,29 +434,33 @@ async def _handle_event(
         elif event_type == "tool_call.started":
             step = await self._find_step(run_id, data["step_index"])
             if step is not None:
-                call = ToolCall(
-                    step_id=step.id,
-                    name=data["name"],
-                    arguments=data.get("arguments", {}),
-                )
+                call_kwargs: dict[str, Any] = {
+                    "step_id": step.id,
+                    "name": data["name"],
+                    "arguments": data.get("arguments", {}),
+                }
+                call_id = data.get("call_id")
+                if isinstance(call_id, str) and call_id:
+                    call_kwargs["id"] = call_id
+                call = ToolCall(**call_kwargs)
                 self.session.add(call)
                 await self.session.commit()
+                # Ensure SSE clients always receive the persisted association key.
+                data = {**data, "call_id": call.id}
         elif event_type == "tool_call.completed":
             step = await self._find_step(run_id, data["step_index"])
             if step is not None:
-                stmt = (
-                    select(ToolCall)
-                    .where(ToolCall.step_id == step.id, ToolCall.name == data["name"])
-                    .order_by(ToolCall.created_at.desc())
-                    .limit(1)
+                call = await self._find_tool_call(
+                    step_id=step.id,
+                    name=data["name"],
+                    call_id=data.get("call_id"),
                 )
-                result = await self.session.execute(stmt)
-                call = result.scalar_one_or_none()
                 if call is not None:
                     call.result = data.get("result")
                     call.error = data.get("error")
                     call.latency_ms = data.get("latency_ms")
                     await self.session.commit()
+                    data = {**data, "call_id": call.id}
                     run = await self._get_run(run_id)
                     record_tool_call(
                         adapter=run.adapter,
@@ -493,6 +497,43 @@ async def _find_step(self, run_id: str, index: int) -> Step | None:
         result = await self.session.execute(stmt)
         return result.scalar_one_or_none()
 
+    async def _find_tool_call(
+        self,
+        *,
+        step_id: str,
+        name: str,
+        call_id: Any = None,
+    ) -> ToolCall | None:
+        """Resolve a ToolCall for ``tool_call.completed``.
+
+        Prefer ``call_id`` (``ToolCall.id``) so parallel / same-name invocations
+        associate correctly. Fall back to the oldest incomplete row with a
+        matching name for legacy emitters that omit ``call_id``.
+        """
+        if isinstance(call_id, str) and call_id:
+            stmt = select(ToolCall).where(
+                ToolCall.id == call_id,
+                ToolCall.step_id == step_id,
+            )
+            result = await self.session.execute(stmt)
+            call = result.scalar_one_or_none()
+            if call is not None:
+                return call
+
+        stmt = (
+            select(ToolCall)
+            .where(
+                ToolCall.step_id == step_id,
+                ToolCall.name == name,
+                ToolCall.result.is_(None),
+                ToolCall.error.is_(None),
+            )
+            .order_by(ToolCall.created_at.asc())
+            .limit(1)
+        )
+        result = await self.session.execute(stmt)
+        return result.scalar_one_or_none()
+
     async def _next_message_index(self, run_id: str) -> int:
         stmt = (
             select(Message.index)

backend/app/worker/executor.py

@@ -27,15 +27,16 @@
     OrchestratorAdapter,
     get_adapter,
 )
-from app.runtime.resume_context import (
-    parse_resume_context,
-    without_resume_metadata,
-)
 from app.core.logging import get_logger
 from app.core.telemetry import trace_adapter_run
 from app.db.session import SessionLocal
 from app.events import EventBus
 from app.models import Agent, RunStatus
+from app.runtime.resume_context import (
+    parse_resume_context,
+    without_resume_metadata,
+)
+from app.schemas.run import EventType
 from app.worker.cancel import CancelRegistry, InMemoryCancelRegistry
 
 logger = get_logger("worker.executor")
@@ -90,6 +91,14 @@ async def execute(self, run_id: str, adapter_name: str) -> None:
             await session.commit()
             await service._broadcast("run.started", run.id, {})
 
+            # Serialize DB writes: adapters may emit from concurrent tasks
+            # (e.g. parallel tool calls) while sharing one AsyncSession.
+            emit_lock = asyncio.Lock()
+
+            async def _emit(event_type: EventType, data: dict[str, Any]) -> None:
+                async with emit_lock:
+                    await service._handle_event(run.id, event_type, data)
+
             ctx = AdapterContext(
                 run_id=run.id,
                 agent_id=agent.id,
@@ -98,9 +107,7 @@ async def execute(self, run_id: str, adapter_name: str) -> None:
                 metadata=run.metadata_,
                 resume=resume_ctx,
                 step_index_base=step_index_base,
-                emit=lambda event_type, data: service._handle_event(
-                    run.id, event_type, data
-                ),
+                emit=_emit,
             )
 
             adapter = get_adapter(adapter_name)

backend/tests/test_adapter_tools.py

@@ -62,8 +62,11 @@ async def test_adapter_tool_surface_builtin_execute():
         completed = [d for e, d in ctx.events if e == "tool_call.completed"]
         assert len(started) == 1
         assert started[0]["name"] == "echo"
+        assert "call_id" in started[0]
         assert len(completed) == 1
         assert completed[0]["result"] == {"text": "ping"}
+        assert completed[0]["call_id"] == started[0]["call_id"]
+        assert isinstance(completed[0].get("latency_ms"), int)
     finally:
         await surface.close()