diff --git a/docs/engineering/coding_agent_android_testing.md b/docs/engineering/coding_agent_android_testing.md new file mode 100644 index 00000000..8d9c19e0 --- /dev/null +++ b/docs/engineering/coding_agent_android_testing.md @@ -0,0 +1,132 @@ +# Optional Android testing for coding-agent instances + +This is a **candidate follow-on** to the prepared instance package. It is not +production enabled or accepted yet. It must not delay the core package. + +The intended job is native Chrome/Gboard investigation without a connected +phone. The simplest baseline is the existing operator-owned Mac SDK/image, +with new disposable AVD data for each task/run. No emulator or account is +created just because a coding instance is provisioned. + +## Authority and routes + +`coding_agent_android` uses the existing `VON_CODING_AGENT_REGISTRY`. An +instance can have an optional `android` entry with a fixed `provisioner_command`, +`host` and `unix_account`. The test host can differ from the coding host. The +service accepts only the authenticated enrolled agent or delegator, in the +exact organisation, with current membership and a current task created by the +enrolled delegator and assigned to that agent. No client actor/org/path/command +is accepted. Reserved coding identities can use this capability without +reprovisioning their worker. + +The fixed command invokes `scripts/codex_von_android.py --binding PRIVATE_JSON`. +The binding uses the core provisioner's host/account/identity validation and +adds `android`: `platform` (OS/architecture pair), `sdk_root`, `system_image` +(relative to SDK), `abi`, `arch`, `state_root`, `ram_mib`, `cores`, `adb_port`, +`emulator_port`, `host_capacity`, `capacity_enrolment_verified`, and a map of +named `fixtures` to operator-approved disposable URLs. Defaults bound failed +boot occupancy to 600 seconds and idle occupancy to 1800 seconds; legitimate +operations renew idle activity. These protect shared test-host resources, not +whole coding-task duration. + +The SDK and image are immutable shared inputs. Mutable AVD data and control +sockets are private to the prepared account/binding and task/run. All test-host +consumers must use the same operator-owned capacity inode; the core package's +single-slot admission serialises consumers. Guest RAM/CPU are explicit launch +arguments; headroom includes host overhead. A reserve check runs during boot +and use. This is not a hard total-RSS guarantee or a defence against arbitrary +same-account processes. Separate untrusted tenants require prepared accounts +and suitable OS confinement. + +For a Mac without incoming SSH, `codex_von_android_transport.py` supports a +private Unix listener and client. An operator can use SSH StreamLocal reverse +forwarding from the Mac to a private DGX socket. Both parent directories must be +owner-only. The DGX registry binds the client command to that exact socket; the +Mac listener binds a fixed host configuration. No TCP listener, remote-login +setting, credentials in model context or extra coding worker is needed. The +trusted service/operator owns the socket. Do not expose it directly to an +untrusted model process, since the host trusts the canonical service's prior +authorisation. Transport lifecycle remains operator-managed; this candidate +does not install another scheduler or silently enable a persistent tunnel. + +## Operations and evidence + +Call `discover` with instance/task IDs. It returns platform, tool versions, +acceleration and configured budgets; it is not proof of available capacity or +a running emulator. `start` also requires a stable `run_id`. Repeating an active +or terminal run returns retained state, rather than launching again. Read +`status` until ready, waiting for capacity, or stopped. A fresh attempt after a +terminal run uses a different run ID, after inspecting retained evidence. + +Ready sessions support native `tap`, bounded plain ASCII `text`, +`open_fixture` by enrolled name, `screenshot`, `report`, and `stop`. Screenshot +and run-report bytes travel to the controller, which checks the digest, +rechecks canonical assignment, and uses canonical task attachment services. +Reports identify actual Android/build/Chrome/Gboard versions and native-emulator +coverage. They do not assert physical-device behaviour. The older Android17 / +Chrome145 / Gboard17.2 evidence does not establish Chrome152 acceptance. + +`android_instance_id` in a worker configuration exposes the optional capability +in task context. The existing controller polls a task/run-specific request file +inside the coding workspace and writes responses there. The coding child gets +no Mongo credential or general host command. Requests cannot select another +task, actor or host. Private controller receipts deduplicate requests; an +interrupted effect remains uncertain and is not automatically repeated. No +additional worker or poller is installed. Task authority is rechecked by the +canonical service for each operation. Mailbox reads/writes reject symlink +redirection from the coding workspace. + +The current single-slot capacity baseline means a coding worker cannot acquire +a second slot for an emulator on its own host while retaining the first. The +intended DGX-to-Mac route uses separate host budgets. Same-host concurrent +coding/emulation needs explicit shared reservation support before activation. + +## Current delivery decision and evidence + +The bounded emulator/controller capability passed live acceptance on 13 September +2026. Core PR #682 merged on 17 September 2026; this separate follow-on +now integrates the accepted core and current task timing/retry changes. No +production Android registry or emulator is enabled by publication. + +131 targeted integration tests pass, covering worker/inbox/provisioner regressions, task and +organisation denial, duplicate requests, retained uncertain effects, shared +capacity contention and symlink-safe controller mailboxes. + +The first admission was correctly refused at 10.17 GiB estimated available +versus the unchanged 14 GiB requirement (8 GiB reserve plus 6 GiB launch headroom). +After Michael requested continuation and headroom reached 19.18 GiB, a fresh +bridge attempt ran live acceptance. The ADB build rejected a hostname-qualified +`-L` endpoint; the candidate now uses its supported `-P` private port with default +loopback binding. Per-run host logs survive disposable AVD cleanup. + +Verified route: this coding agent wrote task-bound requests to the existing +controller handler on DGX, under Codex VS Code's canonical task authority; that +handler invoked the enrolled service through a private SSH StreamLocal relay to +the Mac. No second coding worker or model was started. The normal worker polling +hook is covered by tests; it was not deployed to the production DGX worker. + +Live run `native-acceptance-03` verified: + +- Android17 / SDK37 / build CE2A.260420.019, Chrome145.0.7632.218, + Gboard17.2.2.895242737, 3 GiB guest RAM and two cores; +- native Chrome navigation, text entry and a Gboard key tap; repeating the same + key-tap request returned the same receipt and inserted only one character; +- repeating start retained the same emulator PID; competing admission was denied; +- screenshot and report attachments through canonical task services; all three + downloaded byte hashes matched their canonical source hashes; +- another organisation context was denied before access to the host; +- supervisor interruption terminated emulator and ADB, removed mutable AVD data + and the control socket, and released capacity; repeating terminal start did + not relaunch; post-cleanup available memory was 17.43 GiB. + +Canonical task evidence: + +- `#V#computer_file_copy_8daf6dd0ef6c4a59a64db9eddb30819a` (initial screenshot) +- `#V#computer_file_copy_98e796f288344301b52d685109593166` (native keyboard screenshot) +- `#V#computer_file_copy_08b649de8635494fb152e7103190c67d` (environment/action report) + +The report records host operations, not browser DOM input-event telemetry or a +claim that image pasting is fixed. Chrome152 and physical-device coverage remain +outside this acceptance. For release, reconcile this follow-on against the +accepted core package and run applicable CI. Do not repeat the completed native +acceptance merely because a controller output/comment changed. diff --git a/docs/engineering/coding_agent_instances.md b/docs/engineering/coding_agent_instances.md index 732d9d11..1f32788a 100644 --- a/docs/engineering/coding_agent_instances.md +++ b/docs/engineering/coding_agent_instances.md @@ -176,3 +176,10 @@ shared-account execution is not a security sandbox between users. No permanent additional consumer, production registry, or public deployment was activated. Native task `#V#task_agent_002b293c463ed8e1e52df23cb68f9612` retains the substantive acceptance receipt and its environment/revision provenance. + +## Optional native Android testing + +The separate [Android test-host candidate](coding_agent_android_testing.md) reuses +instance bindings and shared admission. It is optional and is not a release gate +for this core package. Read its current evidence and remaining acceptance before +claiming that a coding instance can use it autonomously. diff --git a/scripts/codex_von_android.py b/scripts/codex_von_android.py new file mode 100644 index 00000000..1bd90872 --- /dev/null +++ b/scripts/codex_von_android.py @@ -0,0 +1,595 @@ +"""Optional disposable Android sessions behind a fixed operator host binding. + +Only the trusted controller may invoke this executable. Requests never select +host paths, SDKs, accounts or shell commands. Same-account Unix access is not a +sandbox; separate untrusted tenants need separate prepared accounts. +""" + +from __future__ import annotations + +import argparse +import base64 +import fcntl +import hashlib +import json +import os +import platform +import re +import shutil +import signal +import socket +import subprocess +import sys +import time +from pathlib import Path + +try: + from .codex_von_instance import binding + from .codex_von_capacity import admission, available_memory_bytes +except ImportError: + from codex_von_instance import binding + from codex_von_capacity import admission, available_memory_bytes + +IDENTITY = ( + "instance_id", + "agent_id", + "delegator_id", + "organisation_id", + "host", + "unix_account", +) + + +def write(path, value): + temporary = path.with_suffix(".tmp") + temporary.write_text(json.dumps(value, indent=2)) + temporary.chmod(0o600) + temporary.replace(path) + + +def validate(spec): + cfg = spec["android"] + if [platform.system(), platform.machine()] != cfg["platform"]: + raise ValueError("Unsupported test-host platform") + if not cfg.get("capacity_enrolment_verified"): + raise ValueError("Enrol test-host consumers in shared capacity first") + if cfg["ram_mib"] < 1024 or cfg["cores"] < 1: + raise ValueError("Invalid emulator resource budget") + if cfg["cores"] > (os.cpu_count() or 1): + raise ValueError("Insufficient host CPU capacity") + if cfg["host_capacity"]["launch_headroom_bytes"] < cfg["ram_mib"] * 1024**2: + raise ValueError("Headroom must include guest RAM and calibrated host overhead") + sdk = Path(cfg["sdk_root"]).resolve() + image = (sdk / cfg["system_image"]).resolve() + image.relative_to(sdk) + for p in ( + sdk / "emulator/emulator", + sdk / "platform-tools/adb", + image / "system.img", + ): + if not p.is_file(): + raise ValueError("Prepared SDK or image unavailable") + root = Path(cfg["state_root"]) + root.mkdir(parents=True, exist_ok=True, mode=0o700) + root.chmod(0o700) + return cfg, root + + +def command(cfg, *args, binary=False): + result = subprocess.run( + [ + str(Path(cfg["sdk_root"]) / "platform-tools/adb"), + "-P", + str(cfg["adb_port"]), + "-s", + "emulator-" + str(cfg["emulator_port"]), + *args, + ], + capture_output=True, + timeout=30, + check=True, + ) + return result.stdout if binary else result.stdout.decode().strip() + + +def ports_free(cfg): + for port in (cfg["adb_port"], cfg["emulator_port"], cfg["emulator_port"] + 1): + with socket.socket() as probe: + probe.bind(("127.0.0.1", port)) + + +def probe(cfg): + sdk = Path(cfg["sdk_root"]) + acceleration = subprocess.run( + [str(sdk / "emulator/emulator"), "-accel-check"], + capture_output=True, + text=True, + timeout=30, + ) + return { + "platform": [platform.system(), platform.machine()], + "acceleration_available": acceleration.returncode == 0, + "emulator_version": subprocess.check_output( + [str(sdk / "emulator/emulator"), "-version"], text=True + ).splitlines()[0], + "adb_version": subprocess.check_output( + [str(sdk / "platform-tools/adb"), "version"], text=True + ).splitlines()[:2], + "ram_mib": cfg["ram_mib"], + "cores": cfg["cores"], + "coverage": "native_android_emulator", + "limitations": [ + "Not physical-device evidence; browser and keyboard versions must be read per run.", + "Disposable data; no enrolled Google account.", + ], + } + + +def current(root): + path = root / "session.json" + return json.loads(path.read_text()) if path.exists() else None + + +def own_session(state, request): + if ( + not state + or state["task_id"] != request["task_id"] + or state["run_id"] != request["run_id"] + ): + raise PermissionError("Session does not belong to this task and run") + + +def live_process(state): + if not state or not state.get("emulator_pid"): + return False + p = subprocess.run( + ["ps", "-p", str(state["emulator_pid"]), "-o", "command="], + text=True, + capture_output=True, + ) + return bool( + p.returncode == 0 and state["avd_name"] in p.stdout and "qemu" in p.stdout + ) + + +def stop_owned(cfg, root, state): + # The dedicated ADB ports are operator-reserved and checked before launch. + # Do not touch another session if the retained process identity does not match. + if live_process(state): + os.kill(state["emulator_pid"], signal.SIGTERM) + for _ in range(100): + if not live_process(state): + break + time.sleep(0.1) + if live_process(state): + raise RuntimeError("Owned emulator did not stop; capacity remains occupied") + adb_pid = state.get("adb_pid") + if adb_pid: + p = subprocess.run( + ["ps", "-p", str(adb_pid), "-o", "command="], text=True, capture_output=True + ) + if p.returncode == 0 and "adb" in p.stdout and str(cfg["adb_port"]) in p.stdout: + os.kill(adb_pid, signal.SIGTERM) + run_root = root / state["avd_name"] + if run_root.is_dir(): + shutil.rmtree(run_root) + (root / "control.sock").unlink(missing_ok=True) + state.update(status="stopped", stopped_at=time.time()) + write(root / "session.json", state) + + +def serve(spec, request): + cfg, root = validate(spec) + state = current(root) + own_session(state, request) + stop_request = root / ("stop-" + request["run_id"] + ".json") + with admission({"host_capacity": cfg["host_capacity"]}) as fds: + if fds is None: + state.update(status="waiting_for_capacity") + write(root / "session.json", state) + return + ports_free(cfg) + run_root = root / state["avd_name"] + run_root.mkdir(mode=0o700) + avds = run_root / "avd" + avds.mkdir() + avd = avds / (state["avd_name"] + ".avd") + avd.mkdir() + config = { + "abi.type": cfg["abi"], + "hw.cpu.arch": cfg["arch"], + "hw.cpu.ncore": cfg["cores"], + "hw.ramSize": cfg["ram_mib"], + "hw.lcd.width": 1080, + "hw.lcd.height": 1920, + "hw.lcd.density": 420, + "hw.keyboard": "no", + "hw.mainKeys": "no", + "hw.gpu.enabled": "yes", + "hw.gpu.mode": "host", + "hw.camera.back": "none", + "hw.camera.front": "none", + "image.sysdir.1": str(Path(cfg["sdk_root"]) / cfg["system_image"]) + "/", + "disk.dataPartition.size": "6G", + "showDeviceFrame": "no", + } + (avd / "config.ini").write_text( + "\n".join(f"{k}={v}" for k, v in config.items()) + ) + (avds / (state["avd_name"] + ".ini")).write_text(f"path={avd}\n") + env = { + **os.environ, + "ANDROID_SDK_ROOT": cfg["sdk_root"], + "ANDROID_AVD_HOME": str(avds), + "ANDROID_ADB_SERVER_PORT": str(cfg["adb_port"]), + } + env.pop("ANDROID_SERIAL", None) + log = (root / (state["avd_name"] + "-host.log")).open("ab") + adb = emulator = None + + def interrupted(*_): + raise InterruptedError("Test-host session interrupted") + + signal.signal(signal.SIGTERM, interrupted) + signal.signal(signal.SIGINT, interrupted) + try: + adb = subprocess.Popen( + [ + str(Path(cfg["sdk_root"]) / "platform-tools/adb"), + "-P", + str(cfg["adb_port"]), + "nodaemon", + "server", + ], + env=env, + stdout=log, + stderr=log, + pass_fds=fds, + ) + state["adb_pid"] = adb.pid + write(root / "session.json", state) + time.sleep(1) + if adb.poll() is not None: + raise RuntimeError("Private ADB server failed") + emulator = subprocess.Popen( + [ + str(Path(cfg["sdk_root"]) / "emulator/emulator"), + "-avd", + state["avd_name"], + "-memory", + str(cfg["ram_mib"]), + "-cores", + str(cfg["cores"]), + "-port", + str(cfg["emulator_port"]), + "-no-snapshot", + "-no-audio", + "-no-boot-anim", + "-gpu", + "host", + ], + env=env, + stdout=log, + stderr=log, + pass_fds=fds, + ) + state["emulator_pid"] = emulator.pid + write(root / "session.json", state) + # Protect indefinite failed boot; this bounds occupancy of the test host. + deadline = time.monotonic() + cfg.get("boot_timeout_seconds", 600) + while time.monotonic() < deadline: + if stop_request.exists(): + raise InterruptedError("Task requested stop during boot") + if available_memory_bytes() < cfg["host_capacity"]["reserve_bytes"]: + raise RuntimeError("Host memory reserve crossed during boot") + if emulator.poll() is not None: + raise RuntimeError("Emulator exited before boot") + try: + if command(cfg, "shell", "getprop", "sys.boot_completed") == "1": + break + except subprocess.SubprocessError: + pass + time.sleep(2) + else: + raise RuntimeError("Emulator boot exceeded host occupancy bound") + state.update(status="ready", environment=probe(cfg)) + state["environment"].update( + android=command(cfg, "shell", "getprop", "ro.build.version.release"), + sdk=command(cfg, "shell", "getprop", "ro.build.version.sdk"), + build=command(cfg, "shell", "getprop", "ro.build.id"), + ) + for name, package in [ + ("chrome", "com.android.chrome"), + ("gboard", "com.google.android.inputmethod.latin"), + ]: + versions = command(cfg, "shell", "dumpsys", "package", package) + state["environment"][name] = re.findall( + r"versionName=([^\s]+)", versions + ) + write(root / "session.json", state) + endpoint = root / "control.sock" + endpoint.unlink(missing_ok=True) + with socket.socket(socket.AF_UNIX) as listener: + listener.bind(str(endpoint)) + endpoint.chmod(0o600) + listener.listen(2) + listener.settimeout(1) + active = time.monotonic() + while emulator.poll() is None: + if stop_request.exists(): + break + if available_memory_bytes() < cfg["host_capacity"]["reserve_bytes"]: + state["stop_reason"] = "host_memory_reserve_crossed" + break + if time.monotonic() - active > cfg.get( + "idle_timeout_seconds", 1800 + ): + break + try: + client, _ = listener.accept() + except socket.timeout: + continue + with client: + client.settimeout(5) + data = client.recv(65537) + try: + req = json.loads(data) + own_session(state, req) + result = operate(cfg, state, req) + active = time.monotonic() + client.sendall(json.dumps(result).encode()) + if req["action"] == "stop": + break + except Exception as exc: + try: + client.sendall( + json.dumps({"error": type(exc).__name__}).encode() + ) + except OSError: + pass + finally: + signal.signal(signal.SIGTERM, signal.SIG_IGN) + signal.signal(signal.SIGINT, signal.SIG_IGN) + for child in (emulator, adb): + if child and child.poll() is None: + child.terminate() + try: + child.wait(timeout=20) + except subprocess.TimeoutExpired: + child.kill() + child.wait() + (root / "control.sock").unlink(missing_ok=True) + state.update(status="stopped", stopped_at=time.time()) + write(root / "session.json", state) + shutil.rmtree(run_root) + log.close() + + +def operate(cfg, state, req): + action = req["action"] + if action == "status" or action == "stop": + return dict(state) + if action == "report": + raw = json.dumps(state, indent=2).encode() + return { + **state, + "artifact": { + "filename": "android-run-report.json", + "media_type": "application/json", + "sha256": hashlib.sha256(raw).hexdigest(), + "data_base64": base64.b64encode(raw).decode(), + }, + } + if action == "screenshot": + raw = command(cfg, "exec-out", "screencap", "-p", binary=True) + return { + **state, + "artifact": { + "filename": "android-screen.png", + "media_type": "image/png", + "sha256": hashlib.sha256(raw).hexdigest(), + "data_base64": base64.b64encode(raw).decode(), + }, + } + if action == "tap": + x, y = req["x"], req["y"] + if ( + type(x) is not int + or type(y) is not int + or not (0 <= x < 1080 and 0 <= y < 1920) + ): + raise ValueError("Tap outside display") + command(cfg, "shell", "input", "tap", str(x), str(y)) + elif action == "text": + value = req["text"] + # adb shell joins arguments; only a literal, safely quoted text value. + if ( + not isinstance(value, str) + or len(value) > 2000 + or not re.fullmatch(r"[a-zA-Z0-9 .,!?@_\-]+", value) + ): + raise ValueError( + "Native text input supports plain ASCII letters/digits/punctuation; use fixture tools for richer input" + ) + command(cfg, "shell", "input", "text", "'" + value.replace(" ", "%s") + "'") + elif action == "open_fixture": + url = cfg["fixtures"][req["fixture"]] + if not re.fullmatch(r"https?://[a-zA-Z0-9.:/_?=&%\-]+", url): + raise ValueError("Invalid operator fixture URL") + command( + cfg, + "shell", + "am", + "start", + "-a", + "android.intent.action.VIEW", + "-d", + "'" + url + "'", + "com.android.chrome", + ) + else: + raise ValueError("Unsupported Android action") + state.setdefault("actions", []).append( + {"action": action, "observed_at": time.time()} + ) + state["actions"] = state["actions"][-200:] + return {**state, "performed": action} + + +def lifecycle(spec, request, binding_path): + cfg, root = validate(spec) + identity = {key: spec[key] for key in IDENTITY} + allowed = {"action", "task_id", "run_id", "x", "y", "text", "fixture"} + if request.get("action") not in { + "discover", + "start", + "status", + "stop", + "screenshot", + "report", + "tap", + "text", + "open_fixture", + }: + raise ValueError("Unsupported Android action") + if set(request) - allowed: + raise ValueError("Unrecognised Android request fields") + if request["action"] == "discover": + return { + **identity, + "status": "available", + **probe(cfg), + "fixtures": list(cfg.get("fixtures", {})), + } + if not isinstance(request.get("task_id"), str) or not request["task_id"].startswith( + "#V#" + ): + raise ValueError("Canonical task required") + if not re.fullmatch(r"[a-zA-Z0-9_-]{1,64}", request.get("run_id", "")): + raise ValueError("Stable run ID required") + with (root / "operation.lock").open("a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + state = current(root) + if request["action"] == "start": + if state and state["status"] not in {"stopped", "waiting_for_capacity"}: + own_session(state, request) + return { + **identity, + **state, + } # lost acknowledgement never launches twice + if ( + state + and state["run_id"] == request["run_id"] + and state["status"] == "stopped" + ): + own_session(state, request) + return { + **identity, + **state, + } # explicit new run needed after termination + digest = hashlib.sha256( + ( + spec["organisation_id"] + request["task_id"] + request["run_id"] + ).encode() + ).hexdigest()[:24] + state = { + "task_id": request["task_id"], + "run_id": request["run_id"], + "avd_name": "von_" + digest, + "status": "starting", + "started_at": time.time(), + } + write(root / "session.json", state) + log = (root / "supervisor.log").open("ab") + child = subprocess.Popen( + [ + sys.executable, + str(Path(__file__).resolve()), + "--binding", + str(binding_path), + "--serve", + ], + stdin=subprocess.PIPE, + stdout=log, + stderr=log, + start_new_session=True, + ) + child.stdin.write(json.dumps(request).encode()) + child.stdin.close() + state["supervisor_pid"] = child.pid + # Supervisor records emulator state; do not overwrite its concurrent receipt. + (root / "supervisor.pid").write_text(str(child.pid)) + log.close() + return {**identity, **state} + own_session(state, request) + endpoint = root / "control.sock" + if request["action"] == "stop": + write(root / ("stop-" + request["run_id"] + ".json"), request) + if not endpoint.exists(): + if request["action"] == "stop": + pid_path = root / "supervisor.pid" + if pid_path.exists(): + pid = int(pid_path.read_text()) + process = subprocess.run( + ["ps", "-p", str(pid), "-o", "command="], + text=True, + capture_output=True, + ) + if ( + "--serve" in process.stdout + and str(binding_path) in process.stdout + ): + os.kill(pid, signal.SIGTERM) + return {**identity, **state, "status": "stopping"} + stop_owned(cfg, root, state) + return {**identity, **state, "supervisor_ready": False} + with socket.socket(socket.AF_UNIX) as client: + client.settimeout(40) + try: + client.connect(str(endpoint)) + except (ConnectionRefusedError, FileNotFoundError): + if request["action"] == "stop": + stop_owned(cfg, root, state) + return {**identity, **state} + return { + **identity, + **state, + "supervisor_ready": False, + "recovery": "Controller disconnected; request stop before a new run", + } + client.sendall(json.dumps(request).encode()) + parts = [] + while data := client.recv(65536): + parts.append(data) + if sum(map(len, parts)) > 32 * 1024**2: + raise ValueError("Host evidence exceeds response bound") + result = json.loads(b"".join(parts)) + if "error" in result: + raise RuntimeError(result["error"]) + return {**identity, **result} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--binding", required=True) + parser.add_argument("--serve", action="store_true") + args = parser.parse_args() + os.umask(0o077) + spec = binding(args.binding) + request = json.load(sys.stdin) + if args.serve: + try: + serve(spec, request) + except Exception as exc: + root = Path(spec["android"]["state_root"]) + state = current(root) + if state and state["run_id"] == request["run_id"]: + state.update(status="stopped", error=type(exc).__name__) + write(root / "session.json", state) + raise + else: + print(json.dumps(lifecycle(spec, request, args.binding))) + + +if __name__ == "__main__": + main() diff --git a/scripts/codex_von_android_control.py b/scripts/codex_von_android_control.py new file mode 100644 index 00000000..ee80fb99 --- /dev/null +++ b/scripts/codex_von_android_control.py @@ -0,0 +1,131 @@ +"""Existing coding controller services one task-bound Android request file. + +The coding child writes a request in its workspace; credentials remain in the +controller. Retained pending intents never replay uncertain native gestures. +""" + +import hashlib +import json +import os +import uuid +from contextlib import contextmanager +from pathlib import Path + +try: + from .codex_von_worker import write_json +except ImportError: + from codex_von_worker import write_json + + +def prepare(config, state, worktree): + if not config.get("android_instance_id"): + return None + root = Path(worktree) / ".run" / ("android-" + state["attempt"]) + root.mkdir(parents=True, exist_ok=True) + return root + + +@contextmanager +def mailbox(root): + # Walk workspace/.run/mailbox without following child-controlled symlinks. + fd = os.open(root.parents[1], os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + for component in (root.parent.name, root.name): + following = os.open( + component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd + ) + os.close(fd) + fd = following + yield fd + finally: + os.close(fd) + + +def respond(root, result): + with mailbox(root) as directory: + temporary = ".response-" + uuid.uuid4().hex + fd = os.open( + temporary, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, + dir_fd=directory, + ) + with os.fdopen(fd, "w") as stream: + json.dump(result, stream) + os.rename( + temporary, "response.json", src_dir_fd=directory, dst_dir_fd=directory + ) + + +def poll(config, state, root, receipts): + with mailbox(root) as directory: + try: + fd = os.open( + "request.json", + os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, + dir_fd=directory, + ) + except FileNotFoundError: + return + with os.fdopen(fd, "rb") as stream: + import stat + + if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode): + raise ValueError("Android request must be a regular file") + raw = stream.read(65537) + if len(raw) > 65536: + raise ValueError("Android request too large") + value = json.loads(raw) + key = hashlib.sha256(raw).hexdigest() + receipt_path = receipts / (key + ".json") + if receipt_path.exists(): + result = json.loads(receipt_path.read_text()) + respond(root, result) + return + # Persist exact intent before effects. Controller interruption cannot replay + # taps or text. The agent can inspect status and request explicit recovery. + result = { + "request_sha256": key, + "status": "effect_uncertain", + "reason": "Intent retained; inspect session before issuing another gesture.", + } + write_json(receipt_path, result) + try: + if not isinstance(value, dict) or set(value) - { + "action", + "x", + "y", + "text", + "fixture", + "request_id", + }: + raise ValueError("Only native operation fields are accepted") + from src.backend.services.coding_agent_android_service import ( + coding_agent_android, + ) + + args = {k: v for k, v in value.items() if k != "request_id"} + result = { + "request_sha256": key, + "result": coding_agent_android( + instance_id=config["android_instance_id"], + task_id=state["task_id"], + run_id=state["attempt"], + **args, + ), + } + except Exception as exc: + result.update(status="failed_or_uncertain", error=type(exc).__name__) + write_json(receipt_path, result) + respond(root, result) + + +def stop(config, state): + from src.backend.services.coding_agent_android_service import coding_agent_android + + return coding_agent_android( + instance_id=config["android_instance_id"], + task_id=state["task_id"], + run_id=state["attempt"], + action="stop", + ) diff --git a/scripts/codex_von_android_transport.py b/scripts/codex_von_android_transport.py new file mode 100644 index 00000000..7774103a --- /dev/null +++ b/scripts/codex_von_android_transport.py @@ -0,0 +1,70 @@ +"""Private Unix-socket transport for an operator-bound Android host command. + +Use an SSH StreamLocal reverse forward for a test host without inbound SSH. +Both socket parents must be private; only the trusted canonical controller gets +access. No TCP listener, client-selected command, credentials or new scheduler. +""" + +import argparse +import json +import os +import socket +import sys +from pathlib import Path + + +def receive(sock, limit): + parts = [] + length = 0 + while data := sock.recv(65536): + parts.append(data) + length += len(data) + if length > limit: + raise ValueError("Transport payload exceeds bound") + return json.loads(b"".join(parts)) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--socket", required=True) + parser.add_argument("--binding") + args = parser.parse_args() + os.umask(0o077) + path = Path(args.socket) + info = path.parent.stat() + if info.st_uid != os.geteuid() or info.st_mode & 0o077: + raise PermissionError("Transport socket directory must be owner-only") + if not args.binding: + with socket.socket(socket.AF_UNIX) as client: + client.settimeout(120) + client.connect(str(path)) + client.sendall(json.dumps(json.load(sys.stdin)).encode()) + client.shutdown(socket.SHUT_WR) + print(json.dumps(receive(client, 32 * 1024**2))) + return + try: + from .codex_von_android import binding, lifecycle + except ImportError: + from codex_von_android import binding, lifecycle + spec = binding(args.binding) + # An existing socket may be an active listener: never replace it blindly. + with socket.socket(socket.AF_UNIX) as listener: + listener.bind(str(path)) + path.chmod(0o600) + listener.listen(4) + try: + while True: + client, _ = listener.accept() + with client: + client.settimeout(120) + try: + result = lifecycle(spec, receive(client, 65536), args.binding) + except Exception as exc: + result = {"error": type(exc).__name__} + client.sendall(json.dumps(result).encode()) + finally: + path.unlink(missing_ok=True) + + +if __name__ == "__main__": + main() diff --git a/scripts/codex_von_capacity.py b/scripts/codex_von_capacity.py index 8ae27bfb..65effc8d 100644 --- a/scripts/codex_von_capacity.py +++ b/scripts/codex_von_capacity.py @@ -8,12 +8,23 @@ import fcntl import os +import platform +import re +import subprocess import stat from contextlib import contextmanager from pathlib import Path def available_memory_bytes(): + if platform.system() == "Darwin": + output = subprocess.check_output(["/usr/bin/vm_stat"], text=True) + page_size = int(re.search(r"page size of (\d+) bytes", output).group(1)) + pages = sum( + int(re.search(rf"{name}:\s+(\d+)", output).group(1)) + for name in ("Pages free", "Pages inactive", "Pages speculative") + ) + return pages * page_size for line in Path("/proc/meminfo").read_text().splitlines(): if line.startswith("MemAvailable:"): return int(line.split()[1]) * 1024 diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py index 954d865d..98c395dd 100755 --- a/scripts/codex_von_worker.py +++ b/scripts/codex_von_worker.py @@ -18,6 +18,7 @@ import shlex import subprocess import sys +import time import uuid from contextlib import ExitStack from datetime import UTC, datetime @@ -106,6 +107,14 @@ def capability_context(config, *, read_only, worktree=None): "worktree": str(worktree) if worktree else None, "controller_runtime": config.get("runtime_evidence", {"available": False}), "public_served_revision": {"available": False, "reason": "Not probed here"}, + "android_test_host": { + "enrolled": bool(config.get("android_instance_id")), + "instance_id": config.get("android_instance_id"), + "controller_tool": "coding_agent_android", + "readiness": "Must discover through the trusted controller for the current task", + "coverage": "Native emulator, not physical-device evidence", + "limitations": "This declaration does not itself expose a tool or grant host access.", + }, "read_only_host_inspection": True, "run_can_edit_checkout": not read_only, "von_mcp": "disabled; canonical reads/effects belong to the controller", @@ -1251,6 +1260,22 @@ def launch(config, state, inputs, conversation, state_path, lock_fd): ] else: args[2:2] = ["--sandbox", "workspace-write"] + try: + from . import codex_von_android_control as android_control + except ImportError: + try: + import codex_von_android_control as android_control + except ImportError: + from scripts import codex_von_android_control as android_control + android_root = android_control.prepare(config, state, worktree) + if android_root: + context["capabilities"]["android_test_host"].update( + request_path=str(android_root / "request.json"), + response_path=str(android_root / "response.json"), + instruction="Atomically write an operation object with action and a fresh request_id; read the matching response. No task/actor/host fields. Do not repeat uncertain gestures.", + ) + write_json(context_path, context) + next_archive = time.monotonic() + ACTIVITY_CHECKPOINT_SECONDS with ( # noqa: SIM117 - streams must outlive the process context (run_dir / "events.jsonl").open("a") as events, (run_dir / "stderr.log").open("w") as errors, @@ -1277,11 +1302,38 @@ def launch(config, state, inputs, conversation, state_path, lock_fd): process.stdin.close() while True: try: - process.wait(timeout=ACTIVITY_CHECKPOINT_SECONDS) + process.wait( + timeout=2 if android_root else ACTIVITY_CHECKPOINT_SECONDS + ) break except subprocess.TimeoutExpired: - archive_checkpoint(config, state) - write_json(state_path, state) + if android_root: + try: + android_control.poll( + config, + state, + android_root, + run_dir / "android-receipts", + ) + except (OSError, ValueError): + pass # malformed/partially-written request cannot stop coding + if time.monotonic() >= next_archive: + archive_checkpoint(config, state) + write_json(state_path, state) + next_archive = time.monotonic() + ACTIVITY_CHECKPOINT_SECONDS + if android_root: + try: + write_json( + run_dir / "android-stop.json", android_control.stop(config, state) + ) + except Exception as exc: + write_json( + run_dir / "android-stop.json", + { + "error": type(exc).__name__, + "cleanup": "Unverified; test-host idle recovery remains active", + }, + ) state["finished_at"] = datetime.now(UTC).isoformat() write_json( run_dir / "exit.json", @@ -1380,9 +1432,9 @@ def apply_deployment(config, api, state, state_path, lock_fd): receipt ) if success: - result["summary"] += ( - f" Deployed and verified the public server at {commit[:12]}." - ) + result[ + "summary" + ] += f" Deployed and verified the public server at {commit[:12]}." else: result["status"] = "blocked" result["summary"] += ( @@ -1393,9 +1445,9 @@ def apply_deployment(config, api, state, state_path, lock_fd): ) if recover_only: result["status"] = "blocked" - result["summary"] += ( - " The task changed; only the interrupted deployment was reconciled." - ) + result[ + "summary" + ] += " The task changed; only the interrupted deployment was reconciled." write_json(state_path, state) diff --git a/src/backend/integrations/internal_mcp/coding_agent_instance_tools.py b/src/backend/integrations/internal_mcp/coding_agent_instance_tools.py index 80fe506a..d2706cca 100644 --- a/src/backend/integrations/internal_mcp/coding_agent_instance_tools.py +++ b/src/backend/integrations/internal_mcp/coding_agent_instance_tools.py @@ -19,14 +19,40 @@ def _instances(**kwargs): ) as exc: return make_error_response( "coding_agent_instance_unavailable", - type(exc).__name__ + ": " + str(exc) - if isinstance(exc, (PermissionError, ValueError)) - else "Instance operation unavailable; inspect operator readiness and private logs.", + ( + type(exc).__name__ + ": " + str(exc) + if isinstance(exc, (PermissionError, ValueError)) + else "Instance operation unavailable; inspect operator readiness and private logs." + ), ) def build_coding_agent_instance_tools(): return [ + MethodDefinition( + name="coding_agent_android", + handler=_android, + input_schema=Schema( + required={"instance_id": str, "task_id": str}, + optional={ + "action": str, + "run_id": str, + "x": int, + "y": int, + "text": str, + "fixture": str, + }, + allow_unknown=False, + ), + output_schema=None, + category="write", + description=( + "Optional native Android test-host capability for a current assigned task. " + "Discover actual host prerequisites; start/status/stop a disposable run with a stable run_id; " + "tap/text/open_fixture or attach a screenshot. Uses an operator-enrolled host route, " + "never creates another coding worker. Emulator evidence is not physical-device coverage." + ), + ), MethodDefinition( name="coding_agent_instances", handler=_instances, @@ -46,5 +72,28 @@ def build_coding_agent_instance_tools(): "and reasoning_effort; task overrides remain supported. Readiness requires " "actual task and recipient evidence, not a selected release or active timer." ), - ) + ), ] + + +def _android(**kwargs): + from ...services.coding_agent_android_service import coding_agent_android + + try: + return coding_agent_android(**kwargs) + except ( + PermissionError, + ValueError, + TypeError, + KeyError, + OSError, + RuntimeError, + ) as exc: + return make_error_response( + "coding_agent_android_unavailable", + ( + str(exc) + if isinstance(exc, (PermissionError, ValueError)) + else "Android operation unavailable; inspect private operator logs." + ), + ) diff --git a/src/backend/services/coding_agent_android_service.py b/src/backend/services/coding_agent_android_service.py new file mode 100644 index 00000000..b2121e18 --- /dev/null +++ b/src/backend/services/coding_agent_android_service.py @@ -0,0 +1,137 @@ +"""Task-authorised Android capability on an enrolled instance's test host. + +The existing operator instance registry binds the route. The authenticated +agent (or its delegator) must still own a live Michael/owner-created assignment. +No caller-supplied actor, organisation, host command or SDK path is accepted. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import subprocess + +from . import coding_agent_instance_service as instances +from . import task_management_service as tasks + + +def authorised_slot(instance_id, task_id): + from ..security.access_control import get_effective_organisation_concept_id + from .organisation_membership_governance_service import _trusted_actor + from .organisation_membership_service import resolve_user_organisation_membership + + actor, error = _trusted_actor() + if error: + raise PermissionError(error["error_code"]) + org = get_effective_organisation_concept_id() + slot = instances.load_registry()["instances"].get(instance_id) + if ( + not slot + or slot["organisation_id"] != org + or actor not in {slot["agent_id"], slot["delegator_id"]} + ): + raise PermissionError("No enrolled Android capability in this actor scope") + if not resolve_user_organisation_membership(actor, org): + raise PermissionError("Current organisation membership required") + task = tasks.get_task(task_id) + if ( + not task + or task["organisation_concept_id"] != org + or task["assignee_concept_id"] != slot["agent_id"] + or task["created_by_concept_id"] != slot["delegator_id"] + ): + raise PermissionError("Current delegator-created task assignment required") + return actor, slot, task + + +def coding_agent_android( + *, + instance_id, + task_id, + action="discover", + run_id=None, + x=None, + y=None, + text=None, + fixture=None, +): + actor, slot, task = authorised_slot(instance_id, task_id) + actions = { + "discover", + "start", + "status", + "stop", + "screenshot", + "report", + "tap", + "text", + "open_fixture", + } + if action not in actions: + raise ValueError("Unsupported Android action") + if action not in {"status", "stop"} and task["status"] not in { + "pending", + "in_progress", + }: + raise PermissionError("Task is not executable") + capability = slot.get("android") + if not capability: + return { + "status": "unavailable", + "reason": "No optional Android test host enrolled", + } + request = {"action": action, "task_id": task_id, "run_id": run_id} + for key, value in (("x", x), ("y", y), ("text", text), ("fixture", fixture)): + if value is not None: + request[key] = value + argv = capability["provisioner_command"] + if ( + not isinstance(argv, list) + or not argv + or not all(isinstance(v, str) for v in argv) + ): + raise ValueError("Invalid fixed Android host command") + process = subprocess.run( + argv, input=json.dumps(request), text=True, capture_output=True + ) + if process.returncode: + raise RuntimeError( + "Android host operation failed; inspect private operator log" + ) + result = json.loads(process.stdout) + for key in ("instance_id", "agent_id", "delegator_id", "organisation_id"): + if result.get(key) != slot[key]: + raise RuntimeError("Android receipt identity mismatch") + for key in ("host", "unix_account"): + if result.get(key) != capability[key]: + raise RuntimeError("Android receipt test-host mismatch") + if action != "discover" and ( + result.get("task_id") != task_id or result.get("run_id") != run_id + ): + raise RuntimeError("Android receipt task/run mismatch") + artifact = result.pop("artifact", None) + if artifact: + raw = base64.b64decode(artifact["data_base64"], validate=True) + if hashlib.sha256(raw).hexdigest() != artifact["sha256"]: + raise RuntimeError("Android evidence checksum mismatch") + # Recheck assignment after transport and before publishing private bytes. + authorised_slot(instance_id, task_id) + result["attachment"] = tasks.add_task_attachment_bytes( + task_id, + data=raw, + filename=artifact["filename"], + actor_concept_id=actor, + media_type=artifact["media_type"], + note=json.dumps( + { + "producer": slot["agent_id"], + "run_id": run_id, + "host": result["host"], + "sha256": artifact["sha256"], + "environment": result.get("environment"), + "coverage": "native_android_emulator", + } + ), + ) + return result diff --git a/tests/backend/test_coding_agent_android.py b/tests/backend/test_coding_agent_android.py new file mode 100644 index 00000000..b5ff46b7 --- /dev/null +++ b/tests/backend/test_coding_agent_android.py @@ -0,0 +1,284 @@ +"""Boundary tests for optional task-scoped native Android testing.""" + +import base64 +import hashlib +import json +from types import SimpleNamespace + +import pytest + +from scripts import codex_von_android as host +from src.backend.services import coding_agent_android_service as service + + +@pytest.fixture +def scope(monkeypatch): + from src.backend.security import access_control + from src.backend.services import ( + organisation_membership_governance_service as governance, + ) + from src.backend.services import organisation_membership_service as membership + + slot = dict( + instance_id="test", + agent_id="#V#agent", + delegator_id="#V#owner", + organisation_id="#V#org", + android=dict( + host="mac", unix_account="test", provisioner_command=["/trusted/android"] + ), + ) + task = dict( + organisation_concept_id="#V#org", + assignee_concept_id="#V#agent", + created_by_concept_id="#V#owner", + status="in_progress", + ) + monkeypatch.setattr(governance, "_trusted_actor", lambda: ("#V#agent", None)) + monkeypatch.setattr( + access_control, "get_effective_organisation_concept_id", lambda: "#V#org" + ) + monkeypatch.setattr( + membership, + "resolve_user_organisation_membership", + lambda *_: {"role": "member"}, + ) + monkeypatch.setattr( + service.instances, "load_registry", lambda: {"instances": {"test": slot}} + ) + monkeypatch.setattr(service.tasks, "get_task", lambda *_: task) + return slot, task + + +def test_assigned_agent_route_attaches_verified_bytes(scope, monkeypatch): + slot, _ = scope + data = b"example native evidence" + result = { + **{k: v for k, v in slot.items() if k != "android"}, + "host": "mac", + "unix_account": "test", + "task_id": "#V#task", + "run_id": "run", + "status": "ready", + "artifact": dict( + filename="screen.png", + media_type="image/png", + data_base64=base64.b64encode(data).decode(), + sha256=hashlib.sha256(data).hexdigest(), + ), + } + calls = [] + monkeypatch.setattr( + service.subprocess, + "run", + lambda *a, **kw: SimpleNamespace(returncode=0, stdout=json.dumps(result)), + ) + monkeypatch.setattr( + service.tasks, + "add_task_attachment_bytes", + lambda *a, **kw: calls.append(kw) or {"attachment_id": "verified"}, + ) + response = service.coding_agent_android( + instance_id="test", task_id="#V#task", run_id="run", action="screenshot" + ) + assert response["attachment"]["attachment_id"] == "verified" + assert calls[0]["data"] == data and calls[0]["actor_concept_id"] == "#V#agent" + assert "artifact" not in response + + +@pytest.mark.parametrize( + "field,value", + [ + ("organisation_concept_id", "#V#other"), + ("created_by_concept_id", "#V#other"), + ("assignee_concept_id", "#V#other"), + ], +) +def test_other_org_owner_or_reassigned_task_denied_before_host( + scope, monkeypatch, field, value +): + scope[1][field] = value + monkeypatch.setattr( + service.subprocess, "run", lambda *a, **kw: pytest.fail("host called") + ) + with pytest.raises(PermissionError): + service.coding_agent_android( + instance_id="test", task_id="#V#task", run_id="run", action="start" + ) + + +def test_forged_identity_and_revoked_membership(scope, monkeypatch): + from src.backend.services import ( + organisation_membership_governance_service as governance, + ) + + monkeypatch.setattr( + governance, "_trusted_actor", lambda: (None, {"error_code": "untrusted_actor"}) + ) + with pytest.raises(PermissionError, match="untrusted_actor"): + service.authorised_slot("test", "#V#task") + + +def test_terminal_task_can_clean_up_but_cannot_launch(scope, monkeypatch): + scope[1]["status"] = "completed" + with pytest.raises(PermissionError, match="not executable"): + service.coding_agent_android( + instance_id="test", task_id="#V#task", run_id="run", action="start" + ) + monkeypatch.setattr( + service.subprocess, "run", lambda *a, **kw: SimpleNamespace(returncode=1) + ) + with pytest.raises(RuntimeError, match="host operation"): + service.coding_agent_android( + instance_id="test", task_id="#V#task", run_id="run", action="stop" + ) + + +def test_cross_task_session_and_shell_input_are_denied(): + with pytest.raises(PermissionError): + host.own_session( + {"task_id": "#V#a", "run_id": "r"}, {"task_id": "#V#b", "run_id": "r"} + ) + for value in ["$(touch /tmp/pwned)", "x'; echo secret", "a\nb"]: + with pytest.raises(ValueError): + host.operate({}, {}, {"action": "text", "text": value}) + + +def test_duplicate_launch_and_stopped_attempt_never_respawn(tmp_path, monkeypatch): + spec = {key: key for key in host.IDENTITY} + monkeypatch.setattr(host, "validate", lambda _: ({}, tmp_path)) + monkeypatch.setattr( + host.subprocess, "Popen", lambda *a, **kw: pytest.fail("duplicate launch") + ) + request = {"action": "start", "task_id": "#V#task", "run_id": "same"} + for phase in ["starting", "ready", "stopped"]: + host.write( + tmp_path / "session.json", + dict(task_id="#V#task", run_id="same", status=phase), + ) + assert host.lifecycle(spec, request, "binding")["status"] == phase + with pytest.raises(PermissionError): + host.lifecycle(spec, {**request, "task_id": "#V#another"}, "binding") + + +def test_missing_prerequisites_do_not_launch(tmp_path, monkeypatch): + monkeypatch.setattr(host.platform, "system", lambda: "Darwin") + monkeypatch.setattr(host.platform, "machine", lambda: "arm64") + with pytest.raises(ValueError, match="platform"): + host.validate({"android": {"platform": ["Linux", "x86_64"]}}) + + +def test_worker_request_is_fixed_to_assignment_and_deduplicated(tmp_path, monkeypatch): + from scripts import codex_von_android_control as controller + + root = tmp_path / ".run" / "child" + root.mkdir(parents=True) + receipts = tmp_path / "controller" + config = {"android_instance_id": "test"} + state = {"task_id": "#V#fixed", "attempt": "fixed-run"} + calls = [] + monkeypatch.setattr( + service, + "coding_agent_android", + lambda **kw: calls.append(kw) or {"status": "ready"}, + ) + (root / "request.json").write_text( + json.dumps({"action": "tap", "x": 1, "y": 2, "request_id": "gesture1"}) + ) + controller.poll(config, state, root, receipts) + controller.poll(config, state, root, receipts) + assert ( + len(calls) == 1 + and calls[0]["task_id"] == "#V#fixed" + and calls[0]["run_id"] == "fixed-run" + ) + (root / "request.json").write_text( + json.dumps({"action": "start", "task_id": "#V#another"}) + ) + controller.poll(config, state, root, receipts) + assert len(calls) == 1 + + +def test_worker_lost_effect_ack_is_not_repeated(tmp_path, monkeypatch): + from scripts import codex_von_android_control as controller + + root = tmp_path / ".run" / "child" + root.mkdir(parents=True) + receipts = tmp_path / "receipts" + receipts.mkdir() + raw = json.dumps({"action": "text", "text": "Hello", "request_id": "1"}).encode() + (root / "request.json").write_bytes(raw) + digest = hashlib.sha256(raw).hexdigest() + (receipts / (digest + ".json")).write_text( + json.dumps({"status": "effect_uncertain"}) + ) + monkeypatch.setattr( + service, "coding_agent_android", lambda **kw: pytest.fail("repeated effect") + ) + controller.poll( + {"android_instance_id": "test"}, + {"task_id": "#V#t", "attempt": "r"}, + root, + receipts, + ) + assert ( + json.loads((root / "response.json").read_text())["status"] == "effect_uncertain" + ) + + +def test_stale_socket_recovery_preserves_an_unrelated_process(tmp_path, monkeypatch): + import socket + + spec = {key: key for key in host.IDENTITY} + state = { + "task_id": "#V#task", + "run_id": "r", + "avd_name": "owned-avd", + "status": "ready", + "emulator_pid": 1234, + } + host.write(tmp_path / "session.json", state) + # macOS AF_UNIX addresses are shorter than pytest long temporary paths. + import tempfile + + short = tempfile.TemporaryDirectory(prefix="von-", dir="/tmp") + tmp_path = __import__("pathlib").Path(short.name) + host.write(tmp_path / "session.json", state) + endpoint = tmp_path / "control.sock" + sock = socket.socket(socket.AF_UNIX) + sock.bind(str(endpoint)) + sock.close() + monkeypatch.setattr(host, "validate", lambda _: ({}, tmp_path)) + monkeypatch.setattr( + host.subprocess, + "run", + lambda *a, **kw: SimpleNamespace(returncode=0, stdout="unrelated-editor"), + ) + monkeypatch.setattr( + host.os, "kill", lambda *a: pytest.fail("killed unrelated process") + ) + result = host.lifecycle( + spec, {"action": "stop", "task_id": "#V#task", "run_id": "r"}, "binding" + ) + assert result["status"] == "stopped" and not endpoint.exists() + + +def test_child_mailbox_cannot_redirect_controller_writes(tmp_path, monkeypatch): + from scripts import codex_von_android_control as controller + + root = tmp_path / ".run" / "child" + root.mkdir(parents=True) + secret = tmp_path / "private.txt" + secret.write_text("preserve") + (root / "response.json").symlink_to(secret) + controller.respond(root, {"status": "ready"}) + assert secret.read_text() == "preserve" + (root / "request.json").symlink_to(secret) + with pytest.raises(OSError): + controller.poll({}, {}, root, tmp_path / "receipts") + import shutil + + shutil.rmtree(root) + root.symlink_to(tmp_path, target_is_directory=True) + with pytest.raises(OSError): + controller.respond(root, {"status": "ready"})