feat(computer-use): stop resending unchanged screenshots

Port from openclaw/openclaw#129924: a capture whose pixels are
byte-identical to the previous capture of the same target in the same
session returns its full text metadata (element index included) plus an
explicit 'screen unchanged' note instead of the multimodal image block.

Adapted for Hermes: openclaw gates dedup on per-frame context-epoch
tracking; Hermes bounds staleness with a consecutive-omission streak cap
(2) so full pixels are re-delivered before compaction could evict the
referenced image. Dedup state is per-session (no cross-session leaks),
append-only (no history rewrites — prompt cache prefixes untouched),
and skipped entirely when no session_id is present.

tests/tools/test_computer_use_screenshot_dedup.py

@@ -0,0 +1,125 @@
+"""Screenshot dedup — unchanged frames are delivered without their image.
+
+Port of openclaw/openclaw#129924: a capture whose pixels are byte-identical
+to the previous capture of the same target in the same session returns text
+metadata plus an explicit "screen unchanged" note instead of the multimodal
+image block. Changed pixels, changed targets, other sessions, and streak
+exhaustion all deliver full pixels again.
+"""
+
+import base64
+import json
+import struct
+import zlib
+
+import pytest
+
+import tools.computer_use.tool as cu_tool
+from tools.computer_use.backend import CaptureResult
+
+
+def _png_b64(seed: int) -> str:
+    """A tiny real PNG (so dimension sniffing works) with seed-varied bytes."""
+    width = height = 64
+    raw = b"".join(
+        b"\x00" + bytes(((x + y + seed) % 256) for x in range(width * 3))
+        for y in range(height)
+    )
+    def chunk(tag: bytes, data: bytes) -> bytes:
+        return (
+            struct.pack(">I", len(data)) + tag + data
+            + struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF)
+        )
+    png = (
+        b"\x89PNG\r\n\x1a\n"
+        + chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0))
+        + chunk(b"IDAT", zlib.compress(raw))
+        + chunk(b"IEND", b"")
+    )
+    return base64.b64encode(png).decode("ascii")
+
+
+def _cap(seed: int = 1, app: str = "kate", title: str = "notes") -> CaptureResult:
+    return CaptureResult(
+        mode="som", width=64, height=64, png_b64=_png_b64(seed),
+        elements=[], app=app, window_title=title,
+    )
+
+
+@pytest.fixture(autouse=True)
+def _fresh_dedup(monkeypatch):
+    cu_tool._reset_screenshot_dedup()
+    # Keep the aux-vision router out of the way — these tests exercise the
+    # multimodal branch directly.
+    monkeypatch.setattr(cu_tool, "_should_route_through_aux_vision", lambda: False)
+    monkeypatch.setattr(cu_tool, "_persist_capture_image", lambda cap: None)
+    yield
+    cu_tool._reset_screenshot_dedup()
+
+
+def _is_multimodal(resp) -> bool:
+    return isinstance(resp, dict) and resp.get("_multimodal") is True
+
+
+class TestScreenshotDedup:
+    def test_first_capture_delivers_image(self):
+        resp = cu_tool._capture_response(_cap(seed=1), session_id="s1")
+        assert _is_multimodal(resp)
+
+    def test_identical_second_capture_omits_image(self):
+        cu_tool._capture_response(_cap(seed=1), session_id="s1")
+        resp = cu_tool._capture_response(_cap(seed=1), session_id="s1")
+        assert not _is_multimodal(resp)
+        data = json.loads(resp)
+        assert data["screen_unchanged"] is True
+        assert "screen unchanged" in data["summary"]
+
+    def test_changed_pixels_deliver_image_again(self):
+        cu_tool._capture_response(_cap(seed=1), session_id="s1")
+        resp = cu_tool._capture_response(_cap(seed=2), session_id="s1")
+        assert _is_multimodal(resp)
+
+    def test_different_target_delivers_image(self):
+        cu_tool._capture_response(_cap(seed=1, app="kate"), session_id="s1")
+        resp = cu_tool._capture_response(_cap(seed=1, app="kcalc"), session_id="s1")
+        assert _is_multimodal(resp)
+
+    def test_sessions_do_not_share_dedup_state(self):
+        cu_tool._capture_response(_cap(seed=1), session_id="s1")
+        resp = cu_tool._capture_response(_cap(seed=1), session_id="s2")
+        assert _is_multimodal(resp)
+
+    def test_streak_cap_redelivers_pixels(self):
+        cu_tool._capture_response(_cap(seed=1), session_id="s1")
+        omitted = 0
+        for _ in range(cu_tool._SCREENSHOT_DEDUP_MAX_STREAK):
+            resp = cu_tool._capture_response(_cap(seed=1), session_id="s1")
+            assert not _is_multimodal(resp)
+            omitted += 1
+        # Streak exhausted — identical pixels must be re-delivered in full.
+        resp = cu_tool._capture_response(_cap(seed=1), session_id="s1")
+        assert _is_multimodal(resp)
+        assert omitted == cu_tool._SCREENSHOT_DEDUP_MAX_STREAK
+        # And the re-delivery resets the streak: next identical capture
+        # dedups again.
+        resp = cu_tool._capture_response(_cap(seed=1), session_id="s1")
+        assert not _is_multimodal(resp)
+
+    def test_no_session_id_never_dedups(self):
+        cu_tool._capture_response(_cap(seed=1))
+        resp = cu_tool._capture_response(_cap(seed=1))
+        assert _is_multimodal(resp)
+
+    def test_change_resets_streak_state(self):
+        cu_tool._capture_response(_cap(seed=1), session_id="s1")
+        cu_tool._capture_response(_cap(seed=1), session_id="s1")  # omitted
+        cu_tool._capture_response(_cap(seed=3), session_id="s1")  # new pixels
+        resp = cu_tool._capture_response(_cap(seed=3), session_id="s1")
+        assert not _is_multimodal(resp)  # dedups against the NEW frame
+
+    def test_ax_mode_untouched(self):
+        cap = CaptureResult(mode="ax", width=64, height=64, png_b64=None,
+                            elements=[], app="kate", window_title="notes")
+        resp = cu_tool._capture_response(cap, session_id="s1")
+        assert not _is_multimodal(resp)
+        assert "screen_unchanged" not in json.loads(resp)

tools/computer_use/tool.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 import atexit
 import base64
 import contextlib
+import hashlib
 import json
 import logging
 import os
@@ -90,6 +91,38 @@ _AUX_VISION_ROUTE_CACHE: Dict[Tuple[str, str, str], bool] = {}
 _approval_lock = threading.Lock()
 _escalation_warned: set = set()               # sids already warned that a bypass widened the driver mode
 
+# Screenshot dedup: a tight capture→act→capture loop on a static screen resends the same ~1200px image every step.
+# Every delivered frame is hashed (sha256 over mime + base64 payload); when the next capture of the SAME target
+# (app, window) in the SAME session is byte-identical, the tool returns the normal text metadata (element index
+# included) plus an explicit "screen unchanged" note and omits the image block. Append-only — no prior transcript
+# message is rewritten, so prompt-cache prefixes stay intact. Staleness is bounded by a consecutive-omission streak
+# cap: full pixels are re-delivered before compaction (which keeps only the newest image-bearing tool results)
+# could evict the image the note refers to. State is per session; sessionless calls never dedup.
+_screenshot_dedup_lock = threading.Lock()
+_last_screenshot_state: Dict[str, Dict[str, Any]] = {}  # session_id -> {"digest", "target": (app, window), "streak"}
+_SCREENSHOT_DEDUP_MAX_STREAK = 2
+
+def _screenshot_dedup_check(session_id: str, digest: str, target: Tuple[str, str]) -> bool:
+    """True when this capture should be delivered WITHOUT its image: the previous frame for this session had identical
+    bytes for the same target and the omission streak is below _SCREENSHOT_DEDUP_MAX_STREAK. Any miss (new pixels,
+    new target, streak exhausted, first capture) resets the stored state to this digest so the image goes out."""
+    with _screenshot_dedup_lock:
+        state = _last_screenshot_state.get(session_id)
+        if (state is not None and state.get("digest") == digest and state.get("target") == target
+                and int(state.get("streak", 0)) < _SCREENSHOT_DEDUP_MAX_STREAK):
+            state["streak"] = int(state.get("streak", 0)) + 1
+            return True
+        _last_screenshot_state[session_id] = {"digest": digest, "target": target, "streak": 0}
+        return False
+
+def _reset_screenshot_dedup(session_id: Optional[str] = None) -> None:
+    """Forget dedup state (all sessions, or one)."""
+    with _screenshot_dedup_lock:
+        if session_id is None:
+            _last_screenshot_state.clear()
+        else:
+            _last_screenshot_state.pop(session_id, None)
+
 def _cua_permission_mode(session_id: str) -> str:
     """Map Hermes's approval bypass onto Cua's immutable mode; fails closed. Both identity namespaces are consulted
     (DB ``session_id`` and gateway ``session_key`` contextvar) or a gateway ``/yolo`` would be invisible here.
@@ -221,6 +254,7 @@ def _shutdown_backend_atexit() -> None:
 def reset_backend_for_tests() -> None:  # pragma: no cover — tear down the cached backend and per-session state
     _shutdown_backend_atexit()
     _AUX_VISION_ROUTE_CACHE.clear()
+    _reset_screenshot_dedup()
 
 def _noop_stub(name: str, *params: str, result: Any = None):
     # Recording stub: positional args are folded in under *params* (declared params default to None). ``result`` may
@@ -270,7 +304,7 @@ def handle_computer_use(args: Dict[str, Any], **kwargs) -> Any:
         with _backend_lock:
             call_lock = _backend_call_locks.setdefault(session_id, threading.RLock())
         with call_lock:
-            return _dispatch(backend, action, args)
+            return _dispatch(backend, action, args, session_id=session_id or None)
     except Exception as e:
         logger.exception("computer_use %s failed", action)
         return json.dumps({"error": f"{action} failed: {e}"})
@@ -332,12 +366,13 @@ def _do_scroll(backend, action, args, **delivery):
     return backend.scroll(direction=args.get("direction", "down"), amount=int(args.get("amount", 3)),
                           element=args.get("element"), **_scroll_xy(args), modifiers=args.get("modifiers"), **delivery)
 
-def _do_capture(backend, action, args, **_):
+def _do_capture(backend, action, args, session_id=None, **_):
     if (mode := str(args.get("mode", "som"))) not in {"som", "vision", "ax"}:
         return json.dumps({"error": f"bad mode {mode!r}; use som|vision|ax"})
     # pid/window_id forwarded only when given so older backends keep their defaults.
     return _capture_response(backend.capture(mode=mode, app=args.get("app"),
-                                             **{k: args[k] for k in ("pid", "window_id") if args.get(k) is not None}))
+                                             **{k: args[k] for k in ("pid", "window_id") if args.get(k) is not None}),
+                             session_id=session_id)
 
 def _do_listing(backend, action, args, key, **_):
     return json.dumps({key: (items := getattr(backend, action)()), "count": len(items)})
@@ -387,7 +422,7 @@ _ACTION_SUGGESTIONS = {
     "input_text": "type", "screenshot": "capture", "get_window_state": "capture", "left_click": "click", "mouse_click": "click",
 }
 
-def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) -> Any:
+def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any], session_id: Optional[str] = None) -> Any:
     spec = _ACTIONS.get(action)
     if spec is None:
         return json.dumps({"error": f"unknown action {action!r}" + (f" — did you mean {hint!r}? See the action enum in the tool schema."
@@ -400,10 +435,12 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) ->
             f"{action} would go to the current target {mismatch!r}, not {requested_app.strip()!r} "
             "— input actions always hit the sticky target from the last capture/focus_app. "
             f"Call capture(app={requested_app.strip()!r}) or focus_app first, then retry.")})
-    # delivery_mode / bring_to_front thread through every input action (background → foreground ladder).
+    # delivery_mode / bring_to_front thread through every input action (background → foreground ladder); input
+    # handlers forward their kwargs to the backend verbatim, so the dedup session key rides only on read handlers.
     res = spec.handler(backend, action, args, delivery_mode=args.get("delivery_mode"),
-                       bring_to_front=bool(args.get("bring_to_front")))
-    return res if isinstance(res, (str, dict)) else _maybe_follow_capture(backend, res, bool(args.get("capture_after")))
+                       bring_to_front=bool(args.get("bring_to_front")), **({} if spec.input else {"session_id": session_id}))
+    return res if isinstance(res, (str, dict)) else _maybe_follow_capture(backend, res, bool(args.get("capture_after")),
+                                                                         session_id=session_id)
 
 # ── Response shaping ────────────────────────────────────────────────────────
 def _classify_action_result(res: ActionResult) -> Dict[str, Any]:
@@ -546,11 +583,24 @@ def _text_capture_payload(v: SimpleNamespace, summary: str, extra: Optional[Dict
                    bounds_scale=v.bounds_scale),
     })
 
-def _capture_response(cap: CaptureResult, max_elements: int = _DEFAULT_MAX_ELEMENTS) -> Any:
+def _capture_digest(cap: CaptureResult) -> str:
+    return hashlib.sha256((str(cap.image_mime_type or "") + ":").encode("utf-8")
+                          + (cap.png_b64 or "").encode("ascii", "ignore")).hexdigest()
+
+def _capture_response(cap: CaptureResult, max_elements: int = _DEFAULT_MAX_ELEMENTS,
+                      session_id: Optional[str] = None) -> Any:
     v = _capture_view(cap, max_elements)
     lines = _capture_summary_lines(v)
     summary, extra = "\n".join(lines), None  # multimodal/aux paths use this; text paths append notes and rebuild
-    if v.has_image:
+    if v.has_image and session_id and _screenshot_dedup_check(
+            session_id, _capture_digest(cap), (str(cap.app or ""), str(cap.window_title or ""))):
+        # Unchanged frame: same pixels for the same target in this session — no image (and no aux-vision call);
+        # the text metadata is fresh and the note says which screenshot still applies.
+        lines.append("  (screen unchanged since the previous screenshot — image omitted to save context; the prior "
+                     "screenshot still shows the current state. Element indices below are fresh and remain the "
+                     "preferred way to act.)")
+        extra = {"screen_unchanged": True}
+    elif v.has_image:
         # Hand the screenshot to auxiliary.vision (text-only result) when the main model may not consume images
         # natively; returning the multimodal envelope unconditionally tripped HTTP 404/400 at the provider.
         if not _should_route_through_aux_vision():  # envelope carrying the screenshot (not the elements array, so no truncation note)
@@ -581,7 +631,8 @@ def _capture_response(cap: CaptureResult, max_elements: int = _DEFAULT_MAX_ELEME
                      "elements_file — read_file/search_files it, or pass app= to narrow scope)")
     return _text_capture_payload(v, "\n".join(lines), extra)
 
-def _maybe_follow_capture(backend: ComputerUseBackend, res: ActionResult, do_capture: bool) -> Any:
+def _maybe_follow_capture(backend: ComputerUseBackend, res: ActionResult, do_capture: bool,
+                          session_id: Optional[str] = None) -> Any:
     # No follow-up capture after a failed action: a normal-looking screenshot would suggest success.
     if not do_capture or not res.ok:
         return _text_response(res)
@@ -594,7 +645,7 @@ def _maybe_follow_capture(backend: ComputerUseBackend, res: ActionResult, do_cap
     except Exception as e:
         logger.warning("follow-up capture failed: %s", e)
         return _text_response(res)
-    resp, payload = _capture_response(cap), _action_payload(res)
+    resp, payload = _capture_response(cap, session_id=session_id), _action_payload(res)
     if isinstance(resp, dict) and resp.get("_multimodal"):
         # Keep the evidence/verdict contract visible alongside the image — it governs whether input may repeat.
         resp["content"][0]["text"] = resp["text_summary"] = json.dumps(payload) + "\n\n" + resp["text_summary"]