diff --git a/skills/test-preflight/README.md b/skills/test-preflight/README.md
index e339cf7..88e49a3 100644
--- a/skills/test-preflight/README.md
+++ b/skills/test-preflight/README.md
@@ -2,8 +2,10 @@
 
 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.
+artifacts, tester nodes, devices, network, services, mobile album/photo/pinyin
+capabilities, and fixture/cleanup contracts. Capability checks run on the actual
+Android/iOS tester node; authorization environment variables are not accepted as
+a substitute.
 
 Run the self-tests with:
 
diff --git a/skills/test-preflight/SKILL.md b/skills/test-preflight/SKILL.md
index 03fa5ef..6f0eae3 100644
--- a/skills/test-preflight/SKILL.md
+++ b/skills/test-preflight/SKILL.md
@@ -2,8 +2,9 @@
 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.
+  artifacts, tester bindings, HOST/PUBLIC_HOST, service probes, mobile capability
+  probes, and fixture/cleanup contracts. Use before tester runs; it is not a deployment,
+  build, or business-data skill.
 ---
 
 # test-preflight
@@ -40,12 +41,20 @@ Mac or that a deployment happened.
 
 Overall status is `ready`, `blocked`, or `environment-fail`:
 
-- `blocked` means a required declaration, artifact, binding, runner, fixture,
+- `blocked` means a required declaration, artifact, binding, runner, capability, 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.
 
+For every mobile matrix row, preflight requires real tester-node capability evidence for
+`album_picker`, `photo_permission` (including a usable test image), and `ime_pinyin`
+(composition and candidate selection). Android evidence must exercise the Android picker,
+permission state and real Chinese IME composition; iOS evidence must exercise the corresponding
+Photos picker/permission and text-input capabilities on the iOS node. `AUTHORIZE_ALBUM` and
+`IME_PINYIN` are never accepted as substitutes for these probes.
+
 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.
+no client session may be created; the tester owns `not-run` propagation and must write
+one result row per selected testcase with `blocked_by: [test-preflight]`.
diff --git a/skills/test-preflight/references/schema.md b/skills/test-preflight/references/schema.md
index 0af5aac..9096a6d 100644
--- a/skills/test-preflight/references/schema.md
+++ b/skills/test-preflight/references/schema.md
@@ -18,6 +18,10 @@ fixtures:
 cleanup:
   required: true
   steps: [delete-run-data]
+capabilities:
+  # These baseline checks are mandatory for mobile rows and cannot be disabled
+  # with AUTHORIZE_ALBUM or IME_PINYIN.
+  required: [album_picker, photo_permission, ime_pinyin]
 matrix:
   - platform: android
     tester:
@@ -48,6 +52,35 @@ 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.
 
+## Mobile capability contract
+
+Every Android/iOS matrix row checks these capabilities on the actual tester node:
+
+| capability | required evidence |
+|---|---|
+| `album_picker` | `checks.picker`: the real platform album/Photos picker opened and returned a selectable asset |
+| `photo_permission` | `checks.permission` and `checks.test_image`: permission state was exercised and the declared test image was available |
+| `ime_pinyin` | `checks.composition` and `checks.candidates`: real Chinese pinyin composition and candidate selection succeeded |
+
+The default list is always applied; `capabilities.required` can add project-specific
+checks but cannot remove these three. A contract runner receives one request per
+capability:
+
+```json
+{
+  "operation": "capability_probe",
+  "platform": "android",
+  "node": "tester.android",
+  "capability": "ime_pinyin",
+  "required_checks": ["composition", "candidates"]
+}
+```
+
+It must return every requested check with `status`, `observed`, and (when useful)
+`expected`/`remediation`. Missing evidence is `blocked`; a real probe failure is
+`environment-fail`. Environment variables such as `AUTHORIZE_ALBUM` and
+`IME_PINYIN` are not capability evidence.
+
 ## Tester-node contract
 
 `tester.runner.kind: local` executes platform and network probes on the current
diff --git a/skills/test-preflight/scripts/preflight.py b/skills/test-preflight/scripts/preflight.py
index fde8bcb..66dbbd5 100644
--- a/skills/test-preflight/scripts/preflight.py
+++ b/skills/test-preflight/scripts/preflight.py
@@ -31,6 +31,17 @@ except ImportError as exc:  # pragma: no cover
 
 PLATFORMS = {"android", "ios"}
 STATUSES = {"ready", "blocked", "environment-fail"}
+CAPABILITY_CHECKS = {
+    "album_picker": ("picker",),
+    "photo_permission": ("permission", "test_image"),
+    "ime_pinyin": ("composition", "candidates"),
+}
+CAPABILITY_EXPECTED = {
+    "album_picker": "real platform album picker opens and returns a selectable asset",
+    "photo_permission": "photo permission flow is observable and the declared test image is available",
+    "ime_pinyin": "real Chinese pinyin composition produces and selects candidate words",
+}
+DEFAULT_CAPABILITIES = tuple(CAPABILITY_CHECKS)
 EVIDENCE_FIELDS = [
     "check_id", "mode", "platform", "node", "target", "observed", "expected",
     "status", "remediation", "timestamp",
@@ -161,6 +172,27 @@ def validate_shape(config: Mapping[str, Any], evidence: Evidence) -> bool:
     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
+    capabilities = config.get("capabilities")
+    if capabilities is not None and not isinstance(capabilities, (list, dict)):
+        evidence.add(
+            "config-capabilities",
+            target="capabilities",
+            observed=capabilities,
+            expected="list or mapping with required list",
+            status="blocked",
+            remediation="声明 capabilities.required；不能用 AUTHORIZE_ALBUM/IME_PINYIN 代替能力检查",
+        )
+        ok = False
+    elif isinstance(capabilities, dict) and capabilities.get("required") is not None and not isinstance(capabilities.get("required"), list):
+        evidence.add(
+            "config-capabilities-required",
+            target="capabilities.required",
+            observed=capabilities.get("required"),
+            expected="list of capability ids",
+            status="blocked",
+            remediation="将 capabilities.required 写成列表",
+        )
+        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 探针")
@@ -422,6 +454,122 @@ def check_fixture_cleanup(evidence: Evidence, config: Mapping[str, Any], platfor
     return ok
 
 
+def capability_ids(config: Mapping[str, Any]) -> list[str]:
+    """Return mandatory mobile capabilities.
+
+    The three baseline capabilities are never disabled by an authorization
+    environment variable.  A config may add more checks, but it cannot remove
+    the album/photo/pinyin checks required for real client journeys.
+    """
+    raw = config.get("capabilities")
+    requested: Any = raw.get("required") if isinstance(raw, dict) else raw
+    ids = list(DEFAULT_CAPABILITIES)
+    if isinstance(requested, list):
+        for item in requested:
+            name = text(item).strip()
+            if name and name not in ids:
+                ids.append(name)
+    return ids
+
+
+def check_capabilities(
+    evidence: Evidence,
+    config: Mapping[str, Any],
+    row: Mapping[str, Any],
+    platform_name: str,
+    node: str,
+    index: int,
+    *,
+    platform_ok: bool,
+) -> bool:
+    """Run real capability probes on the resolved tester node.
+
+    A contract runner must return each required sub-check.  We intentionally do
+    not infer album, permission, image, or IME behavior from adb/xcrun alone:
+    the adapter must exercise the real picker/permission/UI composition path.
+    """
+    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"}
+    capabilities_ok = True
+    for capability in capability_ids(config):
+        required = CAPABILITY_CHECKS.get(capability, ("capability",))
+        prefix = f"{platform_name}-{index}-capability-{capability}"
+        if not platform_ok:
+            for check in required:
+                evidence.add(
+                    f"{prefix}-{check}",
+                    platform=platform_name,
+                    node=node,
+                    target=capability,
+                    observed="platform prerequisite failed",
+                    expected=CAPABILITY_EXPECTED.get(capability, capability),
+                    status="environment-fail",
+                    remediation="先修复平台/设备闸门，再由 tester 节点重跑真实能力探针",
+                )
+            capabilities_ok = False
+            continue
+        if not (isinstance(runner, dict) and runner.get("kind") == "contract"):
+            for check in required:
+                evidence.add(
+                    f"{prefix}-{check}",
+                    platform=platform_name,
+                    node=node,
+                    target=capability,
+                    observed="no real capability runner",
+                    expected=CAPABILITY_EXPECTED.get(capability, capability),
+                    status="blocked",
+                    remediation="配置 tester 节点的 contract runner，真实打开 picker/权限 UI 并完成拼音 composition/candidate 检查",
+                )
+            capabilities_ok = False
+            continue
+
+        status, observed, result = runner_call(
+            runner,
+            {
+                "operation": "capability_probe",
+                "platform": platform_name,
+                "node": node,
+                "capability": capability,
+                "required_checks": list(required),
+                "artifact": (row.get("artifact") or {}).get("path"),
+            },
+        )
+        checks = result.get("checks") if isinstance(result, dict) else None
+        if not isinstance(checks, dict):
+            checks = {}
+        for check in required:
+            detail = checks.get(check)
+            if detail is None:
+                evidence.add(
+                    f"{prefix}-{check}",
+                    platform=platform_name,
+                    node=node,
+                    target=capability,
+                    observed=observed or "runner omitted required evidence",
+                    expected=CAPABILITY_EXPECTED.get(capability, capability),
+                    status="blocked",
+                    remediation=f"runner 必须返回 checks.{check}，且证据来自真实 {platform_name} 能力操作",
+                )
+                capabilities_ok = False
+                continue
+            detail = detail if isinstance(detail, dict) else {"observed": detail}
+            detail_status = status_from(detail.get("status"), status)
+            evidence.add(
+                f"{prefix}-{check}",
+                platform=platform_name,
+                node=node,
+                target=text(detail.get("target", capability)),
+                observed=detail.get("observed", observed),
+                expected=detail.get("expected", CAPABILITY_EXPECTED.get(capability, capability)),
+                status=detail_status,
+                remediation=text(detail.get("remediation", "")),
+            )
+            if detail_status != "ready":
+                capabilities_ok = False
+    return capabilities_ok
+
+
 def run(config: Mapping[str, Any], mode: str) -> Evidence:
     evidence = Evidence(mode)
     shape_ok = validate_shape(config, evidence)
@@ -479,7 +627,16 @@ def run(config: Mapping[str, Any], mode: str) -> Evidence:
             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
+        capabilities_ok = check_capabilities(
+            evidence,
+            config,
+            row,
+            platform_name,
+            node,
+            index,
+            platform_ok=platform_ok,
+        )
+        row_ok = artifact_ok and fixture_ok and platform_ok and network_ok and capabilities_ok
         if mode == "prepare":
             artifact = row.get("artifact") or {}
             identity = artifact.get("package" if platform_name == "android" else "bundle_id")
diff --git a/skills/test-preflight/tests/fixtures/contract_runner.py b/skills/test-preflight/tests/fixtures/contract_runner.py
index f4d5e95..0747356 100644
--- a/skills/test-preflight/tests/fixtures/contract_runner.py
+++ b/skills/test-preflight/tests/fixtures/contract_runner.py
@@ -50,6 +50,25 @@ elif operation == "network_probe":
             "tcp": {"status": "ready", "observed": "443 open", "expected": "TCP reachable"},
             "http": {"status": "ready", "observed": 200, "expected": "HTTP 2xx-4xx"},
         }}))
+elif operation == "capability_probe":
+    capability = request.get("capability")
+    checks = {
+        "album_picker": {
+            "picker": {"status": "ready", "observed": "real picker opened and returned image-001"},
+        },
+        "photo_permission": {
+            "permission": {"status": "ready", "observed": "permission state checked on device"},
+            "test_image": {"status": "ready", "observed": "image-001 available in test album"},
+        },
+        "ime_pinyin": {
+            "composition": {"status": "ready", "observed": "nihao composed in Chinese IME"},
+            "candidates": {"status": "ready", "observed": "你好 candidate selected"},
+        },
+    }
+    if scenario == "capability-missing" and capability == "album_picker":
+        print(json.dumps({"status": "ready", "checks": {}}))
+    else:
+        print(json.dumps({"status": "ready", "checks": checks.get(capability, {})}))
 elif operation in {"install", "start"}:
     print(json.dumps({"status": "ready", "observed": operation + "ed mock artifact"}))
 else:
diff --git a/skills/test-preflight/tests/test_preflight.py b/skills/test-preflight/tests/test_preflight.py
index 03b7799..c67a60f 100644
--- a/skills/test-preflight/tests/test_preflight.py
+++ b/skills/test-preflight/tests/test_preflight.py
@@ -140,6 +140,24 @@ class PreflightTests(unittest.TestCase):
         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_mobile_capabilities_are_mandatory_and_missing_evidence_blocks(self):
+        data = self.config()
+        config_path = self.write_config(data)
+        env = os.environ.copy()
+        env["PREFLIGHT_RUNNER_SCENARIO"] = "capability-missing"
+        output = self.root / "capabilities.yaml"
+        process = subprocess.run([
+            sys.executable, str(ROOT / "scripts" / "preflight.py"), "verify",
+            "--config", str(config_path), "--yaml", str(output), "--csv", str(self.root / "capabilities.csv"),
+        ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
+        report = __import__("yaml").safe_load(output.read_text(encoding="utf-8"))
+        self.assertEqual(process.returncode, 1)
+        self.assertEqual(report["status"], "blocked")
+        capability_rows = [row for row in report["checks"] if "capability-album_picker" in row["check_id"]]
+        self.assertTrue(capability_rows)
+        self.assertTrue(all(row["status"] == "blocked" for row in capability_rows))
+        self.assertTrue(any("photo_permission" in row["check_id"] for row in report["checks"]))
+
     def test_host_unreachable_from_actual_runner_is_environment_fail(self):
         data = self.config()
         config_path = self.write_config(data)
diff --git a/skills/tester/README.md b/skills/tester/README.md
index 12e0238..38a7d9d 100644
--- a/skills/tester/README.md
+++ b/skills/tester/README.md
@@ -27,7 +27,12 @@
   每轮测试会创建新的 terminal，结束后自动关闭；不会复用或关闭已有 terminal。
 - 测试环境部署由 deployer 在本机 Compose 中触发；正式环境部署由本机 `~/code/app00`
   项目的 fresh Agent terminal 触发，tester 不直接 SSH 或执行正式 Compose。
-- 会动硬件、会改数据的用例带风险级别（`readonly` / `mutating` / `destructive`）。默认只跑最低档 `readonly`。要跑更高风险，需要你明确说，例如「风险到 mutating」；碰哪个子系统另由 `gate`（如 hardware / weld）控制。
+- 会动硬件、会改数据的用例带风险级别（`readonly` / `mutating` / `destructive`）。默认
+  `standard` 只跑最低档 `readonly`。隔离、可重置的 Writes 类环境在 catalog 设置
+  `execution_profile: isolated-full` 后，所有 active 自动化用例都进入执行，不因
+  `AUTHORIZE_MUTATING`、`AUTHORIZE_DESTRUCTIVE`、`AUTHORIZE_ALBUM` 或
+  `AUTHORIZE_PASSWORD` skipped；risk 原值仍保留，必须用 test-preflight 的真实能力检查和
+  cleanup/rollback 保证安全。
 
 ## 使用示例
 
@@ -76,7 +81,8 @@ APK/IPA，为本轮创建隔离 worktree，在 Android 和 iOS tester 节点安
    不执行可用平台；只有全部可用才按目标环境走 deployer 路由、解析 Agent、输出生效
    列表、准备 Orca 项目并创建临时 terminal。脚本类、浏览器和移动模拟器都在对应
    Agent 环境中执行。`builder.<platform>` 不能替代 `tester.<platform>`。
-4. 超出你允许的风险级别会跳过，不会偷偷执行。
+4. `standard` 下超出你允许的风险级别会跳过；`isolated-full` 由 catalog profile 明确
+   全量 active 执行，不把风险悄悄改成 readonly。
 5. 无论测试成功、失败或异常，都会关闭本轮创建的 terminal；跑完给出按分类计数的
    结果，以及 `pass / fail / environment-fail / skipped / not-run`。
 
@@ -117,6 +123,14 @@ cleanup 缺失会在开始前阻断；前置失败后的剩余用例是 `not-run
 `safety` 必须显式选择和授权，不进入默认 smoke/full。它与普通 tester 用例共用原有
 CSV/YAML 运行记录和汇总校验。
 
+移动 preflight 还必须在实际 tester 节点验证相册 picker、照片权限与测试图片，以及中文
+拼音的真实 composition 和候选词选择；缺失时所有 testcase 都保留 `not-run` +
+`blocked_by`，不会静默 `skipped`。
+
+失败回流使用 run YAML 顶层 `defects`/`test_issues` 与 `feedback`：PR 触发先更新 PR/关联
+issue、通知 dispatcher 重新分配开发；需求/bug 触发直接回开发流程。可用
+`scripts/record_issues.py` 生成同名 CSV 的结构化 issue 列。
+
 ### 汇总已有运行结果
 
 对已经生成的同一运行 CSV/YAML，使用显式路径汇总：
diff --git a/skills/tester/SKILL.md b/skills/tester/SKILL.md
index fd90440..c610f46 100644
--- a/skills/tester/SKILL.md
+++ b/skills/tester/SKILL.md
@@ -74,10 +74,10 @@ tester agent 配置。
 - `local.yaml.example` 的 HOST 默认是 `127.0.0.1`；如果目标不是本机，必须在
   `local.yaml` 或环境变量中显式覆盖。`local.yaml` 不存在且没有环境变量时仍不猜
   HOST，停下并记 environment-fail。
-- 默认风险上限是 catalog 里的 `default_max_risk`。超出上限的条目标 `skipped`，禁止执行。
+- 默认执行 profile 是 `standard`：风险上限来自 catalog 的 `default_max_risk`，超出上限的条目标 `skipped`，禁止执行。若 catalog 声明 `execution_profile: isolated-full`，只允许用于隔离且可重置的测试环境：所有 active 自动化用例都进入执行矩阵，不因 `AUTHORIZE_MUTATING`、`AUTHORIZE_DESTRUCTIVE`、`AUTHORIZE_ALBUM` 或 `AUTHORIZE_PASSWORD` skipped；原有 `risk` 值必须保留并进入报告，安全边界改由 test-preflight 能力探针和 cleanup/rollback 契约保证。
 - `risk` 只能是 `readonly` / `mutating` / `destructive`，默认只跑 `readonly`。`tag=smoke` 只能配 `risk=readonly`。规则见 [levels.md](references/levels.md)。
 - `driver=browser` 必须开真实页面；`driver=mobile` 必须挂模拟器 session。二者都不得改成只打 API。
-- `automation=assisted|manual` 需要人在场：无人值守时记 `not-run`，不要自己代跑人工步骤。见 [concepts.md](references/concepts.md)。
+- `automation=assisted|manual` 需要人在场：无人值守时记 `not-run`，不要自己代跑人工步骤。isolated-full 取消的是风险/授权过滤，不会伪造人工操作。见 [concepts.md](references/concepts.md)。
 - 浏览器、移动模拟器按 catalog `sessions` 初始化一次，同一次运行内复用。
 - Agent、环境、项目目录和 Orca terminal 生命周期见 [execution.md](references/execution.md)：可以从 project id/名称或 terminal id 自动解析，更新后输出配置列表；每轮都新建 terminal，结束后关闭本轮创建的 terminal，不复用或关闭已有 terminal。
 - 环境准备见 [deployment.md](references/deployment.md)：测试环境只调用 deployer 的本机 Compose 路径；正式环境只委派给本机 `~/code/app00` 的 fresh Agent terminal，tester 不直接 SSH 或执行生产 Compose。
@@ -189,6 +189,13 @@ python3 <skill-dir>/scripts/catalog.py select --root <root> --suite safety --max
 客户端运行器统一遵守 `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 值。
+能力缺失时 preflight 必须明确 blocked/environment-fail，并为所有选中的 testcase 生成
+`not-run` 结果行；不能把缺失的相册、照片权限、测试图片或中文拼音 composition/candidate
+能力静默变成 `skipped`。独立用例继续执行，依赖失败的用例写 `not-run` 和 `blocked_by`；
+修复后按 targeted rerun，再执行 full integration → journey。成绩仍写现有
+`runs/<run-id>.yaml` 并用 `generate_run_csv.py` 生成同名 CSV；YAML 可带 `defects`/
+`test_issues` 与 `feedback`，记录 testcase、product/test issue 类型、expected/observed、
+artifact hash、environment、证据路径、复现和下一步，不添加新的 result 值。
+
+PR 触发的失败回流必须先更新 PR/关联 issue，再通知 dispatcher 重新分配开发；需求/bug
+触发的失败直接返回开发流程。tester 只生成结构化回流计划，不在本地偷偷修改远端。
diff --git a/skills/tester/references/concepts.md b/skills/tester/references/concepts.md
index 67412d8..a3a5531 100644
--- a/skills/tester/references/concepts.md
+++ b/skills/tester/references/concepts.md
@@ -35,6 +35,7 @@ catalog（整个测试集）
 | `session` | 可复用会话（浏览器 / 模拟器） | pytest fixture；Robot suite variable | 不是一条用例 |
 | `gate` | 项目细分闸门名 → 环境变量（`motion` / `hardware` / `weld`） | 授权开关 | 不是 `risk`，不要拿它代替风险等级 |
 | `authorize` | 风险档 → 环境变量。该档执行前必须为 `true` | pytest `skipif` 条件 | 不是认证（auth），是**授权** |
+| `execution_profile` | 执行环境策略：`standard` 或隔离可重置的 `isolated-full` | pytest marker/fixture policy | 不是风险等级；isolated-full 不会抹掉 `risk` |
 | `assume` | 额外跳过条件（gate / authorize 之外的环境表达式） | JUnit `assumeTrue`；pytest `skipif` | 不是授权开关的重复写法 |
 | `status_probe` | 分类级前置条件：请求一个 URL，命中即视为前置不满足 | pytest session fixture；Robot Suite Setup | 不是产品失败，不是整轮硬闸 |
 | `how` | 怎么测：`steps` / `run` / `url` / `timeout` | test steps | 不是判定标准 |
@@ -55,7 +56,7 @@ catalog（整个测试集）
 | `pass` | 通过 | passed |
 | `fail` | 产品缺陷 | failed |
 | `environment-fail` | 前置/环境导致无法执行，**永远不是产品失败** | pytest error；JUnit aborted |
-| `skipped` | **主动**不跑：`retired` / `pending` / 超风险上限 / 未授权 / `assume` 不成立 | pytest skipped；JUnit disabled |
+| `skipped` | **主动**不跑：standard 下的 `retired` / `pending` / 超风险上限 / 未授权 / `assume` 不成立 | pytest skipped；JUnit disabled |
 | `not-run` | **被动**没轮到：失败即停后剩下的；分类被 `status_probe` 挡住整组没执行；`automation` 需人工而无人值守 | pytest not run；deselected |
 
 ## 2. 命名迁移对照（v1 → v2）
@@ -82,6 +83,8 @@ catalog（整个测试集）
 - 三级**单调**：`readonly ⊂ mutating ⊂ destructive`，所以只需要一个上限 `default_max_risk` 和一个 `--max-risk`，不用组合矩阵。
 - 与 `gate` 正交：`risk` 说「影响多大」，`gate` 说「碰哪个子系统」。点动 vs 焊接这种细分授权用 `gate`，不要再造 risk 名。
 - `tag=smoke` 只能配 `risk=readonly`：冒烟测试不许留下变化。
+- 隔离可重置环境可设置 `execution_profile: isolated-full` 跑所有 active 用例；安全前置由
+  test-preflight capability 与 cleanup/rollback 契约保证，原始 risk 仍必须报告。
 
 ### 3.2 `automation`：为什么是 `automated / assisted / manual`
 
@@ -93,6 +96,7 @@ catalog（整个测试集）
   - `assisted`：Agent 驱动，但至少一步需要人在场；该步在 `how.steps` 里以 `[人工]` 开头（可被 `validate` 机器判定）。
   - `manual`：只能人工执行；禁止 `driver`、禁止 `how.run`，必须有 `how.steps`。
 - 默认只跑 `automated`。无人值守时 `assisted` / `manual` 记 `not-run`；有人在场时 `assisted` 暂停等人确认，`manual` 由人工回报结果。
+- isolated-full 只取消风险/授权过滤，不伪造 assisted/manual 的人工步骤。
 
 ### 3.3 结果为什么拆出 `not-run`
 
diff --git a/skills/tester/references/levels.md b/skills/tester/references/levels.md
index 6f744de..92822b5 100644
--- a/skills/tester/references/levels.md
+++ b/skills/tester/references/levels.md
@@ -36,7 +36,11 @@
 | `destructive` | 难恢复或伤人/伤钱/伤环境 | 删库、真实扣费 | 清数据、内购 | 急停、断电、真焊 |
 
 默认只跑 `readonly`。项目细分闸（点动 vs 焊接）用可选 `gate`，不要再发明 risk 名。
+隔离且可重置的项目可在 catalog 设置 `execution_profile: isolated-full`，此时所有 active
+用例按原始 risk 执行；`risk` 仍用于报告和 cleanup 设计，不能改写为 readonly。
 
 `tag=smoke` 且 `risk!=readonly` 为非法。
 
-不想跑某一档时，`authorize.<risk>` 指向的环境变量保持 false 即可——该档用例记 `skipped`，不是 `not-run`。
+standard 下不想跑某一档时，`authorize.<risk>` 指向的环境变量保持 false 即可——该档用例记
+`skipped`，不是 `not-run`。isolated-full 不使用这些授权开关，改由 preflight 能力探针和
+cleanup/rollback 契约阻断不安全环境。
diff --git a/skills/tester/references/run.md b/skills/tester/references/run.md
index 3383277..9b8cfac 100644
--- a/skills/tester/references/run.md
+++ b/skills/tester/references/run.md
@@ -72,7 +72,11 @@ iOS     | … | … | page.writes | tester.ios     | remote-ios
 
 ## 1. 选范围
 
-未指定范围 → `--tag smoke`，风险上限 = `default_max_risk`，`automation` 默认只跑 `automated`。
+未指定范围 → `--tag smoke`，执行 profile = catalog 的 `execution_profile`（默认 `standard`），
+`standard` 的风险上限 = `default_max_risk`，`automation` 默认只跑 `automated`。
+`execution_profile=isolated-full` 只用于隔离可重置环境：不按风险上限或 `AUTHORIZE_*`/gate
+授权变量过滤 active 用例，保留每条用例原始 `risk`；安全门禁由 test-preflight 能力探针和
+cleanup/rollback 契约承担。
 
 用户说分类、id、tag、风险上限时，原样传给 select：
 
@@ -80,6 +84,7 @@ iOS     | … | … | page.writes | tester.ios     | remote-ios
 python3 <skill-dir>/scripts/catalog.py select --root <root> \
   [--category CAT] [--id ID]... [--tag smoke|full|slow] \
   [--max-risk <risk>] [--driver script|browser|mobile] \
+  [--execution-profile standard|isolated-full] \
   [--automation automated|assisted|manual|any] \
   [--test-type integration|journey] [--suite <client-or-integration-suite>] \
   [--platform android|ios]... \
@@ -111,11 +116,11 @@ python3 <skill-dir>/scripts/catalog.py select --root <root> --suite api-contract
 不能拿到 cases 后绕过闸门开始测试。默认 binding 是 `tester.<platform>`，所以 iOS
 会强制检查 `tester.ios`，不会回退到 `builder.ios` 或通用 `tester.test`。
 
-`--automation` 默认 `automated`（需人工的条目会被跳过，并在 JSON `skipped` 里说明）；用户明确要求现场配合时用 `assisted`，`any` 表示不按人工参与度筛。
+`--automation` 默认 `automated`（需人工的条目会被记为 `not-run`，不会伪造人工步骤）；用户明确要求现场配合时用 `assisted`，`any` 表示不按人工参与度筛。
 
 单根 `select` 的 JSON 字段：`max_risk` / `automation` / `count` / `vars`（已展开的变量，用来替换 `how` 里的 `${…}`）/ `artifacts` / `preflight` / `skipped` / `run_together` / `status_probe`（按分类，`url` 已展开）/ `cases`。从父仓库 fan-out 时返回 `mode: fan-out` 和每个根的独立结果，不能把多个根的 cases 拼成一个列表。
 
-`select` 已排除 `retired` 和风险高于上限的条目（后者若用户点了具体 id，会在 JSON 的 `skipped` 里说明，不执行）。
+`standard` 的 `select` 已排除 `retired` 和风险高于上限的条目（后者若用户点了具体 id，会在 JSON 的 `skipped` 里说明）；`isolated-full` 保留所有 active 条目，不因风险或授权变量 skipped。
 
 若 catalog 有 `run_together`，命中组内任一条就把整组按 `ids` 顺序扩进列表（JSON 带 `run_together`）。这不是擅自扩大范围。
 
@@ -135,8 +140,9 @@ preflight 通过后，先按 catalog 顺序执行 `integration`，再按客户
 ## 3. 运行前闸
 
 - `HOST` 未设（环境变量 / `local.yaml` 都没有）：全部停，environment-fail。
-- 命中条的 `risk` 在 catalog `authorize` 里有对应变量：变量不是 `true` 则该条 `skipped`。有 `gate` 时还要 `catalog.gates` 对应变量为 `true`；`assume` 不成立同样 `skipped`。即使用户点了 id 也不执行。
-- 授权/假设不成立是**主动不跑**（`skipped`），不是 `not-run`：这是我们决定不跑，不是没轮到。
+- `standard`：命中条的 `risk` 在 catalog `authorize` 里有对应变量时，变量不是 `true` 则该条 `skipped`；有 `gate` 时还要对应变量为 `true`，`assume` 不成立同样 `skipped`。即使用户点了 id 也不执行。
+- `isolated-full`：不读取 `AUTHORIZE_MUTATING`、`AUTHORIZE_DESTRUCTIVE`、`AUTHORIZE_ALBUM`、`AUTHORIZE_PASSWORD` 等人工开关；相册、照片权限/测试图片和中文拼音真实 composition/candidate 能力必须由 test-preflight 证明。
+- standard 中授权/假设不成立是**主动不跑**（`skipped`）；isolated-full 的能力缺失是 preflight blocked/environment-fail，并为全部 testcase 写 `not-run` + `blocked_by: [test-preflight]`。
 - `status_probe` 是**分类级前置条件**（业界对应 pytest 的 session fixture / Robot 的 Suite Setup）。用 `select` 返回的 `status_probe` 解析生效值，不要自己读 catalog 猜。返回的 `probe.url` 已用 `vars` 展开，直接请求即可；`select` 同批返回 `vars`：
 
   | `categories[].status_probe` | 生效探测 |
@@ -221,7 +227,7 @@ Android+iOS 默认是全量 preflight 闸门：任一项不满足，另一平台
 
 ## 5. 执行
 
-按 select 返回的顺序，一条一条来。select 按 `catalog.yaml` 的 `categories` 顺序，同一分类内按该 yaml 列表顺序（不是文件名字母序）。默认失败即停（产品 `fail` 或探测到的不安全状态）。用户说继续 / keep-going 才往后跑。
+按 select 返回的顺序，一条一条来。select 按 `catalog.yaml` 的 `categories` 顺序，同一分类内按该 yaml 列表顺序（不是文件名字母序）。同一 phase 的独立用例继续执行；只有显式 `depends_on` 的失败依赖才写 `not-run` + `blocked_by`。用户说 targeted rerun 时只重跑受影响用例，修复后再 full integration → journey。
 
 例外：当前条和下一条同属 `run_together` 且该组 `continue_on_fail: true` 时，前一条产品 `fail` 仍跑后一条。`environment-fail` 仍整轮停。
 
@@ -253,7 +259,7 @@ python3 <skill-dir>/scripts/catalog.py run-script --root <root> --id <id>
 
 ## 6. 记录与报告
 
-每条：`id, result, evidence, notes`。`result` 只能是 `pass` / `fail` / `environment-fail` / `skipped` / `not-run`。`skipped` 是主动不跑（retired / pending / 超风险上限 / 未授权 / `assume` 不成立），`not-run` 是没轮到（失败即停后剩下的 / 被前置整组挡住的 / 需人工而无人值守的）。
+每条：`id, result, evidence, notes`，依赖阻断时另有 `blocked_by`。`result` 只能是 `pass` / `fail` / `environment-fail` / `skipped` / `not-run`。`skipped` 是 standard 中主动不跑（retired / pending / 超风险上限 / 未授权 / `assume` 不成立）；`not-run` 是没轮到或 preflight/依赖阻断。失败用例可在 YAML 顶层 `defects`（兼容 `test_issues`）记录 testcase、`issue_type=product|test`、expected/observed、artifact hash、environment、evidence path、reproduction、next step。
 
 写入 `.pouch/tester/runs/<UTC时间>-<范围>-<随机后缀>.yaml`，可用
 `generate_run_csv.new_run_id(<范围>)` 生成同名 `runId`。记录包含唯一 `runId`、筛选条件、HOST 是否已设（可以写 HOST 值）、session 是否复用、每个分类的 `status_probe` 生效值与观测结果（`source` / `disabled` 原因 / `hit` / `observed`）、逐条结果。记录文件只创建一次，不覆盖已有文件。立刻用同一快照生成 CSV，不要手改：
diff --git a/skills/tester/references/schema.md b/skills/tester/references/schema.md
index 0c51fb1..4f33a49 100644
--- a/skills/tester/references/schema.md
+++ b/skills/tester/references/schema.md
@@ -25,6 +25,7 @@ runs/*.csv             同一次运行的成绩单；generate_run_csv.py 从对
 | `categories` | `{id, title, status_probe?}` 列表，id 与 `cases/<id小写>.yaml` 对应；列表顺序即 `select` / 执行顺序。`status_probe` 见下 |
 | `risks` | 从低到高，只能是 `readonly` → `mutating` → `destructive` 的一段，且以 `readonly` 开头。含义见 [levels.md](levels.md) |
 | `default_max_risk` | 未指定时的运行上限，必须是 `risks` 里的值 |
+| `execution_profile` | `standard`（默认）或 `isolated-full`。后者只用于隔离可重置环境，保留 risk 语义但不按 risk/authorize/gate 过滤 active 用例；安全边界由 preflight capability 与 cleanup/rollback 契约承担 |
 | `vars` | 可替换变量，如 `API_BASE: http://${HOST}:8095` |
 | `artifacts` | 可选的移动产物身份契约。按 `android` / `ios` 声明 `required` 和 `package` / `bundle_id`；每轮文件路径或下载地址由调用方提供，不提交临时产物路径。 |
 | `authorize` | 风险名 → 环境变量名。该档执行前变量必须为 `true`（`readonly` 不需要） |
@@ -36,6 +37,10 @@ runs/*.csv             同一次运行的成绩单；generate_run_csv.py 从对
 | `client_suites` | 可选的标准套件声明；出现时必须完整包含 `client-smoke`、`client-regression`、`safety`。缺省运行仍按 `tag` 兼容旧目录 |
 | `integration_suites` | 可选的集成套件声明；标准值为 `api-contract`、`authorization`、`build-contract` |
 
+`isolated-full` 不得把风险字段改成 `readonly`，也不得把 `AUTHORIZE_ALBUM`、
+`AUTHORIZE_PASSWORD`、`IME_PINYIN` 等变量当成能力门禁。移动用例必须通过 test-preflight
+的真实相册 picker、照片权限/测试图片和中文拼音 composition/candidate 检查。
+
 `local.yaml` 与环境变量覆盖 `HOST` 和授权开关。模板中的 `HOST` 默认是
 `127.0.0.1`；远程目标必须显式覆盖。查找顺序：环境变量 > `local.yaml` >
 `catalog.yaml` 的 vars。
@@ -91,6 +96,7 @@ terminal；运行结束用 `orca terminal close --terminal <handle>` 关闭本
 | `expected` | 是 | 怎么过，至少一条 `{kind, value}` |
 | `maps_to` | 否 | 对人文档的功能项 id |
 | `assume` | 否 | 额外跳过条件（环境变量表达式）。不要重复 `risk` / `gate` 已覆盖的授权变量，否则 `validate` 报错 |
+| `depends_on` | 否 | 依赖的 testcase id 列表；依赖 fail/environment-fail/not-run 时本条写 `not-run` 和 `blocked_by`，独立用例继续 |
 | `setup` | 否 | 人话前置 |
 | `teardown` | 否 | 跑完回退 |
 | `notes` | 否 | 备注 |
@@ -170,3 +176,24 @@ cleanup:
 ## 判定
 
 `pass` / `fail` / `environment-fail` / `skipped` / `not-run`
+
+## 运行失败回流
+
+run YAML 的 `results` 必须保留每条 testcase 行。失败可在顶层 `defects`（兼容别名
+`test_issues`）追加：
+
+```yaml
+defects:
+  - testcase: TC-CLIENT-001
+    issue_type: product # product | test
+    expected: "候选词可选择"
+    observed: "composition 成功但候选列表为空"
+    artifact_hash: sha256:...
+    environment: "Android emulator API 35 / tester.android"
+    evidence_path: .pouch/tester/evidence/TC-CLIENT-001.png
+    reproduction: "打开相册后输入 nihao"
+    next_step: "开发修复候选词渲染后 targeted rerun"
+```
+
+PR 触发的 run 可带 `feedback` 计划：先更新 PR/关联 issue，再通知 dispatcher 重新分配开发；
+需求/bug 触发则 route 直接为 development。tester 只记录计划，不隐式调用远端写操作。
diff --git a/skills/tester/scripts/catalog.py b/skills/tester/scripts/catalog.py
index 7c64323..e9a8ebd 100755
--- a/skills/tester/scripts/catalog.py
+++ b/skills/tester/scripts/catalog.py
@@ -42,6 +42,7 @@ LIFECYCLE_SKIP_REASON = {
 }
 TAGS = ("smoke", "full", "slow")
 CANONICAL_RISKS = ("readonly", "mutating", "destructive")
+EXECUTION_PROFILES = ("standard", "isolated-full")
 AUTOMATIONS = ("automated", "assisted", "manual")
 RESULTS = ("pass", "fail", "environment-fail", "skipped", "not-run")
 EXPECTED_KINDS = (
@@ -351,6 +352,21 @@ def risk_index(meta: dict[str, Any], name: str) -> int:
     return risks.index(name)
 
 
+def resolve_execution_profile(meta: dict[str, Any], requested: str | None = None) -> str:
+    """Resolve the run profile without changing the meaning of ``risk``.
+
+    ``isolated-full`` is an execution policy for a resettable test environment:
+    it disables risk/authorization filtering, but leaves the catalog risk value
+    on every case so reports still describe the side effect that was exercised.
+    """
+    profile = requested or meta.get("execution_profile") or "standard"
+    if profile not in EXECUTION_PROFILES:
+        raise SystemExit(
+            f"未知 execution_profile: {profile!r}（可选: {', '.join(EXECUTION_PROFILES)}）"
+        )
+    return str(profile)
+
+
 def host_set(vars_map: dict[str, str]) -> bool:
     host = (vars_map.get("HOST") or "").strip()
     return bool(host) and "${HOST}" not in host
@@ -570,6 +586,7 @@ def expand_run_together(
     skipped: list[dict[str, str]],
     max_i: int,
     automation_filter: str = "automated",
+    bypass_risk: bool = False,
 ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
     groups = run_together_groups(meta)
     if not groups:
@@ -607,7 +624,7 @@ def expand_run_together(
             except SystemExit:
                 skipped.append({"id": cid, "reason": f"未知风险 {risk}"})
                 continue
-            if r_i > max_i:
+            if not bypass_risk and r_i > max_i:
                 skipped.append({"id": cid, "reason": f"risk={risk} 超过上限"})
                 continue
             members.append(public_case(case))
@@ -770,9 +787,11 @@ def mobile_preflight(
 def cmd_select(args: argparse.Namespace, root: Path) -> int:
     meta = load_catalog_meta(root)
     cases = load_all_cases(root)
+    execution_profile = resolve_execution_profile(meta, args.execution_profile)
+    bypass_risk = execution_profile == "isolated-full"
     max_risk = args.max_risk or meta.get("default_max_risk") or "readonly"
     max_i = risk_index(meta, max_risk)
-    automation_filter = args.automation or "automated"
+    automation_filter = args.automation or ("any" if bypass_risk else "automated")
     wanted = set(parse_ids(args.ids or []))
     selected: list[dict[str, Any]] = []
     skipped: list[dict[str, str]] = []
@@ -802,9 +821,15 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
                 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":
+            # A mixed catalog keeps legacy cases runnable. Standard keeps the
+            # historical client-only implicit default; isolated-full means
+            # exactly what it says and includes every active test type.
+            if (
+                not bypass_risk
+                and not args.test_type
+                and not args.suite
+                and test_type(case) == "integration"
+            ):
                 continue
             if args.driver and case.get("driver") != args.driver:
                 continue
@@ -821,7 +846,7 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
         except SystemExit:
             skipped.append({"id": cid, "reason": f"未知风险 {risk}"})
             continue
-        if r_i > max_i:
+        if not bypass_risk and r_i > max_i:
             skipped.append({"id": cid, "reason": f"risk={risk} 超过上限 {max_risk}"})
             continue
         selected.append(public_case(case))
@@ -833,7 +858,15 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
         )
         return 2
 
-    selected, together = expand_run_together(meta, cases, selected, skipped, max_i, automation_filter)
+    selected, together = expand_run_together(
+        meta,
+        cases,
+        selected,
+        skipped,
+        max_i,
+        automation_filter,
+        bypass_risk=bypass_risk,
+    )
 
     # The historical key is kept in the JSON, but the gate now validates both
     # test types before any caller can start a phase.
@@ -867,6 +900,8 @@ def cmd_select(args: argparse.Namespace, root: Path) -> int:
         json.dumps(
             {
                 "max_risk": max_risk,
+                "execution_profile": execution_profile,
+                "risk_filter": "disabled" if bypass_risk else "enforced",
                 "automation": automation_filter,
                 "count": len(selected),
                 "vars": vars_map,
@@ -947,6 +982,12 @@ def validate(root: Path) -> list[str]:
     default_max = meta.get("default_max_risk")
     if default_max not in risks:
         errors.append(f"default_max_risk={default_max!r} 不在 risks 中")
+    profile = meta.get("execution_profile", "standard")
+    if profile not in EXECUTION_PROFILES:
+        errors.append(
+            f"execution_profile={profile!r} 非法，只能是 {', '.join(EXECUTION_PROFILES)}"
+        )
+    isolated_full = profile == "isolated-full"
     gates = meta.get("gates") or {}
     if gates and not isinstance(gates, dict):
         errors.append("catalog.gates 必须是 mapping")
@@ -1073,8 +1114,17 @@ def validate(root: Path) -> list[str]:
             if case.get("risk") in (None, "readonly"):
                 errors.append(f"{where}: risk=readonly 不要填 gate")
         risk = case.get("risk")
-        if risk and risk != "readonly" and risk not in authorize and not gate:
+        if risk and risk != "readonly" and risk not in authorize and not gate and not isolated_full:
             errors.append(f"{where}: {risk} 需要 authorize.{risk} 或 gate")
+        if isolated_full and case.get("lifecycle", "active") == "active" and risk != "readonly":
+            cleanup = case.get("cleanup")
+            cleanup_ok = isinstance(cleanup, dict) and (
+                bool(cleanup.get("steps"))
+                or bool(str(cleanup.get("run") or "").strip())
+                or (cleanup.get("required") is False and bool(str(cleanup.get("reason") or "").strip()))
+            )
+            if not cleanup_ok:
+                errors.append(f"{where}: isolated-full 的写入型用例必须声明 cleanup/rollback steps/run")
         automation = str(case.get("automation") or "automated")
         if automation not in AUTOMATIONS:
             errors.append(f"{where}: 非法 automation={case.get('automation')}")
@@ -1206,7 +1256,19 @@ def cmd_next_id(root: Path, category: str) -> int:
     return 0
 
 
-def authorized(vars_map: dict[str, str], meta: dict[str, Any], case: dict[str, Any]) -> bool:
+def authorized(
+    vars_map: dict[str, str],
+    meta: dict[str, Any],
+    case: dict[str, Any],
+    *,
+    execution_profile: str | None = None,
+) -> bool:
+    profile = resolve_execution_profile(meta, execution_profile)
+    if profile == "isolated-full":
+        # AUTHORIZE_* and gate variables are intentionally not run-time
+        # switches for an isolated resettable environment.  Capability and
+        # cleanup preflight are the safety boundary for this profile.
+        return True
     risk = str(case.get("risk") or "readonly")
     risks = list(meta.get("risks") or ["readonly"])
     lowest = risks[0] if risks else "readonly"
@@ -1226,6 +1288,7 @@ def authorized(vars_map: dict[str, str], meta: dict[str, Any], case: dict[str, A
 
 def cmd_run_script(root: Path, case_id: str) -> int:
     meta = load_catalog_meta(root)
+    execution_profile = resolve_execution_profile(meta)
     vars_map = resolve_vars(root, meta)
     case = next((c for c in load_all_cases(root) if c.get("id") == case_id), None)
     if not case:
@@ -1263,10 +1326,10 @@ def cmd_run_script(root: Path, case_id: str) -> int:
         )
         return 2
     risk = str(case.get("risk") or "readonly")
-    if not authorized(vars_map, meta, case):
+    if not authorized(vars_map, meta, case, execution_profile=execution_profile):
         print(
             json.dumps(
-                {"id": public_id, "result": "skipped", "evidence": f"风险 {risk} 未授权"},
+                {"id": public_id, "result": "skipped", "evidence": f"风险 {risk} 未授权", "execution_profile": execution_profile},
                 ensure_ascii=False,
             )
         )
@@ -1315,6 +1378,7 @@ def cmd_run_script(root: Path, case_id: str) -> int:
             {
                 "id": public_id,
                 "result": "executed",
+                "execution_profile": execution_profile,
                 "exit": proc.returncode,
                 "evidence": evidence,
             },
@@ -1345,6 +1409,12 @@ def build_parser() -> argparse.ArgumentParser:
         help="按顶层 test_type 筛选；integration 必须显式选择",
     )
     p_sel.add_argument("--max-risk")
+    p_sel.add_argument(
+        "--execution-profile",
+        choices=EXECUTION_PROFILES,
+        default=None,
+        help="执行策略；isolated-full 在隔离可重置环境中不按 risk/authorize 筛选",
+    )
     p_sel.add_argument("--driver", choices=DRIVERS)
     p_sel.add_argument(
         "--automation",
diff --git a/skills/tester/scripts/failure_feedback.py b/skills/tester/scripts/failure_feedback.py
new file mode 100644
index 0000000..1b1c0f7
--- /dev/null
+++ b/skills/tester/scripts/failure_feedback.py
@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+"""Build structured tester issues and the development feedback route.
+
+This module only writes the run contract.  Updating a remote PR/issue or
+notifying a dispatcher remains an explicit caller-side integration step; no
+network mutation is hidden in the tester.
+"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from datetime import datetime, timezone
+from typing import Any, Iterable, Mapping
+
+
+ISSUE_TYPES = ("product", "test")
+TRIGGERS = ("pr", "requirement", "bug")
+ISSUE_RESULTS = frozenset({"fail", "environment-fail"})
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def issue_type_for(result: str) -> str:
+    if result == "fail":
+        return "product"
+    if result == "environment-fail":
+        return "test"
+    raise ValueError(f"只有 fail/environment-fail 可以生成 issue: {result}")
+
+
+def _value(row: Mapping[str, Any], key: str, defaults: Mapping[str, Any]) -> Any:
+    value = row.get(key)
+    return value if value not in (None, "") else defaults.get(key, "")
+
+
+def build_issue(
+    row: Mapping[str, Any],
+    *,
+    defaults: Mapping[str, Any] | None = None,
+    issue_type: str | None = None,
+) -> dict[str, Any]:
+    """Create the stable issue shape required by the run report."""
+    defaults = defaults or {}
+    result = str(row.get("result") or "")
+    if result not in ISSUE_RESULTS:
+        raise ValueError(f"不能为 {result!r} 结果创建 defect: {row.get('id')}")
+    selected_type = issue_type or row.get("issue_type") or issue_type_for(result)
+    if selected_type not in ISSUE_TYPES:
+        raise ValueError(f"issue_type 必须是 product/test: {selected_type}")
+    testcase = str(row.get("testcase") or row.get("id") or "")
+    if not testcase:
+        raise ValueError("issue 缺少 testcase")
+    return {
+        "id": str(row.get("issue_id") or f"TEST-{testcase}-{result}"),
+        "testcase": testcase,
+        "issue_type": selected_type,
+        "result": result,
+        "phase": _value(row, "phase", defaults),
+        "expected": _value(row, "expected", defaults),
+        "observed": _value(row, "observed", defaults),
+        "artifact_hash": _value(row, "artifact_hash", defaults),
+        "environment": _value(row, "environment", defaults),
+        "evidence_path": _value(row, "evidence_path", defaults),
+        "reproduction": _value(row, "reproduction", defaults),
+        "next_step": _value(row, "next_step", defaults),
+        "created_at": str(row.get("created_at") or _now()),
+    }
+
+
+def collect_issues(
+    results: Iterable[Mapping[str, Any]],
+    *,
+    defaults: Mapping[str, Any] | None = None,
+) -> list[dict[str, Any]]:
+    issues: list[dict[str, Any]] = []
+    for row in results:
+        if not isinstance(row, Mapping) or str(row.get("result") or "") not in ISSUE_RESULTS:
+            continue
+        explicit = row.get("issue")
+        source = dict(row)
+        if isinstance(explicit, Mapping):
+            source.update(explicit)
+        issues.append(build_issue(source, defaults=defaults))
+    return issues
+
+
+def feedback_plan(
+    trigger: str,
+    *,
+    target: str = "",
+    related_issue: str = "",
+    dispatcher: str = "dispatcher",
+) -> dict[str, Any]:
+    """Describe the required handoff without performing remote mutations."""
+    if trigger not in TRIGGERS:
+        raise ValueError(f"trigger 必须是 {', '.join(TRIGGERS)}")
+    if trigger == "pr":
+        actions = [
+            {"action": "update_pr", "target": target, "status": "required"},
+            {"action": "update_related_issue", "target": related_issue, "status": "required"},
+            {"action": "notify_dispatcher", "target": dispatcher, "status": "required"},
+            {"action": "reassign_development", "target": dispatcher, "status": "pending"},
+        ]
+        route = "pr-update-issue-dispatcher-reassign"
+    else:
+        actions = [
+            {"action": "return_to_development", "target": target or trigger, "status": "required"},
+        ]
+        route = "development"
+    return {
+        "trigger": trigger,
+        "route": route,
+        "target": target,
+        "related_issue": related_issue,
+        "actions": actions,
+        "remote_mutation": "caller-required",
+        "created_at": _now(),
+    }
+
+
+def attach_feedback(
+    run: Mapping[str, Any],
+    *,
+    trigger: str | None = None,
+    target: str = "",
+    related_issue: str = "",
+    dispatcher: str = "dispatcher",
+    defaults: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Return a new run mapping with results, defects and routing preserved."""
+    output = deepcopy(dict(run))
+    results = output.get("results")
+    if not isinstance(results, list):
+        raise ValueError("run 缺少 results 列表")
+    issues = collect_issues(results, defaults=defaults)
+    output["defects"] = issues
+    # Compatibility name for consumers that call the same structure
+    # ``test-issues``.  It is deliberately a copy, not a second source of truth.
+    output["test_issues"] = deepcopy(issues)
+    if trigger:
+        output["feedback"] = feedback_plan(
+            trigger,
+            target=target,
+            related_issue=related_issue,
+            dispatcher=dispatcher,
+        )
+    return output
+
diff --git a/skills/tester/scripts/generate_run_csv.py b/skills/tester/scripts/generate_run_csv.py
index e7272d3..6b6a554 100644
--- a/skills/tester/scripts/generate_run_csv.py
+++ b/skills/tester/scripts/generate_run_csv.py
@@ -40,6 +40,16 @@ FIELDS = (
     "risk",
     "driver",
     "maps_to",
+    "blocked_by",
+    "issue_id",
+    "issue_type",
+    "expected",
+    "observed",
+    "artifact_hash",
+    "environment",
+    "evidence_path",
+    "reproduction",
+    "next_step",
 )
 
 
@@ -112,6 +122,23 @@ def case_index(root: Path) -> dict[str, dict[str, Any]]:
     return out
 
 
+def issue_index(run: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    """Index canonical defects and compatibility test_issues by testcase."""
+    indexed: dict[str, dict[str, Any]] = {}
+    raw = run.get("defects")
+    if not isinstance(raw, list):
+        raw = run.get("test_issues")
+    if not isinstance(raw, list):
+        return indexed
+    for issue in raw:
+        if not isinstance(issue, dict):
+            continue
+        testcase = str(issue.get("testcase") or issue.get("id") or "")
+        if testcase:
+            indexed.setdefault(testcase, issue)
+    return indexed
+
+
 def export_run_csv(
     root: Path,
     run_path: Path | None = None,
@@ -123,6 +150,7 @@ def export_run_csv(
     path = path.resolve()
     run = load_run(path)
     cases = case_index(root)
+    issues = issue_index(run)
     dest = out or path.with_suffix(".csv")
     dest.parent.mkdir(parents=True, exist_ok=True)
     with dest.open("w", encoding="utf-8-sig", newline="") as handle:
@@ -134,6 +162,9 @@ def export_run_csv(
             cid = str(item["id"])
             result = str(item.get("result") or "")
             case = cases.get(cid) or {}
+            issue = dict(issues.get(cid, {}))
+            if isinstance(item.get("issue"), dict):
+                issue.update(item["issue"])
             client_suite_value = case.get("client_suite") or (
                 case.get("suite") if test_type(case) == "journey" else ""
             )
@@ -154,6 +185,16 @@ def export_run_csv(
                     "risk": case.get("risk") or "",
                     "driver": case.get("driver") or "",
                     "maps_to": case.get("maps_to") or "",
+                    "blocked_by": one_line(item.get("blocked_by")),
+                    "issue_id": issue.get("id") or item.get("issue_id") or "",
+                    "issue_type": issue.get("issue_type") or item.get("issue_type") or "",
+                    "expected": one_line(issue.get("expected") or item.get("expected")),
+                    "observed": one_line(issue.get("observed") or item.get("observed")),
+                    "artifact_hash": issue.get("artifact_hash") or item.get("artifact_hash") or "",
+                    "environment": one_line(issue.get("environment") or item.get("environment")),
+                    "evidence_path": one_line(issue.get("evidence_path") or item.get("evidence_path")),
+                    "reproduction": one_line(issue.get("reproduction") or item.get("reproduction")),
+                    "next_step": one_line(issue.get("next_step") or item.get("next_step")),
                 }
             )
     return dest
diff --git a/skills/tester/scripts/journey.py b/skills/tester/scripts/journey.py
index 10eca98..5060747 100644
--- a/skills/tester/scripts/journey.py
+++ b/skills/tester/scripts/journey.py
@@ -254,6 +254,7 @@ def propagate_results(
     skipped: Mapping[str, str] | None = None,
     continue_on_fail: bool = False,
     policies: Mapping[str, str] | None = None,
+    blocked_by: Mapping[str, Iterable[str]] | None = None,
 ) -> list[dict[str, str]]:
     """Build ordered result rows and mark cases after a blocking failure not-run.
 
@@ -263,15 +264,21 @@ def propagate_results(
     """
     skipped = skipped or {}
     policies = policies or {}
+    blocked_by = blocked_by or {}
     rows: list[dict[str, str]] = []
-    blocked = False
+    blocking_case = ""
     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": "前置用例失败，未轮到执行"})
+        if blocking_case:
+            rows.append({
+                "id": case_id,
+                "result": "not-run",
+                "notes": "前置用例失败，未轮到执行",
+                "blocked_by": ",".join(blocked_by.get(case_id) or (blocking_case,)),
+            })
             continue
         result = observed.get(case_id, "not-run")
         if result not in RESULTS:
@@ -279,5 +286,5 @@ def propagate_results(
         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
+            blocking_case = case_id
     return rows
diff --git a/skills/tester/scripts/mobile_bundle.py b/skills/tester/scripts/mobile_bundle.py
index 3b4c8a9..3f689fa 100755
--- a/skills/tester/scripts/mobile_bundle.py
+++ b/skills/tester/scripts/mobile_bundle.py
@@ -390,6 +390,27 @@ def write_run(root: Path, record: dict[str, Any]) -> Path:
     return path
 
 
+def all_active_result_rows(root: Path, *, blocked_by: str) -> list[dict[str, Any]]:
+    """Keep a testcase row for every active catalog entry when preflight blocks."""
+    try:
+        from generate_cases_csv import load_cases
+
+        cases = load_cases(root)
+    except (ImportError, OSError, ValueError, SystemExit):
+        cases = []
+    rows: list[dict[str, Any]] = []
+    for case in cases:
+        if case.get("lifecycle", "active") != "active":
+            continue
+        rows.append({
+            "id": str(case.get("id") or ""),
+            "result": "not-run",
+            "notes": "test-preflight 未通过，未创建移动 session",
+            "blocked_by": [blocked_by],
+        })
+    return rows
+
+
 def run_test_preflight(
     config_path: Path,
     *,
@@ -507,6 +528,7 @@ def execute(args: argparse.Namespace) -> int:
                     record["platforms"][platform].update({"status": "not-run", "reason": "全量 preflight 闸门被另一平台阻断"})
             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": "修复列出的平台环境后重跑；默认不会静默降级"})
+            record["results"] = all_active_result_rows(context.root, blocked_by="test-preflight")
         else:
             for platform, (result, node, identity) in prepared.items():
                 try:
@@ -527,7 +549,18 @@ def execute(args: argparse.Namespace) -> int:
         record["evidence"] = context.evidence
     except (EnvironmentFailure, ArtifactFailure) as exc:
         record.update({"status": "environment-fail", "failure_class": "automatic-repair-failed", "error": str(exc), "evidence": context.evidence})
+    if record.get("status") != "ready" and "results" not in record:
+        record["results"] = all_active_result_rows(context.root, blocked_by="test-preflight")
     path = write_run(context.root, record)
+    try:
+        from generate_run_csv import export_run_csv
+
+        record["csv"] = str(export_run_csv(context.root, run_path=path))
+    except (OSError, SystemExit, ValueError):
+        # The YAML run remains the source of truth when a partial/provisioning
+        # root has no catalog cases yet.
+        record["csv"] = ""
+    path.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
     output = {"status": record["status"], "run": str(path), "platforms": record["platforms"], "worktree": record.get("worktree")}
     if record.get("error"):
         output["error"] = record["error"]
diff --git a/skills/tester/scripts/record_issues.py b/skills/tester/scripts/record_issues.py
new file mode 100644
index 0000000..dcf262c
--- /dev/null
+++ b/skills/tester/scripts/record_issues.py
@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Append structured tester issue evidence to an existing run snapshot."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from failure_feedback import attach_feedback
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="将 tester 失败结构化回流到 run YAML/CSV")
+    parser.add_argument("--run", required=True, type=Path)
+    parser.add_argument("--trigger", choices=("pr", "requirement", "bug"))
+    parser.add_argument("--target", default="")
+    parser.add_argument("--related-issue", default="")
+    parser.add_argument("--dispatcher", default="dispatcher")
+    parser.add_argument("--testcase", default="", help="为一条失败结果补充 testcase issue 字段")
+    parser.add_argument("--issue-type", choices=("product", "test"))
+    parser.add_argument("--expected", default="")
+    parser.add_argument("--observed", default="")
+    parser.add_argument("--artifact-hash", default="")
+    parser.add_argument("--environment", default="")
+    parser.add_argument("--evidence-path", default="")
+    parser.add_argument("--reproduction", default="")
+    parser.add_argument("--next-step", default="")
+    args = parser.parse_args(argv)
+
+    data: dict[str, Any] = yaml.safe_load(args.run.read_text(encoding="utf-8")) or {}
+    if args.testcase:
+        rows = data.get("results")
+        if not isinstance(rows, list):
+            raise SystemExit("run 缺少 results 列表")
+        row = next((item for item in rows if isinstance(item, dict) and str(item.get("id")) == args.testcase), None)
+        if row is None:
+            raise SystemExit(f"找不到 testcase: {args.testcase}")
+        row["issue"] = {
+            key: value
+            for key, value in {
+                "issue_type": args.issue_type,
+                "expected": args.expected,
+                "observed": args.observed,
+                "artifact_hash": args.artifact_hash,
+                "environment": args.environment,
+                "evidence_path": args.evidence_path,
+                "reproduction": args.reproduction,
+                "next_step": args.next_step,
+            }.items()
+            if value
+        }
+    output = attach_feedback(
+        data,
+        trigger=args.trigger,
+        target=args.target,
+        related_issue=args.related_issue,
+        dispatcher=args.dispatcher,
+    )
+    args.run.write_text(yaml.safe_dump(output, allow_unicode=True, sort_keys=False), encoding="utf-8")
+    csv_path = ""
+    try:
+        from generate_run_csv import export_run_csv
+
+        csv_path = str(export_run_csv(args.run.parents[3], run_path=args.run))
+    except (ImportError, OSError, SystemExit, ValueError):
+        pass
+    print(f"updated {args.run} defects={len(output.get('defects') or [])} csv={csv_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+
diff --git a/skills/tester/scripts/summarize_run.py b/skills/tester/scripts/summarize_run.py
index 9ce1791..df7ed50 100644
--- a/skills/tester/scripts/summarize_run.py
+++ b/skills/tester/scripts/summarize_run.py
@@ -103,6 +103,27 @@ def load_yaml_results(path: Path) -> list[dict[str, Any]]:
         errors.append(f"YAML testcase id 重复: {_format_ids(set(duplicates))}")
     if errors:
         raise SummaryError("YAML 校验失败:\n- " + "\n- ".join(errors))
+    issues = data.get("defects", data.get("test_issues"))
+    if issues is not None:
+        if not isinstance(issues, list):
+            raise SummaryError("YAML defects/test_issues 必须是列表")
+        result_ids = set(ids)
+        issue_errors: list[str] = []
+        for index, issue in enumerate(issues, start=1):
+            if not isinstance(issue, dict):
+                issue_errors.append(f"YAML defects[{index}] 不是 mapping")
+                continue
+            testcase = str(issue.get("testcase") or "")
+            issue_type = str(issue.get("issue_type") or "")
+            if not testcase or testcase not in result_ids:
+                issue_errors.append(f"YAML defects[{index}] testcase 无效: {testcase or '(missing)'}")
+            if issue_type not in {"product", "test"}:
+                issue_errors.append(f"YAML defects[{index}] issue_type 无效: {issue_type or '(missing)'}")
+            for field in ("expected", "observed", "artifact_hash", "environment", "evidence_path", "reproduction", "next_step"):
+                if field not in issue:
+                    issue_errors.append(f"YAML defects[{index}] 缺少 {field}")
+        if issue_errors:
+            raise SummaryError("YAML defects 校验失败:\n- " + "\n- ".join(issue_errors))
     return results
 
 
diff --git a/skills/tester/scripts/two_tier.py b/skills/tester/scripts/two_tier.py
index 521bb30..c2c4ab6 100644
--- a/skills/tester/scripts/two_tier.py
+++ b/skills/tester/scripts/two_tier.py
@@ -12,6 +12,8 @@ from __future__ import annotations
 from collections.abc import Callable, Iterable, Mapping
 from typing import Any
 
+from failure_feedback import collect_issues
+
 RESULTS = ("pass", "fail", "environment-fail", "skipped", "not-run")
 PHASES = ("preflight", "integration", "journey")
 
@@ -29,10 +31,18 @@ def _case_id(case: Mapping[str, Any] | str) -> str:
     return str(case)
 
 
-def _not_run(cases: Iterable[Mapping[str, Any] | str], reason: str) -> list[dict[str, str]]:
+def _not_run(
+    cases: Iterable[Mapping[str, Any] | str],
+    reason: str,
+    *,
+    blocked_by: Iterable[str] = (),
+) -> list[dict[str, Any]]:
     rows: list[dict[str, str]] = []
+    blockers = [str(item) for item in blocked_by if str(item)]
     for case in cases:
-        row = {"id": _case_id(case), "result": "not-run", "notes": reason}
+        row: dict[str, Any] = {"id": _case_id(case), "result": "not-run", "notes": reason}
+        if blockers:
+            row["blocked_by"] = blockers
         if isinstance(case, Mapping):
             row["test_type"] = str(case.get("test_type") or "")
             row["suite"] = str(
@@ -45,6 +55,37 @@ def _not_run(cases: Iterable[Mapping[str, Any] | str], reason: str) -> list[dict
     return rows
 
 
+def _dependencies(
+    case: Mapping[str, Any] | str,
+    *,
+    default: Iterable[str] = (),
+) -> list[str]:
+    if not isinstance(case, Mapping):
+        return [str(item) for item in default]
+    declared = case.get("depends_on", case.get("dependencies"))
+    if declared is None:
+        return [str(item) for item in default]
+    if isinstance(declared, str):
+        return [declared]
+    if isinstance(declared, Iterable):
+        return [str(item) for item in declared if str(item)]
+    raise ValueError(f"depends_on 必须是字符串或列表: {_case_id(case)}")
+
+
+def _failed_dependencies(dependencies: Iterable[str], results: Mapping[str, Mapping[str, Any]]) -> list[str]:
+    return [
+        dependency
+        for dependency in dependencies
+        if str((results.get(dependency) or {}).get("result"))
+        in {"fail", "environment-fail", "not-run"}
+    ]
+
+
+def _issue_structures(results: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
+    issues = collect_issues(results)
+    return {"defects": issues, "test_issues": [dict(issue) for issue in issues]}
+
+
 def run_integration_first(
     *,
     preflight_status: str,
@@ -65,18 +106,29 @@ def run_integration_first(
     journey = list(journey_cases)
     if preflight_status != "ready":
         reason = f"test-preflight={preflight_status}，未创建客户端 session"
+        result_rows = _not_run(integration, reason, blocked_by=("test-preflight",)) + _not_run(journey, reason, blocked_by=("test-preflight",))
         return {
             "status": preflight_status,
             "phase": "preflight",
             "session_created": False,
-            "results": _not_run(integration, reason) + _not_run(journey, reason),
+            "results": result_rows,
+            **_issue_structures(result_rows),
         }
 
     results: list[dict[str, Any]] = []
-    integration_failed = False
-    for index, case in enumerate(integration):
-        if integration_failed:
-            results.extend(_not_run([case], "前置 integration 失败，未轮到执行"))
+    results_by_id: dict[str, dict[str, Any]] = {}
+    for case in integration:
+        case_id = _case_id(case)
+        dependencies = _dependencies(case)
+        blocked_by = _failed_dependencies(dependencies, results_by_id)
+        if blocked_by:
+            rows = _not_run(
+                [case],
+                "依赖的 integration 失败，未执行",
+                blocked_by=blocked_by,
+            )
+            results.extend(rows)
+            results_by_id[case_id] = rows[0]
             continue
         outcome = execute_integration(case)
         row = dict(outcome) if isinstance(outcome, Mapping) else {"result": outcome}
@@ -89,20 +141,26 @@ def run_integration_first(
             raise ValueError(f"未知 integration 结果: {row['id']}={result}")
         row["result"] = result
         results.append(row)
-        if result in ("fail", "environment-fail"):
-            integration_failed = True
+        results_by_id[case_id] = row
 
-    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
+    integration_ids = [_case_id(case) for case in integration]
+    has_runnable_journey = any(
+        not _failed_dependencies(_dependencies(case, default=integration_ids), results_by_id)
+        for case in journey
+    )
+    session = create_journey_session() if has_runnable_journey else None
     for case in journey:
+        dependencies = _dependencies(case, default=integration_ids)
+        blocked_by = _failed_dependencies(dependencies, results_by_id)
+        if blocked_by:
+            row = _not_run(
+                [case],
+                "依赖的用例失败，journey 未执行",
+                blocked_by=blocked_by,
+            )[0]
+            results.append(row)
+            results_by_id[_case_id(case)] = row
+            continue
         outcome = execute_journey(session, case)
         row = dict(outcome) if isinstance(outcome, Mapping) else {"result": outcome}
         row["id"] = _case_id(case)
@@ -114,12 +172,14 @@ def run_integration_first(
             raise ValueError(f"未知 journey 结果: {row['id']}={result}")
         row["result"] = result
         results.append(row)
+        results_by_id[_case_id(case)] = 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),
+        "session_created": has_runnable_journey,
         "results": results,
+        **_issue_structures(results),
     }
diff --git a/skills/tester/templates/catalog.yaml b/skills/tester/templates/catalog.yaml
index 864ac8c..b948041 100644
--- a/skills/tester/templates/catalog.yaml
+++ b/skills/tester/templates/catalog.yaml
@@ -22,6 +22,10 @@ risks:
   - mutating
   - destructive
 default_max_risk: readonly
+# standard enforces the default_max_risk/authorize policy. Writes-style isolated
+# resettable environments may set this to isolated-full; risk values remain on
+# cases and are still reported.
+execution_profile: standard
 client_suites:
   client-smoke:
     default: true
diff --git a/skills/tester/tests/test_two_tier.py b/skills/tester/tests/test_two_tier.py
index 4ebb6aa..ec02fe4 100644
--- a/skills/tester/tests/test_two_tier.py
+++ b/skills/tester/tests/test_two_tier.py
@@ -150,6 +150,86 @@ class TwoTierCatalogTests(unittest.TestCase):
             self.assertEqual(report_rows[0]["test_type"], "integration")
             self.assertEqual(report_rows[0]["suite"], "api-contract")
 
+    def test_isolated_full_keeps_risk_and_ignores_authorize_filter(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            write_mixed_catalog(root)
+            catalog_path = root / ".pouch/tester/catalog.yaml"
+            catalog = yaml.safe_load(catalog_path.read_text(encoding="utf-8"))
+            catalog["authorize"] = {"mutating": "AUTHORIZE_MUTATING"}
+            catalog["execution_profile"] = "isolated-full"
+            catalog_path.write_text(yaml.safe_dump(catalog, sort_keys=False), encoding="utf-8")
+            case_path = root / ".pouch/tester/cases/mixed.yaml"
+            cases = yaml.safe_load(case_path.read_text(encoding="utf-8"))
+            journey = cases["cases"][0]
+            journey.update({"risk": "mutating", "tag": "full", "client_suite": "client-regression"})
+            case_path.write_text(yaml.safe_dump(cases, sort_keys=False), encoding="utf-8")
+
+            selected = self.run_catalog(root, "select", "--suite", "client-regression")
+            self.assertEqual(selected.returncode, 0, selected.stderr)
+            payload = json.loads(selected.stdout)
+            self.assertEqual(payload["execution_profile"], "isolated-full")
+            self.assertEqual(payload["risk_filter"], "disabled")
+            self.assertEqual(payload["count"], 1)
+            self.assertEqual(payload["cases"][0]["risk"], "mutating")
+            self.assertEqual(payload["skipped"], [])
+
+            all_active = self.run_catalog(root, "select")
+            self.assertEqual(all_active.returncode, 0, all_active.stderr)
+            all_payload = json.loads(all_active.stdout)
+            self.assertEqual(all_payload["count"], 2)
+            self.assertEqual(all_payload["test_types"], {"journey": 1, "integration": 1})
+
+    def test_isolated_full_run_script_does_not_use_authorize_switch(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            tester = root / ".pouch" / "tester"
+            (tester / "cases").mkdir(parents=True)
+            (tester / "local.yaml").write_text("HOST: http://127.0.0.1\nAUTHORIZE_MUTATING: false\n", encoding="utf-8")
+            (tester / "catalog.yaml").write_text(
+                yaml.safe_dump(
+                    {
+                        "version": 2,
+                        "id_pattern": "^TC-[A-Z]+-[0-9]+$",
+                        "categories": [{"id": "SCRIPT"}],
+                        "risks": ["readonly", "mutating", "destructive"],
+                        "default_max_risk": "readonly",
+                        "execution_profile": "isolated-full",
+                        "authorize": {"mutating": "AUTHORIZE_MUTATING"},
+                    },
+                    sort_keys=False,
+                ),
+                encoding="utf-8",
+            )
+            (tester / "cases" / "script.yaml").write_text(
+                yaml.safe_dump(
+                    {
+                        "category": "SCRIPT",
+                        "cases": [{
+                            "id": "TC-SCRIPT-001",
+                            "title": "mutating script",
+                            "what": "run isolated script",
+                            "lifecycle": "active",
+                            "tag": "full",
+                            "risk": "mutating",
+                            "driver": "script",
+                            "how": {"run": "exit 0"},
+                            "cleanup": {"steps": ["reset"]},
+                            "expected": [{"kind": "script-exit", "value": 0}],
+                        }],
+                    },
+                    sort_keys=False,
+                ),
+                encoding="utf-8",
+            )
+            result = subprocess.run(
+                [sys.executable, str(CATALOG), "run-script", "--root", str(root), "--id", "TC-SCRIPT-001"],
+                text=True,
+                capture_output=True,
+            )
+            self.assertEqual(result.returncode, 0, result.stderr)
+            self.assertEqual(json.loads(result.stdout)["result"], "executed")
+
 
 class IntegrationFirstTests(unittest.TestCase):
     def test_preflight_block_does_not_create_session(self):
@@ -177,9 +257,38 @@ class IntegrationFirstTests(unittest.TestCase):
             execute_journey=lambda session, case: "pass",
         )
         self.assertEqual(result["status"], "fail")
-        self.assertEqual(calls, ["INT-1"])
+        self.assertEqual(calls, ["INT-1", "INT-2"])
         self.assertFalse(result["session_created"])
-        self.assertEqual([row["result"] for row in result["results"]], ["fail", "not-run", "not-run", "not-run"])
+        self.assertEqual([row["result"] for row in result["results"]], ["fail", "pass", "not-run", "not-run"])
+        self.assertEqual(result["results"][2]["blocked_by"], ["INT-1"])
+
+    def test_independent_integration_continues_and_dependency_is_blocked(self):
+        result = run_integration_first(
+            preflight_status="ready",
+            integration_cases=[
+                {"id": "INT-1"},
+                {"id": "INT-2"},
+                {"id": "INT-3", "depends_on": ["INT-1"]},
+            ],
+            journey_cases=[],
+            execute_integration=lambda case: "fail" if case["id"] == "INT-1" else "pass",
+            create_journey_session=lambda: object(),
+            execute_journey=lambda session, case: "pass",
+        )
+        self.assertEqual([row["result"] for row in result["results"]], ["fail", "pass", "not-run"])
+        self.assertEqual(result["results"][2]["blocked_by"], ["INT-1"])
+
+    def test_preflight_block_creates_every_result_row(self):
+        result = run_integration_first(
+            preflight_status="blocked",
+            integration_cases=[{"id": "INT-1"}],
+            journey_cases=[{"id": "J-1"}],
+            execute_integration=lambda case: "pass",
+            create_journey_session=lambda: object(),
+            execute_journey=lambda session, case: "pass",
+        )
+        self.assertEqual([row["result"] for row in result["results"]], ["not-run", "not-run"])
+        self.assertEqual(result["results"][0]["blocked_by"], ["test-preflight"])
 
     def test_service_unreachable_is_environment_failure(self):
         self.assertEqual(classify_integration(ok=False, service_reachable=False), "environment-fail")
diff --git a/tests/test_tester_run_records.py b/tests/test_tester_run_records.py
index 9d5de4c..3d863e6 100644
--- a/tests/test_tester_run_records.py
+++ b/tests/test_tester_run_records.py
@@ -12,6 +12,7 @@ REPO_ROOT = Path(__file__).resolve().parents[1]
 CASES = REPO_ROOT / "skills" / "tester" / "scripts" / "generate_cases_csv.py"
 RUNS = REPO_ROOT / "skills" / "tester" / "scripts" / "generate_run_csv.py"
 sys.path.insert(0, str(RUNS.parent))
+from failure_feedback import attach_feedback
 
 
 def load_module(name: str, path: Path):
@@ -29,6 +30,7 @@ class TesterRunRecordTests(unittest.TestCase):
             root = Path(temp)
             cases_dir = root / ".pouch" / "tester" / "cases"
             cases_dir.mkdir(parents=True)
+            (root / ".pouch" / "tester" / "runs").mkdir()
             (root / ".pouch" / "tester" / "catalog.yaml").write_text(
                 "categories:\n  - id: smoke\n    title: Smoke\n",
                 encoding="utf-8",
@@ -51,6 +53,63 @@ class TesterRunRecordTests(unittest.TestCase):
         self.assertNotEqual(first, second)
         self.assertRegex(first, r"^\d{8}T\d{6}Z-smoke-[0-9a-f]{6}$")
 
+    def test_run_csv_preserves_all_results_and_structured_issue_fields(self) -> None:
+        runs = load_module("tester_run_csv_issues", RUNS)
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            cases_dir = root / ".pouch" / "tester" / "cases"
+            cases_dir.mkdir(parents=True)
+            (root / ".pouch" / "tester" / "runs").mkdir()
+            (root / ".pouch" / "tester" / "catalog.yaml").write_text(
+                "categories:\n  - id: smoke\n    title: Smoke\n", encoding="utf-8"
+            )
+            (cases_dir / "smoke.yaml").write_text(
+                "category: smoke\ncases:\n"
+                "  - id: TC-1\n    title: One\n    what: one\n    lifecycle: active\n"
+                "    tag: smoke\n    risk: readonly\n    how: {steps: [open]}\n"
+                "    expected: [{kind: visible-text, value: ready}]\n"
+                "  - id: TC-2\n    title: Two\n    what: two\n    lifecycle: active\n"
+                "    tag: smoke\n    risk: readonly\n    how: {steps: [open]}\n"
+                "    expected: [{kind: visible-text, value: ready}]\n",
+                encoding="utf-8",
+            )
+            run = attach_feedback(
+                {
+                    "runId": "issues",
+                    "results": [
+                        {
+                            "id": "TC-1",
+                            "result": "fail",
+                            "issue": {
+                                "expected": "ready",
+                                "observed": "blank",
+                                "artifact_hash": "sha256:abc",
+                                "environment": "Android tester",
+                                "evidence_path": "evidence/1.png",
+                                "reproduction": "open",
+                                "next_step": "targeted rerun",
+                            },
+                        },
+                        {"id": "TC-2", "result": "not-run", "blocked_by": ["TC-1"]},
+                    ],
+                },
+                trigger="pr",
+                target="PR-1",
+                related_issue="ISSUE-1",
+            )
+            run_path = root / ".pouch" / "tester" / "runs" / "issues.yaml"
+            import yaml
+            run_path.write_text(yaml.safe_dump(run, sort_keys=False), encoding="utf-8")
+            out = runs.export_run_csv(root, run_path=run_path)
+            with out.open(encoding="utf-8-sig", newline="") as handle:
+                rows = list(csv.DictReader(handle))
+            self.assertEqual([row["id"] for row in rows], ["TC-1", "TC-2"])
+            self.assertEqual(rows[0]["issue_type"], "product")
+            self.assertEqual(rows[0]["expected"], "ready")
+            self.assertEqual(rows[0]["artifact_hash"], "sha256:abc")
+            self.assertEqual(rows[1]["blocked_by"], "TC-1")
+            self.assertEqual(run["feedback"]["route"], "pr-update-issue-dispatcher-reassign")
+
 
 if __name__ == "__main__":
     unittest.main()