Skip to content

fix(run_agent): 7 bug fixes — db flush, interrupt, tool error handling, retry loop #76

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
0xbyt4 wants to merge 1 commit into from
Closed

fix(run_agent): 7 bug fixes — db flush, interrupt, tool error handling, retry loop #76

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 22 additions & 10 deletions run_agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -581,7 +581,7 @@ def _flush_messages_to_session_db(self, messages: List[Dict], conversation_histo
if not self._session_db:
return
try:
start_idx = (len(conversation_history) if conversation_history else 0) + 1
start_idx = len(conversation_history) if conversation_history else 0
for msg in messages[start_idx:]:
role = msg.get("role", "unknown")
content = msg.get("content")
Expand Down Expand Up @@ -929,7 +929,6 @@ def _clean_session_content(content: str) -> str:
return content
content = convert_scratchpad_to_think(content)
# Strip extra newlines before/after think blocks
import re
content = re.sub(r'\n+(<think>)', r'\n\1', content)
content = re.sub(r'(</think>)\n+', r'\1\n', content)
return content.strip()
Expand Down Expand Up @@ -1050,8 +1049,7 @@ def _hydrate_todo_store(self, history: List[Dict[str, Any]]) -> None:
self._todo_store.write(last_todo_response, merge=False)
if not self.quiet_mode:
print(f"{self.log_prefix}📋 Restored {len(last_todo_response)} todo item(s) from history")
_set_interrupt(False)


@property
def is_interrupted(self) -> bool:
"""Check if an interrupt has been requested."""
Expand Down Expand Up @@ -1290,7 +1288,8 @@ def flush_memories(self, messages: list = None, min_turns: int = None):
"[System: The session is being compressed. "
"Please save anything worth remembering to your memories.]"
)
flush_msg = {"role": "user", "content": flush_content}
_sentinel = f"__flush_{id(self)}_{time.monotonic()}"
flush_msg = {"role": "user", "content": flush_content, "_flush_sentinel": _sentinel}
messages.append(flush_msg)

try:
Expand Down Expand Up @@ -1352,10 +1351,13 @@ def flush_memories(self, messages: list = None, min_turns: int = None):
except Exception as e:
logger.debug("Memory flush API call failed: %s", e)
finally:
# Strip flush artifacts: remove everything from the flush message onward
while messages and messages[-1] is not flush_msg and len(messages) > 0:
# Strip flush artifacts: remove everything from the flush message onward.
# Use sentinel marker instead of identity check for robustness.
while messages and messages[-1].get("_flush_sentinel") != _sentinel:
messages.pop()
if messages and messages[-1] is flush_msg:
if not messages:
break
if messages and messages[-1].get("_flush_sentinel") == _sentinel:
messages.pop()

def _compress_context(self, messages: list, system_message: str, *, approx_tokens: int = None) -> tuple:
Expand Down Expand Up @@ -1547,12 +1549,19 @@ def _execute_tool_calls(self, assistant_message, messages: list, effective_task_
try:
function_result = handle_function_call(function_name, function_args, effective_task_id)
_spinner_result = function_result
except Exception as tool_error:
function_result = f"Error executing tool '{function_name}': {tool_error}"
logger.error("handle_function_call raised for %s: %s", function_name, tool_error)
finally:
tool_duration = time.time() - tool_start_time
cute_msg = _get_cute_tool_message_impl(function_name, function_args, tool_duration, result=_spinner_result)
spinner.stop(cute_msg)
else:
function_result = handle_function_call(function_name, function_args, effective_task_id)
try:
function_result = handle_function_call(function_name, function_args, effective_task_id)
except Exception as tool_error:
function_result = f"Error executing tool '{function_name}': {tool_error}"
logger.error("handle_function_call raised for %s: %s", function_name, tool_error)
tool_duration = time.time() - tool_start_time

result_preview = function_result[:200] if len(function_result) > 200 else function_result
Expand Down Expand Up @@ -1859,7 +1868,7 @@ def run_conversation(
retry_count = 0
max_retries = 6 # Increased to allow longer backoff periods

while retry_count <= max_retries:
while retry_count < max_retries:
try:
api_kwargs = self._build_api_kwargs(api_messages)

Expand Down Expand Up @@ -1953,6 +1962,7 @@ def run_conversation(
if self._interrupt_requested:
print(f"{self.log_prefix}⚡ Interrupt detected during retry wait, aborting.")
self._persist_session(messages, conversation_history)
self.clear_interrupt()
return {
"final_response": "Operation interrupted.",
"messages": messages,
Expand Down Expand Up @@ -2055,6 +2065,7 @@ def run_conversation(
if self._interrupt_requested:
print(f"{self.log_prefix}⚡ Interrupt detected during error handling, aborting retries.")
self._persist_session(messages, conversation_history)
self.clear_interrupt()
return {
"final_response": "Operation interrupted.",
"messages": messages,
Expand Down Expand Up @@ -2142,6 +2153,7 @@ def run_conversation(
if self._interrupt_requested:
print(f"{self.log_prefix}⚡ Interrupt detected during retry wait, aborting.")
self._persist_session(messages, conversation_history)
self.clear_interrupt()
return {
"final_response": "Operation interrupted.",
"messages": messages,
Expand Down