OpenAI computer_use Action Loop (Batched Actions, Screenshot Submission)

import base64
import asyncio
from openai import AsyncOpenAI
from playwright.async_api import async_playwright, Page

client = AsyncOpenAI()


MAX_ITERATIONS = 30                    # never let a task exceed this


async def execute_action(page: Page, action: dict, viewport: dict):
    """Execute a single computer_use action in Playwright."""
    t = action.get("type")

    if t == "click":
        button = action.get("button", "left")
        await page.mouse.click(action["x"], action["y"], button=button)

    elif t == "double_click":
        await page.mouse.dblclick(action["x"], action["y"])

    elif t == "move":
        await page.mouse.move(action["x"], action["y"])

    elif t == "drag":
        # GA drag has a path array
        path = action.get("path", [])
        if path:
            await page.mouse.move(path[0]["x"], path[0]["y"])
            await page.mouse.down()
            for pt in path[1:]:
                await page.mouse.move(pt["x"], pt["y"])
            await page.mouse.up()

    elif t == "type":
        await page.keyboard.type(action["text"])

    elif t == "key" or t == "keypress":
        keys = action.get("keys") or [action.get("key")]
        for k in keys:
            await page.keyboard.press(k)

    elif t == "scroll":
        # scroll_x and scroll_y are the deltas
        await page.mouse.wheel(action.get("scroll_x", 0), action.get("scroll_y", 0))

    elif t == "screenshot":
        # No-op — we take a screenshot after every action anyway
        pass

    elif t == "wait":
        await asyncio.sleep(action.get("duration_ms", 1000) / 1000)

    else:
        print(f"unknown action type: {t}")


async def capture_screenshot(page: Page, viewport: dict) -> str:
    """Screenshot the page, return as base64-encoded PNG at declared dims."""
    raw = await page.screenshot(full_page=False)   # visible viewport only

    # Optional: verify dimensions match declared
    # (Retina and DPR quirks may require resizing)
    from PIL import Image
    import io
    img = Image.open(io.BytesIO(raw))
    if img.size != (viewport["width"], viewport["height"]):
        img = img.resize((viewport["width"], viewport["height"]), Image.LANCZOS)
    out = io.BytesIO()
    img.save(out, format="PNG")
    return base64.b64encode(out.getvalue()).decode()


async def computer_use_loop(page: Page, goal: str, viewport: dict = {"width": 1280, "height": 800}):
    """Full action loop until task complete or iteration cap hit."""

    # 1. Initial call with goal + starting screenshot
    screenshot_b64 = await capture_screenshot(page, viewport)

    response = await client.responses.create(
        model="gpt-5.4",
        tools=[{
            "type": "computer_use_preview",
            "display_width": viewport["width"],
            "display_height": viewport["height"],
            "environment": "browser",
        }],
        input=[{
            "role": "user",
            "content": [
                {"type": "input_text", "text": goal},
                {"type": "input_image", "image_url": f"data:image/png;base64,{screenshot_b64}"},
            ],
        }],
        truncation="auto",
    )

    for iteration in range(MAX_ITERATIONS):
        # 2. Extract computer_call items from response.output
        computer_calls = [item for item in response.output if item.type == "computer_call"]

        if not computer_calls:
            # 3a. No more actions — task complete
            print(f"task completed at iteration {iteration}")
            final_message = next((item for item in response.output if item.type == "message"), None)
            if final_message:
                for part in final_message.content:
                    if part.type == "output_text":
                        return part.text
            return "(no final message)"

        # 3b. Process each computer_call
        output_items = []
        for call in computer_calls:
            # GA: iterate actions[]; preview: single action
            actions = call.actions if hasattr(call, "actions") and call.actions else [call.action]

            for action_obj in actions:
                # action may be a Pydantic model — convert to dict
                action = action_obj if isinstance(action_obj, dict) else action_obj.model_dump()
                await execute_action(page, action, viewport)

            # Take fresh screenshot AFTER all batched actions
            await page.wait_for_load_state("domcontentloaded", timeout=5_000)
            screenshot_b64 = await capture_screenshot(page, viewport)

            # 4. Build computer_call_output item
            output_items.append({
                "type": "computer_call_output",
                "call_id": call.call_id,
                "output": {
                    "type": "computer_screenshot",
                    "image_url": f"data:image/png;base64,{screenshot_b64}",
                },
                # acknowledged_safety_checks (see #157) added if needed
            })

        # 5. Chain to next Response with previous_response_id
        response = await client.responses.create(
            model="gpt-5.4",
            previous_response_id=response.id,
            tools=[{
                "type": "computer_use_preview",
                "display_width": viewport["width"],
                "display_height": viewport["height"],
                "environment": "browser",
            }],
            input=output_items,
            truncation="auto",
        )

    raise RuntimeError(f"hit MAX_ITERATIONS={MAX_ITERATIONS} without task completion")


# ✅ Full run
async def main():
    p = await async_playwright().start()
    browser = await p.chromium.launch(headless=False)
    context = await browser.new_context(viewport={"width": 1280, "height": 800})
    page = await context.new_page()
    await page.goto("https://example.com")

    result = await computer_use_loop(
        page,
        goal="Find the \"About\" link, click it, and report the first paragraph of the page.",
    )
    print("Final result:", result)

    await browser.close()
    await p.stop()


if __name__ == "__main__":
    asyncio.run(main())

from typing import Any


def extract_actions(computer_call) -> list[dict]:
    """Return list of action dicts, whether the call is GA (batched) or preview (single).
    Works with SDK model objects and raw dicts."""

    def to_dict(x):
        if isinstance(x, dict):
            return x
        if hasattr(x, "model_dump"):
            return x.model_dump()
        return dict(x)

    call = to_dict(computer_call) if not isinstance(computer_call, dict) else computer_call

    # GA: actions is an array
    if "actions" in call and call["actions"]:
        return [to_dict(a) for a in call["actions"]]

    # Preview: single action
    if "action" in call and call["action"]:
        return [to_dict(call["action"])]

    return []


# ✅ Example — GA batched call
ga_call = {
    "type": "computer_call",
    "id": "cu_abc123",
    "call_id": "call_xyz789",
    "actions": [
        {"type": "move", "x": 640, "y": 400},
        {"type": "click", "x": 640, "y": 400, "button": "left"},
        {"type": "type", "text": "hello world"},
    ],
    "pending_safety_checks": [],
    "status": "completed",
}
actions = extract_actions(ga_call)
# → [{"type": "move", ...}, {"type": "click", ...}, {"type": "type", ...}]


# ✅ Example — legacy preview call
preview_call = {
    "type": "computer_call",
    "id": "cu_def456",
    "call_id": "call_uvw012",
    "action": {"type": "click", "x": 100, "y": 200, "button": "left"},
    "pending_safety_checks": [],
    "status": "completed",
}
actions = extract_actions(preview_call)
# → [{"type": "click", "x": 100, "y": 200, "button": "left"}]


# ✅ Ordered execution — the order matters (move BEFORE click, click BEFORE type)
async def execute_batched_actions(page, actions: list[dict], between_delay_ms: int = 50):
    """Execute in the order provided. Small delay between actions helps flaky UIs settle."""
    for i, action in enumerate(actions):
        await execute_action(page, action, viewport={"width": 1280, "height": 800})
        if i < len(actions) - 1 and between_delay_ms:
            import asyncio; await asyncio.sleep(between_delay_ms / 1000)


# ✅ Full action-type coverage — every action a model might emit
async def execute_action(page, action: dict, viewport: dict):
    t = action.get("type")

    # Mouse actions
    if t == "click":
        await page.mouse.click(action["x"], action["y"], button=action.get("button", "left"))
    elif t == "double_click":
        await page.mouse.dblclick(action["x"], action["y"])
    elif t == "right_click":
        await page.mouse.click(action["x"], action["y"], button="right")
    elif t == "move" or t == "mouse_move":
        await page.mouse.move(action["x"], action["y"])
    elif t == "drag":
        path = action.get("path", [])
        if path:
            await page.mouse.move(path[0]["x"], path[0]["y"])
            await page.mouse.down()
            for pt in path[1:]:
                await page.mouse.move(pt["x"], pt["y"])
            await page.mouse.up()

    # Keyboard actions
    elif t == "type":
        await page.keyboard.type(action["text"], delay=action.get("delay", 20))
    elif t == "key" or t == "keypress" or t == "key_press":
        keys = action.get("keys") or ([action.get("key")] if action.get("key") else [])
        for k in keys:
            # Convert model names to Playwright names if needed
            k_norm = normalize_key(k)
            await page.keyboard.press(k_norm)

    # Scrolling
    elif t == "scroll":
        # scroll_x / scroll_y are pixel deltas
        await page.mouse.wheel(action.get("scroll_x", 0), action.get("scroll_y", 0))

    # Waits / no-ops
    elif t == "wait":
        import asyncio; await asyncio.sleep(action.get("duration_ms", 1000) / 1000)
    elif t == "screenshot":
        pass  # we screenshot after the batch anyway

    else:
        print(f"[warn] unknown action type: {t}, action: {action}")


def normalize_key(k: str) -> str:
    """Map model key names to Playwright key names."""
    mapping = {
        "ENTER": "Enter", "RETURN": "Enter",
        "ESC": "Escape", "ESCAPE": "Escape",
        "TAB": "Tab",
        "SPACE": " ",
        "BACKSPACE": "Backspace",
        "DELETE": "Delete",
        "ARROW_UP": "ArrowUp", "ARROW_DOWN": "ArrowDown",
        "ARROW_LEFT": "ArrowLeft", "ARROW_RIGHT": "ArrowRight",
        "HOME": "Home", "END": "End",
        "PAGE_UP": "PageUp", "PAGE_DOWN": "PageDown",
        "CTRL": "Control", "CMD": "Meta", "COMMAND": "Meta",
        "SHIFT": "Shift", "ALT": "Alt",
    }
    return mapping.get(k.upper(), k)


# ✅ Testing — verify handler works with both shapes
def test_extract_actions():
    ga = {"actions": [{"type": "click", "x": 1, "y": 2}]}
    preview = {"action": {"type": "click", "x": 3, "y": 4}}
    empty = {}

    assert len(extract_actions(ga)) == 1
    assert len(extract_actions(preview)) == 1
    assert len(extract_actions(empty)) == 0
    assert extract_actions(ga)[0]["x"] == 1
    assert extract_actions(preview)[0]["x"] == 3

import time
from dataclasses import dataclass, field


@dataclass
class LoopState:
    iteration: int = 0
    last_action_hash: str | None = None
    same_action_count: int = 0
    start_time: float = field(default_factory=time.monotonic)

    max_iterations: int = 30
    max_seconds: int = 300
    same_action_threshold: int = 3          # kill loop if same action 3× in a row

    def check_stop_conditions(self, current_actions: list[dict]) -> str | None:
        """Return a stop-reason string, or None to continue."""
        self.iteration += 1

        if self.iteration > self.max_iterations:
            return f"hit max_iterations={self.max_iterations}"

        elapsed = time.monotonic() - self.start_time
        if elapsed > self.max_seconds:
            return f"hit max_seconds={self.max_seconds} (elapsed {elapsed:.0f}s)"

        # Detect action loops
        import hashlib, json
        current_hash = hashlib.sha256(json.dumps(current_actions, sort_keys=True).encode()).hexdigest()
        if current_hash == self.last_action_hash:
            self.same_action_count += 1
            if self.same_action_count >= self.same_action_threshold:
                return f"same action repeated {self.same_action_count} times — likely stuck"
        else:
            self.last_action_hash = current_hash
            self.same_action_count = 1

        return None


async def robust_computer_use_loop(page, goal: str, viewport: dict):
    """Loop with iteration cap, timeout, and stuck-detection."""
    from openai import AsyncOpenAI
    client = AsyncOpenAI()

    state = LoopState(max_iterations=30, max_seconds=300)

    screenshot_b64 = await capture_screenshot(page, viewport)
    response = await client.responses.create(
        model="gpt-5.4",
        tools=[{"type": "computer_use_preview", "display_width": viewport["width"],
                "display_height": viewport["height"], "environment": "browser"}],
        input=[{
            "role": "user",
            "content": [
                {"type": "input_text", "text": goal},
                {"type": "input_image", "image_url": f"data:image/png;base64,{screenshot_b64}"},
            ],
        }],
        truncation="auto",
    )

    while True:
        computer_calls = [item for item in response.output if item.type == "computer_call"]

        if not computer_calls:
            # Task done
            final = next((it for it in response.output if it.type == "message"), None)
            return extract_message_text(final) if final else "(no message)"

        # Check stop conditions BEFORE acting
        all_actions = []
        for call in computer_calls:
            all_actions.extend(extract_actions(call))
        stop_reason = state.check_stop_conditions(all_actions)
        if stop_reason:
            print(f"stopping: {stop_reason}")
            return f"[incomplete] {stop_reason}"

        # Execute
        output_items = []
        for call in computer_calls:
            for action_obj in extract_actions(call):
                await execute_action(page, action_obj, viewport)
            await page.wait_for_load_state("domcontentloaded", timeout=5_000)
            screenshot_b64 = await capture_screenshot(page, viewport)
            output_items.append({
                "type": "computer_call_output",
                "call_id": call.call_id,
                "output": {"type": "computer_screenshot",
                           "image_url": f"data:image/png;base64,{screenshot_b64}"},
            })

        # Chain
        response = await client.responses.create(
            model="gpt-5.4",
            previous_response_id=response.id,
            tools=[{"type": "computer_use_preview", "display_width": viewport["width"],
                    "display_height": viewport["height"], "environment": "browser"}],
            input=output_items,
            truncation="auto",
        )


def extract_message_text(msg_item) -> str:
    if not msg_item:
        return ""
    parts = []
    for part in msg_item.content:
        if part.type == "output_text":
            parts.append(part.text)
    return "".join(parts)


# ✅ Alternative — use Conversations for durable state instead of previous_response_id
async def with_conversation(page, goal: str, viewport: dict, conv_id: str = None):
    """Use a Conversation resource so the loop can be resumed across process restarts."""
    from openai import AsyncOpenAI
    client = AsyncOpenAI()

    if not conv_id:
        conv = await client.conversations.create(metadata={"task": "computer_use"})
        conv_id = conv.id

    # First call attaches to the conversation
    screenshot_b64 = await capture_screenshot(page, viewport)
    response = await client.responses.create(
        model="gpt-5.4",
        conversation=conv_id,                  # ← persistent conversation
        tools=[...],
        input=[{
            "role": "user",
            "content": [
                {"type": "input_text", "text": goal},
                {"type": "input_image", "image_url": f"data:image/png;base64,{screenshot_b64}"},
            ],
        }],
    )
    # ... loop continues, always passing conversation=conv_id
    # If the process dies, restart with the same conv_id — history preserved server-side
    return conv_id


# ✅ Progress reporting during long loops
class ProgressReporter:
    def __init__(self):
        self.actions_executed = 0
        self.screenshots_sent = 0

    def report(self):
        elapsed = 0  # calculate from state
        print(f"  [progress] iteration={self.actions_executed} screenshots={self.screenshots_sent}")