Unify hosted rollout task targeting

hud/eval/runtime/hosted.py

@@ -132,30 +132,49 @@ async def _submit_and_await(
                 **spec,
                 "config": {**spec.get("config", {}), **task.agent_config},
             }
+        config = dict(spec.get("config", {}))
+        model = config.pop("model", None)
+        if not isinstance(model, str) or not model:
+            raise ValueError("hosted agent config requires a model")
+        max_steps = config.pop("max_steps", 100)
+        if not isinstance(max_steps, int) or isinstance(max_steps, bool) or max_steps < 1:
+            raise ValueError("hosted agent config requires max_steps to be a positive integer")
         platform = PlatformClient.from_settings()
         if not platform.api_key:
             raise RuntimeError("HUD-hosted execution requires HUD_API_KEY")
         payload: dict[str, Any] = {
             # The SDK's hex ids travel as canonical UUID strings.
             "trace_id": str(uuid.UUID(trace_id)),
             "job_id": str(uuid.UUID(job_id)),
-            "env": task.env,
-            "task": task.id,
-            "slug": task.slug,
-            "args": task.args,
-            "agent": spec,
+            "target": self._target(task),
+            "agent": {"model": model, "config": config},
+            "max_steps": max_steps,
         }
         if group_id is not None:
             payload["group_id"] = group_id
         if task.runtime_config is not None:
             runtime_config = task.runtime_config.model_dump(mode="json", exclude_unset=True)
             if runtime_config:
                 payload["runtime_config"] = runtime_config
-        if task.verifier is not None:
-            payload["verifier"] = task.verifier.model_dump(mode="json", exclude_none=True)
-        await platform.apost("/rollouts/submit", json=payload)
+        await platform.apost("/rollouts", json=payload)
         return await self._await_terminal(platform, payload["trace_id"])
 
+    @staticmethod
+    def _target(task: Task) -> dict[str, Any]:
+        task_version_id = task._current_platform_version_id()
+        if task_version_id is not None:
+            return {"type": "task_version", "task_version_id": task_version_id}
+        target: dict[str, Any] = {
+            "type": "inline_task",
+            "env": task.env,
+            "task": task.id,
+            "slug": task.slug,
+            "args": task.args,
+        }
+        if task.verifier is not None:
+            target["verifier"] = task.verifier.model_dump(mode="json", exclude_none=True)
+        return target
+
     @staticmethod
     def _fold(state: dict[str, Any], trace_id: str) -> Run:
         """Build the local view of a remotely-executed rollout from its trace state."""

hud/eval/sync.py

@@ -113,7 +113,7 @@ def fetch_taskset_tasks(
 
 def _record_to_task(record: dict[str, Any]) -> Task:
     """Map one platform export record onto the portable row shape."""
-    return Task.model_validate(
+    task = Task.model_validate(
         {
             "env": record.get("env"),
             "id": record.get("scenario") or "",
@@ -126,6 +126,10 @@ def _record_to_task(record: dict[str, Any]) -> Task:
             "verifier": record.get("verifier"),
         }
     )
+    task_version_id = record.get("task_version_id")
+    if isinstance(task_version_id, str):
+        task._bind_platform_version(task_version_id)
+    return task
 
 
 # ─── upload ─────────────────────────────────────────────────────────────

hud/eval/task.py

@@ -38,6 +38,8 @@ class Task(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
 
     _env: Environment | None = PrivateAttr(default=None)
+    _platform_task_version_id: str | None = PrivateAttr(default=None)
+    _platform_signature: str | None = PrivateAttr(default=None)
 
     env: str = Field(min_length=1)
     id: str = Field(min_length=1)
@@ -93,6 +95,23 @@ def _serialize_runtime_config(
             else None
         )
 
+    def _content_signature(self) -> str:
+        return json.dumps(
+            self.model_dump(mode="json", exclude_none=True),
+            sort_keys=True,
+            default=str,
+            separators=(",", ":"),
+        )
+
+    def _bind_platform_version(self, task_version_id: str) -> None:
+        self._platform_task_version_id = task_version_id
+        self._platform_signature = self._content_signature()
+
+    def _current_platform_version_id(self) -> str | None:
+        if self._platform_signature != self._content_signature():
+            return None
+        return self._platform_task_version_id
+
     # ─── execution ────────────────────────────────────────────────────
 
     async def run(

hud/eval/tests/test_hosted.py

@@ -227,30 +227,69 @@ async def test_run_submits_and_polls_to_terminal(monkeypatch: pytest.MonkeyPatch
     assert run.group_id == "g1"
     assert platform.polled == 3
     (path, payload) = platform.posts[0]
-    assert path == "/rollouts/submit"
+    assert path == "/rollouts"
     # Hex ids travel as canonical UUID strings.
     assert payload["trace_id"] == str(uuid.UUID(trace_id))
     assert payload["job_id"] == str(uuid.UUID(job_id))
-    assert payload["env"] == "sums"
-    assert payload["task"] == "add"
-    assert payload["slug"] == "sums-add"
-    assert payload["args"] == {"a": 1, "b": 2}
+    assert payload["target"] == {
+        "type": "inline_task",
+        "env": "sums",
+        "task": "add",
+        "slug": "sums-add",
+        "args": {"a": 1, "b": 2},
+        "verifier": {
+            "env": "judge",
+            "id": "verify",
+            "args": {"expected": 3},
+            "slug": "verify-5579a3e5",
+            "runtime_config": {"resources": {"memory_mb": 4096}},
+        },
+    }
     assert payload["runtime_config"] == {
         "image": "registry.example/sums:latest",
         "resources": {"cpu": 2.0, "gpu": {"type": "L4", "count": 1}},
         "limits": {"startup_timeout_s": 120, "run_timeout_s": 900},
     }
-    assert payload["verifier"] == {
-        "env": "judge",
-        "id": "verify",
-        "args": {"expected": 3},
-        "slug": "verify-5579a3e5",
-        "runtime_config": {"resources": {"memory_mb": 4096}},
-    }
     assert payload["group_id"] == "g1"
-    assert payload["agent"]["type"] == "openai_compatible"
-    assert payload["agent"]["config"]["model"] == "test-model"
+    assert payload["agent"]["model"] == "test-model"
     assert payload["agent"]["config"]["timeout_seconds"] == 45.0
+    assert "max_steps" not in payload["agent"]["config"]
+    assert payload["max_steps"] == 10
+
+
+@pytest.mark.asyncio
+async def test_run_uses_stored_version_only_while_platform_task_is_unchanged(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    platform = _FakePlatform(
+        [
+            {"status": "completed", "reward": 1.0},
+            {"status": "completed", "reward": 1.0},
+        ]
+    )
+    monkeypatch.setattr(
+        "hud.eval.runtime.hosted.PlatformClient.from_settings", classmethod(lambda cls: platform)
+    )
+    version_id = str(uuid.uuid4())
+    task = Task(env="sums", id="add", slug="one", args={"a": 1})
+    task._bind_platform_version(version_id)
+    hosted = HostedRuntime(poll_interval=0.0)
+
+    await hosted.run(task, _agent(), job_id=uuid.uuid4().hex, trace_id=uuid.uuid4().hex)
+    task.args["a"] = 2
+    await hosted.run(task, _agent(), job_id=uuid.uuid4().hex, trace_id=uuid.uuid4().hex)
+
+    assert platform.posts[0][1]["target"] == {
+        "type": "task_version",
+        "task_version_id": version_id,
+    }
+    assert platform.posts[1][1]["target"] == {
+        "type": "inline_task",
+        "env": "sums",
+        "task": "add",
+        "slug": "one",
+        "args": {"a": 2},
+    }
 
 
 @pytest.mark.asyncio
@@ -358,7 +397,7 @@ async def test_omitted_rollout_timeout_allows_long_environment_timeout(
     )
 
     assert run.trace.status == "completed"
-    assert platform.posts[0][0] == "/rollouts/submit"
+    assert platform.posts[0][0] == "/rollouts"
 
 
 @pytest.mark.asyncio
@@ -409,7 +448,7 @@ async def test_taskset_rollout_timeout_reaches_hosted_runtime(
     )
 
     assert job.runs[0].trace.status == "completed"
-    assert platform.posts[0][0] == "/rollouts/submit"
+    assert platform.posts[0][0] == "/rollouts"
 
 
 @pytest.mark.asyncio
@@ -541,7 +580,7 @@ async def test_submit_timeout_requests_platform_cancel(monkeypatch: pytest.Monke
     class _StuckSubmitPlatform(_FakePlatform):
         async def apost(self, path: str, *, json: Any | None = None) -> Any:
             self.posts.append((path, json or {}))
-            if path == "/rollouts/submit":
+            if path == "/rollouts":
                 await never.wait()
             return {"status": "queued"}
 

hud/eval/tests/test_sync.py

@@ -58,8 +58,21 @@ def test_fetched_tasks_map_canonical_export_fields(
         "taskset_id": "ts-id",
         "name": "demo",
         "tasks": [
-            {"scenario": "solve", "env": "myenv", "name": "a", "args": {"n": 1}},
-            {"scenario": "fix_bug", "env": "other", "name": "b"},
+            {
+                "task_id": "task-a",
+                "task_version_id": "version-a",
+                "scenario": "solve",
+                "env": "myenv",
+                "name": "a",
+                "args": {"n": 1},
+            },
+            {
+                "task_id": "task-b",
+                "task_version_id": "version-b",
+                "scenario": "fix_bug",
+                "env": "other",
+                "name": "b",
+            },
         ],
     }
 
@@ -77,6 +90,10 @@ def fake_request(method: str, url: str, **kwargs: object) -> dict[str, Any]:
         ("myenv", "solve", "a"),
         ("other", "fix_bug", "b"),
     ]
+    assert [task._current_platform_version_id() for task in tasks] == [
+        "version-a",
+        "version-b",
+    ]
 
 
 def test_resolve_taskset_id_looks_up_by_name(monkeypatch: pytest.MonkeyPatch) -> None: