feat(tester): add integration-first journey execution

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()

skills/tester/README.md

@@ -85,3 +85,50 @@ APK/IPA，为本轮创建隔离 worktree，在 Android 和 iOS tester 节点安
 - **列出**：能看到分类表和总条数；`.pouch/tester/cases.csv` 已由 skill 里的 `generate_cases_csv.py` 从 `cases/` 重新生成。每次运行的结果保存在独立的 `.pouch/tester/runs/<run-id>.*` 文件中。
 - **维护**：目标 id 能 `查看` 到，且检查通过。
 - **运行**：有本次 `runs/*.yaml` 快照和同名 `.csv` 成绩单（`executed` 列区分已跑 / 未跑）；没填 HOST 时会停，而不是编造失败。
+
+### 客户端用户旅程
+
+测试用例支持两种顶层类型：`test_type: integration` 和 `test_type: journey`。集成套件
+至少包括 `api-contract`、`authorization`、`build-contract`；客户端套件仍是
+`client-smoke`、`client-regression`、`safety`。没有显式范围时默认不选择 integration。
+
+执行顺序固定为 `test-preflight -> integration -> journey`。preflight 状态为 blocked 或
+environment-fail 时不创建客户端 session；integration 失败时依赖 journey 为 `not-run`。
+服务不可达是 environment-fail，契约断言错误是 integration fail，正确账号登录失败仍是
+journey fail。`tester run`/bundle 必须传入 caller 提供的 matrix：
+
+```bash
+<skill-dir>/tester run --preflight-config /abs/preflight.yaml \
+  --artifact android=/abs/app.apk --artifact ios=/abs/app.ipa
+```
+
+没有真实 deployer、Mac 或设备时，只能使用 test-preflight 的 command-contract mock fixture
+验证协议，不能把 mock 结果写成真实运行成功。
+
+客户端 E2E 用例使用 `client_suite: client-smoke | client-regression | safety`，并明确
+`journey`、`fixture`、`session`、`cleanup`、`platform`。标准生命周期是：
+
+```text
+preflight → 启动 → 登录 → 旅程 → 断言 → 清理 → 报告
+```
+
+正确测试账号登录失败记 `fail`；设备、产物或服务不可达记 `environment-fail`。fixture 或
+cleanup 缺失会在开始前阻断；前置失败后的剩余用例是 `not-run`，明确排除才是 `skipped`。
+`safety` 必须显式选择和授权，不进入默认 smoke/full。它与普通 tester 用例共用原有
+CSV/YAML 运行记录和汇总校验。
+
+### 汇总已有运行结果
+
+对已经生成的同一运行 CSV/YAML，使用显式路径汇总：
+
+```bash
+python3 <skill-dir>/scripts/summarize_run.py \
+  --csv .pouch/tester/runs/<run-id>.csv \
+  --yaml .pouch/tester/runs/<run-id>.yaml
+```
+
+脚本只从 CSV 的 `result` 列计算 `pass`、`fail`、`environment-fail`、`skipped`、
+`not-run` 五类状态，并要求五类计数和等于 CSV 数据行数。YAML 只用于校验
+`results` 数量和 testcase id 集合；数量、集合、重复/缺失 id 或 CSV 缺失/未知状态
+时返回非零并输出诊断。校验成功才表示这份汇总可作为运行报告；非零表示报告不一致，
+不能据此判定产品失败。

skills/tester/SKILL.md

@@ -145,3 +145,50 @@ python3 <skill-dir>/scripts/catalog.py show --root <project-root> --id <CASE_ID>
 4. 只有矩阵所有行都可用，才创建本轮测试 terminal/session，按 `select` 结果执行并记录成绩。
 
 测试环境由本机 deployer 触发，正式环境由 app00 Agent terminal 委派。随后按 agent 解析目标项目、目录和 Agent，更新配置并输出生效列表。移动端每个平台的产物位置必须由调用方给出，值为本地文件或 URL；缺了就停，不要猜 builder、CI release 或本机旧包。矩阵行记产物位置、身份和 `tester.<platform>` binding，并用 `<skill-dir>/tester run --artifact android=<path-or-url> --artifact ios=<path-or-url> --platform android,ios` 下载或取用、安装并准备本轮隔离 worktree。Android 在 `tester.android`、iOS 在 `tester.ios` 节点执行。未指定范围时等同「tag=smoke 且风险上限=默认且 automation=automated」。`select` 的 `run_together` 组按 [run.md](references/run.md) 整组执行。无论部署或测试成功、失败、超时还是中断，都要清理各自本轮创建的 terminal。
+
+### 客户端用户旅程标准
+
+### 双层测试与执行顺序
+
+用例顶层类型是 `test_type: integration | journey`。旧目录没有该字段时保持兼容：已有
+`client_suite`/`journey` 视为 journey，已有 `integration_suite` 视为 integration。集成
+套件至少包括 `api-contract`、`authorization`、`build-contract`；旅程套件仍固定为
+`client-smoke`、`client-regression`、`safety`。集成用例不需要客户端 session、platform
+或 journey 字段，不能被客户端套件选中。
+
+一次运行严格遵守：
+
+```text
+test-preflight (blocked/environment-fail -> 不创建 session)
+  -> integration
+  -> journey (integration fail -> not-run，不创建 session)
+```
+
+`test-preflight` 的报告是环境证据，不计入 testcase 总数。服务不可达记
+`environment-fail`；服务已达但契约断言错误记 integration `fail`；正确测试账号登录失败
+仍记 journey `fail`。tester 的 `run`/bundle 入口必须接收 caller 提供的 preflight matrix，
+没有 matrix 不得绕过闸门或伪造成功。
+
+声明 `client_suite`（迁移期也接受 `suite`）的用例采用固定的三套客户端集合：
+`client-smoke`、`client-regression`、`safety`。前两者分别覆盖最短只读主路径和隔离测试
+环境中的普通用户回归；`safety` 覆盖危险/权限边界场景，必须显式选择和授权，不能被
+默认 smoke/full 扩入。选择某套客户端旅程可用：
+
+```bash
+python3 <skill-dir>/scripts/catalog.py select --root <root> --suite client-smoke
+python3 <skill-dir>/scripts/catalog.py select --root <root> --suite client-regression --max-risk mutating
+python3 <skill-dir>/scripts/catalog.py select --root <root> --suite safety --max-risk destructive
+```
+
+客户端用例必须声明 `journey`、`fixture`、`session`、`cleanup` 和 `platform`。旅程包含
+`goal`、`entry_state`、`actions`、`assertions`、`final_state`、`failure_policy`；fixture
+至少声明测试账号；session 指向 catalog 中可复用的 browser/mobile session；cleanup 要有
+步骤或命令（只读也要显式说明无需清理）；platform 只能是 Android/iOS。具体字段见
+`references/schema.md`。平台适配只能改操作层，不能把真实客户端操作换成 API 调用。
+
+客户端运行器统一遵守 `preflight → 启动 → 登录 → 旅程 → 断言 → 清理 → 报告`。正确测试
+账号登录失败是产品 `fail`；设备、产物、服务或 session 不可达才是 `environment-fail`。
+fixture 或 cleanup 缺失会在 `catalog.py select` 的 preflight 阶段阻断，不产生虚假逐条结果。
+前置用例失败后剩余条目写 `not-run`，主动排除/未授权写 `skipped`。成绩仍写现有
+`runs/<run-id>.yaml` 并用 `generate_run_csv.py` 生成同名 CSV；清理失败作为 cleanup 证据
+单独记录，不添加新的 result 值。

skills/tester/references/concepts.md

@@ -40,6 +40,13 @@ catalog（整个测试集）
 | `how` | 怎么测：`steps` / `run` / `url` / `timeout` | test steps | 不是判定标准 |
 | `expected` | 怎么算过：`{kind, value}` 列表 | pytest `assert`；Robot `Should …` | 不是步骤 |
 | `run_together` | 必须成组执行的用例（有序、可选 continue_on_fail） | 场景组（scenario） | 不是 suite，不是 tag |
+| `client_suite` | 客户端用户旅程套件：`client-smoke` / `client-regression` / `safety` | mobile/browser E2E suite | 不是 `category`，也不是通用 tag；迁移期可读 `suite` 别名 |
+| `test_type` | 顶层测试类型：`journey` 或 `integration` | E2E vs service/contract test | 不是 driver；旧目录可由既有字段推断 |
+| `integration_suite` | 集成套件：`api-contract` / `authorization` / `build-contract` | API/contract test suite | 不是客户端 `client_suite`；迁移期可读 `suite` 别名 |
+| `journey` | 用户目标、入口、操作、UI 断言、最终状态和失败继续策略 | Cucumber scenario / Robot task | 不是 API contract |
+| `fixture` | 旅程开始前声明的测试账号与隔离数据 | pytest fixture | 不是运行中临时猜出的数据 |
+| `cleanup` | 本轮数据的 teardown 契约；缺失会在 preflight 阻断 | pytest teardown / Robot teardown | 不是测试通过后的可选备注 |
+| `platform` | 旅程适用的 `android`/`ios` 平台列表 | device matrix | 不是 tester Agent binding；binding 仍在命令参数/运行环境 |
 
 ### 结果词
 
@@ -99,6 +106,19 @@ catalog（整个测试集）
 - 因此它是**分类级前置条件**：`catalog.status_probe` 是全局默认，`categories[].status_probe` 可覆盖（缺省继承 / `false` 关闭 / mapping 整体替换）。对应 pytest 的 session fixture、Robot 的 Suite Setup。
 - 命中或不可达只影响用它的分类；只有所有选中分类都被挡住，整轮结论才是 `environment-fail`。`select` 会按分类返回解析后的生效值，Agent 不要自己读 catalog 猜。
 
+### 3.5 客户端套件和生命周期
+
+客户端套件固定为：
+
+- `client-smoke`：启动、正确账号登录和少量只读主路径，通常配 `tag=smoke`、`risk=readonly`。
+- `client-regression`：完整的普通用户旅程，可包含隔离环境中的 `mutating` 操作；不等于所有 full 用例。
+- `safety`：删除、权限边界、改密、停用或故障注入等危险旅程；必须显式选择和授权，不进入默认运行。
+
+每条旅程按 `preflight → 启动 → 登录 → 旅程 → 断言 → 清理 → 报告` 执行。登录失败在
+客户端和正确测试账号都已就绪时记 `fail`；设备、产物、服务或 session 不可达才记
+`environment-fail`。fixture/cleanup 缺失是 contract block，不伪装成逐条结果。前置失败
+后的剩余条目是 `not-run`，主动排除仍是 `skipped`。
+
 ## 4. 不采用的词
 
 | 词 | 不用它的原因 |

skills/tester/references/run.md

@@ -9,6 +9,11 @@ Agent、执行目标和临时 terminal 的完整协议见 [execution.md](executi
 
 ## 0. 建立测试矩阵并完成全量 preflight
 
+`test-preflight` 是 integration 和 journey 的共同硬闸。tester 必须先调用其 `verify`
+或 `prepare`，保存独立的 preflight YAML/CSV；状态为 `blocked` 或 `environment-fail`
+时不创建 browser/mobile session，所有尚未执行的 testcase 记 `not-run`。preflight 证据
+不算 testcase，也不进入 testcase CSV/YAML 的总数。
+
 测试不是“找到一个能跑的设备就开始”。每轮先建立一份测试矩阵；移动端每个目标产物
 必须单独占一行：
 
@@ -76,11 +81,30 @@ python3 <skill-dir>/scripts/catalog.py select --root <root> \
   [--category CAT] [--id ID]... [--tag smoke|full|slow] \
   [--max-risk <risk>] [--driver script|browser|mobile] \
   [--automation automated|assisted|manual|any] \
+  [--test-type integration|journey] [--suite <client-or-integration-suite>] \
   [--platform android|ios]... \
   [--artifact android=<path-or-url>] [--artifact ios=<path-or-url>] \
   [--binding android=tester.android] [--binding ios=tester.ios]
 ```
 
+客户端旅程还可按标准套件选择：
+
+```bash
+python3 <skill-dir>/scripts/catalog.py select --root <root> --suite client-smoke
+```
+
+`client-smoke`、`client-regression`、`safety` 是 `client_suite` 的固定值，不替换既有
+`tag=smoke|full|slow`。`safety` 不参与未指定范围的默认选择。
+
+集成测试显式选择：
+
+```bash
+python3 <skill-dir>/scripts/catalog.py select --root <root> --test-type integration
+python3 <skill-dir>/scripts/catalog.py select --root <root> --suite api-contract
+```
+
+未指定 `--test-type`/`--suite` 时，默认选择不会包含 `test_type=integration`。
+
 选中任一 `driver=mobile` 用例时，`--platform`、每个平台的 `--artifact` 和对应
 `tester` binding 是硬性参数；`catalog.py select` 会检查产物文件/下载地址和
 `.pouch/agent.yaml` 中的 binding。缺任何一项直接返回非零并打印 `preflight: blocked`，
@@ -97,7 +121,18 @@ python3 <skill-dir>/scripts/catalog.py select --root <root> \
 
 无命中：停下说明筛选条件，不要扩大范围。
 
-## 2. 运行前闸
+## 2. 双层执行
+
+preflight 通过后，先按 catalog 顺序执行 `integration`，再按客户端套件顺序执行
+`journey`。集成服务不可达是 `environment-fail`；服务已达但 API/权限/数据契约断言错误
+是 `fail`。任何 integration `fail` 都阻断依赖的 journey，并把它们记为 `not-run`；此时
+不得创建客户端 session。正确测试账号登录失败发生在客户端已启动后，仍是产品 `fail`。
+
+运行器可复用 `scripts/two_tier.py` 的 `run_integration_first`，它只接收外部 driver
+回调，不实现 API、浏览器或移动执行器；mock fixture 只能验证调用顺序和失败传播，不能
+冒充真实部署、Mac 或设备成功。
+
+## 3. 运行前闸
 
 - `HOST` 未设（环境变量 / `local.yaml` 都没有）：全部停，environment-fail。
 - 命中条的 `risk` 在 catalog `authorize` 里有对应变量：变量不是 `true` 则该条 `skipped`。有 `gate` 时还要 `catalog.gates` 对应变量为 `true`；`assume` 不成立同样 `skipped`。即使用户点了 id 也不执行。
@@ -114,7 +149,7 @@ python3 <skill-dir>/scripts/catalog.py select --root <root> \
 - 探测不可达或命中 `stop_if`：在 preflight 阶段记录对应分类/平台的不可用项并停止整轮；
   先返回用户处理，不创建后续测试 session。只有所有本轮目标环境通过，才进入用例执行。
 
-## 2.1 移动端平台环境闸门与用户引导
+## 3.1 移动端平台环境闸门与用户引导
 
 `driver=mobile` 不是一个可以互换的平台。运行前必须根据 catalog session、用例变量或用户指定，明确本轮是 Android 还是 iOS；无法判断时先停下询问平台，不得猜测。两类平台分别执行以下闸门：
 
@@ -175,7 +210,7 @@ Android+iOS 默认是全量 preflight 闸门：任一项不满足，另一平台
 
 环境引导只负责让用户补齐前置条件；除 catalog 已明确授权的 session 初始化外，不自动安装 SDK、配对真机、修改系统安全设置或上传产物。
 
-## 3. Session
+## 4. Session
 
 看命中条用到的 `session` 键：
 
@@ -184,7 +219,7 @@ Android+iOS 默认是全量 preflight 闸门：任一项不满足，另一平台
 - `reuse: run`：整轮结束再关。`persist`：不关。
 - 中途 session 挂了：后续同 driver 标 environment-fail，不要标产品失败。
 
-## 4. 执行
+## 5. 执行
 
 按 select 返回的顺序，一条一条来。select 按 `catalog.yaml` 的 `categories` 顺序，同一分类内按该 yaml 列表顺序（不是文件名字母序）。默认失败即停（产品 `fail` 或探测到的不安全状态）。用户说继续 / keep-going 才往后跑。
 
@@ -216,7 +251,7 @@ python3 <skill-dir>/scripts/catalog.py run-script --root <root> --id <id>
 
 挂到已有 session，执行 `how.steps`，对照 `expected`。必须留下可见文本、URL 或交互证据。
 
-## 5. 记录与报告
+## 6. 记录与报告
 
 每条：`id, result, evidence, notes`。`result` 只能是 `pass` / `fail` / `environment-fail` / `skipped` / `not-run`。`skipped` 是主动不跑（retired / pending / 超风险上限 / 未授权 / `assume` 不成立），`not-run` 是没轮到（失败即停后剩下的 / 被前置整组挡住的 / 需人工而无人值守的）。
 
@@ -227,7 +262,23 @@ python3 <skill-dir>/scripts/catalog.py run-script --root <root> --id <id>
 python3 <skill-dir>/scripts/generate_run_csv.py --root <root> --run <yaml>
 ```
 
-这会写出与快照同名的 `.pouch/tester/runs/<run-id>.csv`（`executed=yes|no`），不会修改 `cases.csv`。对话里两个路径都报。
+这会写出与快照同名的 `.pouch/tester/runs/<run-id>.csv`（`executed=yes|no`），不会修改
+`cases.csv`。对话里两个路径都报。
+
+需要复核已有成绩单时，显式同时传入同一次运行的 CSV 和 YAML：
+
+```bash
+python3 <skill-dir>/scripts/summarize_run.py \
+  --csv .pouch/tester/runs/<run-id>.csv \
+  --yaml .pouch/tester/runs/<run-id>.yaml
+```
+
+汇总状态只读取 CSV `result` 列，必须输出五类 `pass / fail / environment-fail /
+skipped / not-run`，且计数和等于 CSV 数据行数。汇总前会检查 CSV 数据行数、CSV 唯一
+testcase id 数、YAML `results` 数量和 CSV/YAML testcase id 集合；重复或缺失 id、缺失或
+未知 CSV 状态、任一数量或集合不一致都会非零退出并给出诊断。YAML 的 `counts` 或其它
+历史汇总字段不会参与状态计数。退出码 0 表示汇总校验成功，非零表示报告不一致或输入
+无效，不等同于产品测试失败。
 
 对话报告：
 

skills/tester/references/schema.md

@@ -14,6 +14,10 @@ runs/<run-id>.yaml     每次运行快照（执行源）；run-id 必须唯一
 runs/*.csv             同一次运行的成绩单；generate_run_csv.py 从对应 yaml 生成
 ```
 
+两类用例共用同一套 cases/runs 和五态报告，不另造成绩单。新目录显式声明顶层
+`test_type: journey | integration`；旧目录没有该字段时按已有 `client_suite`/`journey`
+或 `integration_suite` 推断。`category` 仍是文件/执行顺序容器，不被套件字段替代。
+
 ## catalog.yaml
 
 | 字段 | 含义 |
@@ -29,6 +33,8 @@ runs/*.csv             同一次运行的成绩单；generate_run_csv.py 从对
 | `id_pattern` | 用例 id 正则 |
 | `run_together` | 可选。成组执行：`[{ids: [TC-…, TC-…], continue_on_fail: true}]`。点到组内任一条，`select` 会按 `ids` 顺序把整组扩进命中列表。`continue_on_fail: true` 时组内前一条产品 fail 仍跑后一条（返回/收尾）。一个 id 只能属于一组。 |
 | `status_probe` | 可选。顶层默认值，一个**分类级前置条件**：`{url, stop_if: {字段: [命中值]}}`。运行前请求 `url`，响应命中任一 `stop_if` 条件即视为前置不满足。分类项可用同名键覆盖：缺省继承顶层、`false` 关闭、mapping 整体替换。命中或不可达只影响用它的分类，不是产品失败。业界对应 pytest 的 session fixture / Robot 的 Suite Setup。`select` 会按分类返回解析后的生效值，`url` 已用 `vars` 展开 |
+| `client_suites` | 可选的标准套件声明；出现时必须完整包含 `client-smoke`、`client-regression`、`safety`。缺省运行仍按 `tag` 兼容旧目录 |
+| `integration_suites` | 可选的集成套件声明；标准值为 `api-contract`、`authorization`、`build-contract` |
 
 `local.yaml` 与环境变量覆盖 `HOST` 和授权开关。模板中的 `HOST` 默认是
 `127.0.0.1`；远程目标必须显式覆盖。查找顺序：环境变量 > `local.yaml` >
@@ -70,6 +76,7 @@ terminal；运行结束用 `orca terminal close --terminal <handle>` 关闭本
 | 字段 | 必填 | 含义 |
 |------|------|------|
 | `id` | 是 | 匹配 `id_pattern` |
+| `test_type` | 新用例是 | `journey` 或 `integration`；缺省兼容旧目录并按字段推断 |
 | `category` | 是 | 必须等于文件声明的 category |
 | `title` | 是 | 短名 |
 | `what` | 是 | 一句可观察的「测什么」 |
@@ -88,6 +95,62 @@ terminal；运行结束用 `orca terminal close --terminal <handle>` 关闭本
 | `teardown` | 否 | 跑完回退 |
 | `notes` | 否 | 备注 |
 
+### 集成测试用例契约
+
+`test_type: integration` 的用例不创建 browser/mobile session，也不能声明 `platform` 或
+客户端 `journey`。必须有 `integration_suite`（迁移期可用 `suite`）且值为
+`api-contract`、`authorization` 或 `build-contract`；`fixture` 必须是非空 mapping；
+`request`、`input`、`call`、`integration` 之一或旧式 `how` 用于描述调用；`expected` 是
+协议/数据断言。`mutating`/`destructive` 集成用例必须提供 `cleanup.steps` 或
+`cleanup.run`（只读探针可用 `required: false` + `reason`）。
+
+选择 `--suite api-contract` 或 `--test-type integration` 才会进入集成阶段；无显式类型
+时的默认选择不会包含 integration。
+
+### 客户端旅程用例契约
+
+声明 `client_suite` 后，以下字段都必须存在；`catalog.py validate` 和 `select` 使用同一
+套检查。契约缺失是 `preflight: blocked`，不生成逐条产品失败。
+
+| 字段 | 必填 | 约束 |
+|------|------|------|
+| `client_suite` | 是 | `client-smoke`、`client-regression` 或 `safety`；安全套件必须显式选择，不属于默认 smoke/full |
+| `journey.goal` | 是 | 用户目标，不写接口目标 |
+| `journey.entry_state` | 是 | 启动/登录前的可观察入口状态 |
+| `journey.actions` | 是 | 非空用户操作列表；真实 App/browser 驱动从这里开始 |
+| `journey.assertions` | 是 | 非空可观察 UI 断言列表；最终业务状态可另由 `expected` 的 api-field/api-status 佐证 |
+| `journey.final_state` | 是 | 旅程完成后的业务状态 |
+| `journey.failure_policy` | 是 | `stop` 或 `continue`；默认不隐式吞掉前置失败 |
+| `fixture` | 是 | mapping，至少有 `account`；账号和数据由 app catalog 提供，skill 不写具体账号 |
+| `session` | 是 | 复用的 browser/mobile session 名，必须在 catalog.sessions 声明 |
+| `cleanup` | 是 | mapping，必须有非空 `steps`/`run`；只读旅程也要用 `required: false` + `reason` 显式说明无需清理 |
+| `platform` | 是 | `android`/`ios` 的字符串或列表，不能重复；平台差异只能进入执行适配层 |
+
+一个最小形状如下（值是占位符，不是 app 规则）：
+
+```yaml
+client_suite: client-smoke
+platform: [android, ios]
+journey:
+  goal: "用户完成一个可观察的客户端目标"
+  entry_state: logged_out
+  actions: ["启动客户端", "输入测试账号并登录", "完成目标操作"]
+  assertions: ["看到目标页面", "看到成功提示"]
+  final_state: "目标状态已保存"
+  failure_policy: stop
+fixture:
+  account: test-role
+  data: []
+session: mobile
+cleanup:
+  steps: ["删除本轮 run-id 创建的数据"]
+```
+
+账号登录属于 session 初始化：正确测试账号登录失败是产品 `fail`；设备、产物或服务
+不可达才是 `environment-fail`。fixture/cleanup 缺失必须在启动、登录和任何旅程操作前
+阻断。旅程失败后，尚未轮到的用例记 `not-run`；风险、生命周期或明确范围排除记
+`skipped`。清理失败单独写入运行快照的 cleanup 证据，不改造为一个新的 result 值。
+
 ## how
 
 按 `driver` 分派；`automation=manual` 时没有 `driver`，只有 `how.steps`（人工步骤）。

skills/tester/scripts/catalog.py

@@ -18,6 +18,16 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
+from journey import (
+    CLIENT_SUITES,
+    INTEGRATION_SUITES,
+    TEST_TYPES,
+    contract_errors,
+    preflight as journey_preflight,
+    suite as case_suite,
+    test_type,
+)
+
 try:
     import yaml
 except ImportError:  # pragma: no cover
@@ -368,7 +378,7 @@ def cmd_list(root: Path) -> int:
     pending = [c for c in cases if c.get("lifecycle") == "pending"]
     print(f"## 测试目录  active {len(active)} / pending {len(pending)} / retired {len(retired)}")
     print()
-    headers = ["分类", "标题", "条数", "smoke"] + risks + list(DRIVERS)
+    headers = ["分类", "标题", "条数", "journey", "integration", "smoke"] + risks + list(DRIVERS)
     print("| " + " | ".join(headers) + " |")
     print("| " + " | ".join("---" for _ in headers) + " |")
 
@@ -386,7 +396,14 @@ def cmd_list(root: Path) -> int:
         title = (cats_meta.get(cat) or {}).get("title") or (
             group[0].get("_file_title") if group else ""
         )
-        row = [cat, str(title), str(len(group)), str(sum(1 for c in group if c.get("tag") == "smoke"))]
+        row = [
+            cat,
+            str(title),
+            str(len(group)),
+            str(sum(1 for c in group if test_type(c) == "journey")),
+            str(sum(1 for c in group if test_type(c) == "integration")),
+            str(sum(1 for c in group if c.get("tag") == "smoke")),
+        ]
         for risk in risks:
             row.append(str(sum(1 for c in group if c.get("risk") == risk)))
         for driver in DRIVERS:
@@ -781,6 +798,14 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
                 continue
             if args.tag and case.get("tag") != args.tag:
                 continue
+            if args.test_type and test_type(case) != args.test_type:
+                continue
+            if args.suite and case_suite(case) != args.suite:
+                continue
+            # A mixed catalog keeps legacy cases runnable, but the implicit
+            # default is the client journey surface. Integration is opt-in.
+            if not args.test_type and not args.suite and test_type(case) == "integration":
+                continue
             if args.driver and case.get("driver") != args.driver:
                 continue
         automation = str(case.get("automation") or "automated")
@@ -810,6 +835,13 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
 
     selected, together = expand_run_together(meta, cases, selected, skipped, max_i, automation_filter)
 
+    # The historical key is kept in the JSON, but the gate now validates both
+    # test types before any caller can start a phase.
+    journey_gate = journey_preflight(selected, sessions=meta.get("sessions") or {})
+    if journey_gate["status"] == "blocked":
+        print(json.dumps({"preflight": journey_gate, "journey_preflight": journey_gate}, ensure_ascii=False, indent=2))
+        return 2
+
     preflight = mobile_preflight(
         root,
         meta,
@@ -840,6 +872,11 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
                 "vars": vars_map,
                 "artifacts": meta.get("artifacts") or {},
                 "preflight": preflight,
+                "journey_preflight": journey_gate,
+                "test_types": {
+                    "journey": sum(1 for case in selected if test_type(case) == "journey"),
+                    "integration": sum(1 for case in selected if test_type(case) == "integration"),
+                },
                 "skipped": skipped,
                 "run_together": together,
                 "status_probe": status_probes,
@@ -922,6 +959,28 @@ def validate(root: Path) -> list[str]:
     if not isinstance(sessions, dict):
         errors.append("catalog.sessions 必须是 mapping")
         sessions = {}
+    client_suites = meta.get("client_suites")
+    if client_suites is not None:
+        if not isinstance(client_suites, dict):
+            errors.append("catalog.client_suites 必须是 mapping")
+        else:
+            unknown = sorted(set(client_suites) - set(CLIENT_SUITES))
+            if unknown:
+                errors.append("catalog.client_suites 含非法套件: " + ", ".join(unknown))
+            missing_suites = sorted(set(CLIENT_SUITES) - set(client_suites))
+            if missing_suites:
+                errors.append("catalog.client_suites 缺少标准套件: " + ", ".join(missing_suites))
+    integration_suites = meta.get("integration_suites")
+    if integration_suites is not None:
+        if not isinstance(integration_suites, dict):
+            errors.append("catalog.integration_suites 必须是 mapping")
+        else:
+            unknown = sorted(set(integration_suites) - set(INTEGRATION_SUITES))
+            if unknown:
+                errors.append("catalog.integration_suites 含非法套件: " + ", ".join(unknown))
+            missing_suites = sorted(set(INTEGRATION_SUITES) - set(integration_suites))
+            if missing_suites:
+                errors.append("catalog.integration_suites 缺少标准套件: " + ", ".join(missing_suites))
     artifacts = meta.get("artifacts")
     if artifacts is not None:
         if not isinstance(artifacts, dict):
@@ -998,6 +1057,7 @@ def validate(root: Path) -> list[str]:
             errors.append(f"{where}: category={cat} 与文件 category={file_cat} 不一致")
         if cat and cat_ids and cat not in cat_ids:
             errors.append(f"{where}: category={cat} 不在 catalog.categories")
+        errors.extend(contract_errors(case, sessions=sessions))
         if case.get("lifecycle", "active") not in LIFECYCLES:
             errors.append(f"{where}: 非法 lifecycle")
         if case.get("tag") not in TAGS:
@@ -1275,6 +1335,15 @@ def build_parser() -> argparse.ArgumentParser:
     p_sel.add_argument("--category")
     p_sel.add_argument("--id", action="append", dest="ids", default=[])
     p_sel.add_argument("--tag", choices=TAGS)
+    p_sel.add_argument(
+        "--suite", "--client-suite", dest="suite",
+        choices=[*CLIENT_SUITES, *INTEGRATION_SUITES],
+        help="客户端或集成测试套件；不传时默认不选 integration",
+    )
+    p_sel.add_argument(
+        "--test-type", choices=TEST_TYPES, default=None,
+        help="按顶层 test_type 筛选；integration 必须显式选择",
+    )
     p_sel.add_argument("--max-risk")
     p_sel.add_argument("--driver", choices=DRIVERS)
     p_sel.add_argument(

skills/tester/scripts/generate_cases_csv.py

@@ -13,6 +13,8 @@ import sys
 from pathlib import Path
 from typing import Any
 
+from journey import suite as case_suite, test_type
+
 try:
     import yaml
 except ImportError:  # pragma: no cover
@@ -22,6 +24,10 @@ except ImportError:  # pragma: no cover
 FIELDS = (
     "category",
     "id",
+    "test_type",
+    "suite",
+    "client_suite",
+    "platform",
     "maps_to",
     "title",
     "what",
@@ -156,10 +162,17 @@ def export_cases_csv(root: Path, out: Path | None = None) -> Path:
         writer.writeheader()
         for case in cases:
             cid = str(case.get("id") or "")
+            client_suite_value = case.get("client_suite") or (
+                case.get("suite") if test_type(case) == "journey" else ""
+            )
             writer.writerow(
                 {
                     "category": case.get("category") or "",
                     "id": cid,
+                    "test_type": test_type(case),
+                    "suite": case_suite(case),
+                    "client_suite": client_suite_value or "",
+                    "platform": one_line(case.get("platform")),
                     "maps_to": case.get("maps_to") or "",
                     "title": case.get("title") or "",
                     "what": one_line(case.get("what")),

skills/tester/scripts/generate_run_csv.py

@@ -18,6 +18,7 @@ except ImportError:  # pragma: no cover
     raise SystemExit(2)
 
 from generate_cases_csv import load_cases, one_line, resolve_root, tester_dir
+from journey import suite as case_suite, test_type
 
 RESULTS = ("pass", "fail", "environment-fail", "skipped", "not-run")
 RUN_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
@@ -25,6 +26,10 @@ EXECUTED_YES = frozenset({"pass", "fail", "environment-fail"})
 EXECUTED_NO = frozenset({"skipped", "not-run"})
 FIELDS = (
     "id",
+    "test_type",
+    "suite",
+    "client_suite",
+    "platform",
     "result",
     "executed",
     "evidence",
@@ -129,9 +134,16 @@ def export_run_csv(
             cid = str(item["id"])
             result = str(item.get("result") or "")
             case = cases.get(cid) or {}
+            client_suite_value = case.get("client_suite") or (
+                case.get("suite") if test_type(case) == "journey" else ""
+            )
             writer.writerow(
                 {
                     "id": cid,
+                    "test_type": test_type(case),
+                    "suite": case_suite(case),
+                    "client_suite": client_suite_value or "",
+                    "platform": one_line(case.get("platform")),
                     "result": result,
                     "executed": executed_cell(result),
                     "evidence": one_line(item.get("evidence")),

skills/tester/scripts/journey.py

@@ -0,0 +1,283 @@
+#!/usr/bin/env python3
+"""Shared client-journey contract and result lifecycle helpers.
+
+The mobile/browser driver still owns the real UI interaction.  This module
+owns the parts that must be identical for every client catalog: contract
+validation, preflight blocking, and result propagation.  It deliberately
+returns the existing tester result vocabulary instead of introducing a second
+report format.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Iterable, Mapping
+
+
+TEST_TYPES = ("journey", "integration")
+CLIENT_SUITES = ("client-smoke", "client-regression", "safety")
+INTEGRATION_SUITES = ("api-contract", "authorization", "build-contract")
+PLATFORMS = ("android", "ios")
+FAILURE_POLICIES = ("stop", "continue")
+RESULTS = ("pass", "fail", "environment-fail", "skipped", "not-run")
+
+
+def client_suite(case: Mapping[str, Any]) -> str:
+    """Return the canonical client suite, accepting ``suite`` as a migration alias."""
+    value = case.get("client_suite")
+    alias = case.get("suite")
+    if value and alias and str(value) != str(alias):
+        return ""
+    return str(value or alias or "")
+
+
+def integration_suite(case: Mapping[str, Any]) -> str:
+    """Return the integration suite, accepting ``suite`` during migration."""
+    value = case.get("integration_suite")
+    alias = case.get("suite")
+    if value and alias and str(value) != str(alias):
+        return ""
+    return str(value or alias or "")
+
+
+def test_type(case: Mapping[str, Any]) -> str:
+    """Return an explicit type or infer one for an old catalog entry."""
+    declared = str(case.get("test_type") or "").strip().lower()
+    if declared:
+        return declared
+    if client_suite(case) or case.get("journey") is not None:
+        return "journey"
+    if integration_suite(case):
+        return "integration"
+    return ""
+
+
+def suite(case: Mapping[str, Any]) -> str:
+    """Return the report/selection suite for either test type."""
+    kind = test_type(case)
+    return client_suite(case) if kind == "journey" else integration_suite(case)
+
+
+def normalize_platforms(value: Any) -> list[str]:
+    if isinstance(value, str):
+        values = [value]
+    elif isinstance(value, (list, tuple)):
+        values = list(value)
+    else:
+        return []
+    return [str(item).strip().lower() for item in values if str(item).strip()]
+
+
+def _non_empty_list(value: Any) -> bool:
+    return isinstance(value, list) and bool(value) and all(str(item).strip() for item in value)
+
+
+def _has_account(fixture: Mapping[str, Any]) -> bool:
+    account = fixture.get("account")
+    if isinstance(account, Mapping):
+        return bool(str(account.get("role") or account.get("username") or "").strip())
+    return bool(str(account or "").strip())
+
+
+def contract_errors(case: Mapping[str, Any], sessions: Mapping[str, Any] | None = None) -> list[str]:
+    """Validate a client journey without requiring app-specific field names.
+
+    A case becomes a client journey when it declares ``client_suite`` (or the
+    temporary ``suite`` alias).  Legacy non-client cases keep the existing
+    schema.  Once opted in, all lifecycle inputs are explicit so a missing
+    fixture/cleanup can stop the run before any case result is fabricated.
+    """
+    kind = test_type(case)
+    declared_kind = str(case.get("test_type") or "").strip().lower()
+    if declared_kind and declared_kind not in TEST_TYPES:
+        return [f"{case.get('id') or '?'}: test_type 必须是 journey 或 integration"]
+    if not kind:
+        return []
+    if kind == "integration":
+        return integration_contract_errors(case)
+
+    suite_name = client_suite(case)
+    journey = case.get("journey")
+    errors: list[str] = []
+    cid = str(case.get("id") or "?")
+    where = f"{cid}: client journey"
+    if case.get("client_suite") and case.get("suite") and str(case["client_suite"]) != str(case["suite"]):
+        errors.append(f"{where}: client_suite 与 suite 不一致")
+    if suite_name not in CLIENT_SUITES:
+        errors.append(f"{where}: client_suite 必须是 {', '.join(CLIENT_SUITES)}")
+
+    if not isinstance(journey, Mapping):
+        errors.append(f"{where}: journey 必须是 mapping")
+        journey = {}
+    for key in ("goal", "entry_state", "final_state"):
+        if not str(journey.get(key) or "").strip():
+            errors.append(f"{where}: journey.{key} 必填")
+    if not _non_empty_list(journey.get("actions")):
+        errors.append(f"{where}: journey.actions 必须是非空列表")
+    if not _non_empty_list(journey.get("assertions")):
+        errors.append(f"{where}: journey.assertions 必须是非空列表")
+    policy = str(journey.get("failure_policy") or "")
+    if policy not in FAILURE_POLICIES:
+        errors.append(f"{where}: journey.failure_policy 必须是 stop 或 continue")
+
+    fixture = case.get("fixture")
+    if not isinstance(fixture, Mapping):
+        errors.append(f"{where}: fixture 缺失，必须在 preflight 前声明")
+    elif not _has_account(fixture):
+        errors.append(f"{where}: fixture.account 缺失，必须声明测试账号或角色")
+
+    cleanup = case.get("cleanup")
+    if not isinstance(cleanup, Mapping):
+        errors.append(f"{where}: cleanup 缺失，必须在 preflight 前声明")
+    else:
+        steps = cleanup.get("steps")
+        command = cleanup.get("run")
+        explicitly_noop = cleanup.get("required") is False and str(cleanup.get("reason") or "").strip()
+        if not explicitly_noop and not _non_empty_list(steps) and not str(command or "").strip():
+            errors.append(f"{where}: cleanup 必须有非空 steps/run；只读用例也要显式声明 reason")
+
+    platforms = normalize_platforms(case.get("platform"))
+    if not platforms:
+        errors.append(f"{where}: platform 必须声明 android/ios")
+    elif any(platform not in PLATFORMS for platform in platforms):
+        errors.append(f"{where}: platform 只能是 {', '.join(PLATFORMS)}")
+    elif len(platforms) != len(set(platforms)):
+        errors.append(f"{where}: platform 不能重复")
+
+    driver = case.get("driver")
+    if driver not in ("mobile", "browser"):
+        errors.append(f"{where}: 客户端旅程 driver 必须是 mobile 或 browser")
+    if driver and driver in ("mobile", "browser"):
+        session = case.get("session")
+        if not str(session or "").strip():
+            errors.append(f"{where}: session 必须声明可复用客户端会话")
+        elif sessions is not None and session not in sessions:
+            errors.append(f"{where}: session={session} 未在 catalog.sessions 声明")
+    if suite_name == "safety" and str(case.get("tag") or "") == "smoke":
+        errors.append(f"{where}: safety 不得标记为 smoke")
+    return errors
+
+
+def integration_contract_errors(case: Mapping[str, Any]) -> list[str]:
+    """Validate the non-client half of the tester catalog.
+
+    Integration cases deliberately do not require a browser/mobile session.  A
+    ``how`` block is the legacy call declaration; newer catalogs may use
+    ``request``/``input``/``call`` or an ``integration`` mapping as well.
+    ``expected`` remains the common protocol/data assertion list.
+    """
+    cid = str(case.get("id") or "?")
+    where = f"{cid}: integration"
+    errors: list[str] = []
+    value = case.get("integration_suite")
+    alias = case.get("suite")
+    if value and alias and str(value) != str(alias):
+        errors.append(f"{where}: integration_suite 与 suite 不一致")
+    suite_name = integration_suite(case)
+    if suite_name not in INTEGRATION_SUITES:
+        errors.append(
+            f"{where}: integration_suite 必须是 {', '.join(INTEGRATION_SUITES)}"
+        )
+
+    fixture = case.get("fixture")
+    if not isinstance(fixture, Mapping) or not fixture:
+        errors.append(f"{where}: fixture 缺失，必须在 preflight 前声明")
+
+    declared_call = any(
+        key in case and case.get(key) not in (None, "", [], {})
+        for key in ("request", "input", "call", "integration")
+    )
+    how = case.get("how")
+    if not declared_call and not isinstance(how, Mapping):
+        errors.append(f"{where}: 必须声明 request/input/call/integration 或 how")
+
+    # Integration writes need an explicit rollback/cleanup contract.  Read-only
+    # probes may state why cleanup is unnecessary, matching test-preflight.
+    risk = str(case.get("risk") or "")
+    cleanup = case.get("cleanup")
+    cleanup_steps = isinstance(cleanup, Mapping) and (
+        bool(cleanup.get("steps")) or bool(str(cleanup.get("run") or "").strip())
+    )
+    cleanup_noop = isinstance(cleanup, Mapping) and cleanup.get("required") is False and bool(
+        str(cleanup.get("reason") or "").strip()
+    )
+    if risk != "readonly" and not (cleanup_steps or cleanup_noop):
+        errors.append(f"{where}: 写入型 integration 必须声明 cleanup/rollback steps/run")
+    elif cleanup is not None and not (cleanup_steps or cleanup_noop):
+        errors.append(f"{where}: cleanup 必须有 steps/run 或 required=false + reason")
+
+    forbidden = []
+    if case.get("client_suite") or case.get("journey") is not None:
+        forbidden.append("client_suite/journey")
+    if case.get("session") or case.get("platform"):
+        forbidden.append("session/platform")
+    if forbidden:
+        errors.append(f"{where}: integration 不应声明 " + ", ".join(forbidden))
+    return errors
+
+
+def preflight(cases: Iterable[Mapping[str, Any]], sessions: Mapping[str, Any] | None = None) -> dict[str, Any]:
+    """Return a contract preflight result; no per-case result is created on block."""
+    cases = list(cases)
+    errors: list[str] = []
+    for case in cases:
+        errors.extend(contract_errors(case, sessions=sessions))
+    return {
+        "required": bool(errors or any(client_suite(case) for case in cases)),
+        "status": "blocked" if errors else "ready",
+        "failure_class": "contract" if errors else None,
+        "errors": errors,
+    }
+
+
+def classify_outcome(
+    *,
+    ok: bool,
+    environment_ready: bool = True,
+    phase: str = "journey",
+) -> str:
+    """Classify one observed phase using the tester's five result states.
+
+    A valid test account is part of the client journey: a failed login is a
+    product ``fail``.  Only an unavailable environment is ``environment-fail``.
+    """
+    if not environment_ready:
+        return "environment-fail"
+    if not ok:
+        return "fail"
+    return "pass"
+
+
+def propagate_results(
+    case_ids: Iterable[str],
+    observed: Mapping[str, str],
+    *,
+    skipped: Mapping[str, str] | None = None,
+    continue_on_fail: bool = False,
+    policies: Mapping[str, str] | None = None,
+) -> list[dict[str, str]]:
+    """Build ordered result rows and mark cases after a blocking failure not-run.
+
+    ``skipped`` is intentional selection (risk/lifecycle/assumption); it is
+    never converted to ``not-run``.  A product or environment failure stops
+    subsequent cases unless the caller explicitly opts into continuation.
+    """
+    skipped = skipped or {}
+    policies = policies or {}
+    rows: list[dict[str, str]] = []
+    blocked = False
+    for raw_id in case_ids:
+        case_id = str(raw_id)
+        if case_id in skipped:
+            rows.append({"id": case_id, "result": "skipped", "notes": skipped[case_id]})
+            continue
+        if blocked:
+            rows.append({"id": case_id, "result": "not-run", "notes": "前置用例失败，未轮到执行"})
+            continue
+        result = observed.get(case_id, "not-run")
+        if result not in RESULTS:
+            raise ValueError(f"未知结果: {case_id}={result}")
+        rows.append({"id": case_id, "result": result})
+        policy = policies.get(case_id)
+        if result in ("fail", "environment-fail") and not (continue_on_fail or policy == "continue"):
+            blocked = True
+    return rows

skills/tester/scripts/mobile_bundle.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 import argparse
 import hashlib
+import importlib.util
 import json
 import os
 import plistlib
@@ -33,6 +34,7 @@ from typing import Any
 
 COMMIT = re.compile(r"^[0-9a-fA-F]{40}$")
 PLATFORMS = {"android", "ios"}
+PREFLIGHT_ROOT = Path(__file__).resolve().parents[2] / "test-preflight" / "scripts" / "preflight.py"
 
 
 class EnvironmentFailure(RuntimeError):
@@ -388,6 +390,33 @@ def write_run(root: Path, record: dict[str, Any]) -> Path:
     return path
 
 
+def run_test_preflight(
+    config_path: Path,
+    *,
+    mode: str = "verify",
+    yaml_path: Path | None = None,
+    csv_path: Path | None = None,
+) -> dict[str, Any]:
+    """Run the shared preflight contract before any client session work."""
+    if mode not in {"verify", "prepare"}:
+        raise ArtifactFailure("preflight mode 只能是 verify 或 prepare")
+    if not PREFLIGHT_ROOT.is_file():
+        raise EnvironmentFailure(f"找不到 test-preflight skill: {PREFLIGHT_ROOT}")
+    spec = importlib.util.spec_from_file_location("pouch_test_preflight", PREFLIGHT_ROOT)
+    if spec is None or spec.loader is None:
+        raise EnvironmentFailure("无法加载 test-preflight skill")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    try:
+        config = module.load_config(config_path)
+        evidence = module.run(config, mode)
+        yaml_path = yaml_path or config_path.with_name("preflight.yaml")
+        csv_path = csv_path or config_path.with_name("preflight.csv")
+        return module.write_evidence(evidence, yaml_path, csv_path)
+    except (OSError, ValueError, module.yaml.YAMLError) as exc:
+        return {"status": "blocked", "error": str(exc), "yaml": str(yaml_path or ""), "csv": str(csv_path or "")}
+
+
 def execute(args: argparse.Namespace) -> int:
     requested = platforms(args.platform)
     context = RuntimeContext(Path(args.root).resolve())
@@ -417,7 +446,7 @@ def execute(args: argparse.Namespace) -> int:
         artifact_dir = Path(tempfile.mkdtemp(prefix="pouch-tester-artifacts-"))
         record["artifact_dir"] = str(artifact_dir)
         expected = catalog_identities(context.root)
-        prepared: dict[str, tuple[dict[str, Any], dict[str, str], str]] = {}
+        prepared: dict[str, tuple[dict[str, Any], dict[str, str] | None, str]] = {}
         failures: dict[str, str] = {}
         for platform in requested:
             result: dict[str, Any] = {"node": "tester.android" if platform == "android" else "tester.ios", "status": "preparing"}
@@ -429,22 +458,65 @@ def execute(args: argparse.Namespace) -> int:
                 wanted = expected.get(platform)
                 if wanted and identity != wanted:
                     raise EnvironmentFailure(f"{platform} 身份不匹配: expected={wanted} actual={identity}")
+                result.update({"identity": identity, "status": "artifact-ready"})
+                prepared[platform] = (result, None, identity)
+            except (EnvironmentFailure, ArtifactFailure) as exc:
+                failures[platform] = str(exc)
+                result.update({"status": "environment-fail", "failure_class": "automatic-repair-failed", "error": str(exc)})
+
+        # This is the hard gate shared by integration and journey callers.
+        # There is no safe implicit matrix: without a caller-supplied config we
+        # cannot prove HOST, fixtures, tester bindings, or node reachability.
+        preflight_config = Path(args.preflight_config).expanduser() if args.preflight_config else None
+        if not failures:
+            if preflight_config is None:
+                preflight_result = {
+                    "status": "blocked",
+                    "error": "缺少 --preflight-config；不能跳过 test-preflight",
+                }
+            else:
+                preflight_result = run_test_preflight(
+                    preflight_config,
+                    mode=args.preflight_mode,
+                    yaml_path=Path(args.preflight_yaml).expanduser() if args.preflight_yaml else None,
+                    csv_path=Path(args.preflight_csv).expanduser() if args.preflight_csv else None,
+                )
+            record["preflight"] = preflight_result
+            if preflight_result.get("status") != "ready":
+                failures["preflight"] = str(preflight_result.get("error") or preflight_result.get("status"))
+
+        if not failures and args.preflight_mode == "prepare":
+            # test-preflight prepare owns platform probing, installation and
+            # launch; do not install/start the same artifact a second time.
+            for result, _, _ in prepared.values():
+                result.update({"status": "preflight-prepared"})
+        elif not failures:
+            for platform in requested:
+                result, _, identity = prepared[platform]
+                try:
                     node = android_preflight(context) if platform == "android" else ios_preflight(context)
                     check_network(context, node)
-                result.update({"identity": identity, "node_preflight": node, "status": "preflight-passed"})
+                    result.update({"node_preflight": node, "status": "preflight-passed"})
                     prepared[platform] = (result, node, identity)
-            except (EnvironmentFailure, ArtifactFailure) as exc:
+                except EnvironmentFailure as exc:
                     failures[platform] = str(exc)
                     result.update({"status": "environment-fail", "failure_class": "automatic-repair-failed", "error": str(exc)})
         if failures and not args.partial:
             for platform in requested:
                 if platform not in failures:
                     record["platforms"][platform].update({"status": "not-run", "reason": "全量 preflight 闸门被另一平台阻断"})
-            record.update({"status": "environment-fail", "failure_class": "automatic-repair-failed", "next_action": "修复列出的平台环境后重跑；默认不会静默降级"})
+            overall = record.get("preflight", {}).get("status") if "preflight" in record else "environment-fail"
+            record.update({"status": overall if overall in {"blocked", "environment-fail"} else "environment-fail", "failure_class": "test-preflight" if "preflight" in record else "automatic-repair-failed", "next_action": "修复列出的平台环境后重跑；默认不会静默降级"})
         else:
             for platform, (result, node, identity) in prepared.items():
                 try:
-                    installed = install_android(node, Path(result["artifact"]["path"]), identity, context) if platform == "android" else install_ios(node, Path(result["artifact"]["path"]), identity, context)
+                    installed = (
+                        {"delegated": "test-preflight.prepare"}
+                        if args.preflight_mode == "prepare"
+                        else install_android(node, Path(result["artifact"]["path"]), identity, context)
+                        if platform == "android"
+                        else install_ios(node, Path(result["artifact"]["path"]), identity, context)
+                    )
                     result.update({"status": "ready", "install": installed, "failure_class": None})
                 except EnvironmentFailure as exc:
                     failures[platform] = str(exc)
@@ -476,6 +548,10 @@ def parser() -> argparse.ArgumentParser:
     run.add_argument("--root", default=".")
     run.add_argument("--worktree-root", default="")
     run.add_argument("--binding", default="")
+    run.add_argument("--preflight-config", default="", help="test-preflight matrix YAML（必填，不得跳过）")
+    run.add_argument("--preflight-mode", choices=("verify", "prepare"), default="verify")
+    run.add_argument("--preflight-yaml", default="", help="preflight YAML 证据输出路径")
+    run.add_argument("--preflight-csv", default="", help="preflight CSV 证据输出路径")
     run.set_defaults(func=execute)
     return root
 

skills/tester/scripts/summarize_run.py

@@ -0,0 +1,175 @@
+#!/usr/bin/env python3
+"""Validate a tester CSV/YAML pair and summarize statuses from CSV only."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import sys
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+try:
+    import yaml
+except ImportError:  # pragma: no cover
+    print("需要 PyYAML: python3 -m pip install pyyaml", file=sys.stderr)
+    raise SystemExit(2)
+
+
+RESULTS = ("pass", "fail", "environment-fail", "skipped", "not-run")
+RESULT_SET = frozenset(RESULTS)
+
+
+class SummaryError(ValueError):
+    """Raised when the CSV/YAML pair cannot be safely summarized."""
+
+
+def _format_ids(ids: set[str]) -> str:
+    return ", ".join(sorted(ids)) or "(none)"
+
+
+def _duplicate_ids(ids: list[str]) -> list[str]:
+    counts = Counter(ids)
+    return sorted(case_id for case_id, count in counts.items() if count > 1)
+
+
+def load_csv(path: Path) -> list[dict[str, str]]:
+    """Load CSV rows and reject malformed IDs/results with diagnostics."""
+    try:
+        handle = path.open(encoding="utf-8-sig", newline="")
+    except OSError as exc:
+        raise SummaryError(f"无法读取 CSV {path}: {exc}") from exc
+
+    with handle:
+        reader = csv.DictReader(handle)
+        if not reader.fieldnames:
+            raise SummaryError(f"CSV {path} 缺少表头")
+        missing_fields = [field for field in ("id", "result") if field not in reader.fieldnames]
+        if missing_fields:
+            raise SummaryError(f"CSV {path} 缺少必需列: {', '.join(missing_fields)}")
+        rows = list(reader)
+
+    ids: list[str] = []
+    errors: list[str] = []
+    for line_number, row in enumerate(rows, start=2):
+        case_id = (row.get("id") or "").strip()
+        result = (row.get("result") or "").strip()
+        ids.append(case_id)
+        if not case_id:
+            errors.append(f"CSV 第 {line_number} 行 testcase id 缺失")
+        if not result:
+            errors.append(f"CSV 第 {line_number} 行 result 缺失 (testcase id={case_id or '(missing)'})")
+        elif result not in RESULT_SET:
+            errors.append(
+                f"CSV 第 {line_number} 行 result 未知: {result!r} "
+                f"(testcase id={case_id or '(missing)'}; 允许值={', '.join(RESULTS)})"
+            )
+
+    duplicates = _duplicate_ids([case_id for case_id in ids if case_id])
+    if duplicates:
+        errors.append(f"CSV testcase id 重复: {_format_ids(set(duplicates))}")
+    if errors:
+        raise SummaryError("CSV 校验失败:\n- " + "\n- ".join(errors))
+    return rows
+
+
+def load_yaml_results(path: Path) -> list[dict[str, Any]]:
+    """Load only YAML results structure needed for count/ID validation."""
+    try:
+        data = yaml.safe_load(path.read_text(encoding="utf-8"))
+    except OSError as exc:
+        raise SummaryError(f"无法读取 YAML {path}: {exc}") from exc
+    except yaml.YAMLError as exc:
+        raise SummaryError(f"YAML 解析失败 {path}: {exc}") from exc
+
+    if not isinstance(data, dict) or not isinstance(data.get("results"), list):
+        raise SummaryError(f"YAML {path} 缺少 results 列表")
+
+    results = data["results"]
+    ids: list[str] = []
+    errors: list[str] = []
+    for index, item in enumerate(results, start=1):
+        if not isinstance(item, dict):
+            errors.append(f"YAML results[{index}] 不是 mapping")
+            continue
+        case_id = str(item.get("id") or "").strip()
+        ids.append(case_id)
+        if not case_id:
+            errors.append(f"YAML results[{index}] testcase id 缺失")
+
+    duplicates = _duplicate_ids([case_id for case_id in ids if case_id])
+    if duplicates:
+        errors.append(f"YAML testcase id 重复: {_format_ids(set(duplicates))}")
+    if errors:
+        raise SummaryError("YAML 校验失败:\n- " + "\n- ".join(errors))
+    return results
+
+
+def summarize(csv_path: Path, yaml_path: Path) -> tuple[Counter[str], dict[str, int]]:
+    """Validate the pair and return counts calculated from CSV result values."""
+    rows = load_csv(csv_path)
+    yaml_results = load_yaml_results(yaml_path)
+    csv_ids = [(row.get("id") or "").strip() for row in rows]
+    yaml_ids = [str(item.get("id") or "").strip() for item in yaml_results]
+    csv_id_set = set(csv_ids)
+    yaml_id_set = set(yaml_ids)
+    diagnostics: list[str] = []
+
+    if len(rows) != len(csv_id_set):
+        diagnostics.append(
+            f"CSV 数据行总数与唯一 testcase id 数不一致: rows={len(rows)} unique_ids={len(csv_id_set)}"
+        )
+    if len(rows) != len(yaml_results):
+        diagnostics.append(
+            f"CSV 数据行总数与 YAML results 数量不一致: csv_rows={len(rows)} yaml_results={len(yaml_results)}"
+        )
+    if csv_id_set != yaml_id_set:
+        diagnostics.append(
+            "CSV/YAML testcase id 集合不一致: "
+            f"仅 CSV 有 [{_format_ids(csv_id_set - yaml_id_set)}]; "
+            f"仅 YAML 有 [{_format_ids(yaml_id_set - csv_id_set)}]"
+        )
+    if diagnostics:
+        raise SummaryError("CSV/YAML 汇总校验失败:\n- " + "\n- ".join(diagnostics))
+
+    counts: Counter[str] = Counter((row.get("result") or "").strip() for row in rows)
+    total = sum(counts[name] for name in RESULTS)
+    if total != len(rows):
+        raise SummaryError(
+            "CSV 状态计数和不等于数据行总数: "
+            f"count_sum={total} csv_rows={len(rows)}"
+        )
+    metadata = {
+        "csv_rows": len(rows),
+        "csv_unique_ids": len(csv_id_set),
+        "yaml_results": len(yaml_results),
+        "count_sum": total,
+    }
+    return counts, metadata
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="校验 tester CSV/YAML 并从 CSV result 汇总状态")
+    parser.add_argument("--csv", required=True, type=Path, help="运行成绩单 CSV")
+    parser.add_argument("--yaml", required=True, type=Path, help="对应运行快照 YAML")
+    args = parser.parse_args(argv)
+    try:
+        counts, metadata = summarize(args.csv, args.yaml)
+    except SummaryError as exc:
+        print(f"汇总失败: {exc}", file=sys.stderr)
+        return 1
+
+    bits = " ".join(f"{name}={counts[name]}" for name in RESULTS)
+    print(f"summary source=CSV.result {bits} total={metadata['count_sum']}")
+    print(
+        "validated "
+        f"csv_rows={metadata['csv_rows']} "
+        f"csv_unique_ids={metadata['csv_unique_ids']} "
+        f"yaml_results={metadata['yaml_results']}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

skills/tester/scripts/two_tier.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""Shared execution policy for tester's integration-first runs.
+
+This module intentionally has no browser, mobile, deployment, or API driver.
+Those adapters belong to the caller.  It only controls the phase gate and the
+stable five-state testcase result propagation so a caller cannot create a
+journey session before preflight and integration have passed.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable, Mapping
+from typing import Any
+
+RESULTS = ("pass", "fail", "environment-fail", "skipped", "not-run")
+PHASES = ("preflight", "integration", "journey")
+
+
+def classify_integration(*, ok: bool, service_reachable: bool = True) -> str:
+    """Separate a contract assertion failure from an unavailable service."""
+    if not service_reachable:
+        return "environment-fail"
+    return "pass" if ok else "fail"
+
+
+def _case_id(case: Mapping[str, Any] | str) -> str:
+    if isinstance(case, Mapping):
+        return str(case.get("id") or "")
+    return str(case)
+
+
+def _not_run(cases: Iterable[Mapping[str, Any] | str], reason: str) -> list[dict[str, str]]:
+    rows: list[dict[str, str]] = []
+    for case in cases:
+        row = {"id": _case_id(case), "result": "not-run", "notes": reason}
+        if isinstance(case, Mapping):
+            row["test_type"] = str(case.get("test_type") or "")
+            row["suite"] = str(
+                case.get("client_suite")
+                or case.get("integration_suite")
+                or case.get("suite")
+                or ""
+            )
+        rows.append(row)
+    return rows
+
+
+def run_integration_first(
+    *,
+    preflight_status: str,
+    integration_cases: Iterable[Mapping[str, Any] | str],
+    journey_cases: Iterable[Mapping[str, Any] | str],
+    execute_integration: Callable[[Mapping[str, Any] | str], str | Mapping[str, Any]],
+    create_journey_session: Callable[[], Any],
+    execute_journey: Callable[[Any, Mapping[str, Any] | str], str | Mapping[str, Any]],
+) -> dict[str, Any]:
+    """Execute integration cases before creating any journey session.
+
+    ``preflight_status`` must be the status emitted by test-preflight.  A
+    blocked/environment-fail gate records only ``not-run`` testcase rows and
+    never invokes either adapter.  An integration failure stops the phase and
+    propagates ``not-run`` to all journey cases.
+    """
+    integration = list(integration_cases)
+    journey = list(journey_cases)
+    if preflight_status != "ready":
+        reason = f"test-preflight={preflight_status}，未创建客户端 session"
+        return {
+            "status": preflight_status,
+            "phase": "preflight",
+            "session_created": False,
+            "results": _not_run(integration, reason) + _not_run(journey, reason),
+        }
+
+    results: list[dict[str, Any]] = []
+    integration_failed = False
+    for index, case in enumerate(integration):
+        if integration_failed:
+            results.extend(_not_run([case], "前置 integration 失败，未轮到执行"))
+            continue
+        outcome = execute_integration(case)
+        row = dict(outcome) if isinstance(outcome, Mapping) else {"result": outcome}
+        row["id"] = _case_id(case)
+        if isinstance(case, Mapping):
+            row.setdefault("test_type", str(case.get("test_type") or "integration"))
+            row.setdefault("suite", str(case.get("integration_suite") or case.get("suite") or ""))
+        result = str(row.get("result") or "not-run")
+        if result not in RESULTS:
+            raise ValueError(f"未知 integration 结果: {row['id']}={result}")
+        row["result"] = result
+        results.append(row)
+        if result in ("fail", "environment-fail"):
+            integration_failed = True
+
+    if integration_failed:
+        results.extend(_not_run(journey, "integration 失败，依赖的 journey 未执行"))
+        return {
+            "status": "fail" if any(row["result"] == "fail" for row in results) else "environment-fail",
+            "phase": "integration",
+            "session_created": False,
+            "results": results,
+        }
+
+    session = create_journey_session() if journey else None
+    for case in journey:
+        outcome = execute_journey(session, case)
+        row = dict(outcome) if isinstance(outcome, Mapping) else {"result": outcome}
+        row["id"] = _case_id(case)
+        if isinstance(case, Mapping):
+            row.setdefault("test_type", str(case.get("test_type") or "journey"))
+            row.setdefault("suite", str(case.get("client_suite") or case.get("suite") or ""))
+        result = str(row.get("result") or "not-run")
+        if result not in RESULTS:
+            raise ValueError(f"未知 journey 结果: {row['id']}={result}")
+        row["result"] = result
+        results.append(row)
+    status = "fail" if any(row["result"] == "fail" for row in results) else (
+        "environment-fail" if any(row["result"] == "environment-fail" for row in results) else "pass"
+    )
+    return {
+        "status": status,
+        "phase": "journey" if journey else "integration",
+        "session_created": bool(journey),
+        "results": results,
+    }

skills/tester/templates/catalog.yaml

@@ -22,6 +22,20 @@ risks:
   - mutating
   - destructive
 default_max_risk: readonly
+client_suites:
+  client-smoke:
+    default: true
+  client-regression:
+    default: false
+  safety:
+    default: false
+integration_suites:
+  api-contract:
+    default: false
+  authorization:
+    default: false
+  build-contract:
+    default: false
 vars: {}
 authorize:
   mutating: AUTHORIZE_MUTATING

skills/tester/tests/test_catalog_journey.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+import json
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+import yaml
+
+
+ROOT = Path(__file__).resolve().parents[1]
+CATALOG = ROOT / "scripts" / "catalog.py"
+
+
+def write_fixture(root: Path, *, complete: bool = True) -> None:
+    tester = root / ".pouch" / "tester"
+    (tester / "cases").mkdir(parents=True)
+    (tester / "runs").mkdir()
+    (tester / "catalog.yaml").write_text(
+        yaml.safe_dump(
+            {
+                "version": 2,
+                "id_pattern": "^TC-[A-Z]+-[0-9]+$",
+                "categories": [{"id": "CLIENT"}],
+                "risks": ["readonly", "mutating", "destructive"],
+                "default_max_risk": "readonly",
+                "client_suites": {
+                    "client-smoke": {"default": True},
+                    "client-regression": {"default": False},
+                    "safety": {"default": False},
+                },
+                "sessions": {"browser": {"kind": "browser", "reuse": "run"}},
+            },
+            sort_keys=False,
+        ),
+        encoding="utf-8",
+    )
+    case = {
+        "id": "TC-CLIENT-001",
+        "category": "CLIENT",
+        "title": "journey",
+        "what": "client journey",
+        "lifecycle": "active",
+        "tag": "smoke",
+        "risk": "readonly",
+        "driver": "browser",
+        "session": "browser",
+        "client_suite": "client-smoke",
+        "platform": ["android", "ios"],
+        "journey": {
+            "goal": "goal",
+            "entry_state": "logged_out",
+            "actions": ["start", "login"],
+            "assertions": ["visible"],
+            "final_state": "done",
+            "failure_policy": "stop",
+        },
+        "fixture": {"account": "test-role"},
+        "cleanup": {"steps": ["cleanup"]},
+        "how": {"steps": ["start", "login"]},
+        "expected": [{"kind": "visible-text", "value": "done"}],
+    }
+    if not complete:
+        case.pop("fixture")
+        case.pop("cleanup")
+    (tester / "cases" / "client.yaml").write_text(
+        yaml.safe_dump({"category": "CLIENT", "cases": [case]}, sort_keys=False),
+        encoding="utf-8",
+    )
+
+
+class CatalogJourneyTests(unittest.TestCase):
+    def run_catalog(self, root: Path, *args: str):
+        return subprocess.run(
+            [sys.executable, str(CATALOG), *args, "--root", str(root)],
+            text=True,
+            capture_output=True,
+        )
+
+    def test_valid_client_catalog_and_suite_filter(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            write_fixture(root)
+            valid = self.run_catalog(root, "validate")
+            self.assertEqual(valid.returncode, 0, valid.stderr)
+            selected = self.run_catalog(root, "select", "--suite", "client-smoke")
+            self.assertEqual(selected.returncode, 0, selected.stderr)
+            self.assertEqual(json.loads(selected.stdout)["count"], 1)
+
+    def test_missing_fixture_or_cleanup_blocks_before_results(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            write_fixture(root, complete=False)
+            selected = self.run_catalog(root, "select", "--suite", "client-smoke")
+            self.assertNotEqual(selected.returncode, 0)
+            payload = json.loads(selected.stdout)
+            self.assertEqual(payload["preflight"]["failure_class"], "contract")
+            self.assertTrue(any("fixture 缺失" in error for error in payload["preflight"]["errors"]))
+            self.assertEqual(payload["preflight"]["status"], "blocked")
+
+
+if __name__ == "__main__":
+    unittest.main()

skills/tester/tests/test_journey.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
+from journey import classify_outcome, contract_errors, propagate_results
+
+
+def valid_case(**overrides):
+    case = {
+        "id": "TC-CLIENT-001",
+        "client_suite": "client-smoke",
+        "driver": "browser",
+        "session": "browser",
+        "platform": ["android", "ios"],
+        "journey": {
+            "goal": "完成客户端目标",
+            "entry_state": "logged_out",
+            "actions": ["启动", "登录", "完成目标"],
+            "assertions": ["看到目标页面"],
+            "final_state": "目标已完成",
+            "failure_policy": "stop",
+        },
+        "fixture": {"account": "test-role", "data": []},
+        "cleanup": {"steps": ["清理本轮数据"]},
+    }
+    case.update(overrides)
+    return case
+
+
+class JourneyContractTests(unittest.TestCase):
+    def test_contract_accepts_shared_client_shape(self):
+        self.assertEqual(contract_errors(valid_case(), {"browser": {}}), [])
+
+    def test_missing_fixture_and_cleanup_are_contract_errors(self):
+        errors = contract_errors(
+            valid_case(fixture=None, cleanup=None),
+            {"browser": {}},
+        )
+        self.assertTrue(any("fixture 缺失" in error for error in errors))
+        self.assertTrue(any("cleanup 缺失" in error for error in errors))
+
+    def test_login_failure_is_product_failure_but_unavailable_environment_is_not(self):
+        self.assertEqual(classify_outcome(ok=False, phase="login"), "fail")
+        self.assertEqual(
+            classify_outcome(ok=False, phase="login", environment_ready=False),
+            "environment-fail",
+        )
+
+    def test_failed_case_blocks_remaining_cases_and_keeps_explicit_skip(self):
+        rows = propagate_results(
+            ["TC-1", "TC-2", "TC-3"],
+            {"TC-1": "fail", "TC-2": "pass", "TC-3": "pass"},
+            skipped={"TC-3": "未授权"},
+        )
+        self.assertEqual([row["result"] for row in rows], ["fail", "not-run", "skipped"])
+
+    def test_continue_policy_allows_next_case(self):
+        rows = propagate_results(
+            ["TC-1", "TC-2"],
+            {"TC-1": "fail", "TC-2": "pass"},
+            policies={"TC-1": "continue"},
+        )
+        self.assertEqual([row["result"] for row in rows], ["fail", "pass"])
+
+
+if __name__ == "__main__":
+    unittest.main()

skills/tester/tests/test_run_summary.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+import csv
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
+import summarize_run
+
+
+class RunSummaryTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.csv_path = self.root / "run.csv"
+        self.yaml_path = self.root / "run.yaml"
+        self.rows = [
+            {"id": "TC-001", "result": "pass"},
+            {"id": "TC-002", "result": "fail"},
+            {"id": "TC-003", "result": "environment-fail"},
+            {"id": "TC-004", "result": "skipped"},
+            {"id": "TC-005", "result": "not-run"},
+        ]
+        self.write_csv(self.rows)
+        self.write_yaml([{"id": row["id"], "result": row["result"]} for row in self.rows])
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def write_csv(self, rows):
+        with self.csv_path.open("w", encoding="utf-8", newline="") as handle:
+            writer = csv.DictWriter(handle, fieldnames=["id", "result"])
+            writer.writeheader()
+            writer.writerows(rows)
+
+    def write_yaml(self, results):
+        import yaml
+
+        self.yaml_path.write_text(
+            yaml.safe_dump(
+                {"counts": {"pass": 999}, "results": results},
+                sort_keys=False,
+            ),
+            encoding="utf-8",
+        )
+
+    def run_cli(self):
+        return subprocess.run(
+            [
+                sys.executable,
+                str(Path(summarize_run.__file__).resolve()),
+                "--csv",
+                str(self.csv_path),
+                "--yaml",
+                str(self.yaml_path),
+            ],
+            text=True,
+            capture_output=True,
+        )
+
+    def test_valid_summary_uses_csv_results(self):
+        result = self.run_cli()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("source=CSV.result", result.stdout)
+        self.assertIn("pass=1 fail=1 environment-fail=1 skipped=1 not-run=1 total=5", result.stdout)
+
+    def test_yaml_counts_do_not_drive_status_summary(self):
+        self.write_csv([{"id": "TC-001", "result": "fail"}])
+        self.write_yaml([{"id": "TC-001", "result": "fail"}])
+        result = self.run_cli()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("pass=0 fail=1", result.stdout)
+        self.assertNotIn("pass=999", result.stdout)
+
+    def test_yaml_result_count_mismatch_is_nonzero(self):
+        self.write_yaml([{"id": row["id"], "result": row["result"]} for row in self.rows[:-1]])
+        result = self.run_cli()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("YAML results 数量不一致", result.stderr)
+
+    def test_id_set_mismatch_is_nonzero(self):
+        self.write_yaml([{"id": row["id"], "result": row["result"]} for row in self.rows[:-1]] + [{"id": "TC-999"}])
+        result = self.run_cli()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("testcase id 集合不一致", result.stderr)
+
+    def test_duplicate_csv_id_is_nonzero(self):
+        rows = self.rows.copy()
+        rows[-1] = {"id": "TC-001", "result": "not-run"}
+        self.write_csv(rows)
+        result = self.run_cli()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("CSV testcase id 重复", result.stderr)
+
+    def test_missing_csv_id_is_nonzero(self):
+        rows = self.rows.copy()
+        rows[0] = {"id": "", "result": "pass"}
+        self.write_csv(rows)
+        result = self.run_cli()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("CSV 第 2 行 testcase id 缺失", result.stderr)
+
+    def test_duplicate_or_missing_yaml_id_is_nonzero(self):
+        cases = [
+            ([{"id": "TC-001"}, {"id": "TC-001"}] + [{"id": row["id"]} for row in self.rows[2:]], "YAML testcase id 重复"),
+            ([{"id": ""}] + [{"id": row["id"]} for row in self.rows[1:]], "YAML results[1] testcase id 缺失"),
+        ]
+        for results, expected in cases:
+            with self.subTest(expected=expected):
+                self.write_yaml(results)
+                result = self.run_cli()
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn(expected, result.stderr)
+
+    def test_unknown_or_missing_csv_result_is_nonzero(self):
+        for bad_result, expected in [("unexpected", "result 未知"), ("", "result 缺失")]:
+            with self.subTest(bad_result=bad_result):
+                rows = self.rows.copy()
+                rows[0] = {"id": "TC-001", "result": bad_result}
+                self.write_csv(rows)
+                result = self.run_cli()
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn(expected, result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

skills/tester/tests/test_two_tier.py

@@ -0,0 +1,190 @@
+import csv
+import json
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+import yaml
+
+SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
+CATALOG = SCRIPTS / "catalog.py"
+CASES_CSV = SCRIPTS / "generate_cases_csv.py"
+RUN_CSV = SCRIPTS / "generate_run_csv.py"
+sys.path.insert(0, str(SCRIPTS))
+from two_tier import classify_integration, run_integration_first
+
+
+def write_mixed_catalog(root: Path) -> None:
+    tester = root / ".pouch" / "tester"
+    (tester / "cases").mkdir(parents=True)
+    (tester / "runs").mkdir()
+    (tester / "catalog.yaml").write_text(
+        yaml.safe_dump(
+            {
+                "version": 2,
+                "id_pattern": "^TC-[A-Z]+-[0-9]+$",
+                "categories": [{"id": "MIXED"}],
+                "risks": ["readonly", "mutating", "destructive"],
+                "default_max_risk": "readonly",
+                "client_suites": {
+                    "client-smoke": {"default": True},
+                    "client-regression": {"default": False},
+                    "safety": {"default": False},
+                },
+                "sessions": {"browser": {"kind": "browser", "reuse": "run"}},
+            },
+            sort_keys=False,
+        ),
+        encoding="utf-8",
+    )
+    journey = {
+        "id": "TC-JOURNEY-001",
+        "test_type": "journey",
+        "category": "MIXED",
+        "title": "journey",
+        "what": "client path",
+        "lifecycle": "active",
+        "tag": "smoke",
+        "risk": "readonly",
+        "driver": "browser",
+        "automation": "automated",
+        "session": "browser",
+        "client_suite": "client-smoke",
+        "platform": "android",
+        "journey": {
+            "goal": "goal",
+            "entry_state": "logged_out",
+            "actions": ["login"],
+            "assertions": ["visible"],
+            "final_state": "done",
+            "failure_policy": "stop",
+        },
+        "fixture": {"account": "test-role"},
+        "cleanup": {"required": False, "reason": "read-only"},
+        "how": {"steps": ["open"]},
+        "expected": [{"kind": "visible-text", "value": "done"}],
+    }
+    integration = {
+        "id": "TC-INTEGRATION-001",
+        "test_type": "integration",
+        "category": "MIXED",
+        "title": "contract",
+        "what": "API contract",
+        "lifecycle": "active",
+        "tag": "slow",
+        "risk": "readonly",
+        "driver": "script",
+        "automation": "automated",
+        "integration_suite": "api-contract",
+        "fixture": {"references": ["isolated-account"]},
+        "request": {"method": "GET", "path": "/health"},
+        "how": {"run": "true"},
+        "cleanup": {"required": False, "reason": "read-only"},
+        "expected": [{"kind": "api-status", "value": 200}],
+    }
+    (tester / "cases" / "mixed.yaml").write_text(
+        yaml.safe_dump({"category": "MIXED", "cases": [journey, integration]}, sort_keys=False),
+        encoding="utf-8",
+    )
+
+
+class TwoTierCatalogTests(unittest.TestCase):
+    def run_catalog(self, root: Path, *args: str) -> subprocess.CompletedProcess[str]:
+        return subprocess.run(
+            [sys.executable, str(CATALOG), *args, "--root", str(root)],
+            text=True,
+            capture_output=True,
+        )
+
+    def test_default_selection_excludes_integration_and_explicit_selection_includes_it(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            write_mixed_catalog(root)
+            valid = self.run_catalog(root, "validate")
+            self.assertEqual(valid.returncode, 0, valid.stderr)
+
+            default = self.run_catalog(root, "select")
+            self.assertEqual(default.returncode, 0, default.stderr)
+            default_payload = json.loads(default.stdout)
+            self.assertEqual(default_payload["count"], 1)
+            self.assertEqual(default_payload["test_types"], {"journey": 1, "integration": 0})
+
+            integration = self.run_catalog(root, "select", "--test-type", "integration")
+            self.assertEqual(integration.returncode, 0, integration.stderr)
+            integration_payload = json.loads(integration.stdout)
+            self.assertEqual(integration_payload["count"], 1)
+            self.assertEqual(integration_payload["cases"][0]["test_type"], "integration")
+
+            csv_result = subprocess.run(
+                [sys.executable, str(CASES_CSV), "--root", str(root)],
+                text=True,
+                capture_output=True,
+            )
+            self.assertEqual(csv_result.returncode, 0, csv_result.stderr)
+            with (root / ".pouch/tester/cases.csv").open(encoding="utf-8-sig", newline="") as handle:
+                rows = list(csv.DictReader(handle))
+            self.assertEqual({row["test_type"] for row in rows}, {"journey", "integration"})
+            self.assertEqual({row["suite"] for row in rows}, {"client-smoke", "api-contract"})
+
+            run_path = root / ".pouch/tester/runs/two-tier.yaml"
+            run_path.write_text(
+                yaml.safe_dump(
+                    {"runId": "two-tier", "results": [
+                        {"id": "TC-INTEGRATION-001", "result": "pass"},
+                        {"id": "TC-JOURNEY-001", "result": "not-run"},
+                    ]},
+                    sort_keys=False,
+                ),
+                encoding="utf-8",
+            )
+            run_csv = subprocess.run(
+                [sys.executable, str(RUN_CSV), "--root", str(root), "--run", str(run_path)],
+                text=True,
+                capture_output=True,
+            )
+            self.assertEqual(run_csv.returncode, 0, run_csv.stderr)
+            with run_path.with_suffix(".csv").open(encoding="utf-8-sig", newline="") as handle:
+                report_rows = list(csv.DictReader(handle))
+            self.assertEqual(report_rows[0]["test_type"], "integration")
+            self.assertEqual(report_rows[0]["suite"], "api-contract")
+
+
+class IntegrationFirstTests(unittest.TestCase):
+    def test_preflight_block_does_not_create_session(self):
+        calls: list[str] = []
+        result = run_integration_first(
+            preflight_status="blocked",
+            integration_cases=["INT-1"],
+            journey_cases=["J-1"],
+            execute_integration=lambda case: calls.append("integration") or "pass",
+            create_journey_session=lambda: calls.append("session"),
+            execute_journey=lambda session, case: "pass",
+        )
+        self.assertEqual(result["results"][0]["result"], "not-run")
+        self.assertFalse(result["session_created"])
+        self.assertEqual(calls, [])
+
+    def test_integration_failure_propagates_journey_not_run(self):
+        calls: list[str] = []
+        result = run_integration_first(
+            preflight_status="ready",
+            integration_cases=["INT-1", "INT-2"],
+            journey_cases=["J-1", "J-2"],
+            execute_integration=lambda case: calls.append(case) or ("fail" if case == "INT-1" else "pass"),
+            create_journey_session=lambda: calls.append("session"),
+            execute_journey=lambda session, case: "pass",
+        )
+        self.assertEqual(result["status"], "fail")
+        self.assertEqual(calls, ["INT-1"])
+        self.assertFalse(result["session_created"])
+        self.assertEqual([row["result"] for row in result["results"]], ["fail", "not-run", "not-run", "not-run"])
+
+    def test_service_unreachable_is_environment_failure(self):
+        self.assertEqual(classify_integration(ok=False, service_reachable=False), "environment-fail")
+        self.assertEqual(classify_integration(ok=False, service_reachable=True), "fail")
+
+
+if __name__ == "__main__":
+    unittest.main()