Add ac_wait_for_image and ac_wait_for_pixel

Polling wait helpers fill the gap left by ac_wait_for_window /
ac_wait_for_text — useful for 'click then wait until the spinner
disappears' style flows. Both tools accept a ToolCallContext, so
they emit progress notifications and abort on
notifications/cancelled. Timeout becomes a TimeoutError surfaced
as a clean tool error to the model.

Author: JE-Chen

je_auto_control/utils/mcp_server/tools/_factories.py

@@ -143,6 +143,38 @@ def screen_tools() -> List[MCPTool]:
             handler=h.get_pixel,
             annotations=READ_ONLY,
         ),
+        MCPTool(
+            name="ac_wait_for_image",
+            description=("Poll the screen until ``image_path`` appears, "
+                         "returning its centre [x, y]. Raises after "
+                         "``timeout`` seconds. Cancellable: clients can "
+                         "send notifications/cancelled to abort."),
+            input_schema=schema({
+                "image_path": {"type": "string"},
+                "timeout": {"type": "number"},
+                "poll": {"type": "number"},
+                "detect_threshold": {"type": "number"},
+            }, required=["image_path"]),
+            handler=h.wait_for_image,
+            annotations=READ_ONLY,
+        ),
+        MCPTool(
+            name="ac_wait_for_pixel",
+            description=("Poll pixel (x, y) until it matches ``target_rgb`` "
+                         "within ``tolerance`` per channel. Returns the "
+                         "actual [r, g, b] reading on match."),
+            input_schema=schema({
+                "x": {"type": "integer"},
+                "y": {"type": "integer"},
+                "target_rgb": {"type": "array",
+                                "items": {"type": "integer"}},
+                "tolerance": {"type": "integer"},
+                "timeout": {"type": "number"},
+                "poll": {"type": "number"},
+            }, required=["x", "y", "target_rgb"]),
+            handler=h.wait_for_pixel,
+            annotations=READ_ONLY,
+        ),
         MCPTool(
             name="ac_diff_screenshots",
             description=("Compare two screenshots and return the bounding "

je_auto_control/utils/mcp_server/tools/_handlers.py

@@ -178,6 +178,60 @@ def get_pixel(x: int, y: int) -> List[int]:
     return [int(component) for component in pixel]
 
 
+def wait_for_image(image_path: str, timeout: float = 10.0,
+                   poll: float = 0.5,
+                   detect_threshold: float = 1.0,
+                   ctx: Any = None) -> List[int]:
+    """Poll for ``image_path`` on screen; return its centre [x, y] or raise."""
+    import time as _time
+    from je_auto_control.utils.exception.exceptions import ImageNotFoundException
+    from je_auto_control.wrapper.auto_control_image import locate_image_center as _loc
+    poll_seconds = max(0.05, float(poll))
+    deadline = _time.monotonic() + float(timeout)
+    while _time.monotonic() < deadline:
+        if ctx is not None:
+            ctx.check_cancelled()
+            ctx.progress(_time.monotonic() - (deadline - float(timeout)),
+                          total=float(timeout),
+                          message=f"waiting for {image_path}")
+        try:
+            cx, cy = _loc(image_path,
+                          detect_threshold=float(detect_threshold))
+            return [int(cx), int(cy)]
+        except ImageNotFoundException:
+            _time.sleep(poll_seconds)
+    raise TimeoutError(
+        f"wait_for_image timed out after {timeout}s: {image_path!r}"
+    )
+
+
+def wait_for_pixel(x: int, y: int, target_rgb: List[int],
+                   tolerance: int = 8, timeout: float = 10.0,
+                   poll: float = 0.25,
+                   ctx: Any = None) -> List[int]:
+    """Poll until pixel ``(x, y)`` matches ``target_rgb`` within ``tolerance``."""
+    import time as _time
+    from je_auto_control.wrapper.auto_control_screen import get_pixel as _pixel
+    if len(target_rgb) < 3:
+        raise ValueError("target_rgb must contain at least 3 channels")
+    target = [int(c) for c in target_rgb[:3]]
+    tol = max(0, int(tolerance))
+    poll_seconds = max(0.05, float(poll))
+    deadline = _time.monotonic() + float(timeout)
+    while _time.monotonic() < deadline:
+        if ctx is not None:
+            ctx.check_cancelled()
+        raw = _pixel(int(x), int(y))
+        if raw is not None and len(raw) >= 3:
+            channels = [int(raw[i]) for i in range(3)]
+            if all(abs(channels[i] - target[i]) <= tol for i in range(3)):
+                return channels
+        _time.sleep(poll_seconds)
+    raise TimeoutError(
+        f"wait_for_pixel timed out after {timeout}s at ({x}, {y})"
+    )
+
+
 def diff_screenshots(image_path_a: str,
                      image_path_b: str,
                      threshold: int = 16,

test/unit_test/headless/test_mcp_server.py

@@ -1239,6 +1239,68 @@ def test_logging_set_level_rejects_unknown_name():
     assert response["error"]["code"] == -32602
 
 
+def test_wait_for_image_returns_center_when_template_found(monkeypatch):
+    import je_auto_control.utils.mcp_server.tools._handlers as handlers
+    import je_auto_control.wrapper.auto_control_image as image_module
+    monkeypatch.setattr(image_module, "locate_image_center",
+                        lambda image_path, detect_threshold=1.0: (42, 84))
+    by_name = {tool.name: tool for tool in build_default_tool_registry()}
+    coords = by_name["ac_wait_for_image"].invoke({
+        "image_path": "needle.png", "timeout": 1.0, "poll": 0.05,
+    })
+    assert coords == [42, 84]
+
+
+def test_wait_for_image_times_out_when_template_missing(monkeypatch):
+    from je_auto_control.utils.exception.exceptions import (
+        ImageNotFoundException,
+    )
+    import je_auto_control.wrapper.auto_control_image as image_module
+
+    def always_miss(image_path, detect_threshold=1.0):
+        raise ImageNotFoundException("nope")
+
+    monkeypatch.setattr(image_module, "locate_image_center", always_miss)
+    by_name = {tool.name: tool for tool in build_default_tool_registry()}
+    try:
+        by_name["ac_wait_for_image"].invoke({
+            "image_path": "missing.png",
+            "timeout": 0.2, "poll": 0.05,
+        })
+    except TimeoutError as error:
+        assert "timed out" in str(error)
+    else:
+        raise AssertionError("expected TimeoutError")
+
+
+def test_wait_for_pixel_returns_when_match(monkeypatch):
+    import je_auto_control.wrapper.auto_control_screen as screen_module
+    monkeypatch.setattr(screen_module, "get_pixel",
+                        lambda x, y, hwnd=None: (255, 0, 0))
+    by_name = {tool.name: tool for tool in build_default_tool_registry()}
+    rgb = by_name["ac_wait_for_pixel"].invoke({
+        "x": 1, "y": 2, "target_rgb": [255, 0, 0],
+        "tolerance": 2, "timeout": 0.5, "poll": 0.05,
+    })
+    assert rgb == [255, 0, 0]
+
+
+def test_wait_for_pixel_times_out_when_color_never_matches(monkeypatch):
+    import je_auto_control.wrapper.auto_control_screen as screen_module
+    monkeypatch.setattr(screen_module, "get_pixel",
+                        lambda x, y, hwnd=None: (0, 0, 0))
+    by_name = {tool.name: tool for tool in build_default_tool_registry()}
+    try:
+        by_name["ac_wait_for_pixel"].invoke({
+            "x": 1, "y": 2, "target_rgb": [255, 0, 0],
+            "tolerance": 2, "timeout": 0.2, "poll": 0.05,
+        })
+    except TimeoutError:
+        pass
+    else:
+        raise AssertionError("expected TimeoutError")
+
+
 def test_default_registry_lists_core_automation_tools():
     names = {tool.name for tool in build_default_tool_registry()}
     expected = {