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/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..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 @@ -73,7 +74,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 +89,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 +134,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 +154,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 +169,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): @@ -273,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 @@ -359,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 @@ -525,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 ( @@ -679,6 +694,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 +726,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 +772,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 +808,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() @@ -824,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: @@ -1115,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/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 @@