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