MervinPraison/PraisonAI
1
0 Fork 0
mirror of https://github.com/MervinPraison/PraisonAI.git synced 2026-11-07 22:52:58 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

fix: resolve all 8 hard blockers and CodeRabbit issues

Browse Source 
- B1: Wire error classifier into retry paths with proper attempt handling
- B2: Add interrupt controller checks to sync run_autonomous()
- B3: Fix silent failure - fail closed when path overlap unavailable
- B4: Enhanced path overlap detection with custom write tool support
- B5: Skip empty path strings (already implemented)
- B6: Fix SessionMessage attribute access + async I/O blocking
- B7: Handle asyncio.run() in running loop context
- B8: Implement recursive sanitization for nested tool call arguments

Additional fixes:
- Add missing type imports (InterruptController)
- Replace undefined MemoryManager with Any
- Fix test determinism with threading.Event coordination
- Add missing __all__ exports for error_classifier
- Add InterruptControllerProtocol for extension point

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Claude <[email protected]>
This commit is contained in:
praisonai-triage-agent[bot]andClaude committed 2026-04-21 10:20:53 +00:00
1 parent af66f595d3
commit 76264e280c
10 files changed
+217 -95
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/praisonai-agents/praisonaiagents/agent/agent.py
+16 -2
View File
@@ -245,6 +245,7 @@ if TYPE_CHECKING:
from ..context.models import ContextConfig
from ..context.manager import ContextManager
from ..knowledge.knowledge import Knowledge
from .interrupt import InterruptController
from ..agent.autonomy import AutonomyConfig
from ..task.task import Task
from .handoff import Handoff, HandoffConfig, HandoffResult
@@ -531,7 +532,7 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
# CONSOLIDATED FEATURE PARAMS (agent-centric API)
# Each follows: False=disabled, True=defaults, Config=custom
# ============================================================
memory: Optional[Union[bool, str, 'MemoryConfig', 'MemoryManager']] = None,
memory: Optional[Union[bool, str, 'MemoryConfig', Any]] = None,
knowledge: Optional[Union[bool, str, List[str], 'KnowledgeConfig', 'Knowledge']] = None,
planning: Optional[Union[bool, str, 'PlanningConfig']] = False,
reflection: Optional[Union[bool, str, 'ReflectionConfig']] = None,
@@ -576,7 +577,7 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
memory: Memory system configuration. Accepts:
- bool: True enables defaults, False disables
- MemoryConfig: Custom configuration
- MemoryManager: Pre-configured instance
- Any: Pre-configured memory instance
knowledge: Knowledge sources. Accepts:
- bool: True enables defaults
- List[str]: File paths, URLs, or text content
@@ -2838,6 +2839,19 @@ Summary:"""
started_at=started_at,
)
# G2: Check for interrupt request (cooperative cancellation) - sync version
if self.interrupt_controller and self.interrupt_controller.is_set():
reason = self.interrupt_controller.reason or "unknown"
return AutonomyResult(
success=False,
output=f"Task interrupted: {reason}",
completion_reason="interrupted",
iterations=iterations,
stage=stage,
actions=actions_taken,
duration_seconds=time_module.time() - start_time,
started_at=started_at,
)
# Execute one turn using the agent's chat method
# Always use the original prompt (prompt re-injection)
src/praisonai-agents/praisonaiagents/agent/interrupt.py
+27 -2
View File
@@ -6,10 +6,35 @@ operations. Follows protocol-driven design with zero overhead when not used.
"""
import threading
from typing import Optional
from typing import Optional, Protocol
from dataclasses import dataclass, field
__all__ = ["InterruptController"]
__all__ = ["InterruptControllerProtocol", "InterruptController"]
class InterruptControllerProtocol(Protocol):
"""Protocol for interrupt controller extension point."""
def request(self, reason: str = "user") -> None:
"""Request cancellation with optional reason."""
...
def clear(self) -> None:
"""Clear interrupt state."""
...
def is_set(self) -> bool:
"""Check if interrupt was requested."""
...
@property
def reason(self) -> Optional[str]:
"""Get interrupt reason if set."""
...
def check(self) -> None:
"""Check for interrupt and raise if set."""
...
@dataclass
src/praisonai-agents/praisonaiagents/llm/error_classifier.py
+8 -1
View File
@@ -9,7 +9,14 @@ import re
from enum import Enum
from typing import Dict, Tuple, List, Optional
__all__ = ["ErrorCategory", "classify_error", "should_retry", "get_retry_delay"]
__all__ = [
"ErrorCategory",
"classify_error",
"should_retry",
"get_retry_delay",
"extract_retry_after",
"get_error_context",
]
class ErrorCategory(str, Enum):
src/praisonai-agents/praisonaiagents/llm/llm.py
+10 -11
View File
@@ -650,11 +650,12 @@ Respond with ONLY a valid JSON tool call in this format:
return any(indicator in error_str or indicator in error_type for indicator in indicators)
def _classify_error_and_should_retry(self, error: Exception) -> tuple[str, bool, float]:
def _classify_error_and_should_retry(self, error: Exception, attempt: int = 1) -> tuple[str, bool, float]:
"""Classify error and determine retry strategy using G5 error classifier.
Args:
error: Exception to classify
attempt: Current attempt number (1-based)
Returns:
Tuple of (category, should_retry, retry_delay)
@@ -674,8 +675,8 @@ Respond with ONLY a valid JSON tool call in this format:
if retry_after:
return category.value, True, retry_after
# Use category-specific delay calculation
delay = get_retry_delay(category, attempt=1, base_delay=self._retry_delay)
# Use category-specific delay calculation with proper attempt
delay = get_retry_delay(category, attempt=attempt, base_delay=self._retry_delay)
return category.value, True, delay
except ImportError:
@@ -709,17 +710,16 @@ Respond with ONLY a valid JSON tool call in this format:
return func(*args, **kwargs)
except Exception as e:
if not self._is_rate_limit_error(e):
category, can_retry, retry_delay = self._classify_error_and_should_retry(e, attempt + 1)
if not can_retry:
raise
last_error = e
error_str = str(e)
if attempt < self._max_retries:
retry_delay = self._parse_retry_delay(error_str)
logging.warning(
f"Rate limit hit (attempt {attempt + 1}/{self._max_retries + 1}), "
f"{category} error hit (attempt {attempt + 1}/{self._max_retries + 1}), "
f"waiting {retry_delay:.1f}s before retry..."
)
@@ -770,17 +770,16 @@ Respond with ONLY a valid JSON tool call in this format:
return await func(*args, **kwargs)
except Exception as e:
if not self._is_rate_limit_error(e):
category, can_retry, retry_delay = self._classify_error_and_should_retry(e, attempt + 1)
if not can_retry:
raise
last_error = e
error_str = str(e)
if attempt < self._max_retries:
retry_delay = self._parse_retry_delay(error_str)
logging.warning(
f"Rate limit hit (attempt {attempt + 1}/{self._max_retries + 1}), "
f"{category} error hit (attempt {attempt + 1}/{self._max_retries + 1}), "
f"waiting {retry_delay:.1f}s before retry..."
)
src/praisonai-agents/praisonaiagents/llm/sanitize.py
+51 -34
View File
@@ -44,13 +44,57 @@ def strip_surrogates(text: str) -> str:
return re.sub(r'[\uD800-\uDFFF]', '', text)
def _sanitize_value_recursive(value: Any) -> tuple[Any, bool]:
"""Recursively sanitize any nested string values in a data structure.
Args:
value: Value to sanitize (can be dict, list, str, or other)
Returns:
Tuple of (sanitized_value, changed_flag)
"""
if isinstance(value, str):
sanitized = strip_surrogates(value)
return sanitized, sanitized != value
elif isinstance(value, list):
changed = False
sanitized_items = []
for item in value:
sanitized_item, item_changed = _sanitize_value_recursive(item)
sanitized_items.append(sanitized_item)
changed = changed or item_changed
return sanitized_items, changed
elif isinstance(value, dict):
changed = False
sanitized_dict = {}
for key, nested_value in value.items():
# Sanitize the key if it's a string
if isinstance(key, str):
sanitized_key = strip_surrogates(key)
if sanitized_key != key:
changed = True
key = sanitized_key
# Recursively sanitize the value
sanitized_value, value_changed = _sanitize_value_recursive(nested_value)
sanitized_dict[key] = sanitized_value
changed = changed or value_changed
return sanitized_dict, changed
else:
# Return other types unchanged
return value, False
def sanitize_messages(messages: List[Dict[str, Any]]) -> bool:
"""Sanitize message content in-place, removing problematic Unicode.
Processes all string content in message dictionaries, including:
- message.content (string or list)
- message.name
- Any nested string values
- Any nested string values (including tool_calls[].function.arguments)
Args:
messages: List of message dicts to sanitize in-place
@@ -73,39 +117,12 @@ def sanitize_messages(messages: List[Dict[str, Any]]) -> bool:
if not isinstance(message, dict):
continue
# Sanitize content field (most common)
if "content" in message:
content = message["content"]
if isinstance(content, str):
sanitized = strip_surrogates(content)
if sanitized != content:
message["content"] = sanitized
changed = True
elif isinstance(content, list):
# Handle list content (e.g., multimodal messages)
for i, item in enumerate(content):
if isinstance(item, dict) and "text" in item:
text = item["text"]
if isinstance(text, str):
sanitized = strip_surrogates(text)
if sanitized != text:
content[i]["text"] = sanitized
changed = True
elif isinstance(item, str):
sanitized = strip_surrogates(item)
if sanitized != item:
content[i] = sanitized
changed = True
# Sanitize other string fields
for key, value in message.items():
if isinstance(value, str) and key != "content": # Already handled above
sanitized = strip_surrogates(value)
if sanitized != value:
message[key] = sanitized
changed = True
# Recursively sanitize the entire message structure
sanitized_message, message_changed = _sanitize_value_recursive(message)
if message_changed:
message.clear()
message.update(sanitized_message)
changed = True
return changed
src/praisonai-agents/praisonaiagents/session/hierarchy.py
+24 -9
View File
@@ -490,7 +490,10 @@ class HierarchicalSessionStore(DefaultSessionStore):
Returns:
True if title was generated and set, False otherwise
"""
session = self._load_extended_session(session_id)
import asyncio
# Load session in thread to avoid blocking event loop
session = await asyncio.to_thread(self._load_extended_session, session_id)
# Skip if already has a title
if session.title and session.title.strip():
@@ -506,12 +509,18 @@ class HierarchicalSessionStore(DefaultSessionStore):
assistant_msg = None
for msg in messages:
if msg.get("role") == "user" and not user_msg:
# Handle both SessionMessage dataclass and dict formats
if hasattr(msg, 'role'):
role = msg.role
content = msg.content
else:
role = msg.get("role")
content = msg.get("content", "")
if role == "user" and not user_msg:
if isinstance(content, str) and content.strip():
user_msg = content
elif msg.get("role") == "assistant" and not assistant_msg and user_msg:
content = msg.get("content", "")
elif role == "assistant" and not assistant_msg and user_msg:
if isinstance(content, str) and content.strip():
assistant_msg = content
break
@@ -525,12 +534,18 @@ class HierarchicalSessionStore(DefaultSessionStore):
title = await generate_title_async(user_msg, assistant_msg)
if title and title.strip():
session.title = title.strip()
return self._save_extended_session(session)
# Reload session to avoid overwriting concurrent updates
fresh_session = await asyncio.to_thread(self._load_extended_session, session_id)
# Only set title if it's still empty
if not fresh_session.title or not fresh_session.title.strip():
fresh_session.title = title.strip()
return await asyncio.to_thread(self._save_extended_session, fresh_session)
except Exception:
# Title generation failed - not critical
pass
except Exception as e:
# Title generation failed - log with context instead of silent failure
import logging
logger = logging.getLogger(__name__)
logger.debug("Auto title generation failed for session %s: %s", session_id, str(e))
return False
src/praisonai-agents/praisonaiagents/session/title.py
+21 -5
View File
@@ -41,11 +41,27 @@ def generate_title(
fallback_title = _create_fallback_title(user_msg, max_length)
try:
# Run async version in sync context
return asyncio.run(generate_title_async(
user_msg, assistant_msg, llm_model, timeout, max_length
))
except Exception:
# Check if we're already in an event loop
asyncio.get_running_loop()
except RuntimeError:
# No running loop - safe to use asyncio.run
try:
return asyncio.run(generate_title_async(
user_msg, assistant_msg, llm_model, timeout, max_length
))
except Exception:
import logging
logger = logging.getLogger(__name__)
logger.debug("Title generation failed in sync context", exc_info=True)
return fallback_title
else:
# Already in event loop - cannot use asyncio.run
import logging
logger = logging.getLogger(__name__)
logger.warning(
"generate_title() called from running event loop; "
"use generate_title_async() instead. Returning fallback title."
)
return fallback_title
src/praisonai-agents/praisonaiagents/tools/call_executor.py
+10 -6
View File
@@ -141,13 +141,17 @@ class ParallelToolCallExecutor:
# G4: Check for path conflicts - fallback to sequential if conflicts detected
try:
from .path_overlap import has_write_conflicts
if has_write_conflicts(tool_calls):
logger.info(f"Path conflicts detected in {len(tool_calls)} tool calls, using sequential execution")
sequential_executor = SequentialToolCallExecutor()
return sequential_executor.execute_batch(tool_calls, execute_tool_fn)
except ImportError:
# path_overlap module not available, continue with parallel execution
pass
logger.warning(
"Path conflict detection unavailable; using sequential execution for safety"
)
sequential_executor = SequentialToolCallExecutor()
return sequential_executor.execute_batch(tool_calls, execute_tool_fn)
if has_write_conflicts(tool_calls):
logger.info(f"Path conflicts detected in {len(tool_calls)} tool calls, using sequential execution")
sequential_executor = SequentialToolCallExecutor()
return sequential_executor.execute_batch(tool_calls, execute_tool_fn)
def _execute_single_tool(tool_call: ToolCall) -> ToolResult:
"""Execute a single tool call with error handling."""
src/praisonai-agents/praisonaiagents/tools/path_overlap.py
+35 -2
View File
@@ -20,6 +20,13 @@ _WRITE_TOOLS = frozenset({
"mkdir", "rmdir", "move_file", "copy_file"
})
# Write operation hints in tool names for detecting custom write tools
_WRITE_NAME_HINTS = frozenset({
"write", "edit", "patch", "create", "save", "update",
"delete", "remove", "mkdir", "rmdir", "move", "copy", "persist",
"modify", "append"
})
# Argument names that typically contain file paths
_PATH_ARG_NAMES = frozenset({
"path", "file_path", "filepath", "dest", "destination", "target",
@@ -28,6 +35,32 @@ _PATH_ARG_NAMES = frozenset({
})
def _is_potential_write_tool(function_name: str, arguments: dict) -> bool:
"""Check if a tool call is potentially a write operation.
Args:
function_name: Name of the tool function
arguments: Tool call arguments
Returns:
True if the tool might perform write operations
"""
# Check explicit write tools first
if function_name in _WRITE_TOOLS:
return True
# Check for write hints in function name
normalized_name = function_name.lower()
if any(hint in normalized_name for hint in _WRITE_NAME_HINTS):
return True
# Conservative fallback: if tool has path-like arguments, treat as potential writer
if any(arg_name in _PATH_ARG_NAMES for arg_name in arguments.keys()):
return True
return False
def extract_paths(tool_call: ToolCall) -> List[pathlib.Path]:
"""Extract file paths from a tool call.
@@ -39,8 +72,8 @@ def extract_paths(tool_call: ToolCall) -> List[pathlib.Path]:
"""
paths = []
# Only check write tools
if tool_call.function_name not in _WRITE_TOOLS:
# Check if this tool might perform write operations
if not _is_potential_write_tool(tool_call.function_name, tool_call.arguments):
return paths
args = tool_call.arguments or {}
src/praisonai-agents/tests/unit/agent/test_interrupt.py
+15 -23
View File
@@ -52,17 +52,21 @@ class TestInterruptController:
"""Test thread-safe operations."""
controller = InterruptController()
results = []
ready_event = threading.Event()
done_event = threading.Event()
def worker():
# Wait a bit then request interrupt
time.sleep(0.1)
# Wait for checker to be ready
ready_event.wait(timeout=1)
controller.request("thread_cancel")
results.append("requested")
done_event.set()
def checker():
# Keep checking until interrupted
while not controller.is_set():
time.sleep(0.05)
# Signal ready and wait for interrupt
ready_event.set()
done_event.wait(timeout=1)
assert controller.is_set()
results.append(f"cancelled: {controller.reason}")
# Start threads
@@ -105,23 +109,11 @@ class TestInterruptController:
controller.request("new_reason")
assert controller.reason == "new_reason"
def test_zero_overhead_when_not_used(self):
"""Test that creation and is_set() have minimal overhead."""
import time
# Test creation overhead
start = time.perf_counter()
for _ in range(1000):
controller = InterruptController()
creation_time = time.perf_counter() - start
# Test check overhead
def test_is_set_is_stable_when_not_used(self):
"""Test that repeated checks do not mutate state."""
controller = InterruptController()
start = time.perf_counter()
for _ in range(10000):
controller.is_set()
check_time = time.perf_counter() - start
# Should be very fast (< 1ms each)
assert creation_time < 0.001
assert check_time < 0.001
# Should consistently return False and None
for _ in range(1000):
assert not controller.is_set()
assert controller.reason is None