fix(video): add an optional frame cap for universal analysis

artemis/agents/video_analyzer/universal_engine.py

@@ -23,6 +23,7 @@
 
 import asyncio
 import base64
+from bisect import bisect_left, bisect_right
 import glob
 from pathlib import Path
 import re
@@ -59,9 +60,23 @@ async def exec_single_chunk_universal(
 ) -> str:
     """Executes sub-agent chunk analysis via keyframe extraction and LangChain ChatModel."""
     keyframes = _gather_universal_frames(analyzer, raw_path, start_time, end_time, actual_start)
+    source_frame_count = len(keyframes)
+    frame_limit = analyzer.max_total_frames
+    if frame_limit is not None:
+        actions = analyzer._action_timestamps(
+            start_time, end_time if end_time is not None else start_time
+        )
+        keyframes = _select_video_frames(
+            keyframes, frame_limit, [timestamp - actual_start for timestamp in actions]
+        )
 
     user_blocks, audio_block = await _build_universal_user_blocks(
-        analyzer, keyframes, actual_start, prompt_with_context, raw_path
+        analyzer,
+        keyframes,
+        actual_start,
+        prompt_with_context,
+        raw_path,
+        sampling_source_count=source_frame_count if frame_limit is not None else None,
     )
 
     messages = [
@@ -76,6 +91,7 @@ async def exec_single_chunk_universal(
             name="video_sub_agent_universal",
             payload={
                 "num_keyframes": len(keyframes),
+                "source_num_keyframes": source_frame_count,
                 "prompt": prompt_with_context,
                 "system_instruction": analyzer.sub_system_prompt,
             },
@@ -137,15 +153,64 @@ def _gather_universal_frames(
     return sorted(merged_frames.items())
 
 
+def _select_video_frames(
+    frames: list[tuple[float, bytes]], limit: int, action_offsets: list[float]
+) -> list[tuple[float, bytes]]:
+    """Bound sorted extracted frames without changing their timestamps or bytes."""
+    if type(limit) is not int or limit < 2:
+        raise ValueError("Video frame limit must be an integer of at least two")
+    if len(frames) <= limit:
+        return frames
+    times = [timestamp for timestamp, _ in frames]
+    selected = {0, len(frames) - 1}
+    # Favor recent actions, adding a before/after pair only when both fit.
+    # An action outside the extracted range has no evidenced pair.
+    for action in sorted(set(action_offsets), reverse=True):
+        before = bisect_left(times, action) - 1
+        after = bisect_right(times, action)
+        if before < 0 or after >= len(frames):
+            continue
+        pair = {before, after}
+        if len(selected | pair) <= limit:
+            selected.update(pair)
+    # Fill the remaining slots across time, with an earlier-frame tie break.
+    while len(selected) < limit:
+        selected.add(
+            max(
+                (index for index in range(len(frames)) if index not in selected),
+                key=lambda index: (
+                    min(abs(times[index] - times[chosen]) for chosen in selected),
+                    -times[index],
+                ),
+            )
+        )
+    return [frames[index] for index in sorted(selected)]
+
+
 async def _build_universal_user_blocks(
     analyzer,
     keyframes: list,
     actual_start: float,
     prompt_with_context: str,
     raw_path: Path,
+    *,
+    sampling_source_count: int | None = None,
 ) -> tuple[list[dict[str, Any]], dict[str, Any] | None]:
     """Builds the multimodal user content blocks (text, frames, optional audio)."""
     user_blocks: list[dict[str, Any]] = [{"type": "text", "text": prompt_with_context}]
+    if sampling_source_count is not None:
+        user_blocks.append(
+            {
+                "type": "text",
+                "text": (
+                    f"Video frame sampling: selected {len(keyframes)} of "
+                    f"{sampling_source_count} extracted frames in timestamp order. "
+                    "Extracted endpoints and fitting action-adjacent pairs are prioritized. "
+                    "This does not show every video instant or every action; "
+                    "do not infer what happened between the supplied frames."
+                ),
+            }
+        )
 
     for ts_sec, frame_bytes in keyframes:
         b64_str = base64.b64encode(frame_bytes).decode("utf-8")

artemis/agents/video_analyzer/video_analyzer.py

@@ -178,6 +178,7 @@ def __init__(self, ctx: ArtemisContext):
         self.max_dense_action_frames = int(
             getattr(self.video_config, "max_dense_action_frames", 24)
         )
+        self.max_total_frames = getattr(self.video_config, "max_total_frames", None)
         self.native_max_retries = int(getattr(self.video_config, "native_max_retries", 1))
         self.model_call_timeout_seconds = float(
             getattr(self.video_config, "model_call_timeout_seconds", 120.0)

artemis/config/agent.py

@@ -95,6 +95,17 @@ class VideoAnalyzerConfig(BaseModel):
         le=120,
         description="Maximum additional action-proximal keyframes per chunk.",
     )
+    max_total_frames: int | None = Field(
+        default=None,
+        ge=2,
+        le=150,
+        strict=True,
+        description=(
+            "Optional cap on merged frames in one universal video request. "
+            "Keeps extracted endpoints and prioritizes action-adjacent frame pairs; "
+            "None preserves the existing frame selection."
+        ),
+    )
     native_max_retries: int = Field(
         default=1,
         ge=0,

tests/unit/agents/test_video_analyzer_universal.py

@@ -14,6 +14,8 @@
 
 """Unit tests for Universal VideoAnalyzer dual-path execution engine."""
 
+import asyncio
+import base64
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
@@ -67,6 +69,117 @@ def test_universal_tool_schemas():
     assert "analyze_audio_only" in tool_names
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("frame_limit", [None, 8])
+async def test_video_frame_cap_applies_before_model_and_discloses_sampling(
+    mock_context, tmp_path, frame_limit
+):
+    agent = VideoAnalyzer(mock_context)
+    agent.max_total_frames = frame_limit
+    frames = [(float(index), bytes([index])) for index in range(53)]
+    source_bytes = b"unchanged-original-recording"
+    recording = tmp_path / "recording.mp4"
+    recording.write_bytes(source_bytes)
+    prompt = "Check only the supplied test interval. Preserve this full query."
+    response = AIMessage(content="Observed only supplied frames.")
+    with (
+        patch(
+            "artemis.agents.video_analyzer.universal_engine._gather_universal_frames",
+            return_value=frames,
+        ),
+        patch.object(agent, "_action_timestamps", return_value=[125.1, 150.1]),
+        patch(
+            "artemis.agents.video_analyzer.video_analyzer.extract_audio_from_video",
+            new=AsyncMock(side_effect=ValueError("no audio")),
+        ),
+        patch.object(
+            agent, "_invoke_universal_model", new=AsyncMock(return_value=response)
+        ) as invoke,
+        patch(
+            "artemis.agents.video_analyzer.universal_engine._persist_universal_events",
+            new=AsyncMock(),
+        ),
+    ):
+        await agent._exec_single_chunk_universal(
+            compressed_path=recording,
+            raw_path=recording,
+            start_time=100.0,
+            end_time=153.0,
+            actual_start=100.0,
+            prompt_with_context=prompt,
+            specific_query="test interval",
+        )
+    messages, tools = invoke.await_args.args[:2]
+    assert messages[0].content == agent.sub_system_prompt
+    assert messages[1].content[0] == {"type": "text", "text": prompt}
+    assert tools == [UNIVERSAL_SUBMIT_ANSWER_TOOL]
+    parts = messages[1].content
+    images = [part for part in parts if part["type"] == "image_url"]
+    assert len(images) == (8 if frame_limit else 53)
+    decoded = [base64.b64decode(part["image_url"]["url"].split(",", 1)[1]) for part in images]
+    assert decoded == sorted(decoded)
+    assert decoded[0] == frames[0][1] and decoded[-1] == frames[-1][1]
+    if frame_limit:
+        assert {frames[index][1] for index in [25, 26, 50, 51]} <= set(decoded)
+        assert "selected 8 of 53 extracted frames" in parts[1]["text"]
+        assert "does not show every video instant or every action" in parts[1]["text"]
+    else:
+        assert len(parts) == 1 + 2 * len(frames)
+    assert recording.read_bytes() == source_bytes
+    assert len(frames) == 53
+
+
+@pytest.mark.asyncio
+async def test_concurrent_video_chunks_keep_their_own_sampling_counts(mock_context, tmp_path):
+    agent = VideoAnalyzer(mock_context)
+    agent.max_total_frames = 8
+    sent = []
+
+    async def invoke(messages, *_args, **_kwargs):
+        await asyncio.sleep(0)
+        sent.append(messages[1].content)
+        return AIMessage(content="Observed only supplied frames.")
+
+    def gather(_agent, path, *_args):
+        return [(float(index), bytes([index])) for index in range(int(path.stem))]
+
+    with (
+        patch(
+            "artemis.agents.video_analyzer.universal_engine._gather_universal_frames",
+            side_effect=gather,
+        ),
+        patch.object(agent, "_action_timestamps", return_value=[]),
+        patch(
+            "artemis.agents.video_analyzer.video_analyzer.extract_audio_from_video",
+            new=AsyncMock(side_effect=ValueError("no audio")),
+        ),
+        patch.object(agent, "_invoke_universal_model", side_effect=invoke),
+        patch(
+            "artemis.agents.video_analyzer.universal_engine._persist_universal_events",
+            new=AsyncMock(),
+        ),
+    ):
+        await asyncio.gather(
+            *(
+                agent._exec_single_chunk_universal(
+                    compressed_path=tmp_path / f"{count}.mp4",
+                    raw_path=tmp_path / f"{count}.mp4",
+                    start_time=0.0,
+                    end_time=60.0,
+                    actual_start=0.0,
+                    prompt_with_context=f"Source chunk {count}",
+                    specific_query="test interval",
+                )
+                for count in [53, 51]
+            )
+        )
+    assert len(sent) == 2
+    for parts in sent:
+        count = int(parts[0]["text"].split()[-1])
+        assert f"selected 8 of {count} extracted frames" in parts[1]["text"]
+        assert len([part for part in parts if part["type"] == "image_url"]) == 8
+
+
 @pytest.mark.asyncio
 async def test_exec_single_chunk_universal(mock_context, tmp_path):
     """Test keyframe extraction and LangChain tool invocation in universal sub-agent."""

tests/unit/agents/test_video_frame_sampling.py

@@ -0,0 +1,60 @@
+"""Video request bounds without a device, model call, or loss of source media."""
+
+import pytest
+from pydantic import ValidationError
+
+from artemis.agents.video_analyzer.universal_engine import _select_video_frames
+from artemis.config.agent import VideoAnalyzerConfig
+
+
+@pytest.mark.parametrize("count", [0, 1, 2, 7, 8, 9, 51, 53, 54, 150])
+def test_selection_is_bounded_and_preserves_original_order_timestamps_and_bytes(count):
+    frames = [(index / 4, bytes([index])) for index in range(count)]
+    before = frames.copy()
+    selected = _select_video_frames(frames, 8, [1.1, 7.1, 11.1])
+    assert frames == before
+    assert len(selected) == min(count, 8)
+    assert selected == sorted(selected)
+    assert all(frame in frames for frame in selected)
+    assert len({timestamp for timestamp, _ in selected}) == len(selected)
+    if count:
+        assert selected[0] is frames[0] and selected[-1] is frames[-1]
+    if count <= 8:
+        assert selected is frames
+
+
+def test_recent_action_pairs_are_preserved_and_unrelated_ranges_do_not_consume_slots():
+    frames = [(float(index), bytes([index])) for index in range(53)]
+    actions = [-4.0, 25.1, 50.1, 90.0]
+    selected = _select_video_frames(frames, 8, actions)
+    assert {frames[index] for index in [0, 25, 26, 50, 51, 52]} <= set(selected)
+    assert selected == _select_video_frames(frames, 8, list(reversed(actions)))
+    assert selected == _select_video_frames(frames, 8, [25.1, 50.1])
+
+
+def test_tight_limit_preserves_endpoints_and_prioritizes_most_recent_action_pair():
+    frames = [(float(index), bytes([index])) for index in range(20)]
+    assert _select_video_frames(frames, 2, [4.5, 15.5]) == [frames[0], frames[-1]]
+    assert _select_video_frames(frames, 4, [4.5, 15.5]) == [
+        frames[0],
+        frames[15],
+        frames[16],
+        frames[-1],
+    ]
+
+
+def test_exact_action_timestamp_keeps_frames_on_both_sides():
+    frames = [(float(index), bytes([index])) for index in range(12)]
+    selected = _select_video_frames(frames, 4, [6.0])
+    assert selected == [frames[0], frames[5], frames[7], frames[-1]]
+
+
+@pytest.mark.parametrize("limit", [0, 1, 151, True, 8.5])
+def test_total_frame_config_rejects_invalid_limits(limit):
+    with pytest.raises(ValidationError):
+        VideoAnalyzerConfig(max_total_frames=limit)
+
+
+@pytest.mark.parametrize("limit", [None, 2, 8, 150])
+def test_total_frame_config_is_optional_and_bounded(limit):
+    assert VideoAnalyzerConfig(max_total_frames=limit).max_total_frames == limit