diff --git a/configs/tera_crow_nav.toml b/configs/tera_crow_nav.toml new file mode 100644 index 0000000..55520e0 --- /dev/null +++ b/configs/tera_crow_nav.toml @@ -0,0 +1,25 @@ +# tera-crow-nav — fly a crow to a waypoint on sixteen replies. +# See environments/tera_spatial for what it measures and where these two integers +# came from (`measure_hold.py`, not judgement). +# +# uv run --project environments/tera_spatial eval @ configs/tera_crow_nav.toml \ +# --model brain-qwen38-dspark \ +# --client.base-url http://100.127.247.67:8001/v1 \ +# --client.api-key-var SPARK_API_KEY --no-push --no-rich -c 8 +# +# `node` >= 22.18 must be on PATH: the environment is TypeScript and is executed, +# not translated. +num_tasks = 16 +num_rollouts = 2 + +[env.taskset] +id = "tera-crow-nav" +num_tasks = 16 +max_turns = 16 # replies. Sixteen is where the ceiling first becomes reachable. +max_hold = 12 # steps one reply is held for. 16 x 12 = 192 against a 300-step cap, + # so what binds is the turn budget and never the simulator's own. +default_hold = 6 # what an unreadable reply costs. Never zero. +seed_base = 115 + +[env.agent.harness] +id = "null" # one completion per turn, no tools; the env programs the loop diff --git a/environments/tera_spatial/README.md b/environments/tera_spatial/README.md index 955e049..c680827 100644 --- a/environments/tera_spatial/README.md +++ b/environments/tera_spatial/README.md @@ -24,13 +24,148 @@ inside Tera, and only ever carried across the pipe. | `tera_spatial/closure.py` | the import walk both the sync script and the gate use | | `scripts/sync_tera.py` | reproduces the vendoring from a `tera` checkout | | `tests/test_replay.py` | the replay gate | +| `tera_spatial/spatial.py` | the shared half of a spatial taskset: the resident worker, the reply grammar, the episode loop, the reward | +| `tera_crow_nav/` | the first taskset — `tera-crow-nav`, fly a crow to a waypoint on sixteen replies | +| `measure_hold.py` | the hold and turn-budget sweep the two integers came from | +| `tests/test_crow_nav.py` | the taskset's gate: the grammar, the reward, and a flight flown off the rendered panel | -The tasksets are not here yet. This is the bridge and its correctness proof. +`tera-office-nav` and `tera-california-flight` follow. `tera-drive-101` ships +labelled a bridge-correctness environment or not at all: its return is monotone in +throttle and any positive constant scores about 0.95 of the baseline, which is a +number with no house rule 4 behind it. + +## The crux: a language model cannot emit sixty control vectors a second + +These are continuous-control environments and the agent is text. So a reply is a +DECISION, not a frame: one control vector plus the number of simulation steps to +hold it for, and then the model is shown where the vehicle ended up. + +The two integers that decide everything — how many replies, and how long one may +be held — are measured. `measure_hold.py` flies Tera's own scripted baseline at +every refresh rate over sixteen scenario x seed episodes: + +``` + hold return worst steps calls goal turns hold return goal + 1 5.484 5.059 66.1 66.1 100% 1 12 0.031 0% + 2 5.484 5.059 66.1 33.2 100% 2 12 0.317 0% + 3 5.484 5.060 66.1 22.4 100% 4 12 1.039 0% + 5 5.479 5.056 66.1 13.5 100% 6 11 1.700 0% + 8 4.906 0.416 97.6 12.6 94% 8 11 3.420 44% + 12 4.401 -0.467 127.6 11.2 88% 10 8 4.595 75% + 16 3.643 -0.577 153.4 10.1 75% 12 7 4.974 88% + 24 1.392 -0.689 280.1 12.1 44% 14 6 5.409 100% + 32 -2.807 -2.923 300.0 10.0 0% 16 5 5.479 100% + 64 -3.137 -3.658 238.3 4.0 0% 20 4 5.483 100% + 300 -3.764 -4.906 149.7 1.0 0% 32 3 5.484 100% + +inaction floor: -4.161 crow-nav-v1, 16 episodes per row +``` + +(the two tables `measure_hold.py` prints, set side by side; rows at hold 48, 100 +and 150 are dropped for width.) + +The left table is house rule 4's premise: **open-loop fails outright.** One +decision held for the whole flight returns -3.764 against an inaction floor of +-4.161 and reaches the waypoint none of the time — a bird that sets a course and +leaves does no better than a bird that does nothing. A plan is not a policy in a +world with momentum, which is why the grammar is one vector per reply and not a +program. + +The right table is the conclusion, and it is the reason `max_turns = 16`: the +budget BINDS all the way up to it. At eight turns the best possible constant hold +reaches the waypoint 44% of the time, at twelve 88%, and only at fourteen does it +reach every one. Sixteen is the first budget at which the ceiling is reachable, +and it is one turn of slack, not fifteen. + +`max_hold = 12` is the longest hold at which the baseline still reaches most +waypoints. `16 x 12 = 192` against a 300-step simulator cap, so what binds is the +turn budget and never the simulator's own. + +## The rewards + +Two, both products, both from numbers TypeScript computed: + +| | | +|---|---| +| `flight` (0.70) | the episode's return against nine tenths of what Tera's scripted controller returned on the same scenario at the same seed | +| `economy` (0.30) | that quality, multiplied by having arrived and by how directly | + +`flight` is smooth because goal attainment is already the biggest term inside it — +the simulator pays +3 of a ~5.5 return for reaching the waypoint — so a near-miss +scores about four tenths and a wander scores zero. + +⚠️ **Goal attainment multiplies and is never a term of its own.** It is already +inside the return as the success bonus, and a separate additive `gate` beside the +sum double-counts it. That is the exact failure `probe.py` records having paid +for. Efficiency multiplies too: a flight that never arrived cannot be quick about +arriving, and standing beside `flight` it would pay a crash for being fast. + +`gate`, `quality`, `step_ratio`, `ts_return`, `hold_mean`, `malformed_turns` and +`replay_ok` are recorded and never rewarded. + +**`replay_ok` is the bridge's claim asserted once per episode**, not once per +commit: every graded flight is replayed in a fresh TypeScript environment and the +metric is whether it reached the same FNV-1a-64 checksum, the same step count and +the same return. A divergence is recorded, never raised — an episode that flew is +data, and losing it to an assertion would hide the thing the metric exists to +surface. + +## Running it + +```bash +uv run --project environments/tera_spatial eval @ configs/tera_crow_nav.toml \ + --model brain-qwen38-dspark \ + --client.base-url http://100.127.247.67:8001/v1 \ + --client.api-key-var SPARK_API_KEY \ + --no-push --no-rich -c 8 -o outputs/run-/tera-crow-nav +``` + +⚠️ `eval` exits 0 even when every rollout errors. Count non-empty `rewards`; see +`docs/FIRST_EVAL.md`. + +## What it measured + +`outputs/run-20260821-1627/tera-crow-nav`, 16 tasks x 2 rollouts, model +`brain-qwen38-dspark` on spark-1, 12 minutes at `-c 8`. **32 episodes attempted, +32 scored, 0 errored, 0 provider errors.** + +| | | +|---|---| +| `reward_mean_attempted` | **0.4603** | +| `flight` (0.70) | 0.4894 | +| `economy` (0.30) | 0.3926 | +| waypoint reached | 14 of 32 | +| flight-envelope contact | 2 of 32 | +| out of turns, still flying | 16 of 32 | +| malformed replies | **0** of 412 | +| `replay_ok` | **1.000 on all 32** | + +Three things in that table are worth more than the headline. + +**The turn budget binds, in the run and not only in the sweep.** Turns used were +7, 8, 9, 10, 12, 14 and 16 — mean 12.9, and **sixteen of the thirty-two rollouts +spent every reply they had** and were still airborne when the budget ran out. The +one that scored zero on `train-east-crosswind` did so by orbiting: 128 steps +flown, yaw -90°, and 52.8 m still between it and the waypoint on its last reply. + +**The gate carries a gradient.** Goal attainment fired on 43.75% of rollouts. +Across the first seven Arena environments a `gate` component scored exactly 0.000, +max 0.000, in four of them — a quarter to a third of the reward mass with nothing +in it. Here it discriminates, and per-scenario it discriminates sharply: +`train-north-climb` 0.863, `dev-west-return` 0.728, `dev-south-descent` 0.250, +`train-east-crosswind` 0.000. + +**`eval.log` and `traces.jsonl` agree.** The mean over the log's own 32 `reward=` +lines is 0.4603, the same number as `sum(score x weight)` over the file. That is +not free: rewards recorded after `interaction.close()` land in the file and not in +the log, and the first cut of this env printed `reward=0.000` for an episode that +scored 1.000. Grade before closing. ## The gate ```bash -uv run python -m unittest discover -s environments/tera_spatial/tests -v +uv run --project environments/tera_spatial \ + python -m unittest discover -s environments/tera_spatial/tests -v ``` 32 episodes — 4 environments x 4 public scenarios x 2 seeds — each driven from @@ -40,6 +175,13 @@ environment, each tried twice: raw, where the envelope checksum catches it, and **re-sealed with a checksum Tera itself recomputed**, where only `replay()` re-running the simulator can. All must be rejected. +`test_crow_nav.py` adds fifteen more. The reward's floor and ceiling are flown +rather than asserted from a fixture, and the strongest of them is the rendering's +own claim: a policy that reads **nothing but the rendered panel** — the same +characters the model is shown, parsed back out of them with a regex — reaches all +four waypoints inside the shipped budget. An observation encoding that stops +carrying enough to fly on stops passing that test. + `node` >= 22.18 is required — the vendored sources are raw `.ts` and are type-stripped, not compiled. Set `TERA_NODE` to point at a specific binary. diff --git a/environments/tera_spatial/measure_hold.py b/environments/tera_spatial/measure_hold.py new file mode 100644 index 0000000..200d048 --- /dev/null +++ b/environments/tera_spatial/measure_hold.py @@ -0,0 +1,160 @@ +"""How long a crow can be flown blind — the measurement the turn budget is set from. + +A language model does not emit sixty control vectors a second. It emits a few +dozen replies in a whole episode. So a spatial taskset has to decide how many +simulation steps one reply is held for, and that decision is the environment: +hold too little and the model cannot reach the waypoint inside any budget worth +running; hold too much and the episode is open-loop, which is to say the turn +budget does not bind and rule 4 has nothing to say. + +This file measures the frontier rather than guessing at it. It flies Tera's own +scripted baseline — the reward denominator, unmodified — at hold lengths from 1 +to the whole episode, and reports for each one the return, the steps, the +decisions and whether the waypoint was reached. Only the refresh rate changes; +the policy, the scenario and the seed do not. The number that matters is where +the return falls off, because that is the boundary between "the model is +flying" and "the model set a course and left". + + uv run --no-project python environments/tera_spatial/measure_hold.py + uv run --no-project python environments/tera_spatial/measure_hold.py --env drive-101-v1 + +Nothing here is a reward and nothing here runs a model. It is the evidence for +two integers in `configs/tera_crow_nav.toml`. +""" + +from __future__ import annotations + +import argparse +import statistics +from collections.abc import Sequence + +from tera_spatial import ENVIRONMENT_IDS, TeraWorker + +HOLDS = (1, 2, 3, 5, 8, 12, 16, 24, 32, 48, 64, 100, 150, 300) +SEEDS = (115, 2718, 40_902, 7_331) +BUDGETS = (1, 2, 4, 6, 8, 10, 12, 14, 16, 20, 24, 32) + + +def sweep( + worker: TeraWorker, env_id: str, holds: Sequence[int], seeds: Sequence[int] +) -> list[dict[str, float]]: + """One row per hold length, averaged over every public scenario at every seed.""" + manifest = next(m for m in worker.manifests() if m["id"] == env_id) + scenarios = [ + (split, scenario_id) + for split in ("train", "dev") + for scenario_id in manifest["scenarioIds"][split] + ] + rows = [] + for hold in holds: + returns, steps, decisions, reached = [], [], [], 0 + for split, scenario_id in scenarios: + for seed in seeds: + result = worker.oracle( + env_id, seed, {"split": split, "id": scenario_id}, hold=hold + ) + returns.append(result["cumulativeReward"]) + steps.append(result["steps"]) + decisions.append(result["decisions"]) + reached += result["terminalReason"] == "goal" + rows.append( + { + "hold": hold, + "mean_return": statistics.fmean(returns), + "worst_return": min(returns), + "mean_steps": statistics.fmean(steps), + "mean_decisions": statistics.fmean(decisions), + "goal_rate": reached / len(returns), + "episodes": len(returns), + } + ) + return rows + + +def budget_ladder( + worker: TeraWorker, + env_id: str, + turns: Sequence[int], + max_hold: int, + seeds: Sequence[int], +) -> list[dict[str, float]]: + """The best the baseline can do under a turn budget — the rule-4 ladder. + + For each budget the baseline is flown at every legal hold length and the best + result is kept, with the episode cut off at `turns x hold` steps because that + is all the budget buys. A budget that does not change the answer is a budget + that does not bind, and this table is where that shows. + """ + manifest = next(m for m in worker.manifests() if m["id"] == env_id) + scenarios = [ + (split, scenario_id) + for split in ("train", "dev") + for scenario_id in manifest["scenarioIds"][split] + ] + rows = [] + for budget in turns: + best = None + for hold in range(1, max_hold + 1): + returns, reached = [], 0 + for split, scenario_id in scenarios: + for seed in seeds: + result = worker.oracle( + env_id, + seed, + {"split": split, "id": scenario_id}, + max_steps=budget * hold, + hold=hold, + ) + returns.append(result["cumulativeReward"]) + reached += result["terminalReason"] == "goal" + mean = statistics.fmean(returns) + if best is None or mean > best["mean_return"]: + best = { + "turns": budget, + "hold": hold, + "mean_return": mean, + "worst_return": min(returns), + "goal_rate": reached / len(returns), + } + rows.append(best) + return rows + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--env", default="crow-nav-v1", choices=ENVIRONMENT_IDS) + parser.add_argument("--seeds", type=int, nargs="+", default=list(SEEDS)) + parser.add_argument("--max-hold", type=int, default=12) + args = parser.parse_args() + + with TeraWorker() as worker: + rows = sweep(worker, args.env, HOLDS, args.seeds) + floor = worker.oracle(args.env, args.seeds[0], {"split": "train"}, policy="inaction") + ladder = budget_ladder(worker, args.env, BUDGETS, args.max_hold, args.seeds) + + print(f"{args.env} — Tera's scripted baseline, flown at one decision every `hold` steps") + print(f"{'hold':>5} {'return':>9} {'worst':>9} {'steps':>7} {'calls':>7} {'goal':>6}") + for row in rows: + print( + f"{row['hold']:>5} {row['mean_return']:>9.3f} {row['worst_return']:>9.3f} " + f"{row['mean_steps']:>7.1f} {row['mean_decisions']:>7.1f} {row['goal_rate']:>6.0%}" + ) + print( + f"\ninaction floor (seed {args.seeds[0]}, train): " + f"{floor['cumulativeReward']:.3f} over {floor['steps']} steps " + f"({floor['terminalReason']})" + ) + print(f"{rows[0]['episodes']} episodes per row.") + + print(f"\nUnder a turn budget, hold capped at {args.max_hold} — best constant hold per budget") + print(f"{'turns':>6} {'hold':>5} {'return':>9} {'worst':>9} {'goal':>6}") + for row in ladder: + print( + f"{row['turns']:>6} {row['hold']:>5} {row['mean_return']:>9.3f} " + f"{row['worst_return']:>9.3f} {row['goal_rate']:>6.0%}" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/environments/tera_spatial/pyproject.toml b/environments/tera_spatial/pyproject.toml index bf90634..ea72b52 100644 --- a/environments/tera_spatial/pyproject.toml +++ b/environments/tera_spatial/pyproject.toml @@ -1,25 +1,25 @@ [project] name = "tera-spatial" version = "0.1.0" -description = "tera-spatial — Arena's bridge to Tera's four renderer-independent spatial environments, executed in TypeScript and graded in Python." +description = "tera-spatial — Arena's bridge to Tera's four renderer-independent spatial environments, executed in TypeScript and graded in Python, and the tasksets that sit on it." requires-python = ">=3.11" dependencies = ["verifiers"] -# One directory, four tasksets: the environments share a vendored simulator and +# One directory, several tasksets: the environments share a vendored simulator and # a single worker, so splitting them into four wheels would ship the same 283 KB -# of TypeScript four times. B1's discovery reads this key instead of assuming -# one wheel is one taskset. +# of TypeScript four times. Discovery reads this key instead of assuming one wheel +# is one taskset. +# +# Only the ones that exist are listed. `tera-office-nav` and `tera-california-flight` +# follow; `tera-drive-101` ships labelled a bridge-correctness environment or not at +# all, because its return is monotone in throttle and any positive constant scores +# ~0.95 of the baseline — a number with no house rule 4 behind it. [tool.arena] -tasksets = [ - "tera-drive-101", - "tera-office-nav", - "tera-crow-nav", - "tera-california-flight", -] +tasksets = ["tera-crow-nav"] [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["tera_spatial"] +packages = ["tera_spatial", "tera_crow_nav"] diff --git a/environments/tera_spatial/tera_crow_nav/__init__.py b/environments/tera_spatial/tera_crow_nav/__init__.py new file mode 100644 index 0000000..3caa522 --- /dev/null +++ b/environments/tera_spatial/tera_crow_nav/__init__.py @@ -0,0 +1,18 @@ +"""tera-crow-nav — the first Tera spatial taskset, over the NDJSON bridge. + +The taskset id resolves to this module name, so both plugin classes are exported +from here: the `Taskset` the run loads its tasks from, and the `Env` that programs +the flight loop host-side. `TeraSpatialEnv` lives in `tera_spatial` because the +loop is the same for every spatial environment; what makes this one crow-nav is +`CrowNavTaskset`. +""" + +from tera_crow_nav.taskset import CrowNavConfig, CrowNavTaskset +from tera_spatial.spatial import TeraSpatialEnv, TeraSpatialEnvConfig + +__all__ = [ + "CrowNavConfig", + "CrowNavTaskset", + "TeraSpatialEnv", + "TeraSpatialEnvConfig", +] diff --git a/environments/tera_spatial/tera_crow_nav/taskset.py b/environments/tera_spatial/tera_crow_nav/taskset.py new file mode 100644 index 0000000..ff051d4 --- /dev/null +++ b/environments/tera_spatial/tera_crow_nav/taskset.py @@ -0,0 +1,211 @@ +"""tera-crow-nav: fly a crow to a waypoint on sixteen replies. + +Tera's `crow-nav-v1` is a three-dimensional waypoint problem over the same flight +controller the renderer flies — momentum, a bank-coupled turn rate, a finite +flight-energy budget, thermals, drag that rises with the square of airspeed, and +an envelope at 2 m and 40 m that ends the flight if you touch it. The simulator +runs at 10 Hz for up to 300 steps. A language model does not fly at 10 Hz. + +So a reply is a decision, not a frame: one control vector plus the number of +steps to hold it for. Sixteen replies, twelve steps each at most. + +**Those two integers are the environment**, and they are measured. `measure_hold.py` +flies Tera's own scripted baseline at every refresh rate over sixteen scenario x +seed episodes: + + hold return goal turns best hold return goal + 1 5.484 100% 1 12 0.031 0% + 5 5.479 100% 4 12 1.039 0% + 8 4.906 94% 8 11 3.420 44% + 12 4.401 88% 12 7 4.974 88% + 24 1.392 44% 14 6 5.409 100% + 32 -2.807 0% 16 5 5.479 100% + 300 -3.764 0% 20 4 5.483 100% + +Read the left table for house rule 4's premise and the right one for its +conclusion. Open-loop fails outright — one decision for the whole flight returns +-3.764 against an inaction floor of -4.161, which is to say a bird that sets a +course and leaves does no better than a bird that does nothing. And the budget +BINDS all the way to sixteen: at eight turns the best possible constant hold +reaches the waypoint 44% of the time, at twelve 88%, and only at fourteen does it +reach every one. Sixteen is the first budget at which the ceiling is reachable and +fifteen turns of slack is not on offer. + +Two rewards, both products: + + flight the episode's return against nine tenths of what Tera's scripted + controller returned on the same scenario at the same seed. Smooth: + reaching the waypoint is already the biggest term inside it — the + simulator pays +3 of a ~5.5 return for arriving — so a near-miss + scores about four tenths and a wander scores zero. + economy that same quality, multiplied by having arrived and by how directly. + Goal attainment MULTIPLIES and is never a term of its own; standing + beside the sum it would double-count a bonus the return already + contains. Efficiency multiplies too, because a flight that never + arrived cannot be quick about arriving. + +Minimum-power inaction returns -4.161 and scores exactly 0.000 on both. The +scripted controller flown inside this budget scores 1.000 on both. `gate`, +`quality`, `step_ratio` and `replay_ok` are recorded and never rewarded. +""" + +from __future__ import annotations + +import math +from typing import Any, ClassVar + +import verifiers.v1 as vf + +from tera_spatial.spatial import ( + SpatialConfig, + SpatialData, + SpatialTask, + TeraSpatialTaskset, + TurnView, +) + +MISSION = """You are flying a crow to a waypoint. + +The simulator advances in fixed steps of {step:g} s. You do not fly it step by step: +each reply is ONE set of control positions plus a `hold`, and the crow holds those +positions for that many steps before you are shown where it ended up. + + You get {turns} replies. A reply may be held for 1 to {max_hold} steps. + That is {budget} steps of flight in total if you use every one of them. + +The controls, each a number from -1 to 1: + + forward airspeed demand. -1 is minimum power (about 4 m/s), +1 is full + (about 16 m/s). Flapping hard drains flight energy, and a tired crow + has less speed and climb authority than a rested one. + turn rate of turn, up to about 1.75 rad/s at full deflection and full + speed. The crow banks into it, which adds a little more. + pitch nose attitude, up to about 0.72 rad. Pitch trades airspeed for + height and back; it is not the climb control. + climb powered vertical demand, up to about 5 m/s at full deflection. + glide true or false. Gliding recovers flight energy and holds about + 7.5 m/s, but you give up powered climb while you do it. + +The world: + + The waypoint counts as reached within {radius:g} m in three dimensions, and pays a + large bonus when you get there. + You must stay between {floor:g} m and {ceiling:g} m altitude and inside +/-{bound:g} m in x and z. + Touching any of those ends the flight immediately and costs more than the + waypoint was worth. + Every step costs a little time, and power, turn, pitch and climb demand each + cost a little more. Pointing away from the waypoint costs a little. + +Reply with ONE ```json code block and nothing else that matters: + +```json +{{"forward": 0.5, "turn": 0.0, "pitch": 0.0, "climb": 0.0, "glide": false, "hold": 6}} +``` + +Any control you leave out keeps the value it already had. A reply with no readable +JSON object holds your last controls for {default_hold} steps and costs you the turn — there +is no way to pause.""" + + +def _degrees(radians: float) -> float: + return math.degrees(radians) + + +def _wrap(value: float) -> float: + return (value + math.pi) % (2 * math.pi) - math.pi + + +class CrowNavConfig(SpatialConfig): + """Crow-nav's knobs are the shared ones. The defaults are the measured ones and + a config that changes `max_turns` or `max_hold` is changing the environment, not + tuning it — see the two tables in this module's docstring.""" + + +class CrowNavTaskset(TeraSpatialTaskset, vf.Taskset[SpatialTask, CrowNavConfig]): + """Sixteen waypoint flights over the four public crow scenarios. + + The generic form is not decoration: verifiers v1 resolves a taskset's config + through `__orig_bases__`, so `vf.Taskset[SpatialTask, CrowNavConfig]` in the + bases is the whole of the wiring. There is no `CONFIG` classvar to set, and a + taskset that omits this reads every per-environment setting off the base config + and silently ignores the TOML. + """ + + ENV_ID: ClassVar[str] = "crow-nav-v1" + + def mission(self, manifest: dict[str, Any], data: SpatialData) -> str: + return MISSION.format( + step=manifest["fixedStepSeconds"], + turns=data.max_turns, + max_hold=data.max_hold, + default_hold=data.default_hold, + budget=data.max_turns * data.max_hold, + radius=2.5, + floor=2, + ceiling=40, + bound=120, + ) + + def board(self, view: TurnView) -> str: + """The state vector, rendered as a panel a reader could fly on. + + Every number here is the environment's own published observation or plain + arithmetic over it: `distanceToGoalM` and the deltas are fields on the + manifest, and the bearing is `atan2` of two of them against a third. That + trigonometry is given away deliberately. What crow-nav measures is whether a + model can fly a body with momentum to a point inside a turn budget, not + whether it can do `atan2` in its head with thinking switched off — and the + scripted controller the reward is normalised against reads exactly these + fields, so nothing here moves the denominator. + + What it does NOT contain is anything the simulator has not already + published: this is a dict that crossed a JSON pipe, with no handle in it to + walk back to the model. + """ + o = view.observation + bearing = _degrees(_wrap(math.atan2(-o["deltaX"], -o["deltaZ"]) - o["yaw"])) + contact = o["altitudeBoundContact"] + seconds = view.steps * 0.1 + lines = [ + f"turn {view.turn} of {view.turn + view.turns_left - 1}" + f" · step {view.steps} of {view.budget}" + f" · {seconds:.1f} s flown", + "", + f"position x {o['x']:>8.1f} y {o['y']:>6.1f} z {o['z']:>8.1f}" + f" altitude {o['y']:.1f} m of 2–40", + f"attitude yaw {_degrees(o['yaw']):>6.0f}° pitch {_degrees(o['pitch']):>4.0f}°" + f" speed {o['speedMps']:.1f} m/s vertical {o['verticalSpeedMps']:+.1f} m/s", + f"waypoint x {o['goalX']:>8.1f} y {o['goalY']:>6.1f} z {o['goalZ']:>8.1f}", + f" {o['distanceToGoalM']:.1f} m away · {o['deltaY']:+.1f} m of height to make up" + f" · bearing {bearing:+.0f}° (a positive turn closes it)", + f"envelope {'clear' if contact == 'none' else contact + ' altitude bound — you are on it'}", + ] + if view.last is None: + lines.append("controls at rest; you have not flown yet") + else: + action = view.last.action + lines.append( + "you flew " + + " ".join( + f"{name} {_show(action.get(name, 0))}" + for name in ("forward", "turn", "pitch", "climb", "glide") + ) + + f" held {view.last.hold} step{'s' if view.last.hold != 1 else ''}" + ) + if view.last.problem: + lines.append(f"⚠ {view.last.problem}, so those were held again") + lines.append("") + lines.append( + f"{view.turns_left} repl{'y' if view.turns_left == 1 else 'ies'} left, " + f"{view.budget - view.steps} steps of flight. Your controls?" + ) + return "\n".join(lines) + + +def _show(value: Any) -> str: + if isinstance(value, bool): + return "yes" if value else "no" + return f"{float(value):+.2f}" + + +__all__ = ["CrowNavConfig", "CrowNavTaskset"] diff --git a/environments/tera_spatial/tera_spatial/bridge.py b/environments/tera_spatial/tera_spatial/bridge.py index 0be02af..3a33626 100644 --- a/environments/tera_spatial/tera_spatial/bridge.py +++ b/environments/tera_spatial/tera_spatial/bridge.py @@ -164,15 +164,27 @@ class TeraWorker: scenario: str | dict[str, Any], policy: str = "scripted", max_steps: int | None = None, + hold: int = 1, ) -> dict[str, Any]: """Run one of the baselines Tera exports, inside Tera. The scripted return is the denominator every reward here is normalised against, so it is not a thing to translate: `src/arena/index.ts` already exports all four policies and the worker calls them. + + `hold` is how many simulation steps one decision is held for. It is 1 for + the denominator — the baseline flying at the simulator's own rate — and + larger when the question is what that same policy scores at a language + model's refresh rate. See `measure_hold.py`. """ return self.request( - "oracle", env=env_id, seed=seed, scenario=scenario, policy=policy, maxSteps=max_steps + "oracle", + env=env_id, + seed=seed, + scenario=scenario, + policy=policy, + maxSteps=max_steps, + hold=hold, ) # -- lifecycle -------------------------------------------------------- diff --git a/environments/tera_spatial/tera_spatial/spatial.py b/environments/tera_spatial/tera_spatial/spatial.py new file mode 100644 index 0000000..1e2ca0f --- /dev/null +++ b/environments/tera_spatial/tera_spatial/spatial.py @@ -0,0 +1,578 @@ +"""The part of a Tera spatial taskset that is not about crows. + +A language model cannot emit sixty control vectors a second, and pretending +otherwise is the only real design question these environments pose. The answer +this file implements: **one control vector per reply, held for a number of +simulation steps the model itself chooses.** A turn is a decision, not a frame, +and the two integers that decide everything — how many turns, and how long one +of them may be held — are measured, not guessed. `measure_hold.py` is that +measurement. + +The alternative, a short program executed open-loop, was rejected by the same +measurement: Tera's own scripted baseline flown at one decision every 32 steps +scores -2.807 against 5.484 at every frame, and reaches the waypoint 0% of the +time. A plan is not a policy in a world with momentum. + +What lives here is everything a second and third spatial taskset will want +unchanged: the resident worker, the reply grammar, the episode loop, and the +reward mapping. What does not live here is any rendering or any prompt — those +are the environment, and each one writes its own on `TeraSpatialTaskset`. + +Two rules the file exists to keep: + + never recompute in Python a number that came out of TypeScript + the return, the per-step checksums and the denominator all arrive from the + worker. `oracle_return` is Tera's own scripted baseline, run by Tera. + + the model sees the observation and nothing behind it + the dict handed to a renderer is the environment's published observation, + already across a JSON pipe. There is no live handle in it to walk back to + the simulator, which is the leak `grand-exchange-live` found in its own + per-turn view: a `market` attribute that carried the whole held-out series + and paid a foresight policy +9.4% over the oracle. +""" + +from __future__ import annotations + +import asyncio +import atexit +import json +import math +import re +import threading +from dataclasses import dataclass, field +from typing import Any, ClassVar + +from pydantic import Field + +import verifiers.v1 as vf + +from tera_spatial.bridge import TeraEpisode, TeraError, TeraWorker + +_BLOCK = re.compile(r"```(?:json)?\s*\n(.*?)```", re.DOTALL) + +#: How much of the baseline's return counts as full marks. The plateau around a +#: proportional controller and everything above it all score 1.000; the same nine +#: tenths `grand-exchange` uses, for the same reason — a denominator you have to +#: match exactly is an imitation score, not a capability one. +BAND = 0.90 + + +# --- the resident worker --------------------------------------------------- + +_worker: TeraWorker | None = None +_worker_lock = threading.Lock() + + +def shared_worker() -> TeraWorker: + """The one `node` process this eval talks to. + + It costs ~92 ms to start and ~19 us per round trip, so it is opened once and + shared: a worker per rollout would spend more time importing TypeScript than + simulating. `TeraWorker.request` serialises itself, and every caller here + reaches it through `asyncio.to_thread`, so concurrent episodes queue on the + pipe instead of on the event loop. + """ + global _worker + with _worker_lock: + if _worker is None: + _worker = TeraWorker() + atexit.register(_worker.close) + return _worker + + +# --- the reply grammar ----------------------------------------------------- + + +@dataclass(frozen=True) +class Command: + """One reply, as the simulator will receive it.""" + + action: dict[str, Any] + hold: int + problem: str | None = None + """What was unreadable about the reply, in words the model will be shown. None + when the reply was fine.""" + + +def parse_command( + reply: str, + *, + fields: tuple[str, ...], + max_hold: int, + default_hold: int, + previous: Command | None, + inaction: dict[str, Any], +) -> Command: + """The last JSON object in the reply, read against the manifest's action fields. + + Every field is optional and every one that is missing keeps whatever the last + command flew — a crow does not stop being a crow because a reply was short. + `hold` is clipped into `[1, max_hold]`. + + An unreadable reply is not an error and does not raise: it flies the previous + command for `default_hold` steps and says so on the next observation. Silence + is the one thing it must not be, because a reply that costs nothing is a reply + the turn budget does not bind. Raising instead would turn a formatting slip + into a crashed rollout, which scores nothing at all rather than scoring badly. + + `json.loads` is not strict JSON — it accepts `Infinity` and `NaN` as bare + literals and overflows `1e309` to `inf` — so non-finite numbers are dropped + rather than carried into a reward. `RecursionError` and `OverflowError` are + not `ValueError`, and both are reachable from a reply a model can emit. + """ + carried = dict(previous.action) if previous is not None else dict(inaction) + blocks = _BLOCK.findall(reply or "") + raw = blocks[-1] if blocks else (reply or "") + try: + parsed = json.loads(raw.strip()) + except (ValueError, RecursionError): + return Command(carried, default_hold, "your last reply held no JSON object") + if not isinstance(parsed, dict): + return Command(carried, default_hold, "your last reply held no JSON object") + + action = dict(carried) + read = 0 + for name in fields: + if name not in parsed: + continue + value = parsed[name] + if isinstance(value, bool): + action[name] = value + read += 1 + elif isinstance(value, (int, float)): + try: + if math.isfinite(value): + action[name] = float(value) + read += 1 + except OverflowError: + continue + + hold = default_hold + raw_hold = parsed.get("hold") + if isinstance(raw_hold, (int, float)) and not isinstance(raw_hold, bool): + try: + if math.isfinite(raw_hold): + hold = max(1, min(max_hold, int(raw_hold))) + except (OverflowError, ValueError): + hold = default_hold + + if read == 0: + return Command(action, hold, "your JSON named none of the controls") + return Command(action, hold) + + +# --- the episode ----------------------------------------------------------- + + +@dataclass +class Flight: + """One episode inside the worker, plus what Python is allowed to remember of it. + + Every number on this object came out of TypeScript. Nothing here adds two + rewards together or decides that an episode ended — `terminal_reason` is the + simulator's word, carried. + """ + + episode: TeraEpisode + max_turns: int + budget: int + observation: dict[str, Any] = field(default_factory=dict) + ts_return: float = 0.0 + steps: int = 0 + turns: int = 0 + malformed: int = 0 + holds: list[int] = field(default_factory=list) + terminal_reason: str | None = None + done: bool = False + last: Command | None = None + + def fly(self, command: Command) -> None: + """Hold one command for its steps, stopping the moment the simulator does.""" + self.turns += 1 + self.holds.append(command.hold) + self.malformed += command.problem is not None + self.last = command + for _ in range(min(command.hold, self.budget - self.steps)): + result = self.episode.step(command.action) + self.steps += 1 + self.ts_return += result["reward"] + self.observation = result["observation"] + if result["terminated"] or result["truncated"]: + self.terminal_reason = result["info"]["terminalReason"] + self.done = True + return + if self.steps >= self.budget or self.turns >= self.max_turns: + # Out of budget, not out of world: the episode is unfinished and the + # simulator has said nothing. `terminal_reason` stays None, which is + # what makes `gate` zero. + self.done = True + + +# --- the taskset and the env ----------------------------------------------- + + +class SpatialData(vf.TaskData): + """What rebuilds the episode, and nothing the model is not shown. + + The scenario is public and the seed reproduces it exactly, so there is no + held-out corpus to leak here — what the model does not see is the next + hundredth of a second, which no field can carry. `oracle_return` is the + denominator, not the answer: it is what Tera's scripted controller scored on + this scenario, and a model that knew the number would still have to fly. + """ + + env_id: str + seed: int + scenario_id: str + split: str + max_turns: int + max_hold: int + default_hold: int + band: float + oracle_return: float + oracle_steps: int + + +class SpatialTask(vf.Task[SpatialData]): + pass + + +class SpatialConfig(vf.TasksetConfig): + num_tasks: int = Field(16, ge=1) + max_turns: int = Field(16, ge=1) + """Replies the model gets. Measured, not chosen: `measure_hold.py` puts the + scripted baseline at 0.031 return and 0% goals on one turn, 3.420 and 44% on + eight, and 5.479 and 100% on sixteen — against 5.484 at every frame. Sixteen + is where the ceiling becomes reachable and one turn below it is where it is + not, which is house rule 4 in two integers.""" + max_hold: int = Field(12, ge=1) + """Simulation steps one reply may be held for. Twelve is the longest hold at + which the baseline still reaches most waypoints (88%); at sixteen it is 75% + and at thirty-two it is none. `max_turns x max_hold` is 192 against a 300-step + simulator cap, so the budget that binds is the turn budget and never the + simulator's own.""" + default_hold: int = Field(6, ge=1) + """What an unreadable reply is held for. Not zero: a reply that costs no steps + is a free turn, and a free turn is a budget that does not bind.""" + band: float = Field(BAND, gt=0.0, le=1.0) + splits: list[str] = Field(default_factory=lambda: ["train", "dev"]) + seed_base: int = Field(115, ge=0) + + +@dataclass(frozen=True) +class TurnView: + """Everything a renderer may read. Deliberately a plain dict of floats and no + handle: whatever is in here is what the model sees, so nothing that is not + already across the JSON pipe can get in by accident.""" + + observation: dict[str, Any] + turn: int + turns_left: int + steps: int + budget: int + max_hold: int + last: Command | None + terminal_reason: str | None = None + + +class TeraSpatialTaskset(vf.Taskset[SpatialTask, SpatialConfig]): + """Tasks over one Tera environment: a scenario, a seed, and a denominator. + + Subclasses pin `ENV_ID` and write the two things that ARE the environment as + far as a language model is concerned — the mission it is briefed with, and how + a state vector is rendered as something a reader can fly on. Everything else + is the same for every spatial environment and is settled here. + """ + + ENV_ID: ClassVar[str] = "" + + def mission(self, manifest: dict[str, Any], data: SpatialData) -> str: + """The system prompt: the vehicle, the controls, the grammar, the budget.""" + raise NotImplementedError + + def board(self, view: TurnView) -> str: + """One turn's observation, rendered.""" + raise NotImplementedError + + # -- the shared half ---------------------------------------------------- + + def scenarios(self, manifest: dict[str, Any]) -> list[tuple[str, str]]: + return [ + (split, scenario_id) + for split in self.config.splits + for scenario_id in manifest["scenarioIds"].get(split, ()) + ] + + def load(self) -> list[SpatialTask]: + worker = shared_worker() + manifest = next(m for m in worker.manifests() if m["id"] == self.ENV_ID) + rota = self.scenarios(manifest) + if not rota: + raise ValueError( + f"{self.ENV_ID} has no scenarios in splits {self.config.splits!r}" + ) + tasks = [] + for i in range(self.config.num_tasks): + split, scenario_id = rota[i % len(rota)] + seed = self.config.seed_base + i + request = {"split": split, "id": scenario_id} + # The denominator is Tera's, run by Tera, on this exact scenario. A + # Python port of the four exported baselines would put the number every + # reward is divided by on the far side of a hand translation. + oracle = worker.oracle(self.ENV_ID, seed, request) + start = worker.open(self.ENV_ID, seed, request) + observation = start.observation + start.close() + data = SpatialData( + idx=i, + name=f"{scenario_id}@{seed}", + env_id=self.ENV_ID, + seed=seed, + scenario_id=scenario_id, + split=split, + max_turns=self.config.max_turns, + max_hold=self.config.max_hold, + default_hold=self.config.default_hold, + band=self.config.band, + oracle_return=float(oracle["cumulativeReward"]), + oracle_steps=int(oracle["steps"]), + ) + view = TurnView( + observation=observation, + turn=1, + turns_left=self.config.max_turns, + steps=0, + budget=self.config.max_turns * self.config.max_hold, + max_hold=self.config.max_hold, + last=None, + ) + tasks.append( + SpatialTask( + data.model_copy( + update={ + "prompt": self.board(view), + "system_prompt": self.mission(manifest, data), + } + ), + self.config.task, + ) + ) + return tasks + + +class TeraSpatialEnvConfig(vf.EnvConfig): + """One seat, named `agent`. + + The name is the config key: `EnvConfig` is `extra="forbid"`, so a role called + anything else parses `[env.agent.harness]` as an unknown field and the run does + not load. It matches `SingleAgentEnvConfig` on purpose — every Arena config + already addresses `--env.agent.*`. + """ + + agent: vf.AgentConfig = vf.AgentConfig() + + +class TeraSpatialEnv(vf.Env[TeraSpatialEnvConfig]): + """Fly one episode, one reply at a time, and grade what the simulator recorded. + + The prompt is the first observation, so the model speaks first and the opening + reply is taken with a bare `turn()`. Every later turn hands it the state that + its own last command produced. A `terminated` Segment — a context limit, a + harness that gave up — ends the exchange mid-flight; the episode is still + graded, on the steps it actually flew. + """ + + async def run(self, task: vf.Task, agents: vf.Agents) -> None: + data: SpatialData = task.data # type: ignore[assignment] + taskset: TeraSpatialTaskset = self.taskset # type: ignore[assignment] + worker = shared_worker() + manifest = next(m for m in worker.manifests() if m["id"] == data.env_id) + fields = tuple(manifest["actionFields"]) + inaction = _inaction(manifest) + request = {"split": data.split, "id": data.scenario_id} + + episode = await asyncio.to_thread( + worker.open, data.env_id, data.seed, request + ) + flight = Flight( + episode=episode, + max_turns=data.max_turns, + budget=data.max_turns * data.max_hold, + observation=episode.observation, + ) + try: + async with agents.agent.interaction(task) as interaction: + segment = await interaction.turn() + while not segment.terminated and not flight.done: + command = parse_command( + segment.last_reply, + fields=fields, + max_hold=data.max_hold, + default_hold=data.default_hold, + previous=flight.last, + inaction=inaction, + ) + await asyncio.to_thread(flight.fly, command) + if flight.done: + break + view = TurnView( + observation=flight.observation, + turn=flight.turns + 1, + turns_left=data.max_turns - flight.turns, + steps=flight.steps, + budget=flight.budget, + max_hold=data.max_hold, + last=command, + ) + segment = await interaction.turn(taskset.board(view)) + # Graded before the interaction closes, not after. Closing is what + # finishes the rollout and prints its `reward=` line, and a reward + # recorded on the far side of that lands in `traces.jsonl` but not + # in the log — which would make `eval.log` and the file disagree + # about the same episode, and the runbook cross-checks one against + # the other. + envelope = await asyncio.to_thread(episode.trace) + rewards, metrics, info = await asyncio.to_thread( + score, data, flight, envelope, worker + ) + trace = interaction.trace + for name, (value, weight) in rewards.items(): + trace.record_reward(name, value, weight) + trace.record_metrics(metrics) + trace.info["tera"] = info + finally: + with _quiet(): + await asyncio.to_thread(episode.close) + + +def _inaction(manifest: dict[str, Any]) -> dict[str, Any]: + """A neutral action in the manifest's own field names. Only ever the seed for + turn one's carry-forward — the environment's published inaction baseline is a + different thing and belongs to `probe.py`, not to a fallback.""" + return {name: 0.0 for name in manifest["actionFields"]} + + +class _quiet: + """Closing a handle must not lose an episode that already flew.""" + + def __enter__(self) -> None: + return None + + def __exit__(self, exc_type, *_: object) -> bool: + return exc_type is not None and issubclass(exc_type, (TeraError, OSError)) + + +def score( + data: SpatialData, + flight: Flight, + envelope: dict[str, Any], + worker: TeraWorker, +) -> tuple[dict[str, tuple[float, float]], dict[str, float], dict[str, Any]]: + """Two rewards, both products, from numbers TypeScript computed. + + Returns `(rewards, metrics, info)` rather than writing to a trace, so the + arithmetic can be exercised without an agent, a model or a rollout — the + reward is the part of an environment that has to be right when nothing is + watching. + + `flight` is what the episode returned against nine tenths of what Tera's own + controller returned on the same scenario. It is smooth, and goal attainment is + already the largest term inside it — the simulator pays +3 of a ~5.5 return for + reaching the waypoint — so a near-miss scores about four tenths and a wander + scores nothing. + + `economy` is that same quality multiplied by having arrived and by how directly. + Goal attainment MULTIPLIES here and appears nowhere as a term of its own: the + success bonus is already inside the return, and adding it again beside the sum + is the double count `probe.py` records having paid for. Efficiency multiplies + too, because time-to-goal is not a thing a flight that never arrived can be + good at — as a term standing beside `flight` it would pay a crash for being + quick. + + Floor and ceiling: minimum-power inaction returns about -4.2, which clips to + exactly 0.000 on both. The scripted baseline flown inside this budget returns + 5.479 against a 4.936 target and arrives in the baseline's own step count, + which is 1.000 on both. + """ + target = data.band * data.oracle_return + quality = 0.0 if target <= 0 else max(0.0, min(1.0, flight.ts_return / target)) + gate = 1.0 if flight.terminal_reason == "goal" else 0.0 + steps = max(flight.steps, 1) + step_ratio = max(0.0, min(1.0, data.oracle_steps / max(steps, data.oracle_steps))) + + rewards = { + "flight": (quality, 0.70), + "economy": (gate * quality * step_ratio, 0.30), + } + metrics = { + "gate": gate, + "quality": quality, + "step_ratio": step_ratio, + "ts_return": flight.ts_return, + "oracle_return": data.oracle_return, + "oracle_steps": float(data.oracle_steps), + "steps": float(flight.steps), + "turns_used": float(flight.turns), + "turns_granted": float(data.max_turns), + "hold_mean": ( + sum(flight.holds) / len(flight.holds) if flight.holds else 0.0 + ), + "malformed_turns": float(flight.malformed), + "safety_contact": float( + flight.terminal_reason is not None + and flight.terminal_reason not in ("goal", "max-steps") + ), + # The bridge's whole claim, asserted once per episode rather than once + # per commit: the return that just became a reward is a return the + # simulator will reproduce from the actions alone. + "replay_ok": float(_replays(worker, data.env_id, envelope, flight)), + } + info = { + "env_id": data.env_id, + "scenario_id": data.scenario_id, + "split": data.split, + "seed": data.seed, + "terminal_reason": flight.terminal_reason, + "final_state_checksum": envelope["finalStateChecksum"], + "env_hash": envelope["envHash"], + "scenario_hash": envelope["scenarioHash"], + } + return rewards, metrics, info + + +def _replays( + worker: TeraWorker, env_id: str, envelope: dict[str, Any], flight: Flight +) -> bool: + """Re-run the trace in a fresh environment and require the same last bit. + + A divergence is recorded, never raised: an episode that flew is data, and + losing it to an assertion would hide the very thing this metric exists to + surface. `probe.py` is where a zero here becomes an exit code. + """ + try: + replayed = worker.replay(env_id, envelope) + except TeraError: + return False + return ( + replayed["finalStateChecksum"] == envelope["finalStateChecksum"] + and replayed["steps"] == flight.steps + and abs(replayed["cumulativeReward"] - flight.ts_return) < 1e-12 + ) + + +__all__ = [ + "BAND", + "Command", + "Flight", + "SpatialConfig", + "SpatialData", + "SpatialTask", + "TeraSpatialEnv", + "TeraSpatialEnvConfig", + "TeraSpatialTaskset", + "TurnView", + "parse_command", + "score", + "shared_worker", +] diff --git a/environments/tera_spatial/tera_spatial/worker.mjs b/environments/tera_spatial/tera_spatial/worker.mjs index b8a4ca1..fea929a 100644 --- a/environments/tera_spatial/tera_spatial/worker.mjs +++ b/environments/tera_spatial/tera_spatial/worker.mjs @@ -71,21 +71,32 @@ function held(handle) { * This is the reward denominator. Reimplementing these four policies in Python * would put the number every environment is normalised against on the far side * of a hand translation — so the loop lives here and Python receives a result. + * + * `hold` is how many simulation steps one decision is held for. At 1 the policy + * sees every frame, which is the denominator. Above 1 it is the same policy + * flying with a language model's refresh rate, and the sweep of returns over + * `hold` is the evidence for how large a turn budget the taskset has to grant. */ -function rollBaseline(envId, seed, scenario, policy, maxSteps) { +function rollBaseline(envId, seed, scenario, policy, maxSteps, hold) { const { Environment, scripted, inaction } = registration(envId); const env = new Environment(); const start = env.reset(seed, scenario); const cap = maxSteps ?? env.manifest.maxSteps; + const every = Math.max(1, Math.trunc(hold ?? 1)); let observation = start.observation; let cumulativeReward = 0; let steps = 0; + let decisions = 0; let terminated = false; let truncated = false; let terminalReason = null; + let action = null; const components = {}; while (steps < cap) { - const action = policy === "inaction" ? inaction : scripted(observation); + if (steps % every === 0) { + action = policy === "inaction" ? inaction : scripted(observation); + decisions += 1; + } const result = env.step(action); observation = result.observation; cumulativeReward += result.reward; @@ -103,6 +114,8 @@ function rollBaseline(envId, seed, scenario, policy, maxSteps) { policy, seed, scenario, + hold: every, + decisions, steps, cumulativeReward, rewardComponents: components, @@ -149,8 +162,8 @@ const OPS = { return new Environment().replay(trace); }, - oracle: ({ env: envId, seed, scenario, policy = "scripted", maxSteps = null }) => - rollBaseline(envId, seed, scenario, policy, maxSteps), + oracle: ({ env: envId, seed, scenario, policy = "scripted", maxSteps = null, hold = 1 }) => + rollBaseline(envId, seed, scenario, policy, maxSteps, hold), close: ({ handle }) => ({ closed: handles.delete(handle) }), diff --git a/environments/tera_spatial/tests/test_crow_nav.py b/environments/tera_spatial/tests/test_crow_nav.py new file mode 100644 index 0000000..f66b020 --- /dev/null +++ b/environments/tera_spatial/tests/test_crow_nav.py @@ -0,0 +1,305 @@ +"""The taskset's own gate: the grammar, the arithmetic, and one flight flown for real. + +`test_replay.py` proves the bridge carries a simulation faithfully. This file +proves the three things built on top of it are right when no model is watching: + +1. **The reply grammar.** A reply is the only thing a model controls, so every way + of getting it wrong has to land somewhere sensible and none of them may raise — + a formatting slip that crashes a rollout scores nothing at all rather than + scoring badly. +2. **The reward.** Floor, ceiling and the two products, computed against episodes + this file actually flies rather than against a fixture. +3. **The rendering.** The strongest claim a spatial observation can make is that + it is enough to fly on, so the test flies on it: a policy that reads nothing + but the rendered panel — the same characters the model is shown, parsed back + out of them — reaches the waypoint inside the shipped budget. + + uv run --project environments/tera_spatial python -m unittest discover \\ + -s environments/tera_spatial/tests -v +""" + +from __future__ import annotations + +import json +import math +import re +import unittest + +from tera_crow_nav.taskset import CrowNavConfig, CrowNavTaskset +from tera_spatial import TeraWorker +from tera_spatial.spatial import ( + Command, + Flight, + SpatialData, + TurnView, + parse_command, + score, +) + +FIELDS = ("forward", "turn", "pitch", "climb", "glide") +INACTION = dict.fromkeys(FIELDS, 0.0) +MINIMUM_POWER = {"forward": -1.0, "turn": 0.0, "pitch": 0.0, "climb": 0.03, "glide": False} + +_DISTANCE = re.compile(r"^\s+([-\d.]+) m away .* bearing ([+-]\d+)", re.MULTILINE) +_HEIGHT = re.compile(r"([+-][\d.]+) m of height to make up") + + +def read_the_board(board: str) -> tuple[float, float, float]: + """Distance, bearing in degrees and height to make up — from the panel text only. + + Deliberately parsed out of the rendered characters rather than taken from the + observation dict. If this function can fly the crow, the rendering carries + enough; if the rendering ever stops carrying enough, this stops passing. + """ + away, bearing = _DISTANCE.search(board).groups() + (height,) = _HEIGHT.search(board).groups() + return float(away), float(bearing), float(height) + + +def reader_policy(board: str) -> str: + """A reply, in the grammar, from a reader who has seen only the panel. + + Proportional guidance and nothing clever: point at the waypoint, trim the + height, and shorten the hold as it closes, because a crow at 12 m/s covers + 12 m in one held second and the waypoint is 2.5 m wide. + """ + away, bearing, height = read_the_board(board) + hold = 8 if away > 40 else 4 if away > 12 else 2 + return json.dumps( + { + "forward": 0.45, + "turn": max(-1.0, min(1.0, math.radians(bearing) / 0.55)), + "pitch": 0.0, + "climb": max(-0.8, min(0.8, height / 10 - 0.2)), + "glide": False, + "hold": hold, + } + ) + + +class GrammarTests(unittest.TestCase): + """Every reply a model can emit, and where it lands.""" + + def parse(self, reply: str, previous: Command | None = None) -> Command: + return parse_command( + reply, + fields=FIELDS, + max_hold=12, + default_hold=6, + previous=previous, + inaction=INACTION, + ) + + def test_a_fenced_block_is_read(self) -> None: + command = self.parse('```json\n{"forward": 0.5, "turn": -1, "glide": true, "hold": 4}\n```') + self.assertIsNone(command.problem) + self.assertEqual(command.hold, 4) + self.assertEqual(command.action["forward"], 0.5) + self.assertEqual(command.action["turn"], -1.0) + self.assertIs(command.action["glide"], True) + + def test_the_last_block_wins(self) -> None: + """A model that thinks out loud in JSON is answering with its last one.""" + command = self.parse( + '```json\n{"forward": 0.1}\n```\nno, better:\n```json\n{"forward": 0.9}\n```' + ) + self.assertEqual(command.action["forward"], 0.9) + + def test_a_bare_object_is_read(self) -> None: + self.assertEqual(self.parse('{"climb": 0.3}').action["climb"], 0.3) + + def test_omitted_controls_keep_flying(self) -> None: + previous = Command({"forward": 0.8, "turn": 0.2, "pitch": 0.0, "climb": 0.1, "glide": False}, 6) + command = self.parse('{"turn": -0.5}', previous) + self.assertEqual(command.action["forward"], 0.8) + self.assertEqual(command.action["turn"], -0.5) + + def test_an_unreadable_reply_costs_the_default_hold(self) -> None: + """Silence is the one thing a reply must not be. A turn that costs no + simulation steps is a free turn, and a free turn is a budget that does not + bind — which is house rule 4 undone by a parser.""" + previous = Command({**INACTION, "forward": 0.7}, 5) + for reply in ("", "I think we should climb.", "```json\n{not json}\n```", "[1, 2, 3]"): + with self.subTest(reply=reply): + command = self.parse(reply, previous) + self.assertIsNotNone(command.problem) + self.assertEqual(command.hold, 6) + self.assertEqual(command.action, previous.action) + + def test_json_that_names_no_control_is_flagged_and_still_flies(self) -> None: + command = self.parse('{"thinking": "left a bit"}') + self.assertEqual(command.problem, "your JSON named none of the controls") + self.assertEqual(command.action, INACTION) + + def test_hold_is_clipped_never_raised(self) -> None: + for raw, expected in ((0, 1), (-9, 1), (999, 12), (3.7, 3)): + with self.subTest(hold=raw): + self.assertEqual(self.parse(json.dumps({"forward": 0, "hold": raw})).hold, expected) + + def test_non_finite_numbers_are_dropped_not_carried(self) -> None: + """`json.loads` accepts bare Infinity and NaN and overflows 1e309 to inf. + Either one reaching a reward is a corrupted training signal rather than a + bad answer, and both are reachable from a reply a model can emit.""" + previous = Command({**INACTION, "forward": 0.4}, 6) + command = self.parse('{"forward": NaN, "turn": Infinity, "climb": 1e309, "hold": NaN}', previous) + self.assertEqual(command.action["forward"], 0.4) + self.assertEqual(command.action["turn"], 0.0) + self.assertEqual(command.hold, 6) + for value in command.action.values(): + if isinstance(value, float): + self.assertTrue(math.isfinite(value)) + + def test_twenty_thousand_nested_arrays_do_not_crash_the_decoder(self) -> None: + self.assertIsNotNone(self.parse("[" * 20_000).problem) + + +class FlightTests(unittest.TestCase): + """One scenario, flown three ways, graded on what the simulator recorded.""" + + worker: TeraWorker + + @classmethod + def setUpClass(cls) -> None: + cls.worker = TeraWorker() + cls.taskset = CrowNavTaskset(CrowNavConfig(id="tera-crow-nav", num_tasks=4)) + cls.tasks = list(cls.taskset) + + @classmethod + def tearDownClass(cls) -> None: + cls.worker.close() + + def fly(self, data: SpatialData, replies) -> tuple[Flight, dict]: + """Drive a whole episode from Python, exactly as the env does.""" + episode = self.worker.open( + data.env_id, data.seed, {"split": data.split, "id": data.scenario_id} + ) + flight = Flight( + episode=episode, + max_turns=data.max_turns, + budget=data.max_turns * data.max_hold, + observation=episode.observation, + ) + view = TurnView( + observation=flight.observation, + turn=1, + turns_left=data.max_turns, + steps=0, + budget=flight.budget, + max_hold=data.max_hold, + last=None, + ) + while not flight.done: + command = parse_command( + replies(self.taskset.board(view)), + fields=FIELDS, + max_hold=data.max_hold, + default_hold=data.default_hold, + previous=flight.last, + inaction=INACTION, + ) + flight.fly(command) + view = TurnView( + observation=flight.observation, + turn=flight.turns + 1, + turns_left=data.max_turns - flight.turns, + steps=flight.steps, + budget=flight.budget, + max_hold=data.max_hold, + last=command, + ) + envelope = episode.trace() + episode.close() + return flight, envelope + + def test_a_reader_of_the_panel_reaches_every_waypoint(self) -> None: + """The rendering's real claim: it is enough to fly on. + + Nothing in `reader_policy` sees the observation dict — it sees the same + characters the model is shown and parses them back out. Four scenarios, + four waypoints, inside the shipped budget. + """ + for task in self.tasks: + with self.subTest(task=task.data.name): + flight, envelope = self.fly(task.data, reader_policy) + self.assertEqual(flight.terminal_reason, "goal") + self.assertLessEqual(flight.turns, task.data.max_turns) + rewards, metrics, info = score(task.data, flight, envelope, self.worker) + self.assertEqual(metrics["gate"], 1.0) + self.assertEqual(metrics["replay_ok"], 1.0) + self.assertGreater(rewards["flight"][0], 0.0) + self.assertEqual(info["terminal_reason"], "goal") + + def test_minimum_power_scores_exactly_zero(self) -> None: + """House rule 3's floor. A crow that will not fly does not fly, and the + reward for not flying is 0.000 and not 0.001 — the clip is against a + negative return, so there is nothing to round.""" + data = self.tasks[0].data + flight, envelope = self.fly(data, lambda _board: json.dumps({**MINIMUM_POWER, "hold": 12})) + rewards, metrics, _ = score(data, flight, envelope, self.worker) + self.assertNotEqual(flight.terminal_reason, "goal") + self.assertLess(metrics["ts_return"], 0.0) + self.assertEqual(rewards["flight"][0], 0.0) + self.assertEqual(rewards["economy"][0], 0.0) + + def test_a_silent_model_still_spends_its_budget(self) -> None: + """Sixteen unreadable replies cost sixteen turns and ninety-six steps.""" + data = self.tasks[0].data + flight, _ = self.fly(data, lambda _board: "I would rather not.") + self.assertEqual(flight.turns, data.max_turns) + self.assertEqual(flight.malformed, data.max_turns) + self.assertEqual(flight.steps, data.max_turns * data.default_hold) + + def test_the_ceiling_is_reachable_inside_the_budget(self) -> None: + """House rule 3's ceiling, taken from Tera and not from Python. + + The scripted controller — the same function the denominator is measured + with — flown at a hold this taskset permits, cut off at the turns it + grants, must still clear the band. If it cannot, the budget is not tight, + it is impossible. + """ + data = self.tasks[0].data + best = self.worker.oracle( + data.env_id, + data.seed, + {"split": data.split, "id": data.scenario_id}, + max_steps=data.max_turns * 5, + hold=5, + ) + self.assertEqual(best["terminalReason"], "goal") + self.assertGreaterEqual( + best["cumulativeReward"] / (data.band * data.oracle_return), 1.0 + ) + + def test_the_turn_budget_binds(self) -> None: + """House rule 4, in the shipped engine rather than in a spreadsheet. + + The same policy at the same refresh rate, allowed one turn and allowed + sixteen. If the two agree, the budget buys nothing and crow-nav has no + rule-4 claim to make. + """ + data = self.tasks[0].data + request = {"split": data.split, "id": data.scenario_id} + one = self.worker.oracle(data.env_id, data.seed, request, max_steps=12, hold=12) + many = self.worker.oracle(data.env_id, data.seed, request, max_steps=16 * 5, hold=5) + self.assertNotEqual(one["terminalReason"], "goal") + self.assertEqual(many["terminalReason"], "goal") + self.assertGreater(many["cumulativeReward"] - one["cumulativeReward"], 4.0) + + def test_a_tampered_trace_fails_the_episode_replay(self) -> None: + """`replay_ok` is not decoration: it is the bridge's claim, asserted once + per episode. An envelope whose reward was edited and re-sealed with a + checksum Tera itself computed still has to fail.""" + data = self.tasks[0].data + flight, envelope = self.fly(data, reader_policy) + forged = json.loads(json.dumps(envelope)) + forged["steps"][0]["reward"] += 1.0 + forged["cumulativeReward"] += 1.0 + forged["checksum"] = self.worker.checksum( + {k: v for k, v in forged.items() if k != "checksum"} + ) + _, metrics, _ = score(data, flight, forged, self.worker) + self.assertEqual(metrics["replay_ok"], 0.0) + + +if __name__ == "__main__": + unittest.main()