fix(tester): use remote artifact path for installs

skills/test-preflight/README.md

@@ -2,8 +2,9 @@
 
 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, and fixture/cleanup contracts.
+This project assumes mobile album/photo/IME capabilities; journey cases exercise them
+and preflight does not gate them or read authorization environment variables.
 
 Run the self-tests with:
 

skills/test-preflight/SKILL.md

@@ -3,7 +3,8 @@ 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.
+  contracts. Use before tester runs; it is not a deployment,
+  build, or business-data skill.
 ---
 
 # test-preflight
@@ -47,5 +48,11 @@ Overall status is `ready`, `blocked`, or `environment-fail`:
 - `ready` means every recorded check passed. A ready preflight permits the
   caller to continue with `integration`; this skill does not run test cases.
 
+This project assumes the client has the required album picker, photo permission, and Chinese
+IME capabilities. Preflight does not probe or gate these capabilities, and does not read
+`AUTHORIZE_ALBUM` or `IME_PINYIN`. Their real behavior is exercised by journey cases; a failure
+there is a product test failure, not an environment block.
+
 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]`.

skills/test-preflight/references/schema.md

@@ -18,6 +18,9 @@ fixtures:
 cleanup:
   required: true
   steps: [delete-run-data]
+# Optional project metadata. It is not a preflight gate.
+capabilities:
+  assumed: [album_picker, photo_permission, ime_pinyin]
 matrix:
   - platform: android
     tester:
@@ -48,6 +51,13 @@ 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.
 
+## Client capabilities
+
+The project assumes the mobile test nodes provide the album picker, photo permission,
+test image, and Chinese IME. These are not preflight declarations or gates. Journey
+cases exercise the real UI; failures are recorded as product failures. `AUTHORIZE_ALBUM`
+and `IME_PINYIN` are not read.
+
 ## Tester-node contract
 
 `tester.runner.kind: local` executes platform and network probes on the current

skills/test-preflight/scripts/preflight.py

@@ -391,7 +391,8 @@ def check_network(evidence: Evidence, config: Mapping[str, Any], row: Mapping[st
                 statuses.append(status)
                 evidence.add(f"{platform_name}-{index}-network-{probe_index}-{detail['name']}", platform=platform_name, node=node, target=url, observed=detail["observed"], expected=detail["expected"], status=status, remediation=detail.get("remediation", ""))
         elif isinstance(runner, dict) and runner.get("kind") == "contract":
-            status, observed, result = runner_call(runner, {"operation": "network_probe", "platform": platform_name, "node": node, "url": url, "probe": probe})
+            binding = text(tester.get("binding")) if isinstance(tester, dict) else ""
+            status, observed, result = runner_call(runner, {"operation": "network_probe", "platform": platform_name, "node": node, "binding": binding, "url": url, "probe": probe})
             result = result or {"status": status, "observed": observed}
             complete = require_runner_checks(evidence, result, prefix=f"{platform_name}-{index}-network-{probe_index}", platform_name=platform_name, node=node, required={"dns", "tcp", "http"})
             network_status = add_runner_checks(evidence, result, prefix=f"{platform_name}-{index}-network-{probe_index}", platform_name=platform_name, node=node, default_expected="DNS/TCP/HTTP reachable from actual tester node")
@@ -479,6 +480,9 @@ 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)
+        # This isolated project assumes the client capabilities used by its
+        # journeys. Their real behavior is a product assertion, not a preflight
+        # environment gate.
         row_ok = artifact_ok and fixture_ok and platform_ok and network_ok
         if mode == "prepare":
             artifact = row.get("artifact") or {}

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:

skills/test-preflight/tests/test_preflight.py

@@ -140,6 +140,21 @@ 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_assumed_and_not_a_preflight_gate(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, 0)
+        self.assertEqual(report["status"], "ready")
+        self.assertFalse(any("capability-" 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)
@@ -203,6 +218,7 @@ class PreflightTests(unittest.TestCase):
         self.assertEqual(result.overall(), "ready")
         self.assertNotIn("install", calls)
         self.assertNotIn("start", calls)
+        self.assertNotIn("capability_probe", calls)
 
 
 if __name__ == "__main__":

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 原值仍保留。设备、产物、服务和 cleanup/rollback
+  由 test-preflight 保证；客户端相册、权限和中文输入能力默认具备，由 journey 用例验证。
 
 ## 使用示例
 
@@ -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，使用显式路径汇总：

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 检查；相册、照片权限/测试图片和中文拼音能力在本项目中默认具备，由 journey 用例真实验证。
 - `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 值。
+设备、产物、服务或 session 能力缺失时 preflight 必须明确 blocked/environment-fail，并为所有选中的 testcase 生成
+`not-run` 结果行；本项目默认具备相册、照片权限、测试图片或中文拼音 composition/candidate
+能力，不在 preflight 中校验或静默变成 `skipped`，由 journey 真实验证并记录产品失败。独立用例继续执行，依赖失败的用例写 `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 只生成结构化回流计划，不在本地偷偷修改远端。

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 的设备、服务与 cleanup/rollback 契约保证，原始 risk 仍必须报告；客户端能力由 journey 用例验证。
 
 ### 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`
 

skills/tester/references/execution.md

@@ -25,6 +25,16 @@ bindings:
 上执行测试。`tester.test` 只能用于明确的单平台或非移动测试；同一轮包含 Android
 和 iOS 时必须分别解析两个 tester 绑定。
 
+移动 bundle 入口必须把平台绑定显式传入：
+
+```bash
+--binding android=tester.android --binding ios=tester.ios
+```
+
+入口会校验 preflight matrix 的每个平台 binding、node 和 runner。远程 iOS 必须是
+`contract` runner；多平台运行不会读取旧的单一 `local.yaml.agent` 来决定执行节点，
+也不会在协调节点直接执行 Android/iOS 平台命令。
+
 `locations` 和 `profiles` 的定义见共享 `agent-runtime`。tester adapter 只把它 materialize
 到本轮兼容配置；每轮 terminal 仍然新建并在结束时关闭，不能因为 binding 是 persistent
 就复用测试 terminal。

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 契约阻断不安全环境。

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` 等人工开关；相册、照片权限/测试图片和中文拼音能力默认具备，由 journey 用例真实验证，失败记为产品失败。
+- standard 中授权/假设不成立是**主动不跑**（`skipped`）；isolated-full 不因客户端能力授权变量停止，设备、产物、服务或清理前置失败时才为全部 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，不要手改：

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 用例；设备、服务和 cleanup/rollback 契约仍由 preflight 检查，客户端能力由 journey 真实验证 |
 | `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
+的设备、产物、服务和网络检查；相册、权限和输入法能力默认具备，交由 journey 用例验证。
+
 `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 只记录计划，不隐式调用远端写操作。

skills/tester/references/test-bundle.md

@@ -7,11 +7,15 @@ CI release、builder 绑定或本机旧文件猜测产物。
 <skill-dir>/tester run \
   --artifact android=<path-or-url> \
   --artifact ios=<path-or-url> \
-  --platform android,ios
+  --platform android,ios \
+  --binding android=tester.android \
+  --binding ios=tester.ios
 ```
 
-URL 就下载，本地文件就直接用。每个平台都在自己的 `tester.<platform>` 节点校验身份并安装，
-随后由现有 tester catalog 驱动真实移动 session。默认先完成 Android+iOS 双平台
+URL 就下载，本地文件就直接用。每个平台都通过自己的 `tester.<platform>` binding
+校验、探测和安装；`agent-runtime` 负责把 Android 请求送到 Android 节点、把 iOS
+请求送到 Mac 节点。旧的单一 `local.yaml.agent` 只能作为兼容配置，不能作为双平台
+运行目标。随后由现有 tester catalog 驱动真实移动 session。默认先完成 Android+iOS 双平台
 preflight；任一平台无法准备时整轮停止。只有显式加 `--partial` 才允许运行可用平台。
 
 每轮另外创建隔离 Git worktree，不切用户当前分支。`--commit` 缺省为当前 HEAD。

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",

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
+

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

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

skills/tester/scripts/mobile_bundle.py

@@ -54,6 +54,7 @@ class CommandResult:
 @dataclass
 class RuntimeContext:
     root: Path
+    bindings: dict[str, str] = field(default_factory=dict)
     retries: dict[str, int] = field(default_factory=lambda: {"download": 3, "start": 3, "install": 3})
     evidence: list[dict[str, Any]] = field(default_factory=list)
 
@@ -126,6 +127,120 @@ def parse_artifacts(values: list[str]) -> dict[str, str]:
     return pairs
 
 
+def parse_bindings(values: list[str], requested: list[str]) -> dict[str, str]:
+    """Resolve one tester binding per requested platform.
+
+    A single legacy ``agent`` mapping cannot safely represent a multi-platform
+    run.  Keep the platform distinction in the command contract all the way to
+    agent-runtime instead of treating binding as report metadata.
+    """
+    pairs: dict[str, str] = {}
+    for raw in values:
+        platform, sep, value = raw.partition("=")
+        platform = platform.strip().lower()
+        value = value.strip()
+        if not sep or platform not in PLATFORMS or not value or not re.fullmatch(r"[a-z][a-z0-9-]*(?:\.[a-z][a-z0-9-]*)+", value):
+            raise ArtifactFailure(f"--binding 格式必须是 android=tester.<name> 或 ios=tester.<name>: {raw}")
+        if platform in pairs:
+            raise ArtifactFailure(f"--binding 重复平台: {platform}")
+        pairs[platform] = value
+    resolved = {platform: pairs.get(platform, f"tester.{platform}") for platform in requested}
+    if len(set(resolved.values())) != len(resolved) and len(resolved) > 1:
+        raise ArtifactFailure("多平台运行不能复用同一个 tester binding；请分别配置 tester.android/tester.ios")
+    return resolved
+
+
+def load_agent_runtime():
+    candidates = [
+        Path.home() / ".codex" / "skills" / "agent-runtime" / "scripts" / "agent.py",
+        Path.home() / ".pouch" / "skills" / "agent-runtime" / "scripts" / "agent.py",
+        Path(__file__).resolve().parents[2] / "agent-runtime" / "scripts" / "agent.py",
+    ]
+    runtime_path = next((item for item in candidates if item.is_file()), None)
+    if runtime_path is None:
+        raise EnvironmentFailure("找不到共享 agent-runtime: " + ", ".join(str(item) for item in candidates))
+    spec = importlib.util.spec_from_file_location("pouch_agent_runtime_for_mobile_bundle", runtime_path)
+    if spec is None or spec.loader is None:
+        raise EnvironmentFailure(f"无法加载共享 agent-runtime: {runtime_path}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def binding_location(root: Path, binding_id: str) -> dict[str, Any]:
+    runtime = load_agent_runtime()
+    try:
+        value = runtime.binding(runtime.load_config(root), binding_id)
+    except (ValueError, OSError) as exc:
+        raise EnvironmentFailure(f"无法解析 {binding_id}: {exc}") from exc
+    if value.get("adapter") != "tester":
+        raise EnvironmentFailure(f"{binding_id} 不是 tester binding")
+    return value
+
+
+def agent_runner(root: Path, request: dict[str, Any]) -> dict[str, Any]:
+    """Dispatch a platform operation through the resolved tester binding."""
+    runtime = load_agent_runtime()
+    return runtime.execute_runner_request(root, request)
+
+
+def platform_node(root: Path, binding_id: str) -> str:
+    location = binding_location(root, binding_id)["location"]
+    return "local" if location.get("environment") == "local" else str(location.get("environment"))
+
+
+def validate_preflight_matrix(root: Path, config_path: Path, bindings: dict[str, str], requested: list[str]) -> None:
+    """Fail closed when a matrix routes a platform to the wrong node/runner."""
+    try:
+        import yaml
+
+        raw = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
+    except (OSError, ValueError, ImportError) as exc:
+        raise EnvironmentFailure(f"无法读取 preflight matrix: {config_path}: {exc}") from exc
+    rows = raw.get("matrix") if isinstance(raw, dict) else None
+    if not isinstance(rows, list):
+        raise EnvironmentFailure("preflight matrix 缺少 matrix；不能推断平台执行节点")
+    by_platform = {str(row.get("platform", "")).lower(): row for row in rows if isinstance(row, dict)}
+    for platform in requested:
+        expected_binding = bindings[platform]
+        row = by_platform.get(platform)
+        if not row:
+            raise EnvironmentFailure(f"preflight matrix 缺少 {platform} 行；不能创建该平台 session")
+        tester = row.get("tester") if isinstance(row, dict) else None
+        actual_binding = tester.get("binding") if isinstance(tester, dict) else None
+        if actual_binding != expected_binding:
+            raise EnvironmentFailure(f"{platform} preflight binding={actual_binding!r}，期望 {expected_binding!r}")
+        expected_node = platform_node(root, expected_binding)
+        actual_node = tester.get("node") if isinstance(tester, dict) else None
+        if expected_node == "local" and actual_node not in {"local", "localhost", "127.0.0.1"}:
+            raise EnvironmentFailure(f"{platform} binding 是 local，但 preflight node={actual_node!r}")
+        runner = tester.get("runner") if isinstance(tester, dict) else None
+        if expected_node != "local" and (not isinstance(runner, dict) or runner.get("kind") != "contract"):
+            raise EnvironmentFailure(f"{platform} 绑定到 {expected_node}，必须使用 contract runner；禁止在协调节点执行")
+
+
+def preflight_remote_artifacts(config_path: Path, requested: list[str]) -> dict[str, str]:
+    """Return caller-declared artifact paths readable by remote tester nodes."""
+    try:
+        import yaml
+
+        raw = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
+    except (OSError, ValueError, ImportError) as exc:
+        raise EnvironmentFailure(f"无法读取 preflight matrix: {config_path}: {exc}") from exc
+    rows = raw.get("matrix") if isinstance(raw, dict) else None
+    if not isinstance(rows, list):
+        raise EnvironmentFailure("preflight matrix 缺少 matrix；不能解析远程产物路径")
+    by_platform = {str(row.get("platform", "")).lower(): row for row in rows if isinstance(row, dict)}
+    paths: dict[str, str] = {}
+    for platform in requested:
+        row = by_platform.get(platform) or {}
+        artifact = row.get("artifact") if isinstance(row, dict) else None
+        remote_path = artifact.get("remote_path") if isinstance(artifact, dict) else None
+        if isinstance(remote_path, str) and remote_path.strip():
+            paths[platform] = remote_path.strip()
+    return paths
+
+
 def local_path(locator: str) -> Path | None:
     parsed = urllib.parse.urlsplit(locator)
     if parsed.scheme in ("http", "https"):
@@ -211,6 +326,10 @@ def android_identity(path: Path, context: RuntimeContext) -> str:
         result = run_command(command, context=context)
         if result.code == 0:
             match = re.search(r"(?:name|application-id)[=:]['\"]?([A-Za-z0-9_.]+)", result.output)
+            if not match:
+                # apkanalyzer manifest application-id prints the bare package
+                # name, while aapt/aapt2 print a package: name=... record.
+                match = re.fullmatch(r"\s*([A-Za-z0-9_.]+)\s*", result.output)
             if match:
                 return match.group(1)
     raise EnvironmentFailure("APK 身份读取失败；请检查 Android SDK build-tools")
@@ -390,6 +509,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,
     *,
@@ -419,18 +559,21 @@ def run_test_preflight(
 
 def execute(args: argparse.Namespace) -> int:
     requested = platforms(args.platform)
-    context = RuntimeContext(Path(args.root).resolve())
+    root = Path(args.root).resolve()
+    bindings = parse_bindings(args.bindings, requested)
+    context = RuntimeContext(root, bindings=bindings)
     record: dict[str, Any] = {
         "schema_version": 1,
         "started_at": datetime.now(timezone.utc).isoformat(),
         "platforms_requested": requested,
         "partial": bool(args.partial),
-        "binding": args.binding,
+        "bindings": bindings,
         "node": os.environ.get("TESTER_NODE", "local"),
         "terminal": {"handle": os.environ.get("ORCA_TERMINAL_HANDLE"), "created": os.environ.get("TESTER_TERMINAL_CREATED", "false").lower() == "true", "closed": os.environ.get("TESTER_TERMINAL_CLOSED", "unknown")},
         "product_test": {"status": "not-run", "failure_class": "automatic-repair-failed", "note": "准备完成后由 tester catalog 驱动真实移动 session；本工具不以 API 替代移动测试"},
         "platforms": {},
     }
+    remote_artifacts: dict[str, str] = {}
     try:
         locators = parse_artifacts(args.artifacts)
         missing = [platform for platform in requested if platform not in locators]
@@ -475,6 +618,8 @@ def execute(args: argparse.Namespace) -> int:
                     "error": "缺少 --preflight-config；不能跳过 test-preflight",
                 }
             else:
+                validate_preflight_matrix(context.root, preflight_config, bindings, requested)
+                remote_artifacts = preflight_remote_artifacts(preflight_config, requested)
                 preflight_result = run_test_preflight(
                     preflight_config,
                     mode=args.preflight_mode,
@@ -494,10 +639,28 @@ def execute(args: argparse.Namespace) -> int:
             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)
+                    binding_id = bindings[platform]
+                    node_name = platform_node(context.root, binding_id)
+                    probe = agent_runner(context.root, {
+                        "operation": "platform_probe",
+                        "platform": platform,
+                        "node": node_name,
+                        "binding": binding_id,
+                    })
+                    if probe.get("status") != "ready":
+                        raise EnvironmentFailure(str(probe.get("observed") or probe))
+                    host = os.environ.get("HOST", "").strip()
+                    network = agent_runner(context.root, {
+                        "operation": "network_probe",
+                        "platform": platform,
+                        "node": node_name,
+                        "binding": binding_id,
+                        "url": f"http://{host}/v1/me",
+                    }) if host else {"status": "blocked", "observed": "HOST 未设置"}
+                    if network.get("status") != "ready":
+                        raise EnvironmentFailure(str(network.get("observed") or network))
+                    result.update({"node_preflight": probe, "network_preflight": network, "status": "preflight-passed"})
+                    prepared[platform] = (result, None, identity)
                 except EnvironmentFailure as exc:
                     failures[platform] = str(exc)
                     result.update({"status": "environment-fail", "failure_class": "automatic-repair-failed", "error": str(exc)})
@@ -507,16 +670,51 @@ 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:
-                    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)
-                    )
+                    if args.preflight_mode == "prepare":
+                        # test-preflight already ran platform_probe, network_probe,
+                        # install and start through each matrix binding.
+                        installed = {"delegated": "test-preflight.prepare", "binding": bindings[platform]}
+                    else:
+                        binding_id = bindings[platform]
+                        node_name = platform_node(context.root, binding_id)
+                        probe = agent_runner(context.root, {
+                            "operation": "platform_probe",
+                            "platform": platform,
+                            "node": node_name,
+                            "binding": binding_id,
+                        })
+                        if probe.get("status") != "ready":
+                            raise EnvironmentFailure(str(probe.get("observed") or probe))
+                        host = os.environ.get("HOST", "").strip()
+                        if host:
+                            network = agent_runner(context.root, {
+                                "operation": "network_probe",
+                                "platform": platform,
+                                "node": node_name,
+                                "binding": binding_id,
+                                "url": f"http://{host}/v1/me",
+                            })
+                            if network.get("status") != "ready":
+                                raise EnvironmentFailure(str(network.get("observed") or network))
+                        artifact_path = result["artifact"]["path"]
+                        if node_name != "local":
+                            artifact_path = remote_artifacts.get(platform, "")
+                            if not artifact_path:
+                                raise EnvironmentFailure(f"远程 {platform} 安装必须在 matrix.artifact.remote_path 声明节点可读产物")
+                        installed = agent_runner(context.root, {
+                            "operation": "install",
+                            "platform": platform,
+                            "node": node_name,
+                            "binding": binding_id,
+                            "artifact": artifact_path,
+                            "identity": identity,
+                        })
+                        if installed.get("status") != "ready":
+                            raise EnvironmentFailure(str(installed.get("observed") or installed))
                     result.update({"status": "ready", "install": installed, "failure_class": None})
                 except EnvironmentFailure as exc:
                     failures[platform] = str(exc)
@@ -527,7 +725,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"]
@@ -547,7 +756,14 @@ def parser() -> argparse.ArgumentParser:
     run.add_argument("--partial", action="store_true")
     run.add_argument("--root", default=".")
     run.add_argument("--worktree-root", default="")
-    run.add_argument("--binding", default="")
+    run.add_argument(
+        "--binding",
+        action="append",
+        dest="bindings",
+        default=[],
+        metavar="PLATFORM=TESTER_BINDING",
+        help="每个平台的 tester binding，可重复；默认 tester.<platform>",
+    )
     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 证据输出路径")

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

skills/tester/scripts/resolve_agent.py

@@ -180,9 +180,14 @@ def clean_config(config: dict[str, Any]) -> dict[str, Any]:
 def config_from_binding(root: Path, selector: str) -> dict[str, Any]:
     """Materialize a common agent-runtime tester binding for legacy local.yaml."""
 
-    runtime_path = Path(__file__).resolve().parents[3] / "skills" / "agent-runtime" / "scripts" / "agent.py"
-    if not runtime_path.is_file():
-        raise RuntimeError(f"找不到共享 agent-runtime: {runtime_path}")
+    candidates = [
+        Path.home() / ".codex" / "skills" / "agent-runtime" / "scripts" / "agent.py",
+        Path.home() / ".pouch" / "skills" / "agent-runtime" / "scripts" / "agent.py",
+        Path(__file__).resolve().parents[3] / "skills" / "agent-runtime" / "scripts" / "agent.py",
+    ]
+    runtime_path = next((item for item in candidates if item.is_file()), None)
+    if runtime_path is None:
+        raise RuntimeError("找不到共享 agent-runtime: " + ", ".join(str(item) for item in candidates))
     spec = importlib.util.spec_from_file_location("pouch_agent_runtime", runtime_path)
     if spec is None or spec.loader is None:
         raise RuntimeError(f"无法加载共享 agent-runtime: {runtime_path}")

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
 
 

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),
     }

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

skills/tester/tests/test_mobile_bundle.py

@@ -50,6 +50,84 @@ class MobileBundleTests(unittest.TestCase):
         with self.assertRaises(bundle.ArtifactFailure):
             bundle.parse_artifacts(["android=/tmp/a.apk", "android=/tmp/b.apk"])
 
+    def test_parse_bindings_keeps_platforms_separate(self):
+        self.assertEqual(
+            bundle.parse_bindings(["android=tester.android", "ios=tester.ios"], ["android", "ios"]),
+            {"android": "tester.android", "ios": "tester.ios"},
+        )
+        self.assertEqual(
+            bundle.parse_bindings([], ["android"]),
+            {"android": "tester.android"},
+        )
+        with self.assertRaises(bundle.ArtifactFailure):
+            bundle.parse_bindings(["android=tester.test", "ios=tester.test"], ["android", "ios"])
+
+    def test_preflight_matrix_rejects_remote_platform_using_local_runner(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            agent_dir = root / ".pouch"
+            agent_dir.mkdir()
+            (agent_dir / "agent.yaml").write_text(
+                """
+schemaVersion: 1
+locations:
+  local-app:
+    environment: local
+    host: local
+    project: test
+    path: /tmp/test
+  remote-ios:
+    environment: macmini
+    host: runtime:12345678-1234-1234-1234-123456789012
+    project: test
+    path: /tmp/test-ios
+profiles:
+  test-agent:
+    role: test
+    cli: codex
+    tier: standard
+    model: gpt-5
+    reasoningEffort: medium
+    permissionMode: workspace-write
+bindings:
+  tester.android: {adapter: tester, location: local-app, profile: test-agent}
+  tester.ios: {adapter: tester, location: remote-ios, profile: test-agent}
+""".strip()
+                + "\n",
+                encoding="utf-8",
+            )
+            config = root / "matrix.yaml"
+            config.write_text(
+                """
+matrix:
+  - platform: android
+    tester: {binding: tester.android, node: local, runner: {kind: local}}
+  - platform: ios
+    tester: {binding: tester.ios, node: local, runner: {kind: local}}
+""".strip()
+                + "\n",
+                encoding="utf-8",
+            )
+            with self.assertRaisesRegex(bundle.EnvironmentFailure, "ios.*contract runner"):
+                bundle.validate_preflight_matrix(
+                    root,
+                    config,
+                    {"android": "tester.android", "ios": "tester.ios"},
+                    ["android", "ios"],
+                )
+
+    def test_remote_artifact_path_comes_from_preflight_matrix(self):
+        with tempfile.TemporaryDirectory() as temp:
+            config = Path(temp) / "matrix.yaml"
+            config.write_text(
+                "matrix:\n  - platform: ios\n    artifact:\n      remote_path: /Users/tester/app.ipa\n",
+                encoding="utf-8",
+            )
+            self.assertEqual(
+                bundle.preflight_remote_artifacts(config, ["ios"]),
+                {"ios": "/Users/tester/app.ipa"},
+            )
+
     def test_new_worktree_each_run(self):
         with tempfile.TemporaryDirectory() as temp:
             repo, worktrees = Path(temp) / "repo", Path(temp) / "worktrees"
@@ -80,6 +158,16 @@ class MobileBundleTests(unittest.TestCase):
             self.assertTrue(result["verified"])
             self.assertTrue(any("install" in call for call in calls))
 
+    def test_android_identity_accepts_bare_apkanalyzer_output(self):
+        with tempfile.TemporaryDirectory() as temp:
+            apk = Path(temp) / "app.apk"
+            apk.write_bytes(b"apk")
+            context = bundle.RuntimeContext(Path(temp))
+            with mock.patch.object(bundle.shutil, "which", side_effect=lambda name: "apkanalyzer" if name == "apkanalyzer" else None), mock.patch.object(
+                bundle, "run_command", return_value=bundle.CommandResult(0, "page.writes\n")
+            ):
+                self.assertEqual(bundle.android_identity(apk, context), "page.writes")
+
     def test_android_missing_device_is_environment_failure(self):
         context = bundle.RuntimeContext(Path("."))
         with mock.patch.object(bundle.shutil, "which", return_value="/sdk/adb"), mock.patch.object(

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

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