From 6b7884bf6e8cef736523c54f33c62efcd257c41e Mon Sep 17 00:00:00 2001
From: liuzhen <liuzhen@localhost>
Date: Thu, 22 Oct 2026 02:00:13 +0000
Subject: [PATCH 1/2] =?UTF-8?q?=E9=BB=98=E8=AE=A4=E6=A8=A1=E5=9E=8B?=
 =?UTF-8?q?=E5=88=87=E6=8D=A2=E8=87=B3=20OpenRouter=20gpt-6-astra=EF=BC=8C?=
 =?UTF-8?q?=E6=96=B0=E5=A2=9E=E6=9C=BA=E6=A2=B0=E8=87=82=E8=BF=90=E5=8A=A8?=
 =?UTF-8?q?=E8=A7=86=E9=A2=91=E5=BD=95=E5=88=B6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 默认服务商改为 OpenRouter（base_url: openrouter.ai/api/v1），
  默认模型 openai/gpt-6-astra；不包含任何 API key，密钥仍由
  环境变量或界面输入在运行期提供
- 新增 recording/motion.py：物理循环内按仿真时钟快照，
  独立线程离屏渲染双视角并经 ffmpeg 流式合成 motion.mp4；
  controlled 时序下 API 推理期间物理冻结，视频只含机械臂实际运动
- runner 接入录制生命周期（启动创建、步进快照、终止/关闭收尾）

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/maniloop/providers/credentials.py |   2 +-
 src/maniloop/providers/responses.py   |   2 +-
 src/maniloop/recording/motion.py      | 238 ++++++++++++++++++++++++++
 src/maniloop/runtime/runner.py        |  36 +++-
 src/maniloop/ui/chat.html             |   2 +-
 src/maniloop/ui/index.html            |   4 +-
 src/maniloop/ui/workspace.html        |   2 +-
 src/maniloop/ui/workspace.js          |   2 +-
 8 files changed, 279 insertions(+), 9 deletions(-)
 create mode 100644 src/maniloop/recording/motion.py

diff --git a/src/maniloop/providers/credentials.py b/src/maniloop/providers/credentials.py
index 1a45079..cb88933 100644
--- a/src/maniloop/providers/credentials.py
+++ b/src/maniloop/providers/credentials.py
@@ -21,7 +21,7 @@
 from urllib.parse import urlsplit, urlunsplit
 
 MAX_CONFIG_BYTES = 1024 * 1024
-DEFAULT_BASE_URL = "https://api.openai.com/v1"
+DEFAULT_BASE_URL = "https://openrouter.ai/api/v1"
 
 
 class ConfigError(ValueError):
diff --git a/src/maniloop/providers/responses.py b/src/maniloop/providers/responses.py
index bb9a8c4..8b28daa 100644
--- a/src/maniloop/providers/responses.py
+++ b/src/maniloop/providers/responses.py
@@ -25,7 +25,7 @@
     RateLimitError,
 )
 
-DEFAULT_MODEL = "gpt-6-astra"
+DEFAULT_MODEL = "openai/gpt-6-astra"
 from maniloop.core.observations import PUBLIC_OBSERVATION_FIELDS, guard_sensor_tree
 
 ACTION_SCHEMA = {
diff --git a/src/maniloop/recording/motion.py b/src/maniloop/recording/motion.py
new file mode 100644
index 0000000..ce0d2c4
--- /dev/null
+++ b/src/maniloop/recording/motion.py
@@ -0,0 +1,238 @@
+"""Live episode motion video for the in-process MuJoCo backend.
+
+The runner's physics loop hands immutable MjData snapshots to a worker thread
+at a fixed simulation-time cadence. The worker owns its offscreen context and
+streams composited dual-camera frames to ffmpeg, so recording never slows
+physics or inference. Controlled timing freezes physics during API calls, and
+paused/diagnostic periods never reach the physics loop; request latency
+therefore never appears in the video.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import queue
+import shutil
+import subprocess
+import threading
+from pathlib import Path
+
+import mujoco
+from PIL import Image, ImageDraw
+
+CAMERAS = ("external", "wrist")
+_MAX_ERRORS = 5
+
+
+class MotionRecorder:
+    """Dual-camera MP4 of everything physics actually did, nothing it did not."""
+
+    def __init__(
+        self,
+        sim,
+        run_dir,
+        *,
+        fps: int = 20,
+        panel: int = 320,
+        ffmpeg: str | None = None,
+        max_frames: int = 20 * 7200,
+    ):
+        if getattr(sim, "backend", None) != "mujoco":
+            raise ValueError("MotionRecorder supports only the in-process mujoco backend")
+        for name in CAMERAS:
+            sim.model.camera(name)  # Raises early when a scene lacks the cameras.
+        if type(fps) is not int or not 1 <= fps <= 60:
+            raise ValueError("fps must be an integer from 1 to 60")
+        binary = ffmpeg or shutil.which("ffmpeg")
+        if not binary or not Path(binary).is_file():
+            raise ValueError("ffmpeg was not found; motion video requires the ffmpeg executable")
+        self._model = sim.model
+        self._scene_option = getattr(sim, "scene_option", None)
+        self._fps = fps
+        self._panel = panel
+        self._interval = 1.0 / fps
+        self._next_due = None
+        self._max_frames = max_frames
+        self._queue: queue.Queue = queue.Queue(maxsize=8)
+        self._dropped = 0
+        self._captured = 0
+        self._errors = 0
+        self._last_sim_time = -math.inf
+        self._disabled = False
+        self._finalized = False
+        self._thread = threading.Thread(
+            target=self._work, name="motion-recorder", daemon=True
+        )
+        canvas = (2 * panel, panel + 56)
+        self._canvas = canvas
+        run_dir = Path(run_dir)
+        self._partial = run_dir / "motion.partial.mp4"
+        self._destination = run_dir / "motion.mp4"
+        self._process = subprocess.Popen(
+            [
+                str(binary), "-nostdin", "-hide_banner", "-loglevel", "error", "-y",
+                "-f", "rawvideo", "-pixel_format", "rgb24",
+                "-video_size", f"{canvas[0]}x{canvas[1]}",
+                "-framerate", str(fps), "-i", "pipe:0",
+                "-an", "-c:v", "libx264", "-preset", "veryfast",
+                "-crf", "20", "-pix_fmt", "yuv420p", "-movflags", "+faststart",
+                str(self._partial),
+            ],
+            stdin=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        self._thread.start()
+
+    def tick(self, sim) -> None:
+        """Physics-loop hook: snapshot state at the simulation-time cadence.
+
+        Runs in the physics thread; rendering happens in the worker. A short
+        blocking put throttles headless runs to encoder speed instead of
+        dropping motion; only sustained overload drops a snapshot.
+        """
+        if self._disabled or self._finalized or self._captured >= self._max_frames:
+            return
+        sim_time = float(sim.data.time)
+        if self._next_due is None:
+            self._next_due = sim_time
+        if sim_time + 1e-9 < self._next_due:
+            return
+        self._next_due += self._interval
+        snapshot = mujoco.MjData(self._model)
+        mujoco.mj_copyData(snapshot, self._model, sim.data)
+        try:
+            self._queue.put((snapshot, sim_time), timeout=1.0)
+        except queue.Full:
+            self._dropped += 1
+
+    def _work(self) -> None:
+        renderer = None
+        while True:
+            item = self._queue.get()
+            if item is None:
+                return
+            data, sim_time = item
+            try:
+                if renderer is None:
+                    # The OSMesa/EGL context is thread-local; create it here.
+                    renderer = mujoco.Renderer(
+                        self._model, height=self._panel, width=self._panel
+                    )
+                canvas = Image.new("RGB", self._canvas, "#111b24")
+                for offset, camera in enumerate(CAMERAS):
+                    renderer.update_scene(
+                        data, camera=camera, scene_option=self._scene_option
+                    )
+                    rgb = renderer.render()
+                    canvas.paste(
+                        Image.fromarray(rgb).convert("RGB"),
+                        (offset * self._panel, 28),
+                    )
+                draw = ImageDraw.Draw(canvas)
+                draw.text((10, 8), "EXTERNAL CAMERA", fill="white")
+                draw.text((self._panel + 10, 8), "WRIST CAMERA", fill="white")
+                draw.text(
+                    (10, self._panel + 36),
+                    f"simulation {sim_time:.2f}s | {self._fps} fps | motion only (API latency excluded)",
+                    fill="white",
+                )
+                self._process.stdin.write(canvas.tobytes())
+                self._captured += 1
+                self._last_sim_time = sim_time
+            except (BrokenPipeError, OSError, ValueError):
+                self._disabled = True
+                return
+            except Exception:
+                self._errors += 1
+                if self._errors >= _MAX_ERRORS:
+                    self._disabled = True
+                    return
+
+    def finalize(self) -> dict | None:
+        """Flush the encoder once; returns stats for the run log, or None."""
+        if self._finalized:
+            return None
+        self._finalized = True
+        # Drain any queued snapshot, then stop the worker.
+        try:
+            self._queue.put(None, timeout=2)
+        except queue.Full:
+            try:
+                self._queue.get_nowait()
+            except queue.Empty:
+                pass
+            try:
+                self._queue.put(None, timeout=2)
+            except queue.Full:
+                self._disabled = True
+        self._thread.join(timeout=30)
+        if self._process.stdin and not self._process.stdin.closed:
+            try:
+                self._process.stdin.close()
+            except (BrokenPipeError, OSError):
+                pass
+        try:
+            error = self._process.stderr.read().decode("utf-8", errors="replace")
+        except (OSError, ValueError):
+            error = ""
+        try:
+            returncode = self._process.wait(timeout=60)
+        except subprocess.TimeoutExpired:
+            self._process.kill()
+            self._process.wait()
+            returncode = -1
+        stats = {
+            "frames": self._captured,
+            "dropped": self._dropped,
+            "fps": self._fps,
+            "panel": self._panel,
+            "simulation_span_seconds": (
+                round(self._last_sim_time, 3) if self._captured else 0.0
+            ),
+        }
+        ok = (
+            returncode == 0
+            and self._captured > 0
+            and self._partial.is_file()
+            and self._partial.stat().st_size > 0
+        )
+        if ok:
+            self._partial.replace(self._destination)
+            stats["video"] = self._destination.name
+            stats["video_sha256"] = hashlib.sha256(
+                self._destination.read_bytes()
+            ).hexdigest()
+            stats["video_bytes"] = self._destination.stat().st_size
+        else:
+            self._partial.unlink(missing_ok=True)
+            stats["video"] = None
+            if error.strip():
+                stats["encoder_error"] = error.strip()[:300]
+        (self._destination.parent / "motion.json").write_text(
+            json.dumps(stats, ensure_ascii=False, indent=2), encoding="utf-8"
+        )
+        return stats
+
+    def abort(self) -> None:
+        """Kill the encoder without producing a video (runner teardown)."""
+        if self._finalized:
+            return
+        self._finalized = True
+        self._disabled = True
+        try:
+            self._queue.put_nowait(None)
+        except queue.Full:
+            pass
+        if self._process.poll() is None:
+            try:
+                self._process.stdin.close()
+            except (BrokenPipeError, OSError):
+                pass
+            try:
+                self._process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                self._process.kill()
+                self._process.wait()
+        self._partial.unlink(missing_ok=True)
diff --git a/src/maniloop/runtime/runner.py b/src/maniloop/runtime/runner.py
index 89e01d5..37cedcd 100644
--- a/src/maniloop/runtime/runner.py
+++ b/src/maniloop/runtime/runner.py
@@ -73,7 +73,7 @@ def __init__(
         self._last_wall = time.monotonic()
         self._physics_remainder = 0.0
         self.seed = 0
-        self.model = model or os.environ.get("OPENAI_MODEL") or "gpt-6-astra"
+        self.model = model or os.environ.get("OPENAI_MODEL") or DEFAULT_MODEL
         self.default_model = self.model
         self.manual_base_url = None
         self.key = os.environ.get("OPENAI_API_KEY", "")
@@ -88,7 +88,7 @@ def __init__(
             "source": "manual",
             "path": "",
             "provider": "OpenAI",
-            "base_url": "https://api.openai.com/v1",
+            "base_url": "https://openrouter.ai/api/v1",
             "model": self.model,
             "key_configured": bool(self.key),
             "models": [self.model],
@@ -133,6 +133,7 @@ def __init__(
         self.token = 0
         self.future_token = -1
         self.log_file = None
+        self.motion = None
         self.next_request = 0.0
         self.event("info", f"{sim.robot_name} 场景已准备；输入任务后开始闭环")
 
@@ -152,6 +153,7 @@ def event(self, kind, message, **details):
     def stop(self, message="任务已停止", reason="user_stop"):
         if not self.running and self.log_file and self.termination_reason:
             # A completed run is immutable even if Stop/Reset is sent again.
+            self._finalize_motion()
             self.update_review()
             self.sim.hold()
             return
@@ -166,6 +168,7 @@ def stop(self, message="任务已停止", reason="user_stop"):
             self.policy.cancel()
         self.sim.hold()
         self.event("info", message)
+        self._finalize_motion()
         self.update_review()
 
     def update_review(self):
@@ -679,6 +682,12 @@ def begin_episode(self, task, max_steps):
         (run / "manifest.json").write_text(
             json.dumps(self.manifest, ensure_ascii=False, indent=2), encoding="utf-8"
         )
+        self.motion = None
+        try:
+            from maniloop.recording.motion import MotionRecorder
+            self.motion = MotionRecorder(self.sim, run)
+        except Exception as exc:
+            self.event("info", f"运动视频录制不可用：{exc}")
         self.event(
             "start", "开始任务", task=task, model=self.model, max_calls=max_steps or None
         )
@@ -705,6 +714,7 @@ def advance(self, simulation_seconds=0.02):
             self.stop("环境已结束", reason="environment_terminated")
         self.tick()
         if not self.running and self.log_file:
+            self._finalize_motion()
             self.update_review()
         if self.paused:
             self._physics_remainder = 0.0
@@ -750,10 +760,31 @@ def advance(self, simulation_seconds=0.02):
                     if self.sim.backend in ("libero", "robosuite", "robocasa"):
                         break
             self.sim.step()
+            if self.motion is not None:
+                self.motion.tick(self.sim)
             if self.running and self.sim.terminated:
                 self.stop("环境已结束", reason="environment_terminated")
                 break
 
+    def _finalize_motion(self):
+        motion, self.motion = self.motion, None
+        if motion is None:
+            return
+        try:
+            stats = motion.finalize()
+        except Exception as exc:
+            self.event("info", f"运动视频封装失败：{exc}")
+            return
+        if stats is None:
+            return
+        if stats.get("video"):
+            message = f"运动视频已保存 {stats['video']}（{stats['frames']} 帧"
+            if stats.get("dropped"):
+                message += f"，渲染繁忙丢弃 {stats['dropped']} 帧"
+            self.event("info", message + "）")
+        else:
+            self.event("info", "运动视频未生成（无有效帧或编码失败）")
+
     def close_policy(self):
         policy, self.policy = self.policy, None
         if policy is not None and hasattr(policy, "close"):
@@ -765,6 +796,7 @@ def close(self):
         self.running = False
         self.chunk = None
         self.token += 1
+        self._finalize_motion()
         self.close_policy()
         self.pool.shutdown(wait=True, cancel_futures=True)
         self.sim.close()
diff --git a/src/maniloop/ui/chat.html b/src/maniloop/ui/chat.html
index ac2dfcc..e9991eb 100644
--- a/src/maniloop/ui/chat.html
+++ b/src/maniloop/ui/chat.html
@@ -13,7 +13,7 @@
 <main><aside class="panel"><div class="panel-title"><span><span class="number">01</span>连接配置</span><span class="hint">仅保留在本页</span></div>
 <div class="settings" id="settings">
 <div class="field"><label for="source">配置方式</label><select id="source"><option value="manual">手动填写 API 地址和 Key</option><option value="upload">选择 config.toml + Key</option><option value="file">读取本地配置 / CC Switch</option></select></div>
-<div id="manual-fields"><div class="field"><label for="base-url">API 地址（Base URL）</label><input id="base-url" type="url" value="https://api.openai.com/v1" autocomplete="off" spellcheck="false"><p class="hint">填写服务商提供的基础地址，不要附加 /responses 或 /chat/completions。</p></div></div>
+<div id="manual-fields"><div class="field"><label for="base-url">API 地址（Base URL）</label><input id="base-url" type="url" value="https://openrouter.ai/api/v1" autocomplete="off" spellcheck="false"><p class="hint">填写服务商提供的基础地址，不要附加 /responses 或 /chat/completions。</p></div></div>
 <div id="upload-fields" hidden><div class="field"><label for="config-upload">选择 config.toml</label><input id="config-upload" type="file" accept=".toml,text/plain"></div><div class="field"><label for="config-toml">或粘贴 TOML 内容</label><textarea id="config-toml" rows="5" spellcheck="false" autocomplete="off" placeholder="model = &quot;你的模型名称&quot;"></textarea></div></div>
 <div id="file-fields" hidden><div class="field"><label for="config-list">已发现的配置</label><select id="config-list"><option value="">自定义路径…</option></select><button class="button small subtle" id="discover-configs" type="button" style="margin-top:8px">发现本地配置</button></div><div class="field"><label for="config-path">本地文件路径</label><input id="config-path" autocomplete="off" spellcheck="false"><p class="hint">点击发现后查找 Codex / CC Switch 路径；读取配置或发送时才读取所选凭据。</p></div></div>
 <div class="field"><label for="api-key">API Key</label><input id="api-key" type="password" autocomplete="off" spellcheck="false" placeholder="输入密钥；配置已有 Key 时可留空"><p class="hint">不会写入浏览器存储、聊天记录或仓库。</p></div>
diff --git a/src/maniloop/ui/index.html b/src/maniloop/ui/index.html
index 431adbf..1859c06 100644
--- a/src/maniloop/ui/index.html
+++ b/src/maniloop/ui/index.html
@@ -129,7 +129,7 @@
               </div>
               <div id="manual-config-fields">
                 <div class="key-row"><label class="field-label" for="api-key">API 密钥</label><input id="api-key" class="input mono" type="password" placeholder="输入 API 密钥，或使用服务端环境变量" autocomplete="new-password" spellcheck="false" autocapitalize="none" maxlength="1000"></div>
-                <div class="key-row"><label class="field-label" for="base-url">API 地址<span>可选，留空使用服务默认地址</span></label><input id="base-url" class="input model-input" type="url" placeholder="https://api.openai.com/v1" spellcheck="false" autocapitalize="none" maxlength="2048"></div>
+                <div class="key-row"><label class="field-label" for="base-url">API 地址<span>可选，留空使用服务默认地址</span></label><input id="base-url" class="input model-input" type="url" placeholder="https://openrouter.ai/api/v1" spellcheck="false" autocapitalize="none" maxlength="2048"></div>
                 <p class="key-hint">开始提交后清空密钥输入；不写入浏览器存储或实验日志。</p>
               </div>
             </div>
@@ -804,7 +804,7 @@
         $('agent-select').dispatchEvent(new Event('change'));
         updateBackendFields();
         ui.modelSelect.value = '__custom__';
-        ui.model.value = 'gpt-6-astra';
+        ui.model.value = 'openai/gpt-6-astra';
         updateModelInput();
         ui.maxSteps.value = '30';
         $('request-timeout').value = '120';
diff --git a/src/maniloop/ui/workspace.html b/src/maniloop/ui/workspace.html
index 9a3cfd9..79e61b2 100644
--- a/src/maniloop/ui/workspace.html
+++ b/src/maniloop/ui/workspace.html
@@ -158,7 +158,7 @@
               </div>
               <div id="manual-config-fields">
                 <div class="key-row"><label class="field-label" for="api-key">API 密钥</label><input id="api-key" class="input mono" type="password" placeholder="输入 API 密钥，或使用服务端环境变量" autocomplete="new-password" spellcheck="false" autocapitalize="none" maxlength="1000"></div>
-                <div class="key-row"><label class="field-label" for="base-url">API 地址<span>可选，留空使用服务默认地址</span></label><input id="base-url" class="input model-input" type="url" placeholder="https://api.openai.com/v1" spellcheck="false" autocapitalize="none" maxlength="2048"></div>
+                <div class="key-row"><label class="field-label" for="base-url">API 地址<span>可选，留空使用服务默认地址</span></label><input id="base-url" class="input model-input" type="url" placeholder="https://openrouter.ai/api/v1" spellcheck="false" autocapitalize="none" maxlength="2048"></div>
                 <p class="key-hint">开始提交后清空密钥输入；不写入浏览器存储或实验日志。</p>
               </div>
             </div>
diff --git a/src/maniloop/ui/workspace.js b/src/maniloop/ui/workspace.js
index 42a821f..16bc4fc 100644
--- a/src/maniloop/ui/workspace.js
+++ b/src/maniloop/ui/workspace.js
@@ -583,7 +583,7 @@ $('demo-preset').addEventListener('click', async () => {
     $('agent-select').dispatchEvent(new Event('change'));
     updateBackendFields();
     ui.modelSelect.value = '__custom__';
-    ui.model.value = 'gpt-6-astra';
+    ui.model.value = 'openai/gpt-6-astra';
     updateModelInput();
     ui.maxSteps.value = '30';
     $('request-timeout').value = '120';

From 41ff5a9c29ce11d8d03b1a15e6cff955eb81ad53 Mon Sep 17 00:00:00 2001
From: liuzhen <liuzhen@localhost>
Date: Thu, 22 Oct 2026 02:46:05 +0000
Subject: [PATCH 2/2] =?UTF-8?q?=E6=8E=A5=E5=85=A5=E5=88=86=E5=B1=82?=
 =?UTF-8?q?=E5=A4=A7=E8=84=91=20Harness=EF=BC=9A=E7=9B=AE=E6=A0=87?=
 =?UTF-8?q?=E5=88=86=E8=A7=A3=E3=80=81=E8=BD=BB=E9=87=8F=E4=B8=89=E7=BA=A7?=
 =?UTF-8?q?=E9=AA=8C=E8=AF=81=E4=B8=8E=E6=9C=89=E7=95=8C=E5=8F=8D=E6=80=9D?=
 =?UTF-8?q?=E9=87=8D=E8=A7=84=E5=88=92?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 新增 maniloop/harness 包（抽象自 Goal2Skill 上层大脑，剔除任务特化）：
  types/memory/planner/verifier/reflector/bridge/router/vlm/agent
- 双时标回路：VLM 规划为语义子任务（终态可视觉验证），执行器逐步执行；
  L0 运动规则门控 + L1 执行器自报 PHASE 行 + L2 仅在宣称完成时 VLM 确认，
  验证成本与子任务数同量级而非与决策步数同量级
- 失败经独立 reflector 诊断（retry/adjust-param/replan），
  前缀保留式重规划；尝试/重规划/回退次数均有硬上限
- HarnessAgent 实现标准 Agent 协议，runner 主循环零改动；
  策略切换/暂停单步/密钥刷新路径均已兼容
- UI 策略下拉新增「分层大脑 Harness」，复用云端 LLM 凭证与模型选择
- 不包含任何 API key；角色模型默认跟随当前连接模型，可经 harness_roles 覆盖

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 src/maniloop/harness/__init__.py  |  21 +++
 src/maniloop/harness/agent.py     | 199 ++++++++++++++++++++++++
 src/maniloop/harness/bridge.py    | 248 ++++++++++++++++++++++++++++++
 src/maniloop/harness/memory.py    | 200 ++++++++++++++++++++++++
 src/maniloop/harness/planner.py   | 116 ++++++++++++++
 src/maniloop/harness/reflector.py |  86 +++++++++++
 src/maniloop/harness/router.py    |  20 +++
 src/maniloop/harness/types.py     | 105 +++++++++++++
 src/maniloop/harness/verifier.py  | 137 +++++++++++++++++
 src/maniloop/harness/vlm.py       | 134 ++++++++++++++++
 src/maniloop/runtime/runner.py    |  25 ++-
 src/maniloop/ui/workspace.html    |   2 +-
 src/maniloop/ui/workspace.js      |  25 +--
 13 files changed, 1300 insertions(+), 18 deletions(-)
 create mode 100644 src/maniloop/harness/__init__.py
 create mode 100644 src/maniloop/harness/agent.py
 create mode 100644 src/maniloop/harness/bridge.py
 create mode 100644 src/maniloop/harness/memory.py
 create mode 100644 src/maniloop/harness/planner.py
 create mode 100644 src/maniloop/harness/reflector.py
 create mode 100644 src/maniloop/harness/router.py
 create mode 100644 src/maniloop/harness/types.py
 create mode 100644 src/maniloop/harness/verifier.py
 create mode 100644 src/maniloop/harness/vlm.py

diff --git a/src/maniloop/harness/__init__.py b/src/maniloop/harness/__init__.py
new file mode 100644
index 0000000..d3f761c
--- /dev/null
+++ b/src/maniloop/harness/__init__.py
@@ -0,0 +1,21 @@
+"""Layered upper-brain harness: plan, verify, reflect, remember."""
+
+from .agent import HarnessAgent
+from .types import (
+    DEFAULT_LIMITS,
+    BridgeStepOutcome,
+    MemoryState,
+    ReflectionResult,
+    SubTask,
+    VerifyResult,
+)
+
+__all__ = [
+    "HarnessAgent",
+    "SubTask",
+    "VerifyResult",
+    "ReflectionResult",
+    "MemoryState",
+    "BridgeStepOutcome",
+    "DEFAULT_LIMITS",
+]
diff --git a/src/maniloop/harness/agent.py b/src/maniloop/harness/agent.py
new file mode 100644
index 0000000..a74927d
--- /dev/null
+++ b/src/maniloop/harness/agent.py
@@ -0,0 +1,199 @@
+"""HarnessAgent: the layered upper brain as a standard ManiLoop agent.
+
+Wraps any executor implementing the Agent protocol (today: the cloud LLM
+agent). The slow loop plans the goal into semantic subtasks, gates progress
+with L0 motion rules and the executor's L1 self-report, confirms claimed
+completions with an L2 VLM check, and recovers through bounded reflection.
+The runner only sees decide() -> action, so no runner loop changes are needed.
+"""
+
+from __future__ import annotations
+
+import re
+import time
+
+from .bridge import PlanBridge
+from .memory import MemoryManager
+from .planner import SingleTaskPlanner, VLMPlanner
+from .reflector import RuleReflector, VLMReflector
+from .router import RoleRouter
+from .types import DEFAULT_LIMITS
+from .verifier import ProgressGate, RuleVerifier, VLMVerifier
+from .vlm import HarnessVLM
+
+_PHASE_RE = re.compile(r"PHASE:\s*(working|done|stuck)", re.IGNORECASE)
+
+
+def _scene_summary(observation: dict) -> dict:
+    return {
+        "tcp_position": observation.get("tcp_position"),
+        "gripper_opening": observation.get("gripper_opening"),
+        "cameras": list((observation.get("cameras") or {}).keys())
+        if isinstance(observation.get("cameras"), dict)
+        else observation.get("cameras"),
+        "robot_description": str(observation.get("robot_description") or "")[:400],
+        "action_limits": observation.get("action_limits"),
+    }
+
+
+def _subtask_prompt(subtask, goal: str, progress: str) -> str:
+    postconditions = "; ".join(subtask.postconditions) or subtask.instruction
+    prompt = (
+        f"{subtask.instruction}\n\n"
+        f"[Harness protocol] Overall goal: {goal}\n"
+        f"This is subtask {progress} decomposed from the goal; complete ONLY "
+        f"this subtask, then stop moving.\n"
+        f"Done when ALL hold: {postconditions}.\n"
+        f"End your explanation with exactly one status line:\n"
+        f"PHASE: working | done | stuck\n"
+        f"- working: still executing this subtask\n"
+        f"- done: the postconditions are visibly satisfied\n"
+        f"- stuck: you cannot make progress toward them"
+    )
+    if subtask.skill_params:
+        import json
+
+        prompt += f"\nParameters: {json.dumps(subtask.skill_params, ensure_ascii=False)}"
+    return prompt
+
+
+def _split_phase_status(explanation: str) -> tuple[str, str]:
+    match = _PHASE_RE.search(explanation or "")
+    if not match:
+        return "working", explanation
+    cleaned = (explanation[: match.start()] + explanation[match.end() :]).strip()
+    return match.group(1).lower(), cleaned
+
+
+class HarnessAgent:
+    def __init__(self, executor, *, roles: dict | None = None, limits: dict | None = None):
+        self.executor = executor
+        provider = executor.provider  # LLMAgent -> GPTPolicy (owns client+model)
+        self.router = RoleRouter(provider.model, roles)
+        self.limits = {**DEFAULT_LIMITS, **(limits or {})}
+        self._vlms = {
+            role: HarnessVLM(provider.client, self.router.model_for(role))
+            for role in ("planner", "verifier", "reflector")
+        }
+        planner = VLMPlanner(self._vlms["planner"], fallback=SingleTaskPlanner())
+        verifier = VLMVerifier(self._vlms["verifier"], fallback=RuleVerifier())
+        reflector = VLMReflector(self._vlms["reflector"], fallback=RuleReflector())
+        self._roles = (planner, verifier, reflector)
+        self.memory = MemoryManager()
+        self.bridge = PlanBridge(self.memory, planner, verifier, reflector, self.limits)
+        self.gate = ProgressGate(
+            window=int(self.limits["stall_window"]),
+            translation_m=float(self.limits["stall_translation_m"]),
+            gripper_delta=float(self.limits["stall_gripper_delta"]),
+        )
+        self._planned = False
+        self._phase_started = 0.0
+        self.last_usage: dict = {}
+        self.last_latency = 0.0
+        self.last_decision: dict | None = None
+
+    # ------------------------------------------------------ Agent protocol
+    @property
+    def request_options(self):
+        return self.executor.request_options
+
+    @property
+    def last_response_id(self):
+        return self.executor.last_response_id
+
+    def reset(self, seed: int = 0) -> None:
+        self.executor.reset(seed)
+        self.memory = MemoryManager()
+        planner, verifier, reflector = self._roles
+        self.bridge = PlanBridge(self.memory, planner, verifier, reflector, self.limits)
+        self.gate.reset()
+        self._planned = False
+        self._phase_started = 0.0
+        self.last_usage = {}
+        self.last_latency = 0.0
+        self.last_decision = None
+
+    def decide(self, task, observation, images, history, geometry_results):
+        started = time.monotonic()
+        try:
+            if not self._planned:
+                self.memory.begin(task)
+                plan = self.bridge.planner.plan(
+                    task, _scene_summary(observation), self.memory.export_for_planner()
+                )
+                self.bridge.set_plan(plan)
+                self._planned = True
+                self._phase_started = started
+            else:
+                outcome = self.bridge.review(
+                    observation=observation,
+                    images=images,
+                    elapsed_s=started - self._phase_started,
+                    gate=self.gate,
+                    scene=_scene_summary(observation),
+                )
+                if outcome.phase_restarted:
+                    self._phase_started = started
+                    self.gate.reset()
+                if outcome.episode_done or outcome.aborted:
+                    return self._terminal_action(
+                        observation,
+                        outcome.notice or "分层控制结束",
+                    )
+            subtask = self.bridge.current_subtask()
+            if subtask is None:
+                return self._terminal_action(observation, "计划已全部完成")
+            action = self.executor.decide(
+                _subtask_prompt(subtask, task, self.bridge.progress_label()),
+                observation,
+                images,
+                history,
+                geometry_results,
+            )
+            status, explanation = _split_phase_status(str(action.get("explanation", "")))
+            action["explanation"] = (
+                f"[Harness {self.bridge.progress_label()} · {subtask.instruction[:40]}] "
+                f"{explanation}"
+            )
+            self.bridge.note_execution(subtask, action, status)
+            self.gate.update(observation)
+            return action
+        finally:
+            self.last_latency = time.monotonic() - started
+            self.last_usage = self._merged_usage()
+            self.last_decision = self.memory.snapshot()
+
+    def close(self) -> None:
+        # The role VLMs share the executor's client; closing the executor
+        # closes the transport for everyone. Mirror the runner's close_policy
+        # fallback: not every executor exposes close() itself.
+        close = getattr(self.executor, "close", None)
+        if callable(close):
+            close()
+            return
+        client = getattr(self.executor, "client", None)
+        if client is not None:
+            client.close()
+
+    # ----------------------------------------------------------------- parts
+    def _terminal_action(self, observation: dict, message: str) -> dict:
+        gripper = observation.get("gripper_opening")
+        return {
+            "observation_id": observation["observation_id"],
+            "kind": "done",
+            "delta_position": [0.0, 0.0, 0.0],
+            "delta_rotation": [0.0, 0.0, 0.0],
+            "gripper_opening": float(gripper) if type(gripper) in (int, float) else 1.0,
+            "pixel": [0, 0],
+            "camera": "",
+            "explanation": f"[Harness] {message}",
+        }
+
+    def _merged_usage(self) -> dict:
+        merged: dict[str, int] = {}
+        sources = [self.executor.last_usage] + [vlm.last_usage for vlm in self._vlms.values()]
+        for usage in sources:
+            for key, count in (usage or {}).items():
+                if type(count) is int:
+                    merged[key] = merged.get(key, 0) + count
+        return merged
diff --git a/src/maniloop/harness/bridge.py b/src/maniloop/harness/bridge.py
new file mode 100644
index 0000000..425a94a
--- /dev/null
+++ b/src/maniloop/harness/bridge.py
@@ -0,0 +1,248 @@
+"""Plan cursor state machine: advance, retry, rewind, or replan.
+
+The bridge owns the slow-loop decision after each executed chunk: L0 rules
+(timeout, stall) first, then the executor's L1 self-report, and only a claimed
+completion triggers the L2 VLM confirm. Reflection is bounded by attempt,
+replan and rewind budgets from the shared limits.
+"""
+
+from __future__ import annotations
+
+from .memory import MemoryManager
+from .planner import Planner
+from .reflector import Reflector
+from .types import (
+    DEFAULT_LIMITS,
+    BridgeStepOutcome,
+    PhaseStatus,
+    ReflectionResult,
+    SubTask,
+    VerifyResult,
+)
+from .verifier import ProgressGate, Verifier
+
+
+class PlanBridge:
+    def __init__(
+        self,
+        memory: MemoryManager,
+        planner: Planner,
+        verifier: Verifier,
+        reflector: Reflector,
+        limits: dict | None = None,
+    ):
+        self.memory = memory
+        self.planner = planner
+        self.verifier = verifier
+        self.reflector = reflector
+        self.limits = {**DEFAULT_LIMITS, **(limits or {})}
+        self._pending_status: PhaseStatus = "working"
+        self._pending_note = ""
+
+    # ------------------------------------------------------------------ plan
+    def set_plan(self, plan: list[SubTask]) -> None:
+        self.memory.set_plan(plan)
+        self._pending_status = "working"
+
+    def current_subtask(self) -> SubTask | None:
+        return self.memory.current_subtask()
+
+    def progress_label(self) -> str:
+        state = self.memory.state
+        total = len(state.current_plan)
+        return f"{min(state.plan_index + 1, total)}/{total}" if total else "0/0"
+
+    def note_execution(self, subtask: SubTask, action: dict, phase_status: str) -> None:
+        """L1: file the executor's self-report; judged on the next decide()."""
+        if phase_status in ("working", "done", "stuck"):
+            self._pending_status = phase_status
+        else:
+            self._pending_status = "working"
+        self.memory.record_execution(
+            subtask,
+            str(action.get("kind", "?")),
+            str(action.get("explanation", "")),
+            self._pending_status,
+        )
+
+    # ---------------------------------------------------------------- review
+    def review(
+        self,
+        *,
+        observation: dict,
+        images: dict[str, bytes],
+        elapsed_s: float,
+        gate: ProgressGate,
+        scene: dict,
+    ) -> BridgeStepOutcome:
+        subtask = self.memory.current_subtask()
+        if subtask is None:
+            return self._outcome(None, None, episode_done=True, notice="计划已全部完成")
+
+        verify = self._judge(subtask, images=images, elapsed_s=elapsed_s, gate=gate)
+        if verify.completion == "running":
+            return self._outcome(verify, None)
+
+        attempt = self.memory.increment_attempt(subtask.subtask_id)
+        self.memory.record_verification(subtask, verify, attempt=attempt)
+
+        if verify.completion == "success":
+            if not verify.prior_phase_intact:
+                return self._rewind(verify)
+            self.memory.advance()
+            self._pending_status = "working"
+            following = self.memory.current_subtask()
+            return self._outcome(
+                verify,
+                None,
+                advanced=True,
+                episode_done=following is None,
+                current=following,
+                notice=f"子任务完成（{verify.source}）：{verify.evidence[:120]}",
+                phase_restarted=True,
+            )
+
+        # fail / timeout -> independent reflection, bounded by budgets.
+        reflection = self.reflector.reflect(
+            subtask, verify, self.memory.export_for_reflector()
+        )
+        if (
+            reflection.recovery == "retry"
+            and attempt >= self.limits["max_attempts_per_subtask"]
+        ):
+            reflection = ReflectionResult(
+                diagnosis=reflection.diagnosis + "；重试次数用尽",
+                recovery="replan",
+                parameter_updates=reflection.parameter_updates,
+            )
+        self.memory.record_error(subtask, verify, reflection, attempt)
+        return self._recover(subtask, verify, reflection, scene)
+
+    # ----------------------------------------------------------------- parts
+    def _judge(
+        self,
+        subtask: SubTask,
+        *,
+        images: dict[str, bytes],
+        elapsed_s: float,
+        gate: ProgressGate,
+    ) -> VerifyResult:
+        """L0 rules -> L1 self-report -> L2 VLM confirm, cheapest first."""
+        if elapsed_s > subtask.timeout_s:
+            return VerifyResult(
+                completion="timeout",
+                confidence=1.0,
+                evidence=f"子任务超时 {elapsed_s:.0f}s > 预算 {subtask.timeout_s:.0f}s",
+                source="rule",
+            )
+        if gate.stalled:
+            return VerifyResult(
+                completion="fail",
+                confidence=0.6,
+                evidence=f"连续 {self.limits['stall_window']} 次决策无可见运动进展",
+                source="rule",
+            )
+        status = self._pending_status
+        self._pending_status = "working"
+        if status == "working":
+            return VerifyResult(
+                completion="running", confidence=1.0, evidence="执行器仍在推进", source="self_report"
+            )
+        if status == "stuck":
+            return VerifyResult(
+                completion="fail",
+                confidence=0.7,
+                evidence="执行器报告当前子任务无法继续",
+                source="self_report",
+            )
+        # status == "done": the only moment that spends an L2 VLM confirm.
+        return self.verifier.confirm(
+            subtask, images, self.memory.export_for_verifier()
+        )
+
+    def _rewind(self, verify: VerifyResult) -> BridgeStepOutcome:
+        if self.memory.state.rewind_count >= self.limits["max_rewinds"]:
+            return self._outcome(
+                verify,
+                None,
+                aborted=True,
+                notice="先前阶段被推翻且回退次数已用尽，中止",
+            )
+        self.memory.revoke_last()
+        self._pending_status = "working"
+        return self._outcome(
+            verify,
+            None,
+            current=self.memory.current_subtask(),
+            notice=f"先前阶段被视觉证据推翻，回退重做：{verify.evidence[:120]}",
+            phase_restarted=True,
+        )
+
+    def _recover(
+        self,
+        subtask: SubTask,
+        verify: VerifyResult,
+        reflection: ReflectionResult,
+        scene: dict,
+    ) -> BridgeStepOutcome:
+        if reflection.recovery in ("retry", "adjust-param"):
+            if reflection.recovery == "adjust-param" and reflection.parameter_updates:
+                subtask.skill_params.update(reflection.parameter_updates)
+            self._pending_status = "working"
+            return self._outcome(
+                verify,
+                reflection,
+                current=subtask,
+                notice=f"{reflection.recovery}：{reflection.diagnosis[:120]}",
+                phase_restarted=True,
+            )
+        # replan
+        state = self.memory.state
+        if state.replan_count >= self.limits["max_replans"]:
+            return self._outcome(
+                verify,
+                reflection,
+                aborted=True,
+                notice=f"重规划次数已用尽，中止：{reflection.diagnosis[:120]}",
+            )
+        suffix = self.planner.replan(
+            state.current_goal, scene, self.memory.export_for_planner()
+        )
+        state.replan_count += 1
+        self.memory.set_plan(suffix, keep_prefix=True)
+        self._pending_status = "working"
+        return self._outcome(
+            verify,
+            reflection,
+            replanned=True,
+            current=self.memory.current_subtask(),
+            notice=f"已重规划剩余步骤：{reflection.diagnosis[:120]}",
+            phase_restarted=True,
+        )
+
+    def _outcome(
+        self,
+        verify: VerifyResult | None,
+        reflection: ReflectionResult | None,
+        *,
+        episode_done: bool = False,
+        aborted: bool = False,
+        replanned: bool = False,
+        advanced: bool = False,
+        current: SubTask | None = None,
+        notice: str = "",
+        phase_restarted: bool = False,
+    ) -> BridgeStepOutcome:
+        if current is None and not (episode_done or aborted):
+            current = self.memory.current_subtask()
+        return BridgeStepOutcome(
+            verify=verify,
+            reflection=reflection,
+            episode_done=episode_done,
+            aborted=aborted,
+            replanned=replanned,
+            advanced=advanced,
+            current_subtask=current,
+            notice=notice,
+            phase_restarted=phase_restarted,
+        )
diff --git a/src/maniloop/harness/memory.py b/src/maniloop/harness/memory.py
new file mode 100644
index 0000000..7346d0f
--- /dev/null
+++ b/src/maniloop/harness/memory.py
@@ -0,0 +1,200 @@
+"""Episode memory for the harness: one fact source, per-role minimal views."""
+
+from __future__ import annotations
+
+from .types import (
+    ErrorRecord,
+    ExecutionRecord,
+    MemoryState,
+    ReflectionResult,
+    SubTask,
+    VerifyResult,
+)
+
+
+class MemoryManager:
+    def __init__(self, history_window: int = 12):
+        self.state = MemoryState()
+        self.history_window = history_window
+
+    def begin(self, goal: str) -> None:
+        self.state.current_goal = goal
+
+    def set_plan(self, plan: list[SubTask], *, keep_prefix: bool = False) -> None:
+        if keep_prefix:
+            prefix = self.state.current_plan[: self.state.plan_index]
+            taken = {s.subtask_id for s in prefix}
+            suffix = []
+            for item in plan:
+                # Attempt counters key on subtask_id, so suffix ids must not
+                # collide with the preserved prefix (fallback planners reuse
+                # plain ids like "s1").
+                base, serial, candidate = item.subtask_id, 2, item.subtask_id
+                while candidate in taken:
+                    candidate = f"{base}~{serial}"
+                    serial += 1
+                if candidate != item.subtask_id:
+                    item = SubTask(
+                        subtask_id=candidate,
+                        instruction=item.instruction,
+                        postconditions=item.postconditions,
+                        timeout_s=item.timeout_s,
+                        skill_params=item.skill_params,
+                    )
+                taken.add(candidate)
+                suffix.append(item)
+            self.state.current_plan = prefix + suffix
+        else:
+            self.state.current_plan = list(plan)
+            self.state.plan_index = 0
+            self.state.completed_steps = []
+
+    def current_subtask(self) -> SubTask | None:
+        plan = self.state.current_plan
+        if 0 <= self.state.plan_index < len(plan):
+            return plan[self.state.plan_index]
+        return None
+
+    def advance(self) -> SubTask | None:
+        current = self.current_subtask()
+        if current is not None:
+            self.state.completed_steps.append(current.subtask_id)
+            self.state.attempt_counter.pop(current.subtask_id, None)
+        self.state.plan_index += 1
+        return self.current_subtask()
+
+    def revoke_last(self) -> None:
+        """Rewind the cursor one phase; the revoked phase must be redone."""
+        if self.state.completed_steps:
+            self.state.completed_steps.pop()
+        self.state.plan_index = max(0, self.state.plan_index - 1)
+        self.state.rewind_count += 1
+
+    def increment_attempt(self, subtask_id: str) -> int:
+        count = self.state.attempt_counter.get(subtask_id, 0) + 1
+        self.state.attempt_counter[subtask_id] = count
+        return count
+
+    def attempts(self, subtask_id: str) -> int:
+        return self.state.attempt_counter.get(subtask_id, 0)
+
+    def record_execution(
+        self, subtask: SubTask, action_kind: str, explanation: str, phase_status: str
+    ) -> None:
+        self.state.episodic_history.append(
+            ExecutionRecord(
+                subtask_id=subtask.subtask_id,
+                instruction=subtask.instruction,
+                action_kind=action_kind,
+                explanation=explanation[:400],
+                phase_status=phase_status,
+            )
+        )
+        del self.state.episodic_history[: -self.history_window]
+
+    def record_verification(
+        self, subtask: SubTask, verify: VerifyResult, *, attempt: int
+    ) -> None:
+        self.state.evidence_chain.append(
+            {
+                "subtask_id": subtask.subtask_id,
+                "attempt": attempt,
+                "completion": verify.completion,
+                "confidence": round(verify.confidence, 3),
+                "source": verify.source,
+                "evidence": verify.evidence[:400],
+            }
+        )
+
+    def record_error(
+        self, subtask: SubTask, verify: VerifyResult, reflection: ReflectionResult, attempt: int
+    ) -> None:
+        self.state.error_register.append(
+            ErrorRecord(
+                subtask_id=subtask.subtask_id,
+                attempt_idx=attempt,
+                completion=verify.completion,
+                diagnosis=reflection.diagnosis[:400],
+                recovery=reflection.recovery,
+            )
+        )
+
+    def export_for_planner(self) -> dict:
+        state = self.state
+        return {
+            "goal": state.current_goal,
+            "completed": [
+                {"id": s.subtask_id, "instruction": s.instruction}
+                for s in state.current_plan[: state.plan_index]
+            ],
+            "recent_errors": [vars(e) for e in state.error_register[-4:]],
+            "recent_history": [vars(r) for r in state.episodic_history[-6:]],
+            "replans_so_far": state.replan_count,
+        }
+
+    def export_for_verifier(self) -> dict:
+        state = self.state
+        current = self.current_subtask()
+        return {
+            "goal": state.current_goal,
+            "current_subtask": (
+                {
+                    "id": current.subtask_id,
+                    "instruction": current.instruction,
+                    "postconditions": current.postconditions,
+                }
+                if current
+                else None
+            ),
+            "completed_instructions": [
+                s.instruction for s in state.current_plan[: state.plan_index]
+            ],
+            "attempt": self.attempts(current.subtask_id) if current else 0,
+        }
+
+    def export_for_reflector(self) -> dict:
+        state = self.state
+        current = self.current_subtask()
+        return {
+            "goal": state.current_goal,
+            "current_subtask": (
+                {
+                    "id": current.subtask_id,
+                    "instruction": current.instruction,
+                    "postconditions": current.postconditions,
+                    "skill_params": current.skill_params,
+                }
+                if current
+                else None
+            ),
+            "attempt": self.attempts(current.subtask_id) if current else 0,
+            "recent_errors": [vars(e) for e in state.error_register[-4:]],
+            "remaining_instructions": [
+                s.instruction for s in state.current_plan[state.plan_index + 1 :]
+            ],
+            "replans_so_far": state.replan_count,
+        }
+
+    def snapshot(self) -> dict:
+        state = self.state
+        current = self.current_subtask()
+        return {
+            "goal": state.current_goal,
+            "plan": [
+                {
+                    "id": s.subtask_id,
+                    "instruction": s.instruction,
+                    "status": (
+                        "done"
+                        if i < state.plan_index
+                        else "active" if i == state.plan_index else "pending"
+                    ),
+                }
+                for i, s in enumerate(state.current_plan)
+            ],
+            "current": current.subtask_id if current else None,
+            "replans": state.replan_count,
+            "rewinds": state.rewind_count,
+            "errors": len(state.error_register),
+            "evidence": state.evidence_chain[-6:],
+        }
diff --git a/src/maniloop/harness/planner.py b/src/maniloop/harness/planner.py
new file mode 100644
index 0000000..540240e
--- /dev/null
+++ b/src/maniloop/harness/planner.py
@@ -0,0 +1,116 @@
+"""Planner role: goal -> linear list of semantic subtasks, plus replanning."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from .types import SubTask
+from .vlm import HarnessVLM, HarnessVLMError
+
+PLAN_INSTRUCTIONS = """You are the high-level planner of a robot manipulation system.
+Decompose the goal into a short linear sequence of semantic subtasks.
+
+Rules:
+- Each subtask ends at a visually checkable state change (an object grasped,
+  placed, a door opened), never at a low-level motion like "move 3 cm".
+- 2 to 6 subtasks for typical tabletop tasks. Do not over-decompose.
+- The executor is a vision-language policy that receives your instruction
+  verbatim and controls the arm step by step; write instructions it can ground
+  directly from camera images (name objects by color/shape/position).
+- postconditions list what must be visibly true when the subtask completes;
+  a verifier that only sees camera images will judge exactly these.
+- timeout_s is the wall-clock budget for this subtask (60-300).
+- Reply with JSON only: {"subtasks": [{"subtask_id": "s1", "instruction":
+  "...", "postconditions": ["..."], "timeout_s": 120}]}."""
+
+REPLAN_INSTRUCTIONS = """You are the high-level planner of a robot manipulation system.
+The previous plan failed partway. The completed prefix is preserved; write a
+REPLACEMENT SUFFIX for the remaining work only.
+
+Rules:
+- Do not repeat any completed instruction; the suffix continues from the
+  current scene state shown in the images.
+- Same granularity and output contract as an initial plan; subtask ids must be
+  fresh (r1_s1, r1_s2, ...), 1 to 5 subtasks.
+- Reply with JSON only: {"subtasks": [...]}."""
+
+
+class Planner(ABC):
+    @abstractmethod
+    def plan(self, goal: str, scene: dict, context: dict) -> list[SubTask]:
+        raise NotImplementedError
+
+    def replan(self, goal: str, scene: dict, context: dict) -> list[SubTask]:
+        return self.plan(goal, scene, context)
+
+
+class SingleTaskPlanner(Planner):
+    """Offline envelope: the whole goal as one subtask (degraded but safe)."""
+
+    def plan(self, goal: str, scene: dict, context: dict) -> list[SubTask]:
+        del scene, context
+        return [
+            SubTask(
+                subtask_id="s1",
+                instruction=goal,
+                postconditions=["The goal is visibly achieved"],
+                timeout_s=600.0,
+            )
+        ]
+
+
+def _parse_subtasks(data: dict, *, id_prefix: str = "s") -> list[SubTask]:
+    raw = data.get("subtasks")
+    if not isinstance(raw, list):
+        raise HarnessVLMError("planner: 响应缺少 subtasks 列表")
+    subtasks: list[SubTask] = []
+    for i, item in enumerate(raw[:8]):
+        if not isinstance(item, dict):
+            continue
+        instruction = item.get("instruction")
+        if not isinstance(instruction, str) or not instruction.strip():
+            continue
+        postconditions = item.get("postconditions")
+        if not isinstance(postconditions, list):
+            postconditions = [str(postconditions)] if postconditions else []
+        try:
+            timeout_s = float(item.get("timeout_s", 120.0))
+        except (TypeError, ValueError):
+            timeout_s = 120.0
+        timeout_s = min(max(timeout_s, 30.0), 600.0)
+        subtasks.append(
+            SubTask(
+                subtask_id=str(item.get("subtask_id") or f"{id_prefix}{i + 1}")[:40],
+                instruction=instruction.strip()[:300],
+                postconditions=[str(p)[:200] for p in postconditions[:4]],
+                timeout_s=timeout_s,
+            )
+        )
+    if not subtasks:
+        raise HarnessVLMError("planner: 未解析出任何有效子任务")
+    return subtasks
+
+
+class VLMPlanner(Planner):
+    def __init__(self, vlm: HarnessVLM, fallback: Planner | None = None):
+        self.vlm = vlm
+        self.fallback = fallback or SingleTaskPlanner()
+
+    def plan(self, goal: str, scene: dict, context: dict) -> list[SubTask]:
+        try:
+            data = self.vlm.query_json(
+                PLAN_INSTRUCTIONS, {"goal": goal, "scene": scene, "memory": context}
+            )
+            return _parse_subtasks(data)
+        except HarnessVLMError:
+            return self.fallback.plan(goal, scene, context)
+
+    def replan(self, goal: str, scene: dict, context: dict) -> list[SubTask]:
+        try:
+            data = self.vlm.query_json(
+                REPLAN_INSTRUCTIONS,
+                {"goal": goal, "scene": scene, "memory": context},
+            )
+            return _parse_subtasks(data, id_prefix=f"r{context.get('replans_so_far', 0) + 1}_s")
+        except HarnessVLMError:
+            return self.fallback.plan(goal, scene, context)
diff --git a/src/maniloop/harness/reflector.py b/src/maniloop/harness/reflector.py
new file mode 100644
index 0000000..45d0174
--- /dev/null
+++ b/src/maniloop/harness/reflector.py
@@ -0,0 +1,86 @@
+"""Reflector role: independent failure diagnosis, separate from the verifier."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from .types import ReflectionResult, SubTask, VerifyResult
+from .vlm import HarnessVLM, HarnessVLMError
+
+REFLECT_INSTRUCTIONS = """You are the failure diagnostician of a robot manipulation system.
+A subtask failed or timed out. Diagnose the likely cause and choose one
+recovery:
+- retry: same subtask again (transient miss, e.g. grasp slipped).
+- adjust-param: same subtask with corrected skill_params (e.g. a wrong target
+  description); put the corrections in parameter_updates.
+- replan: the remaining plan is wrong or the scene changed; replace it.
+
+You did not verify the failure yourself, so stay factual: what deviated, what
+is the cheapest plausible fix. Reply with JSON only:
+{"diagnosis": "...", "recovery": "retry|adjust-param|replan",
+ "parameter_updates": {...}}."""
+
+
+class Reflector(ABC):
+    @abstractmethod
+    def reflect(
+        self, subtask: SubTask, verify: VerifyResult, context: dict
+    ) -> ReflectionResult:
+        raise NotImplementedError
+
+
+class RuleReflector(Reflector):
+    def reflect(
+        self, subtask: SubTask, verify: VerifyResult, context: dict
+    ) -> ReflectionResult:
+        attempt = int(context.get("attempt", 1))
+        if verify.completion == "timeout":
+            return ReflectionResult(
+                diagnosis="subtask exceeded its time budget without completion",
+                recovery="replan" if attempt >= 2 else "retry",
+            )
+        if verify.source == "self_report":
+            return ReflectionResult(
+                diagnosis="executor reported it cannot make progress",
+                recovery="replan" if attempt >= 2 else "retry",
+            )
+        return ReflectionResult(
+            diagnosis=f"postcondition not satisfied ({verify.evidence[:120]})",
+            recovery="retry" if attempt < 2 else "replan",
+        )
+
+
+class VLMReflector(Reflector):
+    def __init__(self, vlm: HarnessVLM, fallback: Reflector | None = None):
+        self.vlm = vlm
+        self.fallback = fallback or RuleReflector()
+
+    def reflect(
+        self, subtask: SubTask, verify: VerifyResult, context: dict
+    ) -> ReflectionResult:
+        payload = {
+            "subtask": {
+                "instruction": subtask.instruction,
+                "postconditions": subtask.postconditions,
+                "skill_params": subtask.skill_params,
+            },
+            "verification": {
+                "completion": verify.completion,
+                "evidence": verify.evidence,
+                "source": verify.source,
+            },
+            "memory": context,
+        }
+        try:
+            data = self.vlm.query_json(REFLECT_INSTRUCTIONS, payload)
+        except HarnessVLMError:
+            return self.fallback.reflect(subtask, verify, context)
+        recovery = data.get("recovery")
+        if recovery not in ("retry", "adjust-param", "replan"):
+            return self.fallback.reflect(subtask, verify, context)
+        updates = data.get("parameter_updates")
+        return ReflectionResult(
+            diagnosis=str(data.get("diagnosis") or "no diagnosis")[:400],
+            recovery=recovery,
+            parameter_updates=updates if isinstance(updates, dict) else {},
+        )
diff --git a/src/maniloop/harness/router.py b/src/maniloop/harness/router.py
new file mode 100644
index 0000000..b2b7a85
--- /dev/null
+++ b/src/maniloop/harness/router.py
@@ -0,0 +1,20 @@
+"""Role -> model routing. Defaults to the connected model for every role."""
+
+from __future__ import annotations
+
+ROLES = ("planner", "verifier", "reflector")
+
+
+class RoleRouter:
+    """Per-role model overrides, e.g. a cheaper model for verification."""
+
+    def __init__(self, default_model: str, overrides: dict | None = None):
+        self.default_model = default_model
+        clean = {}
+        for role, model in (overrides or {}).items():
+            if role in ROLES and isinstance(model, str) and model.strip():
+                clean[role] = model.strip()[:120]
+        self.overrides = clean
+
+    def model_for(self, role: str) -> str:
+        return self.overrides.get(role, self.default_model)
diff --git a/src/maniloop/harness/types.py b/src/maniloop/harness/types.py
new file mode 100644
index 0000000..9d7d97b
--- /dev/null
+++ b/src/maniloop/harness/types.py
@@ -0,0 +1,105 @@
+"""Contracts for the layered harness: plan, verify, reflect, remember.
+
+Abstracted from the Goal2Skill upper brain (robomme custom_high_level) and
+trimmed to the portable core: no task-specific plan templates, no local model
+backends, no oracle state. Every role reads the same memory through its own
+minimal view.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+Completion = Literal["success", "running", "fail", "timeout"]
+RecoveryType = Literal["retry", "adjust-param", "replan"]
+PhaseStatus = Literal["working", "done", "stuck"]
+
+# Hard bounds that keep the slow loop from cycling forever.
+DEFAULT_LIMITS = {
+    "max_attempts_per_subtask": 3,
+    "max_replans": 2,
+    "max_rewinds": 1,
+    # L0 progress gate: this many consecutive decide() calls without visible
+    # motion marks the subtask stalled, no API call needed.
+    "stall_window": 6,
+    "stall_translation_m": 0.002,
+    "stall_gripper_delta": 0.02,
+}
+
+
+@dataclass
+class SubTask:
+    """One semantic phase; granularity matches a visually checkable event."""
+
+    subtask_id: str
+    instruction: str
+    postconditions: list[str] = field(default_factory=list)
+    timeout_s: float = 120.0
+    skill_params: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class VerifyResult:
+    completion: Completion
+    confidence: float
+    evidence: str
+    # False only when an already-completed phase was visibly undone again.
+    prior_phase_intact: bool = True
+    # rule | self_report | vlm -- where this verdict came from.
+    source: str = "rule"
+
+
+@dataclass
+class ReflectionResult:
+    diagnosis: str
+    recovery: RecoveryType
+    parameter_updates: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class ExecutionRecord:
+    subtask_id: str
+    instruction: str
+    action_kind: str
+    explanation: str
+    phase_status: str
+
+
+@dataclass
+class ErrorRecord:
+    subtask_id: str
+    attempt_idx: int
+    completion: str
+    diagnosis: str
+    recovery: str
+
+
+@dataclass
+class MemoryState:
+    current_goal: str = ""
+    current_plan: list[SubTask] = field(default_factory=list)
+    plan_index: int = 0
+    completed_steps: list[str] = field(default_factory=list)
+    episodic_history: list[ExecutionRecord] = field(default_factory=list)
+    error_register: list[ErrorRecord] = field(default_factory=list)
+    attempt_counter: dict[str, int] = field(default_factory=dict)
+    evidence_chain: list[dict[str, Any]] = field(default_factory=list)
+    replan_count: int = 0
+    rewind_count: int = 0
+
+
+@dataclass
+class BridgeStepOutcome:
+    verify: VerifyResult | None
+    reflection: ReflectionResult | None
+    episode_done: bool
+    aborted: bool
+    replanned: bool
+    advanced: bool
+    current_subtask: SubTask | None
+    notice: str = ""
+    # True whenever the active phase (re)starts with a fresh budget: advanced,
+    # retried, adjusted, replanned or rewound. Only a plain "running" leaves
+    # the phase clock untouched.
+    phase_restarted: bool = False
diff --git a/src/maniloop/harness/verifier.py b/src/maniloop/harness/verifier.py
new file mode 100644
index 0000000..a4c3cbc
--- /dev/null
+++ b/src/maniloop/harness/verifier.py
@@ -0,0 +1,137 @@
+"""Verifier role and the zero-cost L0 progress gate.
+
+Verification is layered so the expensive VLM confirm (L2) runs only at a
+claimed subtask boundary: L0 rules watch motion feedback every step for free,
+L1 is the executor's own status line parsed from its reply, and L2 confirms a
+claimed completion from the camera image.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+import math
+
+from .types import SubTask, VerifyResult
+from .vlm import HarnessVLM, HarnessVLMError
+
+VERIFY_INSTRUCTIONS = """You are the execution verifier of a robot manipulation system.
+Judge ONLY from the camera image whether the current subtask is complete.
+The executor claims it finished; confirm or reject that claim.
+
+Rules:
+- Check every postcondition against what is visible; name what you see.
+- completion is "success" only if ALL postconditions are visibly satisfied.
+- prior_phase_intact is false only if an earlier completed result (listed in
+  memory) was visibly undone, e.g. a placed object fell or was dropped.
+- Reply with JSON only: {"completion": "success|fail", "confidence": 0.0-1.0,
+  "evidence": "what the image shows", "prior_phase_intact": true}."""
+
+
+class ProgressGate:
+    """L0: marks a subtask stalled after a window of motion-free decisions."""
+
+    def __init__(
+        self,
+        window: int = 6,
+        translation_m: float = 0.002,
+        gripper_delta: float = 0.02,
+    ):
+        self.window = window
+        self.translation_m = translation_m
+        self.gripper_delta = gripper_delta
+        self._motion: list[float] = []
+        self._last_gripper: float | None = None
+
+    def reset(self) -> None:
+        self._motion = []
+        self._last_gripper = None
+
+    def update(self, observation: dict) -> None:
+        feedback = observation.get("last_feedback")
+        if not isinstance(feedback, dict) or feedback.get("status") not in (
+            "accepted",
+            "completed",
+        ):
+            return
+        motion = 0.0
+        delta = feedback.get("actual_delta_position")
+        if isinstance(delta, (list, tuple)) and len(delta) == 3:
+            try:
+                motion += math.sqrt(sum(float(d) ** 2 for d in delta))
+            except (TypeError, ValueError):
+                pass
+        gripper = feedback.get("gripper_opening")
+        if type(gripper) in (int, float):
+            if self._last_gripper is not None:
+                motion += abs(float(gripper) - self._last_gripper)
+            self._last_gripper = float(gripper)
+        self._motion.append(motion)
+        del self._motion[: -self.window]
+
+    @property
+    def stalled(self) -> bool:
+        return (
+            len(self._motion) >= self.window
+            and max(self._motion) < self.translation_m + self.gripper_delta
+        )
+
+
+class Verifier(ABC):
+    @abstractmethod
+    def confirm(
+        self, subtask: SubTask, images: dict[str, bytes], context: dict
+    ) -> VerifyResult:
+        raise NotImplementedError
+
+
+class RuleVerifier(Verifier):
+    """Degraded L2: trust the executor's claim, at low confidence."""
+
+    def confirm(
+        self, subtask: SubTask, images: dict[str, bytes], context: dict
+    ) -> VerifyResult:
+        del images, context
+        return VerifyResult(
+            completion="success",
+            confidence=0.4,
+            evidence="VLM 确认不可用，采信执行器自报完成",
+            source="self_report",
+        )
+
+
+class VLMVerifier(Verifier):
+    def __init__(self, vlm: HarnessVLM, fallback: Verifier | None = None):
+        self.vlm = vlm
+        self.fallback = fallback or RuleVerifier()
+
+    def confirm(
+        self, subtask: SubTask, images: dict[str, bytes], context: dict
+    ) -> VerifyResult:
+        try:
+            data = self.vlm.query_json(
+                VERIFY_INSTRUCTIONS,
+                {
+                    "subtask": {
+                        "instruction": subtask.instruction,
+                        "postconditions": subtask.postconditions,
+                    },
+                    "memory": context,
+                },
+                images=images,
+            )
+        except HarnessVLMError:
+            return self.fallback.confirm(subtask, images, context)
+        completion = data.get("completion")
+        if completion not in ("success", "fail"):
+            return self.fallback.confirm(subtask, images, context)
+        try:
+            confidence = min(max(float(data.get("confidence", 0.5)), 0.0), 1.0)
+        except (TypeError, ValueError):
+            confidence = 0.5
+        return VerifyResult(
+            completion=completion,
+            confidence=confidence,
+            evidence=str(data.get("evidence") or "")[:400],
+            prior_phase_intact=data.get("prior_phase_intact") is not False,
+            source="vlm",
+        )
diff --git a/src/maniloop/harness/vlm.py b/src/maniloop/harness/vlm.py
new file mode 100644
index 0000000..584cbcf
--- /dev/null
+++ b/src/maniloop/harness/vlm.py
@@ -0,0 +1,134 @@
+"""Shared JSON-in/JSON-out VLM transport for harness roles.
+
+Reuses the executor policy's authenticated OpenAI-compatible client instead of
+holding credentials of its own. Responses are parsed leniently: a reply cut
+off mid-sentence is repaired by closing the brackets it left hanging (the
+fields that matter come first; truncation lands in the trailing prose).
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import re
+import time
+from typing import Any
+
+
+class HarnessVLMError(RuntimeError):
+    """The role's model never produced a parseable JSON reply."""
+
+
+def _close_truncated_json(cleaned: str) -> str:
+    out: list[str] = []
+    stack: list[str] = []
+    in_string = False
+    escaped = False
+    for char in cleaned:
+        out.append(char)
+        if in_string:
+            if escaped:
+                escaped = False
+            elif char == "\\":
+                escaped = True
+            elif char == '"':
+                in_string = False
+            continue
+        if char == '"':
+            in_string = True
+        elif char in "{[":
+            stack.append("}" if char == "{" else "]")
+        elif char in "}]" and stack:
+            stack.pop()
+    if in_string:
+        out.append('"')
+    text = "".join(out).rstrip()
+    text = re.sub(r",\s*$", "", text)
+    text = re.sub(r',\s*"[^"]*"?\s*:?\s*$', "", text)
+    return text + "".join(reversed(stack))
+
+
+def parse_json_object(raw: str) -> dict | None:
+    """First {...} block of the reply, repaired if the output was truncated."""
+    if not isinstance(raw, str):
+        return None
+    start = raw.find("{")
+    end = raw.rfind("}")
+    if start < 0:
+        return None
+    cleaned = raw[start : end + 1] if end > start else raw[start:]
+    try:
+        value = json.loads(cleaned)
+    except (ValueError, RecursionError):
+        try:
+            value = json.loads(_close_truncated_json(cleaned))
+        except (ValueError, RecursionError):
+            return None
+    return value if isinstance(value, dict) else None
+
+
+class HarnessVLM:
+    """One Responses-API JSON query channel bound to a role's model."""
+
+    def __init__(self, client, model: str, *, max_output_tokens: int = 4096):
+        self.client = client
+        self.model = model
+        self.max_output_tokens = max_output_tokens
+        self.last_usage: dict = {}
+        self.last_latency = 0.0
+
+    def query_json(
+        self,
+        instructions: str,
+        payload: dict,
+        images: dict[str, bytes] | None = None,
+    ) -> dict:
+        self.last_usage = {}
+        self.last_latency = 0.0
+        content: list[dict] = [
+            {"type": "input_text", "text": json.dumps(payload, ensure_ascii=False, default=str)}
+        ]
+        for name, data in (images or {}).items():
+            if not isinstance(data, (bytes, bytearray)) or not data:
+                continue
+            mime = "image/png" if data.startswith(b"\x89PNG") else "image/jpeg"
+            content.append({"type": "input_text", "text": f"Camera: {name}."})
+            content.append(
+                {
+                    "type": "input_image",
+                    "image_url": f"data:{mime};base64,"
+                    + base64.b64encode(bytes(data)).decode("ascii"),
+                    "detail": "high",
+                }
+            )
+        request = {
+            "model": self.model,
+            "instructions": instructions,
+            "input": [{"role": "user", "content": content}],
+            "store": False,
+            "max_output_tokens": self.max_output_tokens,
+        }
+        started = time.monotonic()
+        try:
+            response = self.client.responses.create(**request)
+        except Exception as exc:
+            raise HarnessVLMError(f"{self.model}: 请求失败（{type(exc).__name__}）") from None
+        finally:
+            self.last_latency = time.monotonic() - started
+        usage = getattr(response, "usage", None)
+        if usage is not None:
+            for field_name in ("input_tokens", "output_tokens", "total_tokens"):
+                count = getattr(usage, field_name, None)
+                if type(count) is int and count >= 0:
+                    self.last_usage[field_name] = count
+        texts: list[str] = []
+        for item in getattr(response, "output", None) or []:
+            if getattr(item, "type", None) != "message":
+                continue
+            for part in getattr(item, "content", None) or []:
+                if getattr(part, "type", None) == "output_text":
+                    texts.append(getattr(part, "text", ""))
+        parsed = parse_json_object("\n".join(texts))
+        if parsed is None:
+            raise HarnessVLMError(f"{self.model}: 响应不是可解析的 JSON")
+        return parsed
diff --git a/src/maniloop/runtime/runner.py b/src/maniloop/runtime/runner.py
index 37cedcd..2de6abf 100644
--- a/src/maniloop/runtime/runner.py
+++ b/src/maniloop/runtime/runner.py
@@ -66,6 +66,7 @@ def __init__(
         self.representation = SensorRepresentation()
         self.chunk = None
         self.policy_kind = "llm_cloud"
+        self.harness_roles = {}
         self.sim_budget = 120.0
         self.wall_budget = 600.0
         self.started_wall = None
@@ -276,6 +277,9 @@ def activate_connection(self, config, model, source):
             **({"request_options": self.request_options} if self.request_options else {}),
         )
         self.close_policy()
+        if self.policy_kind == "harness":
+            from maniloop.harness import HarnessAgent
+            policy = HarnessAgent(policy, roles=self.harness_roles)
         self.policy = policy
         self.key = config.api_key
         self.model = model
@@ -362,7 +366,7 @@ def command(self, name, payload):
         elif name in {"pause", "resume", "step"}:
             if not self.running or self.diagnostic_stage:
                 raise ValueError("请先开始一个机器人演示")
-            if self.policy_kind != "llm_cloud":
+            if self.policy_kind not in ("llm_cloud", "harness"):
                 raise ValueError("单步与暂停目前仅用于 LLM 演示")
             if name == "pause":
                 self.human_pause_used = True
@@ -528,8 +532,16 @@ def command(self, name, payload):
                 self.credential_source = "mock"
                 self.begin_episode(task, int(payload.get("max_steps", 30)))
                 return
-            self.policy_kind = "llm_cloud"
+            harness_mode = payload.get("agent") == "harness"
+            self.policy_kind = "harness" if harness_mode else "llm_cloud"
             source = payload.get("credential_source", "manual")
+            if harness_mode and source == "codex":
+                raise ValueError("分层大脑暂不支持 Codex 登录，请使用 API 密钥或配置文件")
+            if harness_mode:
+                roles = payload.get("harness_roles", {})
+                if type(roles) is not dict:
+                    raise ValueError("harness_roles 必须是对象")
+                self.harness_roles = dict(roles)
             config = None if source == "codex" else self.resolve_connection(payload)
             override = payload.get("model") or None
             if override is not None and (
@@ -856,10 +868,12 @@ def tick(self):
                 try:
                     action = future.result()
                 finally:
-                    if self.policy_kind == "jev":
+                    if self.policy_kind in ("jev", "harness"):
                         self.policy_decision = deepcopy(self.policy.last_decision)
                         if self.policy_decision:
-                            self.event("provider_decision", "Jev 类型化选择（无生成式思考文本）",
+                            self.event("provider_decision",
+                                "Jev 类型化选择（无生成式思考文本）" if self.policy_kind == "jev"
+                                else "分层大脑状态（计划/验证/记忆快照）",
                                 decision=self.policy_decision, request_index=self.api_calls,
                                 usage=self.policy.last_usage)
                 if self.diagnostic_stage:
@@ -1147,7 +1161,8 @@ def publish(self, render=True):
             "typesafe_browser_key_configured": bool(self.typesafe_key),
             "typesafe_key_source": ("browser" if self.typesafe_key else
                 "environment" if os.environ.get("TYPESAFE_API_KEY", "").strip() else "none"),
-            "policy_decision": deepcopy(self.policy_decision) if self.policy_kind == "jev" else None,
+            "policy_decision": (deepcopy(self.policy_decision)
+                if self.policy_kind in ("jev", "harness") else None),
             "sim_time": float(self.sim.simulation_time),
             "tcp_position": self.sim.tcp_position.tolist(),
             "gripper_opening": self.sim.gripper_opening,
diff --git a/src/maniloop/ui/workspace.html b/src/maniloop/ui/workspace.html
index 79e61b2..412b90f 100644
--- a/src/maniloop/ui/workspace.html
+++ b/src/maniloop/ui/workspace.html
@@ -99,7 +99,7 @@
           <div class="panel-header"><div class="panel-title"><span class="number-tag">01</span>任务与模型</div><span class="small">由你发起每次实验</span></div>
           <form id="task-form" class="task-body" autocomplete="off">
             <label class="field-label" for="agent-select">策略类型</label>
-            <select id="agent-select" class="input" style="margin-bottom:14px"><option value="llm_cloud">云端 LLM API</option><option value="jev">Jev · 文本原语控制</option><option value="mock_vla">模拟 VLA · 离线接口检查</option><option value="lerobot">本地 LeRobot · 真实模型</option></select>
+            <select id="agent-select" class="input" style="margin-bottom:14px"><option value="llm_cloud">云端 LLM API</option><option value="harness">分层大脑 Harness · 规划+验证</option><option value="jev">Jev · 文本原语控制</option><option value="mock_vla">模拟 VLA · 离线接口检查</option><option value="lerobot">本地 LeRobot · 真实模型</option></select>
 <div id="jev-fields" hidden>
   <p class="session-notice">纯 Jev 文本控制：每行一个动作，Jev 选择原语，本地跟踪目标。相机仅供人类查看，不发送给 Jev。序列结束不等于任务成功。</p>
   <div class="key-row">
diff --git a/src/maniloop/ui/workspace.js b/src/maniloop/ui/workspace.js
index 16bc4fc..484b104 100644
--- a/src/maniloop/ui/workspace.js
+++ b/src/maniloop/ui/workspace.js
@@ -237,7 +237,7 @@ function updateControls() {
   const pauseRequested = !!(state && state.pause_requested);
   const inFlight = !!(state && state.pending_request);
   const moving = !!(state && state.motion_busy);
-  const cloud = $('agent-select').value === 'llm_cloud';
+  const cloud = ['llm_cloud', 'harness'].includes($('agent-select').value);
   const steppable = cloud || $('agent-select').value === 'jev';
   const ready = !!(connected && state && state.ready);
   const configLocked = pending || configBusy || running || inFlight;
@@ -322,7 +322,7 @@ function renderState(data) {
     ui.notice.textContent = data.paused ? '已暂停：不会请求下一次模型决策。可检查画面与反馈，选择执行下一步或继续连续执行。' : data.pause_requested ? '等待本次决策和动作结束，随后暂停；不会额外请求下一次决策。' : (data.timing === 'controlled' ? '受控时序：模型思考期间暂停物理，动作执行后重新观察。' : '实时模式：模型思考期间物理持续推进。');
     ui.notice.className = 'session-notice running';
   } else {
-    ui.notice.textContent = $('agent-select').value === 'jev' ? 'Jev 仅读取指令与本体状态；每行请求一次选择，本地执行后才处理下一行。' : $('agent-select').value === 'lerobot' ? '使用本机已安装的真实模型；ACT / DP 不使用语言。默认按官方任务与所选初始化执行。' : $('agent-select').value === 'mock_vla' ? '模拟 VLA 只做小幅关节动作来验证接口，不调用 API，也不代表真实模型能力。' : ui.source.value === 'codex' ? '预览和手动调试不调用模型。单步、连续执行或诊断通过 Codex 登录请求模型，使用账号额度。' : '无需密钥即可预览与手动调试。开始执行会向所选服务商发送相机观测和任务指令。';
+    ui.notice.textContent = $('agent-select').value === 'jev' ? 'Jev 仅读取指令与本体状态；每行请求一次选择，本地执行后才处理下一行。' : $('agent-select').value === 'harness' ? '分层大脑：上层把任务分解为语义子任务、验证完成并在失败时反思重规划；下层由所选云端模型逐步执行。密钥配置与云端 LLM 相同。' : $('agent-select').value === 'lerobot' ? '使用本机已安装的真实模型；ACT / DP 不使用语言。默认按官方任务与所选初始化执行。' : $('agent-select').value === 'mock_vla' ? '模拟 VLA 只做小幅关节动作来验证接口，不调用 API，也不代表真实模型能力。' : ui.source.value === 'codex' ? '预览和手动调试不调用模型。单步、连续执行或诊断通过 Codex 登录请求模型，使用账号额度。' : '无需密钥即可预览与手动调试。开始执行会向所选服务商发送相机观测和任务指令。';
     ui.notice.className = 'session-notice';
   }
   $('observation-content').textContent = data.observation && Object.keys(data.observation).length ? text(data.observation) : '等待第一份观测…';
@@ -608,12 +608,13 @@ $('demo-preset').addEventListener('click', async () => {
 });
 $('agent-select').addEventListener('change', () => {
   const local = $('agent-select').value === 'lerobot';
+  const cloud = ['llm_cloud', 'harness'].includes($('agent-select').value);
   $('local-policy-fields').hidden = !local;
-  $('cloud-controls').hidden = $('agent-select').value !== 'llm_cloud';
-  ui.modelSelect.parentElement.hidden = $('agent-select').value !== 'llm_cloud';
-  document.querySelector('.model-tools').hidden = $('agent-select').value !== 'llm_cloud';
-  if ($('agent-select').value !== 'llm_cloud') $('custom-model-fields').hidden = true;
-  document.querySelector('.credentials-section').hidden = $('agent-select').value !== 'llm_cloud';
+  $('cloud-controls').hidden = !cloud;
+  ui.modelSelect.parentElement.hidden = !cloud;
+  document.querySelector('.model-tools').hidden = !cloud;
+  if (!cloud) $('custom-model-fields').hidden = true;
+  document.querySelector('.credentials-section').hidden = !cloud;
   ui.maxSteps.max = local ? '1000' : '100';
   ui.maxSteps.value = local ? '500' : '40';
   updateControls();
@@ -713,11 +714,11 @@ async function startEpisode(singleStep = false) {
   const model = selectedModel();
   const maxSteps = Number(ui.maxSteps.value);
   if (!task) { showError(new Error('请先输入任务指令。')); ui.task.focus(); return; }
-  if ($('agent-select').value === 'llm_cloud' && ui.modelSelect.value === '__custom__' && !model) { showError(new Error('请填写自定义模型名称。')); ui.model.focus(); return; }
+  if (['llm_cloud', 'harness'].includes($('agent-select').value) && ui.modelSelect.value === '__custom__' && !model) { showError(new Error('请填写自定义模型名称。')); ui.model.focus(); return; }
   if (!Number.isInteger(maxSteps) || maxSteps < 0 || maxSteps > Number(ui.maxSteps.max)) { showError(new Error('决策次数超出当前策略允许范围；0 表示不限次数。')); return; }
   const agent = $('agent-select').value;
   const body = {task, max_steps: maxSteps, agent, single_step: singleStep};
-  if (agent === 'llm_cloud') Object.assign(body, selectedCredentials(), environmentSettings(), {
+  if (agent === 'llm_cloud' || agent === 'harness') Object.assign(body, selectedCredentials(), environmentSettings(), {
     model, request_options: requestSettings(), reset_on_start: $('reset-before-start').checked,
     context_mode: $('context-mode').value,
     max_wall_seconds: Number($('wall-budget').value), max_sim_seconds: Number($('sim-budget').value)
@@ -727,9 +728,9 @@ async function startEpisode(singleStep = false) {
     catch (error) { showError(error); return; }
   }
   if (agent === 'lerobot') Object.assign(body, {local_model: $('local-model').value, device: 'auto'});
-  if (agent === 'llm_cloud' && !validateConfigSelection(body)) return;
-  if (agent === 'llm_cloud' && body.credential_source === 'manual' && !body.api_key && !(state && state.manual_key_configured)) { showError(new Error('请填写 API 密钥，或选择本地配置文件。')); ui.key.focus(); return; }
-  if (agent === 'llm_cloud') ui.key.value = '';
+  if ((agent === 'llm_cloud' || agent === 'harness') && !validateConfigSelection(body)) return;
+  if ((agent === 'llm_cloud' || agent === 'harness') && body.credential_source === 'manual' && !body.api_key && !(state && state.manual_key_configured)) { showError(new Error('请填写 API 密钥，或选择本地配置文件。')); ui.key.focus(); return; }
+  if (agent === 'llm_cloud' || agent === 'harness') ui.key.value = '';
   try {
     const result = await perform('/api/start', body, singleStep ? '已提交一次模型决策；动作结束后暂停。' : '任务已提交，正在准备第一次模型决策。');
     if (agent === 'jev') clearSubmittedJevKey(body, result);