fix: resolve 3 runtime showstoppers in motion graphics module

🔴 Fix AgentTeam instantiation error
- Change from leader= to process='hierarchical' with manager_llm=
- Remove unused protocols import in team.py

🔴 Implement proper workspace safety validation
- Add base_dir parameter to HtmlRenderBackend constructor
- Implement real containment check using Path.relative_to()
- Validate against base directory and temp directories
- Reject path traversal attempts

🔴 Wire render_iterate into bounded retry system
- Add max_retries parameter to RenderTools
- Implement retry logic in render_composition()
- Add render_with_bounded_retries() method exposing render_iterate
- Fix workspace path mismatch (coordinator expects /renders)

🔧 Minor cleanups
- Remove dead code in git_tools.py _validate_file_path
- Align render workspace paths between coordinator and animator

🧪 Add comprehensive unit tests
- test_backend_safety.py for workspace containment validation
- test_render_retries.py for bounded retry functionality

Fixes #27

Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>

Author: github-actions[bot]

praisonai_tools/tools/git_tools.py

@@ -447,17 +447,7 @@ def _validate_file_path(self, file_path: str) -> str:
         if ".." in safe_path:
             raise ValueError("Invalid file path")
 
-        # Additional safety: resolve path and check it doesn't escape
-        try:
-            from pathlib import Path
-            resolved = Path(safe_path).resolve()
-            # If it tries to go outside current dir, reject
-            if str(resolved).startswith("/") and not str(resolved).startswith(str(Path.cwd())):
-                # This is an absolute path that escapes - but we allow relative paths
-                pass
-        except Exception:
-            # If path resolution fails, reject
-            raise ValueError("Invalid file path")
+        # Basic path validation - above checks already prevent most traversal attacks
 
         return safe_path
 

praisonai_tools/video/motion_graphics/agent.py

@@ -20,9 +20,10 @@
 class RenderTools:
     """Tools for motion graphics rendering."""
 
-    def __init__(self, backend: RenderBackendProtocol, workspace: Path):
+    def __init__(self, backend: RenderBackendProtocol, workspace: Path, max_retries: int = 3):
         self.backend = backend
         self.workspace = workspace
+        self.max_retries = max_retries
 
     async def lint_composition(self, strict: bool = False) -> dict:
         """Lint the motion graphics composition.
@@ -46,7 +47,7 @@ async def render_composition(
         fps: int = 30,
         quality: str = "standard"
     ) -> dict:
-        """Render the motion graphics composition to MP4.
+        """Render the motion graphics composition to MP4 with bounded retries.
         
         Args:
             output_name: Output filename
@@ -64,7 +65,67 @@ async def render_composition(
             quality=quality
         )
 
+        # Direct render with bounded retries for transient failures
+        # Note: For composition authoring retries, the agent should use render_iterate 
+        # with appropriate write/lint/patch functions
         result = await self.backend.render(self.workspace, opts)
+        attempts = 1
+        
+        while not result.ok and attempts < self.max_retries:
+            # Retry for transient failures (e.g., ffmpeg timeouts)
+            result = await self.backend.render(self.workspace, opts)
+            attempts += 1
+        
+        return {
+            "ok": result.ok,
+            "output_path": str(result.output_path) if result.output_path else None,
+            "size_kb": result.size_kb,
+            "stderr": result.stderr,
+            "attempts": attempts,
+            "bytes": result.bytes_
+        }
+    
+    async def render_with_bounded_retries(
+        self,
+        write_fn,
+        patch_fn,
+        output_name: str = "video.mp4", 
+        fps: int = 30,
+        quality: str = "standard",
+        **kwargs
+    ) -> dict:
+        """Use render_iterate for full composition authoring with retries.
+        
+        Args:
+            write_fn: Function to write initial composition
+            patch_fn: Function to patch composition based on error
+            output_name: Output filename
+            fps: Frames per second  
+            quality: Quality setting
+            **kwargs: Arguments passed to write_fn
+            
+        Returns:
+            Dict with render results
+        """
+        from ._render_loop import render_iterate
+        from .protocols import RenderOpts
+        
+        opts = RenderOpts(output_name=output_name, fps=fps, quality=quality)
+        
+        async def lint_fn():
+            return await self.backend.lint(self.workspace, strict=False)
+            
+        async def render_fn():
+            return await self.backend.render(self.workspace, opts)
+        
+        result = await render_iterate(
+            write_fn=write_fn,
+            lint_fn=lint_fn,
+            render_fn=render_fn,
+            patch_fn=patch_fn,
+            max_retries=self.max_retries,
+            **kwargs
+        )
 
         return {
             "ok": result.ok,
@@ -134,7 +195,7 @@ def create_motion_graphics_agent(
 
     # Create tools
     file_tools = FileTools(base_dir=str(workspace))
-    render_tools = RenderTools(render_backend, workspace)
+    render_tools = RenderTools(render_backend, workspace, max_retries)
 
     # Create agent
     agent = Agent(

praisonai_tools/video/motion_graphics/backend_html.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import os
 import shutil
 import subprocess
 import tempfile
@@ -36,7 +37,7 @@ class HtmlRenderBackend:
     - Subprocess timeout limits
     """
 
-    def __init__(self):
+    def __init__(self, base_dir: Optional[Path] = None):
         if async_playwright is None:
             raise ImportError(
                 "Playwright not installed. Install with: pip install playwright"
@@ -45,6 +46,13 @@ def __init__(self):
             raise ImportError(
                 "imageio-ffmpeg not installed. Install with: pip install imageio-ffmpeg"
             )
+        
+        # Set allowed base directory for workspace safety
+        if base_dir is None:
+            # Default to current working directory and temp directory as allowed roots
+            self._allowed_base = Path.cwd()
+        else:
+            self._allowed_base = Path(base_dir).resolve()
 
     async def lint(self, workspace: Path, strict: bool = False) -> LintResult:
         """Lint HTML composition for common issues."""
@@ -128,10 +136,29 @@ async def render(self, workspace: Path, opts: RenderOpts) -> RenderResult:
     def _is_safe_workspace(self, workspace: Path) -> bool:
         """Check if workspace path is safe (prevents path traversal)."""
         try:
-            workspace_abs = workspace.resolve()
-            # Basic check - workspace should be under a temp directory or user's project
-            return True  # Add more sophisticated checks as needed
-        except Exception:
+            workspace_abs = workspace.resolve(strict=True)
+            allowed_base_abs = self._allowed_base.resolve()
+            
+            # Also allow temp directories
+            temp_base = Path(tempfile.gettempdir()).resolve()
+            
+            # Check if workspace is under allowed base or temp directory
+            try:
+                # Check if workspace is relative to allowed base
+                workspace_abs.relative_to(allowed_base_abs)
+                return True
+            except ValueError:
+                pass
+                
+            try:
+                # Check if workspace is relative to temp directory
+                workspace_abs.relative_to(temp_base)
+                return True
+            except ValueError:
+                pass
+                
+            return False
+        except (OSError, ValueError):
             return False
 
     async def _render_with_playwright(self, workspace: Path, opts: RenderOpts) -> RenderResult:

praisonai_tools/video/motion_graphics/team.py

@@ -15,7 +15,6 @@
     search_web = None
 
 from .agent import create_motion_graphics_agent
-from ..motion_graphics import protocols
 
 
 def motion_graphics_team(
@@ -58,6 +57,8 @@ def motion_graphics_team(
         workspace = Path(workspace)
 
     workspace.mkdir(parents=True, exist_ok=True)
+    renders_dir = workspace / "renders"
+    renders_dir.mkdir(parents=True, exist_ok=True)
 
     # Create agents
     agents = []
@@ -74,7 +75,7 @@ def motion_graphics_team(
 4. Return final results to users
 
 CRITICAL OUTPUT VALIDATION RULES:
-- A render succeeded ONLY IF the reply contains a concrete file path (e.g., '/renders/video_123.mp4') AND no error indicators
+- A render succeeded ONLY IF the reply contains a concrete file path under '{renders_dir}' AND no error indicators
 - Never fabricate file paths or claim success without concrete evidence
 - Surface all errors from the Animator
 - Stop work after maximum retry budget is exceeded
@@ -137,16 +138,17 @@ def motion_graphics_team(
     # Animator - the core specialist
     animator = create_motion_graphics_agent(
         backend=backend,
-        workspace=workspace / "animations",
+        workspace=renders_dir,
         llm=llm
     )
     animator.name = "animator"
     agents.append(animator)
 
-    # Create team with coordinator as leader
+    # Create team with coordinator as manager via hierarchical process
     team = AgentTeam(
         agents=agents,
-        leader=coordinator,
+        process="hierarchical",
+        manager_llm=llm,
         **team_kwargs
     )
 

tests/unit/video/test_backend_safety.py

@@ -0,0 +1,66 @@
+"""Tests for backend workspace safety."""
+
+import pytest
+import tempfile
+from pathlib import Path
+from unittest.mock import Mock, AsyncMock
+
+try:
+    from praisonai_tools.video.motion_graphics.backend_html import HtmlRenderBackend
+    from praisonai_tools.video.motion_graphics.protocols import RenderOpts
+    BACKEND_AVAILABLE = True
+except ImportError:
+    BACKEND_AVAILABLE = False
+
+
+@pytest.mark.skipif(not BACKEND_AVAILABLE, reason="Motion graphics backend not available")
+class TestBackendSafety:
+    """Test workspace safety features."""
+    
+    def setup_method(self):
+        """Set up test fixtures."""
+        self.temp_dir = Path(tempfile.mkdtemp())
+        self.backend = HtmlRenderBackend(base_dir=self.temp_dir)
+    
+    def teardown_method(self):
+        """Clean up test fixtures."""
+        import shutil
+        shutil.rmtree(self.temp_dir, ignore_errors=True)
+    
+    def test_safe_workspace_under_base_dir(self):
+        """Test workspace under base directory is considered safe."""
+        safe_workspace = self.temp_dir / "workspace"
+        safe_workspace.mkdir()
+        
+        assert self.backend._is_safe_workspace(safe_workspace)
+    
+    def test_unsafe_workspace_outside_base_dir(self):
+        """Test workspace outside base directory is rejected."""
+        # Try to escape to parent directory
+        unsafe_workspace = self.temp_dir.parent / "escaped"
+        
+        assert not self.backend._is_safe_workspace(unsafe_workspace)
+    
+    def test_workspace_in_temp_dir_allowed(self):
+        """Test workspace in temp directory is allowed."""
+        import tempfile
+        temp_workspace = Path(tempfile.gettempdir()) / "test_workspace"
+        
+        assert self.backend._is_safe_workspace(temp_workspace)
+    
+    def test_nonexistent_workspace_rejected(self):
+        """Test non-existent workspace is rejected."""
+        nonexistent = Path("/does/not/exist/workspace")
+        
+        assert not self.backend._is_safe_workspace(nonexistent)
+    
+    @pytest.mark.asyncio
+    async def test_unsafe_workspace_blocks_render(self):
+        """Test unsafe workspace blocks render operation."""
+        unsafe_workspace = Path("/etc")  # System directory
+        opts = RenderOpts(output_name="test.mp4", fps=30, quality="standard")
+        
+        result = await self.backend.render(unsafe_workspace, opts)
+        
+        assert not result.ok
+        assert "Unsafe workspace path" in result.stderr