refactor(validator): split validator responsibilities into modules

artemis/agents/validator/action_execution.py

@@ -0,0 +1,98 @@
+# 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.
+
+"""Single-action execution and turn-input helpers for the Validator.
+
+Extracted from ``validator.py``; the public entry points remain the
+``ValidatorNode`` methods, which delegate here.
+"""
+
+import asyncio
+import base64
+import json
+from pathlib import Path
+
+from artemis.context import ArtemisContext
+from artemis.graph.state import State
+from artemis.mcp.action_names import to_canonical_call
+from artemis.mcp.action_session import ActionSession
+from artemis.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+def parse_decisions(structured_decisions: str) -> tuple[list[dict] | None, str | None]:
+    """Parses the Operator's structured decisions JSON into a list of actions."""
+    if not structured_decisions:
+        return None, "No structured decisions found, nothing to execute."
+    try:
+        return json.loads(structured_decisions), None
+    except json.JSONDecodeError as e:
+        return None, f"Failed to parse structured decisions: {e}"
+
+
+async def read_initial_screenshot(state: State) -> tuple[str, str | None]:
+    """Loads the turn-initial screenshot (what the Operator saw) from state."""
+    screenshot_path = getattr(state, "latest_screenshot", None)
+    if not screenshot_path:
+        logger.error("No screenshot path found in state.latest_screenshot")
+        raise ValueError("No screenshot path found in state.latest_screenshot")
+
+    if not Path(screenshot_path).exists():
+        logger.error(f"Screenshot file does not exist: {screenshot_path}")
+        raise FileNotFoundError(f"Screenshot file does not exist: {screenshot_path}")
+
+    try:
+        with open(screenshot_path, "rb") as f:
+            screenshot_b64 = base64.b64encode(f.read()).decode("utf-8")
+    except Exception as e:
+        logger.error(f"Failed to read screenshot from {screenshot_path}: {e}")
+        raise e
+
+    screenshot_name = Path(screenshot_path).stem
+    return screenshot_b64, screenshot_name
+
+
+async def exec_action(
+    ctx: ArtemisContext, session: ActionSession, action_item: dict
+) -> tuple[bool, str]:
+    """Executes one Operator action item through the unified action session.
+
+    The Operator's internal verbs (tap, long_press_on, focus_and_input_text, ...)
+    are translated to canonical tool calls in ``to_canonical_call`` using the
+    ``normalized_coordinates`` the Operator already computed. Success comes from
+    the structured ``ActionResult.ok`` -- no string parsing.
+    """
+    action_name = action_item.get("action")
+
+    if action_name == "wait_for_delay":
+        time_in_ms = action_item.get("time_in_ms", 0)
+        logger.info(f"Validator performing wait_for_delay for {time_in_ms}ms...")
+        await asyncio.sleep(time_in_ms / 1000.0)
+        return True, ""
+
+    width = getattr(ctx.device, "device_width", 1080) if ctx.device else 1080
+    height = getattr(ctx.device, "device_height", 2400) if ctx.device else 2400
+
+    try:
+        tool_name, wire_args = to_canonical_call(action_item, dims=(width, height))
+    except ValueError as e:
+        return False, str(e)
+
+    try:
+        res = await session.call(tool_name, wire_args)
+    except Exception as e:
+        return False, f"MCP call failed: {e}"
+
+    return (True, "") if res.ok else (False, res.message)

artemis/agents/validator/execution_loop.py

@@ -0,0 +1,445 @@
+# 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.
+
+"""Per-turn action execution loop for the Validator.
+
+Orchestrates, for each Operator decision: precondition safety-net checks
+(XML-first with pixel fallback), local execution with retries, screenshot and
+trace bookkeeping, and FailureAnalyzer-driven local repair.
+
+Extracted from ``validator.py``; the public entry point remains
+``ValidatorNode.__call__``, which delegates here. All patchable seams
+(_exec_action, _validate_action_precondition*, _get_initial_screenshot, ...)
+are dispatched through the node so existing test patches keep working.
+"""
+
+import asyncio
+import base64
+from dataclasses import dataclass
+from pathlib import Path
+import time
+import uuid
+from uuid import UUID
+
+from artemis.agents.validator.failure_analyzer import (
+    FailureAnalyzer,
+    ValidationErrorCategory,
+)
+from artemis.data_engine.trace import CURRENT_TRACE_ID
+from artemis.graph.state import State
+from artemis.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class _ActionOutcome:
+    """Result of processing a single action item (pre-check + execution)."""
+
+    action_item: dict
+    success: bool = False
+    error_msg: str = ""
+    error_category: ValidationErrorCategory = ValidationErrorCategory.GENERAL
+    post_screenshot_b64: str | None = None
+
+
+async def _run_precondition_gate(
+    node, session, action_item: dict, state: State, pre_screenshot_b64: str, original_coords: list
+) -> tuple[bool, ValidationErrorCategory, str]:
+    """Runs the pre-execution Safety Net: XML-first with pixel fallback."""
+    is_explorer_candidate = action_item.get("target_class") == "ExplorerCandidate"
+    is_ocr_element = "[OCR]" in str(action_item.get("target_class") or "")
+    has_index_metadata = bool(
+        (
+            action_item.get("target_text")
+            or action_item.get("target_bounds")
+            or action_item.get("target_resource_id")
+        )
+        and not is_explorer_candidate
+        and not is_ocr_element
+    )
+
+    if not has_index_metadata:
+        # Fall back to Pixel-based validation for pure coordinate/non-index interactions
+        return await node._validate_action_precondition_pixel(
+            session, action_item, pre_screenshot_b64, original_coords, state=state
+        )
+
+    # Index-based validation is fastest and safest when metadata is present
+    (
+        validation_passed,
+        validation_category,
+        validation_error,
+    ) = await node._validate_action_precondition(session, action_item, state=state)
+
+    if validation_passed:
+        return validation_passed, validation_category, validation_error
+
+    if validation_category == ValidationErrorCategory.XML_BYPASSED:
+        logger.info(
+            "Pre-execution XML-based validation bypassed"
+            f" ({validation_error}). Falling back to"
+            " Pixel-based validation..."
+        )
+        return await node._validate_action_precondition_pixel(
+            session, action_item, pre_screenshot_b64, original_coords, state=state
+        )
+
+    logger.info(
+        "Pre-execution XML-based validation failed:"
+        f" {validation_error}. Attempting Pixel-based"
+        " validation fallback..."
+    )
+    (
+        pixel_passed,
+        pixel_category,
+        pixel_error,
+    ) = await node._validate_action_precondition_pixel(
+        session, action_item, pre_screenshot_b64, original_coords, state=state
+    )
+    if pixel_passed and pixel_category == ValidationErrorCategory.NONE:
+        logger.success(
+            "Pixel-based validation fallback PASSED! Overriding XML validation failure."
+        )
+        return True, ValidationErrorCategory.NONE, ""
+
+    return validation_passed, validation_category, validation_error
+
+
+async def _attempt_local_execution(
+    node, session, action_item: dict, action_name, attempts_log: list
+) -> tuple[bool, str]:
+    """Executes the action locally with a short retry loop."""
+    max_local_retries = 1 if action_name == "launch_app" else 2
+    success = False
+    error_msg = ""
+    for attempt in range(max_local_retries):
+        try:
+            exec_success, exec_error = await node._exec_action(session, action_item)
+        except Exception as e:
+            exec_success = False
+            exec_error = str(e)
+
+        # Skip polling: assume execution success implies action success
+        if exec_success:
+            success = True
+            error_msg = ""
+            attempts_log.append("Success")
+            break
+        else:
+            success = False
+            error_msg = exec_error
+            attempts_log.append(error_msg)
+
+        if attempt < max_local_retries - 1:
+            await asyncio.sleep(0.5)
+
+    return success, error_msg
+
+
+async def _capture_live_screenshot(session, context_desc: str) -> str | None:
+    try:
+        return await session.screenshot_b64()
+    except Exception as e:
+        logger.error(f"Failed to capture {context_desc}: {e!r}")
+        return None
+
+
+async def _process_action(
+    node, state: State, session, action_item: dict, action_name, pre_screenshot_b64: str
+) -> _ActionOutcome:
+    """Runs precondition checks and execution for one action item."""
+    action_item = dict(action_item)  # Make a copy to avoid mutating original state actions
+    outcome = _ActionOutcome(action_item=action_item)
+    attempts_log = []
+
+    # 1. Pre-execution validation (Safety Net)
+    original_coords = list(action_item.get("coordinates") or [])
+    (
+        validation_passed,
+        validation_category,
+        validation_error,
+    ) = await _run_precondition_gate(
+        node, session, action_item, state, pre_screenshot_b64, original_coords
+    )
+
+    if not validation_passed:
+        outcome.success = False
+        outcome.error_msg = f"Pre-execution validation failed: {validation_error}"
+        outcome.error_category = validation_category
+        attempts_log.append(outcome.error_msg)
+
+        # Capture the live mismatch screenshot to provide context for FailureAnalyzer
+        outcome.post_screenshot_b64 = await _capture_live_screenshot(
+            session, "live screenshot for failure analysis"
+        )
+    else:
+        # 2. Local execution attempts
+        outcome.success, outcome.error_msg = await _attempt_local_execution(
+            node, session, action_item, action_name, attempts_log
+        )
+        if not outcome.success and not outcome.post_screenshot_b64:
+            logger.info("Action failed, capturing failure screenshot...")
+            outcome.post_screenshot_b64 = await _capture_live_screenshot(
+                session, "failure screenshot"
+            )
+
+    # Enrich the executed action record
+    if len(attempts_log) > 1 or not outcome.success:
+        action_item["attempts"] = list(attempts_log)
+
+    return outcome
+
+
+def _absorb_post_screenshot(ctx, state: State, post_screenshot_b64: str) -> str | None:
+    """Persists the post-action screenshot and points state at it.
+
+    Returns the stored image name (None when no data engine is attached).
+    """
+    decoded_bytes = base64.b64decode(post_screenshot_b64)
+    logger.info(f"Successfully decoded post_screenshot_b64 ({len(decoded_bytes)} bytes)")
+    if not ctx.data_engine:
+        return None
+
+    post_image_name = ctx.data_engine.get_or_create_image(decoded_bytes)
+    screenshot_path = str(ctx.data_engine.get_image_path(post_image_name))
+    state.latest_screenshot = screenshot_path
+    logger.info(f"Validator updated state.latest_screenshot to: {screenshot_path}")
+    return post_image_name
+
+
+def _record_action_start(ctx, step_id, action_name, action_item, parent_id, action_trace_id):
+    if ctx.data_engine and step_id:
+        ctx.data_engine.record_trace(
+            type="action",
+            name=action_name,
+            payload={"action": action_item, "status": "running"},
+            status="running",
+            parent_trace_id=parent_id,
+            step_id=step_id,
+            trace_id=action_trace_id,
+        )
+
+
+def _record_action_end(
+    ctx,
+    step_id,
+    action_name,
+    outcome: _ActionOutcome,
+    post_image_name,
+    parent_id,
+    action_trace_id,
+    start_time: float,
+):
+    duration = time.time() - start_time
+    if ctx.data_engine and step_id:
+        relative_time = ctx.data_engine.get_relative_time(time.time())
+        ctx.data_engine.record_trace(
+            type="action",
+            name=action_name,
+            payload={
+                "action": outcome.action_item,
+                "success": outcome.success,
+                "error_msg": outcome.error_msg,
+                "post_screenshot": post_image_name,
+                "timestamp": time.time(),
+                "relative_time": relative_time,
+            },
+            status="success" if outcome.success else "failed",
+            duration=duration,
+            parent_trace_id=parent_id,
+            step_id=step_id,
+            trace_id=action_trace_id,
+        )
+
+
+def _reload_post_repair_screenshot(
+    ctx, state: State, last_screenshot_b64: str, last_screenshot_name: str | None
+) -> tuple[str, str | None]:
+    """Reloads the latest screenshot updated by FailureAnalyzer during repair."""
+    try:
+        screenshot_path = getattr(state, "latest_screenshot", None)
+        if screenshot_path:
+            if Path(screenshot_path).exists():
+                with open(screenshot_path, "rb") as f:
+                    last_screenshot_b64 = base64.b64encode(f.read()).decode("utf-8")
+                if ctx.data_engine:
+                    last_screenshot_name = ctx.data_engine.get_or_create_image(
+                        base64.b64decode(last_screenshot_b64)
+                    )
+                logger.info(f"Validator reloaded post-repair screenshot from: {screenshot_path}")
+    except Exception as e:
+        logger.error(f"Failed to reload post-repair screenshot: {e}")
+    return last_screenshot_b64, last_screenshot_name
+
+
+async def _analyze_failure(
+    node,
+    state: State,
+    outcome: _ActionOutcome,
+    *,
+    decision_screenshot_b64,
+    decision_screenshot_name,
+    post_image_name,
+    executed_actions,
+    unexecuted_actions,
+) -> dict:
+    """Triggers the FailureAnalyzer for a failed action and records its analysis."""
+    logger.warning(f"Action failed: {outcome.action_item}. Triggering failure analysis.")
+
+    analyzer = FailureAnalyzer(node.ctx)
+    analysis_result = await analyzer.analyze(
+        state,
+        outcome.action_item,
+        outcome.error_msg,
+        pre_screenshot=decision_screenshot_b64,
+        post_screenshot=outcome.post_screenshot_b64,
+        pre_screenshot_name=decision_screenshot_name,
+        post_screenshot_name=post_image_name,
+        executed_actions=executed_actions,
+        unexecuted_actions=unexecuted_actions,
+        error_category=outcome.error_category,
+    )
+    outcome.action_item["repair"] = analysis_result.get("analysis", "No analysis provided.")
+    return analysis_result
+
+
+async def run_validation_loop(node, state: State) -> dict:
+    """Executes all pending Operator decisions and returns the execution report.
+
+    ``node`` is the ValidatorNode facade; every overridable seam is invoked
+    through it.
+    """
+    ctx = node.ctx
+
+    actions, error_msg = node._parse_decisions(state.structured_decisions)
+    if error_msg:
+        return {}
+
+    execution: list[dict] = []
+    failed_action = None
+    actions_to_execute = list(actions)
+
+    session = await node._get_mcp_session()
+
+    try:
+        last_screenshot_b64, last_screenshot_name = await node._get_initial_screenshot(
+            session, state
+        )
+        # Preserve the turn-initial screenshot (what the operator saw when making decisions)
+        decision_screenshot_b64 = last_screenshot_b64
+        decision_screenshot_name = last_screenshot_name
+    except Exception:
+        return {}
+
+    step_id = UUID(state.current_step_id) if state.current_step_id else None
+
+    analysis_result: dict = {}
+    success = False
+
+    while actions_to_execute:
+        action_item = actions_to_execute.pop(0)
+        action_name = action_item.get("action")
+
+        logger.info(f"Executing action: {action_name}")
+
+        action_trace_id = uuid.uuid4()
+        parent_id = CURRENT_TRACE_ID.get()
+        token = CURRENT_TRACE_ID.set(action_trace_id)
+        start_time = time.time()
+
+        _record_action_start(ctx, step_id, action_name, action_item, parent_id, action_trace_id)
+
+        try:
+            outcome = await _process_action(
+                node, state, session, action_item, action_name, last_screenshot_b64
+            )
+            success = outcome.success
+            execution.append(outcome.action_item)
+
+            post_image_name = None
+            if outcome.post_screenshot_b64:
+                post_image_name = _absorb_post_screenshot(
+                    ctx, state, outcome.post_screenshot_b64
+                )
+                last_screenshot_b64 = outcome.post_screenshot_b64
+                if post_image_name:
+                    last_screenshot_name = post_image_name
+
+            _record_action_end(
+                ctx,
+                step_id,
+                action_name,
+                outcome,
+                post_image_name,
+                parent_id,
+                action_trace_id,
+                start_time,
+            )
+        finally:
+            CURRENT_TRACE_ID.reset(token)
+
+        if not success:
+            failed_action = outcome.action_item
+
+            analysis_result = await _analyze_failure(
+                node,
+                state,
+                outcome,
+                decision_screenshot_b64=decision_screenshot_b64,
+                decision_screenshot_name=decision_screenshot_name,
+                post_image_name=post_image_name,
+                executed_actions=execution[:-1],
+                unexecuted_actions=list(actions_to_execute),
+            )
+
+            if analysis_result.get("status") == "fixed":
+                logger.success("Failure repaired locally!")
+                actions_to_execute = []
+                last_screenshot_b64, last_screenshot_name = _reload_post_repair_screenshot(
+                    ctx, state, last_screenshot_b64, last_screenshot_name
+                )
+                continue
+            else:
+                break
+
+    # Append any skipped actions to the execution log
+    for skipped in actions_to_execute:
+        skipped_copy = dict(skipped)
+        skipped_copy["attempts"] = ["Skipped"]
+        execution.append(skipped_copy)
+
+    report = {
+        "execution": execution,
+        "status": (
+            "success"
+            if success or (failed_action and analysis_result.get("status") == "fixed")
+            else "failed"
+        ),
+        "repair_status": (analysis_result.get("status") if failed_action else None),
+    }
+
+    # Determine if a distinct post-action screenshot was captured
+    distinct_post_image_name = (
+        last_screenshot_name
+        if (last_screenshot_name and last_screenshot_name != decision_screenshot_name)
+        else None
+    )
+
+    if ctx.data_engine and step_id:
+        ctx.data_engine.update_step_execution_result(
+            step_id, report, post_image_name=distinct_post_image_name
+        )
+
+    return {"last_execution_result": report}

artemis/agents/validator/precondition_pixel.py

@@ -0,0 +1,286 @@
+# 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.
+
+"""Pixel-level Safety Net: VLM comparison of the target crop before execution.
+
+Validates that the visual target button/component at the action coordinates is
+still present, visible, and interactive on the live screen compared to the
+original (decision-time) screenshot.
+
+Extracted from ``validator.py``; the public entry point remains the
+``ValidatorNode`` method, which delegates here. The ``get_llm_fn`` dependency
+is injected by the facade so that tests patching
+``artemis.agents.validator.validator.get_llm`` keep working.
+"""
+
+import asyncio
+import base64
+from pathlib import Path
+
+from langchain_core.messages import HumanMessage, SystemMessage
+
+from artemis.agents.validator.failure_analyzer import ValidationErrorCategory
+from artemis.context import ArtemisContext
+from artemis.graph.state import State
+from artemis.services.llm import acomplete_structured
+from artemis.utils import visualization
+from artemis.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+# Only interactive, element-focused actions are subject to pixel validation.
+PIXEL_VALIDATED_ACTIONS = (
+    "tap",
+    "long_press_on",
+    "focus_and_input_text",
+    "input_text",
+    "click_coordinate",
+    "click",
+    "long_press",
+)
+
+_MAX_ATTEMPTS = 3
+_RETRY_DELAY_SECONDS = 0.3
+
+
+def _init_llm_and_prompt(ctx: ArtemisContext, get_llm_fn):
+    """Resolves the VLM and loads the prompt template (once per validation)."""
+    try:
+        llm = get_llm_fn(ctx, name="validator_pixel_safety_net")
+    except Exception:
+        llm = get_llm_fn(ctx, name="validator")
+
+    prompt_path = Path(__file__).parent.joinpath("pixel_safety_net.md")
+    prompt = prompt_path.read_text(encoding="utf-8")
+    return llm, prompt
+
+
+def _build_messages(
+    prompt: str,
+    orig_crop_bytes: bytes,
+    live_crop_bytes: bytes,
+    action_name,
+    state: State | None,
+) -> list:
+    """Formulates the multi-modal verification message pair."""
+    orig_b64 = base64.b64encode(orig_crop_bytes).decode("utf-8")
+    live_b64 = base64.b64encode(live_crop_bytes).decode("utf-8")
+
+    user_content = [
+        {"type": "text", "text": "[Image 1 (Reference)]"},
+        {
+            "type": "image_url",
+            "image_url": {"url": f"data:image/jpeg;base64,{orig_b64}"},
+        },
+        {"type": "text", "text": "[Image 2 (Current State)]"},
+        {
+            "type": "image_url",
+            "image_url": {"url": f"data:image/jpeg;base64,{live_b64}"},
+        },
+    ]
+    if state:
+        thoughts = []
+        native_thought = getattr(state, "operator_native_thinking", None)
+        if native_thought and native_thought.strip():
+            thoughts.append(native_thought.strip())
+        raw_thought = getattr(state, "operator_raw_thinking", None)
+        if raw_thought and raw_thought.strip():
+            thoughts.append(raw_thought.strip())
+
+        operator_thought = "\n\n".join(thoughts)
+        if operator_thought.strip():
+            context_text = (
+                "[Planned Action & Original Thinking]\nAction:"
+                f" {action_name}\nOriginal"
+                f" Thinking:\n{operator_thought.strip()}"
+            )
+            user_content.append({"type": "text", "text": f"\n{context_text.strip()}"})
+
+    return [
+        SystemMessage(content=prompt),
+        HumanMessage(content=user_content),
+    ]
+
+
+async def _run_attempt(
+    session,
+    llm,
+    prompt: str,
+    orig_crop_bytes: bytes,
+    current_coords,
+    action_name,
+    state: State | None,
+    attempt: int,
+) -> tuple[bool, ValidationErrorCategory, str] | None:
+    """Runs one pixel validation attempt.
+
+    Returns a final (passed, category, reason) verdict, or None when the
+    attempt yielded no verdict (never happens today: every branch is final).
+    Raises on infrastructure errors so the caller's retry loop can decide.
+    """
+    # A. Take fresh live screenshot (captures state changes / dynamic loading settling)
+    live_screenshot_b64 = None
+    try:
+        live_screenshot_b64 = await session.screenshot_b64()
+        success_img = True
+    except Exception:
+        success_img, live_screenshot_b64 = False, None
+
+    if not success_img or not live_screenshot_b64:
+        raise Exception("Failed to acquire live screenshot from controller/MCP tool.")
+
+    # B. Decode and crop the live screen target
+    live_bytes = base64.b64decode(live_screenshot_b64)
+    live_crop_bytes = visualization.crop_and_annotate_target(
+        live_bytes, current_coords, crop_size=None, dot_radius=15
+    )
+
+    # C. Formulate multi-modal verification message
+    messages = _build_messages(prompt, orig_crop_bytes, live_crop_bytes, action_name, state)
+
+    # D. Invoke Universal VLM. Parsing (fences, repair) and one
+    # corrective re-ask on unparseable JSON are handled by the
+    # structured-output layer; a StructuredOutputError lands in
+    # the attempt loop's except like any other attempt failure.
+    res_json = await acomplete_structured(llm, messages)
+    if not isinstance(res_json, dict):
+        raise ValueError(
+            "Pixel validation response parsed to"
+            f" {type(res_json).__name__}, expected a JSON object."
+        )
+    reasoning = res_json.get("reasoning", "")
+    is_present = res_json.get("is_present", True)
+    confidence = res_json.get("confidence", 1.0)
+
+    logger.info(
+        f"Pixel validation attempt {attempt}/{_MAX_ATTEMPTS} result:"
+        f" is_present={is_present}, confidence={confidence:.2f}, reasoning={reasoning}"
+    )
+
+    # E. Branching based on outcome
+    if is_present:
+        logger.info(f"Pixel validation SUCCESS on attempt {attempt}/{_MAX_ATTEMPTS}.")
+        return True, ValidationErrorCategory.NONE, ""
+
+    # Target not present: fail or bypass based on confidence without retry!
+    if confidence >= 0.7:
+        reason_msg = (
+            f"Pixel-level validation failed: {reasoning}"
+            if reasoning
+            else (
+                "Pixel-level validation failed: The target UI"
+                " element at the exact coordinates has disappeared,"
+                " changed, or is blocked."
+            )
+        )
+        return (
+            False,
+            ValidationErrorCategory.PIXEL_TARGET_DISAPPEARED,
+            reason_msg,
+        )
+
+    logger.warning(
+        "Pixel validation failed but confidence"
+        f" ({confidence:.2f}) is below 0.7. Bypassing check."
+    )
+    return True, ValidationErrorCategory.PIXEL_BYPASSED, ""
+
+
+async def validate_action_precondition_pixel(
+    ctx: ArtemisContext,
+    session,
+    action_item: dict,
+    pre_screenshot_b64: str,
+    original_coords: list[int] | None = None,
+    state: State | None = None,
+    *,
+    get_llm_fn,
+) -> tuple[bool, ValidationErrorCategory, str]:
+    """Pixel-level Safety Net entry point (see module docstring)."""
+    action_name = action_item.get("action")
+
+    # 1. White-list filtering: Only validate interactive, element-focused actions
+    if action_name not in PIXEL_VALIDATED_ACTIONS:
+        return True, ValidationErrorCategory.PIXEL_BYPASSED, ""
+
+    current_coords = action_item.get("coordinates")
+    orig_coords = original_coords or current_coords
+
+    if not orig_coords or len(orig_coords) != 2:
+        logger.info("Pixel safety net skipped: no valid coordinates.")
+        return True, ValidationErrorCategory.PIXEL_BYPASSED, ""
+
+    if not current_coords or len(current_coords) != 2:
+        current_coords = orig_coords
+
+    logger.info(
+        "Pre-execution pixel validation: checking target position"
+        f" {orig_coords} in original image vs {current_coords} on live"
+        " screen..."
+    )
+
+    # 2. Decode and crop the static reference (original) image ONCE outside the loop to save CPU
+    try:
+        orig_bytes = base64.b64decode(pre_screenshot_b64)
+        orig_crop_bytes = visualization.crop_and_annotate_target(
+            orig_bytes, orig_coords, crop_size=None, dot_radius=15
+        )
+    except Exception as e:
+        logger.error(
+            f"Failed to prepare original crop for pixel safety net: {e}. Bypassing check."
+        )
+        return True, ValidationErrorCategory.PIXEL_BYPASSED, ""
+
+    # 3. Get LLM Configuration once outside the loop
+    try:
+        llm, prompt = _init_llm_and_prompt(ctx, get_llm_fn)
+    except Exception as e:
+        logger.error(
+            f"Failed to initialize LLM/prompt for pixel safety net: {e}. Bypassing check."
+        )
+        return True, ValidationErrorCategory.PIXEL_BYPASSED, ""
+
+    # 4. Start high-precision execution validation retry loop
+    last_error = None
+    for attempt in range(1, _MAX_ATTEMPTS + 1):
+        logger.info(f"Pre-execution pixel validation: Attempt {attempt}/{_MAX_ATTEMPTS}...")
+        try:
+            verdict = await _run_attempt(
+                session,
+                llm,
+                prompt,
+                orig_crop_bytes,
+                current_coords,
+                action_name,
+                state,
+                attempt,
+            )
+            if verdict is not None:
+                return verdict
+        except Exception as attempt_err:
+            logger.warning(
+                f"Error during pixel validation attempt {attempt}/{_MAX_ATTEMPTS}: {attempt_err}"
+            )
+            last_error = attempt_err
+            if attempt < _MAX_ATTEMPTS:
+                await asyncio.sleep(_RETRY_DELAY_SECONDS)
+                continue
+            logger.error(
+                "Pixel validation repeatedly failed with errors. Last"
+                f" error: {last_error}. Bypassing check to avoid"
+                " blocking flow."
+            )
+            return True, ValidationErrorCategory.PIXEL_BYPASSED, ""
+
+    return True, ValidationErrorCategory.PIXEL_BYPASSED, ""

artemis/agents/validator/precondition_xml.py

@@ -0,0 +1,667 @@
+# 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.
+
+"""XML-hierarchy Safety Net: pre-execution validation of the action target.
+
+Checks that the element the Operator decided to interact with is still present
+on the live screen (via the UI hierarchy) and substantially consistent, with
+coordinate self-healing for small drifts and a structured failure taxonomy
+(shifted / occupied / disappeared) consumed by the FailureAnalyzer.
+
+Extracted from ``validator.py``; the public entry points remain the
+``ValidatorNode`` methods, which delegate here.
+"""
+
+import asyncio
+import difflib
+import math
+import re
+
+from artemis.agents.validator.failure_analyzer import ValidationErrorCategory
+from artemis.constants import VALIDATOR_UI_HIERARCHY_TIMEOUT
+from artemis.context import ArtemisContext
+from artemis.graph.state import State
+from artemis.utils import visualization
+from artemis.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+# Only interactive, element-focused actions are subject to XML validation.
+XML_VALIDATED_ACTIONS = (
+    "tap",
+    "long_press_on",
+    "focus_and_input_text",
+    "focus_and_clear_text",
+)
+
+# Signal weights for candidate scoring.
+W_ID = 0.5
+W_TEXT = 0.4
+W_BOUNDS = 0.3
+W_COORD = 0.3
+
+
+async def validate_action_precondition(
+    node, session, action_item: dict, state: State | None = None
+) -> tuple[bool, ValidationErrorCategory, str]:
+    """Retry wrapper around the single-shot XML precondition validation.
+
+    A short retry loop accommodates screen transitions and loading delays.
+    ``node`` is the ValidatorNode; the single-shot check is dispatched through
+    it so tests can patch the method.
+    """
+    max_pre_attempts = 3
+    pre_retry_delay = 0.4
+
+    passed = False
+    category = ValidationErrorCategory.NONE
+    reason = ""
+
+    for pre_attempt in range(1, max_pre_attempts + 1):
+        passed, category, reason = await node._validate_action_precondition_single(
+            session, action_item, state
+        )
+        if passed:
+            return True, category, reason
+
+        if category == ValidationErrorCategory.XML_BYPASSED:
+            # Do not retry on XML bypass/timeout; return immediately
+            # so it falls back to Pixel-based VLM validation
+            return False, category, reason
+
+        if pre_attempt < max_pre_attempts:
+            logger.info(
+                "Pre-execution validation failed on attempt"
+                f" {pre_attempt}/{max_pre_attempts}: {reason}. Retrying in"
+                f" {pre_retry_delay}s..."
+            )
+            await asyncio.sleep(pre_retry_delay)
+        else:
+            break
+
+    return passed, category, reason
+
+
+def _resolve_screen_dims(ctx: ArtemisContext, state: State | None) -> tuple[int, int]:
+    """Resolves the screen dimensions the Operator's coordinates refer to."""
+    width = 1080
+    height = 2400
+    operator_raw_data = getattr(state, "operator_raw_data", {}) or {}
+    w_raw = operator_raw_data.get("width")
+    h_raw = operator_raw_data.get("height")
+    if isinstance(w_raw, int) and isinstance(h_raw, int):
+        width = w_raw
+        height = h_raw
+    else:
+        device = getattr(ctx, "device", None)
+        if device:
+            width = getattr(device, "device_width", 1080)
+            height = getattr(device, "device_height", 2400)
+    return width, height
+
+
+async def _fetch_live_elements(session):
+    """Pulls the live XML tree with a strict timeout.
+
+    Returns ``(elements, None)`` on success, or ``(None, (False, category,
+    reason))`` when the check should be bypassed to Pixel-based validation.
+    """
+    try:
+        try:
+            # The timeout must ride inside the session (MCP read timeout), never an
+            # outer asyncio.wait_for: cancelling call_tool mid-flight corrupts the
+            # in-memory transport and bricks the session for all later callers.
+            elements = await session.ui_hierarchy(timeout=VALIDATOR_UI_HIERARCHY_TIMEOUT)
+        except Exception as e:
+            logger.warning(
+                f"Failed to get live XML via MCP: {e!r}. Falling back to"
+                " Pixel-based validation."
+            )
+            return None, (
+                False,
+                ValidationErrorCategory.XML_BYPASSED,
+                f"XML hierarchy fetch timed out or errored: {e!r}",
+            )
+
+        if not elements or not isinstance(elements, list):
+            logger.warning(
+                "Live hierarchy empty or invalid. Falling back to Pixel-based validation."
+            )
+            return None, (
+                False,
+                ValidationErrorCategory.XML_BYPASSED,
+                "Live hierarchy empty or invalid.",
+            )
+
+    except Exception as e:
+        logger.error(
+            f"Unexpected error during live XML fetch: {e}. Falling back to"
+            " Pixel-based validation."
+        )
+        return None, (
+            False,
+            ValidationErrorCategory.XML_BYPASSED,
+            f"Unexpected error during live XML fetch: {e}",
+        )
+
+    return elements, None
+
+
+def _normalize_text(s: str | None) -> str:
+    if not s:
+        return ""
+    s = s.lower()
+    s = re.sub(r"\(\d+\+?\)", "", s)
+    s = re.sub(r"\[\d+\+?\]", "", s)
+    s = re.sub(r"[^\w\s\d]", "", s)
+    return s.strip()
+
+
+def _aggregate_text(elem: dict) -> str:
+    """Recursively aggregates text from a candidate container node."""
+    text = elem.get("text") or elem.get("content-desc") or elem.get("accessibilityText") or ""
+    if not isinstance(text, str):
+        text = str(text)
+    children = elem.get("children") or []
+    if children:
+        for child in children:
+            if isinstance(child, dict):
+                text += " " + _aggregate_text(child)
+    return text
+
+
+def _text_similarity(norm_target_text: str, elem_norm_text: str) -> float:
+    """Scores text similarity between normalized target and element text."""
+    if not (norm_target_text and elem_norm_text):
+        return 0.0
+    if norm_target_text == elem_norm_text:
+        return 1.0
+    if norm_target_text in elem_norm_text or elem_norm_text in norm_target_text:
+        return 0.8
+    return difflib.SequenceMatcher(None, norm_target_text, elem_norm_text).ratio()
+
+
+def _bounds_match(target_bounds, elem_bounds) -> tuple[float, bool]:
+    """Returns (IoU bounds score, size-mismatch flag) for the bounds pair."""
+    bounds_score = 0.0
+    size_mismatch = False
+    if target_bounds and elem_bounds and len(target_bounds) == 4 and len(elem_bounds) == 4:
+        l1, t1, r1, b1 = target_bounds
+        l2, t2, r2, b2 = elem_bounds
+
+        # Check size ratio mismatch to prevent matching small widget to giant container
+        w1, h1 = r1 - l1, b1 - t1
+        w2, h2 = r2 - l2, b2 - t2
+        if w1 > 0 and h1 > 0 and w2 > 0 and h2 > 0:
+            size_ratio_w = max(w1, w2) / min(w1, w2)
+            size_ratio_h = max(h1, h2) / min(h1, h2)
+            if size_ratio_w > 2.5 or size_ratio_h > 2.5:
+                size_mismatch = True
+
+        li = max(l1, l2)
+        ti = max(t1, t2)
+        ri = min(r1, r2)
+        bi = min(b1, b2)
+        if ri > li and bi > ti:
+            inter_area = (ri - li) * (bi - ti)
+            union_area = (r1 - l1) * (b1 - t1) + (r2 - l2) * (b2 - t2) - inter_area
+            if union_area > 0:
+                bounds_score = inter_area / union_area
+
+    return bounds_score, size_mismatch
+
+
+def _combine_signals(
+    *,
+    target_resource_id,
+    target_text,
+    target_bounds,
+    original_coords,
+    id_match: bool,
+    elem_res_id: str,
+    norm_target_text: str,
+    elem_norm_text: str,
+    text_score: float,
+    bounds_score: float,
+    contains_coord: bool,
+) -> tuple[float, float]:
+    """Combines the weighted match signals into (score, identity_score)."""
+    signals = []
+    weights = []
+    identity_signals = []
+    identity_weights = []
+
+    # 1. Resource ID Signal
+    if target_resource_id:
+        weights.append(W_ID)
+        identity_weights.append(W_ID)
+        if id_match:
+            signals.append(W_ID * 1.0)
+            identity_signals.append(W_ID * 1.0)
+        elif not elem_res_id:
+            signals.append(W_ID * 0.3)  # Soft penalty for missing ID
+            identity_signals.append(W_ID * 0.3)
+        else:
+            signals.append(W_ID * -0.5)  # Hard penalty for mismatched ID
+            identity_signals.append(W_ID * -0.5)
+
+    # 2. Text Signal
+    if target_text:
+        weights.append(W_TEXT)
+        identity_weights.append(W_TEXT)
+        if norm_target_text and elem_norm_text:
+            signals.append(W_TEXT * text_score)
+            identity_signals.append(W_TEXT * text_score)
+        elif not elem_norm_text:
+            signals.append(W_TEXT * 0.0)  # Penalty for missing expected text
+            identity_signals.append(W_TEXT * 0.0)
+
+    # 3. Bounds Signal
+    if target_bounds:
+        weights.append(W_BOUNDS)
+        signals.append(W_BOUNDS * bounds_score)
+
+    # 4. Coordinate Containment Signal
+    if original_coords:
+        weights.append(W_COORD)
+        signals.append(W_COORD * (1.0 if contains_coord else 0.0))
+
+    score = sum(signals) / sum(weights) if weights else 1.0
+    identity_score = (
+        sum(identity_signals) / sum(identity_weights) if identity_weights else 1.0
+    )
+    return score, identity_score
+
+
+def _score_element(
+    elem: dict,
+    action_item: dict,
+    *,
+    target_text,
+    target_bounds,
+    target_resource_id,
+    norm_target_text: str,
+) -> dict | None:
+    """Scores one hierarchy element against the action target metadata.
+
+    Returns a candidate dict, or None if the element has no usable bounds.
+    """
+    elem_bounds_str = elem.get("bounds")
+    elem_bounds = visualization.parse_bounds(elem_bounds_str)
+    if not elem_bounds:
+        return None
+
+    elem_raw_text = _aggregate_text(elem)
+    elem_norm_text = _normalize_text(elem_raw_text)
+    elem_res_id = elem.get("resource-id") or elem.get("resourceId") or ""
+
+    elem_cx, elem_cy = visualization.get_center_coordinates(*elem_bounds)
+
+    original_coords = action_item.get("coordinates") or [0, 0]
+    dist = math.sqrt((elem_cx - original_coords[0]) ** 2 + (elem_cy - original_coords[1]) ** 2)
+
+    id_match = target_resource_id and elem_res_id and target_resource_id == elem_res_id
+
+    text_score = _text_similarity(norm_target_text, elem_norm_text)
+    bounds_score, size_mismatch = _bounds_match(target_bounds, elem_bounds)
+
+    # Calculate Coordinate Containment
+    contains_coord = False
+    if original_coords and len(original_coords) == 2 and elem_bounds and len(elem_bounds) == 4:
+        cx, cy = original_coords
+        l2, t2, r2, b2 = elem_bounds
+        contains_coord = l2 <= cx <= r2 and t2 <= cy <= b2
+
+    # Apply size mismatch penalty
+    if size_mismatch and not id_match:
+        text_score = 0.0
+        bounds_score = 0.0
+        contains_coord = False
+
+    score, identity_score = _combine_signals(
+        target_resource_id=target_resource_id,
+        target_text=target_text,
+        target_bounds=target_bounds,
+        original_coords=original_coords,
+        id_match=id_match,
+        elem_res_id=elem_res_id,
+        norm_target_text=norm_target_text,
+        elem_norm_text=elem_norm_text,
+        text_score=text_score,
+        bounds_score=bounds_score,
+        contains_coord=contains_coord,
+    )
+
+    decay = max(0.5, 1.0 - (dist / 800.0))
+    score *= decay
+
+    return {
+        "element": elem,
+        "center": [elem_cx, elem_cy],
+        "bounds": elem_bounds,
+        "text": elem_raw_text,
+        "resource_id": elem_res_id,
+        "score": score,
+        "identity_score": identity_score,
+        "distance": dist,
+        "text_score": text_score,
+        "id_match": id_match,
+    }
+
+
+def _handle_match_success(
+    action_item: dict, best: dict, target_text, scale_factor: float
+) -> tuple[bool, ValidationErrorCategory, str]:
+    """Handles a passing match: self-heals coordinates for moderate drifts."""
+    if best["distance"] <= 200 * scale_factor:
+        new_coords = best["center"]
+        old_coords = action_item.get("coordinates")
+        if old_coords != new_coords:
+            logger.success(
+                "Pre-execution validation SUCCESS"
+                f" (score={best['score']:.2f}). Target element"
+                f" '{target_text}' shifted by {best['distance']:.1f}px."
+                f" Self-healing: correcting coordinates {old_coords} ->"
+                f" {new_coords}."
+            )
+            action_item["coordinates"] = new_coords
+            # Drop the now-stale normalized coordinates so the canonical
+            # translation re-derives them from the healed pixel center.
+            action_item.pop("normalized_coordinates", None)
+    else:
+        logger.info(
+            "Pre-execution validation SUCCESS"
+            f" (score={best['score']:.2f}). Target element matched but"
+            f" shifted by {best['distance']:.1f}px"
+            f" (>{200 * scale_factor:.1f}px). Bypassing coordinate"
+            " self-healing to remain conservative."
+        )
+    return True, ValidationErrorCategory.NONE, ""
+
+
+def _is_valid_shift(best: dict, scale_factor: float) -> bool:
+    """Applies the strict shift criteria (preferring disappeared over shifted).
+
+    We must be conservative to avoid confusing disappearance as shift. Shift
+    misclassification as disappeared is acceptable, but disappeared as shift
+    causes errors.
+    """
+    max_allowed_distance = 100.0 * scale_factor
+    if best.get("id_match", False):
+        # Allow larger distance threshold only for strong resource ID matches
+        max_allowed_distance = 300.0 * scale_factor
+
+    if best["distance"] <= max_allowed_distance:
+        if best["identity_score"] >= 0.85:
+            # High confidence identity match (e.g. exact text or exact ID)
+            return True
+        if best["identity_score"] >= 0.5 and best.get("id_match", False):
+            # Moderate confidence with exact resource ID match
+            return True
+    return False
+
+
+def _element_is_interactive(e: dict) -> bool:
+    clickable = e.get("clickable")
+    long_clickable = e.get("long-clickable") or e.get("longClickable")
+    focusable = e.get("focusable")
+    enabled = e.get("enabled")
+
+    def to_bool(v) -> bool:
+        if v is None:
+            return False
+        if isinstance(v, bool):
+            return v
+        return str(v).lower() == "true"
+
+    return to_bool(clickable) or to_bool(long_clickable) or to_bool(focusable) or to_bool(enabled)
+
+
+def _find_occupant(elements: list, original_coords) -> tuple | None:
+    """Finds the most specific element containing the expected coordinates."""
+    occupant = None
+    if original_coords and len(original_coords) == 2:
+        orig_cx, orig_cy = original_coords
+        for elem in elements:
+            if not isinstance(elem, dict):
+                continue
+            elem_bounds_str = elem.get("bounds")
+            elem_bounds = visualization.parse_bounds(elem_bounds_str)
+            if elem_bounds:
+                left, top, right, bottom = elem_bounds
+                if left <= orig_cx <= right and top <= orig_cy <= bottom:
+                    elem_area = (right - left) * (bottom - top)
+                    if not occupant or elem_area < occupant[2]:
+                        occupant = (elem, elem_bounds, elem_area)
+    return occupant
+
+
+def _coords_hit_interactive_element(
+    elements: list, original_coords, width: int, height: int
+) -> bool:
+    """True if any non-full-screen interactive element covers the coordinates."""
+    if not (original_coords and len(original_coords) == 2):
+        return False
+    orig_cx, orig_cy = original_coords
+    for e in elements:
+        if not isinstance(e, dict):
+            continue
+        eb_str = e.get("bounds")
+        eb = visualization.parse_bounds(eb_str)
+        if eb:
+            e_left, e_top, e_right, e_bottom = eb
+            if e_left <= orig_cx <= e_right and e_top <= orig_cy <= e_bottom:
+                # Exclude full-screen root background elements
+                e_width = e_right - e_left
+                e_height = e_bottom - e_top
+                is_full_screen = e_width >= width - 10 and e_height >= height - 10
+
+                if _element_is_interactive(e) and not is_full_screen:
+                    return True
+    return False
+
+
+def _describe_occupation(
+    elements: list,
+    occupant: tuple | None,
+    original_coords,
+    target_bounds,
+    width: int,
+    height: int,
+) -> tuple[bool, str, list | None]:
+    """Decides whether the expected coordinates are occupied by another element.
+
+    Returns (is_occupied, occupant_desc, occupant_bounds).
+    """
+    is_occupied = False
+    occupant_desc = "anonymous element"
+    occupant_bounds = None
+
+    if occupant:
+        elem, elem_bounds, _ = occupant
+        occupant_text = (
+            elem.get("text") or elem.get("content-desc") or elem.get("accessibilityText") or ""
+        )
+        occupant_res_id = elem.get("resource-id") or elem.get("resourceId") or ""
+        occupant_bounds = elem_bounds
+
+        # Check if the occupant has content
+        has_content = bool(occupant_text or occupant_res_id)
+
+        # Check if this occupant or any container covering this point is interactive
+        coords_interactive = _coords_hit_interactive_element(
+            elements, original_coords, width, height
+        )
+
+        # Exclude full-screen or giant parent containers from being considered blockers
+        is_occupant_full_screen = False
+        if occupant_bounds:
+            o_w = occupant_bounds[2] - occupant_bounds[0]
+            o_h = occupant_bounds[3] - occupant_bounds[1]
+            is_occupant_full_screen = o_w >= width - 10 and o_h >= height - 10
+
+            # If target bounds are known, check if occupant is a giant container
+            if target_bounds:
+                t_w = target_bounds[2] - target_bounds[0]
+                t_h = target_bounds[3] - target_bounds[1]
+                t_area = t_w * t_h
+                o_area = o_w * o_h
+                if t_area > 0 and o_area > 3.0 * t_area:
+                    is_occupant_full_screen = True
+
+        if (has_content or coords_interactive) and not is_occupant_full_screen:
+            is_occupied = True
+            occupant_desc = (
+                f"'{occupant_text}' ({occupant_res_id})"
+                if has_content
+                else "interactive anonymous element"
+            )
+
+    return is_occupied, occupant_desc, occupant_bounds
+
+
+def _classify_failure(
+    elements: list,
+    action_item: dict,
+    best: dict,
+    *,
+    target_text,
+    target_bounds,
+    target_resource_id,
+    width: int,
+    height: int,
+    scale_factor: float,
+) -> tuple[bool, ValidationErrorCategory, str]:
+    """Classifies a below-threshold match as shifted / occupied / disappeared."""
+    original_coords = action_item.get("coordinates")
+
+    # 1. Shifted Case
+    if _is_valid_shift(best, scale_factor):
+        reason = (
+            f"Target element '{target_text}' ({target_resource_id})"
+            f" expected at {original_coords} (bounds: {target_bounds})"
+            f" has shifted.\n- New location: {best['center']} (bounds:"
+            f" {best['bounds']})"
+        )
+        return False, ValidationErrorCategory.TARGET_SHIFTED, reason
+
+    # 2. Occupied Case
+    occupant = _find_occupant(elements, original_coords)
+    is_occupied, occupant_desc, occupant_bounds = _describe_occupation(
+        elements, occupant, original_coords, target_bounds, width, height
+    )
+    if is_occupied:
+        reason = (
+            f"The expected position {original_coords} for"
+            f" '{target_text}' ({target_resource_id}) is"
+            " occupied/intercepted by a different element:"
+            f" {occupant_desc} at bounds {occupant_bounds}."
+        )
+        return False, ValidationErrorCategory.TARGET_OCCUPIED, reason
+
+    # 3. Completely Missing Case
+    reason = (
+        f"Target element '{target_text}' ({target_resource_id})"
+        f" expected at {original_coords} (bounds:"
+        f" {target_bounds}) was not found on the screen."
+    )
+    return False, ValidationErrorCategory.TARGET_DISAPPEARED, reason
+
+
+async def validate_action_precondition_single(
+    ctx: ArtemisContext, session, action_item: dict, state: State | None = None
+) -> tuple[bool, ValidationErrorCategory, str]:
+    """Single-shot Safety Net check against the live XML hierarchy.
+
+    Returns (True, ValidationErrorCategory.NONE, "") if validation passes or
+    is skipped. Returns (False, category, error_msg) if validation fails.
+    """
+    action_name = action_item.get("action")
+
+    # 1. White-list filtering: Only validate interactive, element-focused actions
+    if action_name not in XML_VALIDATED_ACTIONS:
+        return True, ValidationErrorCategory.NONE, ""
+
+    target_text = action_item.get("target_text")
+    target_bounds = action_item.get("target_bounds")  # [l, t, r, b]
+    target_resource_id = action_item.get("target_resource_id")
+
+    width, height = _resolve_screen_dims(ctx, state)
+
+    # Calculate diagonal-based scale factor for resolution-independent thresholds
+    ref_diagonal = math.sqrt(1080**2 + 2400**2)  # ~2631.8
+    current_diagonal = math.sqrt(width**2 + height**2)
+    scale_factor = current_diagonal / ref_diagonal
+
+    # 2. Skip validation if there's no target metadata to match against
+    if not target_text and not target_bounds and not target_resource_id:
+        return True, ValidationErrorCategory.NONE, ""
+
+    logger.info(
+        f"Pre-execution validation: checking target '{target_text}'"
+        f" ({target_resource_id}) at {target_bounds}..."
+    )
+
+    # 3. Pull live XML tree with strict timeout to avoid hanging the execution flow
+    elements, bypass = await _fetch_live_elements(session)
+    if bypass is not None:
+        return bypass
+
+    # 4. Matching Heuristics & Scoring
+    norm_target_text = _normalize_text(target_text)
+
+    candidates = []
+    for elem in elements:
+        if not isinstance(elem, dict):
+            continue
+        candidate = _score_element(
+            elem,
+            action_item,
+            target_text=target_text,
+            target_bounds=target_bounds,
+            target_resource_id=target_resource_id,
+            norm_target_text=norm_target_text,
+        )
+        if candidate is not None:
+            candidates.append(candidate)
+
+    if not candidates:
+        logger.warning(
+            "No candidates with valid bounds found. Falling back to Pixel-based validation."
+        )
+        return (
+            False,
+            ValidationErrorCategory.XML_BYPASSED,
+            "No candidates with valid bounds found.",
+        )
+
+    candidates.sort(key=lambda x: x["score"], reverse=True)
+    best = candidates[0]
+
+    threshold = 0.55 if best["distance"] <= 150 * scale_factor else 0.75
+
+    if best["score"] >= threshold:
+        return _handle_match_success(action_item, best, target_text, scale_factor)
+
+    return _classify_failure(
+        elements,
+        action_item,
+        best,
+        target_text=target_text,
+        target_bounds=target_bounds,
+        target_resource_id=target_resource_id,
+        width=width,
+        height=height,
+        scale_factor=scale_factor,
+    )