fix: implement ArtifactStore for retrievable tool output overflow

src/praisonai-agents/praisonaiagents/agent/agent.py

@@ -592,6 +592,7 @@ def __init__(
         runtime: Optional[Union[str, Dict[str, Any], 'AgentRuntimeConfig']] = None,  # Model-scoped runtime configuration
         interrupt_controller: Optional['InterruptController'] = None,  # G2: Cooperative cancellation
         tool_search: Optional[Union[bool, str, Dict[str, Any], 'ToolSearchConfig']] = False,  # Progressive tool disclosure
+        tool_output: Optional[Union[bool, Dict[str, Any], 'ToolOutputConfig']] = None,  # Tool output handling and artifact storage
         message_steering: Optional[Union[bool, 'MessageSteeringProtocol']] = False,  # Real-time message steering during execution
         sandbox: Optional[Union[bool, 'SandboxConfig']] = None,  # Sandbox for safe code execution
     ):
@@ -1506,6 +1507,53 @@ def __init__(
                     "a dict of ToolSearchConfig fields, or ToolSearchConfig"
                 )
 
+        # Process tool_output config (artifact storage for large outputs)
+        self._artifact_store = None
+        self._tool_output_config = tool_output  # Store the original config for cloning
+        # Track if tool_output_limit was customized via tool_output config
+        _custom_tool_output_limit = None
+
+        if tool_output is None or tool_output is False:
+            # Disabled - use default truncation only
+            pass
+        elif tool_output is True:
+            # Enabled with defaults
+            from ..config.feature_configs import ToolOutputConfig
+            from ..context.artifact_store import FileSystemArtifactStore
+            config = ToolOutputConfig()
+            _custom_tool_output_limit = config.max_bytes
+            if config.enable_artifacts:
+                self._artifact_store = config.artifact_store or FileSystemArtifactStore(
+                    retention_days=config.retention_days,
+                    redact_secrets=config.redact_secrets
+                )
+        elif isinstance(tool_output, dict):
+            # Dict -> config overrides
+            from ..config.feature_configs import ToolOutputConfig
+            from ..context.artifact_store import FileSystemArtifactStore
+            config = ToolOutputConfig(**tool_output)
+            _custom_tool_output_limit = config.max_bytes
+            if config.enable_artifacts:
+                self._artifact_store = config.artifact_store or FileSystemArtifactStore(
+                    retention_days=config.retention_days,
+                    redact_secrets=config.redact_secrets
+                )
+        else:
+            from ..config.feature_configs import ToolOutputConfig
+            if isinstance(tool_output, ToolOutputConfig):
+                config = tool_output
+                _custom_tool_output_limit = config.max_bytes
+                if config.enable_artifacts:
+                    from ..context.artifact_store import FileSystemArtifactStore
+                    self._artifact_store = config.artifact_store or FileSystemArtifactStore(
+                        retention_days=config.retention_days,
+                        redact_secrets=config.redact_secrets
+                    )
+            else:
+                raise TypeError(
+                    "tool_output must be False/None, True, a dict of ToolOutputConfig fields, or ToolOutputConfig"
+                )
+
         # ============================================================
         # END CONSOLIDATED PARAMS EXTRACTION
         # ============================================================
@@ -1744,7 +1792,8 @@ def __init__(
         self._init_message_steering(message_steering)
         self.verbose = verbose
         self._has_explicit_output_config = _has_explicit_output  # Track if user set output mode
-        self.tool_output_limit = tool_output_limit  # Configurable tool output limit
+        # Use custom tool_output_limit if set via tool_output config, otherwise use parameter value
+        self.tool_output_limit = _custom_tool_output_limit if _custom_tool_output_limit is not None else tool_output_limit
         self.allow_delegation = allow_delegation
         self.step_callback = step_callback
         # Token budget guard (zero overhead when _max_budget is None)
@@ -2151,6 +2200,7 @@ def clone_for_channel(self) -> "Agent":
             'approval': getattr(self, '_approval_config', None),
             'learn': getattr(self, '_learn_config', None),
             'tool_search': getattr(self, '_tool_search_config', None),
+            'tool_output': getattr(self, '_tool_output_config', None),
 
             # Tool configuration - use consolidated config when available  
             'tool_config': getattr(self, '_tool_config', None),
@@ -5515,6 +5565,18 @@ def __del__(self):
                 memory = getattr(self, "_memory_instance", None)
                 if memory and hasattr(memory, 'close_connections'):
                     memory.close_connections()
+
+                # Clean up old artifacts if artifact store is configured
+                artifact_store = getattr(self, "_artifact_store", None)
+                if artifact_store and hasattr(artifact_store, 'cleanup_old_artifacts'):
+                    try:
+                        artifact_store.cleanup_old_artifacts()
+                    except Exception as e:
+                        # Log the error for debugging but don't fail cleanup
+                        import logging
+                        logging.debug(
+                            f"Failed to cleanup artifacts for agent {self.name}: {e}"
+                        )
             except Exception as exc:  # noqa: BLE001 - finalizers must not raise
                 import contextlib
                 with contextlib.suppress(Exception):

src/praisonai-agents/praisonaiagents/agent/tool_execution.py

@@ -26,6 +26,35 @@
 
 class ToolExecutionMixin:
     """Mixin providing toolexecution methods for the Agent class."""
+
+    def _register_artifact_tools(self):
+        """Register artifact retrieval tools when artifacts are first created."""
+        try:
+            from ..tools import artifact_tools
+
+            # Set the store reference for the tools
+            artifact_tools.set_artifact_store(self._artifact_store)
+
+            # Add the retrieval tools
+            tools_to_add = [
+                artifact_tools.artifact_head,
+                artifact_tools.artifact_tail,
+                artifact_tools.artifact_grep,
+                artifact_tools.artifact_chunk,
+                artifact_tools.artifact_load,
+                artifact_tools.artifact_list,
+            ]
+
+            # Only add if not already present
+            existing_tool_names = {getattr(t, '__name__', str(t)) for t in self.tools}
+            for tool in tools_to_add:
+                tool_name = getattr(tool, '__name__', str(tool))
+                if tool_name not in existing_tool_names:
+                    self.tools.append(tool)
+
+            logging.debug("Registered artifact retrieval tools")
+        except Exception as e:
+            logging.warning(f"Failed to register artifact tools: {e}")
 
     def _get_existing_stream_emitter(self):
         """Return an already-initialized stream emitter without creating one."""
@@ -310,28 +339,78 @@ def execute_with_context():
                 try:
                     result_str = str(result)
 
-                    if self.context_manager:
-                        # Use context-aware truncation with configured budget
-                        truncated = self._truncate_tool_output(function_name, result_str)
-                    else:
-                        # Apply default limit even without context management
-                        # This prevents runaway tool outputs from causing overflow
-                        limit = getattr(self, 'tool_output_limit', 16000)
-                        if len(result_str) > limit:
-                            # Use smart truncation format that judge recognizes as OK
-                            tail_size = min(limit // 5, 2000)
-                            head = result_str[:limit - tail_size]
-                            tail = result_str[-tail_size:] if tail_size > 0 else ""
+                    # Get configured limit
+                    limit = getattr(self, 'tool_output_limit', 16000)
+
+                    # Check if we need to spill to artifact store
+                    if len(result_str) > limit:
+                        # Try to use artifact store if available
+                        artifact_ref = None
+                        if hasattr(self, '_artifact_store') and self._artifact_store is not None:
+                            try:
+                                from ..context.artifacts import ArtifactMetadata
+
+                                # Create metadata for this artifact
+                                metadata = ArtifactMetadata(
+                                    agent_id=self.name,
+                                    run_id=getattr(self, '_current_run_id', 'unknown'),
+                                    tool_name=function_name,
+                                    turn_id=getattr(self, '_turn_counter', 0),
+                                )
+
+                                # Store the full output
+                                artifact_ref = self._artifact_store.store(result_str, metadata)
+                                logging.debug(f"Stored {function_name} output ({len(result_str)} bytes) as artifact {artifact_ref.artifact_id}")
+
+                                # Register artifact retrieval tools if not already registered
+                                if not hasattr(self, '_artifact_tools_registered'):
+                                    self._register_artifact_tools()
+                                    self._artifact_tools_registered = True
+                            except Exception as e:
+                                logging.debug(f"Failed to store artifact: {e}")
+
+                        # Generate truncated preview
+                        tail_size = min(limit // 5, 2000)
+                        head = result_str[:limit - tail_size]
+                        tail = result_str[-tail_size:] if tail_size > 0 else ""
+
+                        # If we stored an artifact, include reference in the output
+                        if artifact_ref:
+                            truncated = (
+                                f"{head}\n"
+                                f"...[{len(result_str):,} chars total, showing first/last portions]...\n"
+                                f"{tail}\n\n"
+                                f"{artifact_ref.to_inline()}"
+                            )
+                        else:
+                            # Fallback to simple truncation
                             truncated = f"{head}\n...[{len(result_str):,} chars, showing first/last portions]...\n{tail}"
+                    else:
+                        truncated = result_str
+
+                    if self.context_manager and hasattr(self, '_truncate_tool_output'):
+                        # Use context-aware truncation if available, but preserve artifact reference
+                        if artifact_ref:
+                            # Extract the artifact reference from the truncated string
+                            artifact_inline = artifact_ref.to_inline()
+                            # Remove the artifact reference before context truncation
+                            truncated_without_ref = truncated.replace(artifact_inline, "").rstrip()
+                            # Apply context truncation
+                            truncated_without_ref = self._truncate_tool_output(function_name, truncated_without_ref)
+                            # Re-append the artifact reference
+                            truncated = f"{truncated_without_ref}\n\n{artifact_inline}"
                         else:
-                            truncated = result_str
+                            truncated = self._truncate_tool_output(function_name, truncated)
 
                     if len(truncated) < len(result_str):
                         logging.debug(f"Truncated {function_name} output from {len(result_str)} to {len(truncated)} chars")
                         # For dicts, truncate large string fields (e.g., raw_content from search)
                         if isinstance(result, dict):
                             max_field_chars = getattr(self, 'tool_output_limit', 16000) if not self.context_manager else None
                             result = self._truncate_dict_fields(result, function_name, max_field_chars)
+                            # Add artifact reference to dict result if available
+                            if artifact_ref:
+                                result["_artifact_ref"] = artifact_ref.to_dict()
                         else:
                             result = truncated
                 except Exception as e:

src/praisonai-agents/praisonaiagents/config/feature_configs.py

@@ -42,6 +42,7 @@
 from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from ..compaction.strategy import CompactionStrategy
+    from ..context.artifact_store import FileSystemArtifactStore
 
 
 class MemoryBackend(str, Enum):
@@ -1506,6 +1507,79 @@ def resolve_tools(value: ToolParam) -> Optional[ToolConfig]:
     return value
 
 
+@dataclass
+class ToolOutputConfig:
+    """
+    Configuration for tool output handling and artifact storage.
+
+    Controls when and how large tool outputs are stored as artifacts
+    instead of being truncated and lost.
+
+    Args:
+        max_bytes: Maximum bytes before spilling to artifact store (default: 16000)
+        max_lines: Maximum lines before spilling (default: None, bytes-only)
+        direction: Truncation direction - "head", "tail", or "both" (default: "both")
+        retention_days: Days to retain artifacts before garbage collection (default: 7)
+        enable_artifacts: Whether to enable artifact storage (default: True)
+        artifact_store: Custom artifact store instance (default: FileSystemArtifactStore)
+        redact_secrets: Whether to redact secrets from artifacts (default: True)
+
+    Example:
+        agent = Agent(
+            instructions="...",
+            tool_output=ToolOutputConfig(
+                max_bytes=32000,
+                direction="tail",
+                retention_days=14,
+            )
+        )
+    """
+    max_bytes: int = 16000
+    max_lines: Optional[int] = None
+    direction: str = "both"  # "head", "tail", or "both"
+    retention_days: int = 7
+    enable_artifacts: bool = True
+    artifact_store: Optional[Any] = None  # FileSystemArtifactStore instance
+    redact_secrets: bool = True
+
+    def __post_init__(self) -> None:
+        """Validate configuration values."""
+        if self.max_bytes <= 0:
+            raise ValueError("max_bytes must be > 0. Use False/None to disable artifact spilling.")
+        if self.max_lines is not None and self.max_lines <= 0:
+            raise ValueError("max_lines must be > 0 when provided.")
+        if self.direction not in {"head", "tail", "both"}:
+            raise ValueError("direction must be one of: 'head', 'tail', 'both'.")
+        if self.retention_days < 0:
+            raise ValueError("retention_days must be >= 0.")
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary."""
+        return {
+            "max_bytes": self.max_bytes,
+            "max_lines": self.max_lines,
+            "direction": self.direction,
+            "retention_days": self.retention_days,
+            "enable_artifacts": self.enable_artifacts,
+            "redact_secrets": self.redact_secrets,
+        }
+
+
+# Type alias for tool output parameter
+ToolOutputParam = Union[bool, ToolOutputConfig, Any]
+
+
+def resolve_tool_output(value: Optional[ToolOutputParam]) -> Optional[ToolOutputParam]:
+    """Resolve tool output configuration with precedence ladder."""
+    if value is None or value is False:
+        return None
+    if value is True:
+        return ToolOutputConfig()
+    if isinstance(value, ToolOutputConfig):
+        return value
+    return value
+
+
 __all__ = [
     # Enums
     "MemoryBackend",
@@ -1533,6 +1607,7 @@ def resolve_tools(value: ToolParam) -> Optional[ToolConfig]:
     "SkillsConfig",
     "AutonomyConfig",
     "ToolSearchConfig",
+    "ToolOutputConfig",
     # Config classes (Multi-Agent)
     "MultiAgentHooksConfig",
     "MultiAgentOutputConfig",
@@ -1555,6 +1630,7 @@ def resolve_tools(value: ToolParam) -> Optional[ToolConfig]:
     "AutonomyParam",
     "ToolSearchParam", 
     "ToolParam",
+    "ToolOutputParam",
     # Precedence ladder resolvers
     "resolve_memory",
     "resolve_knowledge",
@@ -1566,4 +1642,5 @@ def resolve_tools(value: ToolParam) -> Optional[ToolConfig]:
     "resolve_autonomy",
     "resolve_tool_search",
     "resolve_tools",
+    "resolve_tool_output",
 ]

src/praisonai-agents/praisonaiagents/context/__init__.py

@@ -180,6 +180,7 @@ def format_percent(value: float) -> str:
     "TerminalLoggerProtocol",
     "compute_checksum",
     "generate_summary",
+    "FileSystemArtifactStore",
     # Context Compaction Policy
     "ContextCompactionPolicy",
     "ContextCompactionPolicyProtocol",
@@ -295,6 +296,11 @@ def __getattr__(name: str):
         from . import artifacts
         return getattr(artifacts, name)
 
+    # FileSystemArtifactStore
+    if name == "FileSystemArtifactStore":
+        from .artifact_store import FileSystemArtifactStore
+        return FileSystemArtifactStore
+
     # Session Context Tracking (Agno pattern)
     if name in ("SessionContextTracker", "SessionState"):
         from . import session_tracker