diff --git a/skills/agent-runtime/SKILL.md b/skills/agent-runtime/SKILL.md index 0dd75dc..9a19d12 100644 --- a/skills/agent-runtime/SKILL.md +++ b/skills/agent-runtime/SKILL.md @@ -52,6 +52,22 @@ python3 /scripts/agent.py run stop --root --run /scripts/agent.py run exec --root +``` + +入口只接受 `platform_probe`、`network_probe`、`install`、`start` 四种受控操作，命令 +由 adapter 固定生成，不读取配置中的 command/argv/shell/env/token。local binding 可执行 +固定的只读平台/网络探针和 test-preflight 已授权的 Android 安装/启动；非 local binding +先按 `environment + host + project + path` 解析 Orca worktree，再通过 Orca 的临时 terminal +执行固定的节点 runner，并从协议标记读取一个 JSON 结果，结束后关闭该 terminal。远程 iOS +的安装/启动要求调用方把产物放在 Mac 节点可读路径，并在 matrix 的 `artifact.remote_path` +中声明；协调机上的本地路径不会被错误地当成 Mac 路径。任何 transport、节点工具链或远端 +产物缺失都返回结构化 `environment-fail`，绝不在协调节点执行平台命令或伪造 `ready`。 + ## 验证 - `python3 /scripts/agent.py validate --root ` 通过。 diff --git a/skills/agent-runtime/scripts/agent.py b/skills/agent-runtime/scripts/agent.py index 7130b9a..9c1a041 100755 --- a/skills/agent-runtime/scripts/agent.py +++ b/skills/agent-runtime/scripts/agent.py @@ -8,15 +8,24 @@ shell, and it only creates builder sessions from a fixed make invocation. from __future__ import annotations import argparse +import base64 import copy +import gzip import hashlib import json import os +import platform as host_platform import re +import shlex +import shutil +import socket import subprocess import sys import time import uuid +import urllib.error +import urllib.parse +import urllib.request from pathlib import Path from typing import Any @@ -38,6 +47,9 @@ TARGET_RE = re.compile(r"^[A-Za-z0-9_.-]{1,96}$") MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/+@-]{0,127}$") MAKEFILE_RE = re.compile(r"^[A-Za-z0-9_.-]+(?:/[A-Za-z0-9_.-]+)*$") FORBIDDEN_KEYS = {"command", "argv", "shell", "env", "environmentVariables", "token", "secret", "password"} +RUNNER_OPERATIONS = {"platform_probe", "network_probe", "install", "start", "maestro", "maestro_batch", "artifact_download"} +RUNNER_PLATFORMS = {"android", "ios"} +LOCAL_NODE_NAMES = {"local", "localhost", "127.0.0.1", "::1"} class ConfigError(ValueError): @@ -223,7 +235,17 @@ def orca_json(args: list[str], environment: str | None = None) -> dict[str, Any] if environment and environment != "local": command.extend(["--environment", environment]) command.append("--json") - proc = subprocess.run(command, capture_output=True, text=True, check=False) + # A remote terminal read is a poll, not a blocking operation. Some Orca + # providers keep the CLI open when a PTY has no new output, so bound the + # read and let the caller poll again. Other commands may legitimately + # take longer (for example terminal creation), hence the larger default. + timeout = 10 if args[:2] == ["terminal", "read"] else 120 + try: + proc = subprocess.run(command, capture_output=True, text=True, check=False, timeout=timeout) + except subprocess.TimeoutExpired as exc: + if args[:2] == ["terminal", "read"]: + return {} + raise ConfigError(f"Orca command timed out after {timeout}s: {' '.join(command)}") from exc raw = (proc.stdout or "").strip() try: data = json.loads(raw) @@ -282,12 +304,19 @@ def worktree_for(binding_value: dict[str, Any]) -> dict[str, Any]: if location["environment"] == "local" and expected_host == "local" and str(item.get("hostId")) != "local": continue candidates.append(item) - if len(candidates) != 1: - if not candidates: - raise ConfigError( - f"找不到精确 worktree: environment={environment}, host={location_host(location)}, " - f"project={location['project']}, path={location['path']}；先创建或登记该 worktree" - ) + if not candidates: + raise ConfigError( + f"找不到精确 worktree: environment={environment}, host={location_host(location)}, " + f"project={location['project']}, path={location['path']}；先创建或登记该 worktree" + ) + if len(candidates) > 1: + # Orca can retain duplicate records for the same physical worktree + # after an environment is paired again. They are safe to collapse only + # when every execution-relevant field is identical; divergent records + # must still fail closed instead of guessing a node or checkout. + identity = {(str(item.get("path")), str(item.get("projectId")), str(item.get("head")), str(item.get("hostId"))) for item in candidates} + if len(identity) == 1: + return candidates[0] raise ConfigError(f"binding 匹配到多个 worktree: {len(candidates)}") return candidates[0] @@ -358,6 +387,673 @@ def make_command(value: dict[str, Any]) -> str: return f"exec make -f {value.get('makefile', 'makefile.builder')} {value['target']}" +def runner_response(status: str, observed: Any = "", *, checks: dict[str, Any] | None = None, remediation: str = "", **extra: Any) -> dict[str, Any]: + """Build the JSON response shared with test-preflight's runner contract.""" + + response: dict[str, Any] = {"status": status, "observed": observed} + if checks: + response["checks"] = checks + if remediation: + response["remediation"] = remediation + response.update(extra) + return response + + +def run_fixed_command(command: list[str], *, cwd: str | None = None) -> tuple[int, str]: + """Run an adapter-owned argv; never invoke a shell or print command output.""" + + try: + process = subprocess.run( + command, + cwd=cwd, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + except OSError as exc: + return 127, str(exc) + return process.returncode, (process.stdout or "").strip() + + +def _runner_check(status: str, observed: Any, expected: str, remediation: str = "") -> dict[str, Any]: + value: dict[str, Any] = {"status": status, "observed": observed, "expected": expected} + if remediation: + value["remediation"] = remediation + return value + + +def _local_platform_probe(platform_name: str) -> dict[str, Any]: + if platform_name == "android": + adb_code, adb_output = run_fixed_command(["adb", "version"]) + if adb_code != 0: + failure = "adb unavailable on the local tester node" + return runner_response("environment-fail", failure, checks={ + "adb": _runner_check("environment-fail", failure, "adb responds", "Install or expose adb on the tester node"), + "device": _runner_check("environment-fail", "not checked", "authorized Android device", "Make adb available first"), + "boot": _runner_check("environment-fail", "not checked", "Android device boot complete", "Make adb available first"), + }) + devices_code, devices_output = run_fixed_command(["adb", "devices"]) + devices: list[str] = [] + if devices_code == 0: + for line in devices_output.splitlines(): + if "\tdevice" in line: + devices.append(line.split("\t", 1)[0].strip()) + if devices_code != 0 or not devices: + observed = "adb responded but no authorized device is available" + return runner_response("environment-fail", observed, checks={ + "adb": _runner_check("ready" if devices_code == 0 else "environment-fail", "adb responded" if devices_code == 0 else "adb devices failed", "adb responds"), + "device": _runner_check("environment-fail", observed, "authorized Android device", "Connect or start an authorized tester device"), + "boot": _runner_check("environment-fail", "not checked", "Android device boot complete", "Provide a booted authorized device"), + }) + serial = devices[0] + boot_code, boot_output = run_fixed_command(["adb", "-s", serial, "shell", "getprop", "sys.boot_completed"]) + booted = boot_code == 0 and boot_output.strip() == "1" + return runner_response("ready" if booted else "environment-fail", "adb/device probe completed", checks={ + "adb": _runner_check("ready", "adb responded", "adb responds"), + "device": _runner_check("ready", "authorized", "authorized Android device"), + "boot": _runner_check("ready" if booted else "environment-fail", "boot complete" if booted else "device is not booted", "Android device boot complete", "Boot the Android tester device" if not booted else ""), + }) + + if platform_name != "ios": + return runner_response("blocked", "unsupported platform", remediation="Use platform android or ios") + if host_platform.system() != "Darwin": + observed = "iOS runner requires a macOS tester node; xcrun was not invoked" + return runner_response("environment-fail", observed, checks={ + "macos": _runner_check("environment-fail", observed, "Darwin tester node", "Route iOS tests to a Mac tester binding"), + "xcrun": _runner_check("environment-fail", "not invoked", "xcrun available", "Route iOS tests to a Mac tester binding"), + "simctl": _runner_check("environment-fail", "not invoked", "simctl available", "Route iOS tests to a Mac tester binding"), + "boot": _runner_check("environment-fail", "not checked", "booted iOS simulator", "Route iOS tests to a Mac tester binding"), + }) + xcrun_code, _ = run_fixed_command(["xcrun", "--version"]) + simctl_code, simctl_output = run_fixed_command(["xcrun", "simctl", "list", "devices", "--json"]) + booted = False + if simctl_code == 0: + try: + data = json.loads(simctl_output) + booted = any( + isinstance(device, dict) and device.get("state") == "Booted" + for devices in (data.get("devices") or {}).values() + if isinstance(devices, list) + for device in devices + ) + except (TypeError, json.JSONDecodeError): + booted = False + status = "ready" if xcrun_code == 0 and simctl_code == 0 and booted else "environment-fail" + return runner_response(status, "macOS/simctl probe completed", checks={ + "macos": _runner_check("ready", "Darwin", "Darwin tester node"), + "xcrun": _runner_check("ready" if xcrun_code == 0 else "environment-fail", "available" if xcrun_code == 0 else "unavailable", "xcrun available"), + "simctl": _runner_check("ready" if simctl_code == 0 else "environment-fail", "available" if simctl_code == 0 else "unavailable", "simctl available"), + "boot": _runner_check("ready" if booted else "environment-fail", "booted simulator" if booted else "no booted simulator", "booted iOS simulator", "Boot an iOS simulator" if not booted else ""), + }) + + +def _network_probe(request: dict[str, Any]) -> dict[str, Any]: + raw_url = request.get("url") + if not isinstance(raw_url, str) or not raw_url.strip(): + return runner_response("blocked", "network_probe requires url", remediation="Provide an http(s) URL") + parsed = urllib.parse.urlsplit(raw_url) + if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password: + return runner_response("blocked", "URL must be http(s) without embedded credentials", remediation="Provide a credential-free probe URL") + try: + port = parsed.port or (443 if parsed.scheme == "https" else 80) + except ValueError: + return runner_response("blocked", "URL contains an invalid port", remediation="Provide a valid http(s) probe URL") + try: + addresses = socket.getaddrinfo(parsed.hostname, port, type=socket.SOCK_STREAM) + except (OSError, ValueError) as exc: + return runner_response("environment-fail", "DNS probe failed", checks={ + "dns": _runner_check("environment-fail", str(exc), "resolvable hostname", "Check tester-node DNS/VPN") + }) + checks: dict[str, Any] = {"dns": _runner_check("ready", sorted({item[4][0] for item in addresses}), "resolvable hostname")} + try: + import socket as _socket + with _socket.create_connection((parsed.hostname, port), timeout=5): + pass + checks["tcp"] = _runner_check("ready", f"{parsed.hostname}:{port}", "TCP connection") + except OSError as exc: + checks["tcp"] = _runner_check("environment-fail", str(exc), f"TCP {parsed.hostname}:{port}", "Check the tester-node route and firewall") + return runner_response("environment-fail", "TCP probe failed", checks=checks) + try: + with urllib.request.urlopen(urllib.request.Request(raw_url, method="GET"), timeout=5) as response: + code = response.status + except urllib.error.HTTPError as exc: + code = exc.code + except (OSError, urllib.error.URLError) as exc: + checks["http"] = _runner_check("environment-fail", str(exc), "HTTP response", "Check service listener, proxy and TLS") + return runner_response("environment-fail", "HTTP probe failed", checks=checks) + expected = request.get("probe", {}).get("expected_status") if isinstance(request.get("probe"), dict) else None + matches = not isinstance(expected, list) or code in expected + checks["http"] = _runner_check("ready" if 200 <= code < 500 and matches else "environment-fail", code, expected or "HTTP 2xx-4xx", "Check service health or expected_status" if not matches or code >= 500 else "") + return runner_response("ready" if checks["http"]["status"] == "ready" else "environment-fail", "DNS/TCP/HTTP probe completed", checks=checks) + + +def _request_artifact(request: dict[str, Any]) -> tuple[Path | None, str | None]: + value = request.get("artifact") + if not isinstance(value, str) or not value.strip(): + return None, "install requires an artifact path" + path = Path(value) + if not path.is_absolute() or ".." in path.parts or path == Path("/"): + return None, "artifact must be a safe absolute path" + return path, None + + +def _request_flow(request: dict[str, Any]) -> tuple[str | None, str | None]: + flow = request.get("flow") + if not isinstance(flow, str) or not flow.strip(): + return None, "maestro requires a non-empty flow" + if len(flow.encode("utf-8")) > 120_000: + return None, "maestro flow is too large" + env = request.get("env", {}) + if env is not None: + if not isinstance(env, dict) or any( + not isinstance(key, str) or not re.fullmatch(r"[A-Z][A-Z0-9_]{0,63}", key) + or not isinstance(value, str) or len(value) > 4096 + for key, value in env.items() + ): + return None, "maestro env must contain safe uppercase names and text values" + return flow, None + + +def _maestro_env(request: dict[str, Any]) -> dict[str, str]: + values = request.get("env") if isinstance(request.get("env"), dict) else {} + return {str(key): str(value) for key, value in values.items()} + + +def _local_maestro(platform_name: str, request: dict[str, Any]) -> dict[str, Any]: + flow, error = _request_flow(request) + if error: + return runner_response("blocked", error, remediation="Generate a bounded Maestro flow from the catalog case") + if platform_name != "android": + return runner_response("blocked", "local Maestro runner only supports Android", remediation="Route iOS Maestro to tester.ios") + maestro = Path("/home/ace/.maestro/bin/maestro") + if not maestro.is_file(): + maestro_path = shutil.which("maestro") + maestro = Path(maestro_path) if maestro_path else maestro + if not maestro.is_file(): + return runner_response("environment-fail", "maestro is not installed on the local tester node", remediation="Install Maestro and rerun test-preflight") + device_code, device_output = run_fixed_command(["adb", "devices"]) + serial = next((line.split("\t", 1)[0].strip() for line in device_output.splitlines() if "\tdevice" in line), "") if device_code == 0 else "" + if not serial: + return runner_response("environment-fail", "no authorized Android device for Maestro") + flow_path = Path("/tmp") / f"pouch-maestro-{uuid.uuid4().hex}.yaml" + flow_path.write_text(flow, encoding="utf-8") + try: + command = [str(maestro), "test", "--device", serial, "--no-color", str(flow_path)] + for key, value in _maestro_env(request).items(): + command.extend(["-e", f"{key}={value}"]) + env = os.environ.copy() + env["MAESTRO_CLI_NO_ANALYTICS"] = "true" + process = subprocess.run(command, capture_output=True, text=True, check=False, timeout=float(request.get("timeout") or 180), env=env) + output = ((process.stdout or "") + ("\n" + process.stderr if process.stderr else "")).strip()[-6000:] + return runner_response("ready", "maestro completed", checks={"maestro": _runner_check("ready", output, "Maestro flow executed")}, test_result="pass" if process.returncode == 0 else "fail", exit=process.returncode, output=output) + except subprocess.TimeoutExpired as exc: + return runner_response("ready", "maestro timed out", checks={"maestro": _runner_check("ready", "timeout", "Maestro flow executed")}, test_result="environment-fail", exit=124, output=str(exc)) + finally: + flow_path.unlink(missing_ok=True) + + +def _local_mutation(platform_name: str, operation: str, request: dict[str, Any]) -> dict[str, Any]: + if platform_name == "ios": + return runner_response("blocked", "local iOS mutation requires an explicit Mac transport", remediation="Use a non-local tester binding with a supported node transport") + if platform_name != "android": + return runner_response("blocked", "unsupported platform", remediation="Use platform android or ios") + path, error = _request_artifact(request) if operation == "install" else (None, None) + identity = request.get("identity") + if operation == "install" and error: + return runner_response("blocked", error, remediation="Provide the caller-supplied APK path") + if not isinstance(identity, str) or not re.fullmatch(r"[A-Za-z0-9_.-]{1,192}", identity): + return runner_response("blocked", "identity must be a safe Android package id", remediation="Provide the declared artifact package") + if operation == "install" and not path.is_file(): + return runner_response("blocked", "artifact does not exist", remediation="Provide an existing caller-supplied APK") + probe = _local_platform_probe("android") + if probe.get("status") != "ready": + return probe + device_code, device_output = run_fixed_command(["adb", "devices"]) + serial = next((line.split("\t", 1)[0].strip() for line in device_output.splitlines() if "\tdevice" in line), "") if device_code == 0 else "" + if not serial: + return runner_response("environment-fail", "no authorized Android device") + command = ["adb", "-s", serial, "install", "-r", str(path)] if operation == "install" else ["adb", "-s", serial, "shell", "monkey", "-p", identity, "1"] + code, _ = run_fixed_command(command) + if code != 0: + return runner_response("environment-fail", f"{operation} command failed", remediation=f"Inspect the tester-node {operation} evidence") + return runner_response("ready", f"{operation} completed", checks={operation: _runner_check("ready", "completed", "operation succeeded")}) + + +REMOTE_RUNNER_SOURCE = r''' +import base64, gzip, json, os, re, shutil, socket, subprocess, sys, urllib.error, urllib.parse, urllib.request, uuid, zipfile +from pathlib import Path + +request_bytes = base64.b64decode(sys.stdin.read()) +try: + request = json.loads(request_bytes.decode("utf-8")) +except UnicodeDecodeError: + request = json.loads(gzip.decompress(request_bytes).decode("utf-8")) +marker = request.pop("_marker") + +def check(status, observed, expected, remediation=""): + item = {"status": status, "observed": observed, "expected": expected} + if remediation: + item["remediation"] = remediation + return item + +def run(argv, env=None, timeout=180): + try: + proc = subprocess.run(argv, env=env, timeout=timeout, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False) + except subprocess.TimeoutExpired as exc: + return 124, (str(exc) or "command timed out").strip() + except OSError as exc: + return 127, str(exc).strip() + return proc.returncode, (proc.stdout or "").strip() + +def response(status, observed, checks=None, remediation=""): + item = {"status": status, "observed": observed} + if checks: + item["checks"] = checks + if remediation: + item["remediation"] = remediation + return item + +def ios_platform(): + uname_code, uname_output = run(["uname", "-s"]) + darwin = uname_code == 0 and uname_output.splitlines()[-1:] == ["Darwin"] + checks = {"macos": check("ready" if darwin else "environment-fail", uname_output or "unavailable", "Darwin tester node", "Route iOS tests to the Mac tester node" if not darwin else "")} + if not darwin: + checks.update({ + "xcrun": check("environment-fail", "not invoked", "xcrun available", "Use a macOS tester node"), + "simctl": check("environment-fail", "not invoked", "simctl available", "Use a macOS tester node"), + "boot": check("environment-fail", "not checked", "booted iOS simulator", "Use a macOS tester node"), + }) + return response("environment-fail", "remote node is not macOS", checks) + xcrun_code, xcrun_output = run(["xcrun", "--version"]) + simctl_code, simctl_output = run(["xcrun", "simctl", "list", "devices", "--json"]) + booted = False + if simctl_code == 0: + try: + data = json.loads(simctl_output) + booted = any( + isinstance(device, dict) and device.get("state") == "Booted" + for devices in (data.get("devices") or {}).values() + if isinstance(devices, list) + for device in devices + ) + except (TypeError, json.JSONDecodeError): + booted = False + checks.update({ + "xcrun": check("ready" if xcrun_code == 0 else "environment-fail", xcrun_output or "unavailable", "xcrun available", "Install/select Xcode command-line tools" if xcrun_code else ""), + "simctl": check("ready" if simctl_code == 0 else "environment-fail", "available" if simctl_code == 0 else simctl_output, "simctl available", "Check Xcode simulator runtime" if simctl_code else ""), + "boot": check("ready" if booted else "environment-fail", "booted simulator" if booted else "no booted simulator", "booted iOS simulator", "Boot an iOS simulator" if not booted else ""), + }) + status = "ready" if all(item["status"] == "ready" for item in checks.values()) else "environment-fail" + return response(status, "macOS/xcrun/simctl probe completed", checks) + +def network_probe(): + raw_url = request.get("url") + parsed = urllib.parse.urlsplit(raw_url or "") + if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password: + return response("blocked", "URL must be http(s) without embedded credentials", remediation="Provide a credential-free probe URL") + port = parsed.port or (443 if parsed.scheme == "https" else 80) + try: + addresses = socket.getaddrinfo(parsed.hostname, port, type=socket.SOCK_STREAM) + except (OSError, ValueError) as exc: + return response("environment-fail", "DNS probe failed", {"dns": check("environment-fail", str(exc), "resolvable hostname", "Check tester-node DNS/VPN")}) + checks = {"dns": check("ready", sorted({item[4][0] for item in addresses}), "resolvable hostname")} + try: + with socket.create_connection((parsed.hostname, port), timeout=5): + pass + checks["tcp"] = check("ready", f"{parsed.hostname}:{port}", "TCP connection") + except OSError as exc: + checks["tcp"] = check("environment-fail", str(exc), f"TCP {parsed.hostname}:{port}", "Check tester-node route and firewall") + return response("environment-fail", "TCP probe failed", checks) + try: + with urllib.request.urlopen(urllib.request.Request(raw_url, method="GET"), timeout=5) as result: + code = result.status + except urllib.error.HTTPError as exc: + code = exc.code + except (OSError, urllib.error.URLError) as exc: + checks["http"] = check("environment-fail", str(exc), "HTTP response", "Check service listener, proxy and TLS") + return response("environment-fail", "HTTP probe failed", checks) + expected = request.get("probe", {}).get("expected_status") if isinstance(request.get("probe"), dict) else None + matches = not isinstance(expected, list) or code in expected + checks["http"] = check("ready" if 200 <= code < 500 and matches else "environment-fail", code, expected or "HTTP 2xx-4xx", "Check service health or expected_status" if not matches or code >= 500 else "") + return response("ready" if checks["http"]["status"] == "ready" else "environment-fail", "DNS/TCP/HTTP probe completed", checks) + +operation = request.get("operation") +if operation == "platform_probe" and request.get("platform") == "ios": + result = ios_platform() +elif operation == "network_probe": + result = network_probe() +elif operation == "maestro" and request.get("platform") == "ios": + flow = request.get("flow") + values = request.get("env") if isinstance(request.get("env"), dict) else {} + if not isinstance(flow, str) or not flow.strip() or len(flow.encode("utf-8")) > 120000: + result = response("blocked", "maestro requires a bounded non-empty flow") + elif any(not isinstance(key, str) or not key.isidentifier() or not key.isupper() or not isinstance(value, str) or len(value) > 4096 for key, value in values.items()): + result = response("blocked", "maestro env must contain uppercase text values") + else: + flow_path = Path("/tmp") / ("pouch-maestro-" + uuid.uuid4().hex + ".yaml") + try: + flow_path.write_text(flow, encoding="utf-8") + env = os.environ.copy() + env["PATH"] = "/Users/laily/bin:/Users/laily/.maestro/bin:/opt/homebrew/bin:" + env.get("PATH", "") + env["JAVA_HOME"] = "/opt/homebrew/opt/openjdk@17/libexec/openjdk.jdk/Contents/Home" + env["MAESTRO_CLI_NO_ANALYTICS"] = "true" + device = "booted" + sim_code, sim_output = run(["xcrun", "simctl", "list", "devices", "--json"], env=env, timeout=30) + if sim_code == 0: + try: + sim_data = json.loads(sim_output) + booted = [ + item.get("udid") for devices in (sim_data.get("devices") or {}).values() + if isinstance(devices, list) + for item in devices + if isinstance(item, dict) and item.get("state") == "Booted" and item.get("udid") + ] + if booted: + device = str(booted[0]) + # Maestro's iOS driver uses IDB for simulator I/O. The + # companion is deliberately connected through the user + # local path; Homebrew cannot create /usr/local/bin here. + run(["idb", "connect", device, "--json"], env=env, timeout=20) + except (TypeError, json.JSONDecodeError): + pass + command = ["/Users/laily/.maestro/bin/maestro", "test", "--device", device, "--no-color", str(flow_path)] + for key, value in values.items(): + command.extend(["-e", key + "=" + value]) + code, output = run(command, env=env, timeout=float(request.get("timeout") or 180)) + result = response("ready" if code not in {124, 127} else "environment-fail", "maestro completed" if code == 0 else ("maestro command unavailable" if code == 127 else "maestro timed out"), {"maestro": check("ready" if code == 0 else ("environment-fail" if code in {124, 127} else "ready"), output[-6000:], "Maestro flow executed")}) + result.update({"test_result": "pass" if code == 0 else ("environment-fail" if code in {124, 127} else "fail"), "exit": code, "output": output[-6000:]}) + finally: + try: + flow_path.unlink() + except OSError: + pass +elif operation == "maestro_batch" and request.get("platform") == "ios": + cases = request.get("cases") + if not isinstance(cases, list) or not cases or len(cases) > 64: + result = response("blocked", "maestro_batch requires 1-64 cases") + else: + values = request.get("env", {}) + if not isinstance(values, dict) or any(not isinstance(key, str) or not key.isidentifier() or not key.isupper() or not isinstance(value, str) or len(value) > 4096 for key, value in values.items()): + result = response("blocked", "maestro env must contain uppercase text values") + else: + env = os.environ.copy() + env["PATH"] = "/Users/laily/bin:/Users/laily/.maestro/bin:/opt/homebrew/bin:" + env.get("PATH", "") + env["JAVA_HOME"] = "/opt/homebrew/opt/openjdk@17/libexec/openjdk.jdk/Contents/Home" + env["MAESTRO_CLI_NO_ANALYTICS"] = "true" + device = "booted" + sim_code, sim_output = run(["xcrun", "simctl", "list", "devices", "--json"], env=env, timeout=30) + if sim_code == 0: + try: + sim_data = json.loads(sim_output) + booted = [ + item.get("udid") for devices in (sim_data.get("devices") or {}).values() + if isinstance(devices, list) + for item in devices + if isinstance(item, dict) and item.get("state") == "Booted" and item.get("udid") + ] + if booted: + device = str(booted[0]) + run(["idb", "connect", device, "--json"], env=env, timeout=20) + except (TypeError, json.JSONDecodeError): + pass + results = [] + for item in cases: + case_id = item.get("id") if isinstance(item, dict) else None + flow = item.get("flow") if isinstance(item, dict) else None + case_env = item.get("env", values) if isinstance(item, dict) else values + if not isinstance(case_id, str) or not case_id or not isinstance(flow, str) or not flow.strip() or len(flow.encode("utf-8")) > 120000: + results.append({"id": str(case_id or ""), "test_result": "environment-fail", "exit": 2, "output": "invalid maestro batch case"}) + continue + if not isinstance(case_env, dict) or any(not isinstance(key, str) or not key.isidentifier() or not key.isupper() or not isinstance(value, str) or len(value) > 4096 for key, value in case_env.items()): + results.append({"id": case_id, "test_result": "environment-fail", "exit": 2, "output": "invalid maestro batch env"}) + continue + flow_path = Path("/tmp") / ("pouch-maestro-batch-" + uuid.uuid4().hex + ".yaml") + try: + flow_path.write_text(flow, encoding="utf-8") + command = ["/Users/laily/.maestro/bin/maestro", "test", "--device", device, "--no-color", str(flow_path)] + for key, value in case_env.items(): + command.extend(["-e", key + "=" + value]) + code, output = run(command, env=env, timeout=float(request.get("case_timeout") or 180)) + test_result = "pass" if code == 0 else ("environment-fail" if code in {124, 127} else "fail") + results.append({"id": case_id, "test_result": test_result, "exit": code, "output": output[-6000:]}) + except Exception as exc: + results.append({"id": case_id, "test_result": "environment-fail", "exit": 124, "output": str(exc)}) + finally: + try: + flow_path.unlink() + except OSError: + pass + has_env_failure = any(item.get("test_result") == "environment-fail" for item in results) + result = response("environment-fail" if has_env_failure else "ready", "maestro batch completed", {"maestro": check("environment-fail" if has_env_failure else "ready", f"{len(results)} cases", "Maestro batch executed")}) + result.update({"results": results}) +elif operation == "artifact_download": + url = request.get("url") + destination = request.get("destination") + if not isinstance(url, str) or not url.startswith("https://git.duguying.net/"): + result = response("blocked", "artifact_download only accepts the configured Gitea host") + elif not isinstance(destination, str) or not destination.startswith("/Users/laily/orca/workspaces/app/"): + result = response("blocked", "artifact destination must stay under the Mac tester artifact root") + else: + dest = Path(destination) + archive = dest.with_suffix(".artifact.zip") + try: + config_paths = [ + Path.home() / ".config" / "tea" / "config.yml", + Path("/Users/duguying/.config/tea/config.yml"), + ] + config_path = next((item for item in config_paths if item.is_file()), None) + if config_path is None: + raise RuntimeError("Mac mini tea login for git.duguying.net is missing") + config_text = config_path.read_text(encoding="utf-8") + token_match = re.search(r"name:\s*git\.duguying\.net[\s\S]*?token:\s*([^\s]+)", config_text) + token = token_match.group(1) if token_match else "" + if not token: + raise RuntimeError(f"Mac mini tea login for git.duguying.net is missing in {config_path}") + request_obj = urllib.request.Request(url, headers={"Authorization": f"token {token}", "User-Agent": "pouch-tester/1"}) + dest.parent.mkdir(parents=True, exist_ok=True) + with urllib.request.urlopen(request_obj, timeout=120) as source, archive.open("wb") as target: + shutil.copyfileobj(source, target) + with zipfile.ZipFile(archive) as artifact_archive: + nested = next((name for name in artifact_archive.namelist() if name.endswith("ios-simulator.zip")), None) + if not nested: + raise RuntimeError("Gitea artifact does not contain ios-simulator.zip") + nested_bytes = artifact_archive.read(nested) + nested_archive = dest.with_suffix(".ios-simulator.zip") + nested_archive.write_bytes(nested_bytes) + with zipfile.ZipFile(nested_archive) as app_archive: + app_archive.extractall(dest.parent) + app = dest.parent / "Writes.app" + if not app.is_dir() or not (app / "Info.plist").is_file(): + raise RuntimeError("downloaded iOS artifact does not contain Writes.app/Info.plist") + result = response("ready", "iOS simulator artifact downloaded", {"artifact": check("ready", str(app), "remote artifact present")}) + except Exception as exc: + result = response("environment-fail", "iOS simulator artifact download failed", remediation=str(exc)) +elif operation in {"install", "start"} and request.get("platform") == "ios": + identity = request.get("identity") + artifact = request.get("artifact") + if operation == "install": + if not isinstance(artifact, str) or not (Path(artifact).is_file() or Path(artifact).is_dir()): + result = response("environment-fail", "remote iOS artifact is not present", remediation="Provide artifact.remote_path on the Mac tester node") + else: + sim_code, sim_output = run(["xcrun", "simctl", "list", "devices", "--json"]) + device = "booted" + if sim_code == 0: + try: + sim_data = json.loads(sim_output) + booted = [ + item.get("udid") for devices in (sim_data.get("devices") or {}).values() + if isinstance(devices, list) + for item in devices + if isinstance(item, dict) and item.get("state") == "Booted" and item.get("udid") + ] + if booted: + device = str(booted[0]) + except (TypeError, json.JSONDecodeError): + pass + code, output = run(["xcrun", "simctl", "install", device, artifact]) + result = response("ready" if code == 0 else "environment-fail", "iOS app installed" if code == 0 else output, {"install": check("ready" if code == 0 else "environment-fail", output or "installed", "simctl install succeeds", "Check IPA and simulator" if code else "")}) + elif not isinstance(identity, str) or not identity: + result = response("blocked", "start requires bundle id") + else: + code, output = run(["xcrun", "simctl", "launch", "booted", identity]) + result = response("ready" if code == 0 else "environment-fail", "iOS app started" if code == 0 else output, {"start": check("ready" if code == 0 else "environment-fail", output or "started", "simctl launch succeeds", "Check bundle id and installed app" if code else "")}) +else: + result = response("blocked", "unsupported remote runner operation", remediation="Use iOS platform_probe, network_probe, install, or start") + +print(marker + base64.b64encode(json.dumps(result, separators=(",", ":")).encode("utf-8")).decode("ascii")) +''' + + +def _terminal_output(result: dict[str, Any]) -> str: + terminal = result.get("terminal") if isinstance(result, dict) else None + if not isinstance(terminal, dict): + terminal = result + tail = terminal.get("tail") if isinstance(terminal, dict) else None + if isinstance(tail, list): + return "\n".join(str(item) for item in tail) + if isinstance(tail, str): + return tail + output = terminal.get("output") if isinstance(terminal, dict) else None + return output if isinstance(output, str) else "" + + +def remote_runner_transport(binding_value: dict[str, Any], request: dict[str, Any], worktree: dict[str, Any]) -> dict[str, Any]: + environment = str(binding_value["location"]["environment"]) + marker = "POUCH_RUNNER_RESULT_" + uuid.uuid4().hex + source = REMOTE_RUNNER_SOURCE + source_b64 = base64.b64encode(gzip.compress(source.encode("utf-8"), mtime=0)).decode("ascii") + payload = dict(request) + payload["_marker"] = marker + payload_b64 = base64.b64encode(gzip.compress(json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8"), mtime=0)).decode("ascii") + handle = "" + try: + def send_text(text: str) -> None: + # Keep each PTY write below the provider's line-editor limit, then + # submit it explicitly. Long base64 lines are accepted as input + # but may remain unsubmitted in fish. + orca_json(["terminal", "send", "--terminal", handle, "--text", text], environment) + orca_json(["terminal", "send", "--terminal", handle, "--text", "\r"], environment) + # The remote PTY acknowledges accepted bytes before fish has + # drained its input buffer. Leave room between chunks so no + # base64 data is dropped on slower Mac sessions. + time.sleep(0.2) + + created = orca_json(["terminal", "create", "--worktree", f"id:{worktree['id']}", "--title", "tester-node-runner"], environment) + terminal = created.get("terminal") if isinstance(created, dict) else None + handle = str(terminal.get("handle")) if isinstance(terminal, dict) and terminal.get("handle") else "" + if not handle: + return runner_response("environment-fail", "remote runner terminal was not created", remediation="Check the Orca macmini terminal service") + remote_base = f"/tmp/pouch-remote-runner-{uuid.uuid4().hex}" + source_file = remote_base + ".b64" + payload_file = remote_base + ".payload.b64" + send_text(f": > {source_file}") + for offset in range(0, len(source_b64), 400): + send_text(f"printf '%s' '{source_b64[offset:offset + 400]}' >> {source_file}") + send_text( + "python3 -c 'import base64,gzip;open(\"%s.py\",\"wb\").write(gzip.decompress(base64.b64decode(open(\"%s\",\"rb\").read())))'" + % (remote_base, source_file) + ) + send_text(f": > {payload_file}") + for offset in range(0, len(payload_b64), 400): + send_text(f"printf '%s' '{payload_b64[offset:offset + 400]}' >> {payload_file}") + # The embedded runner accepts base64(gzip(JSON)) on stdin directly. + send_text(f"python3 {remote_base}.py < {payload_file}") + runner_wait_ms = min( + 900000, + max(15000, int(float(request.get("timeout") or request.get("case_timeout") or 180) * 1000) + 15000), + ) + output = "" + deadline = time.monotonic() + runner_wait_ms / 1000 + while time.monotonic() < deadline: + output = _terminal_output(orca_json(["terminal", "read", "--terminal", handle, "--limit", "20000"], environment)) + for line in reversed(output.splitlines()): + if line.startswith(marker): + try: + result = json.loads(base64.b64decode(line[len(marker):]).decode("utf-8")) + except (ValueError, json.JSONDecodeError): + break + if isinstance(result, dict): + result.setdefault("checks", {}) + result["checks"].setdefault("binding", _runner_check("ready", binding_value["id"], "tester binding resolved")) + result["checks"].setdefault("worktree", _runner_check("ready", worktree.get("path", "resolved"), "exact Orca worktree")) + return result + time.sleep(1) + return runner_response("environment-fail", "remote runner did not return a protocol result", checks={ + "binding": _runner_check("ready", binding_value["id"], "tester binding resolved"), + "worktree": _runner_check("ready", worktree.get("path", "resolved"), "exact Orca worktree"), + "transport": _runner_check("environment-fail", output[-1000:], "node runner protocol result", "Inspect the macmini terminal output"), + }) + except (ConfigError, OSError) as exc: + return runner_response("environment-fail", "remote runner transport failed", remediation=str(exc)) + finally: + if handle: + try: + orca_json(["terminal", "close", "--terminal", handle], environment) + except ConfigError: + pass + + +def execute_runner_request(root: Path, request: Any) -> dict[str, Any]: + if not isinstance(request, dict): + return runner_response("blocked", "runner request must be a JSON object", remediation="Send one JSON object on stdin") + binding_id = request.get("binding") + operation = request.get("operation") + platform_name = request.get("platform") + if not isinstance(binding_id, str) or not BINDING_RE.fullmatch(binding_id): + return runner_response("blocked", "binding must be a safe namespace.name", remediation="Use the resolved tester. binding") + if not isinstance(operation, str) or operation not in RUNNER_OPERATIONS: + return runner_response("blocked", "unknown runner operation", remediation="Use platform_probe, network_probe, install, or start") + if not isinstance(platform_name, str) or platform_name not in RUNNER_PLATFORMS: + return runner_response("blocked", "platform must be android or ios", remediation="Use the matrix platform") + try: + value = binding(load_config(root), binding_id) + except (ConfigError, OSError) as exc: + return runner_response("blocked", "tester binding could not be resolved", remediation=str(exc)) + if value.get("adapter") != "tester": + return runner_response("blocked", "binding is not a tester binding", remediation="Use a binding with adapter: tester") + location = value["location"] + environment = str(location.get("environment")) + requested_node = request.get("node") + if requested_node is not None and (not isinstance(requested_node, str) or not requested_node.strip()): + return runner_response("blocked", "node must be non-empty text when provided") + if environment == "local": + if requested_node and requested_node not in LOCAL_NODE_NAMES: + return runner_response("blocked", "local binding does not match requested tester node", remediation="Use node: local or a local alias") + if operation == "platform_probe": + return _local_platform_probe(platform_name) + if operation == "network_probe": + return _network_probe(request) + if operation == "maestro": + return _local_maestro(platform_name, request) + return _local_mutation(platform_name, operation, request) + try: + worktree = worktree_for(value) + except ConfigError as exc: + return runner_response("blocked", "remote tester worktree could not be resolved", checks={ + "binding": _runner_check("ready", binding_id, "tester binding resolved"), + "worktree": _runner_check("blocked", "unresolved", "exact Orca worktree", str(exc)), + }, remediation="Register one exact environment/host/project/path worktree before using the remote tester") + return remote_runner_transport(value, request, worktree) + + +def cmd_run_exec(args: argparse.Namespace) -> int: + root = Path(args.root).resolve() + try: + request = json.load(sys.stdin) + except (json.JSONDecodeError, OSError): + response = runner_response("blocked", "runner stdin is not one valid JSON object", remediation="Send exactly one JSON object on stdin") + else: + try: + response = execute_runner_request(root, request) + except Exception: + print("agent-runtime runner: unexpected adapter failure", file=sys.stderr) + response = runner_response("environment-fail", "runner adapter failed before producing evidence", remediation="Inspect the tester binding and adapter environment") + print(json.dumps(response, ensure_ascii=False, separators=(",", ":"))) + return 0 + + def cmd_run_start(args: argparse.Namespace) -> int: root = Path(args.root).resolve() value = binding(load_config(root), args.binding) @@ -474,6 +1170,8 @@ def parser() -> argparse.ArgumentParser: start.add_argument("--binding", required=True) start.add_argument("--wait-ms", type=int, default=0) start.set_defaults(func=cmd_run_start) + execute = run_sub.add_parser("exec", parents=[root_parent]) + execute.set_defaults(func=cmd_run_exec) for name, func in (("status", cmd_run_status), ("attach", cmd_run_attach), ("resume", cmd_run_resume), ("stop", cmd_run_stop)): item = run_sub.add_parser(name, parents=[root_parent]) item.add_argument("--run", required=True) diff --git a/skills/agent-runtime/tests/test_runner.py b/skills/agent-runtime/tests/test_runner.py new file mode 100644 index 0000000..2637abf --- /dev/null +++ b/skills/agent-runtime/tests/test_runner.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import importlib.util +import base64 +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "agent.py" +SPEC = importlib.util.spec_from_file_location("agent_runtime_runner", SCRIPT) +assert SPEC and SPEC.loader +agent = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(agent) + + +class RunnerTests(unittest.TestCase): + def setUp(self) -> None: + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name) + pouch = self.root / ".pouch" + pouch.mkdir() + (pouch / "agent.yaml").write_text( + """schemaVersion: 1 +locations: + local-app: + environment: local + host: local + project: local-project + path: /tmp/writes + remote-mac: + environment: macmini + host: runtime:0123456789abcdef0123 + project: remote-project + path: /Users/tester/writes +profiles: + test-agent: + cli: codex + model: null + reasoningEffort: medium + permissionMode: workspace-write +bindings: + tester.android: + adapter: tester + location: local-app + profile: test-agent + tester.ios: + adapter: tester + location: remote-mac + profile: test-agent +""", + encoding="utf-8", + ) + + def tearDown(self) -> None: + self.temp.cleanup() + + def test_local_binding_routes_to_fixed_probe(self): + def command(argv, **_kwargs): + if argv == ["adb", "version"]: + return 0, "Android Debug Bridge version 1" + if argv == ["adb", "devices"]: + return 0, "List of devices attached\nemulator-5554\tdevice\n" + self.assertEqual(argv, ["adb", "-s", "emulator-5554", "shell", "getprop", "sys.boot_completed"]) + return 0, "1" + + with mock.patch.object(agent, "run_fixed_command", side_effect=command) as runner: + result = agent.execute_runner_request(self.root, { + "binding": "tester.android", "operation": "platform_probe", "platform": "android", "node": "local", + }) + self.assertEqual(result["status"], "ready") + self.assertEqual(set(result["checks"]), {"adb", "device", "boot"}) + self.assertEqual(runner.call_count, 3) + + def test_remote_binding_uses_orca_terminal_transport(self): + marker = "POUCH_RUNNER_RESULT_" + ("0" * 31) + "1" + payload = {"status": "ready", "observed": "Darwin/xcrun/simctl probe completed", "checks": {"macos": {"status": "ready"}}} + + def fake_orca(argv, _environment=None): + if argv[:2] == ["terminal", "create"]: + return {"terminal": {"handle": "term-1"}} + if argv[:2] == ["terminal", "read"]: + return {"terminal": {"tail": [marker + base64.b64encode(json.dumps(payload).encode()).decode()]}} + return {} + + with mock.patch.object(agent, "worktree_for", return_value={"id": "wt-1", "path": "/Users/tester/writes"}) as resolve, \ + mock.patch.object(agent, "orca_json", side_effect=fake_orca), \ + mock.patch.object(agent.uuid, "uuid4", return_value=mock.Mock(hex="0" * 31 + "1")): + result = agent.execute_runner_request(self.root, { + "binding": "tester.ios", "operation": "platform_probe", "platform": "ios", "node": "macmini", + }) + self.assertEqual(result["status"], "ready") + self.assertEqual(result["checks"]["binding"]["status"], "ready") + resolve.assert_called_once() + + def test_remote_transport_submits_startup_command_when_first_read_has_no_marker(self): + marker = "POUCH_RUNNER_RESULT_" + ("0" * 31) + "1" + payload = {"status": "ready", "observed": "Darwin/xcrun/simctl probe completed", "checks": {"macos": {"status": "ready"}}} + reads = [ + {"terminal": {"tail": ["Welcome to fish"]}}, + {"terminal": {"tail": [marker + base64.b64encode(json.dumps(payload).encode()).decode()]}}, + ] + + def fake_orca(argv, _environment=None): + if argv[:2] == ["terminal", "create"]: + return {"terminal": {"handle": "term-1"}} + if argv[:2] == ["terminal", "read"]: + return reads.pop(0) + return {} + + with mock.patch.object(agent, "worktree_for", return_value={"id": "wt-1", "path": "/Users/tester/writes"}), \ + mock.patch.object(agent, "orca_json", side_effect=fake_orca) as orca, \ + mock.patch.object(agent.uuid, "uuid4", return_value=mock.Mock(hex="0" * 31 + "1")): + result = agent.execute_runner_request(self.root, { + "binding": "tester.ios", "operation": "platform_probe", "platform": "ios", "node": "macmini", + }) + self.assertEqual(result["status"], "ready") + self.assertTrue(any(call.args[0][:2] == ["terminal", "send"] for call in orca.call_args_list)) + + def test_unknown_operation_and_command_failure_are_structured(self): + unknown = agent.execute_runner_request(self.root, { + "binding": "tester.android", "operation": "exec", "platform": "android", + }) + self.assertEqual(unknown["status"], "blocked") + with mock.patch.object(agent, "run_fixed_command", return_value=(127, "adb missing")): + failed = agent.execute_runner_request(self.root, { + "binding": "tester.android", "operation": "platform_probe", "platform": "android", + }) + self.assertEqual(failed["status"], "environment-fail") + + def test_json_stdout_is_single_protocol_object(self): + process = subprocess.run( + [sys.executable, str(SCRIPT), "run", "exec", "--root", str(self.root)], + input=json.dumps({"binding": "tester.android", "operation": "unknown", "platform": "android"}), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + self.assertEqual(process.returncode, 0) + self.assertEqual(json.loads(process.stdout)["status"], "blocked") + self.assertEqual(process.stderr, "") + + +if __name__ == "__main__": + unittest.main() diff --git a/skills/deployer/SKILL.md b/skills/deployer/SKILL.md index 07dc960..32532e7 100644 --- a/skills/deployer/SKILL.md +++ b/skills/deployer/SKILL.md @@ -83,6 +83,14 @@ ACK 的「运行测试环境」和回归前布环境会加载本 skill，按「C 5. `prod` 环境：停止本轨道的 Compose 操作，读取 [production-agent.md](references/production-agent.md)，在本机 `~/code/app00` 项目创建 fresh Agent terminal 并委派正式部署；当前 Agent 不得直接执行 `sync.py`、`remote.py` 或 SSH。 6. 只有明确点名的非正式远程 Compose 环境，才允许 `sync.py ` 后调用 `remote.py recreate`；只重启调用 `remote.py restart`，并用 `remote.py ps` 验证。 +test-preflight 的一次性部署入口为 `scripts/deploy/prepare.py`。它从 stdin 接收 +`{"operation":"prepare","environment":"test","project_root":"/abs/project","host":"...","public_host":"..."}`， +只解析项目 `.pouch/deployer/test`，复用 `lib.service_info` 的 `_config.yaml` 继承和 node +规则，确认 node 是本机后才按 `docker compose config`、`up -d --force-recreate`、`ps` +顺序准备单个 test 服务。输出始终是 `{status, observed, deployment_id?, checks?}` JSON； +prod、缺配置、非本机 node 为 `blocked`，Compose/命令失败为 `environment-fail`。诊断只写 +stderr，不输出 `.env` 或命令输出中的凭据。 + compose 用 `image:` 且用户说刚合并了代码：现有 tag 对不上这次提交时，**一句问**要不要先重建镜像，然后停。不要自己改 tag、不要走 builder。 新增/下线、deb、镜像 tag 升级、NAS 失败：再读 [compose.md](references/compose.md)。字段见 [config-reference.md](references/config-reference.md)。 diff --git a/skills/deployer/scripts/deploy/prepare.py b/skills/deployer/scripts/deploy/prepare.py new file mode 100755 index 0000000..62fcfb5 --- /dev/null +++ b/skills/deployer/scripts/deploy/prepare.py @@ -0,0 +1,193 @@ +#!/usr/bin/env python3 +"""One-shot JSON adapter for preparing the project's test Compose service. + +The adapter owns no deployment policy of its own: service discovery and +``_config.yaml`` inheritance stay in ``lib.service_info``. The command runner +is deliberately injectable so unit tests never need Docker, SSH, or a network. +""" + +from __future__ import annotations + +import json +import os +import socket +import subprocess +import sys +from pathlib import Path +from typing import Any, Callable + +_SCRIPT_DIR = Path(__file__).resolve().parent +if str(_SCRIPT_DIR) not in sys.path: + sys.path.insert(0, str(_SCRIPT_DIR)) + +import lib # noqa: E402 + + +STATUSES = {"ready", "blocked", "environment-fail"} +LOCAL_NODE_NAMES = {"local", "localhost", "127.0.0.1", "::1"} +CommandRunner = Callable[[list[str], Path], int] +NodeChecker = Callable[[str], bool] + + +def response(status: str, observed: str, *, deployment_id: str = "", checks: dict[str, Any] | None = None, remediation: str = "") -> dict[str, Any]: + if status not in STATUSES: + raise ValueError(f"invalid status: {status}") + value: dict[str, Any] = {"status": status, "observed": observed} + if deployment_id: + value["deployment_id"] = deployment_id + if checks: + value["checks"] = checks + if remediation: + value["remediation"] = remediation + return value + + +def _check(status: str, observed: str, expected: str, remediation: str = "") -> dict[str, str]: + value = {"status": status, "observed": observed, "expected": expected} + if remediation: + value["remediation"] = remediation + return value + + +def run_compose(command: list[str], cwd: Path) -> int: + """Run adapter-owned Compose argv without a shell or protocol pollution.""" + + try: + process = subprocess.run( + command, + cwd=str(cwd), + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + except OSError: + return 127 + return process.returncode + + +def execute_compose(executor: CommandRunner, command: list[str], cwd: Path) -> int: + """Normalize mock/OS command failures into an environment-fail exit code.""" + + try: + return int(executor(command, cwd)) + except (OSError, TypeError, ValueError): + return 127 + + +def node_is_local(node: str) -> bool: + """Accept only a local node; ``ssh -G`` is config inspection, never SSH.""" + + if node in LOCAL_NODE_NAMES: + return True + local_names = {socket.gethostname(), socket.getfqdn(), os.environ.get("HOSTNAME", "")} + if node in {name for name in local_names if name}: + return True + try: + process = subprocess.run( + ["ssh", "-G", node], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + except OSError: + return False + if process.returncode != 0: + return False + configured_host = "" + for line in (process.stdout or "").splitlines(): + key, _, value = line.partition(" ") + if key == "hostname": + configured_host = value.strip() + break + return configured_host in LOCAL_NODE_NAMES or configured_host in {name for name in local_names if name} + + +def _project_root(value: Any) -> Path | None: + if not isinstance(value, str) or not value.strip(): + return None + root = Path(value) + if not root.is_absolute() or ".." in root.parts or os.path.normpath(value) != value: + return None + return root + + +def prepare_test(request: Any, *, executor: CommandRunner = run_compose, local_node: NodeChecker = node_is_local) -> dict[str, Any]: + if not isinstance(request, dict): + return response("blocked", "prepare request must be a JSON object", remediation="Send one JSON object on stdin") + if request.get("operation") != "prepare": + return response("blocked", "only operation=prepare is supported", remediation="Use the test deployer prepare contract") + environment = request.get("environment") + if environment != "test": + return response("blocked", "only the test environment is permitted", remediation="Use environment: test; production is never accepted by this adapter") + root = _project_root(request.get("project_root")) + if root is None or not root.is_dir(): + return response("blocked", "project_root must be an existing absolute project directory", remediation="Provide the project root containing .pouch/deployer/test") + host = request.get("host") + if not isinstance(host, str) or not host.strip(): + return response("blocked", "host is required", remediation="Provide the test service host used by preflight") + public_host = request.get("public_host") + if public_host is not None and (not isinstance(public_host, str) or not public_host.strip()): + return response("blocked", "public_host must be non-empty text when provided") + + deploy_root = root / ".pouch" / "deployer" + test_dir = deploy_root / "test" + if not deploy_root.is_dir() or not test_dir.is_dir(): + return response("blocked", "test deployer directory is missing", remediation="Create .pouch/deployer/test with its Compose files and _config.yaml") + lib.PROJECT_ROOT = deploy_root + try: + info = lib.service_info("test", strict=False) + except (OSError, ValueError, SystemExit): + info = None + if not info: + return response("blocked", "test Compose service is not discoverable", remediation="Ensure test/compose.yaml and inherited _config.yaml resolve a node") + node = str(info.get("node") or "") + if not node or not local_node(node): + return response("blocked", "test Compose node is not confirmed local", remediation=f"Resolve node {node or ''!r} to this machine; this adapter never falls back to SSH") + compose_file = test_dir / "compose.yaml" + if not compose_file.is_file(): + return response("blocked", "test compose.yaml is missing", remediation="Provide .pouch/deployer/test/compose.yaml") + + deployment_id = f"test:{info['name']}" + base = ["docker", "compose", "-f", "compose.yaml"] + checks: dict[str, Any] = {} + code = execute_compose(executor, [*base, "config", "--quiet"], test_dir) + if code != 0: + checks["compose_config"] = _check("environment-fail", f"docker compose config exited {code}", "valid test Compose configuration", "Fix the test Compose file or its local environment") + return response("environment-fail", "test Compose configuration failed", deployment_id=deployment_id, checks=checks) + checks["compose_config"] = _check("ready", "validated", "valid test Compose configuration") + + code = execute_compose(executor, [*base, "up", "-d", "--force-recreate"], test_dir) + if code != 0: + checks["compose_up"] = _check("environment-fail", f"docker compose up exited {code}", "test services started", "Fix the local Docker/image/network failure and retry") + return response("environment-fail", "test Compose prepare failed", deployment_id=deployment_id, checks=checks) + checks["compose_up"] = _check("ready", "completed", "test services started") + + code = execute_compose(executor, [*base, "ps"], test_dir) + if code != 0: + checks["compose_ps"] = _check("environment-fail", f"docker compose ps exited {code}", "test services report status", "Inspect local test service health") + return response("environment-fail", "test Compose status check failed", deployment_id=deployment_id, checks=checks) + checks["compose_ps"] = _check("ready", "completed", "test services report status") + return response("ready", "test Compose service prepared", deployment_id=deployment_id, checks=checks) + + +def main() -> int: + try: + request = json.load(sys.stdin) + except (json.JSONDecodeError, OSError): + result = response("blocked", "prepare stdin is not one valid JSON object", remediation="Send exactly one JSON object on stdin") + else: + try: + result = prepare_test(request) + except Exception: + print("deployer prepare: unexpected adapter failure", file=sys.stderr) + result = response("environment-fail", "prepare adapter failed before producing evidence", remediation="Inspect the test deployer configuration") + print(json.dumps(result, ensure_ascii=False, separators=(",", ":"))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/deployer/tests/test_prepare.py b/skills/deployer/tests/test_prepare.py new file mode 100644 index 0000000..9d5f115 --- /dev/null +++ b/skills/deployer/tests/test_prepare.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import importlib.util +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "deploy" / "prepare.py" +SPEC = importlib.util.spec_from_file_location("deployer_prepare", SCRIPT) +assert SPEC and SPEC.loader +prepare = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(prepare) + + +class PrepareTests(unittest.TestCase): + def setUp(self) -> None: + self.temp = tempfile.TemporaryDirectory() + self.root = Path(self.temp.name) + deployer = self.root / ".pouch" / "deployer" / "test" + deployer.mkdir(parents=True) + (deployer / "compose.yaml").write_text("services:\n app:\n image: example:test\n", encoding="utf-8") + (deployer / "_config.yaml").write_text("node: devbox\nname: writes-test\n", encoding="utf-8") + self.request = {"operation": "prepare", "environment": "test", "project_root": str(self.root), "host": "http://test.local"} + + def tearDown(self) -> None: + self.temp.cleanup() + + def test_prod_is_rejected_without_inspecting_or_executing_compose(self): + calls = [] + result = prepare.prepare_test({**self.request, "environment": "prod"}, executor=lambda command, cwd: calls.append(command) or 0) + self.assertEqual(result["status"], "blocked") + self.assertEqual(calls, []) + + def test_missing_config_is_blocked(self): + (self.root / ".pouch" / "deployer" / "test" / "_config.yaml").unlink() + result = prepare.prepare_test(self.request, local_node=lambda node: True) + self.assertEqual(result["status"], "blocked") + self.assertIn("discoverable", result["observed"]) + + def test_mock_compose_prepare_is_ready(self): + calls = [] + + def executor(command, cwd): + calls.append((command, cwd)) + return 0 + + result = prepare.prepare_test(self.request, executor=executor, local_node=lambda node: node == "devbox") + self.assertEqual(result["status"], "ready") + self.assertEqual(calls[0][0][-2:], ["config", "--quiet"]) + self.assertEqual(calls[1][0][-3:], ["up", "-d", "--force-recreate"]) + self.assertEqual(calls[2][0][-1], "ps") + self.assertEqual(calls[0][1], self.root / ".pouch" / "deployer" / "test") + + def test_compose_failures_are_environment_fail(self): + for failed_step in range(3): + calls = [] + + def executor(command, cwd): + del cwd + calls.append(command) + return 1 if len(calls) == failed_step + 1 else 0 + + result = prepare.prepare_test(self.request, executor=executor, local_node=lambda node: True) + self.assertEqual(result["status"], "environment-fail") + self.assertNotEqual(result["status"], "ready") + + def test_json_stdout_has_no_diagnostics(self): + process = subprocess.run( + [sys.executable, str(SCRIPT)], + input=json.dumps({**self.request, "environment": "prod"}), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + self.assertEqual(process.returncode, 0) + self.assertEqual(json.loads(process.stdout)["status"], "blocked") + self.assertEqual(process.stderr, "") + + +if __name__ == "__main__": + unittest.main() diff --git a/skills/test-preflight/scripts/preflight.py b/skills/test-preflight/scripts/preflight.py index fde8bcb..3f7e9b0 100644 --- a/skills/test-preflight/scripts/preflight.py +++ b/skills/test-preflight/scripts/preflight.py @@ -391,7 +391,11 @@ def check_network(evidence: Evidence, config: Mapping[str, Any], row: Mapping[st statuses.append(status) evidence.add(f"{platform_name}-{index}-network-{probe_index}-{detail['name']}", platform=platform_name, node=node, target=url, observed=detail["observed"], expected=detail["expected"], status=status, remediation=detail.get("remediation", "")) elif isinstance(runner, dict) and runner.get("kind") == "contract": - status, observed, result = runner_call(runner, {"operation": "network_probe", "platform": platform_name, "node": node, "url": url, "probe": probe}) + # The runner is selected by the tester binding. Keep the binding + # in the request just like platform_probe, otherwise the shared + # agent-runtime adapter cannot resolve the actual node/worktree. + binding = text(tester.get("binding")) if isinstance(tester, dict) else "" + status, observed, result = runner_call(runner, {"operation": "network_probe", "platform": platform_name, "node": node, "binding": binding, "url": url, "probe": probe}) result = result or {"status": status, "observed": observed} complete = require_runner_checks(evidence, result, prefix=f"{platform_name}-{index}-network-{probe_index}", platform_name=platform_name, node=node, required={"dns", "tcp", "http"}) network_status = add_runner_checks(evidence, result, prefix=f"{platform_name}-{index}-network-{probe_index}", platform_name=platform_name, node=node, default_expected="DNS/TCP/HTTP reachable from actual tester node") @@ -488,8 +492,8 @@ def run(config: Mapping[str, Any], mode: str) -> Evidence: evidence.add(f"{platform_name}-{index}-install", platform=platform_name, node=node, target=target_path, observed="not-run", expected="verify and deployment ready", status="blocked", remediation="先修复前置检查；prepare 不伪造安装成功") evidence.add(f"{platform_name}-{index}-start", platform=platform_name, node=node, target=text(identity), observed="not-run", expected="install ready", status="blocked", remediation="安装前置失败，未启动 app") else: - install_request = {"operation": "install", "platform": platform_name, "node": node, "artifact": target_path, "identity": identity} - start_request = {"operation": "start", "platform": platform_name, "node": node, "identity": identity} + install_request = {"operation": "install", "platform": platform_name, "node": node, "binding": binding, "artifact": target_path, "identity": identity} + start_request = {"operation": "start", "platform": platform_name, "node": node, "binding": binding, "identity": identity} if is_local and (not isinstance(runner, dict) or runner.get("kind", "local") == "local"): if platform_name == "android": adb = shutil.which("adb") or "adb" diff --git a/skills/tester/scripts/catalog.py b/skills/tester/scripts/catalog.py index 7c64323..1c0df72 100755 --- a/skills/tester/scripts/catalog.py +++ b/skills/tester/scripts/catalog.py @@ -777,6 +777,10 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int: selected: list[dict[str, Any]] = [] skipped: list[dict[str, str]] = [] missing = set(wanted) + # Resolve authorization gates during selection so an explicit safety or + # mutating request is represented as skipped before any runner/session is + # created. run-script keeps the same check as a defense in depth. + vars_map = resolve_vars(root, meta) for case in cases: cid = str(case.get("id") or "") @@ -824,6 +828,9 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int: if r_i > max_i: skipped.append({"id": cid, "reason": f"risk={risk} 超过上限 {max_risk}"}) continue + if not authorized(vars_map, meta, case): + skipped.append({"id": cid, "reason": f"风险 {risk} 未授权"}) + continue selected.append(public_case(case)) if wanted and missing: @@ -855,7 +862,6 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int: return 2 # 按分类解析生效的 status_probe，让 Agent 不再自己按 catalog 猜作用范围。 - vars_map = resolve_vars(root, meta) status_probes: dict[str, dict[str, Any]] = {} for case in selected: cat = str(case.get("category") or "") diff --git a/skills/tester/scripts/mobile_bundle.py b/skills/tester/scripts/mobile_bundle.py index 3b4c8a9..d89bf51 100755 --- a/skills/tester/scripts/mobile_bundle.py +++ b/skills/tester/scripts/mobile_bundle.py @@ -210,6 +210,9 @@ def android_identity(path: Path, context: RuntimeContext) -> str: for command in commands: result = run_command(command, context=context) if result.code == 0: + plain = result.output.strip() + if re.fullmatch(r"[A-Za-z][A-Za-z0-9_.-]*", plain): + return plain match = re.search(r"(?:name|application-id)[=:]['\"]?([A-Za-z0-9_.]+)", result.output) if match: return match.group(1)