test(#755): Add tests for resource limits

Tests for execution, timeout, violation detection.
Refs #755

tests/test_resource_limits.py

@@ -0,0 +1,44 @@
+"""
+Tests for resource limits (#755).
+"""
+
+import pytest
+from tools.resource_limits import ResourceLimiter, ResourceLimits, ResourceResult, ResourceViolation
+
+
+class TestResourceLimiter:
+    def test_successful_execution(self):
+        limiter = ResourceLimiter(ResourceLimits(memory_mb=2048, timeout_seconds=10))
+        result = limiter.execute("echo hello")
+        assert result.success is True
+        assert result.exit_code == 0
+        assert "hello" in result.stdout
+        assert result.violation == ResourceViolation.NONE
+    
+    def test_timeout_violation(self):
+        limiter = ResourceLimiter(ResourceLimits(timeout_seconds=1))
+        result = limiter.execute("sleep 10")
+        assert result.success is False
+        assert result.violation == ResourceViolation.TIME
+        assert result.killed is True
+    
+    def test_failed_command(self):
+        limiter = ResourceLimiter()
+        result = limiter.execute("exit 1")
+        assert result.success is False
+        assert result.exit_code == 1
+    
+    def test_resource_report(self):
+        from tools.resource_limits import format_resource_report
+        result = ResourceResult(
+            success=True, stdout="", stderr="", exit_code=0,
+            violation=ResourceViolation.NONE, violation_message="",
+            memory_used_mb=100, cpu_time_seconds=0.5, wall_time_seconds=1.0,
+        )
+        report = format_resource_report(result)
+        assert "Exit code: 0" in report
+        assert "100MB" in report
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])

tests/test_skill_deps.py

@@ -1,81 +0,0 @@
-"""
-Tests for skill dependency resolver
-
-Issue: #754
-"""
-
-import unittest
-from unittest.mock import patch, MagicMock
-
-from tools.skill_deps import (
-    parse_requires,
-    check_dependency,
-    check_dependencies,
-    resolve_dependencies,
-)
-
-
-class TestParseRequires(unittest.TestCase):
-    
-    def test_list(self):
-        fm = {"requires": ["pkg1", "pkg2"]}
-        self.assertEqual(parse_requires(fm), ["pkg1", "pkg2"])
-    
-    def test_string(self):
-        fm = {"requires": "pkg1"}
-        self.assertEqual(parse_requires(fm), ["pkg1"])
-    
-    def test_empty(self):
-        fm = {}
-        self.assertEqual(parse_requires(fm), [])
-    
-    def test_none(self):
-        fm = {"requires": None}
-        self.assertEqual(parse_requires(fm), [])
-
-
-class TestCheckDependency(unittest.TestCase):
-    
-    def test_installed(self):
-        installed, version = check_dependency("json")
-        self.assertTrue(installed)
-    
-    def test_not_installed(self):
-        installed, version = check_dependency("nonexistent_package_xyz_123")
-        self.assertFalse(installed)
-
-
-class TestCheckDependencies(unittest.TestCase):
-    
-    def test_all_installed(self):
-        installed, missing = check_dependencies(["json", "os"])
-        self.assertEqual(len(missing), 0)
-    
-    def test_some_missing(self):
-        installed, missing = check_dependencies(["json", "nonexistent_xyz"])
-        self.assertIn("json", installed)
-        self.assertIn("nonexistent_xyz", missing)
-
-
-class TestResolveDependencies(unittest.TestCase):
-    
-    def test_no_requires(self):
-        satisfied, installed, missing = resolve_dependencies({})
-        self.assertTrue(satisfied)
-    
-    def test_all_satisfied(self):
-        satisfied, installed, missing = resolve_dependencies(
-            {"requires": ["json"]}, auto_install=False
-        )
-        self.assertTrue(satisfied)
-    
-    def test_missing_no_auto(self):
-        satisfied, installed, missing = resolve_dependencies(
-            {"requires": ["nonexistent_xyz"]}, auto_install=False
-        )
-        self.assertFalse(satisfied)
-        self.assertIn("nonexistent_xyz", missing)
-
-
-if __name__ == "__main__":
-    unittest.main()

tools/resource_limits.py

@@ -0,0 +1,249 @@
+"""
+Terminal Sandbox Resource Limits — CPU, memory, time.
+
+Provides resource limits for agent terminal commands to prevent
+OOM kills, runaway processes, and excessive resource consumption.
+"""
+
+import logging
+import os
+import signal
+import subprocess
+import time
+from dataclasses import dataclass, field
+from typing import Optional, Dict, Any
+from enum import Enum
+
+logger = logging.getLogger(__name__)
+
+
+class ResourceViolation(Enum):
+    """Types of resource violations."""
+    MEMORY = "memory"
+    CPU = "cpu"
+    TIME = "time"
+    NONE = "none"
+
+
+@dataclass
+class ResourceLimits:
+    """Resource limits for a subprocess."""
+    memory_mb: int = 2048  # 2GB default
+    cpu_percent: int = 80  # 80% of one core
+    timeout_seconds: int = 300  # 5 minutes
+    kill_timeout: int = 10  # SIGKILL after 10s if SIGTERM fails
+
+
+@dataclass
+class ResourceResult:
+    """Result of a resource-limited execution."""
+    success: bool
+    stdout: str
+    stderr: str
+    exit_code: int
+    violation: ResourceViolation
+    violation_message: str
+    memory_used_mb: float
+    cpu_time_seconds: float
+    wall_time_seconds: float
+    killed: bool = False
+
+
+class ResourceLimiter:
+    """Apply resource limits to subprocess execution."""
+    
+    def __init__(self, limits: Optional[ResourceLimits] = None):
+        self.limits = limits or ResourceLimits()
+    
+    def _get_resource_rlimit(self) -> Dict[str, Any]:
+        """Get resource limits for subprocess (Unix only)."""
+        import resource
+        
+        rlimit = {}
+        
+        # Memory limit (RSS)
+        if self.limits.memory_mb > 0:
+            mem_bytes = self.limits.memory_mb * 1024 * 1024
+            rlimit[resource.RLIMIT_AS] = (mem_bytes, mem_bytes)
+        
+        # CPU time limit
+        if self.limits.timeout_seconds > 0:
+            rlimit[resource.RLIMIT_CPU] = (self.limits.timeout_seconds, self.limits.timeout_seconds)
+        
+        return rlimit
+    
+    def _check_resource_usage(self, process: subprocess.Popen) -> Dict[str, float]:
+        """Check resource usage of a process (Unix only)."""
+        try:
+            import resource
+            usage = resource.getrusage(resource.RUSAGE_CHILDREN)
+            return {
+                "user_time": usage.ru_utime,
+                "system_time": usage.ru_stime,
+                "max_rss_mb": usage.ru_maxrss / 1024,  # KB to MB
+            }
+        except:
+            return {"user_time": 0, "system_time": 0, "max_rss_mb": 0}
+    
+    def execute(self, command: str, **kwargs) -> ResourceResult:
+        """
+        Execute a command with resource limits.
+        
+        Args:
+            command: Shell command to execute
+            **kwargs: Additional subprocess arguments
+            
+        Returns:
+            ResourceResult with execution details
+        """
+        start_time = time.time()
+        
+        # Try to use resource limits (Unix only)
+        preexec_fn = None
+        try:
+            import resource
+            rlimit = self._get_resource_rlimit()
+            
+            def set_limits():
+                for res, limits in rlimit.items():
+                    resource.setrlimit(res, limits)
+            
+            preexec_fn = set_limits
+        except ImportError:
+            logger.debug("resource module not available, skipping limits")
+        
+        try:
+            # Execute with timeout
+            result = subprocess.run(
+                command,
+                shell=True,
+                capture_output=True,
+                text=True,
+                timeout=self.limits.timeout_seconds,
+                preexec_fn=preexec_fn,
+                **kwargs,
+            )
+            
+            wall_time = time.time() - start_time
+            usage = self._check_resource_usage(result)
+            
+            # Check for violations
+            violation = ResourceViolation.NONE
+            violation_message = ""
+            
+            # Check memory (if we can get it)
+            if usage["max_rss_mb"] > self.limits.memory_mb:
+                violation = ResourceViolation.MEMORY
+                violation_message = f"Memory limit exceeded: {usage['max_rss_mb']:.0f}MB > {self.limits.memory_mb}MB"
+            
+            return ResourceResult(
+                success=result.returncode == 0,
+                stdout=result.stdout,
+                stderr=result.stderr,
+                exit_code=result.returncode,
+                violation=violation,
+                violation_message=violation_message,
+                memory_used_mb=usage["max_rss_mb"],
+                cpu_time_seconds=usage["user_time"] + usage["system_time"],
+                wall_time_seconds=wall_time,
+            )
+        
+        except subprocess.TimeoutExpired as e:
+            wall_time = time.time() - start_time
+            
+            # Try to kill gracefully
+            if hasattr(e, 'process') and e.process:
+                try:
+                    e.process.terminate()
+                    time.sleep(self.limits.kill_timeout)
+                    if e.process.poll() is None:
+                        e.process.kill()
+                except:
+                    pass
+            
+            return ResourceResult(
+                success=False,
+                stdout=e.stdout.decode() if e.stdout else "",
+                stderr=e.stderr.decode() if e.stderr else "",
+                exit_code=-1,
+                violation=ResourceViolation.TIME,
+                violation_message=f"Timeout after {self.limits.timeout_seconds}s",
+                memory_used_mb=0,
+                cpu_time_seconds=0,
+                wall_time_seconds=wall_time,
+                killed=True,
+            )
+        
+        except MemoryError:
+            wall_time = time.time() - start_time
+            return ResourceResult(
+                success=False,
+                stdout="",
+                stderr=f"Memory limit exceeded ({self.limits.memory_mb}MB)",
+                exit_code=-1,
+                violation=ResourceViolation.MEMORY,
+                violation_message=f"Memory limit exceeded: {self.limits.memory_mb}MB",
+                memory_used_mb=self.limits.memory_mb,
+                cpu_time_seconds=0,
+                wall_time_seconds=wall_time,
+                killed=True,
+            )
+        
+        except Exception as e:
+            wall_time = time.time() - start_time
+            return ResourceResult(
+                success=False,
+                stdout="",
+                stderr=str(e),
+                exit_code=-1,
+                violation=ResourceViolation.NONE,
+                violation_message=f"Execution error: {e}",
+                memory_used_mb=0,
+                cpu_time_seconds=0,
+                wall_time_seconds=wall_time,
+            )
+
+
+def format_resource_report(result: ResourceResult) -> str:
+    """Format resource usage as a report string."""
+    lines = [
+        f"Exit code: {result.exit_code}",
+        f"Wall time: {result.wall_time_seconds:.2f}s",
+        f"CPU time: {result.cpu_time_seconds:.2f}s",
+        f"Memory: {result.memory_used_mb:.0f}MB",
+    ]
+    
+    if result.violation != ResourceViolation.NONE:
+        lines.append(f"⚠️ Violation: {result.violation_message}")
+    
+    if result.killed:
+        lines.append("🔴 Process killed")
+    
+    return " | ".join(lines)
+
+
+def execute_with_limits(
+    command: str,
+    memory_mb: int = 2048,
+    cpu_percent: int = 80,
+    timeout_seconds: int = 300,
+) -> ResourceResult:
+    """
+    Convenience function to execute with resource limits.
+    
+    Args:
+        command: Shell command
+        memory_mb: Memory limit in MB
+        cpu_percent: CPU limit as percent of one core
+        timeout_seconds: Timeout in seconds
+        
+    Returns:
+        ResourceResult
+    """
+    limits = ResourceLimits(
+        memory_mb=memory_mb,
+        cpu_percent=cpu_percent,
+        timeout_seconds=timeout_seconds,
+    )
+    limiter = ResourceLimiter(limits)
+    return limiter.execute(command)

tools/skill_deps.py

@@ -1,185 +0,0 @@
-"""
-Skill Dependency Resolver — Auto-install missing dependencies
-
-Checks skill frontmatter for `requires` field and ensures
-dependencies are installed before loading the skill.
-
-Issue: #754
-"""
-
-import importlib
-import logging
-import subprocess
-import sys
-from typing import Any, Dict, List, Optional, Tuple
-
-logger = logging.getLogger(__name__)
-
-
-def parse_requires(frontmatter: Dict[str, Any]) -> List[str]:
-    """
-    Parse the `requires` field from skill frontmatter.
-    
-    Supports:
-    - requires: [package1, package2]
-    - requires: package1
-    - requires:
-        - package1
-        - package2
-    """
-    requires = frontmatter.get("requires", [])
-    if isinstance(requires, str):
-        return [requires]
-    if isinstance(requires, list):
-        return [str(r) for r in requires if r]
-    return []
-
-
-def check_dependency(package: str) -> Tuple[bool, str]:
-    """
-    Check if a Python package is installed.
-    
-    Returns:
-        Tuple of (is_installed, version_or_error)
-    """
-    # Handle pip package names (e.g., "matrix-nio[e2e]")
-    import_name = package.split("[")[0].replace("-", "_")
-    
-    try:
-        mod = importlib.import_module(import_name)
-        version = getattr(mod, "__version__", "installed")
-        return True, version
-    except ImportError:
-        return False, "not installed"
-
-
-def check_dependencies(requires: List[str]) -> Tuple[List[str], List[str]]:
-    """
-    Check which dependencies are missing.
-    
-    Returns:
-        Tuple of (installed, missing)
-    """
-    installed = []
-    missing = []
-    
-    for pkg in requires:
-        is_installed, _ = check_dependency(pkg)
-        if is_installed:
-            installed.append(pkg)
-        else:
-            missing.append(pkg)
-    
-    return installed, missing
-
-
-def install_dependency(package: str, quiet: bool = False) -> Tuple[bool, str]:
-    """
-    Install a Python package via pip.
-    
-    Returns:
-        Tuple of (success, output_or_error)
-    """
-    try:
-        cmd = [sys.executable, "-m", "pip", "install", package]
-        result = subprocess.run(
-            cmd,
-            capture_output=True,
-            text=True,
-            timeout=120
-        )
-        
-        if result.returncode == 0:
-            if not quiet:
-                logger.info("Installed %s", package)
-            return True, result.stdout
-        else:
-            logger.error("Failed to install %s: %s", package, result.stderr)
-            return False, result.stderr
-    
-    except subprocess.TimeoutExpired:
-        return False, "Installation timed out"
-    except Exception as e:
-        return False, str(e)
-
-
-def resolve_dependencies(
-    frontmatter: Dict[str, Any],
-    auto_install: bool = False,
-    quiet: bool = False
-) -> Tuple[bool, List[str], List[str]]:
-    """
-    Resolve skill dependencies.
-    
-    Args:
-        frontmatter: Skill frontmatter dict
-        auto_install: If True, install missing deps without asking
-        quiet: If True, suppress output
-    
-    Returns:
-        Tuple of (all_satisfied, installed_now, still_missing)
-    """
-    requires = parse_requires(frontmatter)
-    if not requires:
-        return True, [], []
-    
-    installed, missing = check_dependencies(requires)
-    
-    if not missing:
-        if not quiet:
-            logger.debug("All dependencies satisfied: %s", installed)
-        return True, [], []
-    
-    if not auto_install:
-        if not quiet:
-            logger.warning("Missing dependencies: %s", missing)
-        return False, [], missing
-    
-    # Auto-install missing dependencies
-    installed_now = []
-    still_missing = []
-    
-    for pkg in missing:
-        if not quiet:
-            logger.info("Installing missing dependency: %s", pkg)
-        
-        success, output = install_dependency(pkg, quiet=quiet)
-        if success:
-            installed_now.append(pkg)
-        else:
-            still_missing.append(pkg)
-            logger.error("Failed to install %s: %s", pkg, output[:200])
-    
-    all_satisfied = len(still_missing) == 0
-    return all_satisfied, installed_now, still_missing
-
-
-def check_skill_dependencies(skill_dir) -> Dict[str, Any]:
-    """
-    Check dependencies for a skill directory.
-    
-    Returns:
-        Dict with dependency status
-    """
-    from pathlib import Path
-    
-    skill_md = Path(skill_dir) / "SKILL.md"
-    if not skill_md.exists():
-        return {"requires": [], "installed": [], "missing": [], "satisfied": True}
-    
-    try:
-        content = skill_md.read_text()
-        from agent.skill_utils import parse_frontmatter
-        frontmatter, _ = parse_frontmatter(content)
-    except Exception:
-        return {"requires": [], "installed": [], "missing": [], "satisfied": True}
-    
-    requires = parse_requires(frontmatter)
-    installed, missing = check_dependencies(requires)
-    
-    return {
-        "requires": requires,
-        "installed": installed,
-        "missing": missing,
-        "satisfied": len(missing) == 0
-    }