From 42b615c9ba53a30bd334e53b1020da58d48310ba Mon Sep 17 00:00:00 2001 From: tutudouzi12 <18865007342@163.com> Date: Wed, 5 Aug 2026 22:49:55 +0800 Subject: [PATCH] feat: enhance tool call lifecycle management with call_id - Introduced a stable `call_id` for `tool_call.started` and `tool_call.completed` events to associate parallel or same-name invocations within a step. - Updated the `emit_tool_call_started` and `emit_tool_call_completed` methods to accept and return `call_id`. - Modified adapters to utilize `call_id` for better tracking of tool call results and errors. - Enhanced documentation to reflect changes in the tool call lifecycle and the usage of `call_id`. --- backend/app/adapters/adapter_tools.py | 28 ++- backend/app/adapters/base.py | 55 +++-- backend/app/adapters/echo_adapter.py | 8 +- backend/app/adapters/langgraph_adapter.py | 13 +- backend/app/services/run_service.py | 65 ++++-- backend/app/worker/executor.py | 21 +- backend/tests/test_adapter_tools.py | 3 + backend/tests/test_tool_call_lifecycle.py | 234 ++++++++++++++++++++++ docs/api-contract.md | 16 ++ docs/data-model.md | 4 + frontend/lib/run-events.ts | 92 +++++++-- 11 files changed, 479 insertions(+), 60 deletions(-) create mode 100644 backend/tests/test_tool_call_lifecycle.py diff --git a/backend/app/adapters/adapter_tools.py b/backend/app/adapters/adapter_tools.py index fa4329c..4fbf25f 100644 --- a/backend/app/adapters/adapter_tools.py +++ b/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 diff --git a/backend/app/adapters/base.py b/backend/app/adapters/base.py index 49579cc..496f684 100644 --- a/backend/app/adapters/base.py +++ b/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( diff --git a/backend/app/adapters/echo_adapter.py b/backend/app/adapters/echo_adapter.py index 0ac1de4..ca64ac8 100644 --- a/backend/app/adapters/echo_adapter.py +++ b/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} diff --git a/backend/app/adapters/langgraph_adapter.py b/backend/app/adapters/langgraph_adapter.py index 223c5af..8698a50 100644 --- a/backend/app/adapters/langgraph_adapter.py +++ b/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( { diff --git a/backend/app/services/run_service.py b/backend/app/services/run_service.py index c7a3aa9..8cf9bca 100644 --- a/backend/app/services/run_service.py +++ b/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) diff --git a/backend/app/worker/executor.py b/backend/app/worker/executor.py index d70baaf..b8f88c9 100644 --- a/backend/app/worker/executor.py +++ b/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) diff --git a/backend/tests/test_adapter_tools.py b/backend/tests/test_adapter_tools.py index 3dac4c5..a0274f2 100644 --- a/backend/tests/test_adapter_tools.py +++ b/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() diff --git a/backend/tests/test_tool_call_lifecycle.py b/backend/tests/test_tool_call_lifecycle.py new file mode 100644 index 0000000..976c416 --- /dev/null +++ b/backend/tests/test_tool_call_lifecycle.py @@ -0,0 +1,234 @@ +"""ToolCall started/completed association — including parallel same-name calls.""" + +from __future__ import annotations + +import asyncio + +import pytest +from sqlalchemy import select + +from app.adapters.base import AdapterContext, AdapterResult, OrchestratorAdapter, new_tool_call_id +from app.db.base import Base +from app.db.session import SessionLocal, engine +from app.events import get_event_bus +from app.models import Agent, RunStatus, Step, ToolCall +from app.schemas.run import RunCreate +from app.services.run_service import RunService +from app.worker.cancel import InMemoryCancelRegistry +from app.worker.executor import RunExecutor + + +@pytest.fixture(autouse=True) +async def schema(): + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.drop_all) + await conn.run_sync(Base.metadata.create_all) + yield + + +async def _make_agent(session, adapter: str = "echo") -> Agent: + agent = Agent(name=f"agent-{id(session)}", adapter=adapter, config={}) + session.add(agent) + await session.commit() + await session.refresh(agent) + return agent + + +@pytest.mark.asyncio +async def test_tool_call_completed_matches_by_call_id_not_latest_name(): + """Out-of-order same-name completions must not clobber the wrong row.""" + bus = get_event_bus() + + async with SessionLocal() as session: + agent = await _make_agent(session) + service = RunService(session=session, bus=bus) + run = await service.create_run( + RunCreate(agent_id=agent.id, input={"prompt": "x"}) + ) + + await service._handle_event( + run.id, "step.started", {"index": 0, "node": "tools"} + ) + + first_id = new_tool_call_id() + second_id = new_tool_call_id() + + await service._handle_event( + run.id, + "tool_call.started", + { + "step_index": 0, + "name": "echo", + "arguments": {"text": "a"}, + "call_id": first_id, + }, + ) + await service._handle_event( + run.id, + "tool_call.started", + { + "step_index": 0, + "name": "echo", + "arguments": {"text": "b"}, + "call_id": second_id, + }, + ) + + # Complete the *second* call first (parallel out-of-order). + await service._handle_event( + run.id, + "tool_call.completed", + { + "step_index": 0, + "name": "echo", + "call_id": second_id, + "result": {"text": "b"}, + "latency_ms": 5, + }, + ) + await service._handle_event( + run.id, + "tool_call.completed", + { + "step_index": 0, + "name": "echo", + "call_id": first_id, + "result": {"text": "a"}, + "latency_ms": 9, + }, + ) + + async with SessionLocal() as session: + step = ( + await session.execute( + select(Step).where(Step.run_id == run.id, Step.index == 0) + ) + ).scalar_one() + calls = ( + await session.execute( + select(ToolCall) + .where(ToolCall.step_id == step.id) + .order_by(ToolCall.created_at.asc()) + ) + ).scalars().all() + + assert len(calls) == 2 + by_id = {c.id: c for c in calls} + assert by_id[first_id].arguments == {"text": "a"} + assert by_id[first_id].result == {"text": "a"} + assert by_id[first_id].latency_ms == 9 + assert by_id[second_id].arguments == {"text": "b"} + assert by_id[second_id].result == {"text": "b"} + assert by_id[second_id].latency_ms == 5 + + +@pytest.mark.asyncio +async def test_tool_call_completed_fallback_uses_oldest_incomplete(): + bus = get_event_bus() + + async with SessionLocal() as session: + agent = await _make_agent(session) + service = RunService(session=session, bus=bus) + run = await service.create_run( + RunCreate(agent_id=agent.id, input={"prompt": "x"}) + ) + await service._handle_event( + run.id, "step.started", {"index": 0, "node": "tools"} + ) + await service._handle_event( + run.id, + "tool_call.started", + {"step_index": 0, "name": "echo", "arguments": {"n": 1}}, + ) + await service._handle_event( + run.id, + "tool_call.started", + {"step_index": 0, "name": "echo", "arguments": {"n": 2}}, + ) + # Legacy completed event without call_id → oldest pending. + await service._handle_event( + run.id, + "tool_call.completed", + {"step_index": 0, "name": "echo", "result": {"n": 1}}, + ) + + async with SessionLocal() as session: + step = ( + await session.execute( + select(Step).where(Step.run_id == run.id, Step.index == 0) + ) + ).scalar_one() + calls = ( + await session.execute( + select(ToolCall) + .where(ToolCall.step_id == step.id) + .order_by(ToolCall.created_at.asc()) + ) + ).scalars().all() + assert calls[0].result == {"n": 1} + assert calls[1].result is None + + +class _ParallelSameNameAdapter(OrchestratorAdapter): + name = "parallel_tools" + + async def run(self, ctx: AdapterContext) -> AdapterResult: + await ctx.emit_step_started(index=0, node="parallel") + + async def _one(label: str, delay: float) -> None: + call_id = await ctx.emit_tool_call_started( + step_index=0, + name="echo", + arguments={"text": label}, + ) + await asyncio.sleep(delay) + await ctx.emit_tool_call_completed( + step_index=0, + name="echo", + call_id=call_id, + result={"text": label}, + ) + + # Faster "b" finishes first; call_id must keep results on the right rows. + await asyncio.gather(_one("a", 0.05), _one("b", 0.01)) + await ctx.emit_step_completed(index=0, node="parallel", output={"ok": True}) + return AdapterResult(status=RunStatus.SUCCEEDED, output={"ok": True}) + + +@pytest.mark.asyncio +async def test_parallel_tool_emits_associate_via_executor_lock(): + from app.adapters import register_adapter + + adapter = _ParallelSameNameAdapter() + try: + register_adapter("parallel_tools", adapter) + except ValueError: + pass + + bus = get_event_bus() + async with SessionLocal() as session: + agent = await _make_agent(session, adapter="parallel_tools") + service = RunService(session=session, bus=bus) + run = await service.create_run( + RunCreate(agent_id=agent.id, input={"prompt": "x"}) + ) + run_id = run.id + + executor = RunExecutor(bus=bus, cancel_registry=InMemoryCancelRegistry()) + await executor.execute(run_id, "parallel_tools") + + async with SessionLocal() as session: + step = ( + await session.execute( + select(Step).where(Step.run_id == run_id, Step.index == 0) + ) + ).scalar_one() + calls = ( + await session.execute( + select(ToolCall).where(ToolCall.step_id == step.id) + ) + ).scalars().all() + assert len(calls) == 2 + results = {c.arguments["text"]: c.result for c in calls} + assert results == {"a": {"text": "a"}, "b": {"text": "b"}} + assert all(c.error is None for c in calls) diff --git a/docs/api-contract.md b/docs/api-contract.md index ab91c53..491bc74 100644 --- a/docs/api-contract.md +++ b/docs/api-contract.md @@ -212,6 +212,22 @@ before `step.completed`: { "index": 0, "tokens_in": 42, "tokens_out": 128, "cost_usd": 0.00012, "latency_ms": 1200 } ``` +`tool_call.started` / `tool_call.completed` share a stable ``call_id`` (ULID, +also the persisted ``ToolCall.id``) so parallel or same-name invocations +within one step associate correctly: + +```json +{ "step_index": 0, "name": "echo", "arguments": { "text": "hi" }, "call_id": "01HZ..." } +``` + +```json +{ "step_index": 0, "name": "echo", "call_id": "01HZ...", "result": { "text": "hi" }, "error": null, "latency_ms": 12 } +``` + +Adapters should pass the ``call_id`` returned from ``emit_tool_call_started`` +into ``emit_tool_call_completed``. When ``call_id`` is omitted on completed, +the runtime falls back to the oldest incomplete call with a matching name. + ## Schemas ### `Agent` diff --git a/docs/data-model.md b/docs/data-model.md index 4ef0563..a80c0dd 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -117,6 +117,10 @@ stateDiagram-v2 - **`Message.step_id` is optional** so an adapter can attach a message to a specific node tick when it makes sense, while keeping the run-level ordering authoritative. +- **`ToolCall.id` is the lifecycle association key.** SSE + ``tool_call.started`` / ``tool_call.completed`` carry the same ``call_id`` + (equal to ``ToolCall.id``) so parallel or same-name tool invocations within + one step do not clobber each other. ## Indexes diff --git a/frontend/lib/run-events.ts b/frontend/lib/run-events.ts index 4577fed..55c96a6 100644 --- a/frontend/lib/run-events.ts +++ b/frontend/lib/run-events.ts @@ -181,8 +181,12 @@ function appendMessage(messages: Message[], data: Record, at: s } function appendToolCall(step: Step, data: Record): ToolCall { + const callId = + typeof data.call_id === "string" && data.call_id + ? data.call_id + : `sse-tc-${step.index}-${step.tool_calls.length}`; return { - id: `sse-tc-${step.index}-${step.tool_calls.length}`, + id: callId, name: typeof data.name === "string" ? data.name : "tool", arguments: asRecord(data.arguments), result: null, @@ -191,33 +195,81 @@ function appendToolCall(step: Step, data: Record): ToolCall { }; } -function patchToolCall( +function isPendingToolCall(call: ToolCall): boolean { + return call.result == null && call.error == null; +} + +function applyToolCallPatch( + call: ToolCall, + data: Record, +): ToolCall { + return { + ...call, + result: + "result" in data + ? (data.result as Record | null) + : call.result, + error: typeof data.error === "string" ? data.error : call.error, + latency_ms: + typeof data.latency_ms === "number" ? data.latency_ms : call.latency_ms, + }; +} + +/** Always append on started so parallel / same-name tools stay distinct. */ +function startToolCall( + step: Step, + data: Record, + at: string, +): Step { + const callId = + typeof data.call_id === "string" && data.call_id ? data.call_id : null; + if (callId) { + const existing = step.tool_calls.findIndex((call) => call.id === callId); + if (existing !== -1) { + const toolCalls = [...step.tool_calls]; + toolCalls[existing] = { + ...toolCalls[existing], + name: typeof data.name === "string" ? data.name : toolCalls[existing].name, + arguments: + "arguments" in data + ? asRecord(data.arguments) + : toolCalls[existing].arguments, + }; + return { ...step, tool_calls: toolCalls, updated_at: at }; + } + } + return { + ...step, + tool_calls: [...step.tool_calls, appendToolCall(step, data)], + updated_at: at, + }; +} + +/** Match completed events by call_id, else oldest pending same-name call. */ +function completeToolCall( step: Step, data: Record, at: string, ): Step { - const name = typeof data.name === "string" ? data.name : ""; + const callId = + typeof data.call_id === "string" && data.call_id ? data.call_id : null; const toolCalls = [...step.tool_calls]; - let index = toolCalls.findIndex((call) => call.name === name); + let index = -1; + if (callId) { + index = toolCalls.findIndex((call) => call.id === callId); + } + if (index === -1) { + const name = typeof data.name === "string" ? data.name : ""; + index = toolCalls.findIndex( + (call) => call.name === name && isPendingToolCall(call), + ); + } if (index === -1) { toolCalls.push(appendToolCall(step, data)); index = toolCalls.length - 1; } - toolCalls[index] = { - ...toolCalls[index], - result: - "result" in data - ? (data.result as Record | null) - : toolCalls[index].result, - error: - typeof data.error === "string" ? data.error : toolCalls[index].error, - latency_ms: - typeof data.latency_ms === "number" - ? data.latency_ms - : toolCalls[index].latency_ms, - }; - + toolCalls[index] = applyToolCallPatch(toolCalls[index], data); return { ...step, tool_calls: toolCalls, updated_at: at }; } @@ -368,7 +420,7 @@ export function applyRunEvent(run: Run, event: RunEvent): Run { if (typeof stepIndex !== "number") return next; const step = findStep(run.steps, stepIndex); if (!step) return next; - const updated = patchToolCall(step, data, at); + const updated = startToolCall(step, data, at); return { ...next, steps: upsertStep(run.steps, updated), @@ -379,7 +431,7 @@ export function applyRunEvent(run: Run, event: RunEvent): Run { if (typeof stepIndex !== "number") return next; const step = findStep(run.steps, stepIndex); if (!step) return next; - const updated = patchToolCall(step, data, at); + const updated = completeToolCall(step, data, at); return { ...next, steps: upsertStep(run.steps, updated),