When Gemini Computer Use Acts on a Stale Screen and Fails Quietly — Field Notes on Guarding the Loop

import time
from google import genai
from google.genai import types
 
client = genai.Client()
MODEL = "gemini-2.5-computer-use-preview"  # confirm the exact model name in the latest docs
 
def capture():
    """Environment-specific. Returns screenshot bytes and logical size."""
    png, width, height = grab_screen()  # your implementation
    return png, width, height
 
def run_step(prev_png, goal, history):
    """One reasoning step. Hand the model the latest screen and the goal."""
    resp = client.models.generate_content(
        model=MODEL,
        contents=[
            types.Part.from_bytes(data=prev_png, mime_type="image/png"),
            types.Part.from_text(text=f"Goal: {goal}\nSo far: {history}\nReturn only the next single action"),
        ],
        config=types.GenerateContentConfig(
            tools=[types.Tool(computer_use=types.ComputerUse())],
            temperature=0.0,
        ),
    )
    return extract_action(resp)  # pull out {"type": "click", "x":..,"y":..} etc.

def perceptual_hash(png):
    """Hash the look of the screen. A lightweight fingerprint for stale detection."""
    return dhash(png)  # difference hash, used for nearness rather than exact match
 
def execute_with_guard(action, observed_png, observed_hash):
    # 1) Capture again right before executing
    fresh_png, w, h = capture()
    if hamming(observed_hash, perceptual_hash(fresh_png)) > STALE_THRESHOLD:
        # Screen differs from observation time -> abort the action on the old frame
        return {"status": "stale", "fresh_png": fresh_png}
 
    # 2) Compare only the local region around the click target (more sensitive than full-screen)
    if action["type"] == "click":
        roi = crop(fresh_png, action["x"], action["y"], radius=48)
        roi_at_observe = crop(observed_png, action["x"], action["y"], radius=48)
        if local_changed(roi, roi_at_observe):
            return {"status": "stale_local", "fresh_png": fresh_png}
 
    # 3) Execute -> capture after -> confirm the screen actually changed
    do_action(action)
    time.sleep(SETTLE_MS / 1000)
    after_png, _, _ = capture()
    progressed = hamming(perceptual_hash(fresh_png), perceptual_hash(after_png)) > NOOP_THRESHOLD
    return {"status": "ok" if progressed else "noop", "after_png": after_png}

DESTRUCTIVE = {"delete", "submit", "purchase", "send"}
 
def guarded_destructive(action, fresh_png):
    label = action.get("label", "")
    if action["type"] not in DESTRUCTIVE:
        return execute_with_guard(action, fresh_png, perceptual_hash(fresh_png))
 
    # Preconditions: is the button text what we expect? is a confirm dialog present?
    text_here = ocr_near(fresh_png, action["x"], action["y"], radius=64)
    if label and label not in text_here:
        # The model meant to press "Delete" but the real screen shows another label -> abort
        return {"status": "precondition_failed", "expected": label, "found": text_here}
 
    # For irreversible actions, split into two steps and require a confirmation screen
    return {"status": "needs_confirm_screen", "action": action}

from collections import Counter
 
class RunMetrics:
    def __init__(self):
        self.counts = Counter()
        self.recent_hashes = []
 
    def record(self, result, screen_hash):
        self.counts[result["status"]] += 1
        self.recent_hashes.append(screen_hash)
        self.recent_hashes = self.recent_hashes[-5:]
 
    def stuck_len(self):
        if not self.recent_hashes:
            return 0
        last = self.recent_hashes[-1]
        return sum(1 for h in reversed(self.recent_hashes)
                   if hamming(h, last) <= NOOP_THRESHOLD)
 
    def success_rate(self):
        ok = self.counts["ok"]
        total = sum(self.counts.values()) or 1
        return ok / total

def run_task(goal, max_steps=40):
    m = RunMetrics()
    history = []
    for step in range(max_steps):
        png, w, h = capture()
        h_now = perceptual_hash(png)
        if m.stuck_len() >= 3:
            return escalate("stuck", goal, history)       # treading water on one screen
        if m.success_rate() < 0.5 and step > 8:
            return escalate("low_success", goal, history)  # no traction early on
        action = run_step(png, goal, history)
        if action["type"] == "done":
            return {"status": "completed", "steps": step}
        result = (guarded_destructive(action, png)
                  if action["type"] in DESTRUCTIVE else
                  execute_with_guard(action, png, h_now))
        m.record(result, h_now)
        history.append({"step": step, "action": action["type"], "result": result["status"]})
    return escalate("budget_exhausted", goal, history)