diff --git a/DEVIATIONS.md b/DEVIATIONS.md
index c7e8200..40a0823 100644
--- a/DEVIATIONS.md
+++ b/DEVIATIONS.md
@@ -83,3 +83,13 @@
     `demo/` 同步角色化：run 文档 `selector_backend` → `fast_model`（新增 `slow_model` 名），
     排序调用计数 `jev_calls` → `fast_calls`、`jev_latency_ms_total` → `fast_latency_ms_total`
     （这两个是 demo 层自算字段，不在结果目录契约里）。
+
+## 可选阶段执行（2026-10-24）
+
+28. 【行为，可选】`execution_mode=stage` 时，慢模型一次规划当前阶段的多个里程碑，
+    快模型依据每帧新观察重新选择动作；规则后置条件优先，证据不足才调用模型 Verify。
+    阶段完成、目标无法绑定、执行失败或阶段动作上限触发后返回慢模型。
+    不采用严格只规划一次的模式；阶段完成不等于任务完成，原有独立终态检查保留。
+    默认 `step` 路径不变，动作/观察/回执协议不变。
+    此为旧阶段执行实验的组件化移植，不包含其传输重试、部署或环境准备逻辑。
+    详细实现边界见 [STAGE_EXECUTION.md](docs/STAGE_EXECUTION.md)。
diff --git a/README.md b/README.md
index 2ecb20d..9ba793e 100644
--- a/README.md
+++ b/README.md
@@ -293,3 +293,9 @@ python -m http.server 8899 --directory demo/site    # 本地预览
 | [DEVIATIONS.md](DEVIATIONS.md) | 与旧仓库 SA-CUAharness 的已知行为差异清单 |
 | [AGENTS.md](AGENTS.md) | 开发纪律与目录地图 |
 | [demo/README.md](demo/README.md) | demo 站点说明 |
+
+## 阶段执行（可选）
+
+默认保持逐步规划；启用 `--execution-mode stage` 后，由慢模型规划阶段、快模型连续选择动作，
+规则验证优先，仅在必要时调用模型验证或重规划。
+配置、执行边界、开发入口和验收方法见 [阶段执行说明](docs/STAGE_EXECUTION.md)。
diff --git a/docs/CLUSTER_RUNTIME.md b/docs/CLUSTER_RUNTIME.md
index e44f84d..d929451 100644
--- a/docs/CLUSTER_RUNTIME.md
+++ b/docs/CLUSTER_RUNTIME.md
@@ -41,6 +41,7 @@ worker 入口、共享资源与已知坑。
 | `harness.fast_model.apis.<名字>` | 快模型 API 条目：`backend`（`jev` / `kev`）+ `endpoint` / `endpoint_file` + `timeout_seconds` |
 | `harness.fast_model.image_stage.write` / `.read` | 截图中转目录的**两侧视角**：写入侧在 adpc，读取侧在 evobox（服务端宿主机要能读到） |
 | `harness.route.policy` / `.allow_cli` / `.cli_timeout_seconds` | 路由策略与 CLI 开关 |
+| `harness.execution.mode` / `.stage_max_actions` | 可选：step（默认）或 stage；阶段内动作上限默认 24。见 [阶段执行说明](STAGE_EXECUTION.md) |
 | `harness.budget.max_steps` / `.max_wall_seconds` | 单题预算默认值（命令行可覆盖） |
 
 **慢模型与快模型都是命名 API 登记表**：登记多套大脑/小脑，启动命令用 `--slow-model` /
@@ -135,6 +136,7 @@ cleanup  stop（释放资源）+ delete（永久删除 rjob 记录）——默
 | `--slow-model NAME` | 慢模型（大脑）API 名，取自 `harness.slow_model.apis`；缺省用 `default` |
 | `--fast-model NAME` | 快模型（小脑）API 名，取自 `harness.fast_model.apis`；缺省用 `default` |
 | `--route-policy` / `--allow-cli` | 路由策略与 CLI 开关 |
+| `--execution-mode` / `--stage-max-actions` | step/stage 模式与阶段动作上限 |
 | `--max-steps` / `--max-wall` | 单题预算（步数 / 墙钟秒） |
 | `--slow-reasoning` | `off` / `low`（默认）/ `medium` / `high` |
 | `--slow-max-tokens` | 慢模型单次响应上限 |
diff --git a/docs/RESULT_DIR_CONTRACT.md b/docs/RESULT_DIR_CONTRACT.md
index b975157..904270c 100644
--- a/docs/RESULT_DIR_CONTRACT.md
+++ b/docs/RESULT_DIR_CONTRACT.md
@@ -62,6 +62,7 @@ OSWorld 2.1 显式配置替代 judge/用户模拟器时，`code_version.json.res
 |------|--------|---------|
 | `task_requirements` | verifier.task_requirements | episode 开始时抽取任务需求（1-12 条） |
 | `plan_step` | planner.plan_step | 每步规划（截图+可见控件+需求+近期动作→下一步决策） |
+| `plan_stage` | stage_planner.plan_stage | 仅 stage 模式：初始阶段、阶段完成或请求修复时生成阶段计划 |
 | `verify_observation` | verifier.verify_observation | 每步动作执行后独立验证（BEFORE/AFTER 截图→achieved/not_achieved/uncertain）|
 
 `verify_observation` 不信任 planner 的自我宣称：它拿前后截图+预期变化独立裁决，驱动 recent 历史、no_action 连击计数、toggle 检查、以及终态双确认（terminal_claim 模式下逐条核验 requirements + 输出文件存在性）。
@@ -75,8 +76,17 @@ OSWorld 2.1 显式配置替代 judge/用户模拟器时，`code_version.json.res
 | `cli` | planner 的 route=cli 决策 + 命令 |
 | `gui_verified` / `cli_verified` | 验证裁决（outcome/task_status/evidence/state_changed）+ 回执 |
 | `terminal` | 终态裁决（planner 声称完成后的 verifier 双确认全文） |
+| `stage_local` | 消费已生成的阶段计划；含 stage_goal、milestone 和当帧重绑定后的 plan，不代表新的模型调用 |
+| `stage_progress` | 里程碑推进，含 stage_goal、milestone、验证 source |
+| `stage_replan` | 阶段修复请求，含原因及当前里程碑；不等于调用已经完成 |
 | `action`（kind 字段） | runner 写的执行记录 `{step, kind: "action", action: {…协议动作…}}` |
 
+stage 模式的 planner 事件只在真实规划调用时写入，增加 mode、reason、seconds；
+gui/cli 中的 plan 带 planner_called，表示本轮是否刚调用规划模型。
+gui_verified/cli_verified 的 verification 增加 source（rule_receipt / rule_a11y / model）
+与 seconds。规则检查不写 llm_calls，不能计为一次慢模型 Verify。
+其他文件和动作/观察/回执协议不变；详细触发规则见 [STAGE_EXECUTION.md](STAGE_EXECUTION.md)。
+
 ## failed_unscored 纪律
 
 API/VM/评分异常 ≠ 0 分——异常题不写 `result.txt`，`episode.json.status=failed_unscored`，run 继续下一题。`summary.json.failed_unscored` 单独计数，不计入均分。
diff --git a/docs/STAGE_EXECUTION.md b/docs/STAGE_EXECUTION.md
new file mode 100644
index 0000000..60c90f9
--- /dev/null
+++ b/docs/STAGE_EXECUTION.md
@@ -0,0 +1,3 @@
+# 阶段执行
+
+通过 `--execution-mode stage` 启用：慢模型规划阶段，JEV/Kev 连续选择 GUI 动作，CLI 执行计划中的命令；优先使用规则验证，证据不足时调用模型 Verify，失败或状态偏离时按需重规划，最终完成仍需独立验证。默认保留原有 `step` 模式。
diff --git a/infra/config.example.json b/infra/config.example.json
index 33c965d..fc7784f 100644
--- a/infra/config.example.json
+++ b/infra/config.example.json
@@ -49,6 +49,10 @@
       "allow_cli": true,
       "cli_timeout_seconds": 90
     },
+    "execution": {
+      "mode": "step",
+      "stage_max_actions": 24
+    },
     "budget": {
       "max_steps": 30,
       "max_wall_seconds": 1800
diff --git a/src/tandem/components.py b/src/tandem/components.py
index 3cc0ae8..92b4c0e 100644
--- a/src/tandem/components.py
+++ b/src/tandem/components.py
@@ -14,6 +14,7 @@
 from .planner import plan_step
 from .proposals import focused_candidates
 from .selector import JevOmniPolicy
+from .stage_planner import plan_stage
 from .targets import A11yTargetBuilder
 from .verifier import task_requirements, verify_observation
 
@@ -42,7 +43,7 @@ def build_default(*, client, endpoint: str, config: SessionConfig,
                                  image_stage=image_stage)
     return Components(
         planner=partial(
-            plan_step,
+            plan_stage if config.execution_mode == "stage" else plan_step,
             client,
             allow_cli=config.allow_cli and config.route_policy != "gui",
             route_policy=config.route_policy,
diff --git a/src/tandem/config.py b/src/tandem/config.py
index 6aeb453..7a43bb6 100644
--- a/src/tandem/config.py
+++ b/src/tandem/config.py
@@ -15,6 +15,8 @@ class SessionConfig:
     allow_cli: bool = True
     route_policy: str = "auto"
     cli_timeout_seconds: int = 90
+    execution_mode: str = "step"
+    stage_max_actions: int = 24
     slow_max_tokens: int = 16384
     slow_timeout_seconds: float = 120
     slow_direct_api: bool = False
@@ -23,6 +25,10 @@ class SessionConfig:
     fast_timeout_seconds: float = 300
 
     def __post_init__(self):
+        if self.execution_mode not in {"step", "stage"}:
+            raise ValueError("execution_mode must be step or stage")
+        if type(self.stage_max_actions) is not int or self.stage_max_actions < 1:
+            raise ValueError("stage_max_actions must be a positive integer")
         if self.route_policy not in {"auto", "gui", "cli"}:
             raise ValueError("route_policy must be auto, gui or cli")
         if not 1 <= self.cli_timeout_seconds <= 120:
diff --git a/src/tandem/episode.py b/src/tandem/episode.py
index dc87a6f..b0643e0 100644
--- a/src/tandem/episode.py
+++ b/src/tandem/episode.py
@@ -117,12 +117,10 @@ def verify_pending(session, observation: Observation, tree: str | None) -> None:
     receipt = observation.last_receipt
     if receipt is None:
         raise RuntimeError("环境未返回上一步执行回执（协议违例）")
-    screenshot_after = Path(observation.screenshot.path)
     screenshot_before = Path(pending["before_screenshot"])
     if pending["kind"] == "cli":
-        verification = session.components.verifier(
-            session.instruction,
-            screenshot_after,
+        verification = _verify_action(
+            session, observation, tree, pending,
             previous=screenshot_before,
             expected=pending["plan"]["expected_observation"],
             action=pending["action"],
@@ -166,9 +164,8 @@ def verify_pending(session, observation: Observation, tree: str | None) -> None:
         and tree is not None
         else None
     )
-    verification = session.components.verifier(
-        session.instruction,
-        screenshot_after,
+    verification = _verify_action(
+        session, observation, tree, pending,
         previous=screenshot_before,
         expected=pending["plan"]["expected_observation"],
         action=winner.compact(),
@@ -217,3 +214,11 @@ def verify_pending(session, observation: Observation, tree: str | None) -> None:
             "control_state_check": control_check,
         }
     )
+
+
+def _verify_action(session, observation, tree, pending, **kwargs):
+    if session.stage is not None:
+        return session.stage.verify(session, observation, tree, pending, **kwargs)
+    return session.components.verifier(
+        session.instruction, Path(observation.screenshot.path), **kwargs,
+    )
diff --git a/src/tandem/session.py b/src/tandem/session.py
index 78d40f2..15dc41c 100644
--- a/src/tandem/session.py
+++ b/src/tandem/session.py
@@ -17,6 +17,7 @@
 from .focus import prune_stalled
 from .protocol.models import Action, CliAction, FailAction, GuiAction, Observation, WaitAction
 from .selector import ranking_dict
+from .stages import StageExecution
 from .stopping import NO_ACTION_LIMIT, budget_exhausted
 
 
@@ -40,6 +41,9 @@ def begin(self, instruction: str = "", requirements: list | None = None) -> None
         self.failed_toggles: dict[str, tuple[str, object]] = {}
         self.pending: dict | None = None
         self.done_probe: dict | None = None
+        self.stage = (
+            StageExecution(self.config.stage_max_actions) if self.config.execution_mode == "stage" else None
+        )
 
     # ------------------------------------------------------------------ 主入口
     def act(self, observation: Observation) -> Action:
@@ -64,17 +68,20 @@ def act(self, observation: Observation) -> Action:
         # 同一观察内最多重规划 NO_ACTION_LIMIT+3 次：完成被拒/无候选会递增 no_action 并重规划，
         # 与旧循环 continue 语义一致（旧版重取 obs，本帧无动作时内容等价）。
         for _ in range(NO_ACTION_LIMIT + 3):
-            plan = self.components.planner(
-                self.instruction,
-                screenshot,
-                targets,
-                self.recent,
-                tree if tree is not None else "<empty/>",
-                self.verified_focus,
-                self.requirements,
-                observation.budget.steps_remaining,
-            )
-            self.events.append({"step": observation.step_idx, "route": "planner", "plan": plan})
+            if self.stage is not None:
+                plan = self.stage.plan(self, observation, targets, tree, screenshot)
+                if plan is None:
+                    self.consecutive_no_action += 1
+                    if self.consecutive_no_action >= NO_ACTION_LIMIT:
+                        return FailAction(reason="no_action_limit")
+                    continue
+            else:
+                plan = self.components.planner(
+                    self.instruction, screenshot, targets, self.recent,
+                    tree if tree is not None else "<empty/>", self.verified_focus,
+                    self.requirements, observation.budget.steps_remaining,
+                )
+                self.events.append({"step": observation.step_idx, "route": "planner", "plan": plan})
             if plan["done"]:
                 terminal = handle_done(self, plan, observation, tree)
                 if terminal is not None:
@@ -99,6 +106,7 @@ def _emit_cli(self, plan: dict, observation: Observation) -> CliAction:
             "action": action.to_dict(),
             "plan": plan,
             "before_screenshot": observation.screenshot.path,
+            "before_tree": self._read_tree(observation),
         }
         self.consecutive_no_action = 0
         self.events.append(
@@ -146,7 +154,7 @@ def _plan_gui(
         for service_retry in range(3):
             try:
                 ranking = self.components.selector.rank(
-                    stage_goal=self.instruction,
+                    stage_goal=plan.get("stage_goal", self.instruction),
                     intent=local_goal,
                     active_window="",
                     targets=targets,
@@ -175,6 +183,8 @@ def _plan_gui(
             }
         )
         if compiled is None:
+            if self.stage is not None:
+                self.stage.invalidate(self, "no_executable_candidate", observation.step_idx)
             self.consecutive_no_action += 1
             self.recent.append(
                 {
diff --git a/src/tandem/stage_planner.py b/src/tandem/stage_planner.py
new file mode 100644
index 0000000..b04d6aa
--- /dev/null
+++ b/src/tandem/stage_planner.py
@@ -0,0 +1,168 @@
+"""One bounded stage per slow-model call; plans are internal, not wire-protocol actions."""
+
+from __future__ import annotations
+
+import base64
+import json
+
+from .focus import input_and_menu_context
+
+OPERATIONS = {"click", "double_click", "type_text", "keypress", "scroll_up", "scroll_down",
+              "set_slider", "wait"}
+
+
+def valid_stage(value, *, allow_cli=True, route_policy="auto"):
+    if not isinstance(value, dict) or type(value.get("done")) is not bool:
+        return False
+    if value["done"]:
+        return (value.get("task_status") in {"completed", "infeasible"}
+                and isinstance(value.get("current_goal"), str) and bool(value["current_goal"].strip()))
+    milestones = value.get("milestones")
+    if (not isinstance(value.get("stage_goal"), str) or not value["stage_goal"].strip()
+            or not isinstance(milestones, list) or not 1 <= len(milestones) <= 8):
+        return False
+    for item in milestones:
+        if not isinstance(item, dict) or not all(
+            isinstance(item.get(key), str) and item[key].strip()
+            for key in ("current_goal", "expected_observation")
+        ):
+            return False
+        route = item.get("route")
+        if route == "cli":
+            if (not allow_cli or route_policy == "gui" or not isinstance(item.get("command"), str)
+                    or not item["command"].strip()):
+                return False
+        elif route == "gui" and route_policy != "cli":
+            proposals = item.get("action_proposals")
+            if not isinstance(proposals, list) or not 1 <= len(proposals) <= 4:
+                return False
+            for proposal in proposals:
+                if not isinstance(proposal, dict) or proposal.get("operation") not in OPERATIONS:
+                    return False
+                if "parent_contains" in proposal and not isinstance(proposal["parent_contains"], str):
+                    return False
+                if "target_name" in proposal and not all(
+                    isinstance(proposal.get(key), str) for key in ("target_name", "target_role")
+                ):
+                    return False
+                operation = proposal["operation"]
+                if operation in {"click", "double_click", "scroll_up", "scroll_down", "set_slider"}:
+                    if not all(isinstance(proposal.get(key), str) and proposal[key]
+                               for key in ("target_name", "target_role")):
+                        return False
+                if operation == "type_text" and (
+                    not isinstance(proposal.get("text"), str)
+                    or proposal.get("input_mode") not in {"insert", "replace"}
+                ):
+                    return False
+                if operation == "keypress":
+                    sequence = proposal.get("key_sequence")
+                    if (not isinstance(sequence, list) or not 1 <= len(sequence) <= 8
+                            or not all(isinstance(keys, list) and 1 <= len(keys) <= 4
+                                       and all(isinstance(key, str) for key in keys) for keys in sequence)):
+                        return False
+                if operation == "set_slider" and (
+                    type(proposal.get("value")) not in (int, float) or not 0 <= proposal["value"] <= 1
+                ):
+                    return False
+        else:
+            return False
+        checks = item.get("checks", [])
+        if not isinstance(checks, list) or len(checks) > 4:
+            return False
+        for check in checks:
+            if not isinstance(check, dict) or not isinstance(check.get("target"), dict):
+                return False
+            if not all(isinstance(check["target"].get(key), str) for key in ("name", "role")):
+                return False
+            if ("parent_contains" in check["target"]
+                    and not isinstance(check["target"]["parent_contains"], str)):
+                return False
+            prop = check.get("property")
+            if prop not in {"visible", "value", "checked", "selected", "focused", "expanded"}:
+                return False
+            if prop == "value":
+                if not isinstance(check.get("expected"), str):
+                    return False
+            elif type(check.get("expected")) is not bool:
+                return False
+    return True
+
+
+def plan_stage(client, instruction, screenshot, targets, recent, accessibility_tree,
+               verified_focus=None, requirements=None, remaining_steps=None, *,
+               allow_cli=True, route_policy="auto"):
+    inputs, menus = input_and_menu_context(accessibility_tree)
+    context = {
+        "task": instruction, "requirements": requirements or [], "remaining_actions": remaining_steps,
+        "visible_controls": [
+            {"name": target.name, "role": target.role, "value": target.value,
+             "parents": target.parent_context, "focused": target.focused,
+             "checked": target.checked, "selected": target.selected} for target in targets
+        ],
+        "focused_inputs": [verified_focus] if verified_focus else inputs,
+        "menu_paths": menus, "recent_actions_and_evidence": recent[-6:],
+        "allow_cli": allow_cli, "route_policy": route_policy,
+    }
+    prompt = """Plan the next bounded desktop stage, not just one action and not the entire unseen task.
+Return only JSON. For a running stage use:
+{"done":false,"stage_goal":"...","milestones":[{"route":"gui",
+"current_goal":"one concrete transition","expected_observation":"specific resulting state",
+"action_proposals":[{"operation":"click","target_name":"exact name","target_role":"push_button"}],
+"checks":[{"target":{"name":"result control","role":"entry"},"property":"focused","expected":true}]}]}.
+Prefer 2-4 milestones for the CURRENT stage, with a hard maximum of eight. If the whole task needs
+more, return only its first bounded stage; subsequent calls will plan the rest from fresh evidence.
+Do not append the rest of the task or extra audits beyond that limit. Stop at an uncertain boundary instead of
+inventing unseen UI. Each milestone is one action, except a bounded scroll-until-visible transition.
+action_proposals are alternatives executable from the SAME before-state, never an ordered sequence.
+Opening an input, typing into it, and submitting are separate milestones, not three proposals in one.
+Future targets need exact semantic names/roles, optional parent_contains for duplicate names, never
+cached target_id or coordinates. Targets are rebound from each new observation. GUI operations:
+click, double_click, scroll_up, scroll_down, type_text, keypress, set_slider, wait.
+type_text requires exact text and input_mode insert/replace. Omit target fields only for a confirmed
+keyboard-focused input. keypress requires key_sequence, e.g. [["ctrl","l"]]. set_slider needs value 0..1.
+checks are optional; use [] when independent model verification is needed. The property field must
+be exactly one of ["visible","value","checked","selected","focused","expanded"], never "exact value".
+Use property="value" with expected:string for exact text, otherwise expected:boolean, on an exact target
+{name,role,parent_contains?}. Check the RESULT, not the pre-existing button being clicked.
+Unknown or duplicate controls cannot prove success. Do not invent conditions to bypass verification.
+For CLI use route=cli, command (bounded Bash script in /home/user), current_goal, expected_observation,
+checks:[]. Actual stdout/stderr will be verified; exit=0 or printed success alone proves nothing.
+When CLI is allowed under auto routing, prefer it for deterministic public-file/data edits or audits;
+use GUI for application state and inspect the real application after CLI document edits.
+Honor allow_cli/route_policy. Never inspect evaluators, reference answers, rewards, unrelated secrets,
+or modify scoring. Do not install packages or broadly delete files unless the task requires it.
+After CLI document edits, reopen/reload and inspect the real application before completion.
+On a repair request, use the fresh observation and recorded failure, preserve verified work, and
+replace only remaining work. Do not repeat a failed method unchanged. Missing candidates do not
+make a task infeasible. An unexpected UI requires a revised grounded plan, not invented controls.
+Preserve the user's exact scope and literal values. Do not introduce unrequested matching modes,
+word boundaries, selection filters, case restrictions, or exceptions. A narrower self-chosen goal
+cannot satisfy a broader original requirement. Check commands against the original instruction,
+not just their own success messages; all occurrences means all literal occurrences unless specified.
+The original task is authoritative; extracted requirements are derived notes, not permission to
+reinterpret the task or add restrictions. Resolve any mismatch in favor of the original wording.
+Use {"done":true,"task_status":"completed" or "infeasible","current_goal":"observed reason"}
+only when the original task is fully satisfied (including saving) or demonstrably infeasible.
+Completion is independently verified; a stage finishing is NOT proof of task completion."""
+    messages = [{"role": "system", "content": prompt}, {"role": "user", "content": [
+        {"type": "text", "text": json.dumps(context, ensure_ascii=False)},
+        {"type": "image_url", "image_url": {"url": "data:image/png;base64,"
+         + base64.b64encode(screenshot.read_bytes()).decode("ascii")}},
+    ]}]
+    def validate(value):
+        valid = valid_stage(value, allow_cli=allow_cli, route_policy=route_policy)
+        if not valid:
+            # Use the client's existing attempts; do not truncate, coerce or execute invalid plans.
+            messages.extend([
+                {"role": "assistant", "content": json.dumps(value, ensure_ascii=False)},
+                {"role": "user", "content": (
+                    "Invalid stage schema. Return only the current stage with 1-8 milestones, respecting "
+                    "allow_cli/route_policy. Each check property must be exactly visible, value, checked, "
+                    "selected, focused or expanded (not 'exact value'). Preserve the ORIGINAL task scope. "
+                    "Correct the JSON using the field names and types specified above."
+                )},
+            ])
+        return valid
+
+    return client.json_request(messages, validate, role="plan_stage")
diff --git a/src/tandem/stage_rules.py b/src/tandem/stage_rules.py
new file mode 100644
index 0000000..6249129
--- /dev/null
+++ b/src/tandem/stage_rules.py
@@ -0,0 +1,75 @@
+"""Observable stage postconditions. Unknown or ambiguous evidence never proves success."""
+
+from __future__ import annotations
+
+import xml.etree.ElementTree as ET
+
+
+def controls(tree: str | None) -> list[dict]:
+    """Read active, visible controls, including non-actionable labels and paragraphs."""
+    if tree is None:
+        return []
+    result = []
+
+    def visit(node, parents=(), active=False, hidden=False):
+        attrs = {key.rsplit("}", 1)[-1]: value for key, value in node.attrib.items()}
+        active = active or attrs.get("active") == "true"
+        hidden = hidden or attrs.get("showing") == "false" or attrs.get("visible") == "false"
+        name = attrs.get("name", "")
+        if active and not hidden:
+            result.append({
+                "name": name, "role": node.tag.rsplit("}", 1)[-1].replace("-", "_"),
+                "parents": parents, "value": attrs.get("value", node.text or ""),
+                **{key: attrs.get(key) == "true" if key in attrs else None
+                   for key in ("focused", "selected", "checked", "expanded")},
+            })
+        for child in node:
+            visit(child, parents + ((name,) if name else ()), active, hidden)
+
+    visit(ET.fromstring(tree))
+    return result
+
+
+def matches(items, query):
+    return [item for item in items
+            if item["role"] == query["role"] and item["name"] == query["name"]
+            and (not query.get("parent_contains")
+                 or any(query["parent_contains"] in parent for parent in item["parents"]))]
+
+
+def check_state(items, check) -> str:
+    found = matches(items, check["target"])
+    if len(found) != 1:
+        return "uncertain"
+    observed = True if check["property"] == "visible" else found[0].get(check["property"])
+    if observed is None:
+        return "uncertain"
+    return "met" if observed == check["expected"] else "not_met"
+
+
+def rule_verification(pending, receipt, tree):
+    """Return None for semantic model verification; never use exit=0 as success proof."""
+    base = {"task_status": "incomplete", "path_input_focused": False, "state_changed": False}
+    if not receipt.acknowledged or receipt.returncode != 0:
+        return {**base, "outcome": "not_achieved", "evidence": "Execution receipt failed.",
+                "source": "rule_receipt"}
+    checks = pending["plan"].get("checks", [])
+    if not checks:
+        return None
+    before, after = controls(pending.get("before_tree")), controls(tree)
+    states = [check_state(after, check) for check in checks]
+    if "uncertain" in states:
+        return None
+    changed = before != after
+    if "not_met" in states:
+        # Only scrolling may repeat locally; never blindly repeat clicks, typing or commands.
+        repeat = pending["kind"] == "gui" and pending["winner"].operation == "scroll" and changed
+        return {**base, "outcome": "uncertain" if repeat else "not_achieved",
+                "state_changed": changed, "continue_local": repeat,
+                "evidence": f"Observable postconditions: {states}.", "source": "rule_a11y"}
+    # A condition already true before the action is not evidence of its effect.
+    if not any(check_state(before, check) != "met" for check in checks):
+        return None
+    return {**base, "outcome": "achieved", "state_changed": changed,
+            "evidence": "All postconditions met, including a newly observed condition.",
+            "source": "rule_a11y"}
diff --git a/src/tandem/stages.py b/src/tandem/stages.py
new file mode 100644
index 0000000..85973b0
--- /dev/null
+++ b/src/tandem/stages.py
@@ -0,0 +1,93 @@
+"""Session-local stage progress; every action still uses current observations and receipts."""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+from .stage_rules import rule_verification
+
+
+class StageExecution:
+    def __init__(self, max_actions):
+        self.max_actions = max_actions
+        self.milestones = []
+        self.index = 0
+        self.actions = 0
+        self.goal = ""
+        self.reason = "initial_stage"
+
+    def invalidate(self, session, reason, step):
+        session.events.append({"route": "stage_replan", "step": step, "reason": reason,
+                               "stage_goal": self.goal, "milestone": self.index})
+        session.recent.append({"operation": "stage_replan", "outcome": "no_action",
+                               "goal": self.goal, "reason": reason,
+                               "verified_milestones": [item["current_goal"]
+                                                       for item in self.milestones[:self.index]]})
+        self.milestones = []
+        self.reason = reason
+
+    def plan(self, session, observation, targets, tree, screenshot):
+        called = False
+        if self.milestones and self.actions >= self.max_actions:
+            self.invalidate(session, "stage_action_limit", observation.step_idx)
+        if not self.milestones:
+            started = time.perf_counter()
+            plan = session.components.planner(
+                session.instruction, screenshot, targets, session.recent, tree or "<empty/>",
+                session.verified_focus, session.requirements, observation.budget.steps_remaining,
+            )
+            session.events.append({"route": "planner", "step": observation.step_idx,
+                                   "plan": plan, "reason": self.reason, "mode": "stage",
+                                   "seconds": time.perf_counter() - started})
+            called = True
+            if plan["done"]:
+                return plan
+            self.goal = plan["stage_goal"]
+            self.milestones, self.index, self.actions = plan["milestones"], 0, 0
+        milestone = self.milestones[self.index]
+        plan = {**milestone, "done": False, "task_status": "running",
+                "stage_goal": self.goal, "milestone": self.index, "planner_called": called}
+        if plan["route"] == "gui":
+            bound = []
+            for proposal in plan["action_proposals"]:
+                proposal = {key: value for key, value in proposal.items() if key != "target_id"}
+                if proposal.get("target_name") is not None and proposal["operation"] != "wait":
+                    found = [target for target in targets
+                             if target.name == proposal["target_name"]
+                             and target.role == proposal.get("target_role")
+                             and (not proposal.get("parent_contains")
+                                  or any(proposal["parent_contains"] in p for p in target.parent_context))]
+                    if len(found) != 1:
+                        continue
+                    proposal["target_id"] = found[0].id
+                bound.append(proposal)
+            if not bound:
+                self.invalidate(session, "target_unbound_or_ambiguous", observation.step_idx)
+                return None
+            plan["action_proposals"] = bound
+        session.events.append({"route": "stage_local", "step": observation.step_idx,
+                               "stage_goal": self.goal, "milestone": self.index, "plan": plan})
+        return plan
+
+    def verify(self, session, observation, tree, pending, **kwargs):
+        started = time.perf_counter()
+        result = rule_verification(pending, observation.last_receipt, tree)
+        if result is None:
+            result = session.components.verifier(
+                session.instruction, Path(observation.screenshot.path), **kwargs,
+            )
+            result = {**result, "source": "model"}
+        result = {**result, "seconds": time.perf_counter() - started}
+        self.actions += 1
+        if result["outcome"] == "achieved":
+            session.events.append({"route": "stage_progress", "step": observation.step_idx,
+                                   "stage_goal": self.goal, "milestone": self.index,
+                                   "source": result["source"]})
+            self.index += 1
+            if self.index == len(self.milestones):
+                self.milestones = []
+                self.reason = "stage_completed"
+        elif not result.get("continue_local"):
+            self.invalidate(session, "execution_failed_or_unverified", observation.step_idx)
+        return result
diff --git a/src/tandem_eval/__main__.py b/src/tandem_eval/__main__.py
index d9e2e43..e3ffc95 100644
--- a/src/tandem_eval/__main__.py
+++ b/src/tandem_eval/__main__.py
@@ -36,6 +36,9 @@ def main(argv: list[str] | None = None) -> int:
     run_cmd.add_argument("--allow-cli", choices=["true", "false"], default=None,
                          help="harness 是否允许 CLI 路由")
     run_cmd.add_argument("--route-policy", choices=["auto", "gui", "cli"], default=None)
+    run_cmd.add_argument("--execution-mode", choices=["step", "stage"], default=None)
+    run_cmd.add_argument("--stage-max-actions", type=int, default=None,
+                         help="阶段内动作数上限，达到后请求剩余工作的重规划")
     run_cmd.add_argument("--slow-model", default=None, metavar="NAME",
                          help="慢模型（大脑）API 名：config 的 harness.slow_model.apis 里选；默认取 default")
     run_cmd.add_argument("--fast-model", default=None, metavar="NAME",
@@ -91,6 +94,10 @@ def main(argv: list[str] | None = None) -> int:
         if models:
             overrides["models"] = models
         harness_config: dict = {}
+        if args.execution_mode is not None:
+            harness_config["execution_mode"] = args.execution_mode
+        if args.stage_max_actions is not None:
+            harness_config["stage_max_actions"] = args.stage_max_actions
         if args.allow_cli is not None:
             harness_config["allow_cli"] = args.allow_cli == "true"
         if args.route_policy:
diff --git a/src/tandem_eval/config_schema.py b/src/tandem_eval/config_schema.py
index 0dab909..9736f42 100644
--- a/src/tandem_eval/config_schema.py
+++ b/src/tandem_eval/config_schema.py
@@ -213,6 +213,15 @@ def validate(config: dict) -> list[str]:
     problems += _validate_fast_models(config)
     problems += _validate_absolute_paths(config)
     problems += _validate_osworld21(config)
+    execution = get_path(config, "harness.execution") or {}
+    if not isinstance(execution, dict):
+        problems.append("harness.execution（应为对象）")
+    else:
+        if execution.get("mode", "step") not in {"step", "stage"}:
+            problems.append("harness.execution.mode（应为 step / stage）")
+        limit = execution.get("stage_max_actions", 24)
+        if type(limit) is not int or limit < 1:
+            problems.append("harness.execution.stage_max_actions（应为正整数）")
     return problems
 
 
diff --git a/src/tandem_eval/entry.py b/src/tandem_eval/entry.py
index 5edbe6c..22bb2c3 100644
--- a/src/tandem_eval/entry.py
+++ b/src/tandem_eval/entry.py
@@ -98,15 +98,16 @@ def git_describe(repo_root: Path) -> str:
 class _ScriptedPlanner:
     """mock 模式的确定性 planner：先点第一个目标，再声称完成。"""
 
-    def __init__(self):
+    def __init__(self, *, stage_mode=False):
         self.calls = 0
+        self.stage_mode = stage_mode
 
     def __call__(self, instruction, screenshot, targets, recent, tree,
                  verified_focus, requirements, remaining_steps):
         self.calls += 1
         if self.calls == 1 and targets:
             target = targets[0]
-            return {
+            plan = {
                 "done": False, "task_status": "running",
                 "current_goal": "操作第一个控件", "expected_observation": "界面变化",
                 "action_proposals": [{
@@ -114,6 +115,10 @@ def __call__(self, instruction, screenshot, targets, recent, tree,
                     "target_name": target.name, "target_role": target.role,
                 }],
             }
+            if self.stage_mode:
+                plan.update(route="gui", checks=[])
+                return {"done": False, "stage_goal": "mock stage", "milestones": [plan]}
+            return plan
         return {"done": True, "task_status": "completed", "current_goal": "完成",
                 "expected_observation": ""}
 
@@ -141,14 +146,14 @@ def rank(self, **kwargs):
         )
 
 
-def mock_components() -> Components:
+def mock_components(*, stage_mode=False) -> Components:
     from tandem.candidates import ActionCandidateBuilder
     from tandem.compile import gui_action
     from tandem.proposals import focused_candidates
     from tandem.targets import A11yTargetBuilder
 
     return Components(
-        planner=_ScriptedPlanner(), verifier=_ScriptedVerifier(),
+        planner=_ScriptedPlanner(stage_mode=stage_mode), verifier=_ScriptedVerifier(),
         requirements=lambda instruction: [], selector=_FirstCandidateSelector(),
         target_builder=A11yTargetBuilder(), candidate_builder=ActionCandidateBuilder(),
         focus_candidates=focused_candidates, grounder=None, compile_gui=gui_action,
@@ -159,7 +164,7 @@ def build_session(spec: dict, *, llm_tap: LlmTap | None = None) -> TandemSession
     config = SessionConfig(**spec.get("harness_config", {}))
     benchmark = spec.get("benchmark")
     if benchmark == "mock":
-        return TandemSession(mock_components(), config=config)
+        return TandemSession(mock_components(stage_mode=config.execution_mode == "stage"), config=config)
     if benchmark in {"osworld1", "osworld2", "osworld21"}:
         from tandem.components import build_default
         from tandem.providers import SlowModelClient
diff --git a/src/tandem_eval/rjob_spec.py b/src/tandem_eval/rjob_spec.py
index 3b81493..b33194e 100644
--- a/src/tandem_eval/rjob_spec.py
+++ b/src/tandem_eval/rjob_spec.py
@@ -102,7 +102,10 @@ def _secrets_section(benchmark: str, config: dict, slow_api: dict, fast_api: dic
 def _harness_config_section(config: dict, slow_api: dict, fast_api: dict) -> dict:
     """行为旋钮段：SessionConfig 的字段（与 harness 侧 dataclass 一一对应）。"""
     route = config["harness"]["route"]
+    execution = config["harness"].get("execution", {})
     return {
+        "execution_mode": execution.get("mode", "step"),
+        "stage_max_actions": execution.get("stage_max_actions", 24),
         "route_policy": route.get("policy", "auto"),
         "allow_cli": bool(route.get("allow_cli", True)),
         "cli_timeout_seconds": int(route.get("cli_timeout_seconds", 90)),
@@ -176,7 +179,7 @@ def build_job_spec(*, config: dict, run_id: str, repo_root: Path,
         "image_stage": _image_stage_section(config),
         "harness": {
             "name": "tandem",
-            "mode": "planner_led",
+            "mode": "stage" if harness_config["execution_mode"] == "stage" else "planner_led",
             "models": {"slow": slow_name, "fast": fast_name},
         },
         "harness_config": harness_config,
diff --git a/tests/eval/test_mock_chain.py b/tests/eval/test_mock_chain.py
index 529c119..a158723 100644
--- a/tests/eval/test_mock_chain.py
+++ b/tests/eval/test_mock_chain.py
@@ -7,6 +7,8 @@
 import sys
 from pathlib import Path
 
+import pytest
+
 from tandem_eval.entry import entry_main, run_from_spec
 from tandem_eval.env import MockEnv
 from tandem_eval.manifest import TaskRef
@@ -39,8 +41,9 @@ def _spec(tmp_path: Path, **overrides) -> dict:
     return base
 
 
-def test_full_mock_chain_produces_renderable_result_dir(tmp_path):
-    episodes = run_from_spec(_spec(tmp_path))
+@pytest.mark.parametrize("mode", ["step", "stage"])
+def test_full_mock_chain_produces_renderable_result_dir(tmp_path, mode):
+    episodes = run_from_spec(_spec(tmp_path, harness_config={"execution_mode": mode}))
     assert episodes and episodes[0]["status"] == "scored"
     run_root = tmp_path / "run"
     episode_dir = run_root / "episodes" / "mock-mock-000"