OpenAI computer_use Long-Horizon Context Management (Screenshot Inflation)

from openai import AsyncOpenAI

client = AsyncOpenAI()


# ✅ ALWAYS set truncation="auto" on every Response call in a computer_use loop
async def call_with_truncation(previous_response_id: str, input_items: list, viewport):
    return await client.responses.create(
        model="gpt-5.4",
        previous_response_id=previous_response_id,
        tools=[{
            "type": "computer_use_preview",
            "display_width": viewport["width"],
            "display_height": viewport["height"],
            "environment": "browser",
        }],
        input=input_items,
        truncation="auto",                      # ← critical
    )


# ✅ Options for truncation parameter
# "auto"       — server prunes when context pressure builds; retains most-recent
# "disabled"   — never prune; you get 400 if you exceed the window
# omitted      — defaults to "disabled" in some SDK versions; ALWAYS set explicitly


# ✅ What "auto" prunes first
# In practice:
# 1. Oldest screenshots (they're the largest items and often least relevant)
# 2. Old computer_call items
# 3. Old messages
# The most-recent items and the initial task text are preserved as long as possible


# ✅ Monitor when truncation kicks in — usage.input_tokens_details tells you
async def call_and_monitor(prev_id, input_items, viewport):
    resp = await call_with_truncation(prev_id, input_items, viewport)

    usage = resp.usage
    if usage:
        input_tokens = usage.input_tokens
        cached = getattr(usage.input_tokens_details, "cached_tokens", 0) if usage.input_tokens_details else 0
        # If cached fraction is low over time, context isn't stable — rotation may be needed
        if input_tokens > 100_000:
            print(f"[warn] high context: {input_tokens} tokens ({cached} cached)")

    return resp


# ✅ Downscale screenshots for lower-token cost per image
from PIL import Image
import io, base64

def compress_screenshot(raw_png: bytes, max_dim: int = 1024, quality: int = 85) -> str:
    """Downscale to max_dim on longest side, encode as JPEG (~50% smaller)."""
    img = Image.open(io.BytesIO(raw_png))

    # Downscale if larger than max_dim
    if max(img.size) > max_dim:
        img.thumbnail((max_dim, max_dim), Image.LANCZOS)

    # Convert to JPEG for smaller payload
    if img.mode == "RGBA":
        img = img.convert("RGB")

    out = io.BytesIO()
    img.save(out, format="JPEG", quality=quality, optimize=True)
    return base64.b64encode(out.getvalue()).decode()


# ✅ Trade-off: smaller screenshots = fewer tokens but lower detail
# For form-heavy UIs where every input matters, keep quality high (95+).
# For dashboards / lists where visual layout dominates, quality 75-85 works fine.


# ✅ Skip redundant screenshots — only send when state actually changed
class ScreenshotDeduplicator:
    """Send a fresh screenshot only when the page content likely changed."""

    def __init__(self):
        self.last_url = None
        self.last_action_type = None

    def should_screenshot(self, current_url: str, actions: list[dict]) -> bool:
        """Decide if new screenshot is worth sending."""
        # URL changed → definitely screenshot
        if current_url != self.last_url:
            self.last_url = current_url
            return True

        # Any action other than pure wait or noop → screenshot
        action_types = {a.get("type") for a in actions}
        if action_types - {"wait", "screenshot"}:
            return True

        # Only waits — no visual change likely
        return False


# ✅ Batch screenshot capture — take once per batch of actions, not per action
# This is the default in the loop (see #156), but worth reiterating: for a
# computer_call with 3 batched actions, take ONE screenshot at the end.

from dataclasses import dataclass
from openai import AsyncOpenAI

client = AsyncOpenAI()


@dataclass
class SessionState:
    task: str
    current_url: str
    progress: str
    completed_steps: list[str]
    remaining_hint: str
    last_screenshot_b64: str


ROTATION_EVERY_N_ITERATIONS = 15
ROTATION_TOKEN_THRESHOLD = 100_000    # rotate if input_tokens exceeds this


async def summarize_and_rotate(response, page, viewport, task: str) -> SessionState:
    """Ask the model to summarize its progress in text, then start fresh."""

    # Ask for a structured progress summary
    summary_resp = await client.responses.create(
        model="gpt-5.4",
        previous_response_id=response.id,
        input=[{
            "role": "user",
            "content": (
                "SUMMARIZE PROGRESS: Do not take any more actions. Instead, "
                "produce a concise text summary of:\n"
                "1. What steps you have completed so far\n"
                "2. What is currently on screen\n"
                "3. What the next 2-3 steps should be\n"
                "4. Any state I need to know (form values, IDs, etc.)\n\n"
                "Respond in plain text, no actions."
            ),
        }],
        truncation="auto",
    )

    summary_text = extract_text(summary_resp)

    # Take fresh screenshot
    screenshot_b64 = await capture_screenshot(page, viewport)

    return SessionState(
        task=task,
        current_url=page.url,
        progress=summary_text,
        completed_steps=[],           # parsed from summary if you want structure
        remaining_hint=summary_text,
        last_screenshot_b64=screenshot_b64,
    )


async def start_rotated_session(state: SessionState, viewport) -> "Response":
    """Start a fresh Response with the summarized state — no prior context."""
    return await client.responses.create(
        model="gpt-5.4",
        tools=[{
            "type": "computer_use_preview",
            "display_width": viewport["width"],
            "display_height": viewport["height"],
            "environment": "browser",
        }],
        # No previous_response_id — this is a fresh conversation
        input=[{
            "role": "user",
            "content": [
                {"type": "input_text", "text": (
                    f"ORIGINAL TASK: {state.task}\n\n"
                    f"PROGRESS SO FAR:\n{state.progress}\n\n"
                    f"Current URL: {state.current_url}\n\n"
                    "Continue from where the previous session left off. "
                    "Take actions to complete the remaining work."
                )},
                {"type": "input_image",
                 "image_url": f"data:image/jpeg;base64,{state.last_screenshot_b64}"},
            ],
        }],
        truncation="auto",
    )


# ✅ Full loop with rotation
async def long_horizon_loop(page, goal: str, viewport):
    iterations_since_rotation = 0
    response = None

    # Initial call
    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/jpeg;base64,{screenshot_b64}"},
            ],
        }],
        truncation="auto",
    )

    total_iterations = 0
    MAX_TOTAL = 60

    while total_iterations < MAX_TOTAL:
        total_iterations += 1
        iterations_since_rotation += 1

        # Check rotation triggers
        input_tokens = response.usage.input_tokens if response.usage else 0
        should_rotate = (
            iterations_since_rotation >= ROTATION_EVERY_N_ITERATIONS
            or input_tokens >= ROTATION_TOKEN_THRESHOLD
        )

        if should_rotate:
            print(f"[rotate] iter={total_iterations} input_tokens={input_tokens}")
            state = await summarize_and_rotate(response, page, viewport, goal)
            response = await start_rotated_session(state, viewport)
            iterations_since_rotation = 0
            continue

        # Normal loop iteration
        computer_calls = [it for it in response.output if it.type == "computer_call"]
        if not computer_calls:
            return extract_text(response)

        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/jpeg;base64,{screenshot_b64}"},
            })

        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("hit MAX_TOTAL without completion")


def extract_text(resp):
    for item in resp.output:
        if item.type == "message":
            for part in item.content:
                if part.type == "output_text":
                    return part.text
    return ""

def extract_actions(call): pass
async def execute_action(p, a, v): pass
async def capture_screenshot(p, v): pass

from dataclasses import dataclass, field
from typing import Callable, Any


@dataclass
class SubTask:
    name: str
    goal: str
    success_criteria: str
    max_iterations: int = 15


@dataclass
class HandoffState:
    """State passed between sub-tasks."""
    extracted_values: dict[str, Any] = field(default_factory=dict)
    current_url: str = ""
    notes: list[str] = field(default_factory=list)


async def run_sub_task(page, sub_task: SubTask, handoff: HandoffState, viewport) -> HandoffState:
    """Run one sub-task in its own Response chain. Returns updated handoff."""
    from openai import AsyncOpenAI
    client = AsyncOpenAI()

    # Build task input including relevant prior state
    context_lines = [f"SUB-TASK: {sub_task.name}"]
    context_lines.append(f"GOAL: {sub_task.goal}")
    context_lines.append(f"SUCCESS: {sub_task.success_criteria}")

    if handoff.extracted_values:
        context_lines.append("\nCONTEXT FROM PRIOR STEPS:")
        for k, v in handoff.extracted_values.items():
            context_lines.append(f"  {k}: {v}")

    if handoff.notes:
        context_lines.append("\nNOTES:")
        for n in handoff.notes[-5:]:            # last 5 notes only
            context_lines.append(f"  - {n}")

    context_lines.append("\nOn completion, return a JSON object with any values extracted.")

    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": "\n".join(context_lines)},
                {"type": "input_image", "image_url": f"data:image/jpeg;base64,{screenshot_b64}"},
            ],
        }],
        truncation="auto",
    )

    # Run the sub-task's own action loop
    for iteration in range(sub_task.max_iterations):
        computer_calls = [it for it in response.output if it.type == "computer_call"]

        if not computer_calls:
            # Sub-task complete — extract the final message
            final_text = extract_text(response)
            new_handoff = HandoffState(
                extracted_values=dict(handoff.extracted_values),
                current_url=page.url,
                notes=list(handoff.notes) + [f"[{sub_task.name}] {final_text[:200]}"],
            )
            # Attempt to parse structured output
            import json, re
            json_match = re.search(r"\{.*\}", final_text, re.DOTALL)
            if json_match:
                try:
                    parsed = json.loads(json_match.group(0))
                    new_handoff.extracted_values.update(parsed)
                except json.JSONDecodeError:
                    pass

            return new_handoff

        # Normal iteration
        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/jpeg;base64,{screenshot_b64}"},
            })

        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"sub-task {sub_task.name!r} hit max_iterations={sub_task.max_iterations}")


# ✅ Example — a form-filling workflow decomposed into stages
async def onboarding_workflow(page, user_data: dict, viewport):
    sub_tasks = [
        SubTask(
            name="navigate_to_signup",
            goal="Navigate to the signup page from the current landing page.",
            success_criteria="URL contains '/signup' or a signup form is visible.",
            max_iterations=5,
        ),
        SubTask(
            name="fill_personal_info",
            goal=f"Fill in the personal info fields: name={user_data['name']}, "
                 f"email={user_data['email']}. Do NOT click submit.",
            success_criteria="Both fields visibly contain the correct values.",
            max_iterations=8,
        ),
        SubTask(
            name="select_plan",
            goal="Select the 'Starter' plan (do not select paid plans).",
            success_criteria="Starter plan is highlighted/selected.",
            max_iterations=6,
        ),
        SubTask(
            name="submit_and_verify",
            goal="Click submit. Report the confirmation message or any error shown.",
            success_criteria="Confirmation page or error message visible.",
            max_iterations=10,
        ),
    ]

    handoff = HandoffState()

    for st in sub_tasks:
        print(f"\n=== Running sub-task: {st.name} ===")
        handoff = await run_sub_task(page, st, handoff, viewport)
        print(f"  extracted: {handoff.extracted_values}")

    return handoff.extracted_values


def extract_text(resp): pass
def extract_actions(call): pass
async def execute_action(p, a, v): pass
async def capture_screenshot(p, v): pass