From daa03715198ae021e1998bb68258c5c80df54c22 Mon Sep 17 00:00:00 2001 From: opencode Date: Sat, 8 Aug 2026 21:58:57 +0300 Subject: [PATCH] search challenge: triangle marker on a crate's back face (rendered in vision, in digest), report tool with distance verification, smooth turn(duration), --free/--search autonomous modes, look_at crates; verified with gpt-5.6-luna --- testbed/README.md | 19 ++++++++ testbed/bridge.py | 85 +++++++++++++++++++++++++++++++++--- testbed/chat.py | 73 +++++++++++++++++++++++++------ testbed/demo.py | 8 ++++ testbed/llm_agent.py | 62 ++++++++++++++++++++------ testbed/room/render.py | 76 ++++++++++++++++++++++++++++++++ testbed/room/world.py | 57 ++++++++++++++++++++++++ testbed/tests/test_bridge.py | 50 +++++++++++++++++++++ testbed/tests/test_world.py | 48 ++++++++++++++++++++ 9 files changed, 444 insertions(+), 34 deletions(-) diff --git a/testbed/README.md b/testbed/README.md index b473931..4c66a0e 100644 --- a/testbed/README.md +++ b/testbed/README.md @@ -64,6 +64,25 @@ taken); stop the old one with `fuser -k 8765/tcp` or Ctrl-C in its terminal. The demo prints a full transcript of tool calls/results to stdout and saves the capsule's final first-person frame to `demo_final_frame.png`. +### Search challenge + +A small orange triangle is painted on the **back side** of one of the crates +(random per run, always on a face hidden from the spawn corner). The agent must +explore the room, look at the crate faces, spot the triangle with `vision`, +and `report` it (the bridge verifies the report by distance): + +```bash +scripts/run_chat.sh --provider polza --search +# autonomous exploration, real vision, smooth turns, cruise gliding; +# success = report verified +``` + +Search missions run in `--free` mode: no re-aim corrections, no collision +hints — the model plans its own exploration. `--free` also applies to beacon +missions if you want fewer guardrails. `look_at` now accepts crates too +(crate_red/crate_blue/crate_olive). Verified: gpt-5.6-luna explored the room +with smooth moves/turns and reported the triangle at 0.5 m. + ### Smooth & proactive movement The capsule no longer teleports or stops-and-thinks: diff --git a/testbed/bridge.py b/testbed/bridge.py index 1ed0205..2d139c4 100644 --- a/testbed/bridge.py +++ b/testbed/bridge.py @@ -113,6 +113,15 @@ class WorldQueryOutput(BaseModel): tick: int +class ReportOutput(BaseModel): + discovery: str + success: bool + verified: bool + distance: float + message: str + tick: int + + class EchoOutput(BaseModel): value: str @@ -166,6 +175,17 @@ class RoomBridge(Bridge): def _tick(self) -> int: return self.world.advance_tick() + async def _animate_turn( + self, yaw_deg: float, pitch_deg: float, duration: float + ) -> None: + """Rotate smoothly over ``duration`` seconds (observable mid-turn).""" + n = max(1, min(int(duration / 0.1), 100)) + dt = duration / n + for i in range(n): + self.world.turn(yaw_deg / n, pitch_deg / n) + if i < n - 1: + await asyncio.sleep(dt) + async def _animate_move( self, forward: float, duration: float ) -> tuple[float, Any | None]: @@ -333,24 +353,48 @@ class RoomBridge(Bridge): @self.tool( cls=ToolClass.ACTUATOR, - description="Rotate the camera: yaw_deg turns left/right, pitch_deg tilts up/down.", + description=( + "Rotate the camera: yaw_deg turns left/right, pitch_deg tilts up/down. " + "duration (seconds) animates the rotation smoothly." + ), ) - async def turn(yaw_deg: float = 0.0, pitch_deg: float = 0.0) -> TurnOutput: + async def turn( + yaw_deg: float = 0.0, pitch_deg: float = 0.0, duration: float = 0.0 + ) -> TurnOutput: tick = self._tick() - world.turn(yaw_deg, pitch_deg) + if duration < 0.0 or duration > 10.0: + raise ValueError(f"duration must be in [0, 10] s, got {duration}") + if duration > 0.0: + await self._animate_turn(yaw_deg, pitch_deg, duration) + else: + world.turn(yaw_deg, pitch_deg) return TurnOutput(rotation=self._rotation(), tick=tick).model_dump() @self.tool( cls=ToolClass.ACTUATOR, - description="Orient the camera toward a named target (e.g. 'beacon').", + description=( + "Orient the camera toward a named target: 'beacon' or a crate " + "(crate_red, crate_blue, crate_olive)." + ), ) async def look_at(target: str) -> LookAtOutput: tick = self._tick() - if target != "beacon": + box = next((b for b in world.boxes if b.id == target), None) + if target == "beacon": + yaw, pitch = world.look_at_beacon() + elif box is not None: + dx = box.cx - world.capsule.x + dz = box.cz - world.capsule.z + dist = math.hypot(dx, dz) or 1.0 + yaw = math.degrees(math.atan2(dx, dz)) + pitch = math.degrees( + math.atan2(box.height / 2 - world.capsule.eye_height, dist) + ) + world.face(yaw, pitch) + else: raise ValueError( - f"unknown target {target!r}; available targets: beacon" + f"unknown target {target!r}; available: beacon, crate_red, crate_blue, crate_olive" ) - yaw, pitch = world.look_at_beacon() return LookAtOutput( target=target, rotation=Rotation( @@ -383,6 +427,33 @@ class RoomBridge(Bridge): target=target, success=success, message=message, tick=tick ).model_dump() + @self.tool( + cls=ToolClass.ACTUATOR, + description=( + "Report a discovery to the operator (used in search missions): " + "say what you found. The report is verified against your position." + ), + ) + async def report(discovery: str = "triangle") -> ReportOutput: + tick = self._tick() + dist = world.distance_to_marker() + verified = world.marker is not None and dist <= 6.0 + if verified: + message = f"report verified: {discovery} sighted {dist:.1f} m away" + else: + message = ( + f"not verified: nothing matching {discovery!r} within 6 m " + f"(closest check {dist:.1f} m) — keep exploring" + ) + return ReportOutput( + discovery=discovery, + success=True, + verified=verified, + distance=round(dist, 2), + message=message, + tick=tick, + ).model_dump() + # -- conformance tools (design doc: register alongside world tools) -- @self.tool( diff --git a/testbed/chat.py b/testbed/chat.py index c5f0541..9c41f27 100644 --- a/testbed/chat.py +++ b/testbed/chat.py @@ -141,16 +141,30 @@ async def chat_loop(client: AICCClient, manifest, args: argparse.Namespace) -> i async def mission_runner(goal_text: str) -> None: from testbed.llm_agent import run_llm_agent_loop - controller.messages.append( - { - "role": "user", - "content": ( - f"MISSION (set by the user): {goal_text}. " - "Keep calling tools and do not stop until the goal is achieved. " - "Report only when done." - ), - } - ) + if args.search: + from testbed.llm_agent import SEARCH_MISSION + + controller.messages = [ + {"role": "system", "content": SEARCH_MISSION}, + { + "role": "user", + "content": ( + f"MISSION: {goal_text}. " + "Keep exploring with tools until the triangle is found and reported." + ), + }, + ] + else: + controller.messages.append( + { + "role": "user", + "content": ( + f"MISSION (set by the user): {goal_text}. " + "Keep calling tools and do not stop until the goal is achieved. " + "Report only when done." + ), + } + ) saver = asyncio.create_task(map_saver()) try: for attempt in range(1, args.mission_retries + 2): @@ -165,12 +179,27 @@ async def chat_loop(client: AICCClient, manifest, args: argparse.Namespace) -> i nudge_limit=args.mission_nudges, look_every=args.look_every, cruise=args.cruise, + autonomous=args.free or args.search, + is_success=( + ( + lambda turn: any( + c.name == "report" + and c.ok + and isinstance(c.output, dict) + and c.output.get("verified") + for c in turn.calls + ) + ) + if args.search + else None + ), ) except RuntimeError as exc: print(f" [mission] LLM error: {exc}") break save_map() - if summary["interacted"]: + done = summary.get("success") if args.search else summary["interacted"] + if done: print(f"\n[mission] DONE: {summary['result']}") return print( @@ -226,7 +255,12 @@ async def chat_loop(client: AICCClient, manifest, args: argparse.Namespace) -> i print(" (no mission running)") await cmd_state() - if args.mission: + if args.mission or args.search: + if args.search: + mission_goal = ( + "explore the room, find the orange triangle painted on the back " + "of one of the crates, and report it" + ) start_mission() while True: @@ -428,6 +462,17 @@ def main() -> int: action="store_true", help="start an autonomous mission on connect: reach and activate the beacon", ) + parser.add_argument( + "--search", + action="store_true", + help="search challenge: explore the room and find the orange triangle " + "on the back of a crate, then report it (autonomous mode, fewer guardrails)", + ) + parser.add_argument( + "--free", + action="store_true", + help="fewer restrictions: no re-aim corrections or collision hints; the model plans freely", + ) parser.add_argument( "--mission-steps", type=int, @@ -479,7 +524,9 @@ def main() -> int: model=args.model, ) if args.provider: - print(f"[chat] provider: {args.provider} -> {args.base_url} model: {args.model}") + print( + f"[chat] provider: {args.provider} -> {args.base_url} model: {args.model}" + ) except ValueError as exc: print(f"[chat] {exc}") return 1 diff --git a/testbed/demo.py b/testbed/demo.py index 0c97def..fb327bf 100644 --- a/testbed/demo.py +++ b/testbed/demo.py @@ -143,6 +143,7 @@ async def run_llm_agent( digest: bool | None = None, look_every: int = 0, cruise: float = 0.0, + autonomous: bool = False, ) -> dict[str, Any]: """Autonomous LLM run: controller + nudge/correct loop (see llm_agent).""" from testbed.llm_agent import LLMController, run_llm_agent_loop @@ -317,6 +318,7 @@ async def run_demo(args: argparse.Namespace) -> dict[str, Any]: digest=args.digest, look_every=args.look_every, cruise=args.cruise, + autonomous=args.free, ) elif args.agent == "scripted": summary = await run_scripted_agent( @@ -341,6 +343,7 @@ async def run_demo(args: argparse.Namespace) -> dict[str, Any]: digest=args.digest, look_every=args.look_every, cruise=args.cruise, + autonomous=args.free, ) if summary.get("interacted"): return summary @@ -440,6 +443,11 @@ def main() -> int: default=1.0, help="proactive motion: glide forward (meters) while the LLM thinks (0 disables)", ) + parser.add_argument( + "--free", + action="store_true", + help="fewer restrictions: no re-aim corrections or collision hints; the model plans freely", + ) parser.add_argument( "--frame", default="demo_final_frame.png", diff --git a/testbed/llm_agent.py b/testbed/llm_agent.py index b1752a4..72df002 100644 --- a/testbed/llm_agent.py +++ b/testbed/llm_agent.py @@ -68,6 +68,19 @@ NUDGE = ( "then call a tool (turn or move, or interact if within 1.6 m)." ) +SEARCH_MISSION = ( + "You are an explorer in a 16x16 m room. Three wooden crates stand in the room, " + "and a small ORANGE TRIANGLE is painted on the BACK side of ONE of them — the " + "side that faces away from the room's entrance, so it is only visible once you " + "walk around the crates. The glowing beacon in the corner is irrelevant: ignore it. " + "Your mission: explore the room until you see the orange triangle with your own " + 'eyes (vision), then walk near it and call report(discovery="triangle"). ' + "Strategy: walk a loop around the crates; every few steps look at the crate faces " + "with vision; check faces on all sides. The CURRENT STATE note in tool results " + "tells you your position and heading. Do not stop until you have seen and " + "reported the triangle." +) + # --------------------------------------------------------------------------- # Provider presets (OpenAI-compatible endpoints) # --------------------------------------------------------------------------- @@ -188,6 +201,7 @@ def looks_multimodal(model: str) -> bool: COLOR_NAMES = [ ("beacon_cyan", (120, 210, 235)), ("beacon_yellow", (255, 190, 90)), + ("marker_orange", (255, 150, 40)), ("crate_red", (178, 64, 54)), ("crate_blue", (64, 96, 178)), ("crate_olive", (128, 128, 60)), @@ -606,14 +620,20 @@ async def run_llm_agent_loop( nudge_limit: int = 1, look_every: int = 0, cruise: float = 0.0, + autonomous: bool = False, + is_success: Callable[[TurnResult], bool] | None = None, ) -> dict[str, Any]: - """Drive the controller until the mission is done or steps run out. + """Drive the controller until the goal is done or steps run out. ``look_every``: attach a fresh camera frame every N steps so the model always sees recent visual context without asking (0 disables). ``cruise``: proactive motion — while the model is thinking, the capsule keeps gliding forward (meters per think, 0 disables). Collisions stop the drift and are reported to the model. + ``autonomous``: fewer guardrails — no re-aim corrections and no collision + hints; the model plans its own exploration (used for search missions). + ``is_success``: custom goal predicate over a model turn; by default the + mission ends when interact succeeds. """ summary: dict[str, Any] = { "steps": 0, @@ -730,23 +750,37 @@ async def run_llm_agent_loop( and c.output.get("collision") ): last_collision_step = step - obj = c.output.get("collision_normal") - hint_msg = ( - f"You collided with an obstacle (normal {obj}). Turn yaw_deg=90 and move " - "forward twice to get around it, then follow the CURRENT STATE note to re-aim." - ) - if not ( - controller.messages - and controller.messages[-1].get("content") == hint_msg - ): - controller.messages.append({"role": "user", "content": hint_msg}) - log("agent", "(collision: go around)") + if not autonomous: + obj = c.output.get("collision_normal") + hint_msg = ( + f"You collided with an obstacle (normal {obj}). Turn yaw_deg=90 and move " + "forward twice to get around it, then follow the CURRENT STATE note to re-aim." + ) + if not ( + controller.messages + and controller.messages[-1].get("content") == hint_msg + ): + controller.messages.append( + {"role": "user", "content": hint_msg} + ) + log("agent", "(collision: go around)") # Show the model what it just bumped into. await controller.auto_frame( "You just bumped into something. Look at what is in front of you." ) - maybe_correct(step) - if turn.interacted: + if not autonomous: + maybe_correct(step) + if is_success is not None: + if is_success(turn): + summary["success"] = True + for c in turn.calls: + if c.name == "report" and c.ok and isinstance(c.output, dict): + summary["result"] = c.output.get("message", "reported") + break + else: + summary["result"] = "goal achieved" + return summary + elif turn.interacted: summary["interacted"] = True summary["result"] = turn.message return summary diff --git a/testbed/room/render.py b/testbed/room/render.py index c1800d5..737f878 100644 --- a/testbed/room/render.py +++ b/testbed/room/render.py @@ -114,6 +114,7 @@ class Raycaster: self._draw_beacon(room, pixels, depth, fx, fz, rx, rz, eye, tan_pitch) img = Image.frombuffer("RGB", (w, h), bytes(pixels), "raw", "RGB", 0, 1) + self._draw_marker(room, img, depth, fx, fz, rx, rz, eye, tan_pitch) grid = self._depth_grid(depth, max_depth) return img, grid @@ -260,6 +261,65 @@ class Raycaster: col = FLOOR_LINE return _fog(col, t) + def _draw_marker( + self, + room: Room, + img: Image.Image, + depth: list[list[float]], + fx: float, + fz: float, + rx: float, + rz: float, + eye: float, + tan_pitch: float, + ) -> None: + """Draw the search-challenge triangle (a sprite on the crate's back + face), only when the anchor pixel is actually that face.""" + m = room.marker + if m is None: + return + pos = room.marker_world_pos() + if pos is None: + return + w, h = self.width, self.height + cam = room.capsule + rel_x = pos[0] - cam.x + rel_z = pos[2] - cam.z + along = rel_x * fx + rel_z * fz + if along < 0.3: + return + right = rel_x * rx + rel_z * rz + col_c = (right / along / self.tan_fx + 1.0) / 2.0 * w + vv = (pos[1] - eye) / along - tan_pitch + row_c = (1.0 - vv / self.tan_fy) / 2.0 * h + r0, c0 = round(row_c), round(col_c) + if not (0 <= r0 < h and 0 <= c0 < w): + return + if abs(depth[r0][c0] - along) > 0.25: + return # the anchor pixel is not the marker's face (occluded) + half = max(2.5, m["size"] / along / self.tan_fx * w / 2.0) + draw = ImageDraw.Draw(img) + color = tuple(m["color"]) + glow = tuple(min(255, c + 90) for c in color) + draw.ellipse( + [ + col_c - half * 2.2, + row_c - half * 2.2, + col_c + half * 2.2, + row_c + half * 2.2, + ], + fill=glow, + ) + draw.polygon( + [ + (col_c, row_c - half), + (col_c - half * 0.9, row_c + half * 0.7), + (col_c + half * 0.9, row_c + half * 0.7), + ], + fill=color, + outline=(255, 255, 255), + ) + def _draw_beacon( self, room: Room, @@ -405,6 +465,22 @@ def render_topdown(room: Room, size: int = 400) -> Image.Image: fx, fz = c.forward() tip = xy(c.x + fx * 0.7, c.z + fz * 0.7) draw.line([(cx, cz), tip], fill=(30, 30, 40), width=3) + + m = room.marker + if m is not None: + pos = room.marker_world_pos() + if pos is not None: + mx, mz = xy(pos[0], pos[2]) + r = m["size"] * scale + draw.polygon( + [ + (mx, mz - r * 1.4), + (mx - r, mz + r), + (mx + r, mz + r), + ], + fill=tuple(m["color"]), + outline=(20, 20, 26), + ) return img diff --git a/testbed/room/world.py b/testbed/room/world.py index a385eaf..c734dfa 100644 --- a/testbed/room/world.py +++ b/testbed/room/world.py @@ -8,6 +8,7 @@ from __future__ import annotations import math from dataclasses import dataclass +from typing import Any ROOM_SIZE = 16.0 ROOM_HEIGHT = 3.0 @@ -124,6 +125,62 @@ class Room: ] # Pending audio events, drained by the hear sensor. self._audio: list[dict] = [] + # Search challenge: a small triangle painted on the BACK side of one + # of the crates (the face pointing away from the spawn corner). The + # agent must find it by exploring and looking with vision. + self.marker: dict[str, Any] = self._place_marker() + + # ---------- search marker ---------- + + def _place_marker(self) -> dict[str, Any]: + """Put the triangle on a random crate, on the face most hidden from + the spawn corner (1.5, 1.5).""" + import random + + box = random.choice(self.boxes) + to_spawn = (1.5 - box.cx, 1.5 - box.cz) + normals = { + "x0": (-1.0, 0.0), + "x1": (1.0, 0.0), + "z0": (0.0, -1.0), + "z1": (0.0, 1.0), + } + faces = sorted( + normals, + key=lambda f: normals[f][0] * to_spawn[0] + normals[f][1] * to_spawn[1], + )[:2] # the two faces pointing away from spawn + return { + "box": box.id, + "face": random.choice(faces), + "u": 0.5, + "v": 0.55, + "size": 0.42, + "color": (255, 150, 40), + } + + def marker_world_pos(self) -> tuple[float, float, float] | None: + """World coordinates (x, y, z) of the marker center, or None.""" + m = self.marker + if m is None: + return None + box = next((b for b in self.boxes if b.id == m["box"]), None) + if box is None: + return None + if m["face"] == "x0": + x, z = box.x_min(), box.cz - box.half_d + m["u"] * 2 * box.half_d + elif m["face"] == "x1": + x, z = box.x_max(), box.cz - box.half_d + m["u"] * 2 * box.half_d + elif m["face"] == "z0": + x, z = box.cx - box.half_w + m["u"] * 2 * box.half_w, box.z_min() + else: + x, z = box.cx - box.half_w + m["u"] * 2 * box.half_w, box.z_max() + return (x, m["v"] * box.height, z) + + def distance_to_marker(self) -> float: + pos = self.marker_world_pos() + if pos is None: + return float("inf") + return math.hypot(pos[0] - self.capsule.x, pos[2] - self.capsule.z) # ---------- ticks ---------- diff --git a/testbed/tests/test_bridge.py b/testbed/tests/test_bridge.py index ec18286..5966229 100644 --- a/testbed/tests/test_bridge.py +++ b/testbed/tests/test_bridge.py @@ -247,3 +247,53 @@ async def test_move_accepts_duration_zero_default(bridge): res = (await client.call_tool("move", {"forward": 1.0})).output assert res["moved"] == pytest.approx(1.0) assert res["duration"] == 0.0 + + +async def test_report_verifies_near_marker(bridge): + t = InProcessTransport.start(bridge) + async with t: + client = AICCClient(t) + async with client: + await client.handshake() + out = (await client.call_tool("report", {"discovery": "triangle"})).output + assert out["success"] is True + assert out["verified"] is False # far away from spawn + pos = bridge.world.marker_world_pos() + bridge.world.capsule.x = pos[0] + bridge.world.capsule.z = pos[2] + out2 = (await client.call_tool("report", {})).output + assert out2["verified"] is True + assert "verified" in out2["message"] + + +async def test_turn_with_duration_is_smooth(bridge): + t1 = InProcessTransport.start(bridge) + t2 = InProcessTransport.start(bridge) + async with t1, t2: + mover = AICCClient(t1) + viewer = AICCClient(t2) + async with mover, viewer: + await mover.handshake() + await viewer.handshake() + task = asyncio.create_task( + mover.call_tool("turn", {"yaw_deg": 90.0, "duration": 0.8}) + ) + await asyncio.sleep(0.3) + mid = (await viewer.call_tool("proprioception", {})).output["rotation"]["yaw_deg"] + assert 0.0 < mid < 90.0, f"expected intermediate rotation, got {mid}" + res = (await task).output + assert res["rotation"]["yaw_deg"] == pytest.approx(135.0, abs=0.5) + + +async def test_look_at_crate(bridge): + t = InProcessTransport.start(bridge) + async with t: + client = AICCClient(t) + async with client: + await client.handshake() + out = (await client.call_tool("look_at", {"target": "crate_red"})).output + dx = bridge.world.boxes[0].cx - bridge.world.capsule.x + dz = bridge.world.boxes[0].cz - bridge.world.capsule.z + assert out["rotation"]["yaw_deg"] == pytest.approx( + math.degrees(math.atan2(dx, dz)) % 360.0, abs=0.5 + ) diff --git a/testbed/tests/test_world.py b/testbed/tests/test_world.py index 8af0701..a5a925d 100644 --- a/testbed/tests/test_world.py +++ b/testbed/tests/test_world.py @@ -131,3 +131,51 @@ def test_describe_layout(): assert d["room"]["width"] == ROOM_SIZE assert len(d["obstacles"]) == 3 assert d["beacon"]["id"] == "beacon" + + +def test_marker_is_on_a_back_face(): + """The triangle must sit on a face pointing away from the spawn corner.""" + for _ in range(20): + room = Room() + pos = room.marker_world_pos() + assert pos is not None + box = next(b for b in room.boxes if b.id == room.marker["box"]) + face = room.marker["face"] + # spawn is at (1.5, 1.5); the marker face normal must point away from it + to_spawn = (1.5 - box.cx, 1.5 - box.cz) + normal = { + "x0": (-1, 0), "x1": (1, 0), "z0": (0, -1), "z1": (0, 1), + }[face] + assert normal[0] * to_spawn[0] + normal[1] * to_spawn[1] < 0 + assert pos[0] == pytest.approx(box.x_max()) if face == "x1" else True + + +def test_marker_not_visible_from_spawn(): + """Looking toward the beacon from spawn must not show the triangle.""" + from testbed.room.render import Raycaster + + room = Room() + # Spawn looks at the beacon; the marker is on a back face, so the anchor + # pixel of the marker must be occluded (different surface) or off-frame. + pos = room.marker_world_pos() + assert pos is not None + # Stand on the marker's side of the crate, facing it. + face = room.marker["face"] + offsets = { + "x0": (-1.0, 0.0, 90.0), # west face: stand west, face +x + "x1": (1.0, 0.0, 270.0), # east face: stand east, face -x + "z0": (0.0, -1.0, 0.0), # south face: stand south, face +z + "z1": (0.0, 1.0, 180.0), # north face: stand north, face -z + }[face] + room.capsule.x = pos[0] + offsets[0] + room.capsule.z = pos[2] + offsets[1] + room.capsule.yaw_deg = offsets[2] + img, _ = Raycaster().render(room) + px = img.load() + orange = 0 + for y in range(0, 120, 2): + for x in range(0, 160, 2): + r, g, b = px[x, y] + if r > 200 and 100 < g < 190 and b < 90: + orange += 1 + assert orange > 3, "triangle must be visible when looking at the marker face"