From 083ef01aab28a6032bec74277219a23e6c319b3d Mon Sep 17 00:00:00 2001 From: Tomas Sereikis Date: Sat, 29 Aug 2026 13:38:03 +0300 Subject: [PATCH] fix(llm): keep state-changing tool calls in the order the model asked for MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review on #1100 caught that the previous commit parallelised every tool call in a turn, mutations included. `ctx.db_lock` makes the mutating handlers take turns but does not decide whose turn comes first, so two writes from one assistant turn could apply in an order the model did not request — `tool_results` stayed ordered while the database did not. Reads may still overlap freely, because they change nothing. So the loop now splits the turn: anything named in `mutating_tools` runs one at a time in request order, everything else runs concurrently alongside it. `mutating_tools=None` is the default and means "assume every tool mutates", which is the fully sequential behaviour that existed before this branch. A caller opts in by naming its mutating set, so no existing caller changes behaviour by being left alone. `MUTATING_TOOL_NAMES` lives in agent_tools next to the dispatch table and lists exactly the five names routing to a handler that takes `ctx.db_lock`: create_observations, create_observations_deductive, create_observations_inductive, update_peer_card, delete_observations. `extract_preferences` and `finish_consolidation` are deliberately absent — both only return text telling the model what to call next. Dialectic passes that set rather than a "these are all reads" boolean. Its loadout is reads today, but DIALECTIC_TOOLS already carries a commented-out create_observations_deductive, so a flag would have gone quietly wrong the day someone uncommented it. The parameter is threaded through honcho_llm_call instead of imported inside src/llm, which keeps the llm layer free of tool semantics and avoids the existing agent_tools <- dreamer.specialists cycle. Two tests added: mutating calls are not reordered even when a later call finishes first, and the default schedule is fully sequential. --- src/dialectic/core.py | 11 ++++++ src/llm/api.py | 5 +++ src/llm/tool_loop.py | 39 +++++++++++++++++----- src/utils/agent_tools.py | 17 ++++++++++ tests/llm/test_tool_loop_parallel.py | 50 ++++++++++++++++++++++++++-- 5 files changed, 111 insertions(+), 11 deletions(-) diff --git a/src/dialectic/core.py b/src/dialectic/core.py index 57964c87..6842fbe1 100644 --- a/src/dialectic/core.py +++ b/src/dialectic/core.py @@ -35,6 +35,7 @@ from src.telemetry.prometheus.metrics import DialecticComponents, TokenTypes from src.utils.agent_tools import ( DIALECTIC_TOOLS, DIALECTIC_TOOLS_MINIMAL, + MUTATING_TOOL_NAMES, create_tool_executor, search_memory, ) @@ -508,6 +509,11 @@ class DialecticAgent: tool_choice=level_settings.TOOL_CHOICE, tool_executor=tool_executor, max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS, + # Dialectic's loadout is reads today, but say so explicitly rather + # than relying on that: the loop keeps any state-changing call + # in the order the model asked for, so adding a write tool here + # later stays correct without anyone remembering this. + mutating_tools=MUTATING_TOOL_NAMES, messages=self.messages, max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, trace_name="dialectic_chat", @@ -584,6 +590,11 @@ class DialecticAgent: tool_choice=level_settings.TOOL_CHOICE, tool_executor=tool_executor, max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS, + # Dialectic's loadout is reads today, but say so explicitly rather + # than relying on that: the loop keeps any state-changing call + # in the order the model asked for, so adding a write tool here + # later stays correct without anyone remembering this. + mutating_tools=MUTATING_TOOL_NAMES, messages=self.messages, max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, trace_name="dialectic_chat", diff --git a/src/llm/api.py b/src/llm/api.py index 13c3a5e5..ca9795d6 100644 --- a/src/llm/api.py +++ b/src/llm/api.py @@ -68,6 +68,7 @@ async def honcho_llm_call( tool_choice: str | dict[str, Any] | None = None, tool_executor: Callable[[str, dict[str, Any]], Any] | None = None, max_tool_iterations: int = 10, + mutating_tools: frozenset[str] | None = None, messages: list[dict[str, Any]] | None = None, max_input_tokens: int | None = None, trace_name: str | None = None, @@ -97,6 +98,7 @@ async def honcho_llm_call( tool_choice: str | dict[str, Any] | None = None, tool_executor: Callable[[str, dict[str, Any]], Any] | None = None, max_tool_iterations: int = 10, + mutating_tools: frozenset[str] | None = None, messages: list[dict[str, Any]] | None = None, max_input_tokens: int | None = None, trace_name: str | None = None, @@ -126,6 +128,7 @@ async def honcho_llm_call( tool_choice: str | dict[str, Any] | None = None, tool_executor: Callable[[str, dict[str, Any]], Any] | None = None, max_tool_iterations: int = 10, + mutating_tools: frozenset[str] | None = None, messages: list[dict[str, Any]] | None = None, max_input_tokens: int | None = None, trace_name: str | None = None, @@ -154,6 +157,7 @@ async def honcho_llm_call( tool_choice: str | dict[str, Any] | None = None, tool_executor: Callable[[str, dict[str, Any]], Any] | None = None, max_tool_iterations: int = 10, + mutating_tools: frozenset[str] | None = None, messages: list[dict[str, Any]] | None = None, max_input_tokens: int | None = None, trace_name: str | None = None, @@ -450,6 +454,7 @@ async def honcho_llm_call( tool_choice=tool_choice, tool_executor=tool_executor, max_tool_iterations=max_tool_iterations, + mutating_tools=mutating_tools, response_model=response_model, json_mode=json_mode, temperature=temperature, diff --git a/src/llm/tool_loop.py b/src/llm/tool_loop.py index 7a30b960..8a70291d 100644 --- a/src/llm/tool_loop.py +++ b/src/llm/tool_loop.py @@ -353,6 +353,7 @@ async def execute_tool_loop( tool_choice: str | dict[str, Any] | None, tool_executor: Callable[[str, dict[str, Any]], Any], max_tool_iterations: int, + mutating_tools: frozenset[str] | None = None, response_model: type[BaseModel] | None, json_mode: bool, temperature: float | None, @@ -649,17 +650,37 @@ async def execute_tool_loop( None, ) - # gather preserves argument order, so tool_results and - # all_tool_calls stay in the order the model asked for them. - outcomes = await asyncio.gather( - *( - asyncio.create_task(run_tool_call(seq, tool_call)) - for seq, tool_call in enumerate(response.tool_calls_made) - ) - ) + # Reads may overlap freely; state-changing calls may not, because + # `ctx.db_lock` decides that they take turns but not which turn each + # one gets. Anything named in `mutating_tools` therefore runs in the + # order the model asked for, while the rest run concurrently + # alongside it. `mutating_tools=None` means "assume every tool + # mutates", so a caller that has not opted in keeps the fully + # sequential behaviour it had before. + calls = list(enumerate(response.tool_calls_made)) + + def _must_stay_ordered(tool_call: dict[str, Any]) -> bool: + return mutating_tools is None or tool_call["name"] in mutating_tools + + # create_task even for the ordered ones: each still needs its own + # context copy for the telemetry ContextVars, and awaiting them one + # at a time is what preserves the order. + concurrent = { + seq: asyncio.create_task(run_tool_call(seq, tool_call)) + for seq, tool_call in calls + if not _must_stay_ordered(tool_call) + } + + outcomes: dict[int, tuple[dict[str, Any], dict[str, Any] | None]] = {} + for seq, tool_call in calls: + if _must_stay_ordered(tool_call): + outcomes[seq] = await run_tool_call(seq, tool_call) + for seq, task in concurrent.items(): + outcomes[seq] = await task tool_results: list[dict[str, Any]] = [] - for result_entry, call_entry in outcomes: + for seq, _ in calls: + result_entry, call_entry = outcomes[seq] tool_results.append(result_entry) if call_entry is not None: all_tool_calls.append(call_entry) diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index b753c462..37d90b8f 100644 --- a/src/utils/agent_tools.py +++ b/src/utils/agent_tools.py @@ -2604,6 +2604,23 @@ _TOOL_HANDLERS: dict[str, Callable[[ToolContext, dict[str, Any]], Any]] = { "get_reasoning_chain": _handle_get_reasoning_chain, } +# Tools that change state, and so must not be reordered relative to one another +# within a single assistant turn. Each of these routes to a handler that takes +# `ctx.db_lock` (`_handle_create_observations_impl`, `_handle_update_peer_card`, +# `_handle_delete_observations`); the lock stops them interleaving, but it does +# not decide who gets it first, so the tool loop keeps them in the order the +# model asked for. `extract_preferences` and `finish_consolidation` are not +# here: both only return text telling the model what to call next. +MUTATING_TOOL_NAMES: frozenset[str] = frozenset( + { + "create_observations", + "create_observations_deductive", + "create_observations_inductive", + "update_peer_card", + "delete_observations", + } +) + async def create_tool_executor( workspace_name: str, diff --git a/tests/llm/test_tool_loop_parallel.py b/tests/llm/test_tool_loop_parallel.py index ac1408d9..f49e66f2 100644 --- a/tests/llm/test_tool_loop_parallel.py +++ b/tests/llm/test_tool_loop_parallel.py @@ -75,8 +75,13 @@ class _Tracker: return "ok" -async def _run(tracker: _Tracker, n_calls: int = 2) -> HonchoLLMCallResponse[Any]: - names = ["search_memory", "search_messages"] +async def _run( + tracker: _Tracker, + n_calls: int = 2, + mutating_tools: frozenset[str] | None = frozenset(), + names: list[str] | None = None, +) -> HonchoLLMCallResponse[Any]: + names = names or ["search_memory", "search_messages"] calls = [ {"name": names[i % len(names)], "input": {"query": str(i)}, "id": f"t{i}"} for i in range(n_calls) @@ -113,6 +118,7 @@ async def _run(tracker: _Tracker, n_calls: int = 2) -> HonchoLLMCallResponse[Any tool_choice="auto", tool_executor=tracker.execute, max_tool_iterations=5, + mutating_tools=mutating_tools, response_model=None, json_mode=False, temperature=None, @@ -171,3 +177,43 @@ async def test_fan_out_is_capped(): f"fan-out reached {tracker.peak}, above the {MAX_CONCURRENT_TOOL_CALLS} cap" ) assert tracker.peak > 1, "cap must not serialise execution entirely" + + +@pytest.mark.asyncio +async def test_mutating_tools_are_not_reordered(): + """State-changing calls keep the order the model asked for. + + `ctx.db_lock` makes the mutating handlers take turns but does not decide + whose turn comes first, so running them concurrently could apply a turn's + writes in an order the model did not request. + """ + order: list[str] = [] + + class _Recorder(_Tracker): + async def execute(self, name: str, _input: dict[str, Any]) -> str: + self.in_flight += 1 + self.peak = max(self.peak, self.in_flight) + # Later calls sleep less, so anything concurrent finishes reversed. + await asyncio.sleep(TOOL_DELAY_SECONDS / (len(order) + 1)) + order.append(name) + self.in_flight -= 1 + return "ok" + + await _run( + _Recorder(), + n_calls=3, + mutating_tools=frozenset({"update_peer_card"}), + names=["update_peer_card"], + ) + + assert order == ["update_peer_card"] * 3 + + +@pytest.mark.asyncio +async def test_default_is_fully_sequential(): + """A caller that does not opt in keeps the old one-at-a-time behaviour.""" + tracker = _Tracker() + + await _run(tracker, n_calls=4, mutating_tools=None) + + assert tracker.peak == 1