feat: harden video analysis and runtime resilience

artemis/agents/flash/context_compressor.py

@@ -14,9 +14,9 @@
 
 """Universal Context Compressor for Artemis Flash profile.
 
-Replaces pruned historical screenshots with high-density visual step summaries
-when ready, while seamlessly falling back to standard pruned action text if
-the background summary is still in flight.
+Replaces historical screenshots with high-density visual step summaries only
+after they are ready. Screenshots remain in context while summarization or
+retry is still in flight.
 """
 
 from typing import Any
@@ -28,6 +28,17 @@ from artemis.utils.logger import get_logger
 
 logger = get_logger(__name__)
 
+_HISTORY_SUMMARY_PREFIX = "--- Historical Visual Transition ---\n"
+_UI_LIST_MARKER = "--- UI Element List ---"
+
+
+def _without_historical_ui_list(text: str) -> str:
+    """Remove a historical UI list without discarding text before its marker."""
+    marker_index = text.find(_UI_LIST_MARKER)
+    if marker_index < 0:
+        return text
+    return text[:marker_index].rstrip()
+
 
 def compress_flash_messages(
     messages: list[BaseMessage],
@@ -40,8 +51,8 @@ def compress_flash_messages(
     1. Keeps the latest screenshot and live UI hierarchy completely intact.
     2. For earlier steps, if a VisualStepSummarizer has completed the step's summary,
        replaces the intermediate image block with the objective summary text.
-    3. If the background summary is not ready yet, gracefully falls back to the existing
-       behavior (omitting the image block and retaining the original action text).
+    3. If the background summary is not ready yet, retains the original image until
+       a successful summary can replace it.
     4. If prune_history_xml is True, removes heavy outdated UI Element lists from past turns.
     """
     if not messages:
@@ -61,7 +72,9 @@ def compress_flash_messages(
                 last_img_msg_idx = idx
                 break
 
-    # 2. Iterate through historical messages and apply compression/fallback
+    # 2. Iterate through historical messages and apply compression/fallback.
+    # ``tool_call_id`` is the stable identity of an action result. The legacy
+    # ordinal is retained only for summaries produced by older callers/tests.
     tool_step_counter = 0
     for idx in range(len(messages)):
         msg = messages[idx]
@@ -74,43 +87,77 @@ def compress_flash_messages(
             continue
 
         # Count tool steps strictly corresponding to FlashRunner action turns
-        is_tool_msg = getattr(msg, "tool_call_id", None) is not None
+        tool_call_id = getattr(msg, "tool_call_id", None)
+        is_tool_msg = tool_call_id is not None
         if is_tool_msg:
             tool_step_counter += 1
-            current_step_num = tool_step_counter
+            legacy_step_num = tool_step_counter
         else:
-            current_step_num = None
+            legacy_step_num = None
 
-        new_blocks: list[dict[str, Any]] = []
-        summary_injected = False
+        summary_text = None
+        summary_pending = False
+        if summarizer and is_tool_msg:
+            # New FlashRunner instances key summaries by tool_call_id, so generic
+            # tools, no-tool turns, and parallel tool calls cannot shift memories.
+            action_key = str(tool_call_id)
+            summary_text = summarizer.get_summary(action_key)
+            summary_pending = summarizer.is_pending(action_key)
+            if (
+                summary_text is None
+                and not summarizer.has_job(action_key)
+                and legacy_step_num is not None
+            ):
+                summary_text = summarizer.get_summary(legacy_step_num)
+                summary_pending = summarizer.is_pending(legacy_step_num)
+
+        new_blocks: list[Any] = []
 
         for b in msg_content:
             if not isinstance(b, dict):
+                # LangChain permits string parts in multimodal content lists.
+                # They are lightweight historical evidence and must not vanish.
+                new_blocks.append(b)
                 continue
             b_type = b.get("type", "")
             b_text = b.get("text", "")
 
-            # Filter heavy outdated XML hierarchy from past steps to save tokens
-            if prune_history_xml and b_type == "text" and "--- UI Element List ---" in b_text:
+            if b_type == "text" and b_text.startswith(_HISTORY_SUMMARY_PREFIX):
+                # Rebuild this generated block on every pass. This makes the
+                # compressor idempotent and lets a late summary replace itself.
+                continue
+
+            # Filter heavy outdated XML hierarchy from past steps to save tokens.
+            # Some adapters combine action text and the hierarchy in one block;
+            # retain the useful prefix instead of dropping the entire block.
+            if prune_history_xml and b_type == "text" and _UI_LIST_MARKER in b_text:
+                retained_text = _without_historical_ui_list(b_text)
+                if retained_text:
+                    retained_block = dict(b)
+                    retained_block["text"] = retained_text
+                    new_blocks.append(retained_block)
                 continue
 
             if b_type in ("image_url", "image"):
-                # Check if this is a tool execution step and background summarizer has a ready summary
-                if (
-                    current_step_num
-                    and summarizer
-                    and summarizer.has_summary(current_step_num)
-                    and not summary_injected
-                ):
-                    summary_text = summarizer.get_summary(current_step_num)
-                    if summary_text:
-                        new_blocks.append({"type": "text", "text": summary_text})
-                        summary_injected = True
-                # If summarizer is not ready or initial screen, gracefully omit the image block
+                # A submitted image remains visible until its summary is ready.
+                # This temporarily favors context size over losing visual evidence.
+                if summary_pending:
+                    new_blocks.append(b)
                 continue
 
             new_blocks.append(b)
 
+        # Do not tie injection to the presence of an image: the image is normally
+        # pruned before the background summary finishes. A stable tool-call key lets
+        # a later compression pass backfill the summary into the correct result.
+        if summary_text:
+            new_blocks.append(
+                {
+                    "type": "text",
+                    "text": f"{_HISTORY_SUMMARY_PREFIX}{summary_text}",
+                }
+            )
+
         # Defensive fallback: never leave ToolMessage or HumanMessage content empty
         if not new_blocks:
             new_blocks = [{"type": "text", "text": "Action completed." if is_tool_msg else ""}]

artemis/agents/flash/runner.py

@@ -155,6 +155,7 @@ class FlashRunner:
         ]
 
         turns = 0
+        action_sequence = 0
         final_report = None
         current_pre_screenshot_bytes = img_bytes
         current_xml_list = xml_list
@@ -186,15 +187,13 @@ class FlashRunner:
                 except Exception as e:
                     logger.warning(f"Failed to check injected instruction in FlashRunner: {e}")
 
-            # Compress previous history before LLM call to save context and inject objective summaries
-            if self.summarizer:
-                compress_flash_messages(
-                    messages,
-                    summarizer=self.summarizer,
-                    prune_history_xml=self.step_summarizer_cfg.prune_history_xml,
-                )
-            else:
-                prune_intermediate_screenshots(messages)
+            # Compress history even when visual summarization is disabled, so
+            # screenshot and historical XML pruning have identical semantics.
+            compress_flash_messages(
+                messages,
+                summarizer=self.summarizer,
+                prune_history_xml=self.step_summarizer_cfg.prune_history_xml,
+            )
 
             # Tool restriction on the final turn
             if turns == self.max_turns:
@@ -418,6 +417,7 @@ class FlashRunner:
                             )
 
                     # Record telemetry / step in DataEngine
+                    recorded_step_id = None
                     if self.ctx.data_engine and name in ACTION_TOOL_NAMES:
                         try:
                             if self.ctx.data_engine.current_step_id is None:
@@ -475,7 +475,7 @@ class FlashRunner:
                                 action_dict["normalized_start_coordinates"] = norm_start
                                 action_dict["normalized_end_coordinates"] = norm_end
 
-                            self.ctx.data_engine.record_step(
+                            recorded_step_id = self.ctx.data_engine.record_step(
                                 pre_screenshot_bytes=current_pre_screenshot_bytes,
                                 post_screenshot_bytes=post_img_bytes,
                                 ui_tree=(exec_result.ui_elements_text or current_xml_list),
@@ -489,13 +489,16 @@ class FlashRunner:
 
                     # ⚡ Non-blocking dispatch of objective visual transition summarizer
                     if self.summarizer and name in ACTION_TOOL_NAMES:
+                        action_sequence += 1
                         self.summarizer.dispatch(
-                            step_number=turns,
+                            step_number=action_sequence,
                             action_name=name,
                             action_args=args,
                             pre_img_bytes=current_pre_screenshot_bytes,
                             post_img_bytes=post_img_bytes,
                             exec_outcome=exec_result.text_summary,
+                            action_key=str(tc_id),
+                            data_engine_step_id=recorded_step_id,
                         )
 
                     if post_img_bytes:

artemis/agents/flash/summarizer.py

@@ -16,13 +16,14 @@
 
 Processes Before/After screenshot pairs and action metadata in the background
 to generate high-density, strictly objective visual state transition summaries.
-Ensures zero blocking latency for the main runner and includes out-of-order self-healing retries.
+Ensures zero blocking latency for the main runner and independently retries failed jobs.
 """
 
 import asyncio
 import base64
 from pathlib import Path
 from typing import Any
+from uuid import UUID
 
 from jinja2 import Template
 from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
@@ -40,18 +41,17 @@ class VisualStepSummarizer:
 
     Key design properties:
     1. Zero-blocking dispatch: The main Flash runner dispatches and proceeds immediately.
-    2. Graceful fallback: If a step summary is not ready, it cleanly falls back to the original action text.
-    3. Self-healing re-trigger: If a previous step's summary is stalled or lost while subsequent steps
-       complete, it automatically re-dispatches the stalled step.
+    2. Lossless pending state: The compressor retains the source image until its summary is ready.
+    3. Independent retry: Every action owns a retry loop and does not depend on later actions arriving.
     """
 
     def __init__(self, ctx: ArtemisContext, model_name: str | None = None):
         self.ctx = ctx
-        self._summaries: dict[int, str] = {}
-        self._pending_tasks: dict[int, asyncio.Task] = {}
-        self._step_inputs: dict[int, dict[str, Any]] = {}
-        self._retry_counts: dict[int, int] = {}
-        self._max_retries = 2
+        self._summaries: dict[int | str, str] = {}
+        self._pending_tasks: dict[int | str, asyncio.Task] = {}
+        self._step_inputs: dict[int | str, dict[str, Any]] = {}
+        self._retry_counts: dict[int | str, int] = {}
+        self._retry_delays = (0.0, 0.5, 1.0, 2.0, 3.0)
 
         # Initialize lightweight VLM: prioritize explicit model_name
         target_model = model_name or "gemini-2.5-flash-lite"
@@ -83,44 +83,50 @@ class VisualStepSummarizer:
         pre_img_bytes: bytes | None,
         post_img_bytes: bytes | None,
         exec_outcome: str,
+        *,
+        action_key: str | None = None,
+        data_engine_step_id: UUID | str | None = None,
     ) -> None:
         """Dispatches an asynchronous summarization task without blocking the caller."""
-        self._step_inputs[step_number] = {
+        key: int | str = action_key or step_number
+        self._step_inputs[key] = {
+            "step_number": step_number,
             "action_name": action_name,
             "action_args": action_args,
             "pre_img_bytes": pre_img_bytes,
             "post_img_bytes": post_img_bytes,
             "exec_outcome": exec_outcome,
+            "data_engine_step_id": data_engine_step_id,
         }
 
-        task = asyncio.create_task(self._run_summary(step_number))
-        self._pending_tasks[step_number] = task
+        task = asyncio.create_task(self._run_summary_until_ready(key))
+        self._pending_tasks[key] = task
 
-        # Check for and re-trigger any previous stalled steps
-        self._check_and_retrigger_stalled_steps(current_step=step_number)
+    async def _run_summary_until_ready(self, action_key: int | str) -> None:
+        """Retry one action independently until a non-empty summary is committed."""
+        while action_key not in self._summaries:
+            if await self._run_summary_once(action_key):
+                return
 
-    def _check_and_retrigger_stalled_steps(self, current_step: int) -> None:
-        """Re-dispatches any previous steps that remain unsummarized while later steps arrive."""
-        for past_step in range(1, current_step):
-            if past_step not in self._summaries and past_step in self._step_inputs:
-                task = self._pending_tasks.get(past_step)
-                retries = self._retry_counts.get(past_step, 0)
-                if (task is None or task.done()) and retries < self._max_retries:
-                    self._retry_counts[past_step] = retries + 1
-                    logger.info(
-                        f"VisualStepSummarizer: Retriggering stalled summary for Step {past_step} "
-                        f"(attempt {self._retry_counts[past_step] + 1})"
-                    )
-                    self._pending_tasks[past_step] = asyncio.create_task(
-                        self._run_summary(past_step)
-                    )
+            retry_count = self._retry_counts.get(action_key, 0) + 1
+            self._retry_counts[action_key] = retry_count
+            delay = self._retry_delays[min(retry_count - 1, len(self._retry_delays) - 1)]
+            input_data = self._step_inputs.get(action_key)
+            display_step = input_data.get("step_number") if input_data else action_key
+            logger.info(
+                f"VisualStepSummarizer: Retrying Step {display_step} "
+                f"(attempt {retry_count + 1}) after {delay:.1f}s"
+            )
+            if delay:
+                await asyncio.sleep(delay)
 
-    async def _run_summary(self, step_number: int) -> None:
-        """Executes the lightweight VLM call to summarize the step transition."""
-        input_data = self._step_inputs.get(step_number)
+    async def _run_summary_once(self, action_key: int | str) -> bool:
+        """Execute one lightweight VLM attempt for a step transition."""
+        input_data = self._step_inputs.get(action_key)
         if not input_data:
-            return
+            return False
 
+        step_number = input_data["step_number"]
         action_name = input_data["action_name"]
         action_args = input_data["action_args"]
         pre_bytes = input_data["pre_img_bytes"]
@@ -183,44 +189,60 @@ class VisualStepSummarizer:
 
             summary_text = summary_raw.strip()
             if summary_text:
-                self._summaries[step_number] = summary_text
+                self._summaries[action_key] = summary_text
                 logger.info(
                     f"VisualStepSummarizer: Generated summary for Step {step_number}: {summary_text[:80]}..."
                 )
 
                 # Free binary image buffers from memory once summary is secured
-                if step_number in self._step_inputs:
-                    self._step_inputs[step_number]["pre_img_bytes"] = None
-                    self._step_inputs[step_number]["post_img_bytes"] = None
+                if action_key in self._step_inputs:
+                    self._step_inputs[action_key]["pre_img_bytes"] = None
+                    self._step_inputs[action_key]["post_img_bytes"] = None
 
                 # Update DataEngine telemetry if active
                 if self.ctx.data_engine:
                     try:
-                        self.ctx.data_engine.update_step_summary(step_number, summary_text)
+                        target_step = input_data.get("data_engine_step_id") or step_number
+                        self.ctx.data_engine.update_step_summary(target_step, summary_text)
                     except Exception as de_err:
                         logger.debug(f"DataEngine step summary update skipped: {de_err}")
+                return True
 
         except Exception as e:
             logger.warning(
                 f"VisualStepSummarizer: Error generating summary for step {step_number}: {e}"
             )
-            if self._retry_counts.get(step_number, 0) >= self._max_retries:
-                # Max retries reached, free memory buffer
-                if step_number in self._step_inputs:
-                    self._step_inputs[step_number]["pre_img_bytes"] = None
-                    self._step_inputs[step_number]["post_img_bytes"] = None
+        return False
 
-    def get_summary(self, step_number: int, fallback_text: str | None = None) -> str | None:
-        """Retrieves summarized state for step_number, or fallback_text if not ready yet."""
-        return self._summaries.get(step_number, fallback_text)
+    def get_summary(
+        self, action_key: int | str, fallback_text: str | None = None
+    ) -> str | None:
+        """Retrieve the summary for a stable action key, or fallback text."""
+        return self._summaries.get(action_key, fallback_text)
 
-    def has_summary(self, step_number: int) -> bool:
-        """Checks if a step summary has completed."""
-        return step_number in self._summaries
+    def has_summary(self, action_key: int | str) -> bool:
+        """Check whether an action summary has completed."""
+        return action_key in self._summaries
 
-    async def flush(self) -> None:
-        """Flushes and awaits all pending background summary tasks upon completion."""
+    def has_job(self, action_key: int | str) -> bool:
+        """Check whether an action has been submitted for visual summarization."""
+        return action_key in self._step_inputs
+
+    def is_pending(self, action_key: int | str) -> bool:
+        """Check whether an action still needs a summary."""
+        return self.has_job(action_key) and not self.has_summary(action_key)
+
+    async def flush(self, timeout_seconds: float = 30.0) -> None:
+        """Wait for active jobs up to a bound, then cancel remaining retry loops."""
         tasks = [t for t in self._pending_tasks.values() if not t.done()]
         if tasks:
             logger.info(f"VisualStepSummarizer: Flushing {len(tasks)} pending summary tasks...")
-            await asyncio.gather(*tasks, return_exceptions=True)
+            _, pending = await asyncio.wait(tasks, timeout=timeout_seconds)
+            if pending:
+                logger.warning(
+                    f"VisualStepSummarizer: Cancelling {len(pending)} unfinished summary tasks "
+                    f"after {timeout_seconds:.1f}s flush timeout."
+                )
+                for task in pending:
+                    task.cancel()
+                await asyncio.gather(*pending, return_exceptions=True)

artemis/agents/operator/prompts.py

@@ -58,6 +58,17 @@ _LEGACY_TROUBLESHOOTER_CHALLENGE = """   - *Challenge Checker*: If you visually
 _TROUBLESHOOTER_CHALLENGE = """   - *Challenge Checker*: If current evidence clearly proves the subgoal is already achieved, invoke `reply_to_checker` directly. This branch takes precedence over diagnosis; do not call `ask_diagnoser` first.
    - *Clear Local Correction*: If the rejection cause is explicit and a safe correction is evident from the current screen, perform that correction directly. This branch also takes precedence over generic diagnosis triggers. Use `ask_diagnoser` only for ambiguity, conflicting evidence, or repeated failure."""
 
+_LEGACY_LARGE_LIST_STRATEGY = """- **Large List & Long Text Exploration Strategy (Zero-Miss, Zero-Duplication Single-Pass Principle)**: When traversing large lists, browsing long text bodies, or scanning continuous content feeds, strictly adhere to the single-pass exploration standard:
+  1. **Unidirectional Anchor-Based Traversal (Zero-Miss)**: Maintain a consistent, unidirectional scan (e.g., uniform downward scrolling). Before each scroll, identify the bottom-most visible item/text segment as your visual anchor. Calibrate your scroll distance so that this anchor remains visible near the top of the subsequent screen, creating a seamless visual overlap that guarantees zero missed items. Terminate exploration when a definitive boundary is reached (e.g., list bottom reached or no new content appears after scrolling).
+  2. **Memory & Plan-Driven Deduplication (Zero-Duplication)**: Systematically log extracted item names, identifiers, or processing states into notes or memory as you progress. After every screen transition, cross-reference visible items against your recorded ledger and strictly interact only with newly surfaced, unrecorded items to eliminate redundant actions on overlapping elements. Proactively maintain subtasks in `task_plan` (e.g., segmenting by batch, category, or alphabetical ranges like `a-d`), tracking exact paths and progress to prevent backtracking.
+  3. **Single-Pass Trust & Anti-Oscillation (No Over-Verification)**: Treat recorded notes as verified ground truth. Once a segment is traversed and logged, consider it permanently resolved. Do NOT scroll back-and-forth or perform redundant re-scans for "double-checking". Only perform incremental re-verification if the list has been refreshed or content has mutated, while maintaining strict single-pass handling of new events."""
+
+_LARGE_LIST_STRATEGY = """- **Large List & Long Text Exploration Strategy (Zero-Miss, Zero-Duplication Single-Pass Principle)**: When traversing large lists, browsing long text bodies, or scanning continuous content feeds, strictly adhere to the single-pass exploration standard:
+  1. **Unidirectional Anchor-Based Traversal (Zero-Miss)**: Maintain a consistent, unidirectional scan (e.g., uniform downward scrolling). Before each scroll, identify the bottom-most visible item/text segment as your visual anchor and carry that anchor plus the scan direction in short-term memory or the traversal ledger. After the scroll, evaluate this handoff before doing anything unrelated: if the anchor remains visible with new content beyond it, record the new content and continue in the same direction without scrolling back to verify; if the anchor is missing, continuity is uncertain, so perform at most one minimal reverse recovery to re-establish overlap, then resume the original direction. While the traversal's declared exit condition remains unresolved, keep this traversal active and do not advance to another milestone or mark it complete.
+  2. **Boundary-Proven Completion**: A single scroll that reveals no new content is not sufficient evidence that the list is complete. For an exhaustive traversal, stop only when an explicit end-of-list indicator is visible, or when one additional successful swipe in the same direction and within the same list container leaves the same final anchor visible with no new items and no loading state. This boundary probe must remain unidirectional; never reverse direction merely to prove completion. If the task declares an earlier exit condition (for example, find one verified match or process N items), that explicit condition may end the traversal without reaching the list boundary.
+  3. **Memory & Plan-Driven Deduplication (Zero-Duplication)**: Systematically log extracted item names, identifiers, or processing states into notes or memory as you progress. After every screen transition, cross-reference visible items against your recorded ledger and strictly interact only with newly surfaced, unrecorded items to eliminate redundant actions on overlapping elements. Proactively maintain subtasks in `task_plan` (e.g., segmenting by batch, category, or alphabetical ranges like `a-d`), tracking exact paths and progress to prevent backtracking.
+  4. **Single-Pass Trust & Anti-Oscillation (No Over-Verification)**: Treat recorded notes as verified ground truth. Once a segment is traversed with confirmed anchor continuity and logged, consider it permanently resolved. Do NOT scroll back-and-forth or perform redundant re-scans for "double-checking", even if older execution history has been pruned. Resume from the last recorded anchor and direction. Only the single bounded recovery above, or a refreshed/mutated list, permits limited re-verification; otherwise continue the single pass over new content."""
+
 
 def apply_operator_prompt_contract(prompt_template: str) -> str:
     """Align prompt-level tool and recovery rules with Operator runtime semantics."""
@@ -77,6 +88,10 @@ def apply_operator_prompt_contract(prompt_template: str) -> str:
         _LEGACY_TROUBLESHOOTER_CHALLENGE,
         _TROUBLESHOOTER_CHALLENGE,
     )
+    prompt_template = prompt_template.replace(
+        _LEGACY_LARGE_LIST_STRATEGY,
+        _LARGE_LIST_STRATEGY,
+    )
     return prompt_template
 
 

artemis/agents/video_analyzer/blackboard.py

@@ -0,0 +1,821 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Persistent, video-scoped memory for video analysis agents."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from dataclasses import dataclass
+import hashlib
+import json
+from pathlib import Path
+import shutil
+import sqlite3
+import threading
+import time
+from typing import Any, Iterator, Literal
+from uuid import uuid4
+
+from artemis.context import ArtemisContext
+from artemis.utils.logger import get_logger
+from artemis.utils.video import get_active_session
+
+logger = get_logger(__name__)
+
+
+ClaimState = Literal["claimed", "cached", "in_progress"]
+
+
+@dataclass(frozen=True)
+class SegmentClaim:
+    """Result of atomically attempting to acquire one analysis segment."""
+
+    state: ClaimState
+    lease_owner: str | None = None
+    summary: str | None = None
+    analysis: str | None = None
+
+
+def _normalise_query(query: str | None) -> str:
+    return " ".join(str(query or "").strip().lower().split())
+
+
+def _query_key(query: str | None) -> str:
+    return hashlib.sha256(_normalise_query(query).encode("utf-8")).hexdigest()
+
+
+def _to_ms(value: float | int) -> int:
+    return int(round(float(value) * 1000.0))
+
+
+def _to_seconds(value: int) -> float:
+    return round(value / 1000.0, 3)
+
+
+class VideoBlackboard:
+    """SQLite-backed blackboard with a context-local in-memory fallback."""
+
+    def __init__(
+        self,
+        board_key: str,
+        *,
+        db_path: Path | None = None,
+        evidence_dir: Path | None = None,
+        lease_seconds: float = 300.0,
+    ) -> None:
+        self.board_key = board_key
+        self.db_path = Path(db_path) if db_path is not None else None
+        self.evidence_dir = Path(evidence_dir) if evidence_dir is not None else None
+        self.lease_seconds = max(0.01, float(lease_seconds))
+        self._lock = threading.RLock()
+        self._memory_observations: dict[str, dict[str, Any]] = {}
+        self._memory_segments: dict[tuple[str, str, int, int], dict[str, Any]] = {}
+
+        if self.db_path is not None:
+            self.db_path.parent.mkdir(parents=True, exist_ok=True)
+            self._init_db()
+        if self.evidence_dir is not None:
+            self.evidence_dir.mkdir(parents=True, exist_ok=True)
+
+    @contextmanager
+    def _connection(self) -> Iterator[sqlite3.Connection]:
+        if self.db_path is None:
+            raise RuntimeError("SQLite connection requested for in-memory blackboard")
+        with self._lock:
+            conn = sqlite3.connect(self.db_path, timeout=30.0)
+            conn.row_factory = sqlite3.Row
+            try:
+                yield conn
+            finally:
+                conn.close()
+
+    def _init_db(self) -> None:
+        with self._connection() as conn:
+            conn.execute("PRAGMA journal_mode=WAL")
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS video_analysis_segments (
+                    board_key TEXT NOT NULL,
+                    query_key TEXT NOT NULL,
+                    query_text TEXT NOT NULL,
+                    modality TEXT NOT NULL,
+                    start_ms INTEGER NOT NULL,
+                    end_ms INTEGER NOT NULL,
+                    status TEXT NOT NULL,
+                    lease_owner TEXT,
+                    lease_expires_at REAL,
+                    attempt_count INTEGER NOT NULL DEFAULT 0,
+                    summary TEXT,
+                    analysis TEXT,
+                    error TEXT,
+                    error_category TEXT,
+                    model_name TEXT,
+                    source_generation INTEGER,
+                    created_at REAL NOT NULL,
+                    updated_at REAL NOT NULL,
+                    PRIMARY KEY (board_key, query_key, modality, start_ms, end_ms)
+                )
+                """
+            )
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS video_analysis_observations (
+                    observation_id TEXT PRIMARY KEY,
+                    board_key TEXT NOT NULL,
+                    start_ms INTEGER NOT NULL,
+                    end_ms INTEGER,
+                    query_text TEXT NOT NULL,
+                    summary TEXT NOT NULL,
+                    confidence REAL,
+                    modality TEXT NOT NULL,
+                    screenshot_path TEXT,
+                    extra_json TEXT,
+                    created_at REAL NOT NULL
+                )
+                """
+            )
+            columns = {
+                row["name"]
+                for row in conn.execute("PRAGMA table_info(video_analysis_segments)").fetchall()
+            }
+            for name, sql_type in (
+                ("error_category", "TEXT"),
+                ("model_name", "TEXT"),
+                ("source_generation", "INTEGER"),
+            ):
+                if name not in columns:
+                    try:
+                        conn.execute(
+                            f"ALTER TABLE video_analysis_segments ADD COLUMN {name} {sql_type}"
+                        )
+                    except sqlite3.OperationalError as error:
+                        if "duplicate column" not in str(error).lower():
+                            raise
+            conn.execute(
+                """
+                CREATE INDEX IF NOT EXISTS idx_video_analysis_segments_lookup
+                ON video_analysis_segments (board_key, query_key, modality, status, start_ms, end_ms)
+                """
+            )
+            conn.execute(
+                """
+                CREATE INDEX IF NOT EXISTS idx_video_analysis_observations_board
+                ON video_analysis_observations (board_key, start_ms, end_ms)
+                """
+            )
+            conn.commit()
+
+    def _persist_evidence(self, entry: dict[str, Any], observation_id: str) -> str | None:
+        candidate = entry.get("screenshot") or entry.get("screenshot_path")
+        if not candidate:
+            return None
+        source = Path(str(candidate))
+        if not source.exists() or not source.is_file() or self.evidence_dir is None:
+            return str(source) if source.exists() else None
+        try:
+            if source.stat().st_size <= 0:
+                return None
+        except OSError:
+            return None
+
+        suffix = source.suffix.lower() or ".jpg"
+        destination = self.evidence_dir / f"{observation_id}{suffix}"
+        if source.resolve() != destination.resolve() and not destination.exists():
+            try:
+                shutil.copy2(source, destination)
+            except OSError as exc:
+                logger.warning(f"Failed to persist video evidence {source}: {exc}")
+                return str(source)
+        return str(destination)
+
+    def add_observation(self, entry: dict[str, Any], *, modality: str = "video") -> dict[str, Any]:
+        """Idempotently persist one evidence-backed observation."""
+
+        start = float(entry.get("start", 0.0))
+        raw_end = entry.get("end")
+        end = None if raw_end in (None, "unknown") else float(raw_end)
+        query_text = str(entry.get("target", ""))
+        summary = str(entry.get("summary", "")).strip()
+        identity = json.dumps(
+            {
+                "board_key": self.board_key,
+                "start_ms": _to_ms(start),
+                "end_ms": _to_ms(end) if end is not None else None,
+                "query": _normalise_query(query_text),
+                "summary": summary,
+                "modality": modality,
+            },
+            sort_keys=True,
+            separators=(",", ":"),
+        )
+        observation_id = hashlib.sha256(identity.encode("utf-8")).hexdigest()
+        screenshot_path = self._persist_evidence(entry, observation_id)
+
+        stored = dict(entry)
+        if screenshot_path:
+            stored["screenshot"] = screenshot_path
+        stored["modality"] = modality
+        stored["observation_id"] = observation_id
+
+        excluded = {
+            "start",
+            "end",
+            "target",
+            "summary",
+            "confidence_score",
+            "screenshot",
+            "screenshot_path",
+            "modality",
+            "observation_id",
+        }
+        extra = {key: value for key, value in stored.items() if key not in excluded}
+        now = time.time()
+
+        if self.db_path is None:
+            with self._lock:
+                self._memory_observations.setdefault(observation_id, stored)
+            return dict(self._memory_observations[observation_id])
+
+        with self._connection() as conn:
+            conn.execute(
+                """
+                INSERT OR IGNORE INTO video_analysis_observations (
+                    observation_id, board_key, start_ms, end_ms, query_text,
+                    summary, confidence, modality, screenshot_path, extra_json, created_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    observation_id,
+                    self.board_key,
+                    _to_ms(start),
+                    _to_ms(end) if end is not None else None,
+                    query_text,
+                    summary,
+                    float(entry.get("confidence_score", 0.0)),
+                    modality,
+                    screenshot_path,
+                    json.dumps(extra, default=str),
+                    now,
+                ),
+            )
+            conn.commit()
+        return stored
+
+    def list_observations(self) -> list[dict[str, Any]]:
+        """Return all observations for this recording in timeline order."""
+
+        if self.db_path is None:
+            with self._lock:
+                entries = [dict(value) for value in self._memory_observations.values()]
+            return sorted(entries, key=lambda value: float(value.get("start", 0.0)))
+
+        with self._connection() as conn:
+            rows = conn.execute(
+                """
+                SELECT * FROM video_analysis_observations
+                WHERE board_key = ?
+                ORDER BY start_ms, created_at
+                """,
+                (self.board_key,),
+            ).fetchall()
+
+        entries: list[dict[str, Any]] = []
+        for row in rows:
+            entry: dict[str, Any] = {
+                "start": _to_seconds(row["start_ms"]),
+                "end": _to_seconds(row["end_ms"]) if row["end_ms"] is not None else "unknown",
+                "target": row["query_text"],
+                "summary": row["summary"],
+                "confidence_score": row["confidence"],
+                "modality": row["modality"],
+                "observation_id": row["observation_id"],
+            }
+            if row["screenshot_path"]:
+                entry["screenshot"] = row["screenshot_path"]
+            if row["extra_json"]:
+                try:
+                    entry.update(json.loads(row["extra_json"]))
+                except (TypeError, json.JSONDecodeError):
+                    pass
+            entries.append(entry)
+        return entries
+
+    def list_ledger_entries(self) -> list[dict[str, Any]]:
+        """Return observations plus successful query-specific segment summaries."""
+
+        entries = self.list_observations()
+        if self.db_path is None:
+            with self._lock:
+                for (query_key, modality, start_ms, end_ms), value in self._memory_segments.items():
+                    if value.get("status") != "succeeded":
+                        continue
+                    entries.append(
+                        {
+                            "start": _to_seconds(start_ms),
+                            "end": _to_seconds(end_ms),
+                            "target": value.get("query_text", ""),
+                            "summary": value.get("summary") or "No summary provided.",
+                            "confidence_score": 0.5,
+                            "modality": modality,
+                            "kind": "segment_result",
+                            "query_key": query_key,
+                        }
+                    )
+        else:
+            with self._connection() as conn:
+                rows = conn.execute(
+                    """
+                    SELECT query_key, query_text, modality, start_ms, end_ms, summary
+                    FROM video_analysis_segments
+                    WHERE board_key = ? AND status = 'succeeded'
+                    ORDER BY start_ms, updated_at
+                    """,
+                    (self.board_key,),
+                ).fetchall()
+            entries.extend(
+                {
+                    "start": _to_seconds(row["start_ms"]),
+                    "end": _to_seconds(row["end_ms"]),
+                    "target": row["query_text"],
+                    "summary": row["summary"] or "No summary provided.",
+                    "confidence_score": 0.5,
+                    "modality": row["modality"],
+                    "kind": "segment_result",
+                    "query_key": row["query_key"],
+                }
+                for row in rows
+            )
+        return sorted(entries, key=lambda value: float(value.get("start", 0.0)))
+
+    def claim_segment(
+        self,
+        start: float,
+        end: float,
+        query: str,
+        *,
+        modality: str = "video",
+        model_name: str | None = None,
+        source_generation: int | None = None,
+    ) -> SegmentClaim:
+        """Atomically acquire work, or return a cached/in-progress result."""
+
+        start_ms = _to_ms(start)
+        end_ms = _to_ms(end)
+        if end_ms <= start_ms:
+            raise ValueError("Video analysis segment end must be greater than start")
+        q_key = _query_key(query)
+        key = (q_key, modality, start_ms, end_ms)
+        now = time.time()
+        owner = str(uuid4())
+        expires = now + self.lease_seconds
+
+        if self.db_path is None:
+            with self._lock:
+                existing = self._memory_segments.get(key)
+                if existing and existing["status"] == "succeeded":
+                    return SegmentClaim(
+                        "cached", summary=existing.get("summary"), analysis=existing.get("analysis")
+                    )
+                if (
+                    existing
+                    and existing["status"] == "running"
+                    and float(existing.get("lease_expires_at") or 0.0) > now
+                ):
+                    return SegmentClaim("in_progress")
+                attempts = int(existing.get("attempt_count", 0)) + 1 if existing else 1
+                self._memory_segments[key] = {
+                    "query_text": query,
+                    "status": "running",
+                    "lease_owner": owner,
+                    "lease_expires_at": expires,
+                    "attempt_count": attempts,
+                    "model_name": model_name,
+                    "source_generation": source_generation,
+                    "start_ms": start_ms,
+                    "end_ms": end_ms,
+                    "updated_at": now,
+                }
+            return SegmentClaim("claimed", lease_owner=owner)
+
+        with self._connection() as conn:
+            conn.execute("BEGIN IMMEDIATE")
+            row = conn.execute(
+                """
+                SELECT * FROM video_analysis_segments
+                WHERE board_key = ? AND query_key = ? AND modality = ?
+                  AND start_ms = ? AND end_ms = ?
+                """,
+                (self.board_key, q_key, modality, start_ms, end_ms),
+            ).fetchone()
+            if row is not None and row["status"] == "succeeded":
+                conn.commit()
+                return SegmentClaim("cached", summary=row["summary"], analysis=row["analysis"])
+            if (
+                row is not None
+                and row["status"] == "running"
+                and float(row["lease_expires_at"] or 0.0) > now
+            ):
+                conn.commit()
+                return SegmentClaim("in_progress")
+
+            if row is None:
+                conn.execute(
+                    """
+                    INSERT INTO video_analysis_segments (
+                        board_key, query_key, query_text, modality, start_ms, end_ms,
+                        status, lease_owner, lease_expires_at, attempt_count,
+                        model_name, source_generation, created_at, updated_at
+                    ) VALUES (?, ?, ?, ?, ?, ?, 'running', ?, ?, 1, ?, ?, ?, ?)
+                    """,
+                    (
+                        self.board_key,
+                        q_key,
+                        query,
+                        modality,
+                        start_ms,
+                        end_ms,
+                        owner,
+                        expires,
+                        model_name,
+                        source_generation,
+                        now,
+                        now,
+                    ),
+                )
+            else:
+                conn.execute(
+                    """
+                    UPDATE video_analysis_segments
+                    SET status = 'running', lease_owner = ?, lease_expires_at = ?,
+                        attempt_count = attempt_count + 1, error = NULL,
+                        error_category = NULL, model_name = ?, source_generation = ?,
+                        updated_at = ?
+                    WHERE board_key = ? AND query_key = ? AND modality = ?
+                      AND start_ms = ? AND end_ms = ?
+                    """,
+                    (
+                        owner,
+                        expires,
+                        model_name,
+                        source_generation,
+                        now,
+                        self.board_key,
+                        q_key,
+                        modality,
+                        start_ms,
+                        end_ms,
+                    ),
+                )
+            conn.commit()
+        return SegmentClaim("claimed", lease_owner=owner)
+
+    def complete_segment(
+        self,
+        start: float,
+        end: float,
+        query: str,
+        lease_owner: str,
+        summary: str,
+        analysis: str | None = None,
+        *,
+        modality: str = "video",
+    ) -> None:
+        """Commit a complete result while preserving lease ownership."""
+
+        key = (_query_key(query), modality, _to_ms(start), _to_ms(end))
+        now = time.time()
+        if self.db_path is None:
+            with self._lock:
+                row = self._memory_segments.get(key)
+                if row is None or row.get("lease_owner") != lease_owner:
+                    return
+                row.update(
+                    {
+                        "status": "succeeded",
+                        "summary": summary,
+                        "analysis": analysis,
+                        "lease_expires_at": None,
+                        "updated_at": now,
+                    }
+                )
+            return
+
+        with self._connection() as conn:
+            conn.execute(
+                """
+                UPDATE video_analysis_segments
+                SET status = 'succeeded', summary = ?, analysis = ?, error = NULL,
+                    lease_expires_at = NULL, updated_at = ?
+                WHERE board_key = ? AND query_key = ? AND modality = ?
+                  AND start_ms = ? AND end_ms = ? AND lease_owner = ?
+                """,
+                (
+                    summary,
+                    analysis,
+                    now,
+                    self.board_key,
+                    key[0],
+                    modality,
+                    key[2],
+                    key[3],
+                    lease_owner,
+                ),
+            )
+            conn.commit()
+
+    def fail_segment(
+        self,
+        start: float,
+        end: float,
+        query: str,
+        lease_owner: str,
+        error: str,
+        *,
+        modality: str = "video",
+        retryable: bool = True,
+        error_category: str | None = None,
+    ) -> None:
+        """Persist a failed attempt without deleting prior observations."""
+
+        key = (_query_key(query), modality, _to_ms(start), _to_ms(end))
+        status = "retryable_failed" if retryable else "permanent_failed"
+        now = time.time()
+        if self.db_path is None:
+            with self._lock:
+                row = self._memory_segments.get(key)
+                if row is None or row.get("lease_owner") != lease_owner:
+                    return
+                row.update(
+                    {
+                        "status": status,
+                        "error": str(error),
+                        "error_category": error_category,
+                        "lease_expires_at": None,
+                        "updated_at": now,
+                    }
+                )
+            return
+
+        with self._connection() as conn:
+            conn.execute(
+                """
+                UPDATE video_analysis_segments
+                SET status = ?, error = ?, error_category = ?,
+                    lease_expires_at = NULL, updated_at = ?
+                WHERE board_key = ? AND query_key = ? AND modality = ?
+                  AND start_ms = ? AND end_ms = ? AND lease_owner = ?
+                """,
+                (
+                    status,
+                    str(error),
+                    error_category,
+                    now,
+                    self.board_key,
+                    key[0],
+                    modality,
+                    key[2],
+                    key[3],
+                    lease_owner,
+                ),
+            )
+            conn.commit()
+
+    def metrics(self) -> dict[str, Any]:
+        """Return a compact operational snapshot for tracing and diagnostics."""
+
+        if self.db_path is None:
+            with self._lock:
+                rows = [dict(value) for value in self._memory_segments.values()]
+                observation_count = len(self._memory_observations)
+        else:
+            with self._connection() as conn:
+                rows = [
+                    dict(row)
+                    for row in conn.execute(
+                        """
+                        SELECT status, attempt_count, start_ms, end_ms, error_category
+                        FROM video_analysis_segments WHERE board_key = ?
+                        """,
+                        (self.board_key,),
+                    ).fetchall()
+                ]
+                observation_count = int(
+                    conn.execute(
+                        "SELECT COUNT(*) FROM video_analysis_observations WHERE board_key = ?",
+                        (self.board_key,),
+                    ).fetchone()[0]
+                )
+
+        statuses: dict[str, int] = {}
+        failure_categories: dict[str, int] = {}
+        successful_seconds = 0.0
+        attempts = 0
+        for row in rows:
+            status = str(row.get("status", "unknown"))
+            statuses[status] = statuses.get(status, 0) + 1
+            attempts += int(row.get("attempt_count", 0) or 0)
+            category = row.get("error_category")
+            if category:
+                failure_categories[str(category)] = failure_categories.get(str(category), 0) + 1
+            if status == "succeeded" and row.get("start_ms") is not None:
+                successful_seconds += max(
+                    0.0,
+                    (float(row.get("end_ms", 0)) - float(row.get("start_ms", 0))) / 1000.0,
+                )
+        return {
+            "board_key": self.board_key,
+            "segments": statuses,
+            "attempts": attempts,
+            "failure_categories": failure_categories,
+            "successful_seconds": round(successful_seconds, 3),
+            "observations": observation_count,
+        }
+
+    def successful_segments(
+        self, start: float, end: float, query: str, *, modality: str = "video"
+    ) -> list[dict[str, Any]]:
+        """Return successful segments overlapping a requested interval."""
+
+        start_ms, end_ms = _to_ms(start), _to_ms(end)
+        q_key = _query_key(query)
+        if self.db_path is None:
+            with self._lock:
+                rows = []
+                for (
+                    row_query,
+                    row_modality,
+                    row_start,
+                    row_end,
+                ), value in self._memory_segments.items():
+                    if (
+                        row_query == q_key
+                        and row_modality == modality
+                        and value.get("status") == "succeeded"
+                        and row_end > start_ms
+                        and row_start < end_ms
+                    ):
+                        rows.append(
+                            {
+                                **value,
+                                "start": _to_seconds(row_start),
+                                "end": _to_seconds(row_end),
+                            }
+                        )
+            return sorted(rows, key=lambda value: value["start"])
+
+        with self._connection() as conn:
+            db_rows = conn.execute(
+                """
+                SELECT start_ms, end_ms, summary, analysis, status
+                FROM video_analysis_segments
+                WHERE board_key = ? AND query_key = ? AND modality = ?
+                  AND status = 'succeeded' AND end_ms > ? AND start_ms < ?
+                ORDER BY start_ms
+                """,
+                (self.board_key, q_key, modality, start_ms, end_ms),
+            ).fetchall()
+        return [
+            {
+                "start": _to_seconds(row["start_ms"]),
+                "end": _to_seconds(row["end_ms"]),
+                "summary": row["summary"],
+                "analysis": row["analysis"],
+                "status": row["status"],
+            }
+            for row in db_rows
+        ]
+
+    def missing_intervals(
+        self, start: float, end: float, query: str, *, modality: str = "video"
+    ) -> list[tuple[float, float]]:
+        """Subtract successful exact-query coverage from a requested interval."""
+
+        start_ms, end_ms = _to_ms(start), _to_ms(end)
+        if end_ms <= start_ms:
+            return []
+        covered = []
+        for row in self.successful_segments(start, end, query, modality=modality):
+            covered.append((max(start_ms, _to_ms(row["start"])), min(end_ms, _to_ms(row["end"]))))
+        covered = sorted((left, right) for left, right in covered if right > left)
+
+        merged: list[tuple[int, int]] = []
+        for left, right in covered:
+            if not merged or left > merged[-1][1]:
+                merged.append((left, right))
+            else:
+                merged[-1] = (merged[-1][0], max(merged[-1][1], right))
+
+        missing: list[tuple[float, float]] = []
+        cursor = start_ms
+        for left, right in merged:
+            if left > cursor:
+                missing.append((_to_seconds(cursor), _to_seconds(left)))
+            cursor = max(cursor, right)
+        if cursor < end_ms:
+            missing.append((_to_seconds(cursor), _to_seconds(end_ms)))
+        return missing
+
+    def format_cached_segments(
+        self, start: float, end: float, query: str, *, modality: str = "video"
+    ) -> list[str]:
+        """Format cached segment results in the child-agent response shape."""
+
+        results = []
+        for row in self.successful_segments(start, end, query, modality=modality):
+            text = (
+                f"[from {row['start']:.1f}s to {row['end']:.1f}s] "
+                f"Summary: {row.get('summary') or 'No summary provided.'}"
+            )
+            if row.get("analysis"):
+                text += f" Analysis: {row['analysis']}"
+            results.append(text)
+        return results
+
+
+def _safe_path(value: Any) -> Path | None:
+    if isinstance(value, Path):
+        return value
+    if isinstance(value, str) and value:
+        return Path(value)
+    return None
+
+
+def _latest_recorded_video_key(db_path: Path | None, session_id: Any) -> str | None:
+    """Resolve a stopped recording so a restarted analyzer keeps its memory key."""
+    if db_path is None or not session_id or not db_path.exists():
+        return None
+    try:
+        conn = sqlite3.connect(db_path, timeout=5.0)
+        try:
+            row = conn.execute(
+                """
+                SELECT video_id FROM video_recordings
+                WHERE session_id = ?
+                ORDER BY start_time DESC LIMIT 1
+                """,
+                (str(session_id),),
+            ).fetchone()
+        finally:
+            conn.close()
+    except sqlite3.Error:
+        return None
+    return f"video:{row[0]}" if row and row[0] else None
+
+
+def get_video_blackboard(ctx: ArtemisContext) -> VideoBlackboard:
+    """Return the blackboard shared by every analyzer in an Artemis context."""
+
+    existing = getattr(ctx, "_video_blackboard", None)
+    if isinstance(existing, VideoBlackboard):
+        return existing
+
+    device_id = str(getattr(getattr(ctx, "device", None), "device_id", "default"))
+    data_engine = getattr(ctx, "data_engine", None)
+    db_path = None
+    base_dir = None
+    storage = getattr(data_engine, "storage", None)
+    if storage is not None:
+        db_path = _safe_path(getattr(storage, "db_path", None))
+    if data_engine is not None:
+        base_dir = _safe_path(getattr(data_engine, "base_dir", None))
+
+    if base_dir is None:
+        execution_setup = getattr(ctx, "execution_setup", None)
+        traces_path = _safe_path(getattr(execution_setup, "traces_path", None))
+        trace_name = getattr(execution_setup, "trace_name", None)
+        if traces_path is not None:
+            base_dir = traces_path / str(trace_name) if trace_name else traces_path
+
+    if db_path is None and base_dir is not None:
+        db_path = base_dir / "video_blackboard.db"
+    evidence_dir = base_dir / "video_blackboard" / "evidence" if base_dir else None
+
+    session = get_active_session(device_id)
+    current_session_id = getattr(data_engine, "current_session_id", None)
+    if session is not None:
+        board_key = f"video:{session.video_id}"
+    else:
+        board_key = _latest_recorded_video_key(db_path, current_session_id) or (
+            f"session:{current_session_id}" if current_session_id else f"device:{device_id}:runtime"
+        )
+
+    board = VideoBlackboard(board_key, db_path=db_path, evidence_dir=evidence_dir)
+    try:
+        setattr(ctx, "_video_blackboard", board)
+    except (AttributeError, ValueError):
+        logger.debug(
+            "Context does not allow caching video_blackboard; SQLite sharing remains active"
+        )
+    return board

artemis/agents/video_analyzer/reliability.py

@@ -0,0 +1,157 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Failure classification and circuit breaking for video model calls."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import StrEnum
+import threading
+import time
+from typing import Any
+
+
+class VideoFailureCategory(StrEnum):
+    RATE_LIMIT = "rate_limit"
+    PROVIDER_UNAVAILABLE = "provider_unavailable"
+    TIMEOUT = "timeout"
+    CONNECTION = "connection"
+    AUTHENTICATION = "authentication"
+    BAD_REQUEST = "bad_request"
+    MEDIA_PROCESSING = "media_processing"
+    CANCELLED = "cancelled"
+    UNKNOWN = "unknown"
+
+
+@dataclass(frozen=True)
+class VideoFailure:
+    category: VideoFailureCategory
+    retryable: bool
+    should_split: bool
+    should_fallback: bool
+
+
+def classify_video_failure(error: BaseException) -> VideoFailure:
+    """Convert provider/transport/media exceptions into recovery decisions."""
+
+    if isinstance(error, KeyboardInterrupt) or type(error).__name__ == "CancelledError":
+        return VideoFailure(VideoFailureCategory.CANCELLED, False, False, False)
+
+    code = getattr(error, "code", None) or getattr(error, "status_code", None)
+    try:
+        code = int(code) if code is not None else None
+    except (TypeError, ValueError):
+        code = None
+
+    message = str(error).lower()
+    if code == 429 or any(marker in message for marker in ("rate limit", "quota", "resource_exhausted")):
+        return VideoFailure(VideoFailureCategory.RATE_LIMIT, True, False, True)
+    if code in {500, 502, 503, 504} or any(
+        marker in message
+        for marker in ("unavailable", "overloaded", "high demand", "service unavailable")
+    ):
+        return VideoFailure(VideoFailureCategory.PROVIDER_UNAVAILABLE, True, False, True)
+    if isinstance(error, TimeoutError) or "timed out" in message or "timeout" in message:
+        return VideoFailure(VideoFailureCategory.TIMEOUT, True, True, True)
+    if isinstance(error, (ConnectionError, OSError)) and not any(
+        marker in message for marker in ("ffmpeg", "video", "audio", "codec", "media")
+    ):
+        return VideoFailure(VideoFailureCategory.CONNECTION, True, False, True)
+    if code in {401, 403} or any(
+        marker in message for marker in ("unauthorized", "forbidden", "api key", "credential")
+    ):
+        return VideoFailure(VideoFailureCategory.AUTHENTICATION, False, False, True)
+    if any(
+        marker in message
+        for marker in ("ffmpeg", "codec", "corrupt", "invalid video", "invalid audio", "media")
+    ):
+        return VideoFailure(VideoFailureCategory.MEDIA_PROCESSING, False, True, True)
+    if code in {400, 404, 413, 415, 422}:
+        return VideoFailure(VideoFailureCategory.BAD_REQUEST, False, code == 413, True)
+    return VideoFailure(VideoFailureCategory.UNKNOWN, True, True, True)
+
+
+@dataclass
+class _CircuitState:
+    consecutive_failures: int = 0
+    open_until: float = 0.0
+    half_open_in_flight: bool = False
+
+
+class VideoCircuitBreaker:
+    """Small context-scoped circuit breaker keyed by provider/model identity."""
+
+    def __init__(self, threshold: int = 3, cooldown_seconds: float = 60.0) -> None:
+        self.threshold = max(1, int(threshold))
+        self.cooldown_seconds = max(0.1, float(cooldown_seconds))
+        self._states: dict[str, _CircuitState] = {}
+        self._lock = threading.RLock()
+
+    def allow(self, key: str, *, now: float | None = None) -> bool:
+        current = time.time() if now is None else now
+        with self._lock:
+            state = self._states.get(key)
+            if state is None:
+                return True
+            if state.open_until:
+                if current < state.open_until or state.half_open_in_flight:
+                    return False
+                # Half-open: permit exactly one trial. Other concurrent calls
+                # continue to bypass the primary until it succeeds or fails.
+                state.half_open_in_flight = True
+                state.consecutive_failures = max(0, self.threshold - 1)
+                return True
+            return not state.half_open_in_flight
+
+    def record_success(self, key: str) -> None:
+        with self._lock:
+            self._states.pop(key, None)
+
+    def record_failure(
+        self,
+        key: str,
+        failure: VideoFailure,
+        *,
+        now: float | None = None,
+    ) -> None:
+        if failure.category not in {
+            VideoFailureCategory.RATE_LIMIT,
+            VideoFailureCategory.PROVIDER_UNAVAILABLE,
+            VideoFailureCategory.TIMEOUT,
+            VideoFailureCategory.CONNECTION,
+        }:
+            with self._lock:
+                state = self._states.get(key)
+                if state is not None and state.half_open_in_flight:
+                    self._states.pop(key, None)
+            return
+        current = time.time() if now is None else now
+        with self._lock:
+            state = self._states.setdefault(key, _CircuitState())
+            state.half_open_in_flight = False
+            state.consecutive_failures += 1
+            if state.consecutive_failures >= self.threshold:
+                state.open_until = current + self.cooldown_seconds
+
+    def snapshot(self) -> dict[str, dict[str, Any]]:
+        with self._lock:
+            return {
+                key: {
+                    "consecutive_failures": state.consecutive_failures,
+                    "open_until": state.open_until,
+                    "half_open_in_flight": state.half_open_in_flight,
+                }
+                for key, state in self._states.items()
+            }