feat(tester): add integration-first journey execution

skills/tester/scripts/catalog.py

@@ -18,7 +18,15 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
-from journey import CLIENT_SUITES, client_suite, contract_errors, preflight as journey_preflight
+from journey import (
+    CLIENT_SUITES,
+    INTEGRATION_SUITES,
+    TEST_TYPES,
+    contract_errors,
+    preflight as journey_preflight,
+    suite as case_suite,
+    test_type,
+)
 
 try:
     import yaml
@@ -370,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) + " |")
 
@@ -388,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:
@@ -783,7 +798,13 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
                 continue
             if args.tag and case.get("tag") != args.tag:
                 continue
-            if args.suite and client_suite(case) != args.suite:
+            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
@@ -814,6 +835,8 @@ 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))
@@ -850,6 +873,10 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
                 "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,
@@ -943,6 +970,17 @@ def validate(root: Path) -> list[str]:
             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):
@@ -1297,8 +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,
-                       help="客户端旅程套件")
+    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,8 @@ except ImportError:  # pragma: no cover
 FIELDS = (
     "category",
     "id",
+    "test_type",
+    "suite",
     "client_suite",
     "platform",
     "maps_to",
@@ -158,11 +162,16 @@ 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,
-                    "client_suite": case.get("client_suite") or case.get("suite") or "",
+                    "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 "",

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,8 @@ EXECUTED_YES = frozenset({"pass", "fail", "environment-fail"})
 EXECUTED_NO = frozenset({"skipped", "not-run"})
 FIELDS = (
     "id",
+    "test_type",
+    "suite",
     "client_suite",
     "platform",
     "result",
@@ -131,10 +134,15 @@ 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,
-                    "client_suite": case.get("client_suite") or case.get("suite") or "",
+                    "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),

skills/tester/scripts/journey.py

@@ -13,7 +13,9 @@ 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")
@@ -28,6 +30,33 @@ def client_suite(case: Mapping[str, Any]) -> str:
     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]
@@ -57,16 +86,23 @@ def contract_errors(case: Mapping[str, Any], sessions: Mapping[str, Any] | None
     schema.  Once opted in, all lifecycle inputs are explicit so a missing
     fixture/cleanup can stop the run before any case result is fabricated.
     """
-    suite = client_suite(case)
-    journey = case.get("journey")
-    if not suite and journey is None:
+    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 not in CLIENT_SUITES:
+    if suite_name not in CLIENT_SUITES:
         errors.append(f"{where}: client_suite 必须是 {', '.join(CLIENT_SUITES)}")
 
     if not isinstance(journey, Mapping):
@@ -116,11 +152,69 @@ def contract_errors(case: Mapping[str, Any], sessions: Mapping[str, Any] | None
             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 == "safety" and str(case.get("tag") or "") == "smoke":
+    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)

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}")
-                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"})
-                prepared[platform] = (result, node, 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({"node_preflight": node, "status": "preflight-passed"})
+                    prepared[platform] = (result, node, identity)
+                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/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,
+    }