Improve OpenAI CUA template logging ergonomics and local runner parity.

This adds CUA-style backend/action event rendering (with JSONL mode support), aligns dotenv/local-run behavior across TypeScript and Python templates, and renames local entry scripts to run_local for clearer usage.

Made-with: Cursor

Author: rgarcia

pkg/templates/python/openai-computer-use/README.md

@@ -11,14 +11,20 @@ You can test against a remote Kernel browser without deploying:
 ```bash
 cp .env.example .env
 # Fill in OPENAI_API_KEY and KERNEL_API_KEY in .env
-uv run test_local.py
+uv run run_local.py
+# JSONL event output
+uv run run_local.py --output jsonl
 ```
 
+The local runner defaults to concise CUA-style logs (`text`), including `kernel>` backend SDK call lines with elapsed timing and `agent>` model output lines. Use `--output jsonl` for one structured event per line (including backend events). Add `--debug` to include verbose in-flight events.
+
 ## Deploy to Kernel
 
 ```bash
 kernel deploy main.py --env-file .env
 kernel invoke python-openai-cua cua-task -p '{"task":"go to https://news.ycombinator.com and list top 5 articles"}'
+# JSONL logs for invocation
+kernel invoke python-openai-cua cua-task -p '{"task":"go to https://news.ycombinator.com and list top 5 articles","output":"jsonl"}'
 ```
 
 See the [docs](https://www.kernel.sh/docs/quickstart) for more information.

pkg/templates/python/openai-computer-use/agent/agent.py

@@ -1,5 +1,6 @@
 import json
-from typing import Callable
+import time
+from typing import Any, Callable
 from computers.kernel_computer import KernelComputer
 from utils import (
     create_response,
@@ -80,13 +81,15 @@ def __init__(
         self.print_steps = True
         self.debug = False
         self.show_images = False
+        self.on_event: Callable[[dict], None] | None = None
+        self._model_request_started_at: float | None = None
         self.acknowledge_safety_check_callback = acknowledge_safety_check_callback
 
         if computer:
             dimensions = computer.get_dimensions()
             self.tools += [
                 {
-                    "type": "computer-preview",
+                    "type": "computer_use_preview",
                     "display_width": dimensions[0],
                     "display_height": dimensions[1],
                     "environment": computer.get_environment(),
@@ -126,6 +129,86 @@ def debug_print(self, *args):
         if self.debug:
             pp(*args)
 
+    def _emit_event(self, event: str, data: dict | None = None) -> None:
+        if self.print_steps and self.on_event:
+            self.on_event({"event": event, "data": data or {}})
+
+    def _current_model_elapsed_ms(self) -> int | None:
+        if self._model_request_started_at is None:
+            return None
+        return int((time.time() - self._model_request_started_at) * 1000)
+
+    def _extract_reasoning_text(self, item: dict[str, Any]) -> str:
+        summary = item.get("summary")
+        if not isinstance(summary, list):
+            return ""
+        pieces: list[str] = []
+        for part in summary:
+            if not isinstance(part, dict):
+                continue
+            text = part.get("text")
+            if isinstance(text, str) and text:
+                pieces.append(text)
+        return " ".join(pieces).strip()
+
+    def _extract_prompt_text(self, item: dict[str, Any]) -> str | None:
+        if item.get("role") != "user":
+            return None
+        content = item.get("content")
+        if isinstance(content, str):
+            return content
+        if not isinstance(content, list):
+            return None
+        parts: list[str] = []
+        for entry in content:
+            if not isinstance(entry, dict):
+                continue
+            text = entry.get("text")
+            if isinstance(text, str) and text:
+                parts.append(text)
+        return " ".join(parts) if parts else None
+
+    def _describe_action(self, action_type: str, action_args: dict[str, Any]) -> str:
+        if action_type == "click":
+            x = int(action_args.get("x", 0))
+            y = int(action_args.get("y", 0))
+            button = action_args.get("button", "left")
+            if button in ("", "left"):
+                return f"click({x}, {y})"
+            return f"click({x}, {y}, {button})"
+        if action_type == "double_click":
+            return f"double_click({int(action_args.get('x', 0))}, {int(action_args.get('y', 0))})"
+        if action_type == "type":
+            text = str(action_args.get("text", ""))
+            if len(text) > 60:
+                text = f"{text[:57]}..."
+            return f"type({text!r})"
+        if action_type == "keypress":
+            keys = action_args.get("keys", [])
+            return f"keypress({keys})"
+        if action_type == "scroll":
+            return (
+                f"scroll({int(action_args.get('x', 0))}, {int(action_args.get('y', 0))}, "
+                f"dx={int(action_args.get('scroll_x', 0))}, dy={int(action_args.get('scroll_y', 0))})"
+            )
+        if action_type == "move":
+            return f"move({int(action_args.get('x', 0))}, {int(action_args.get('y', 0))})"
+        if action_type == "drag":
+            return "drag(...)"
+        if action_type == "wait":
+            return f"wait({int(action_args.get('ms', 1000))}ms)"
+        if action_type == "screenshot":
+            return "screenshot()"
+        return action_type
+
+    def _describe_batch_actions(self, actions: list[dict[str, Any]]) -> str:
+        pieces: list[str] = []
+        for action in actions:
+            action_type = str(action.get("type", "unknown"))
+            action_args = {k: v for k, v in action.items() if k != "type"}
+            pieces.append(self._describe_action(action_type, action_args))
+        return "batch[" + " -> ".join(pieces) + "]"
+
     def _execute_computer_action(self, action_type, action_args):
         if action_type == "click":
             self.computer.click(**action_args)
@@ -150,14 +233,50 @@ def _execute_computer_action(self, action_type, action_args):
 
     def handle_item(self, item):
         """Handle each item; may cause a computer action + screenshot."""
+        if item["type"] == "reasoning":
+            text = self._extract_reasoning_text(item)
+            if text:
+                self._emit_event("reasoning_delta", {"text": text})
+
         if item["type"] == "message":
-            if self.print_steps:
-                print(item["content"][0]["text"])
+            if item.get("role") == "assistant":
+                content = item.get("content", [])
+                if isinstance(content, list):
+                    for part in content:
+                        if isinstance(part, dict) and isinstance(part.get("text"), str):
+                            self._emit_event("text_delta", {"text": part["text"]})
+                self._emit_event("text_done", {})
 
         if item["type"] == "function_call":
             name, args = item["name"], json.loads(item["arguments"])
-            if self.print_steps:
-                print(f"{name}({args})")
+            elapsed_ms = self._current_model_elapsed_ms()
+            if name == BATCH_FUNC_NAME:
+                actions = args.get("actions", [])
+                if isinstance(actions, list):
+                    typed_actions = [a for a in actions if isinstance(a, dict)]
+                    payload = {
+                        "action_type": "batch",
+                        "description": self._describe_batch_actions(typed_actions),
+                        "action": {"type": "batch", "actions": typed_actions},
+                    }
+                    if elapsed_ms is not None:
+                        payload["elapsed_ms"] = elapsed_ms
+                    self._emit_event(
+                        "action",
+                        payload,
+                    )
+            else:
+                payload = {
+                    "action_type": name,
+                    "description": f"{name}({json.dumps(args)})",
+                    "action": args,
+                }
+                if elapsed_ms is not None:
+                    payload["elapsed_ms"] = elapsed_ms
+                self._emit_event(
+                    "action",
+                    payload,
+                )
 
             if name == BATCH_FUNC_NAME:
                 return self._handle_batch_call(item["call_id"], args)
@@ -177,12 +296,26 @@ def handle_item(self, item):
             action = item["action"]
             action_type = action["type"]
             action_args = {k: v for k, v in action.items() if k != "type"}
-            if self.print_steps:
-                print(f"{action_type}({action_args})")
+            elapsed_ms = self._current_model_elapsed_ms()
+            payload = {
+                "action_type": action_type,
+                "description": self._describe_action(action_type, action_args),
+                "action": action,
+            }
+            if elapsed_ms is not None:
+                payload["elapsed_ms"] = elapsed_ms
+            self._emit_event(
+                "action",
+                payload,
+            )
 
             self._execute_computer_action(action_type, action_args)
 
             screenshot_base64 = self.computer.screenshot()
+            self._emit_event(
+                "screenshot",
+                {"captured": True, "bytes_base64": len(screenshot_base64)},
+            )
             if self.show_images:
                 show_image(screenshot_base64)
 
@@ -228,31 +361,55 @@ def _handle_batch_call(self, call_id, args):
         ]
 
     def run_full_turn(
-        self, input_items, print_steps=True, debug=False, show_images=False
+        self,
+        input_items,
+        print_steps=True,
+        debug=False,
+        show_images=False,
+        on_event: Callable[[dict], None] | None = None,
     ):
         self.print_steps = print_steps
         self.debug = debug
         self.show_images = show_images
+        self.on_event = on_event
         new_items = []
+        turns = 0
 
-        while new_items[-1].get("role") != "assistant" if new_items else True:
-            self.debug_print([sanitize_message(msg) for msg in input_items + new_items])
+        for message in input_items:
+            if isinstance(message, dict):
+                prompt = self._extract_prompt_text(message)
+                if prompt:
+                    self._emit_event("prompt", {"text": prompt})
 
-            response = create_response(
-                model=self.model,
-                input=input_items + new_items,
-                tools=self.tools,
-                truncation="auto",
-                instructions=BATCH_INSTRUCTIONS,
-            )
-            self.debug_print(response)
+        try:
+            while new_items[-1].get("role") != "assistant" if new_items else True:
+                turns += 1
+                self.debug_print([sanitize_message(msg) for msg in input_items + new_items])
+
+                self._model_request_started_at = time.time()
+                response = create_response(
+                    model=self.model,
+                    input=input_items + new_items,
+                    tools=self.tools,
+                    truncation="auto",
+                    instructions=BATCH_INSTRUCTIONS,
+                )
+                self.debug_print(response)
+
+                if "output" not in response:
+                    if self.debug:
+                        print(response)
+                    raise ValueError("No output from model")
 
-            if "output" not in response and self.debug:
-                print(response)
-                raise ValueError("No output from model")
-            else:
                 new_items += response["output"]
                 for item in response["output"]:
                     new_items += self.handle_item(item)
+                self._model_request_started_at = None
+                self._emit_event("turn_done", {"turn": turns})
+        except Exception as exc:
+            self._model_request_started_at = None
+            self._emit_event("error", {"message": str(exc)})
+            raise
 
+        self._emit_event("run_complete", {"turns": turns})
         return new_items