feat(test-preflight): add verify and prepare skill

skills/test-preflight/README.md

@@ -0,0 +1,13 @@
+# test-preflight
+
+Reusable preflight for tester matrices. It supports read-only `verify` and
+test-environment-only `prepare`, and writes per-check YAML/CSV evidence for
+artifacts, tester nodes, devices, network, services, and fixture/cleanup
+contracts.
+
+Run the self-tests with:
+
+```bash
+python3 -m unittest discover -s skills/test-preflight/tests
+python3 /home/ace/.codex/skills/.system/skill-creator/scripts/quick_validate.py skills/test-preflight
+```

skills/test-preflight/SKILL.md

@@ -0,0 +1,51 @@
+---
+name: test-preflight
+description: >-
+  Verify or prepare a mobile/integration test environment from a declared matrix:
+  artifacts, tester bindings, HOST/PUBLIC_HOST, service probes, and fixture/cleanup
+  contracts. Use before tester runs; it is not a deployment, build, or business-data skill.
+---
+
+# test-preflight
+
+Use this skill before `integration` or `journey` tests. It answers whether the
+declared environment can start a run and emits one YAML and one CSV evidence
+file. It never logs in, creates business data, builds/downloads an artifact, or
+deploys production.
+
+## Modes
+
+- `verify` is read-only: validate the matrix, artifact identity, binding, actual
+  tester-node platform/device state, DNS/TCP/HTTP probes, backend probes, and
+  fixture/cleanup declarations. It must not install, launch, or deploy.
+- `prepare` is side-effecting only within the test environment: call the
+  caller-supplied deployer command contract, run the same checks, then use the
+  tester-node command contract to install and launch the already supplied APK or
+  IPA. It never calls a builder and refuses `environment: prod`.
+
+The command is:
+
+```bash
+python3 <skill-dir>/scripts/preflight.py verify \
+  --config <matrix.yaml> --yaml <preflight.yaml> --csv <preflight.csv>
+python3 <skill-dir>/scripts/preflight.py prepare \
+  --config <matrix.yaml> --yaml <preflight.yaml> --csv <preflight.csv>
+```
+
+Read [references/schema.md](references/schema.md) for the input contract and
+[references/boundaries.md](references/boundaries.md) before wiring a real
+agent-runtime/deployer adapter. The included command-contract fixture is the
+safe way to test remote-node behavior without pretending that this machine is a
+Mac or that a deployment happened.
+
+Overall status is `ready`, `blocked`, or `environment-fail`:
+
+- `blocked` means a required declaration, artifact, binding, runner, fixture,
+  cleanup contract, or authorization is absent/unsafe.
+- `environment-fail` means a declared environment exists but DNS/TCP/HTTP,
+  device/toolchain, installation, launch, or deployment execution failed.
+- `ready` means every recorded check passed. A ready preflight permits the
+  caller to continue with `integration`; this skill does not run test cases.
+
+Do not treat a preflight report as a product result. A preflight failure means
+no client session may be created; the tester owns `not-run` propagation.

skills/test-preflight/agents/openai.yaml

@@ -0,0 +1,3 @@
+interface:
+  display_name: "Test Preflight"
+  short_description: "Verify or prepare tester environments"

skills/test-preflight/references/api_reference.md

@@ -0,0 +1,4 @@
+# Reference moved
+
+Read [schema.md](schema.md) for the matrix and evidence contract, and
+[boundaries.md](boundaries.md) for agent-runtime/deployer/builder integration.

skills/test-preflight/references/boundaries.md

@@ -0,0 +1,44 @@
+# Integration boundaries
+
+`test-preflight` is an orchestration and evidence layer. It consumes existing
+skills instead of implementing their policy or storage formats.
+
+| Boundary | Caller-supplied contract | What preflight does | What it must not do |
+| --- | --- | --- | --- |
+| agent-runtime | resolved `tester.<name>` binding: binding id, node, environment, project/path, and a node runner | validate that the binding is explicit and route node probes through the runner | parse Orca state, invent a host, or create/reuse a terminal |
+| deployer | `prepare` only: an argv command returning a JSON deployment receipt for `environment: test` | call it once and record its receipt/status | copy Compose/Argo logic, SSH directly, or accept production |
+| builder/obtain | caller-provided local artifact path (and optional remote path), hash, source commit, package/bundle identity | hash and inspect the exact file; pass it to install contract | build, guess an old artifact, or silently download/upload |
+
+## Deployer command contract
+
+`prepare` requires:
+
+```yaml
+deployment:
+  environment: test
+  command: [python3, /path/to/existing-deployer-adapter.py]
+```
+
+The command receives JSON on stdin:
+
+```json
+{"operation":"prepare","environment":"test","project_root":"/abs/project","host":"...","public_host":"..."}
+```
+
+It must return `{ "status": "ready", "observed": "..." }` (an optional
+`checks` mapping is recorded). Non-zero/invalid output is an
+`environment-fail`; absent command or a non-test environment is `blocked`.
+The adapter owns deployer-specific configuration, receipts, cleanup, and
+authorization.
+
+## Device contract
+
+The node runner owns remote execution. For a local node the implementation uses
+read-only `adb`/`xcrun simctl` checks in `verify`; `prepare` alone invokes
+`adb install`/launch or `simctl install`/launch. For a remote node the same
+operations are JSON requests to the runner. Missing xcrun on Linux, Android
+`unauthorized`/`offline`, a missing device, or a failed install/start is an
+`environment-fail`, not a product failure.
+
+No client session is created by this skill. The runner's `install` and `start`
+operations are the only device mutations, and are never called by `verify`.

skills/test-preflight/references/schema.md

@@ -0,0 +1,85 @@
+# test-preflight matrix schema
+
+The input is YAML. `mode` is supplied by the CLI, so a file may be reused for
+`verify` and `prepare`.
+
+```yaml
+schema_version: 1
+project_root: /absolute/project/root
+host: https://test.example.invalid
+public_host: https://public-test.example.invalid # optional
+service_probes:
+  - id: api-health
+    url: "{HOST}/health"
+    expected_status: [200]
+fixtures:
+  required: true
+  references: [isolated-test-account]
+cleanup:
+  required: true
+  steps: [delete-run-data]
+matrix:
+  - platform: android
+    tester:
+      binding: tester.android
+      node: local
+      runner: {kind: local}
+    artifact:
+      path: /absolute/dist/writes.apk
+      sha256: 012345...
+      package: page.writes
+      source_commit: 40-character-commit-sha
+      identity_command: [apkanalyzer, manifest, application-id, "{artifact}"]
+```
+
+Required matrix fields are `platform` (`android` or `ios`), `tester.binding`,
+`tester.node`, and `artifact.path`. Artifact identity is required as
+`package` for Android or `bundle_id` for iOS. `sha256` and `source_commit` are
+optional but recommended; when present they are checked exactly.
+
+`service_probes` must contain at least one URL. `{HOST}` and `{PUBLIC_HOST}`
+are expanded from the top-level values. Each probe is checked from the actual
+tester node as DNS, TCP, and HTTP. A status outside 2xx–4xx fails transport
+readiness; `expected_status` can require an exact allow-list. HTTP 401/403 can
+therefore be a valid reachability result when explicitly expected.
+
+`fixtures` and `cleanup` are declarations, not commands. A required fixture
+needs non-empty `references`; a required cleanup needs non-empty `steps` or
+`run`. For a read-only run, use `required: false` with a non-empty `reason`.
+The skill never executes these values.
+
+## Tester-node contract
+
+`tester.runner.kind: local` executes platform and network probes on the current
+node. It is only valid for `node: local`/`localhost`. A non-local node must use
+`kind: contract` and a non-empty argv `command`. The command receives one JSON
+request on stdin and returns one JSON object on stdout; no shell string is
+accepted.
+
+Requests have `operation` (`platform_probe`, `network_probe`, `install`, or
+`start`), `platform`, `node`, and operation-specific fields. The response is:
+
+```json
+{
+  "status": "ready",
+  "observed": "short evidence",
+  "checks": {
+    "macos": {"status": "ready", "observed": "Darwin 24", "expected": "Darwin"}
+  }
+}
+```
+
+`checks` is optional; each key becomes a separate evidence row. A non-zero,
+invalid, or unavailable runner is `environment-fail`. The adapter is expected
+to execute on the resolved tester node, not on the coordinator. See
+[boundaries.md](boundaries.md).
+
+## Evidence
+
+YAML contains `mode`, overall `status`, timestamps, a summary, and `checks`.
+CSV uses these stable columns:
+
+`check_id,mode,platform,node,target,observed,expected,status,remediation,timestamp`
+
+Every check must use one of `ready`, `blocked`, or `environment-fail`; the
+report does not invent `fail` or `not-run` testcase results.

skills/test-preflight/scripts/preflight.py

@@ -0,0 +1,549 @@
+#!/usr/bin/env python3
+"""Run the test-preflight verify/prepare contract."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import json
+import os
+import platform as host_platform
+import plistlib
+import re
+import shutil
+import socket
+import subprocess
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+import zipfile
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Mapping
+
+try:
+    import yaml
+except ImportError as exc:  # pragma: no cover
+    raise SystemExit("test-preflight 需要 PyYAML；请安装运行依赖后重试") from exc
+
+
+PLATFORMS = {"android", "ios"}
+STATUSES = {"ready", "blocked", "environment-fail"}
+EVIDENCE_FIELDS = [
+    "check_id", "mode", "platform", "node", "target", "observed", "expected",
+    "status", "remediation", "timestamp",
+]
+SHA256 = re.compile(r"^[0-9a-fA-F]{64}$")
+COMMIT = re.compile(r"^[0-9a-fA-F]{40}$")
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def text(value: Any) -> str:
+    if isinstance(value, (dict, list)):
+        return json.dumps(value, ensure_ascii=False, sort_keys=True)
+    return "" if value is None else str(value)
+
+
+def status_from(value: Any, default: str = "environment-fail") -> str:
+    return value if value in STATUSES else default
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for block in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def command_text(command: list[str], *, payload: Mapping[str, Any] | None = None) -> tuple[int, str]:
+    try:
+        process = subprocess.run(
+            command,
+            input=(json.dumps(payload, ensure_ascii=False) if payload is not None else None),
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            check=False,
+        )
+    except OSError as exc:
+        return 127, str(exc)
+    return process.returncode, process.stdout.strip()
+
+
+class Evidence:
+    def __init__(self, mode: str) -> None:
+        self.mode = mode
+        self.started_at = now()
+        self.rows: list[dict[str, str]] = []
+
+    def add(
+        self,
+        check_id: str,
+        *,
+        platform: str = "all",
+        node: str = "coordinator",
+        target: str = "",
+        observed: Any = "",
+        expected: Any = "",
+        status: str,
+        remediation: str = "",
+    ) -> None:
+        if status not in STATUSES:
+            raise ValueError(f"invalid evidence status: {status}")
+        self.rows.append({
+            "check_id": check_id,
+            "mode": self.mode,
+            "platform": platform,
+            "node": node,
+            "target": target,
+            "observed": text(observed),
+            "expected": text(expected),
+            "status": status,
+            "remediation": remediation,
+            "timestamp": now(),
+        })
+
+    def overall(self) -> str:
+        statuses = {row["status"] for row in self.rows}
+        if "environment-fail" in statuses:
+            return "environment-fail"
+        if "blocked" in statuses:
+            return "blocked"
+        return "ready"
+
+    def payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": 1,
+            "mode": self.mode,
+            "status": self.overall(),
+            "started_at": self.started_at,
+            "finished_at": now(),
+            "summary": {state: sum(row["status"] == state for row in self.rows) for state in sorted(STATUSES)},
+            "checks": self.rows,
+        }
+
+
+def load_config(path: Path) -> dict[str, Any]:
+    data = yaml.safe_load(path.read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError("配置根必须是 mapping")
+    return data
+
+
+def expand(value: str, config: Mapping[str, Any]) -> str:
+    return value.replace("{HOST}", text(config.get("host"))).replace("{PUBLIC_HOST}", text(config.get("public_host")))
+
+
+def normalize_url(value: str) -> str:
+    return value if urllib.parse.urlsplit(value).scheme else "http://" + value.lstrip("/")
+
+
+def argv(value: Any) -> list[str] | None:
+    if not isinstance(value, list) or not value or not all(isinstance(item, str) and item for item in value):
+        return None
+    return value
+
+
+def validate_shape(config: Mapping[str, Any], evidence: Evidence) -> bool:
+    ok = True
+    if config.get("schema_version") != 1:
+        evidence.add("config-schema", target="config", observed=config.get("schema_version"), expected="1", status="blocked", remediation="将 schema_version 设为 1")
+        ok = False
+    if not isinstance(config.get("project_root"), str) or not Path(text(config.get("project_root"))).is_absolute():
+        evidence.add("config-project-root", target="project_root", observed=config.get("project_root"), expected="absolute project_root", status="blocked", remediation="提供项目根绝对路径")
+        ok = False
+    if not isinstance(config.get("host"), str) or not config["host"].strip():
+        evidence.add("config-host", target="HOST", expected="non-empty host URL", status="blocked", remediation="提供 HOST")
+        ok = False
+    probes = config.get("service_probes")
+    if not isinstance(probes, list) or not probes:
+        evidence.add("config-service-probes", target="service_probes", expected="at least one probe", status="blocked", remediation="声明服务 URL 探针")
+        ok = False
+    else:
+        for index, probe in enumerate(probes):
+            if not isinstance(probe, dict) or not isinstance(probe.get("url"), str) or not probe["url"].strip():
+                evidence.add(f"config-service-probe-{index}", target=f"service_probes[{index}]", expected="url", status="blocked", remediation="为每个服务探针提供 URL")
+                ok = False
+    matrix = config.get("matrix")
+    if not isinstance(matrix, list) or not matrix:
+        evidence.add("config-matrix", target="matrix", expected="non-empty list", status="blocked", remediation="声明至少一个平台矩阵行")
+        return False
+    seen: set[str] = set()
+    for index, row in enumerate(matrix):
+        prefix = f"matrix-{index}"
+        if not isinstance(row, dict):
+            evidence.add(prefix, target=f"matrix[{index}]", expected="mapping", status="blocked", remediation="修正矩阵行")
+            ok = False
+            continue
+        platform_name = text(row.get("platform")).lower()
+        if platform_name not in PLATFORMS or platform_name in seen:
+            evidence.add(f"{prefix}-platform", platform=platform_name or "unknown", target=f"matrix[{index}].platform", expected="unique android or ios", status="blocked", remediation="为每个平台保留唯一矩阵行")
+            ok = False
+        seen.add(platform_name)
+        tester = row.get("tester")
+        if not isinstance(tester, dict) or not isinstance(tester.get("binding"), str) or not tester["binding"].startswith("tester."):
+            evidence.add(f"{prefix}-binding", platform=platform_name, target=f"matrix[{index}].tester.binding", expected="tester.<name>", status="blocked", remediation="由 agent-runtime 解析并传入 tester binding")
+            ok = False
+        if not isinstance(tester, dict) or not isinstance(tester.get("node"), str) or not tester["node"].strip():
+            evidence.add(f"{prefix}-node", platform=platform_name, target=f"matrix[{index}].tester.node", expected="actual tester node", status="blocked", remediation="提供实际 tester 节点名")
+            ok = False
+        artifact = row.get("artifact")
+        identity_key = "package" if platform_name == "android" else "bundle_id"
+        if not isinstance(artifact, dict) or not isinstance(artifact.get("path"), str) or not artifact["path"].strip():
+            evidence.add(f"{prefix}-artifact", platform=platform_name, target=f"matrix[{index}].artifact.path", expected="caller-provided artifact path", status="blocked", remediation="提供本轮 APK/IPA，不要让 preflight 猜产物")
+            ok = False
+        if not isinstance(artifact, dict) or not isinstance(artifact.get(identity_key), str) or not artifact[identity_key].strip():
+            evidence.add(f"{prefix}-identity", platform=platform_name, target=f"matrix[{index}].artifact.{identity_key}", expected="declared artifact identity", status="blocked", remediation=f"声明 artifact.{identity_key}")
+            ok = False
+        if isinstance(artifact, dict) and artifact.get("sha256") is not None and not SHA256.fullmatch(text(artifact.get("sha256"))):
+            evidence.add(f"{prefix}-sha256", platform=platform_name, target=f"matrix[{index}].artifact.sha256", expected="64 hex characters", status="blocked", remediation="提供正确 SHA-256 或删除字段")
+            ok = False
+        if isinstance(artifact, dict) and artifact.get("source_commit") is not None and not COMMIT.fullmatch(text(artifact.get("source_commit"))):
+            evidence.add(f"{prefix}-commit", platform=platform_name, target=f"matrix[{index}].artifact.source_commit", expected="40-character commit SHA", status="blocked", remediation="提供产物来源 commit SHA")
+            ok = False
+    return ok
+
+
+def artifact_identity(path: Path, platform_name: str, artifact: Mapping[str, Any]) -> tuple[str | None, str]:
+    if platform_name == "ios":
+        try:
+            with zipfile.ZipFile(path) as archive:
+                names = [name for name in archive.namelist() if re.fullmatch(r"Payload/[^/]+\.app/Info\.plist", name)]
+                if len(names) != 1:
+                    return None, "IPA 中没有唯一 Payload/*.app/Info.plist"
+                plist = plistlib.loads(archive.read(names[0]))
+            value = plist.get("CFBundleIdentifier")
+            return (value, "Info.plist CFBundleIdentifier") if isinstance(value, str) else (None, "Info.plist 缺少 CFBundleIdentifier")
+        except (OSError, zipfile.BadZipFile, plistlib.InvalidFileException) as exc:
+            return None, f"IPA 读取失败: {exc}"
+    command = argv(artifact.get("identity_command"))
+    if command:
+        command = [item.replace("{artifact}", str(path)) for item in command]
+    else:
+        tool = shutil.which("apkanalyzer") or shutil.which("aapt2") or shutil.which("aapt")
+        if not tool:
+            return None, "缺少 apkanalyzer/aapt2/aapt；无法读取 APK package identity"
+        command = [tool, "manifest", "application-id", str(path)] if tool.endswith("apkanalyzer") else [tool, "dump", "badging", str(path)]
+    code, output = command_text(command)
+    if code != 0:
+        return None, output[-500:] or f"identity command exited {code}"
+    match = re.search(r"(?:application-id|name)\s*[:=]\s*['\"]?([A-Za-z0-9_.]+)", output)
+    if not match:
+        match = re.search(r"package:\s+name='([^']+)'", output)
+    if not match:
+        match = re.fullmatch(r"\s*([A-Za-z0-9_.]+)\s*", output)
+    return (match.group(1), output[-500:]) if match else (None, "identity command output did not contain package")
+
+
+def record_artifact(evidence: Evidence, row: Mapping[str, Any], platform_name: str, node: str, index: int) -> bool:
+    artifact = row.get("artifact") or {}
+    path = Path(text(artifact.get("path"))).expanduser()
+    prefix = f"{platform_name}-{index}-artifact"
+    if not path.is_file():
+        evidence.add(prefix, platform=platform_name, node=node, target=str(path), expected="existing caller-provided artifact", status="blocked", remediation="补齐 APK/IPA 路径；不要默认构建")
+        return False
+    actual_hash = sha256(path)
+    expected_hash = artifact.get("sha256")
+    if expected_hash and actual_hash.lower() != text(expected_hash).lower():
+        evidence.add(prefix + "-hash", platform=platform_name, node=node, target=str(path), observed=actual_hash, expected=expected_hash, status="blocked", remediation="确认产物路径或 SHA-256 与本轮构建身份一致")
+        return False
+    evidence.add(prefix + "-hash", platform=platform_name, node=node, target=str(path), observed=actual_hash, expected=expected_hash or "file SHA-256 recorded", status="ready")
+    identity, observed = artifact_identity(path, platform_name, artifact)
+    expected_identity = text(artifact.get("package" if platform_name == "android" else "bundle_id"))
+    if identity != expected_identity:
+        evidence.add(prefix + "-identity", platform=platform_name, node=node, target=str(path), observed=identity or observed, expected=expected_identity, status="environment-fail", remediation="使用与 catalog 身份一致的 APK/IPA，并重新校验产物")
+        return False
+    evidence.add(prefix + "-identity", platform=platform_name, node=node, target=str(path), observed=identity, expected=expected_identity, status="ready")
+    if artifact.get("source_commit"):
+        evidence.add(prefix + "-commit", platform=platform_name, node=node, target=str(path), observed=artifact.get("source_commit"), expected=artifact.get("source_commit"), status="ready")
+    return True
+
+
+def local_platform_probe(platform_name: str) -> tuple[str, dict[str, dict[str, Any]]]:
+    if platform_name == "android":
+        adb = shutil.which("adb")
+        if not adb:
+            return "environment-fail", {"adb": {"status": "environment-fail", "observed": "missing", "expected": "Android SDK platform-tools", "remediation": "在 tester.android 节点安装并配置 adb"}}
+        code, output = command_text([adb, "devices", "-l"])
+        states = [line.split() for line in output.splitlines()[1:] if line.strip()]
+        device = next((parts[0] for parts in states if len(parts) > 1 and parts[1] == "device"), "")
+        bad = next((parts for parts in states if len(parts) > 1 and parts[1] in {"unauthorized", "offline"}), None)
+        if bad:
+            return "environment-fail", {"adb": {"status": "ready", "observed": output, "expected": "adb responds"}, "device": {"status": "environment-fail", "observed": " ".join(bad), "expected": "authorized device", "remediation": "解锁 Android 设备并接受 RSA 授权；offline 时重启 adb/设备"}}
+        if code != 0 or not device:
+            return "environment-fail", {"adb": {"status": "environment-fail", "observed": output, "expected": "one authorized device", "remediation": "连接并授权 Android device/emulator"}}
+        boot_code, boot = command_text([adb, "-s", device, "shell", "getprop", "sys.boot_completed"])
+        booted = boot_code == 0 and boot.strip() == "1"
+        return ("ready" if booted else "environment-fail", {
+            "adb": {"status": "ready", "observed": adb, "expected": "adb available"},
+            "device": {"status": "ready", "observed": device, "expected": "authorized device"},
+            "boot": {"status": "ready" if booted else "environment-fail", "observed": boot, "expected": "sys.boot_completed=1", "remediation": "等待 emulator boot complete 后重跑"},
+        })
+    if host_platform.system() != "Darwin":
+        return "environment-fail", {"macos": {"status": "environment-fail", "observed": host_platform.system(), "expected": "Darwin/macOS tester.ios node", "remediation": "将 tester.ios 指向可用 macOS，不要在 Linux 伪装 iOS"}}
+    xcrun = shutil.which("xcrun")
+    if not xcrun:
+        return "environment-fail", {"xcrun": {"status": "environment-fail", "observed": "missing", "expected": "xcrun and simctl", "remediation": "在 Mac 安装 Xcode Command Line Tools/Xcode"}}
+    code, output = command_text([xcrun, "simctl", "list", "devices", "available"])
+    booted = any("(Booted)" in line for line in output.splitlines())
+    return ("ready" if code == 0 and booted else "environment-fail", {
+        "macos": {"status": "ready", "observed": "Darwin", "expected": "Darwin/macOS"},
+        "xcrun": {"status": "ready" if code == 0 else "environment-fail", "observed": output[-500:], "expected": "simctl available", "remediation": "安装 Xcode 并接受许可"},
+        "boot": {"status": "ready" if booted else "environment-fail", "observed": "Booted" if booted else output[-300:], "expected": "Booted simulator", "remediation": "启动目标 Simulator 后重跑"},
+    })
+
+
+def runner_call(runner: Mapping[str, Any], request: Mapping[str, Any]) -> tuple[str, str, dict[str, Any]]:
+    command = argv(runner.get("command"))
+    if not command:
+        return "blocked", "runner command missing", {}
+    code, output = command_text(command, payload=request)
+    if code != 0:
+        return "environment-fail", output[-800:], {}
+    try:
+        result = json.loads(output)
+    except json.JSONDecodeError:
+        return "environment-fail", output[-800:], {}
+    if not isinstance(result, dict):
+        return "environment-fail", "runner returned non-object", {}
+    return status_from(result.get("status")), text(result.get("observed", "")), result
+
+
+def add_runner_checks(evidence: Evidence, result: Mapping[str, Any], *, prefix: str, platform_name: str, node: str, default_expected: str) -> str:
+    checks = result.get("checks")
+    if isinstance(checks, dict) and checks:
+        statuses: list[str] = []
+        for name, detail in checks.items():
+            detail = detail if isinstance(detail, dict) else {"observed": detail}
+            status = status_from(detail.get("status"), status_from(result.get("status")))
+            statuses.append(status)
+            evidence.add(f"{prefix}-{name}", platform=platform_name, node=node, target=text(detail.get("target", name)), observed=detail.get("observed", result.get("observed", "")), expected=detail.get("expected", default_expected), status=status, remediation=text(detail.get("remediation", "")))
+        return "environment-fail" if "environment-fail" in statuses else ("blocked" if "blocked" in statuses else "ready")
+    status = status_from(result.get("status"))
+    evidence.add(prefix, platform=platform_name, node=node, target=prefix, observed=result.get("observed", ""), expected=default_expected, status=status, remediation=text(result.get("remediation", "")))
+    return status
+
+
+def require_runner_checks(evidence: Evidence, result: Mapping[str, Any], *, prefix: str, platform_name: str, node: str, required: set[str]) -> bool:
+    checks = result.get("checks")
+    missing = sorted(required - set(checks)) if isinstance(checks, dict) else sorted(required)
+    if not missing:
+        return True
+    evidence.add(prefix + "-contract", platform=platform_name, node=node, target="runner response", observed={"missing": missing}, expected=sorted(required), status="blocked", remediation="让 tester 节点 adapter 返回每个必需的 adb/boot、macOS/xcrun/simctl 或 DNS/TCP/HTTP 证据")
+    return False
+
+
+def network_local(url: str) -> list[dict[str, Any]]:
+    parsed = urllib.parse.urlsplit(url)
+    host = parsed.hostname or ""
+    port = parsed.port or (443 if parsed.scheme == "https" else 80)
+    try:
+        addresses = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+    except OSError as exc:
+        return [{"name": "dns", "status": "environment-fail", "observed": str(exc), "expected": "resolvable hostname", "remediation": "检查 tester 节点 DNS/VPN/网络"}]
+    rows: list[dict[str, Any]] = [{"name": "dns", "status": "ready", "observed": sorted({item[4][0] for item in addresses}), "expected": "resolvable hostname"}]
+    try:
+        with socket.create_connection((host, port), timeout=5):
+            rows.append({"name": "tcp", "status": "ready", "observed": f"{host}:{port}", "expected": "TCP connection"})
+    except OSError as exc:
+        rows.append({"name": "tcp", "status": "environment-fail", "observed": str(exc), "expected": f"TCP {host}:{port}", "remediation": "检查端口、防火墙和测试节点到 HOST 的路由"})
+        return rows
+    try:
+        with urllib.request.urlopen(urllib.request.Request(url, method="GET"), timeout=5) as response:
+            code = response.status
+        rows.append({"name": "http", "status": "ready" if 200 <= code < 500 else "environment-fail", "observed": code, "expected": "HTTP 2xx-4xx", "remediation": "检查后端健康状态" if code >= 500 else ""})
+    except urllib.error.HTTPError as exc:
+        rows.append({"name": "http", "status": "ready" if 200 <= exc.code < 500 else "environment-fail", "observed": exc.code, "expected": "HTTP 2xx-4xx", "remediation": "检查后端服务，不把 5xx 当成可用" if exc.code >= 500 else ""})
+    except (OSError, urllib.error.URLError) as exc:
+        rows.append({"name": "http", "status": "environment-fail", "observed": str(exc), "expected": "HTTP response", "remediation": "检查服务监听、代理、TLS 和节点网络"})
+    return rows
+
+
+def check_network(evidence: Evidence, config: Mapping[str, Any], row: Mapping[str, Any], platform_name: str, node: str, index: int) -> bool:
+    tester = row.get("tester") or {}
+    runner = tester.get("runner") if isinstance(tester, dict) else None
+    is_local = node in {"local", "localhost", "127.0.0.1"}
+    statuses: list[str] = []
+    for probe_index, probe in enumerate(config.get("service_probes") or []):
+        raw_url = text(probe.get("url"))
+        if "{PUBLIC_HOST}" in raw_url and not text(config.get("public_host")).strip():
+            evidence.add(f"{platform_name}-{index}-network-{probe_index}-public-host", platform=platform_name, node=node, target=raw_url, expected="PUBLIC_HOST configured", status="blocked", remediation="提供 PUBLIC_HOST 或改用 HOST 探针")
+            statuses.append("blocked")
+            continue
+        url = normalize_url(expand(raw_url, config))
+        if not urllib.parse.urlsplit(url).hostname:
+            evidence.add(f"{platform_name}-{index}-network-{probe_index}-url", platform=platform_name, node=node, target=url, expected="valid probe URL", status="blocked", remediation="修正 service_probes URL 或 HOST/PUBLIC_HOST")
+            statuses.append("blocked")
+            continue
+        if is_local and (not isinstance(runner, dict) or runner.get("kind", "local") == "local"):
+            details = network_local(url)
+            expected_statuses = probe.get("expected_status")
+            for detail in details:
+                if detail["name"] == "http" and isinstance(expected_statuses, list) and detail["status"] == "ready" and detail["observed"] not in expected_statuses:
+                    detail = {**detail, "status": "environment-fail", "expected": expected_statuses, "remediation": "检查服务探针的 expected_status 或后端响应"}
+                status = status_from(detail["status"])
+                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})
+            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")
+            statuses.append(network_status if complete else "blocked")
+        else:
+            evidence.add(f"{platform_name}-{index}-network-{probe_index}", platform=platform_name, node=node, target=url, expected="node runner contract", status="blocked", remediation="远端 tester 必须提供 agent-runtime adapter runner")
+            statuses.append("blocked")
+    return "environment-fail" not in statuses and "blocked" not in statuses
+
+
+def check_fixture_cleanup(evidence: Evidence, config: Mapping[str, Any], platform_name: str, node: str, index: int) -> bool:
+    ok = True
+    fixture = config.get("fixtures")
+    valid_fixture = isinstance(fixture, dict) and (not fixture.get("required", True) or bool(fixture.get("references")))
+    if not valid_fixture:
+        evidence.add(f"{platform_name}-{index}-fixture", platform=platform_name, node=node, target="fixtures", expected="required fixture references", status="blocked", remediation="声明隔离测试账号/fixture；不要在运行中临时创建")
+        ok = False
+    else:
+        evidence.add(f"{platform_name}-{index}-fixture", platform=platform_name, node=node, target="fixtures", observed=fixture.get("references"), expected="declared fixture references", status="ready")
+    cleanup = config.get("cleanup")
+    valid_noop = isinstance(cleanup, dict) and cleanup.get("required") is False and bool(text(cleanup.get("reason")).strip())
+    valid_steps = isinstance(cleanup, dict) and (bool(cleanup.get("steps")) or bool(text(cleanup.get("run")).strip()))
+    if not (valid_noop or valid_steps):
+        evidence.add(f"{platform_name}-{index}-cleanup", platform=platform_name, node=node, target="cleanup", expected="steps/run or explicit read-only reason", status="blocked", remediation="声明 teardown/rollback；清理缺失时禁止开始测试")
+        ok = False
+    else:
+        evidence.add(f"{platform_name}-{index}-cleanup", platform=platform_name, node=node, target="cleanup", observed=cleanup, expected="cleanup contract declared", status="ready")
+    return ok
+
+
+def run(config: Mapping[str, Any], mode: str) -> Evidence:
+    evidence = Evidence(mode)
+    shape_ok = validate_shape(config, evidence)
+    deployment_ok = True
+    if mode == "prepare":
+        deployment = config.get("deployment")
+        environment = deployment.get("environment") if isinstance(deployment, dict) else None
+        command = argv(deployment.get("command")) if isinstance(deployment, dict) else None
+        if environment != "test":
+            evidence.add("deployment-environment", target="deployment.environment", observed=environment, expected="test", status="blocked", remediation="prepare 只允许 deployer 的 test 环境")
+            if not command:
+                evidence.add("deployment-command", target="deployment.command", expected="existing deployer adapter command", status="blocked", remediation="提供既有 deployer 的 JSON command contract；不要在 skill 内实现部署")
+            deployment_ok = False
+        elif not command:
+            evidence.add("deployment-command", target="deployment.command", expected="existing deployer adapter command", status="blocked", remediation="提供既有 deployer 的 JSON command contract；不要在 skill 内实现部署")
+            deployment_ok = False
+        else:
+            code, output = command_text(command, payload={"operation": "prepare", "environment": "test", "project_root": config.get("project_root"), "host": config.get("host"), "public_host": config.get("public_host")})
+            if code != 0:
+                evidence.add("deployment", target="test", observed=output[-800:], expected="deployer receipt status=ready", status="environment-fail", remediation="按 deployer 返回证据修复测试后端，再重跑 prepare")
+                deployment_ok = False
+            else:
+                try:
+                    receipt = json.loads(output)
+                except json.JSONDecodeError:
+                    receipt = {}
+                status = status_from(receipt.get("status"), "environment-fail") if isinstance(receipt, dict) else "environment-fail"
+                evidence.add("deployment", target="test", observed=receipt.get("observed", output[-800:]) if isinstance(receipt, dict) else output[-800:], expected="deployer receipt status=ready", status=status, remediation="检查既有 deployer 的测试环境 receipt" if status != "ready" else "")
+                deployment_ok = status == "ready"
+    if not shape_ok:
+        return evidence
+    for index, row in enumerate(config["matrix"]):
+        platform_name = text(row.get("platform")).lower()
+        tester = row.get("tester") or {}
+        node = text(tester.get("node"))
+        binding = text(tester.get("binding"))
+        evidence.add(f"{platform_name}-{index}-binding", platform=platform_name, node=node, target=binding, observed=binding, expected="resolved tester.<name> binding", status="ready")
+        artifact_ok = record_artifact(evidence, row, platform_name, node, index)
+        fixture_ok = check_fixture_cleanup(evidence, config, platform_name, node, index)
+        runner = tester.get("runner") if isinstance(tester, dict) else None
+        is_local = node in {"local", "localhost", "127.0.0.1"}
+        if is_local and (not isinstance(runner, dict) or runner.get("kind", "local") == "local"):
+            platform_status, details = local_platform_probe(platform_name)
+            for name, detail in details.items():
+                evidence.add(f"{platform_name}-{index}-platform-{name}", platform=platform_name, node=node, target=name, observed=detail.get("observed", ""), expected=detail.get("expected", "platform ready"), status=status_from(detail.get("status"), platform_status), remediation=detail.get("remediation", ""))
+            platform_ok = platform_status == "ready"
+        elif isinstance(runner, dict) and runner.get("kind") == "contract":
+            status, observed, result = runner_call(runner, {"operation": "platform_probe", "platform": platform_name, "node": node, "binding": binding})
+            result = result or {"status": status, "observed": observed}
+            required = {"adb", "device", "boot"} if platform_name == "android" else {"macos", "xcrun", "simctl", "boot"}
+            complete = require_runner_checks(evidence, result, prefix=f"{platform_name}-{index}-platform", platform_name=platform_name, node=node, required=required)
+            platform_status = add_runner_checks(evidence, result, prefix=f"{platform_name}-{index}-platform", platform_name=platform_name, node=node, default_expected="platform/device ready")
+            platform_ok = complete and platform_status == "ready"
+        else:
+            evidence.add(f"{platform_name}-{index}-platform-runner", platform=platform_name, node=node, target=node, expected="node runner contract", status="blocked", remediation="远端 tester 必须由 agent-runtime 提供可执行 runner")
+            platform_ok = False
+        network_ok = check_network(evidence, config, row, platform_name, node, index)
+        row_ok = artifact_ok and fixture_ok and platform_ok and network_ok
+        if mode == "prepare":
+            artifact = row.get("artifact") or {}
+            identity = artifact.get("package" if platform_name == "android" else "bundle_id")
+            target_path = text(artifact.get("remote_path") or artifact.get("path"))
+            if not row_ok or not deployment_ok:
+                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}
+                if is_local and (not isinstance(runner, dict) or runner.get("kind", "local") == "local"):
+                    if platform_name == "android":
+                        adb = shutil.which("adb") or "adb"
+                        serial = os.environ.get("ANDROID_SERIAL", "")
+                        install_code, install_output = command_text([adb, "-s", serial, "install", "-r", target_path] if serial else [adb, "install", "-r", target_path])
+                        install_result = {"status": "ready" if install_code == 0 else "environment-fail", "observed": install_output}
+                        launch = [adb, "-s", serial, "shell", "monkey", "-p", text(identity), "1"] if serial else [adb, "shell", "monkey", "-p", text(identity), "1"]
+                        start_code, start_output = command_text(launch) if install_code == 0 else (1, "install failed")
+                        start_result = {"status": "ready" if start_code == 0 else "environment-fail", "observed": start_output}
+                    else:
+                        install_result = {"status": "environment-fail", "observed": "local iOS install requires a macOS runner", "remediation": "use tester.ios contract runner on Mac"}
+                        start_result = {"status": "blocked", "observed": "not-run", "remediation": "install failed"}
+                    add_runner_checks(evidence, install_result, prefix=f"{platform_name}-{index}-install", platform_name=platform_name, node=node, default_expected="artifact installed")
+                    add_runner_checks(evidence, start_result, prefix=f"{platform_name}-{index}-start", platform_name=platform_name, node=node, default_expected="app launched")
+                elif isinstance(runner, dict) and runner.get("kind") == "contract":
+                    install_status, install_observed, install_result = runner_call(runner, install_request)
+                    add_runner_checks(evidence, install_result or {"status": install_status, "observed": install_observed}, prefix=f"{platform_name}-{index}-install", platform_name=platform_name, node=node, default_expected="artifact installed")
+                    if install_status == "ready":
+                        start_status, start_observed, start_result = runner_call(runner, start_request)
+                        add_runner_checks(evidence, start_result or {"status": start_status, "observed": start_observed}, prefix=f"{platform_name}-{index}-start", platform_name=platform_name, node=node, default_expected="app launched")
+                    else:
+                        evidence.add(f"{platform_name}-{index}-start", platform=platform_name, node=node, target=text(identity), observed="not-run", expected="install ready", status="blocked", remediation="安装失败，先修复 tester 节点")
+    return evidence
+
+
+def write_evidence(evidence: Evidence, yaml_path: Path, csv_path: Path) -> dict[str, Any]:
+    payload = evidence.payload()
+    yaml_path.parent.mkdir(parents=True, exist_ok=True)
+    csv_path.parent.mkdir(parents=True, exist_ok=True)
+    yaml_path.write_text(yaml.safe_dump(payload, allow_unicode=True, sort_keys=False), encoding="utf-8")
+    with csv_path.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(handle, fieldnames=EVIDENCE_FIELDS)
+        writer.writeheader()
+        writer.writerows(evidence.rows)
+    return {"status": payload["status"], "yaml": str(yaml_path), "csv": str(csv_path), "checks": len(evidence.rows), "summary": payload["summary"]}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Verify or prepare a test environment and emit YAML/CSV evidence")
+    parser.add_argument("mode", choices=("verify", "prepare"))
+    parser.add_argument("--config", required=True, type=Path)
+    parser.add_argument("--yaml", dest="yaml_path", type=Path, default=Path("preflight.yaml"))
+    parser.add_argument("--csv", dest="csv_path", type=Path, default=Path("preflight.csv"))
+    args = parser.parse_args()
+    try:
+        config = load_config(args.config)
+        evidence = run(config, args.mode)
+        result = write_evidence(evidence, args.yaml_path, args.csv_path)
+    except (OSError, ValueError, yaml.YAMLError) as exc:
+        print(json.dumps({"status": "blocked", "error": str(exc)}, ensure_ascii=False), file=sys.stderr)
+        return 2
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    return 0 if result["status"] == "ready" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

skills/test-preflight/tests/fixtures/contract_runner.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+"""Deterministic runner fixture for test-preflight contract tests."""
+
+import json
+import os
+import sys
+
+request = json.load(sys.stdin)
+scenario = os.environ.get("PREFLIGHT_RUNNER_SCENARIO", "ready")
+operation = request.get("operation")
+
+if operation in {"deployment", "prepare"}:
+    print(json.dumps({"status": "ready", "observed": "mock test deploy receipt"}))
+elif operation == "platform_probe":
+    platform = request.get("platform")
+    if scenario == "android-unauthorized" and platform == "android":
+        print(json.dumps({"status": "environment-fail", "checks": {
+            "adb": {"status": "ready", "observed": "adb 35", "expected": "adb responds"},
+            "device": {"status": "environment-fail", "observed": "emulator-5554 unauthorized", "expected": "authorized device", "remediation": "accept RSA authorization"},
+            "boot": {"status": "environment-fail", "observed": "unknown", "expected": "boot complete"},
+        }}))
+    elif scenario == "mac-unavailable" and platform == "ios":
+        print(json.dumps({"status": "environment-fail", "checks": {
+            "macos": {"status": "environment-fail", "observed": "Mac unavailable", "expected": "Darwin tester node", "remediation": "restore tester.ios Mac"},
+        }}))
+    else:
+        print(json.dumps({"status": "ready", "checks": {
+            "adb": {"status": "ready", "observed": "adb 35", "expected": "adb responds"},
+            "device": {"status": "ready", "observed": "authorized", "expected": "authorized device"},
+            "boot": {"status": "ready", "observed": "boot complete", "expected": "boot complete"},
+            "macos": {"status": "ready", "observed": "Darwin", "expected": "Darwin"},
+            "xcrun": {"status": "ready", "observed": "xcrun", "expected": "xcrun"},
+            "simctl": {"status": "ready", "observed": "available", "expected": "simctl"},
+        }}))
+elif operation == "network_probe":
+    if scenario == "host-unreachable":
+        print(json.dumps({"status": "environment-fail", "checks": {
+            "dns": {"status": "ready", "observed": "10.0.0.2", "expected": "DNS resolves"},
+            "tcp": {"status": "environment-fail", "observed": "connection refused", "expected": "TCP reachable", "remediation": "check HOST route"},
+        }}))
+    elif scenario == "http-failure":
+        print(json.dumps({"status": "environment-fail", "checks": {
+            "dns": {"status": "ready", "observed": "10.0.0.2", "expected": "DNS resolves"},
+            "tcp": {"status": "ready", "observed": "443 open", "expected": "TCP reachable"},
+            "http": {"status": "environment-fail", "observed": 503, "expected": "HTTP 2xx-4xx", "remediation": "check service health"},
+        }}))
+    else:
+        print(json.dumps({"status": "ready", "checks": {
+            "dns": {"status": "ready", "observed": "10.0.0.2", "expected": "DNS resolves"},
+            "tcp": {"status": "ready", "observed": "443 open", "expected": "TCP reachable"},
+            "http": {"status": "ready", "observed": 200, "expected": "HTTP 2xx-4xx"},
+        }}))
+elif operation in {"install", "start"}:
+    print(json.dumps({"status": "ready", "observed": operation + "ed mock artifact"}))
+else:
+    print(json.dumps({"status": "environment-fail", "observed": "unknown operation"}))

skills/test-preflight/tests/test_preflight.py

@@ -0,0 +1,209 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import json
+import os
+import plistlib
+import subprocess
+import sys
+import tempfile
+import unittest
+import zipfile
+from pathlib import Path
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+import preflight  # noqa: E402
+
+
+FIXTURE_RUNNER = Path(__file__).resolve().parent / "fixtures" / "contract_runner.py"
+
+
+class PreflightTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.apk = self.root / "writes.apk"
+        self.apk.write_bytes(b"mock apk")
+        self.ipa = self.root / "writes.ipa"
+        with zipfile.ZipFile(self.ipa, "w") as archive:
+            archive.writestr("Payload/Writes.app/Info.plist", plistlib.dumps({"CFBundleIdentifier": "page.writes"}))
+        self.identity = self.root / "identity.py"
+        self.identity.write_text("print('page.writes')\n", encoding="utf-8")
+
+    def tearDown(self) -> None:
+        self.temp.cleanup()
+
+    def runner(self, scenario: str = "ready") -> dict[str, object]:
+        return {"kind": "contract", "command": [sys.executable, str(FIXTURE_RUNNER)]}
+
+    def config(self, *, platform: str = "android", artifact: Path | None = None, node: str = "tester.android", scenario: str = "ready") -> dict[str, object]:
+        artifact = artifact or self.apk
+        artifact_row: dict[str, object] = {
+            "path": str(artifact),
+            "package": "page.writes" if platform == "android" else None,
+            "bundle_id": "page.writes" if platform == "ios" else None,
+        }
+        if platform == "android":
+            artifact_row["identity_command"] = [sys.executable, str(self.identity)]
+            artifact_row.pop("bundle_id")
+        else:
+            artifact_row.pop("package")
+        return {
+            "schema_version": 1,
+            "project_root": str(self.root),
+            "host": "https://test.example.invalid",
+            "public_host": "https://public.example.invalid",
+            "service_probes": [{"id": "health", "url": "{HOST}/health", "expected_status": [200]}],
+            "fixtures": {"required": True, "references": ["isolated-account"]},
+            "cleanup": {"required": True, "steps": ["cleanup-run-data"]},
+            "matrix": [{
+                "platform": platform,
+                "tester": {"binding": "tester.android" if platform == "android" else "tester.ios", "node": node, "runner": self.runner(scenario)},
+                "artifact": artifact_row,
+            }],
+        }
+
+    def write_config(self, data: dict[str, object]) -> Path:
+        path = self.root / "matrix.yaml"
+        import yaml
+        path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")
+        return path
+
+    def invoke(self, mode: str, data: dict[str, object]) -> tuple[int, dict[str, object], dict[str, object]]:
+        config = self.write_config(data)
+        yaml_path, csv_path = self.root / f"{mode}.yaml", self.root / f"{mode}.csv"
+        env = os.environ.copy()
+        env["PREFLIGHT_RUNNER_SCENARIO"] = "ready"
+        process = subprocess.run([
+            sys.executable, str(ROOT / "scripts" / "preflight.py"), mode,
+            "--config", str(config), "--yaml", str(yaml_path), "--csv", str(csv_path),
+        ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
+        import yaml
+        return process.returncode, yaml.safe_load(yaml_path.read_text(encoding="utf-8")), {"stdout": process.stdout, "stderr": process.stderr}
+
+    def test_missing_artifact_is_blocked_and_report_has_contract_columns(self):
+        data = self.config(artifact=self.root / "missing.apk")
+        code, report, _ = self.invoke("verify", data)
+        self.assertEqual(code, 1)
+        self.assertEqual(report["status"], "blocked")
+        row = next(row for row in report["checks"] if row["check_id"] == "android-0-artifact")
+        self.assertEqual(row["status"], "blocked")
+        self.assertEqual(set(row), set(preflight.EVIDENCE_FIELDS))
+
+    def test_android_unauthorized_is_environment_fail(self):
+        data = self.config()
+        config = self.write_config(data)
+        out = self.root / "unauthorized.yaml"
+        env = os.environ.copy()
+        env["PREFLIGHT_RUNNER_SCENARIO"] = "android-unauthorized"
+        process = subprocess.run([
+            sys.executable, str(ROOT / "scripts" / "preflight.py"), "verify", "--config", str(config), "--yaml", str(out), "--csv", str(self.root / "u.csv")
+        ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
+        report = json.loads(json.dumps(__import__("yaml").safe_load(out.read_text(encoding="utf-8"))))
+        self.assertEqual(process.returncode, 1)
+        self.assertEqual(report["status"], "environment-fail")
+        self.assertEqual(next(row for row in report["checks"] if row["check_id"].endswith("-device"))["status"], "environment-fail")
+
+    def test_linux_without_xcrun_is_environment_fail(self):
+        data = self.config(platform="ios", artifact=self.ipa, node="local")
+        data["matrix"][0]["tester"]["runner"] = {"kind": "local"}
+        with mock.patch.object(preflight.host_platform, "system", return_value="Linux"), mock.patch.object(preflight, "network_local", return_value=[{"name": "dns", "status": "ready", "observed": "ok", "expected": "DNS"}, {"name": "tcp", "status": "ready", "observed": "ok", "expected": "TCP"}, {"name": "http", "status": "ready", "observed": 200, "expected": "HTTP"}]):
+            result = preflight.run(data, "verify")
+        self.assertEqual(result.overall(), "environment-fail")
+        row = next(row for row in result.rows if row["check_id"].endswith("-platform-macos"))
+        self.assertEqual(row["status"], "environment-fail")
+
+    def test_mac_unavailable_contract_is_environment_fail(self):
+        data = self.config(platform="ios", artifact=self.ipa)
+        config_path = self.write_config(data)
+        env = os.environ.copy()
+        env["PREFLIGHT_RUNNER_SCENARIO"] = "mac-unavailable"
+        output = self.root / "mac.yaml"
+        process = subprocess.run([
+            sys.executable, str(ROOT / "scripts" / "preflight.py"), "verify", "--config", str(config_path), "--yaml", str(output), "--csv", str(self.root / "mac.csv")
+        ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
+        import yaml
+        report = yaml.safe_load(output.read_text(encoding="utf-8"))
+        self.assertEqual(process.returncode, 1)
+        self.assertEqual(report["status"], "environment-fail")
+        self.assertTrue(any(row["check_id"].endswith("-macos") and row["status"] == "environment-fail" for row in report["checks"]))
+
+    def test_missing_fixture_or_cleanup_is_blocked(self):
+        data = self.config()
+        data["fixtures"] = {"required": True, "references": []}
+        data["cleanup"] = {"required": True, "steps": []}
+        result = preflight.run(data, "verify")
+        self.assertEqual(result.overall(), "blocked")
+        self.assertTrue(any(row["check_id"].endswith("-fixture") and row["status"] == "blocked" for row in result.rows))
+        self.assertTrue(any(row["check_id"].endswith("-cleanup") and row["status"] == "blocked" for row in result.rows))
+
+    def test_host_unreachable_from_actual_runner_is_environment_fail(self):
+        data = self.config()
+        config_path = self.write_config(data)
+        env = os.environ.copy()
+        env["PREFLIGHT_RUNNER_SCENARIO"] = "host-unreachable"
+        output = self.root / "host.yaml"
+        process = subprocess.run([
+            sys.executable, str(ROOT / "scripts" / "preflight.py"), "verify", "--config", str(config_path), "--yaml", str(output), "--csv", str(self.root / "host.csv")
+        ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
+        import yaml
+        report = yaml.safe_load(output.read_text(encoding="utf-8"))
+        self.assertEqual(process.returncode, 1)
+        self.assertEqual(report["status"], "environment-fail")
+        self.assertTrue(any("network" in row["check_id"] and row["status"] == "environment-fail" for row in report["checks"]))
+
+    def test_http_500_is_environment_fail_not_product_fail(self):
+        data = self.config()
+        config_path = self.write_config(data)
+        env = os.environ.copy()
+        env["PREFLIGHT_RUNNER_SCENARIO"] = "http-failure"
+        output = self.root / "http.yaml"
+        process = subprocess.run([
+            sys.executable, str(ROOT / "scripts" / "preflight.py"), "verify", "--config", str(config_path), "--yaml", str(output), "--csv", str(self.root / "http.csv")
+        ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
+        import yaml
+        report = yaml.safe_load(output.read_text(encoding="utf-8"))
+        self.assertEqual(process.returncode, 1)
+        http_row = next(row for row in report["checks"] if row["check_id"].endswith("-http"))
+        self.assertEqual(http_row["observed"], "503")
+        self.assertEqual(http_row["status"], "environment-fail")
+
+    def test_prepare_without_deployer_is_blocked_and_never_installs(self):
+        data = self.config()
+        code, report, _ = self.invoke("prepare", data)
+        self.assertEqual(code, 1)
+        self.assertEqual(report["status"], "blocked")
+        self.assertFalse(any(row["check_id"].endswith("-install") and row["status"] == "ready" for row in report["checks"]))
+        self.assertTrue(any(row["check_id"] == "deployment-command" for row in report["checks"]))
+
+    def test_prepare_mock_contract_calls_install_and_start(self):
+        data = self.config()
+        data["deployment"] = {"environment": "test", "command": [sys.executable, str(FIXTURE_RUNNER)]}
+        code, report, _ = self.invoke("prepare", data)
+        self.assertEqual(code, 0)
+        self.assertEqual(report["status"], "ready")
+        ids = {row["check_id"] for row in report["checks"]}
+        self.assertIn("android-0-install", ids)
+        self.assertIn("android-0-start", ids)
+
+    def test_verify_never_calls_install_or_start(self):
+        data = self.config()
+        calls: list[str] = []
+        original = preflight.runner_call
+
+        def record(runner, request):
+            calls.append(str(request["operation"]))
+            return original(runner, request)
+
+        with mock.patch.object(preflight, "runner_call", side_effect=record):
+            result = preflight.run(data, "verify")
+        self.assertEqual(result.overall(), "ready")
+        self.assertNotIn("install", calls)
+        self.assertNotIn("start", calls)
+
+
+if __name__ == "__main__":
+    unittest.main()