feat(computer): add intelligent video slicing with activity detection

MediaTool v0.4.0 - Smart video compression for Claude interpretation: NEW ACTIONS: - analyze: Detect activity segments (movement, scene changes) - slice: Extract frames ONLY at activity points - compress_session: Full pipeline for computer use sessions ACTIVITY DETECTION: - Frame differencing (PIL + numpy) for movement detection - FFmpeg scene change detection for major transitions - Activity clustering into segments with scores - Configurable sensitivity thresholds COMPRESSION: - 512px max dimension for session frames (vs 768px default) - 60% JPEG quality for aggressive compression - Target 30 frames per 60-second session - ~15KB per frame, ~450KB total for typical session NEW ENV VARS: HANZO_MEDIA_ACTIVITY_THRESHOLD=0.02 HANZO_MEDIA_SCENE_THRESHOLD=0.3 HANZO_MEDIA_SESSION_MAX_DURATION=60 HANZO_MEDIA_SESSION_TARGET_FRAMES=30 HANZO_MEDIA_SESSION_QUALITY=60 Usage: media(action="compress_session", path="recording.mp4") # Returns compressed keyframes + activity analysis # Perfect for Claude to interpret computer use sessions Added numpy>=1.26.0 dependency for frame differencing.
2026-10-23 20:23:00 +00:00 · 2026-01-11 19:52:05 -08:00
parent e940ca5f1a
commit dbd18eca93
5 changed files with 2476 additions and 30 deletions
@@ -31,7 +31,7 @@ from hanzo_tools.core import BaseTool, ToolRegistry, PermissionManager
 from .computer_tool import ComputerTool
 from .fast_computer import FastComputerTool
 from .video_tool import VideoTool
-from .media_tool import MediaTool, MediaLimits, media_tool
+from .media_tool import MediaTool, MediaLimits, MediaResult, ActivitySegment, media_tool

 # Export list for tool discovery
 TOOLS = [ComputerTool, FastComputerTool, VideoTool, MediaTool]
@@ -42,6 +42,8 @@ __all__ = [
    "VideoTool",
    "MediaTool",
    "MediaLimits",
+    "MediaResult",
+    "ActivitySegment",
    "media_tool",
    "register_tools",
    "TOOLS",
@@ -5,6 +5,8 @@ Handles images and video with configurable limits optimized for Claude's vision
 - Combined payload under 32 MB (configurable)
 - Per-image resolution constraints (configurable)
 - Automatic optimization for Claude vision
+- INTELLIGENT VIDEO SLICING with activity detection
+- Computer use session compression

 Limits based on Claude API constraints:
 - Max 100 images per request
@@ -12,12 +14,19 @@ Limits based on Claude API constraints:
 - Recommended 768-1568px for best quality/speed tradeoff
 - Supports PNG, JPEG, GIF, WebP

+Activity Detection:
+- Frame differencing for movement detection
+- Scene change detection via FFmpeg
+- Configurable sensitivity threshold
+- Only extracts frames during activity periods
+
 Environment configuration:
 - HANZO_MEDIA_MAX_IMAGES: Max images per batch (default: 100)
 - HANZO_MEDIA_MAX_PAYLOAD_MB: Max total payload in MB (default: 32)
 - HANZO_MEDIA_MAX_RESOLUTION: Max image dimension (default: 1568)
 - HANZO_MEDIA_JPEG_QUALITY: JPEG quality 1-100 (default: 85)
 - HANZO_MEDIA_OPTIMAL_SIZE: Target size for optimization (default: 768)
+- HANZO_MEDIA_ACTIVITY_THRESHOLD: Activity detection sensitivity (default: 0.02)
 """

 import os
@@ -61,6 +70,16 @@ class MediaLimits:
    max_frames: int = 100  # Max video frames per extraction
    max_fps: int = 30  # Max frame rate for extraction

+    # Activity detection settings
+    activity_threshold: float = 0.02  # % of pixels changed to count as activity (0.01-0.10)
+    min_activity_gap_ms: int = 200  # Min ms between activity frames
+    scene_change_threshold: float = 0.3  # FFmpeg scene change threshold
+
+    # Session compression settings
+    session_max_duration: int = 60  # Max session duration in seconds
+    session_target_frames: int = 30  # Target frames for session summary
+    session_compression_quality: int = 60  # Aggressive compression for sessions
+
    @classmethod
    def from_env(cls) -> "MediaLimits":
        """Load limits from environment variables."""
@@ -71,6 +90,12 @@ class MediaLimits:
            optimal_size=int(os.environ.get("HANZO_MEDIA_OPTIMAL_SIZE", "768")),
            jpeg_quality=int(os.environ.get("HANZO_MEDIA_JPEG_QUALITY", "85")),
            max_frames=int(os.environ.get("HANZO_MEDIA_MAX_FRAMES", "100")),
+            activity_threshold=float(os.environ.get("HANZO_MEDIA_ACTIVITY_THRESHOLD", "0.02")),
+            min_activity_gap_ms=int(os.environ.get("HANZO_MEDIA_MIN_ACTIVITY_GAP_MS", "200")),
+            scene_change_threshold=float(os.environ.get("HANZO_MEDIA_SCENE_THRESHOLD", "0.3")),
+            session_max_duration=int(os.environ.get("HANZO_MEDIA_SESSION_MAX_DURATION", "60")),
+            session_target_frames=int(os.environ.get("HANZO_MEDIA_SESSION_TARGET_FRAMES", "30")),
+            session_compression_quality=int(os.environ.get("HANZO_MEDIA_SESSION_QUALITY", "60")),
        )

    @property
@@ -261,6 +286,341 @@ def _extract_video_frames(
    return frames


+@dataclass
+class ActivitySegment:
+    """A detected activity segment in video."""
+    start_ms: int
+    end_ms: int
+    activity_score: float  # 0-1, how much activity
+    frame_count: int
+    description: str = ""
+
+
+def _detect_scene_changes(
+    video_path: str,
+    threshold: float = 0.3,
+    max_duration: int = 60,
+) -> list[float]:
+    """Detect scene changes using FFmpeg.
+
+    Returns list of timestamps (in seconds) where scenes change.
+    """
+    import subprocess
+
+    # Use FFmpeg scene detection filter
+    cmd = [
+        "ffprobe", "-v", "quiet",
+        "-show_entries", "frame=pts_time",
+        "-select_streams", "v:0",
+        "-of", "csv=p=0",
+        "-f", "lavfi",
+        f"movie={video_path},select='gt(scene,{threshold})'",
+    ]
+
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+        timestamps = []
+        for line in result.stdout.strip().split('\n'):
+            if line:
+                try:
+                    ts = float(line)
+                    if ts <= max_duration:
+                        timestamps.append(ts)
+                except ValueError:
+                    continue
+        return timestamps
+    except Exception:
+        return []
+
+
+def _compute_frame_difference(
+    frame1_data: bytes,
+    frame2_data: bytes,
+    threshold: float = 0.02,
+) -> tuple[float, bool]:
+    """Compute difference between two frames.
+
+    Returns (difference_ratio, has_significant_activity).
+    difference_ratio is 0-1 representing % of pixels that changed.
+    """
+    try:
+        from PIL import Image, ImageChops
+        import numpy as np
+    except ImportError:
+        return 0.0, False
+
+    try:
+        img1 = Image.open(io.BytesIO(frame1_data)).convert('L')  # Grayscale
+        img2 = Image.open(io.BytesIO(frame2_data)).convert('L')
+
+        # Ensure same size
+        if img1.size != img2.size:
+            img2 = img2.resize(img1.size)
+
+        # Compute difference
+        diff = ImageChops.difference(img1, img2)
+
+        # Count pixels above threshold
+        diff_array = np.array(diff)
+        changed_pixels = np.sum(diff_array > 10)  # Pixel value threshold
+        total_pixels = diff_array.size
+
+        ratio = changed_pixels / total_pixels if total_pixels > 0 else 0
+
+        return ratio, ratio > threshold
+
+    except Exception:
+        return 0.0, False
+
+
+def _analyze_video_activity(
+    video_path: str,
+    activity_threshold: float = 0.02,
+    sample_fps: float = 2.0,
+    max_duration: int = 60,
+    scene_threshold: float = 0.3,
+) -> tuple[list[ActivitySegment], list[float]]:
+    """Analyze video for activity segments.
+
+    Returns (activity_segments, keyframe_timestamps).
+
+    Combines:
+    1. Scene change detection (FFmpeg)
+    2. Frame differencing (PIL)
+    3. Activity clustering
+    """
+    import subprocess
+    import tempfile
+
+    segments = []
+    keyframe_times = []
+
+    # Get video duration
+    probe_cmd = [
+        "ffprobe", "-v", "error",
+        "-show_entries", "format=duration",
+        "-of", "default=noprint_wrappers=1:nokey=1",
+        video_path
+    ]
+    try:
+        result = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=10)
+        duration = min(float(result.stdout.strip()), max_duration)
+    except Exception:
+        duration = max_duration
+
+    # Get scene changes
+    scene_changes = _detect_scene_changes(video_path, scene_threshold, max_duration)
+    keyframe_times.extend(scene_changes)
+
+    # Sample frames for activity detection
+    sample_interval = 1.0 / sample_fps
+    num_samples = int(duration * sample_fps)
+
+    prev_frame = None
+    activity_frames = []
+
+    for i in range(num_samples):
+        timestamp = i * sample_interval
+        if timestamp > duration:
+            break
+
+        # Extract frame
+        with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
+            tmp_path = f.name
+
+        try:
+            cmd = [
+                "ffmpeg", "-y",
+                "-ss", str(timestamp),
+                "-i", video_path,
+                "-frames:v", "1",
+                "-vf", "scale=320:-1",  # Small for comparison
+                "-q:v", "5",
+                tmp_path
+            ]
+            subprocess.run(cmd, capture_output=True, timeout=5)
+
+            if os.path.exists(tmp_path) and os.path.getsize(tmp_path) > 0:
+                with open(tmp_path, "rb") as f:
+                    frame_data = f.read()
+
+                if prev_frame is not None:
+                    diff_ratio, has_activity = _compute_frame_difference(
+                        prev_frame, frame_data, activity_threshold
+                    )
+                    if has_activity:
+                        activity_frames.append({
+                            "timestamp": timestamp,
+                            "diff_ratio": diff_ratio,
+                        })
+                        if timestamp not in keyframe_times:
+                            keyframe_times.append(timestamp)
+
+                prev_frame = frame_data
+
+        finally:
+            if os.path.exists(tmp_path):
+                os.unlink(tmp_path)
+
+    # Cluster activity into segments
+    if activity_frames:
+        current_segment_start = activity_frames[0]["timestamp"]
+        current_segment_score = activity_frames[0]["diff_ratio"]
+        segment_frame_count = 1
+
+        for frame in activity_frames[1:]:
+            # If gap > 1 second, start new segment
+            if frame["timestamp"] - current_segment_start > 1.0 + segment_frame_count * sample_interval:
+                segments.append(ActivitySegment(
+                    start_ms=int(current_segment_start * 1000),
+                    end_ms=int((current_segment_start + segment_frame_count * sample_interval) * 1000),
+                    activity_score=current_segment_score / segment_frame_count,
+                    frame_count=segment_frame_count,
+                ))
+                current_segment_start = frame["timestamp"]
+                current_segment_score = frame["diff_ratio"]
+                segment_frame_count = 1
+            else:
+                current_segment_score += frame["diff_ratio"]
+                segment_frame_count += 1
+
+        # Last segment
+        segments.append(ActivitySegment(
+            start_ms=int(current_segment_start * 1000),
+            end_ms=int((current_segment_start + segment_frame_count * sample_interval) * 1000),
+            activity_score=current_segment_score / segment_frame_count,
+            frame_count=segment_frame_count,
+        ))
+
+    # Sort and deduplicate keyframe times
+    keyframe_times = sorted(set(keyframe_times))
+
+    return segments, keyframe_times
+
+
+def _extract_activity_frames(
+    video_path: str,
+    keyframe_times: list[float],
+    max_frames: int = 30,
+    max_size: int = 512,
+    quality: int = 60,
+) -> list[tuple[bytes, dict]]:
+    """Extract frames at specified timestamps with heavy compression.
+
+    Optimized for minimal payload size.
+    """
+    import subprocess
+    import tempfile
+
+    frames = []
+
+    # Limit frames
+    if len(keyframe_times) > max_frames:
+        # Sample evenly
+        step = len(keyframe_times) / max_frames
+        keyframe_times = [keyframe_times[int(i * step)] for i in range(max_frames)]
+
+    for timestamp in keyframe_times:
+        with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
+            tmp_path = f.name
+
+        try:
+            cmd = [
+                "ffmpeg", "-y",
+                "-ss", str(timestamp),
+                "-i", video_path,
+                "-frames:v", "1",
+                "-vf", f"scale='min({max_size},iw)':'min({max_size},ih)':force_original_aspect_ratio=decrease",
+                "-q:v", str(max(1, min(31, (100 - quality) // 3))),  # FFmpeg quality 1-31
+                tmp_path
+            ]
+            subprocess.run(cmd, capture_output=True, timeout=10)
+
+            if os.path.exists(tmp_path) and os.path.getsize(tmp_path) > 0:
+                with open(tmp_path, "rb") as f:
+                    data = f.read()
+
+                frames.append((data, {
+                    "timestamp_ms": int(timestamp * 1000),
+                    "timestamp_sec": round(timestamp, 2),
+                    "size": len(data),
+                    "format": "jpeg",
+                }))
+
+        finally:
+            if os.path.exists(tmp_path):
+                os.unlink(tmp_path)
+
+    return frames
+
+
+def _compress_session(
+    video_path: str,
+    max_duration: int = 60,
+    target_frames: int = 30,
+    activity_threshold: float = 0.02,
+    scene_threshold: float = 0.3,
+    max_size: int = 512,
+    quality: int = 60,
+) -> tuple[list[tuple[bytes, dict]], list[ActivitySegment], dict]:
+    """Full pipeline: analyze → slice → compress a computer use session.
+
+    Returns (frames, activity_segments, metadata).
+    """
+    # Analyze video
+    segments, keyframe_times = _analyze_video_activity(
+        video_path,
+        activity_threshold=activity_threshold,
+        max_duration=max_duration,
+        scene_threshold=scene_threshold,
+    )
+
+    # If no activity detected, sample evenly
+    if not keyframe_times:
+        import subprocess
+        probe_cmd = [
+            "ffprobe", "-v", "error",
+            "-show_entries", "format=duration",
+            "-of", "default=noprint_wrappers=1:nokey=1",
+            video_path
+        ]
+        try:
+            result = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=10)
+            duration = min(float(result.stdout.strip()), max_duration)
+        except Exception:
+            duration = max_duration
+
+        interval = duration / target_frames
+        keyframe_times = [i * interval for i in range(target_frames)]
+
+    # Extract frames at activity points
+    frames = _extract_activity_frames(
+        video_path,
+        keyframe_times,
+        max_frames=target_frames,
+        max_size=max_size,
+        quality=quality,
+    )
+
+    # Compute metadata
+    total_size = sum(len(f[0]) for f in frames)
+    metadata = {
+        "source": video_path,
+        "activity_segments": len(segments),
+        "keyframes_detected": len(keyframe_times),
+        "frames_extracted": len(frames),
+        "total_size_bytes": total_size,
+        "total_size_kb": round(total_size / 1024, 1),
+        "avg_frame_size_kb": round(total_size / len(frames) / 1024, 1) if frames else 0,
+        "compression_settings": {
+            "max_size": max_size,
+            "quality": quality,
+        },
+    }
+
+    return frames, segments, metadata
+
+
 Action = Literal[
    # Image operations
    "load",           # Load single image
@@ -270,6 +630,10 @@ Action = Literal[
    "info",           # Get image info
    # Video operations
    "extract_frames", # Extract frames from video
+    # Activity detection & slicing (NEW)
+    "analyze",        # Detect activity segments in video
+    "slice",          # Extract frames at activity points only
+    "compress_session",  # Full pipeline for computer use sessions
    # Configuration
    "limits",         # Get/set limits
    "status",         # Get current status
@@ -341,6 +705,27 @@ extract_frames: Extract frames from video
  - interval_ms: Time between frames (default: 1000)
  - optimize: Optimize frames for Claude (default: true)

+analyze: Detect activity segments in video (movement, scene changes)
+  - path: Video file path
+  - activity_threshold: Sensitivity 0.01-0.10 (default: {self.limits.activity_threshold})
+  - scene_threshold: Scene change sensitivity (default: {self.limits.scene_change_threshold})
+  - max_duration: Max seconds to analyze (default: {self.limits.session_max_duration})
+  - Returns activity segments with timestamps and scores
+
+slice: Extract frames ONLY at activity points
+  - path: Video file path
+  - target_frames: Max frames to extract (default: {self.limits.session_target_frames})
+  - activity_threshold: Sensitivity (default: {self.limits.activity_threshold})
+  - max_size: Frame dimension (default: 512 for compression)
+  - quality: JPEG quality (default: {self.limits.session_compression_quality})
+
+compress_session: FULL PIPELINE for computer use video
+  - path: Video file path (e.g., 30-second screen recording)
+  - target_frames: Max frames (default: {self.limits.session_target_frames})
+  - max_size: Frame dimension (default: 512)
+  - quality: Compression quality (default: {self.limits.session_compression_quality})
+  - Returns compressed keyframes + activity analysis for Claude interpretation
+
 info: Get image/video info without loading full data
  - path: File path

@@ -357,12 +742,17 @@ ENVIRONMENT VARIABLES:
  HANZO_MEDIA_MAX_RESOLUTION={self.limits.max_resolution}
  HANZO_MEDIA_OPTIMAL_SIZE={self.limits.optimal_size}
  HANZO_MEDIA_JPEG_QUALITY={self.limits.jpeg_quality}
+  HANZO_MEDIA_ACTIVITY_THRESHOLD={self.limits.activity_threshold}
+  HANZO_MEDIA_SESSION_TARGET_FRAMES={self.limits.session_target_frames}
+  HANZO_MEDIA_SESSION_QUALITY={self.limits.session_compression_quality}

 EXAMPLES:
  media(action="load", path="screenshot.png")
  media(action="load_batch", paths=["img1.png", "img2.jpg", "img3.webp"])
  media(action="extract_frames", path="video.mp4", count=20, interval_ms=500)
-  media(action="optimize", images=[base64_data], max_size=768)
+  media(action="analyze", path="session.mp4")  # Detect activity
+  media(action="slice", path="session.mp4", target_frames=30)  # Activity frames only
+  media(action="compress_session", path="recording.mp4")  # Full pipeline for Claude
  media(action="limits", max_images=50, max_payload_mb=16)
 """

@@ -757,6 +1147,173 @@ EXAMPLES:

                return json.dumps(info)

+            elif action == "analyze":
+                if not path:
+                    return json.dumps({"error": "path required (video file)"})
+
+                if not os.path.exists(path):
+                    return json.dumps({"error": f"File not found: {path}"})
+
+                # Get optional params from kwargs
+                activity_threshold = kwargs.get("activity_threshold", self.limits.activity_threshold)
+                scene_threshold = kwargs.get("scene_threshold", self.limits.scene_change_threshold)
+                max_duration = kwargs.get("max_duration", self.limits.session_max_duration)
+
+                segments, keyframe_times = await loop.run_in_executor(
+                    _EXECUTOR,
+                    _analyze_video_activity,
+                    path,
+                    activity_threshold,
+                    2.0,  # sample_fps
+                    max_duration,
+                    scene_threshold,
+                )
+
+                return json.dumps({
+                    "success": True,
+                    "source": path,
+                    "activity_segments": [
+                        {
+                            "start_ms": s.start_ms,
+                            "end_ms": s.end_ms,
+                            "duration_ms": s.end_ms - s.start_ms,
+                            "activity_score": round(s.activity_score, 4),
+                            "frame_count": s.frame_count,
+                        }
+                        for s in segments
+                    ],
+                    "keyframe_count": len(keyframe_times),
+                    "keyframe_times_sec": [round(t, 2) for t in keyframe_times],
+                    "settings": {
+                        "activity_threshold": activity_threshold,
+                        "scene_threshold": scene_threshold,
+                        "max_duration": max_duration,
+                    },
+                })
+
+            elif action == "slice":
+                if not path:
+                    return json.dumps({"error": "path required (video file)"})
+
+                if not os.path.exists(path):
+                    return json.dumps({"error": f"File not found: {path}"})
+
+                # Get params
+                target_frames = kwargs.get("target_frames", self.limits.session_target_frames)
+                activity_threshold = kwargs.get("activity_threshold", self.limits.activity_threshold)
+                scene_threshold = kwargs.get("scene_threshold", self.limits.scene_change_threshold)
+                max_duration = kwargs.get("max_duration", self.limits.session_max_duration)
+                slice_max_size = kwargs.get("max_size", 512)  # Smaller default for compression
+                slice_quality = kwargs.get("quality", self.limits.session_compression_quality)
+
+                # First analyze
+                segments, keyframe_times = await loop.run_in_executor(
+                    _EXECUTOR,
+                    _analyze_video_activity,
+                    path,
+                    activity_threshold,
+                    2.0,
+                    max_duration,
+                    scene_threshold,
+                )
+
+                # Then extract at activity points
+                frames = await loop.run_in_executor(
+                    _EXECUTOR,
+                    _extract_activity_frames,
+                    path,
+                    keyframe_times,
+                    target_frames,
+                    slice_max_size,
+                    slice_quality,
+                )
+
+                result = MediaResult(success=True)
+
+                for data, info in frames:
+                    if result.total_size + len(data) > self.limits.max_payload_bytes:
+                        result.warnings.append("Payload limit reached")
+                        break
+
+                    result.images.append({
+                        "timestamp_sec": info["timestamp_sec"],
+                        "timestamp_ms": info["timestamp_ms"],
+                        "size": len(data),
+                        "format": info["format"],
+                        "base64": base64.b64encode(data).decode(),
+                    })
+                    result.total_size += len(data)
+                    result.total_count += 1
+
+                return json.dumps({
+                    **result.to_dict(),
+                    "source": path,
+                    "activity_segments": len(segments),
+                    "keyframes_detected": len(keyframe_times),
+                    "compression_settings": {
+                        "max_size": slice_max_size,
+                        "quality": slice_quality,
+                    },
+                })
+
+            elif action == "compress_session":
+                if not path:
+                    return json.dumps({"error": "path required (video file)"})
+
+                if not os.path.exists(path):
+                    return json.dumps({"error": f"File not found: {path}"})
+
+                # Get params
+                target_frames = kwargs.get("target_frames", self.limits.session_target_frames)
+                activity_threshold = kwargs.get("activity_threshold", self.limits.activity_threshold)
+                scene_threshold = kwargs.get("scene_threshold", self.limits.scene_change_threshold)
+                max_duration = kwargs.get("max_duration", self.limits.session_max_duration)
+                session_max_size = kwargs.get("max_size", 512)
+                session_quality = kwargs.get("quality", self.limits.session_compression_quality)
+
+                # Full pipeline
+                frames, segments, metadata = await loop.run_in_executor(
+                    _EXECUTOR,
+                    _compress_session,
+                    path,
+                    max_duration,
+                    target_frames,
+                    activity_threshold,
+                    scene_threshold,
+                    session_max_size,
+                    session_quality,
+                )
+
+                result = MediaResult(success=True)
+
+                for data, info in frames:
+                    if result.total_size + len(data) > self.limits.max_payload_bytes:
+                        result.warnings.append("Payload limit reached")
+                        break
+
+                    result.images.append({
+                        "timestamp_sec": info["timestamp_sec"],
+                        "timestamp_ms": info["timestamp_ms"],
+                        "size": len(data),
+                        "format": info["format"],
+                        "base64": base64.b64encode(data).decode(),
+                    })
+                    result.total_size += len(data)
+                    result.total_count += 1
+
+                return json.dumps({
+                    **result.to_dict(),
+                    **metadata,
+                    "activity_segments": [
+                        {
+                            "start_ms": s.start_ms,
+                            "end_ms": s.end_ms,
+                            "activity_score": round(s.activity_score, 4),
+                        }
+                        for s in segments
+                    ],
+                })
+
            else:
                return json.dumps({"error": f"Unknown action: {action}"})

@@ -782,6 +1339,12 @@ EXAMPLES:
            quality: Annotated[Optional[int], Field(description="JPEG quality 1-100")] = None,
            count: Annotated[int, Field(description="Frame count")] = 10,
            interval_ms: Annotated[int, Field(description="Frame interval ms")] = 1000,
+            # Activity detection params
+            target_frames: Annotated[Optional[int], Field(description="Target frames for activity slicing")] = None,
+            activity_threshold: Annotated[Optional[float], Field(description="Activity sensitivity 0.01-0.10")] = None,
+            scene_threshold: Annotated[Optional[float], Field(description="Scene change sensitivity")] = None,
+            max_duration: Annotated[Optional[int], Field(description="Max seconds to analyze")] = None,
+            # Limit updates
            max_images: Annotated[Optional[int], Field(description="Update max images limit")] = None,
            max_payload_mb: Annotated[Optional[float], Field(description="Update max payload MB")] = None,
            max_resolution: Annotated[Optional[int], Field(description="Update max resolution")] = None,
@@ -804,6 +1367,11 @@ EXAMPLES:
                max_images=max_images,
                max_payload_mb=max_payload_mb,
                max_resolution=max_resolution,
+                # Pass activity params as kwargs
+                target_frames=target_frames,
+                activity_threshold=activity_threshold,
+                scene_threshold=scene_threshold,
+                max_duration=max_duration,
            )


@@ -42,10 +42,13 @@ from hanzo_tools.core import BaseTool, PermissionManager, auto_timeout
 # Thread pool for blocking operations
 _EXECUTOR = ThreadPoolExecutor(max_workers=2, thread_name_prefix="video_")

-# Claude vision optimal settings
-CLAUDE_MAX_FRAMES = 10  # Max frames per request
-CLAUDE_OPTIMAL_SIZE = 768  # Optimal long edge size
-CLAUDE_JPEG_QUALITY = 85  # JPEG quality for compression
+# Claude vision optimal settings (matches API limits)
+CLAUDE_MAX_FRAMES = 100  # Max frames per request (API limit)
+CLAUDE_MAX_PAYLOAD_MB = 32  # Max total payload in MB
+CLAUDE_OPTIMAL_SIZE = 768  # Optimal long edge size for speed/quality balance
+CLAUDE_JPEG_QUALITY_KEYFRAME = 85  # Quality for keyframes (every Nth frame)
+CLAUDE_JPEG_QUALITY_DELTA = 60  # Lower quality for in-between frames (Claude interpolates)
+CLAUDE_KEYFRAME_INTERVAL = 5  # Every 5th frame is a keyframe (higher quality)

 Action = Literal[
    # Recording
@@ -67,13 +70,33 @@ Action = Literal[
 ]


-def _resize_for_claude(image_data: bytes, max_size: int = CLAUDE_OPTIMAL_SIZE) -> bytes:
-    """Resize image for Claude vision API - optimal size and JPEG compression."""
+def _resize_for_claude(
+    image_data: bytes,
+    max_size: int = CLAUDE_OPTIMAL_SIZE,
+    quality: int = CLAUDE_JPEG_QUALITY_KEYFRAME,
+    is_keyframe: bool = True,
+) -> bytes:
+    """Resize image for Claude vision API - optimal size and JPEG compression.
+
+    Args:
+        image_data: Raw image bytes
+        max_size: Max dimension (width or height)
+        quality: JPEG quality (1-100)
+        is_keyframe: If True, use higher quality. If False, use delta compression.
+
+    Keyframe strategy:
+        - Every Nth frame is a keyframe (higher quality, ~100-200KB)
+        - In-between frames use lower quality (~50-100KB) since Claude interpolates
+        - This allows 100 frames at ~10MB total vs ~20MB with uniform quality
+    """
    try:
        from PIL import Image
    except ImportError:
        return image_data  # Return original if PIL not available

+    # Use lower quality for non-keyframes (Claude can interpolate)
+    actual_quality = quality if is_keyframe else min(quality, CLAUDE_JPEG_QUALITY_DELTA)
+
    img = Image.open(io.BytesIO(image_data))

    # Calculate new size maintaining aspect ratio
@@ -87,7 +110,7 @@ def _resize_for_claude(image_data: bytes, max_size: int = CLAUDE_OPTIMAL_SIZE) -
    buffer = io.BytesIO()
    if img.mode in ('RGBA', 'P'):
        img = img.convert('RGB')
-    img.save(buffer, format='JPEG', quality=CLAUDE_JPEG_QUALITY, optimize=True)
+    img.save(buffer, format='JPEG', quality=actual_quality, optimize=True)

    return buffer.getvalue()

@@ -180,11 +203,13 @@ Actions:
  - interval: Time between frames in ms (default 100)

 - frames_claude: RECOMMENDED - Claude-optimized frame capture
+  - Max 100 frames (Claude API limit)
+  - Max 32MB total payload
+  - Keyframe compression: every 5th frame at 85%, others at 60%
+  - Claude interpolates between frames, so delta frames can be smaller
  - Automatically resizes to 768px (optimal for Claude vision)
-  - JPEG compression (~100-200KB per frame)
-  - Max 10 frames to fit context limits
  - Skips duplicate frames automatically
-  - count: Number of frames (max 10)
+  - count: Number of frames (max 100)
  - interval: Time between frames in ms (default 100)

 - stream: Start frame streaming
@@ -204,15 +229,17 @@ Examples:
    video(action="start", fps=30, quality="high")
    video(action="frame")  # Get single frame for AI analysis
    video(action="frames", count=5, interval=200)  # 5 frames, 200ms apart
-    video(action="frames_claude", count=10, interval=500)  # Claude-optimized: 10 frames over 5s
+    video(action="frames_claude", count=100, interval=100)  # 100 frames over 10s
    video(action="stop")
    video(action="gif", output="demo.gif")

 Claude Vision Best Practices:
    - Use frames_claude for video analysis (auto-optimized)
-    - 10 frames max per request (~1-2MB total)
-    - 500-1000ms intervals capture meaningful changes
-    - Frames are 768px JPEG for fast processing
+    - 100 frames max, 32MB max payload
+    - Keyframe strategy: every 5th frame is higher quality
+    - Delta frames use 60% quality (Claude interpolates)
+    - 100-500ms intervals capture meaningful changes
+    - ~10-15MB typical for 100 frames with keyframe compression
 """

    def _capture_frame_native(self, region: Optional[list[int]] = None) -> bytes:
@@ -379,20 +406,29 @@ Claude Vision Best Practices:
                })

            elif action == "frames_claude":
-                # Claude-optimized frame capture
-                # - Max 10 frames to fit context limits
-                # - Resized to 768px for optimal vision processing
-                # - JPEG compressed to ~100-200KB per frame
+                # Claude-optimized frame capture with keyframe compression
+                # - Max 100 frames (Claude API limit)
+                # - Max 32MB total payload
+                # - Keyframe strategy: every 5th frame at 85% quality, others at 60%
+                # - Claude can interpolate between frames, so delta frames can be smaller
                # - Skip duplicate/similar frames
                actual_count = min(count, CLAUDE_MAX_FRAMES)
+                max_payload_bytes = int(CLAUDE_MAX_PAYLOAD_MB * 1024 * 1024)
                frames_data = []
                last_frame_hash = None
                skipped = 0
+                total_size = 0
+                keyframes = 0
+                delta_frames = 0

-                for i in range(actual_count + skipped):
+                for i in range(actual_count * 2):  # Allow for skipping duplicates
                    if len(frames_data) >= actual_count:
                        break

+                    # Check payload limit (leave 1MB buffer)
+                    if total_size >= max_payload_bytes - 1024 * 1024:
+                        break
+
                    if self._check_screencapture():
                        raw_data = await run(self._capture_frame_native, region)
                    else:
@@ -407,33 +443,58 @@ Claude Vision Best Practices:

                    last_frame_hash = frame_hash

-                    # Resize and compress for Claude
-                    optimized = await run(_resize_for_claude, raw_data, CLAUDE_OPTIMAL_SIZE)
+                    # Keyframe strategy: every Nth frame is higher quality
+                    frame_index = len(frames_data)
+                    is_keyframe = (frame_index % CLAUDE_KEYFRAME_INTERVAL == 0)
+
+                    # Resize and compress for Claude with keyframe-aware quality
+                    optimized = await run(
+                        _resize_for_claude,
+                        raw_data,
+                        CLAUDE_OPTIMAL_SIZE,
+                        CLAUDE_JPEG_QUALITY_KEYFRAME,
+                        is_keyframe,
+                    )
+
+                    # Check if this frame would exceed payload
+                    if total_size + len(optimized) > max_payload_bytes:
+                        break
+
+                    if is_keyframe:
+                        keyframes += 1
+                    else:
+                        delta_frames += 1

                    frames_data.append({
-                        "index": len(frames_data),
+                        "index": frame_index,
                        "timestamp_ms": i * interval,
+                        "is_keyframe": is_keyframe,
                        "original_size": len(raw_data),
                        "optimized_size": len(optimized),
                        "format": "jpeg",
                        "base64": base64.b64encode(optimized).decode(),
                    })
+                    total_size += len(optimized)

-                    if i < actual_count + skipped - 1:
+                    if i < actual_count * 2 - 1:
                        await asyncio.sleep(interval / 1000)

-                total_size = sum(f["optimized_size"] for f in frames_data)
-
                return json.dumps({
                    "success": True,
                    "count": len(frames_data),
                    "max_frames": CLAUDE_MAX_FRAMES,
+                    "max_payload_mb": CLAUDE_MAX_PAYLOAD_MB,
                    "optimal_size": CLAUDE_OPTIMAL_SIZE,
+                    "keyframes": keyframes,
+                    "delta_frames": delta_frames,
+                    "keyframe_quality": CLAUDE_JPEG_QUALITY_KEYFRAME,
+                    "delta_quality": CLAUDE_JPEG_QUALITY_DELTA,
                    "skipped_duplicates": skipped,
                    "total_size_kb": round(total_size / 1024, 1),
+                    "total_size_mb": round(total_size / (1024 * 1024), 2),
                    "interval_ms": interval,
                    "frames": frames_data,
-                    "note": f"Frames optimized for Claude vision API ({CLAUDE_OPTIMAL_SIZE}px, JPEG {CLAUDE_JPEG_QUALITY}%)"
+                    "note": f"100 frames max, 32MB limit. Keyframes (1/{CLAUDE_KEYFRAME_INTERVAL}) at {CLAUDE_JPEG_QUALITY_KEYFRAME}%, deltas at {CLAUDE_JPEG_QUALITY_DELTA}%",
                })

            elif action == "start":
@@ -4,8 +4,8 @@ build-backend = "hatchling.build"

 [project]
 name = "hanzo-tools-computer"
-version = "0.3.0"
-description = "Computer control tools for Hanzo AI - automation, video capture, media processing (100 images/32MB configurable limits)"
+version = "0.4.0"
+description = "Computer control tools for Hanzo AI - automation, video capture, media processing with activity detection (100 images/32MB limits)"
 readme = "README.md"
 license = "MIT"
 requires-python = ">=3.12"
@@ -25,6 +25,7 @@ dependencies = [
    "hanzo-tools>=0.3.0",
    "pyautogui>=0.9.54",
    "pillow>=10.0.0",
+    "numpy>=1.26.0",
    "mcp>=1.25.0",
    "pydantic>=2.0.0",
 ]