feat(codex): add Android runner with managed ChatGPT sign-in

README.md

@@ -297,6 +297,15 @@ ARTEMIS supports two execution profiles tailored for different automation requir
 * **Flash Profile (`--profile flash`)**: Fast and token-efficient reactive loop (~3–5s per step): one model observes the live screen, thinks, and acts, with no graph orchestration. Ideal for routine, deterministic UI tasks. The loop is unbounded by default (`agent.flash.max_turns`, 0 = unlimited) because history is compressed rather than capped: Flash shares the Pro session transcript ledger (session-relative `T+mm:ss` clock, screenshots folded into visual summaries, older steps chunked into eras and recallable on demand via `search_history` / `replay_steps`) and can query the session recording through `video_analyzer`. Transient UI (auto-fading control bars, toasts) is handled by chaining taps into one `click_sequence`. *Limitations*: No task plan or notes, no pre-execution safety net, no checkpoint verification or final report, and no ADB shell.
 * **Pro Profile (`--profile pro`)**: A planning and verification workflow (~15–40s per step), built as a multi-agent graph. A **Planner** maintains a living Markdown task plan with milestones and `verify` / `assert` check items; the **Operator** executes it with the full toolset (Explorer grounding whose `flash` / `pro` / `ultra` tier is a user setting per profile — `pro.explorer.mode` / `flash.explorer_mode` in `config/artemis.jsonc` or `--explorer-pro-mode` — never chosen by the agent; notes, history recall, video analysis, ADB diagnostics). Every single action passes a pre-execution **Safety Net** (XML-first, pixel fallback), while multi-action **fast-action bursts** fire back to back to beat turn latency on transient UI. A blocked or failed action opens an **execution incident** that stays in the Operator's context until a later action succeeds, so recovery is handled by the Operator itself with no separate repair agent. A read-only **Checker** verifies plan checkpoints and runs an exit final review against the original goal (`--verification-level`: `off` / `final` (default) / `checkpoints` / `strict`), and plan milestone edits get an advisory review. Handles 100+ step long-horizon workflows, `[Loop:continuous]` monitoring, and an optional written report.
 
+## Codex ChatGPT login (experimental)
+
+Run Android tasks using an existing Codex ChatGPT login, without a model API key
+in ARTEMIS: `uv run artemis codex login`, then
+`uv run artemis codex run "Open Settings and read the battery percentage" --serial YOUR_DEVICE_SERIAL`.
+This separate CLI runner uses the Accessibility Helper and Codex's official
+app-server; Flash/Pro and the Web UI retain their existing configuration.
+See [setup, supported tools, and limitations](docs/codex.md).
+
 ## Roadmap
 
 - [ ] **Android Studio Integration**: Native IDE plugin and workflow integration to enable in-editor debugging, test recording, and automated device control directly within Android Studio.

artemis/integrations/__init__.py

@@ -0,0 +1 @@
+"""Optional integrations with external agent runtimes."""

artemis/integrations/codex/__init__.py

@@ -0,0 +1 @@
+"""Codex-managed ChatGPT authentication and Android automation."""

artemis/integrations/codex/client.py

@@ -0,0 +1,153 @@
+"""Small stdio client for the official Codex app-server protocol.
+
+Authentication stays inside Codex: never read auth.json, extract tokens, or
+send subscription credentials to an OpenAI-compatible API endpoint.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections import deque
+from contextlib import suppress
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+from typing import Any
+
+
+class CodexError(RuntimeError):
+    """An actionable Codex installation, authentication, or protocol error."""
+
+
+class CodexClient:
+    """One owned app-server process; requests are intentionally serialized."""
+
+    def __init__(self, cwd: Path, executable: str = "codex", timeout: float = 30):
+        self.cwd = cwd
+        self.executable = executable
+        self.timeout = timeout
+        self.process: asyncio.subprocess.Process | None = None
+        self._stderr: asyncio.Task | None = None
+        self._next_id = 0
+        self._events: deque[dict[str, Any]] = deque()
+
+    async def __aenter__(self) -> CodexClient:
+        executable = shutil.which(self.executable)
+        if not executable:
+            raise CodexError("Codex CLI not found. Install Codex and put its executable on PATH.")
+        # Avoid accidental API billing when this explicit ChatGPT integration is used.
+        env = os.environ.copy()
+        for key in ("OPENAI_API_KEY", "CODEX_API_KEY"):
+            env.pop(key, None)
+        flags = subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0
+        self.process = await asyncio.create_subprocess_exec(
+            executable,
+            "app-server",
+            "--listen",
+            "stdio://",
+            cwd=self.cwd,
+            env=env,
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            limit=16 * 1024 * 1024,
+            creationflags=flags,
+        )
+        self._stderr = asyncio.create_task(self._drain_stderr())
+        initialized = False
+        try:
+            await self.request(
+                "initialize",
+                {
+                    "clientInfo": {"name": "artemis", "version": "1.0"},
+                    "capabilities": {"experimentalApi": True},
+                },
+            )
+            await self.send({"method": "initialized", "params": {}})
+            initialized = True
+            return self
+        finally:
+            if not initialized:
+                await self.close()
+
+    async def __aexit__(self, *_args) -> None:
+        await self.close()
+
+    async def _drain_stderr(self) -> None:
+        assert self.process and self.process.stderr
+        # Do not persist authentication diagnostics or unbounded subprocess logs.
+        while await self.process.stderr.read(8192):
+            pass
+
+    async def close(self) -> None:
+        if self.process and self.process.returncode is None:
+            self.process.terminate()
+            try:
+                await asyncio.wait_for(self.process.wait(), 5)
+            except TimeoutError:
+                self.process.kill()
+                await self.process.wait()
+        if self._stderr:
+            self._stderr.cancel()
+            with suppress(asyncio.CancelledError):
+                await self._stderr
+
+    async def send(self, payload: dict[str, Any]) -> None:
+        assert self.process and self.process.stdin
+        try:
+            self.process.stdin.write((json.dumps(payload) + "\n").encode())
+            await self.process.stdin.drain()
+        except (BrokenPipeError, ConnectionResetError) as exc:
+            raise CodexError("Codex app-server closed its input stream.") from exc
+
+    async def _read(self) -> dict[str, Any]:
+        assert self.process and self.process.stdout
+        line = await self.process.stdout.readline()
+        if not line:
+            raise CodexError("Codex app-server exited before completing the request.")
+        try:
+            value = json.loads(line)
+        except (ValueError, UnicodeDecodeError) as exc:
+            raise CodexError("Invalid JSON from Codex app-server.") from exc
+        if not isinstance(value, dict):
+            raise CodexError("Invalid message from Codex app-server.")
+        return value
+
+    async def request(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
+        self._next_id += 1
+        request_id = self._next_id
+        await self.send({"id": request_id, "method": method, "params": params})
+        async with asyncio.timeout(self.timeout):
+            while True:
+                message = await self._read()
+                # Server requests have IDs too. Never mistake one for our reply.
+                if "method" not in message and message.get("id") == request_id:
+                    if "error" in message:
+                        raise CodexError(
+                            f"Codex {method}: {message['error'].get('message', 'failed')}"
+                        )
+                    return message.get("result", {})
+                self._events.append(message)
+
+    async def event(self) -> dict[str, Any]:
+        return self._events.popleft() if self._events else await self._read()
+
+    async def reject_request(self, message: dict[str, Any]) -> None:
+        await self.send(
+            {
+                "id": message["id"],
+                "error": {"code": -32601, "message": "Unsupported by the ARTEMIS Codex client"},
+            }
+        )
+
+    async def account(self) -> dict[str, Any]:
+        response = await self.request("account/read", {"refreshToken": False})
+        account = response.get("account") or {}
+        # Deliberately expose no email, account identifier, or credentials.
+        return {"authenticated": account.get("type") == "chatgpt", "auth_type": account.get("type")}
+
+    async def require_chatgpt(self) -> None:
+        if not (await self.account())["authenticated"]:
+            raise CodexError("ChatGPT login required. Run `artemis codex login` first.")

artemis/integrations/codex/device.py

@@ -0,0 +1,175 @@
+"""Validated device tools for a Codex thread, bound to one locked phone."""
+
+from __future__ import annotations
+
+import base64
+from io import BytesIO
+import json
+from pathlib import Path
+import time
+from typing import Literal
+
+from PIL import Image
+from pydantic import BaseModel, ConfigDict, Field
+
+from artemis.clients.accessibility_client import AccessibilityClient
+
+
+class Arguments(BaseModel):
+    model_config = ConfigDict(extra="forbid", strict=True)
+
+
+class Observe(Arguments):
+    pass
+
+
+class Action(Arguments):
+    observation: int = Field(ge=1, description="Observation ID from the latest android_observe.")
+
+
+class Tap(Action):
+    x: int = Field(ge=0)
+    y: int = Field(ge=0)
+
+
+class Swipe(Action):
+    x1: int = Field(ge=0)
+    y1: int = Field(ge=0)
+    x2: int = Field(ge=0)
+    y2: int = Field(ge=0)
+    duration_ms: int = Field(ge=100, le=2000)
+
+
+class TypeText(Action):
+    text: str = Field(min_length=1, max_length=2000)
+
+
+class Key(Action):
+    key: Literal["back", "home", "recents", "notifications", "quick_settings", "enter"]
+
+
+TOOLS = {
+    "android_observe": (
+        Observe,
+        "Read the current Android screenshot and UI tree. Coordinates are pixels.",
+    ),
+    "android_tap": (
+        Tap,
+        "Tap a location in the latest screenshot. Observe again after this action.",
+    ),
+    "android_swipe": (Swipe, "Swipe between pixel coordinates. Observe again after this action."),
+    "android_type": (
+        TypeText,
+        "Append Unicode text to the focused field. Observe again afterwards.",
+    ),
+    "android_key": (Key, "Press an Android navigation key. Observe again afterwards."),
+}
+
+
+def tool_specs() -> list[dict]:
+    return [
+        {
+            "type": "function",
+            "name": name,
+            "description": description,
+            "inputSchema": args.model_json_schema(),
+        }
+        for name, (args, description) in TOOLS.items()
+    ]
+
+
+class DeviceTools:
+    def __init__(self, screen: AccessibilityClient, directory: Path, max_actions: int):
+        self.screen = screen
+        self.directory = directory
+        self.max_actions = max_actions
+        self.actions = 0
+        self.observation = 0
+        self.observed_at = 0.0
+        self.ready = False
+        self.width = self.height = 0
+
+    def call(self, name: str, arguments: dict) -> dict:
+        if name not in TOOLS:
+            raise ValueError(f"Unknown Android tool: {name}")
+        schema, _description = TOOLS[name]
+        args = schema.model_validate(arguments)
+        if isinstance(args, Observe):
+            return self.observe()
+        if not isinstance(args, Action):
+            raise ValueError("Invalid action")
+        if not self.ready or args.observation != self.observation:
+            raise ValueError(
+                "Observe the screen again before acting; observation is stale or consumed."
+            )
+        if time.monotonic() - self.observed_at > 60:
+            self.ready = False
+            raise ValueError("Observation expired. Call android_observe again.")
+        if self.actions >= self.max_actions:
+            raise ValueError("Action limit reached. Stop and report the remaining work.")
+        if isinstance(args, Tap):
+            self._point(args.x, args.y)
+        elif isinstance(args, Swipe):
+            self._point(args.x1, args.y1)
+            self._point(args.x2, args.y2)
+        self.ready = False
+        self.actions += 1
+        if isinstance(args, Tap):
+            success = self.screen.tap(args.x, args.y)
+        elif isinstance(args, Swipe):
+            success = self.screen.swipe(args.x1, args.y1, args.x2, args.y2, args.duration_ms)
+        elif isinstance(args, TypeText):
+            success = self.screen.send_text(args.text)
+        elif isinstance(args, Key):
+            success = self.screen.press_key(args.key)
+        else:
+            raise ValueError("Unsupported action")
+        return {
+            "success": bool(success),
+            "contentItems": [
+                {
+                    "type": "inputText",
+                    "text": json.dumps({"executed": bool(success), "observe_next": True}),
+                }
+            ],
+        }
+
+    def _point(self, x: int, y: int) -> None:
+        if x >= self.width or y >= self.height:
+            raise ValueError(f"Coordinates outside the {self.width}x{self.height} screenshot.")
+
+    def observe(self) -> dict:
+        self.ready = False
+        data = self.screen.get_screen_data()
+        raw = base64.b64decode(data.base64, validate=True)
+        with Image.open(BytesIO(raw)) as image:
+            self.width, self.height = image.size
+            buffer = BytesIO()
+            image.save(buffer, format="PNG")
+            png = buffer.getvalue()
+        self.observation += 1
+        stem = f"observation-{self.observation:04d}"
+        (self.directory / f"{stem}.png").write_bytes(png)
+        (self.directory / f"{stem}.xml").write_text(data.hierarchy_xml, encoding="utf-8")
+        self.ready = True
+        self.observed_at = time.monotonic()
+        description = json.dumps(
+            {
+                "observation": self.observation,
+                "width": self.width,
+                "height": self.height,
+                "hierarchy_xml": data.hierarchy_xml[:60000],
+                "hierarchy_truncated": len(data.hierarchy_xml) > 60000,
+            },
+            ensure_ascii=False,
+        )
+        return {
+            "success": True,
+            "contentItems": [
+                {"type": "inputText", "text": description},
+                {
+                    "type": "inputImage",
+                    "imageUrl": "data:image/png;base64," + base64.b64encode(png).decode(),
+                },
+            ],
+        }