From a32d7a30397235ca8f665a24d4d8c654c5e32c96 Mon Sep 17 00:00:00 2001 From: Zexin Yuan Date: Tue, 30 Jun 2026 11:56:48 +0800 Subject: [PATCH] Make Executor ReAct loop append-only for prompt-prefix caching Rewrite Executor.run to seed [system, user(task)] once and append one assistant turn (via the shared reconstruct_react_text) plus one user observation turn per step, instead of rebuilding a single growing user message each iteration. The static system prompt and all prior turns now form a byte-stable prefix, so provider-side prefix caching hits on every iteration. Previously only the system block cached; the whole rebuilt user message (Previous-Steps JSON + Iteration counter + last response) was a cache miss every step, with hit rate declining as the loop grew. Folded correctness fixes (surfaced in plan-eng-review + /review): - _generate_final_summary: render the string action_summary directly instead of calling .get('name') on it (AttributeError on exhaustion). - tool_retry_counter: reset at run() start. Executors are reused across delegated tasks, so the counter leaked and could poison later runs. - max_syntax_errors=3 budget with reset-on-success: abort a model stuck emitting malformed ReAct instead of spamming up to 15 correction turns. The budget resets after any cleanly-parsed turn, so only 3 CONSECUTIVE bad turns abort (not 3 slips scattered across a long run). Extract reconstruct_react_text into copilotj/multiagent/react_format.py (LeaderAgent and Executor both import it; leader_multiagent imports Executor, so it could not live there without a circular import). Document in _build_enhanced_system_prompt that the tool text is load-bearing: the ReAct wrapper strips the tools= API param before the provider call, so the system-prompt text is the model's only view of the tools. Document pre-existing dead branch in _suggest_tool_based_on_context and deferred correction-path cache trade-off (codex P2 #2) Add copilotj/test/multiagent/test_executor.py (9 cases incl. a post-format prefix-stability cache-contract test that runs messages through the Anthropic formatter rather than asserting raw list-append). Full suite 206 passed; ruff clean. --- copilotj/multiagent/Executor.py | 173 +++++++---- copilotj/multiagent/leader_multiagent.py | 24 +- copilotj/multiagent/react_format.py | 41 +++ copilotj/test/multiagent/test_executor.py | 362 ++++++++++++++++++++++ 4 files changed, 516 insertions(+), 84 deletions(-) create mode 100644 copilotj/multiagent/react_format.py create mode 100644 copilotj/test/multiagent/test_executor.py diff --git a/copilotj/multiagent/Executor.py b/copilotj/multiagent/Executor.py index 4e873db3..39b3d674 100644 --- a/copilotj/multiagent/Executor.py +++ b/copilotj/multiagent/Executor.py @@ -2,11 +2,11 @@ # # SPDX-License-Identifier: Apache-2.0 -import json from typing import Any from copilotj.core import ChatAgent, ModelClient, ModelSyntaxError, TextMessage, Tool from copilotj.multiagent.leader_prompts import build_tool_prompt +from copilotj.multiagent.react_format import reconstruct_react_text __all__ = ["Executor"] @@ -27,11 +27,22 @@ def __init__( self.max_iterations = 15 self.tool_retry_counter = 0 self.max_tool_retry = 3 + # Mirror LeaderAgent's malformed-ReAct budget so a stuck model can't + # spam the prompt with correction turns (which would bloat the very + # prefix we want cache-stable). + self.max_syntax_errors = 3 self.system_prompt = self._build_enhanced_system_prompt(prompt) def _build_enhanced_system_prompt(self, base_prompt: str) -> str: - """Build system prompt that includes available tools information from config""" + """Build system prompt that includes available tools information from config. + + The tool descriptions embedded here are the ONLY way the model sees the + tools: the ReAct wrapper strips the ``tools=`` API param before the + provider call (see ``react_parser.py`` — "dont send tools since we are + parsing the response manually"). Do NOT remove this text expecting the + ``tools=`` param to cover it; it won't reach the provider. + """ if not self.tools: return base_prompt @@ -51,78 +62,107 @@ def _build_enhanced_system_prompt(self, base_prompt: str) -> str: return "\n".join((base_prompt, tools_info, tools_usage)) async def run(self, task: str) -> str: - """Execute the agent task with tool usage and reflection""" + """Execute the agent task with tool usage and reflection. + + Uses an append-only dialog — the static system prompt plus the initial + user task, then one assistant turn (reconstructed ReAct text) plus one + user observation turn per step — so the prompt prefix stays + byte-stable and provider-side prefix caching hits on every iteration + (only the newest turn is uncached). ``steps`` is a side-channel used + solely to build the fallback summary on loop exhaustion; it never + enters the prompt. + """ self.log_info(f"🟢 {self.name} is executing: {task}") self.log_info(f"📋 Available tools: {[t.name for t in self.tools]}") - try: - # Initialize conversation with system prompt and task - conversation_context = { - "task": task, - "status": "in_progress", - "last_tool_response": None, - "steps": [], - } + # Reset per-call counters. tool_retry_counter is instance state and the + # executor instance is reused across delegated tasks + # (LeaderAgent.delegate_task), so it must be reset here or one failed + # task can poison the next. + self.tool_retry_counter = 0 + syntax_error_counter = 0 + + steps: list[dict[str, Any]] = [] + messages: list[TextMessage] = [ + TextMessage(role="system", text=self.system_prompt), + TextMessage(role="user", text=task), + ] + try: for iteration in range(self.max_iterations): self.log_info(f"Iteration {iteration + 1}/{self.max_iterations}") - # Build prompt with context - execution_context = self._build_execution_context(conversation_context, iteration) - - # Get agent response try: - response = await self._create( - TextMessage(role="system", text=self.system_prompt), - TextMessage(role="user", text=execution_context), - tools=self.tools, - ) + response = await self._create(*messages, tools=self.tools) except ModelSyntaxError as e: self.log_error("Agent generated invalid ReAct syntax. Retrying...") - conversation_context["last_tool_response"] = e.message - - thought = ((e.chat_completion and e.chat_completion.reasoning_content) or "No thought provided",) - conversation_context["steps"].append( + syntax_error_counter += 1 + if syntax_error_counter >= self.max_syntax_errors: + return ( + f"❌ {self.name} aborted: too many invalid ReAct responses " + f"({syntax_error_counter}/{self.max_syntax_errors})." + ) + steps.append( { - "thought": thought, - "action": "", + "thought": getattr(e.chat_completion, "reasoning_content", None), "error": e.message, "iteration": iteration + 1, } ) + # NOTE(cache trade-off, deferred fix): appending this correction + # as a second consecutive `user` turn means the provider formatter + # (Anthropic/Gemini) merges same-role messages — it tacks this block + # onto the previous user message, mutating that message's content. So + # the prefix cached up to the previous observation no longer matches + # and THIS retry eats one cache miss (re-cached at the new trailing + # block; subsequent iterations resume cache hits). Rare (only on + # malformed ReAct) and self-healing, so not fixed here. The fix would + # be to first append the model's malformed output as an `assistant` + # turn (via reconstruct_react_text(e.chat_completion)) so roles + # alternate and no merge occurs. + messages.append(TextMessage(role="user", text=self._correction_text(e, iteration))) continue + # Turn parsed cleanly — reset the malformed-ReAct budget so only a + # truly stuck model (3 CONSECUTIVE bad turns) aborts, not 3 slips + # scattered across a long run. + syntax_error_counter = 0 + if not response.content and not response.tool_calls and not response.reasoning_content: await self.print_error("No response from agent") break - # Check for Final Answer + # Final answer? if response.content or self._is_task_complete(response.reasoning_content or ""): - conversation_context["steps"].append( + steps.append( { "thought": response.reasoning_content or "Task completed", "final_answer": response.content, "iteration": iteration + 1, } ) - conversation_context["status"] = "completed" return response.content or response.reasoning_content or "" - # Parse and execute action if present + # Record + append the assistant turn (frozen from here on). + assistant_text = reconstruct_react_text(response) + if assistant_text: + messages.append(TextMessage(role="assistant", text=assistant_text)) + + # No action produced → append a reflection/suggestion user turn. if not response.tool_calls: - # No valid action found, add reflection prompt suggestion = self._suggest_tool_based_on_context(response.reasoning_content or "", task) - conversation_context["last_tool_response"] = f"No valid action identified. {suggestion}" - conversation_context["steps"].append( + steps.append( { - "thought": response.reasoning_content or "No clear thought provided", + "thought": response.reasoning_content or "No clear thought", "reflection_needed": True, "tool_suggestion": suggestion, "iteration": iteration + 1, } ) + messages.append(TextMessage(role="user", text=self._reflection_text(suggestion, iteration))) continue + # Execute the tool, then append a user observation turn. tool_call = response.tool_calls[0] # TODO: handle multiple tool calls action_summary = f"{tool_call.tool.name} with args: {str(tool_call.args)[:100]}" self.log_info(f"🔧 Executing tool: {action_summary}...") @@ -130,8 +170,7 @@ async def run(self, task: str) -> str: tool_response = await self._call_tool(tool_call) self.tool_retry_counter = 0 # Reset on success - conversation_context["last_tool_response"] = tool_response - conversation_context["steps"].append( + steps.append( { "thought": response.reasoning_content or "No thought provided", "action": action_summary, @@ -139,6 +178,7 @@ async def run(self, task: str) -> str: "iteration": iteration + 1, } ) + messages.append(TextMessage(role="user", text=self._observation_text(tool_response, iteration))) except Exception as e: error_msg = f"❌ Error executing action: {str(e)}" @@ -148,8 +188,7 @@ async def run(self, task: str) -> str: if self.tool_retry_counter >= self.max_tool_retry: return f"❌ {self.name} failed after {self.max_tool_retry} attempts: {error_msg}" - conversation_context["last_tool_response"] = error_msg - conversation_context["steps"].append( + steps.append( { "thought": response.reasoning_content or "No thought provided", "action": action_summary, @@ -157,13 +196,38 @@ async def run(self, task: str) -> str: "iteration": iteration + 1, } ) + messages.append(TextMessage(role="user", text=self._observation_text(error_msg, iteration))) - return self._generate_final_summary(conversation_context) + return self._generate_final_summary({"task": task, "status": "in_progress", "steps": steps}) except Exception as e: self.log_error(f"Executor error: {e}") return f"❌ {self.name} encountered an error: {str(e)}" + def _observation_text(self, tool_response: str, iteration: int) -> str: + """User-turn text for a tool observation. + + Volatile content (the response, the step counter) lives only in this + newest turn, so the cached prefix (system + prior turns) is never + invalidated. + """ + return f"Observation:\n{tool_response}\n\nProgress: step {iteration + 1}/{self.max_iterations}" + + def _reflection_text(self, suggestion: str, iteration: int) -> str: + """User-turn text when the model produced a thought but no action.""" + return ( + f"{self._generate_reflection_prompt(iteration)}\n\n{suggestion}\n\n" + f"Progress: step {iteration + 1}/{self.max_iterations}" + ) + + def _correction_text(self, e: ModelSyntaxError, iteration: int) -> str: + """User-turn corrective prompt appended after a malformed ReAct response.""" + return ( + f"Your previous response was not valid ReAct format. Error: {e.message}\n" + f'Reply with Thought: then Action: {{"name": ..., "args": ...}} (or Final Answer:).\n' + f"Progress: step {iteration + 1}/{self.max_iterations}" + ) + def _suggest_tool_based_on_context(self, thought: str, task: str) -> str: """Suggest appropriate tool based on context and descriptions""" context = f"{thought} {task}".lower() @@ -171,6 +235,12 @@ def _suggest_tool_based_on_context(self, thought: str, task: str) -> str: # Use the tool descriptions from config to make suggestions suggested_tools = [] for tool in self.tools: + # FIXME(pre-existing): `tool.name` is a str tested against self.tools (a + # list[Tool]), so this is always False — the keyword-matching body below + # never runs, and this method always returns the generic "choose from all + # tools" fallback. Unrelated to caching; intentionally left unchanged in + # the append-only refactor. Likely meant `if tool in self.tools:` (a + # tautology) or to drop the guard entirely so matching actually fires. if tool.name in self.tools: # Check if context matches tool description keywords desc_words = tool.description.lower().split() @@ -192,28 +262,6 @@ def _suggest_tool_based_on_context(self, thought: str, task: str) -> str: TEMPLATE = 'Please choose from available tools: {{TOOLS}}. Format: {"name": "tool_name", "args": }' return TEMPLATE.replace("{{TOOLS}}", self._tool_names()) - def _build_execution_context(self, conversation_context: dict[str, Any], iteration: int) -> str: - """Build the execution context for the current iteration""" - context = f""" -Task: {conversation_context["task"]} - -Current Status: {conversation_context["status"]} -Iteration: {iteration + 1}/{self.max_iterations} - -Available Tools: {self._tool_names()} - -Previous Steps: -{json.dumps(conversation_context["steps"], indent=2) if conversation_context["steps"] else "No previous steps"} -""" - - if conversation_context["last_tool_response"]: - context += f"\n\nLast Tool Response:\n{conversation_context['last_tool_response']}" - - if iteration > 0: - context += f"\n\n{self._generate_reflection_prompt(iteration)}" - - return context - def _is_task_complete(self, response: str) -> bool: """Check if the task appears to be complete""" completion_indicators = [ @@ -273,7 +321,8 @@ def _generate_final_summary(self, conversation_context: dict[str, Any]) -> str: summary += f"Thought: {step['thought']}\n " if step.get("action"): - summary += f"Action: Used {step['action'].get('name', 'unknown')} tool\n " + # action is the string action_summary ("tool with args: ...") + summary += f"Action: {step['action']}\n " if step.get("response"): response = step["response"][:200] + "..." if len(step["response"]) > 200 else step["response"] diff --git a/copilotj/multiagent/leader_multiagent.py b/copilotj/multiagent/leader_multiagent.py index ec51b94e..c7c7024d 100644 --- a/copilotj/multiagent/leader_multiagent.py +++ b/copilotj/multiagent/leader_multiagent.py @@ -33,6 +33,7 @@ from copilotj.core.config import Config from copilotj.multiagent.agent_loader import load_agent_configs from copilotj.multiagent.Executor import Executor +from copilotj.multiagent.react_format import reconstruct_react_text from copilotj.multiagent.kb_tools import ( _load_macro_plugin_names, kb_build, @@ -76,27 +77,6 @@ SaveWorkflowHandler = Callable[[str, str | None, int | None, list[int] | None], Awaitable[str]] -def _reconstruct_react_text(response: ModelResponse) -> str: - """Rebuild an assistant-side ReAct text block from a parsed ModelResponse. - - The ReAct wrapper parses the model's raw text output into ``reasoning_content`` - (Thought), ``tool_calls`` (Action), and ``content`` (Final Answer). For - multi-turn conversation we re-assemble those parts into the same shape the - model originally produced, so the next turn's context matches the - ReAct-formatted examples in the system prompt. - """ - parts: list[str] = [] - if response.reasoning_content: - parts.append(f"Thought: {response.reasoning_content.strip()}") - if response.tool_calls: - tc = response.tool_calls[0] - args_json = json.dumps(tc.args.model_dump(), ensure_ascii=False) - parts.append(f'Action: {{"name": "{tc.tool.name}", "args": {args_json}}}') - if response.content: - parts.append(f"Final Answer: {response.content.strip()}") - return "\n".join(parts) - - class LeaderAgent(ChatAgent): def __init__( self, @@ -231,7 +211,7 @@ async def continue_dialog( parser's structured output) and a new user message carrying the tool observation plus refreshed ImageJ window info. """ - assistant_text = _reconstruct_react_text(prior_response) + assistant_text = reconstruct_react_text(prior_response) if assistant_text: self._dialog_messages.append(TextMessage(role="assistant", text=assistant_text)) diff --git a/copilotj/multiagent/react_format.py b/copilotj/multiagent/react_format.py new file mode 100644 index 00000000..e17622bd --- /dev/null +++ b/copilotj/multiagent/react_format.py @@ -0,0 +1,41 @@ +# SPDX-FileCopyrightText: Copyright contributors to the CopilotJ project. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Shared ReAct text (de)serialization helpers. + +Used by both :class:`copilotj.multiagent.leader_multiagent.LeaderAgent` and +:class:`copilotj.multiagent.Executor.Executor` to rebuild an assistant-side +ReAct text block from a parsed ``ModelResponse`` so the next turn's context +matches the ReAct-formatted examples in the system prompt. + +Lives in its own module so neither agent has to import the other (Executor is +imported by ``leader_multiagent``, which would create a circular import). +""" + +import json + +from copilotj.core import ModelResponse + +__all__ = ["reconstruct_react_text"] + + +def reconstruct_react_text(response: ModelResponse) -> str: + """Rebuild an assistant-side ReAct text block from a parsed ModelResponse. + + The ReAct wrapper parses the model's raw text output into ``reasoning_content`` + (Thought), ``tool_calls`` (Action), and ``content`` (Final Answer). For + multi-turn conversation we re-assemble those parts into the same shape the + model originally produced, so the next turn's context matches the + ReAct-formatted examples in the system prompt. + """ + parts: list[str] = [] + if response.reasoning_content: + parts.append(f"Thought: {response.reasoning_content.strip()}") + if response.tool_calls: + tc = response.tool_calls[0] + args_json = json.dumps(tc.args.model_dump(), ensure_ascii=False) + parts.append(f'Action: {{"name": "{tc.tool.name}", "args": {args_json}}}') + if response.content: + parts.append(f"Final Answer: {response.content.strip()}") + return "\n".join(parts) diff --git a/copilotj/test/multiagent/test_executor.py b/copilotj/test/multiagent/test_executor.py new file mode 100644 index 00000000..3dc4b1d4 --- /dev/null +++ b/copilotj/test/multiagent/test_executor.py @@ -0,0 +1,362 @@ +# SPDX-FileCopyrightText: Copyright contributors to the CopilotJ project. +# +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for ``Executor.run`` — the append-only ReAct loop. + +Drives async code with ``asyncio.run`` (no pytest-asyncio), mirroring +``test_agent_retry.py``. A scripted ``ModelClient`` pops one canned response +per ``create_stream`` call and records the messages it was handed, so we can +assert both control flow and the prefix-cache contract. +""" + +import asyncio +from collections.abc import AsyncGenerator, Sequence +from typing import Any, override + +import anthropic +import pytest + +from copilotj.core.message import ImageMessage, TextMessage +from copilotj.core.model_client import ModelClient, ModelResponseChunk, ModelSyntaxError, ToolCall +from copilotj.core.model_client.anthropic import AnthropicChatCompletionClient +from copilotj.core.tool import FunctionTool, Tool +from copilotj.multiagent.Executor import Executor + +# Sentinel script: raise a malformed-ReAct error on this create_stream call. +RAISE_SYNTAX = object() + + +# --------------------------------------------------------------------- helpers + + +def _thought(text: str) -> ModelResponseChunk: + return ModelResponseChunk(reasoning_content=text, content=None, finish_reason=None) + + +def _final(text: str) -> ModelResponseChunk: + return ModelResponseChunk(reasoning_content=None, content=text, finish_reason="stop") + + +def _make_tool(*, fail_remaining: int = 0) -> tuple[FunctionTool, dict[str, int]]: + """Build a fake ``lookup(query)`` tool with observable, mutable state. + + ``fail_remaining`` makes the next N calls raise, then succeed — for the + retry / cross-task-reset tests. + """ + state = {"calls": 0, "fail": fail_remaining} + + def fn(query: str) -> str: + state["calls"] += 1 + if state["fail"] > 0: + state["fail"] -= 1 + raise RuntimeError("tool boom") + return f"ok:{query}" + + return FunctionTool(fn, "test lookup tool", name="lookup"), state + + +def _tc(tool: FunctionTool, query: str = "cells") -> ToolCall: + return ToolCall(id="tc1", tool=tool, args=tool.args_type()(query=query)) + + +class _ScriptedClient(ModelClient): + """Pops one canned script per ``create_stream`` call; records all messages. + + A script is either the ``RAISE_SYNTAX`` sentinel or a list of + ``ModelResponseChunk`` / ``ToolCall`` items to yield in order. + """ + + def __init__(self, scripts: list[Any]) -> None: + self._scripts = list(scripts) + self._idx = 0 + self.recorded: list[list[TextMessage | ImageMessage]] = [] + + @override + def get_model(self) -> str: + return "scripted" + + @override + def get_api_key(self) -> str | None: + return None + + @override + async def create( + self, + messages: Sequence[TextMessage | ImageMessage], + *, + tools: list[Tool] | None = None, + extra_args: dict[str, Any] | None = None, + ) -> Any: + raise NotImplementedError + + @override + async def create_stream( + self, + messages: Sequence[TextMessage | ImageMessage], + *, + tools: list[Tool] | None = None, + extra_args: dict[str, Any] | None = None, + ) -> AsyncGenerator[ModelResponseChunk | ToolCall, None]: + self.recorded.append(list(messages)) + script = self._scripts[self._idx] + self._idx += 1 + if script is RAISE_SYNTAX: + raise ModelSyntaxError("malformed ReAct output") + for chunk in script: + yield chunk + + +class _RecordingRuntime: + """Duck-typed Runtime: no-ops everywhere, records tool results.""" + + def __init__(self) -> None: + self.tool_results: list[tuple[str, str]] = [] + + async def update_current_agent(self, agent: str) -> None: # noqa: ARG002 + pass + + async def print_chat(self, agent: str, message: Any) -> None: # noqa: ARG002 + pass + + async def print_info(self, agent: str, message: str) -> None: # noqa: ARG002 + pass + + async def print_error(self, agent: str, message: str) -> None: # noqa: ARG002 + pass + + async def print_retry(self, agent: str, info: Any) -> None: # noqa: ARG002 + pass + + async def print_tool_called(self, agent: str, tool_call_id: str) -> None: # noqa: ARG002 + pass + + async def print_tool_call_result(self, agent: str, tool_call_id: str, status: str, result: str) -> None: # noqa: ARG002 + self.tool_results.append((status, str(result))) + + async def print_handoff(self, agent: str, handoff: Any) -> None: # noqa: ARG002 + pass + + def log_info(self, message: str) -> None: # noqa: ARG002 + pass + + def log_error(self, message: str) -> None: # noqa: ARG002 + pass + + +def _make_executor(client: _ScriptedClient, tools: list[Tool]) -> Executor: + ex = Executor( + name="Tool Agent", description="desc", prompt="You are a test agent.", tools=tools, model_client=client + ) + ex._set_runtime(_RecordingRuntime()) # noqa: SLF001 + return ex + + +def _content_sig(messages: Sequence[TextMessage | ImageMessage]) -> str: + """Content signature of the provider-formatted payload (Anthropic format). + + Runs the messages through ``AnthropicChatCompletionClient._format_messages`` + and concatenates system + each message's text. Cache-control markers are + ignored (only ``text`` is read), so a stable content prefix is detectable + even though the trailing breakpoint marker moves each call. + """ + system, msgs = AnthropicChatCompletionClient._format_messages(messages) # noqa: SLF001 + parts: list[str] = [] + if system is not anthropic.NOT_GIVEN: + for block in system: + parts.append("SYSTEM:" + block.get("text", "")) + for m in msgs: + text = "".join(b.get("text", "") for b in m["content"] if b.get("type") == "text") + parts.append(f"{m['role']}:{text}") + return "\n".join(parts) + + +# --------------------------------------------------------------------- tests + + +def test_happy_path_tool_then_final(): + tool, state = _make_tool() + client = _ScriptedClient([[_thought("need info"), _tc(tool)], [_final("Final Answer: 42")]]) + ex = _make_executor(client, [tool]) + + result = asyncio.run(ex.run("count the cells")) + + assert result == "Final Answer: 42" + assert state["calls"] == 1 # tool executed exactly once + + +def test_append_only_cache_contract(): + """The post-format payload prefix must be byte-stable across iterations. + + Codex #7: asserting Python list-append is not enough — the provider + formatter merges consecutive same-role turns, so we verify stability at the + formatted boundary. + """ + tool, _ = _make_tool() + client = _ScriptedClient( + [ + [_thought("step 1"), _tc(tool)], + [_thought("step 2"), _tc(tool)], + [_final("Final Answer: done")], + ] + ) + ex = _make_executor(client, [tool]) + + asyncio.run(ex.run("multi-step task")) + + assert len(client.recorded) == 3 + # Raw TextMessage list is append-only: each call's prefix is frozen. + assert client.recorded[0] == client.recorded[1][: len(client.recorded[0])] + assert client.recorded[1] == client.recorded[2][: len(client.recorded[1])] + # Formatted content prefix is stable (system text identical + growing prefix). + sigs = [_content_sig(m) for m in client.recorded] + assert sigs[1].startswith(sigs[0]) + assert sigs[2].startswith(sigs[1]) + # System text is identical across all calls. + assert sigs[0].split("\n")[0] == sigs[1].split("\n")[0] == sigs[2].split("\n")[0] + + +def test_tool_error_then_retry_then_success(): + tool, state = _make_tool(fail_remaining=1) # first call raises, second succeeds + client = _ScriptedClient( + [ + [_thought("try once"), _tc(tool)], # tool raises + [_thought("try again"), _tc(tool)], # tool succeeds + [_final("Final Answer: recovered")], + ] + ) + ex = _make_executor(client, [tool]) + + result = asyncio.run(ex.run("flaky task")) + + assert result == "Final Answer: recovered" + assert state["calls"] == 2 + + +def test_exhaustion_returns_summary(): + """Loop exhaustion renders the summary without crashing (F2 fix). + + Without the fix, ``_generate_final_summary`` calls ``.get('name')`` on the + string ``action_summary`` and raises AttributeError. + """ + tool, _ = _make_tool() + client = _ScriptedClient([[_thought("a"), _tc(tool)], [_thought("b"), _tc(tool)]]) + ex = _make_executor(client, [tool]) + ex.max_iterations = 2 # noqa: SLF001 — force exhaustion after 2 steps + + result = asyncio.run(ex.run("never finishes")) + + assert "Task Summary" in result + assert "❌" not in result # not the top-level error path + + +def test_retry_counter_reset_across_runs(): + """C1: tool_retry_counter must reset per run() — executors are reused. + + Run 1 exhausts retries (counter left at 3 in the old code). Run 2 must not + be poisoned: its first failure should NOT immediately abort. + + Both runs share one event loop because the instance-level ``_abort_event`` + (an ``asyncio.Event``) is loop-bound — matching the real lifecycle where + ``LeaderAgent.delegate_task`` reuses the executor on a single loop. + """ + tool, _ = _make_tool(fail_remaining=99) # run 1: always fails + client = _ScriptedClient( + [ + [_thought("a"), _tc(tool)], + [_thought("b"), _tc(tool)], + [_thought("c"), _tc(tool)], + ] + ) + ex = _make_executor(client, [tool]) + + # Run 2: tool fails once, then succeeds, then final answer. + tool2, _ = _make_tool(fail_remaining=1) + client2 = _ScriptedClient([[_thought("x"), _tc(tool2)], [_final("Final Answer: ok")]]) + + async def _both() -> tuple[str, str]: + r1 = await ex.run("doomed task") + # Swap in run 2's client/tool; system_prompt depends on the tool set. + ex._client = client2 # noqa: SLF001 + ex.tools = [tool2] # noqa: SLF001 + ex.system_prompt = ex._build_enhanced_system_prompt("You are a test agent.") # noqa: SLF001 + r2 = await ex.run("recovering task") + return r1, r2 + + r1, r2 = asyncio.run(_both()) + + assert "failed after 3 attempts" in r1 # run 1 exhausted retries (counter now 3) + assert r2 == "Final Answer: ok" # run 2 not poisoned by run 1's counter + + +def test_syntax_budget_aborts(): + """C2: repeated malformed ReAct aborts after max_syntax_errors, not 15.""" + client = _ScriptedClient([RAISE_SYNTAX] * 6) + ex = _make_executor(client, []) + + result = asyncio.run(ex.run("stuck task")) + + assert "aborted: too many invalid ReAct responses" in result + assert len(client.recorded) == ex.max_syntax_errors # stopped at 3, not 15 + + +def test_syntax_budget_resets_on_success(): + """C2 refinement: a successful turn resets the budget, so 3 SCATTERED slips + (not 3 consecutive) do not abort a long run. Without the reset, the 3rd + malformed turn below would abort despite the recovery in between.""" + tool, _ = _make_tool() + client = _ScriptedClient( + [ + RAISE_SYNTAX, + RAISE_SYNTAX, + [_thought("recovered"), _tc(tool)], # success → resets budget to 0 + RAISE_SYNTAX, + [_final("Final Answer: ok")], + ] + ) + ex = _make_executor(client, [tool]) + + result = asyncio.run(ex.run("task")) + + assert result == "Final Answer: ok" # not aborted despite 3 total malformed turns + + +def test_syntax_error_correction_recovers(): + """A malformed response appends a correction turn, then the model recovers.""" + tool, _ = _make_tool() + client = _ScriptedClient( + [ + RAISE_SYNTAX, + [_thought("fixed"), _tc(tool)], + [_final("Final Answer: good")], + ] + ) + ex = _make_executor(client, [tool]) + + result = asyncio.run(ex.run("task")) + + assert result == "Final Answer: good" + # A correction user-turn was appended before the recovery call. + assert any("not valid ReAct format" in getattr(m, "text", "") for m in client.recorded[1]) + + +def test_reflection_no_action_path(): + """A thought without an action appends a reflection turn, then finishes.""" + tool, _ = _make_tool() + client = _ScriptedClient( + [ + [_thought("just thinking, no action yet")], # no tool_calls, no content + [_final("Final Answer: decided")], + ] + ) + ex = _make_executor(client, [tool]) + + result = asyncio.run(ex.run("task")) + + assert result == "Final Answer: decided" + # A reflection user-turn was appended before the final-answer call. + assert any("reflect on your progress" in getattr(m, "text", "") for m in client.recorded[1]) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])