From d105e0d9983b134482764820b66bd4f8a11e8e47 Mon Sep 17 00:00:00 2001 From: luyd Date: Wed, 28 Oct 2026 10:57:10 +0800 Subject: [PATCH 1/6] Fix rollout lifecycle and backend configuration Forward colocate placement to Slime, release sealed trace payloads, and block retries and restarts while cleanup is unresolved. Reuse the embedded rollout loop, drain pending tasks on shutdown, and export bootstrap environment variables. Add focused regressions and lifecycle documentation. Validation: 293 tests passed with one environment-dependent skip; 11 Runner-environment tests passed. Real GPU acceptance completed three rollout rounds each for split and colocate, plus Docker cleanup and restart checks. --- .gitignore | 1 + README.md | 9 +- README_zh.md | 8 +- docs/rollout-runtime-lifecycle.md | 78 +++++++ harborrl/backends/slime_v032/rollout.py | 14 +- harborrl/gateway/trace.py | 20 +- harborrl/platform/slime_train.sh | 3 + harborrl/rollout/harbor_job/coordinator.py | 20 +- harborrl/rollout/harbor_job/sandbox_config.py | 10 +- harborrl/rollout/runtime.py | 181 ++++++++++++--- scripts/bootstrap_backends.sh | 11 +- tests/harborrl/test_backend_bootstrap.py | 51 +++++ tests/harborrl/test_native_gateway.py | 73 +++++- tests/harborrl/test_native_lifecycle.py | 32 +++ tests/harborrl/test_native_terminal.py | 6 +- tests/harborrl/test_native_training.py | 14 +- .../test_rollout_runtime_lifecycle.py | 213 ++++++++++++++++++ tests/harborrl/test_sandbox_runtime.py | 75 +++++- tests/harborrl/test_slime_v032_exporter.py | 24 +- 19 files changed, 767 insertions(+), 76 deletions(-) create mode 100644 docs/rollout-runtime-lifecycle.md create mode 100644 tests/harborrl/test_backend_bootstrap.py create mode 100644 tests/harborrl/test_rollout_runtime_lifecycle.py diff --git a/.gitignore b/.gitignore index 7755b672b..ec61a7863 100644 --- a/.gitignore +++ b/.gitignore @@ -39,6 +39,7 @@ docs/agents/* !docs/configuration-wizard.md !docs/deployment.md !docs/sandbox-rollout.md +!docs/rollout-runtime-lifecycle.md !docs/train-inference-consistency.md !docs/web-console.md practices/**/VALIDATION.md diff --git a/README.md b/README.md index 1b1b8ac7d..bdb9f0e3c 100644 --- a/README.md +++ b/README.md @@ -118,10 +118,15 @@ Expected result: uv executes the installed entry point and tests without activat For native training, install the pinned backend stack in a writable location: ```bash -bash scripts/bootstrap_backends.sh -. /harborrl-backend.env +bash scripts/bootstrap_backends.sh "" +. "/harborrl-backend.env" ``` +The generated file exports `SLIME_DIR`, `MEGATRON_DIR`, and `SGLANG_IMAGE` to child +processes. For files generated by older bootstrap scripts, run `set -a` before +sourcing and `set +a` afterward. See [rollout lifecycle and recovery](docs/rollout-runtime-lifecycle.md) +for cleanup markers and CPU-only regression checks. + ### Train After installation, choose one of the two configuration styles: diff --git a/README_zh.md b/README_zh.md index f75e744a9..22cfcedd9 100644 --- a/README_zh.md +++ b/README_zh.md @@ -108,10 +108,14 @@ uv run pytest 如需原生训练，请把固定版本的 native 训练栈安装到可写位置： ```bash -bash scripts/bootstrap_backends.sh -. /harborrl-backend.env +bash scripts/bootstrap_backends.sh "" +. "/harborrl-backend.env" ``` +生成的文件会将 `SLIME_DIR`、`MEGATRON_DIR` 和 `SGLANG_IMAGE` 导出给子进程。 +对于旧版 bootstrap 脚本生成的文件，请在 source 前执行 `set -a`，之后执行 `set +a`。 +清理标记的恢复步骤和无 GPU 回归命令见[rollout 生命周期与恢复](docs/rollout-runtime-lifecycle.md)。 + ### 训练 完成安装后，从两种配置方式中选择一种： diff --git a/docs/rollout-runtime-lifecycle.md b/docs/rollout-runtime-lifecycle.md new file mode 100644 index 000000000..84a870661 --- /dev/null +++ b/docs/rollout-runtime-lifecycle.md @@ -0,0 +1,78 @@ +# Rollout Lifecycle and Recovery + +## GPU Layout + +`deployment.layout: colocate` requires equal actor, rollout, and total GPU counts. +The Slime launcher passes `--colocate` only for that layout. Split deployments +omit it. The final argv is recorded in the run's command evidence, including in +dry-run mode. + +## Trace Memory + +Closing an attempt revokes its credential. Sealing still requires every pending +request to drain. After `trace.json` is successfully persisted, the registry +releases the full turns and errors. Repeated sealing reads the immutable record +from disk. Failed writes retain the payload so sealing can be retried. + +The registry retains small attempt records to enforce attempt and credential +uniqueness and policy barriers. Training samples retain their current trajectory +evidence for as long as the training backend needs those samples. + +## Cleanup Recovery + +Docker and Sandbox both stop automatic retries when an attempt produces +`cleanup-required.json`. Queued attempts are checked before starting a Runner; +already running attempts finish or are cancelled with cleanup awaited. A runtime +also scans its `native-rollouts` directory before opening the gateway, so restarting +the same run cannot bypass unresolved cleanup. + +1. Read each marker reported by the error, including any nested + `runner/cleanup-required.json`. Coordinator markers include the attempt identity + and Runner host (`null` means local execution). +2. Inspect the matching Harbor Docker or Sandbox resources on that host and + confirm that deletion has completed. Retain the attempt's other evidence. +3. Remove only the markers whose resources have been confirmed clean. Both the + attempt-level and nested Runner markers must be resolved. +4. Restart or resume the run. Admission reopens when no unresolved marker remains. + +There is no automatic deletion of potentially live resources during recovery. + +## Embedded Event Loop + +Successive synchronous Slime rollout calls reuse one event loop and one cached +runtime. The runtime's semaphore and Sandbox admission semaphores belong to that +loop. Changing the runtime configuration requires closing the previous runtime. +Using a cached runtime from another loop raises `RuntimeLifecycleError` immediately. + +`harborrl.rollout.runtime.close_runtime()` cancels and awaits outstanding embedded +tasks, closes the gateway, then closes async generators, the executor, and the +loop. Call it between rollout calls, outside an active event loop; process exit also +registers cleanup. The implementation supports Python 3.10. The manager-side +RolloutService continues to use its own persistent loop. + +## CPU Regression Checks + +These checks use synthetic serving, mock resource allocation, real local Runner +subprocesses, shell dry-runs, and stub bootstrap commands. They do not need CUDA, +model weights, Docker, Harbor, Slime, or the Sandbox SDK. + +```bash +python3 -m pytest -q \ + tests/harborrl/test_backend_bootstrap.py \ + tests/harborrl/test_native_training.py \ + tests/harborrl/test_native_gateway.py \ + tests/harborrl/test_native_lifecycle.py \ + tests/harborrl/test_native_terminal.py \ + tests/harborrl/test_rollout_runtime.py \ + tests/harborrl/test_rollout_runtime_lifecycle.py \ + tests/harborrl/test_sandbox.py \ + tests/harborrl/test_sandbox_runtime.py \ + tests/harborrl/test_slime_v032_exporter.py \ + tests/harborrl/test_rollout_service.py \ + tests/harborrl/test_delivery_barrier.py +``` + +GPU acceptance remains separate: run multi-round split and colocate training, +check serving/training memory behavior, verify policy version transitions, and +exercise cleanup against actual Docker/Sandbox resources. CPU tests establish +the scheduling and lifecycle contracts, not live model or GPU correctness. diff --git a/harborrl/backends/slime_v032/rollout.py b/harborrl/backends/slime_v032/rollout.py index 44c36e2c8..450beb6d5 100644 --- a/harborrl/backends/slime_v032/rollout.py +++ b/harborrl/backends/slime_v032/rollout.py @@ -16,10 +16,10 @@ from harborrl.export.native import export_training_batch from harborrl.rollout.runtime import ( RolloutRuntimeConfig, - close_runtime, current_policy_version, digest, run_group, + run_embedded, runtime, task_entry, ) @@ -195,12 +195,14 @@ async def run_native_groups(args, groups, policy_version, rollout_id): ir_groups = await run_service_groups(args, groups, rollout_id) assert_locked_policy_versions(ir_groups, policy_version) return ir_groups + pending = [asyncio.create_task( + generate_group(args, group, params, rollout_id=rollout_id)) for group in groups] try: - results = await asyncio.gather(*( - generate_group(args, group, params, rollout_id=rollout_id) for group in groups - )) + results = await asyncio.gather(*pending) except BaseException: - close_runtime() + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) raise ir_groups = [] for group in results: @@ -309,7 +311,7 @@ def generate_rollout( if runner is None: async def runner(args_, groups_, policy_version): return await run_native_groups(args_, groups_, policy_version, rollout_id) - ir_groups = asyncio.run(runner(args, groups, None)) + ir_groups = run_embedded(runner(args, groups, None)) policy_version = lock_policy_version(args) assert_locked_policy_versions(ir_groups, policy_version) batch = export_training_batch( diff --git a/harborrl/gateway/trace.py b/harborrl/gateway/trace.py index 5f5ca7c6a..5f3ba0b12 100644 --- a/harborrl/gateway/trace.py +++ b/harborrl/gateway/trace.py @@ -6,7 +6,7 @@ from copy import deepcopy from pathlib import Path -from harborrl.trajectories.native import digest, publish +from harborrl.trajectories.native import digest, publish, read_json class TraceRegistry: @@ -19,6 +19,11 @@ def __init__(self, root): self.accepting = True self.version = None + def _turns(self, state): + if state.get("sealed"): + return read_json(state["root"] / "trace.json")["turns"] + return state["turns"] + def register(self, identity, trial_id, *, credential=None): with self.lock: if not self.accepting or (self.version is not None and self.version != identity.policy_version): @@ -104,7 +109,7 @@ def wait_delivered(self, aid, timeout): raise TimeoutError("policy delivery acknowledgement still pending") self.changed.wait(remaining) if any(turn.get("delivery") not in ("sent", "consumed_confirmed") - for turn in state["turns"]): + for turn in self._turns(state)): raise ValueError("policy response delivery was not confirmed") def consume(self, aid, response_id, content, evidence_ref): @@ -125,9 +130,10 @@ def consume(self, aid, response_id, content, evidence_ref): def unconsumed_responses(self, aid): with self.lock: state = self.attempts[aid] + turns = self._turns(state) return [{"response_id": turn["response_id"], "content": deepcopy(turn["response"]["content"]), - "finish_reason": turn.get("finish_reason"), "is_final": index == len(state["turns"]) - 1} - for index, turn in enumerate(state["turns"]) if turn.get("delivery") != "consumed_confirmed"] + "finish_reason": turn.get("finish_reason"), "is_final": index == len(turns) - 1} + for index, turn in enumerate(turns) if turn.get("delivery") != "consumed_confirmed"] def close(self, aid): with self.lock: @@ -138,6 +144,9 @@ def close(self, aid): def seal(self, aid): with self.lock: state = self.attempts[aid] + if state["sealed"]: + record = read_json(state["root"] / "trace.json") + return record["turns"], record["seal"] if not state["closed"] or state["pending"]: raise ValueError("close admission and drain requests before sealing") turns = deepcopy(state["turns"]) @@ -149,6 +158,9 @@ def seal(self, aid): "turns_digest": digest(turns), "errors": sorted(set(errors))} publish(state["root"] / "trace.json", {"turns": turns, "seal": seal}) state["sealed"] = True + # Keep only admission/identity bookkeeping after durable publication. + state["turns"].clear() + state["errors"].clear() return turns, seal def drain(self, timeout): diff --git a/harborrl/platform/slime_train.sh b/harborrl/platform/slime_train.sh index 8d7972874..db00a0948 100644 --- a/harborrl/platform/slime_train.sh +++ b/harborrl/platform/slime_train.sh @@ -146,6 +146,9 @@ PARALLEL_ARGS=( --attention-backend flash --no-gradient-accumulation-fusion ) +if [[ "${HARBORRL_GPU_LAYOUT:-split}" == "colocate" ]]; then + PARALLEL_ARGS+=(--colocate) +fi CUSTOM_ARGS=( --rollout-function-path harborrl.backends.slime_v032.rollout.generate_rollout diff --git a/harborrl/rollout/harbor_job/coordinator.py b/harborrl/rollout/harbor_job/coordinator.py index 17047cd40..49ea55aae 100644 --- a/harborrl/rollout/harbor_job/coordinator.py +++ b/harborrl/rollout/harbor_job/coordinator.py @@ -92,7 +92,7 @@ async def run_attempt(identity, *, task_path, profile, gateway_url, runner_pytho stderr.close() publish(root / "launch-failed.json", {"identity": identity.to_dict(), "reason": "Runner process did not start"}) raise - registered, trial_id, lifecycle_timeout = False, None, False + registered, trial_id, lifecycle_timeout, sealed = False, None, False, False messages = [] try: req = {"identity": identity.to_dict(), "task_path": str(Path(task_path).resolve()), "profile": profile, @@ -142,7 +142,8 @@ async def lifecycle(): lifecycle_path = root / 'runner' / 'sandbox-lifecycle.json' lifecycle = read_json(lifecycle_path) if lifecycle_path.is_file() else {} if lifecycle.get('state') != 'deleted' or lifecycle.get('runtime_digest') != req['runtime_digest']: - publish(root / 'cleanup-required.json', {'identity': identity.to_dict(), 'reason': 'Sandbox deletion unconfirmed'}) + publish(root / 'cleanup-required.json', {'identity': identity.to_dict(), 'runner_host': runner_host, + 'reason': 'Sandbox deletion unconfirmed'}) raise RuntimeError('sandbox deletion unconfirmed') trial_dir = Path(terminal["trial_dir"]).resolve() result_path = Path(terminal["result_path"]).resolve() @@ -157,6 +158,7 @@ async def lifecycle(): CLAUDE_CODE_HARNESS.adapter.audit_empty_final( trial_dir / "agent" / "claude-code.txt", registry, identity.attempt_id, audit) turns, seal = registry.seal(identity.attempt_id) + sealed = True publish(root / "evaluation.json", receipt) ir = assemble(identity, trial_id, turns, seal, receipt) publish(root / "trajectory.v2.json", ir) @@ -166,8 +168,9 @@ async def lifecycle(): publish(root / "failure.json", {"identity": identity.to_dict(), "error_type": type(exc).__name__}) raise finally: - if sandbox is not None and (root / 'runner' / 'cleanup-required.json').exists(): - publish(root / 'cleanup-required.json', {'identity': identity.to_dict(), 'reason': 'Inspect runner/cleanup-required.json'}) + if (root / 'runner' / 'cleanup-required.json').exists() and not (root / 'cleanup-required.json').exists(): + publish(root / 'cleanup-required.json', {'identity': identity.to_dict(), 'runner_host': runner_host, + 'reason': 'Inspect runner/cleanup-required.json'}) if registered: registry.close(identity.attempt_id) if process.returncode is None: @@ -181,12 +184,15 @@ async def lifecycle(): except ProcessLookupError: pass await process.wait() - publish(root / "cleanup-required.json", {"identity": identity.to_dict(), "trial_id": trial_id, - "reason": "Runner cleanup deadline exceeded; inspect Harbor resources before retry"}) + if not (root / "cleanup-required.json").exists(): + publish(root / "cleanup-required.json", {"identity": identity.to_dict(), "trial_id": trial_id, + "runner_host": runner_host, + "reason": "Runner cleanup deadline exceeded; inspect Harbor resources before retry"}) if (process.returncode != 0 or lifecycle_timeout) and not (root / "cleanup-required.json").exists(): publish(root / "cleanup-required.json", {"identity": identity.to_dict(), "trial_id": trial_id, + "runner_host": runner_host, "reason": "Runner exited abnormally; inspect Harbor resources before retry"}) - if registered: + if registered and not sealed: try: registry.seal(identity.attempt_id) except ValueError: diff --git a/harborrl/rollout/harbor_job/sandbox_config.py b/harborrl/rollout/harbor_job/sandbox_config.py index bb008e184..063cce41e 100644 --- a/harborrl/rollout/harbor_job/sandbox_config.py +++ b/harborrl/rollout/harbor_job/sandbox_config.py @@ -120,7 +120,12 @@ class SandboxAdmission: def __init__(self, cfg): self.total = asyncio.Semaphore(cfg['max_live_instances']) self.bindings = {b['image']: asyncio.Semaphore(b['max_live_instances']) for b in cfg['bindings']} - self.blocked = False + self.cleanup_markers = set() + + @property + def blocked(self): + self.cleanup_markers = {path for path in self.cleanup_markers if path.exists()} + return bool(self.cleanup_markers) @asynccontextmanager async def lease(self, binding, root): @@ -131,5 +136,4 @@ async def lease(self, binding, root): try: yield finally: - if (Path(root) / 'cleanup-required.json').exists(): - self.blocked = True + self.cleanup_markers.update(Path(root).rglob('cleanup-required.json')) diff --git a/harborrl/rollout/runtime.py b/harborrl/rollout/runtime.py index 558bab01d..99bffcf87 100644 --- a/harborrl/rollout/runtime.py +++ b/harborrl/rollout/runtime.py @@ -9,7 +9,9 @@ from __future__ import annotations import asyncio +import atexit import json +import logging import secrets import threading from contextlib import nullcontext @@ -23,6 +25,31 @@ TASK_ENTRY_KEYS = {"id", "revision", "path", "task_digest", "reward_profile"} +class RuntimeLifecycleError(RuntimeError): + """An async resource was used outside its owning runtime lifecycle.""" + + +class CleanupRequiredError(RuntimeError): + """Harbor resources must be inspected before admitting more attempts.""" + + +class CleanupAdmission: + def __init__(self, root): + self.markers = set(Path(root).rglob("cleanup-required.json")) + + def observe(self, root): + self.markers.update(Path(root).rglob("cleanup-required.json")) + + def check(self): + self.markers = {path for path in self.markers if path.exists()} + if self.markers: + paths = ", ".join(str(path) for path in sorted(self.markers)) + raise CleanupRequiredError( + "rollout cleanup unresolved; automatic retry blocked. " + "Inspect Harbor resources, then remove resolved cleanup-required.json " + f"markers before resuming: {paths}") + + @dataclass(frozen=True) class RolloutRuntimeConfig: """Deployment-independent description of one run's rollout machinery.""" @@ -155,6 +182,9 @@ def __init__(self, config, tokenizer, *, backend=None): self.profile = read_json(config.profile_path) self.rollout_root = self.run_root / "native-rollouts" self.rollout_root.mkdir(parents=True, exist_ok=True) + self.cleanup = CleanupAdmission(self.rollout_root) + self.cleanup.check() + self._loop = None self.registry = TraceRegistry(self.run_root / "native-traces") self.semaphore = asyncio.Semaphore(config.concurrency_limit) if config.sandbox: @@ -177,6 +207,13 @@ def __init__(self, config, tokenizer, *, backend=None): port=config.gateway_port) self._start_gateway() + def check_event_loop(self): + loop = asyncio.get_running_loop() + if self._loop is None: + self._loop = loop + elif self._loop is not loop: + raise RuntimeLifecycleError("rollout runtime belongs to a different event loop") + def _start_gateway(self): self.thread = threading.Thread( target=self.server.serve_forever, name="harborrl-native-gateway", daemon=True @@ -200,21 +237,53 @@ def manifest_path(self, identity): NativeRolloutRuntime = RolloutRuntime _RUNTIME = None -_RUNTIME_LOCK = asyncio.Lock() +_RUNTIME_LOCK = threading.RLock() +_EMBEDDED_LOCK = threading.RLock() +_EMBEDDED_LOOP = None +_EMBEDDED_THREAD = None async def runtime(config, *, tokenizer=None, tokenizer_loader=None): """Process-wide runtime cache for the embedded GPU-side rollout path.""" global _RUNTIME - if _RUNTIME is None: - async with _RUNTIME_LOCK: - if _RUNTIME is None: - if tokenizer is None: - if tokenizer_loader is None: - raise ValueError("runtime requires a tokenizer or a loader") - tokenizer = tokenizer_loader() - _RUNTIME = RolloutRuntime(config, tokenizer) - return _RUNTIME + with _RUNTIME_LOCK: + if _RUNTIME is None: + if tokenizer is None: + if tokenizer_loader is None: + raise ValueError("runtime requires a tokenizer or a loader") + tokenizer = tokenizer_loader() + _RUNTIME = RolloutRuntime(config, tokenizer) + elif _RUNTIME.config != config: + raise RuntimeLifecycleError("close the cached rollout runtime before changing its configuration") + _RUNTIME.check_event_loop() + return _RUNTIME + + +def run_embedded(coroutine): + """Run successive synchronous rollout calls on one Python 3.10-compatible loop.""" + global _EMBEDDED_LOOP, _EMBEDDED_THREAD + try: + asyncio.get_running_loop() + except RuntimeError: + pass + else: + coroutine.close() + raise RuntimeLifecycleError("synchronous rollout cannot run inside an active event loop") + with _EMBEDDED_LOCK: + if _EMBEDDED_LOOP is None: + _EMBEDDED_LOOP = asyncio.new_event_loop() + _EMBEDDED_THREAD = threading.get_ident() + elif _EMBEDDED_THREAD != threading.get_ident(): + coroutine.close() + raise RuntimeLifecycleError("embedded rollout must stay on its owning thread") + try: + return _EMBEDDED_LOOP.run_until_complete(coroutine) + except BaseException: + try: + close_runtime() + except Exception: + logging.getLogger(__name__).exception("embedded rollout shutdown failed") + raise def publish_native_version(runtime_, states, expected): @@ -234,10 +303,46 @@ def current_policy_version(runtime_): def close_runtime(): - global _RUNTIME - if _RUNTIME is not None: - _RUNTIME.close() - _RUNTIME = None + """Drain embedded tasks, close the gateway, then release the owning loop.""" + global _RUNTIME, _EMBEDDED_LOOP, _EMBEDDED_THREAD + with _EMBEDDED_LOCK: + loop = _EMBEDDED_LOOP + if loop is not None and loop.is_running(): + raise RuntimeLifecycleError("close embedded rollout between calls") + if loop is not None: + try: + asyncio.get_running_loop() + except RuntimeError: + pass + else: + raise RuntimeLifecycleError("close embedded rollout outside an active event loop") + + async def cancel_pending(): + pending = asyncio.all_tasks() - {asyncio.current_task()} + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + + try: + if loop is not None: + loop.run_until_complete(cancel_pending()) + with _RUNTIME_LOCK: + cached, _RUNTIME = _RUNTIME, None + try: + if cached is not None: + cached.close() + finally: + if loop is not None: + loop.run_until_complete(loop.shutdown_asyncgens()) + loop.run_until_complete(loop.shutdown_default_executor()) + finally: + if loop is not None: + loop.close() + _EMBEDDED_LOOP = None + _EMBEDDED_THREAD = None + + +atexit.register(close_runtime) def task_entry(sample): @@ -291,6 +396,14 @@ async def _run_slot(config, runtime_, slots, slot, entry, reward_profile): """Execute one slot with the group's shared identity and retry bookkeeping.""" workers = config.workers overrides = config.task_worker_overrides + check_loop = getattr(runtime_, "check_event_loop", None) + if check_loop is not None: + check_loop() + if not hasattr(runtime_, "cleanup"): + runtime_.cleanup = CleanupAdmission(runtime_.rollout_root) + cleanup = runtime_.cleanup + cleanup.check() + last_error = None for attempt_index in range(slots.max_attempts): identity = slots.start(str(slot)) runner_host = overrides.get(entry["id"]) @@ -307,27 +420,35 @@ async def _run_slot(config, runtime_, slots, slot, entry, reward_profile): admission = runtime_.sandbox_admission.lease(task_binding(config.sandbox, entry['path']), root) try: async with admission, runtime_.semaphore: - ir = await run_attempt( - identity, - task_path=entry["path"], - profile=runtime_.profile, - gateway_url=config.gateway_url, - runner_python=config.runner_python, - runner_host=runner_host, - root=root, - registry=runtime_.registry, - reward_profile=reward_profile, - max_output_tokens=config.max_output_tokens, - **({'sandbox': config.sandbox} if config.sandbox else {}), - ) + cleanup.check() + try: + ir = await run_attempt( + identity, + task_path=entry["path"], + profile=runtime_.profile, + gateway_url=config.gateway_url, + runner_python=config.runner_python, + runner_host=runner_host, + root=root, + registry=runtime_.registry, + reward_profile=reward_profile, + max_output_tokens=config.max_output_tokens, + **({'sandbox': config.sandbox} if config.sandbox else {}), + ) + finally: + cleanup.observe(root) + cleanup.check() + except (CleanupRequiredError, RuntimeLifecycleError) as exc: + slots.finish(str(slot), error=f"{type(exc).__name__}: {exc}") + raise except (OSError, ValueError, RuntimeError, asyncio.TimeoutError) as exc: slots.finish(str(slot), error=f"{type(exc).__name__}: {exc}") - if config.sandbox and ((root / 'cleanup-required.json').exists() or runtime_.sandbox_admission.blocked): - raise RuntimeError('sandbox cleanup unresolved; automatic retry blocked') from exc + cleanup.check() + last_error = exc continue slots.finish(str(slot), ir) return ir - raise RuntimeError(f"native slot {slot} exhausted its retry budget") + raise RuntimeError(f"native slot {slot} exhausted its retry budget") from last_error async def run_group(config, runtime_, *, entry, group_index, rollout_id, diff --git a/scripts/bootstrap_backends.sh b/scripts/bootstrap_backends.sh index 6e9c18d5b..a3078343d 100755 --- a/scripts/bootstrap_backends.sh +++ b/scripts/bootstrap_backends.sh @@ -28,14 +28,15 @@ clone_pinned() { need git need docker mkdir -p "$PREFIX" +PREFIX="$(cd "$PREFIX" && pwd)" clone_pinned "$SLIME_REPO" "$PREFIX/slime" "$SLIME_COMMIT" clone_pinned "$MEGATRON_REPO" "$PREFIX/Megatron-LM" "$MEGATRON_COMMIT" docker pull "$SGLANG_IMAGE" -cat >"$PREFIX/harborrl-backend.env" <"$PREFIX/harborrl-backend.env" echo "Bootstrap complete. Source $PREFIX/harborrl-backend.env before doctor." diff --git a/tests/harborrl/test_backend_bootstrap.py b/tests/harborrl/test_backend_bootstrap.py new file mode 100644 index 000000000..48607d2bd --- /dev/null +++ b/tests/harborrl/test_backend_bootstrap.py @@ -0,0 +1,51 @@ +"""Exercise bootstrap output without cloning repositories or pulling images.""" +import json +import os +import subprocess +import sys +from pathlib import Path + + +def test_backend_env_exports_quoted_absolute_paths(tmp_path): + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + git = bin_dir / "git" + git.write_text('''#!/usr/bin/env bash +set -eu +if [[ "$1" == clone ]]; then + mkdir -p "$3/.git" +elif [[ "$1" == -C && "$3" == rev-parse ]]; then + case "$2" in + */slime) printf '%s\\n' 3778dbf6d1a533ab478ecf5ddaa11449a47752b2 ;; + */Megatron-LM) printf '%s\\n' 1dcf0dafa884ad52ffb243625717a3471643e087 ;; + esac +fi +''') + docker = bin_dir / "docker" + docker.write_text('#!/usr/bin/env bash\n[[ "$1" == pull ]]\n') + git.chmod(0o755) + docker.chmod(0o755) + env = {key: value for key, value in os.environ.items() + if key in {"PATH", "HOME", "LANG", "LC_ALL", "TMPDIR"}} + env["PATH"] = str(bin_dir) + os.pathsep + env["PATH"] + env["SGLANG_IMAGE"] = "registry/sglang:pinned" + prefix = "backend 'stack' $literal" + script = Path(__file__).resolve().parents[2] / "scripts/bootstrap_backends.sh" + subprocess.run(["bash", str(script), prefix], cwd=tmp_path, env=env, + check=True, capture_output=True, text=True, timeout=10) + expected_root = tmp_path / prefix + env_file = expected_root / "harborrl-backend.env" + assert env_file.read_text().count("export ") == 3 + # Source in a fresh shell from a different directory, then inspect a child. + child = subprocess.run( + ["bash", "-c", '. "$1"\nexec "$2" -c "$3"', "bootstrap-test", + str(env_file), sys.executable, + 'import json, os; print(json.dumps({k: os.environ.get(k) for k in ' + '("SLIME_DIR", "MEGATRON_DIR", "SGLANG_IMAGE")}))'], + cwd=bin_dir, env={key: value for key, value in env.items() if key != "SGLANG_IMAGE"}, + check=True, capture_output=True, text=True, timeout=10) + assert json.loads(child.stdout) == { + "SLIME_DIR": str(expected_root / "slime"), + "MEGATRON_DIR": str(expected_root / "Megatron-LM"), + "SGLANG_IMAGE": "registry/sglang:pinned", + } diff --git a/tests/harborrl/test_native_gateway.py b/tests/harborrl/test_native_gateway.py index 75ab7416e..17a6dbfbf 100644 --- a/tests/harborrl/test_native_gateway.py +++ b/tests/harborrl/test_native_gateway.py @@ -93,13 +93,21 @@ def test_model_output_never_repaired(text): parse_qwen_output(text, response_id="m", tools=[]) -def test_trace_requires_actual_consumption_and_is_immutable(tmp_path): +@pytest.mark.parametrize("delivery", ["sent", "delivery_unknown"]) +def test_trace_requires_actual_consumption_and_is_immutable(tmp_path, delivery): registry, _ident, token, gateway = setup(tmp_path) aid, message = gateway.generate(token, payload()) - registry.delivery(aid, message["id"], "sent") + registry.delivery(aid, message["id"], delivery) + unconsumed = registry.unconsumed_responses(aid) registry.close(aid) _turns, seal = registry.seal(aid) assert seal["errors"] == ["unconfirmed_consumption"] + assert registry.unconsumed_responses(aid) == unconsumed + if delivery == "delivery_unknown": + with pytest.raises(ValueError, match="delivery was not confirmed"): + registry.wait_delivered(aid, .1) + else: + registry.wait_delivered(aid, .1) with pytest.raises(ValueError, match="sealed"): registry.consume(aid, message["id"], message["content"], "session") with pytest.raises(PermissionError): @@ -120,6 +128,67 @@ def test_session_audit_covers_final_response(tmp_path): assert turns[0]["delivery"] == "consumed_confirmed" +def test_sealed_trace_releases_payload_and_remains_idempotent(tmp_path): + registry = TraceRegistry(tmp_path / "traces") + gateway = Gateway(registry, Backend(), model="policy") + for version in range(1, 5): + for slot in range(2): + ident = identity(f"{version}-{slot}", policy_version=str(version)) + token = registry.register(ident, "trial") + aid, message = gateway.generate(token, payload()) + registry.delivery(aid, message["id"], "sent") + registry.consume(aid, message["id"], message["content"], "session") + registry.close(aid) + turns, seal = registry.seal(aid) + persisted = (registry.attempts[aid]["root"] / "trace.json").read_bytes() + assert registry.attempts[aid]["turns"] == [] + assert registry.seal(aid) == (turns, seal) + turns[0]["input_ids"].append(999) + assert registry.seal(aid)[0][0]["input_ids"] == [1, 2] + assert (registry.attempts[aid]["root"] / "trace.json").read_bytes() == persisted + with pytest.raises(ValueError, match="registered"): + registry.register(ident, "other") + with pytest.raises(ValueError, match="unique"): + registry.register(identity(f"reuse-{version}-{slot}", policy_version=str(version)), + "other", credential=token) + registry.drain(.1) + next_version = str(version + 1) + registry.publish_version([{"healthy": True, "weight_version": next_version, + "endpoint": "engine"}], next_version) + assert len(registry.attempts) == 8 + assert all(not state["turns"] for state in registry.attempts.values()) + assert not registry.credentials + assert len(list((tmp_path / "traces").rglob("trace.json"))) == 8 + + +def test_seal_retains_payload_until_delivery_and_durable_write(tmp_path, monkeypatch): + from harborrl.gateway import trace + + registry, _, token, gateway = setup(tmp_path) + aid, message = gateway.generate(token, payload()) + registry.close(aid) + with pytest.raises(ValueError, match="drain"): + registry.seal(aid) + assert len(registry.attempts[aid]["turns"]) == 1 + registry.delivery(aid, message["id"], "sent") + original_publish = trace.publish + + def fail_trace(path, value): + if path.name == "trace.json": + raise OSError("disk unavailable") + return original_publish(path, value) + + with monkeypatch.context() as patch: + patch.setattr(trace, "publish", fail_trace) + with pytest.raises(OSError, match="disk unavailable"): + registry.seal(aid) + assert not registry.attempts[aid]["sealed"] + assert len(registry.attempts[aid]["turns"]) == 1 + turns, seal = registry.seal(aid) + assert len(turns) == 1 and seal["errors"] == ["unconfirmed_consumption"] + assert registry.attempts[aid]["turns"] == [] + + def test_session_audit_merges_split_assistant_blocks(tmp_path): class MultiBlockBackend(Backend): def generate(self, converted, ident, response_id): diff --git a/tests/harborrl/test_native_lifecycle.py b/tests/harborrl/test_native_lifecycle.py index a36ef0fd9..81c76ee56 100644 --- a/tests/harborrl/test_native_lifecycle.py +++ b/tests/harborrl/test_native_lifecycle.py @@ -128,6 +128,38 @@ async def fake_spawn(*args, **kwargs): assert json.loads((root / "runner-events.json").read_text())[0]["type"] == "prepared" +def test_cleanup_timeout_preserves_runner_marker_and_original_error(tmp_path, monkeypatch): + worker = tmp_path / "worker.py" + worker.write_text('''import json, sys, time +from pathlib import Path +req = json.loads(sys.stdin.readline()) +print(json.dumps({"type": "prepared", "trial_id": "trial"}), flush=True) +sys.stdin.readline() +output = Path(req["output"]) +output.mkdir(parents=True) +(output / "cleanup-required.json").write_text('{"reason": "deletion failed"}') +print(json.dumps({"type": "event", "trial_id": "other"}), flush=True) +time.sleep(30) +''') + spawn = asyncio.create_subprocess_exec + + async def fake_spawn(*args, **kwargs): + return await spawn(sys.executable, "-S", str(worker), **kwargs) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_spawn) + root = tmp_path / "attempt" + with pytest.raises(ValueError, match="event trial identity mismatch"): + asyncio.run(run_attempt(identity(), task_path=tmp_path, profile=PROFILE, + gateway_url="http://gateway", runner_python=sys.executable, root=root, + registry=TraceRegistry(tmp_path / "traces"), reward_profile=RewardProfile(), + timeout=2, cleanup_timeout=.1)) + marker = json.loads((root / "cleanup-required.json").read_text()) + assert marker["reason"] == "Inspect runner/cleanup-required.json" + assert marker["runner_host"] is None + assert json.loads((root / "failure.json").read_text())["error_type"] == "ValueError" + assert (root / "runner-events.json").is_file() + + def test_offline_http_to_export_is_explicitly_synthetic(tmp_path): from harborrl.rollout.harbor_job.offline import smoke from harborrl.trajectories.native import require_ready diff --git a/tests/harborrl/test_native_terminal.py b/tests/harborrl/test_native_terminal.py index eb94c8a98..ae7f3767a 100644 --- a/tests/harborrl/test_native_terminal.py +++ b/tests/harborrl/test_native_terminal.py @@ -73,14 +73,18 @@ async def fake_spawn(*args, **kwargs): return await spawn(sys.executable, "-S", str(worker), **kwargs) monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_spawn) + registry = Registry(tmp_path / "trace") ir = asyncio.run(run_attempt( ident, task_path=tmp_path, profile=PROFILE, gateway_url="http://gateway", - runner_python=sys.executable, root=root, registry=Registry(tmp_path / "trace"), + runner_python=sys.executable, root=root, registry=registry, reward_profile=RewardProfile())) assert ir["readiness"] == "CONTRACT_READY" assert json.loads((root / "trajectory.v2.json").read_text()) == ir assert (root / "evaluation.json").is_file() assert not (root / "cleanup-required.json").exists() + assert not (root / "trace-unsealed.json").exists() + assert registry.attempts[ident.attempt_id]["turns"] == [] + assert registry.seal(ident.attempt_id) == (ir["turns"], ir["trace_seal"]) assert all(turn["delivery"] == "consumed_confirmed" for turn in ir["turns"]) diff --git a/tests/harborrl/test_native_training.py b/tests/harborrl/test_native_training.py index ecfd6a24e..14328c2d5 100644 --- a/tests/harborrl/test_native_training.py +++ b/tests/harborrl/test_native_training.py @@ -146,10 +146,13 @@ def test_cli_command_mode_selects_experiment_dimensions(tmp_path, capsys, monkey ([], '0.0001'), (['training.learning_rate=0.000025'], '2.5e-05'), ]) -def test_native_shell_dry_run_uses_official_v032_hooks(tmp_path, overrides, expected_lr): +@pytest.mark.parametrize('layout', ['split', 'colocate']) +def test_native_shell_dry_run_uses_official_v032_hooks(tmp_path, overrides, expected_lr, layout): config_path, task = native_config(tmp_path) from harborrl.data.harbor.native_inspector import inspect_native - config = load_config(config_path, overrides) + total_gpus = '8' if layout == 'split' else '4' + config = load_config(config_path, [*overrides, f'deployment.layout={layout}', + f'deployment.num_gpus={total_gpus}']) catalog_path = Path(config['tasks']['catalog']) entries = json.loads(catalog_path.read_text()) entries[0]['task_digest'] = inspect_native(task)['task_digest'] @@ -171,9 +174,9 @@ def test_native_shell_dry_run_uses_official_v032_hooks(tmp_path, overrides, expe "HARNESS_OPTION": "native-claude-code", "HARBORRL_NATIVE_ROLLOUT": "1", "HARBORRL_SKIP_GLOBAL_CLEANUP": "1", "ROLLOUT_PROMPT_DATA": str(prompt), "HF_CKPT": "/model", "REF_LOAD": "/reference", "MODEL_ARGS_FILE": "qwen3-8B", - "MODEL_TAG": "qwen3-8B", "NUM_GPUS": "8", "ACTOR_GPUS": "4", + "MODEL_TAG": "qwen3-8B", "NUM_GPUS": total_gpus, "ACTOR_GPUS": "4", "ROLLOUT_GPUS": "4", "TP_SIZE": "4", "ROLLOUT_NUM_GPUS_PER_ENGINE": "4", - "HARBORRL_GPU_LAYOUT": "split", "NUM_ROLLOUT": "1", "SAVE_INTERVAL": "1", + "HARBORRL_GPU_LAYOUT": layout, "NUM_ROLLOUT": "1", "SAVE_INTERVAL": "1", "RUN_DIR": str(root), "RUN_ID": "run", "RUNS_ROOT": str(tmp_path), "CKPT_ROOT": str(root / "checkpoints"), "HARBORRL_NATIVE_MAX_ATTEMPTS": "2", "HARBORRL_NATIVE_DRAIN_TIMEOUT": "1", @@ -200,7 +203,7 @@ def test_native_shell_dry_run_uses_official_v032_hooks(tmp_path, overrides, expe '--num-rollout': '1', '--save-interval': '1', '--rollout-batch-size': '1', '--n-samples-per-prompt': '2', '--rollout-max-response-len': '128', '--rollout-max-context-len': '1024', - '--num-gpus-per-node': '8', '--actor-num-gpus-per-node': '4', + '--num-gpus-per-node': total_gpus, '--actor-num-gpus-per-node': '4', '--rollout-num-gpus': '4', '--tensor-model-parallel-size': '4', '--rollout-num-gpus-per-engine': '4', '--hf-checkpoint': '/model', '--ref-load': '/reference', '--prompt-data': str(prompt), @@ -210,6 +213,7 @@ def test_native_shell_dry_run_uses_official_v032_hooks(tmp_path, overrides, expe assert arguments[arguments.index(flag) + 1] == value recorded = json.loads(next((root / 'dry-run-commands').glob('*.json')).read_text()) assert recorded['command'] == arguments + assert arguments.count('--colocate') == int(layout == 'colocate') assert recorded['mode'] == 'dry_run' for argument in ( "--rollout-function-path harborrl.backends.slime_v032.rollout.generate_rollout", diff --git a/tests/harborrl/test_rollout_runtime_lifecycle.py b/tests/harborrl/test_rollout_runtime_lifecycle.py new file mode 100644 index 000000000..b94f33d01 --- /dev/null +++ b/tests/harborrl/test_rollout_runtime_lifecycle.py @@ -0,0 +1,213 @@ +"""CPU regressions for embedded loop ownership, cleanup admission and shutdown.""" +import asyncio +from concurrent.futures import ThreadPoolExecutor +from contextlib import nullcontext +from dataclasses import replace +from types import SimpleNamespace + +import pytest + +from harborrl.rollout import runtime + + +def cached_runtime_fixture(tmp_path, monkeypatch, *, sandbox=None): + from harborrl.gateway import server, sglang_backend + + catalog = tmp_path / "catalog.json" + catalog.write_text("[]") + profile = tmp_path / "profile.json" + profile.write_text('{"model": "policy"}') + tokenizer = SimpleNamespace(apply_chat_template=lambda *a, **k: [1, 2], + get_vocab=lambda: {"token": 1}) + tok, template = runtime.tokenizer_fingerprint(tokenizer) + config = runtime.RolloutRuntimeConfig( + "run", str(tmp_path), str(catalog), str(profile), "127.0.0.1", 0, + "http://gateway", "/python", (), {}, 2, 128, 1024, 1, 2, tok, template, + True, 1, "qwen3-8B", "http://router", sandbox=sandbox, runner_transport="local") + monkeypatch.setattr(server, "make_server", lambda *a, **k: SimpleNamespace( + serve_forever=lambda: None, shutdown=lambda: None, server_close=lambda: None)) + monkeypatch.setattr(sglang_backend, "SGLangBackend", lambda *a, **k: SimpleNamespace()) + return config, tokenizer + + +@pytest.mark.parametrize("sandbox_enabled", [False, True]) +def test_cached_embedded_runtime_survives_contended_rollout_rounds(tmp_path, monkeypatch, + sandbox_enabled): + runtime.close_runtime() + sandbox = {"max_live_instances": 1, + "bindings": [{"image": "image", "max_live_instances": 1}]} if sandbox_enabled else None + config, tokenizer = cached_runtime_fixture(tmp_path, monkeypatch, sandbox=sandbox) + seen, loops = [], [] + active, peak, completed = 0, 0, 0 + + async def round_(): + current = await runtime.runtime(config, tokenizer=tokenizer) + seen.append(current) + loops.append(asyncio.get_running_loop()) + + async def slot(index): + nonlocal active, peak, completed + admission = (current.sandbox_admission.lease(sandbox["bindings"][0], tmp_path / str(index)) + if sandbox else nullcontext()) + async with admission, current.semaphore: + active += 1 + peak = max(peak, active) + await asyncio.sleep(.01) + active -= 1 + completed += 1 + + await asyncio.gather(slot(0), slot(1)) + + try: + for _ in range(3): + runtime.run_embedded(round_()) + assert len({id(item) for item in seen}) == 1 + assert len({id(loop) for loop in loops}) == 1 + assert completed == 6 and peak == 1 + old_loop = loops[0] + runtime.close_runtime() + assert old_loop.is_closed() + runtime.run_embedded(round_()) + assert seen[-1] is not seen[0] and loops[-1] is not old_loop + finally: + runtime.close_runtime() + + +def test_cached_runtime_rejects_foreign_loop_and_config(tmp_path, monkeypatch): + runtime.close_runtime() + config, tokenizer = cached_runtime_fixture(tmp_path, monkeypatch) + try: + asyncio.run(runtime.runtime(config, tokenizer=tokenizer)) + with pytest.raises(runtime.RuntimeLifecycleError, match="different event loop"): + asyncio.run(runtime.runtime(config, tokenizer=tokenizer)) + with pytest.raises(runtime.RuntimeLifecycleError, match="configuration"): + asyncio.run(runtime.runtime(replace(config, run_id="other"), tokenizer=tokenizer)) + finally: + runtime.close_runtime() + + +def test_embedded_shutdown_awaits_tasks_before_closing_gateway(monkeypatch): + runtime.close_runtime() + events = [] + + async def pending(): + try: + await asyncio.sleep(60) + finally: + await asyncio.sleep(0) + events.append("task cleaned") + + async def launch(): + asyncio.create_task(pending()) + await asyncio.sleep(0) + + runtime.run_embedded(launch()) + loop = runtime._EMBEDDED_LOOP + monkeypatch.setattr(runtime, "_RUNTIME", SimpleNamespace(close=lambda: events.append("gateway closed"))) + runtime.close_runtime() + assert events == ["task cleaned", "gateway closed"] + assert loop.is_closed() + assert runtime._RUNTIME is None and runtime._EMBEDDED_LOOP is None + + +def test_embedded_failure_waits_for_other_groups_and_releases_loop(monkeypatch): + from harborrl.backends.slime_v032 import rollout + + runtime.close_runtime() + events = [] + + async def generate(args, group, params, *, rollout_id): + if group == "failed": + await asyncio.sleep(.01) + raise ValueError("original group error") + try: + await asyncio.sleep(60) + finally: + await asyncio.sleep(.01) + events.append("sibling cleaned") + + monkeypatch.delenv("HARBORRL_ROLLOUT_SERVICE_URL", raising=False) + monkeypatch.setattr(rollout, "generate_group", generate) + with pytest.raises(ValueError, match="original group error"): + runtime.run_embedded(rollout.run_native_groups( + SimpleNamespace(rollout_max_response_len=128), ["failed", "pending"], None, 0)) + assert events == ["sibling cleaned"] + assert runtime._EMBEDDED_LOOP is None and runtime._RUNTIME is None + + +@pytest.mark.parametrize("rollout_failed", [False, True]) +def test_embedded_shutdown_error_releases_resources_and_preserves_cause(monkeypatch, caplog, + rollout_failed): + runtime.close_runtime() + + def fail_close(): + raise RuntimeError("gateway shutdown failed") + + loops = [] + + async def round_(): + loops.append(asyncio.get_running_loop()) + monkeypatch.setattr(runtime, "_RUNTIME", SimpleNamespace(close=fail_close)) + if rollout_failed: + raise ValueError("original rollout failure") + + try: + if rollout_failed: + with pytest.raises(ValueError, match="original rollout failure"): + runtime.run_embedded(round_()) + assert "gateway shutdown failed" in caplog.text + else: + runtime.run_embedded(round_()) + with pytest.raises(RuntimeError, match="gateway shutdown failed"): + runtime.close_runtime() + assert loops[0].is_closed() + assert runtime._RUNTIME is None and runtime._EMBEDDED_LOOP is None + runtime.run_embedded(asyncio.sleep(0)) + finally: + runtime.close_runtime() + + +def test_embedded_rejects_foreign_thread_but_allows_idle_shutdown(): + runtime.close_runtime() + + async def round_(): + return asyncio.get_running_loop() + + try: + with ThreadPoolExecutor(max_workers=1) as pool: + loop = pool.submit(runtime.run_embedded, round_()).result(timeout=5) + with pytest.raises(runtime.RuntimeLifecycleError, match="owning thread"): + runtime.run_embedded(round_()) + runtime.close_runtime() + assert loop.is_closed() + finally: + runtime.close_runtime() + + +def test_embedded_rejects_nested_sync_calls_and_active_shutdown(): + runtime.close_runtime() + + async def round_(): + with pytest.raises(runtime.RuntimeLifecycleError, match="active event loop"): + runtime.run_embedded(asyncio.sleep(0)) + with pytest.raises(runtime.RuntimeLifecycleError, match="between calls"): + runtime.close_runtime() + + try: + runtime.run_embedded(round_()) + assert not runtime._EMBEDDED_LOOP.is_closed() + finally: + runtime.close_runtime() + + +@pytest.mark.parametrize("marker_path", ["cleanup-required.json", "runner/cleanup-required.json"]) +def test_runtime_restart_requires_cleanup_confirmation(tmp_path, monkeypatch, marker_path): + config, tokenizer = cached_runtime_fixture(tmp_path, monkeypatch) + marker = tmp_path / "native-rollouts/group/old-attempt" / marker_path + marker.parent.mkdir(parents=True) + marker.write_text('{}') + with pytest.raises(runtime.CleanupRequiredError, match="old-attempt"): + runtime.RolloutRuntime(config, tokenizer) + marker.unlink() + restarted = runtime.RolloutRuntime(config, tokenizer) + restarted.close() diff --git a/tests/harborrl/test_sandbox_runtime.py b/tests/harborrl/test_sandbox_runtime.py index 9f896818f..f558af24b 100644 --- a/tests/harborrl/test_sandbox_runtime.py +++ b/tests/harborrl/test_sandbox_runtime.py @@ -37,36 +37,97 @@ def request(method, path, body=None): assert calls == ['POST', 'GET', 'GET'] -def test_sandbox_cleanup_debt_prevents_retry(tmp_path, monkeypatch): +@pytest.mark.parametrize('provider', ['docker', 'sandbox']) +@pytest.mark.parametrize('marker_path', ['cleanup-required.json', 'runner/cleanup-required.json']) +def test_cleanup_debt_prevents_retry(tmp_path, monkeypatch, provider, marker_path): from harborrl.rollout import runtime from harborrl.rollout.harbor_job import sandbox_config profile = tmp_path / 'profile.json' profile.write_text('{}') binding = {'image': 'image', 'max_live_instances': 1} - sandbox = {'max_live_instances': 1, 'bindings': [binding]} + sandbox = {'max_live_instances': 1, 'bindings': [binding]} if provider == 'sandbox' else None config = RolloutRuntimeConfig('run', str(tmp_path), '', str(profile), '127.0.0.1', 1, 'http://gateway', '/python', (), {}, 2, 128, 1024, 2, 2, 'tok', 'tpl', True, 1, 'qwen3-8B', 'http://router', sandbox=sandbox, runner_transport='local') current = SimpleNamespace(rollout_root=tmp_path, semaphore=asyncio.Semaphore(2), - registry=None, profile={}, sandbox_admission=SandboxAdmission(sandbox)) + registry=None, profile={}) + if sandbox: + current.sandbox_admission = SandboxAdmission(sandbox) calls = [] async def attempt(identity, **kwargs): calls.append(kwargs) assert kwargs['runner_host'] is None - assert kwargs['sandbox'] == sandbox - kwargs['root'].mkdir(parents=True) - (kwargs['root'] / 'cleanup-required.json').write_text('{}') + assert kwargs.get('sandbox') == sandbox + marker = kwargs['root'] / marker_path + marker.parent.mkdir(parents=True) + marker.write_text('{}') raise RuntimeError('deletion unconfirmed') monkeypatch.setattr(runtime, 'run_attempt', attempt) monkeypatch.setattr(sandbox_config, 'task_binding', lambda *args: binding) entry = {'id': 'task', 'path': '/task', 'task_digest': 'digest', 'reward_profile': {}} with pytest.raises(RuntimeError, match='cleanup unresolved'): asyncio.run(run_slot(config, current, 0, 0, 0, entry, 'sampling', '1')) - assert len(calls) == 1 and current.sandbox_admission.blocked + assert len(calls) == 1 + if sandbox: + assert current.sandbox_admission.blocked + with pytest.raises(runtime.CleanupRequiredError, match='cleanup-required.json'): + runtime.CleanupAdmission(tmp_path).check() + (calls[0]['root'] / marker_path).unlink() + current.cleanup.check() + runtime.CleanupAdmission(tmp_path).check() + if sandbox: + assert not current.sandbox_admission.blocked restored = replace(config, workers=('worker',), runner_transport='ssh') assert json.loads(restored.environment()['HARBORRL_NATIVE_SANDBOX']) == sandbox +def test_docker_cleanup_debt_blocks_waiting_slots(tmp_path, monkeypatch): + from harborrl.rollout import runtime + + calls = [] + + async def attempt(identity, **kwargs): + calls.append(identity) + await asyncio.sleep(.01) + kwargs['root'].mkdir(parents=True) + (kwargs['root'] / 'cleanup-required.json').write_text('{}') + raise RuntimeError('Runner cleanup failed') + + monkeypatch.setattr(runtime, 'run_attempt', attempt) + config = SimpleNamespace(run_id='run', group_size=2, max_attempts=3, workers=(), + task_worker_overrides={}, runner_transport='local', sandbox=None, + gateway_url='http://gateway', runner_python='/python', max_output_tokens=128) + current = SimpleNamespace(profile={}, rollout_root=tmp_path, + semaphore=asyncio.Semaphore(1), registry=None) + entry = {'id': 'task', 'path': '/task', 'task_digest': 'digest', 'reward_profile': {}} + with pytest.raises(runtime.CleanupRequiredError, match='automatic retry blocked') as error: + asyncio.run(runtime.run_group(config, current, entry=entry, + group_index=0, rollout_id=0, sampling_digest='x', policy_version='1')) + assert len(calls) == 1 + assert isinstance(error.value.__context__, RuntimeError) + + +def test_runtime_lifecycle_failure_is_not_retried(tmp_path, monkeypatch): + from harborrl.rollout import runtime + + calls = [] + + async def attempt(identity, **kwargs): + calls.append(identity) + raise runtime.RuntimeLifecycleError('wrong event loop') + + monkeypatch.setattr(runtime, 'run_attempt', attempt) + config = SimpleNamespace(run_id='run', group_size=2, max_attempts=3, workers=(), + task_worker_overrides={}, runner_transport='local', sandbox=None, + gateway_url='http://gateway', runner_python='/python', max_output_tokens=128) + current = SimpleNamespace(profile={}, rollout_root=tmp_path, + semaphore=asyncio.Semaphore(1), registry=None) + entry = {'id': 'task', 'path': '/task', 'task_digest': 'digest', 'reward_profile': {}} + with pytest.raises(runtime.RuntimeLifecycleError, match='wrong event loop'): + asyncio.run(run_slot(config, current, 0, 0, 0, entry, 'sampling', '1')) + assert len(calls) == 1 + + def test_group_failure_waits_for_sibling_cleanup(tmp_path, monkeypatch): from harborrl.rollout import runtime events = [] diff --git a/tests/harborrl/test_slime_v032_exporter.py b/tests/harborrl/test_slime_v032_exporter.py index 41ce72ef9..6597a579d 100644 --- a/tests/harborrl/test_slime_v032_exporter.py +++ b/tests/harborrl/test_slime_v032_exporter.py @@ -355,7 +355,12 @@ def test_launcher_hooks_resolve_and_dry_run(): v032_launcher.hook_arguments("slime-legacy") -def test_rollout_hook_locks_policy_version_and_keeps_groups_atomic(tmp_path, monkeypatch): +def test_rollout_hook_locks_policy_version_and_keeps_groups_atomic(tmp_path, monkeypatch, request): + import asyncio + from harborrl.rollout.runtime import close_runtime + + close_runtime() + request.addfinalizer(close_runtime) groups = native_groups(tmp_path) locked = [ [reidentify(json.loads(json.dumps(ir)), policy_version="7") for ir in group] @@ -368,10 +373,18 @@ def test_rollout_hook_locks_policy_version_and_keeps_groups_atomic(tmp_path, mon args = SimpleNamespace( rollout_batch_size=2, n_samples_per_prompt=2, rollout_max_response_len=128, ) - events = [] + events, loops = [], [] + semaphore = asyncio.Semaphore(1) async def runner(args_, groups_, policy_version): events.append(("generate", policy_version)) + loops.append(asyncio.get_running_loop()) + + async def queued_slot(): + async with semaphore: + await asyncio.sleep(.001) + + await asyncio.gather(queued_slot(), queued_slot()) return json.loads(json.dumps(locked)) class SlimeDataSource: @@ -404,6 +417,13 @@ def get_samples(self, count): ) assert all(sample.metadata["native_ir"]["identity"]["policy_version"] == "7" for group in output.samples for sample in group) + second = generate_rollout( + args, 4, SlimeDataSource(), runner=runner, output_factory=SimpleNamespace, + sample_factory=lambda **fields: SimpleNamespace(**fields)) + assert loops[0] is loops[1] + assert second.metrics["native_turn_count"] == 7.0 + assert {sample.rollout_id for group in second.samples for sample in group} == {8, 9} + async def stale_runner(args_, groups_, policy_version): return [[reidentify(json.loads(json.dumps(ir)), policy_version="8") for ir in group] for group in locked] From 7605e7f2a5e8313daf3e8eb28fc6331f73f7f5b9 Mon Sep 17 00:00:00 2001 From: luyd Date: Wed, 28 Oct 2026 12:09:23 +0800 Subject: [PATCH 2/6] Simplify rollout cancellation and cleanup records Share ordered task gathering and sibling cancellation between group and batch rollout. Consolidate cleanup marker publication while preserving the first cause and recording Runner identity. Extend cancellation and remote Runner regressions. Validation: 295 tests passed, one environment-dependent module skipped; git diff --check passed. --- harborrl/backends/slime_v032/rollout.py | 14 +++------- harborrl/rollout/harbor_job/coordinator.py | 28 ++++++++++--------- harborrl/rollout/runtime.py | 26 +++++++++-------- tests/harborrl/test_native_lifecycle.py | 14 +++++++--- .../test_rollout_runtime_lifecycle.py | 7 +++-- 5 files changed, 48 insertions(+), 41 deletions(-) diff --git a/harborrl/backends/slime_v032/rollout.py b/harborrl/backends/slime_v032/rollout.py index 450beb6d5..fd00f1d82 100644 --- a/harborrl/backends/slime_v032/rollout.py +++ b/harborrl/backends/slime_v032/rollout.py @@ -18,8 +18,9 @@ RolloutRuntimeConfig, current_policy_version, digest, - run_group, + gather_or_cancel, run_embedded, + run_group, runtime, task_entry, ) @@ -195,15 +196,8 @@ async def run_native_groups(args, groups, policy_version, rollout_id): ir_groups = await run_service_groups(args, groups, rollout_id) assert_locked_policy_versions(ir_groups, policy_version) return ir_groups - pending = [asyncio.create_task( - generate_group(args, group, params, rollout_id=rollout_id)) for group in groups] - try: - results = await asyncio.gather(*pending) - except BaseException: - for task in pending: - task.cancel() - await asyncio.gather(*pending, return_exceptions=True) - raise + results = await gather_or_cancel( + generate_group(args, group, params, rollout_id=rollout_id) for group in groups) ir_groups = [] for group in results: trajectories = [] diff --git a/harborrl/rollout/harbor_job/coordinator.py b/harborrl/rollout/harbor_job/coordinator.py index 49ea55aae..7316d5bba 100644 --- a/harborrl/rollout/harbor_job/coordinator.py +++ b/harborrl/rollout/harbor_job/coordinator.py @@ -94,6 +94,15 @@ async def run_attempt(identity, *, task_path, profile, gateway_url, runner_pytho raise registered, trial_id, lifecycle_timeout, sealed = False, None, False, False messages = [] + + def mark_cleanup(reason): + marker = root / "cleanup-required.json" + if not marker.exists(): + publish(marker, { + "identity": identity.to_dict(), "trial_id": trial_id, + "runner_host": runner_host, "reason": reason, + }) + try: req = {"identity": identity.to_dict(), "task_path": str(Path(task_path).resolve()), "profile": profile, "gateway_url": gateway_url, "credential": credential, "output": str(root / "runner"), @@ -142,8 +151,7 @@ async def lifecycle(): lifecycle_path = root / 'runner' / 'sandbox-lifecycle.json' lifecycle = read_json(lifecycle_path) if lifecycle_path.is_file() else {} if lifecycle.get('state') != 'deleted' or lifecycle.get('runtime_digest') != req['runtime_digest']: - publish(root / 'cleanup-required.json', {'identity': identity.to_dict(), 'runner_host': runner_host, - 'reason': 'Sandbox deletion unconfirmed'}) + mark_cleanup("Sandbox deletion unconfirmed") raise RuntimeError('sandbox deletion unconfirmed') trial_dir = Path(terminal["trial_dir"]).resolve() result_path = Path(terminal["result_path"]).resolve() @@ -168,9 +176,8 @@ async def lifecycle(): publish(root / "failure.json", {"identity": identity.to_dict(), "error_type": type(exc).__name__}) raise finally: - if (root / 'runner' / 'cleanup-required.json').exists() and not (root / 'cleanup-required.json').exists(): - publish(root / 'cleanup-required.json', {'identity': identity.to_dict(), 'runner_host': runner_host, - 'reason': 'Inspect runner/cleanup-required.json'}) + if (root / "runner" / "cleanup-required.json").exists(): + mark_cleanup("Inspect runner/cleanup-required.json") if registered: registry.close(identity.attempt_id) if process.returncode is None: @@ -184,14 +191,9 @@ async def lifecycle(): except ProcessLookupError: pass await process.wait() - if not (root / "cleanup-required.json").exists(): - publish(root / "cleanup-required.json", {"identity": identity.to_dict(), "trial_id": trial_id, - "runner_host": runner_host, - "reason": "Runner cleanup deadline exceeded; inspect Harbor resources before retry"}) - if (process.returncode != 0 or lifecycle_timeout) and not (root / "cleanup-required.json").exists(): - publish(root / "cleanup-required.json", {"identity": identity.to_dict(), "trial_id": trial_id, - "runner_host": runner_host, - "reason": "Runner exited abnormally; inspect Harbor resources before retry"}) + mark_cleanup("Runner cleanup deadline exceeded; inspect Harbor resources before retry") + if process.returncode != 0 or lifecycle_timeout: + mark_cleanup("Runner exited abnormally; inspect Harbor resources before retry") if registered and not sealed: try: registry.seal(identity.attempt_id) diff --git a/harborrl/rollout/runtime.py b/harborrl/rollout/runtime.py index 99bffcf87..9288ebef8 100644 --- a/harborrl/rollout/runtime.py +++ b/harborrl/rollout/runtime.py @@ -451,6 +451,18 @@ async def _run_slot(config, runtime_, slots, slot, entry, reward_profile): raise RuntimeError(f"native slot {slot} exhausted its retry budget") from last_error +async def gather_or_cancel(coroutines): + """Return ordered results, awaiting sibling cleanup before propagating failure.""" + tasks = [asyncio.create_task(coroutine) for coroutine in coroutines] + try: + return await asyncio.gather(*tasks) + except BaseException: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise + + async def run_group(config, runtime_, *, entry, group_index, rollout_id, sampling_digest, policy_version, group_size=None): """Execute one complete group and return per-slot IRs in slot order.""" @@ -458,14 +470,6 @@ async def run_group(config, runtime_, *, entry, group_index, rollout_id, slots, reward_profile = _group_slots( config, runtime_, group_index, rollout_id, entry, sampling_digest, policy_version, size) - tasks = [asyncio.create_task( - _run_slot(config, runtime_, slots, slot, entry, reward_profile)) - for slot in range(size) - ] - try: - return list(await asyncio.gather(*tasks)) - except BaseException: - for task in tasks: - task.cancel() - await asyncio.gather(*tasks, return_exceptions=True) - raise + return await gather_or_cancel( + _run_slot(config, runtime_, slots, slot, entry, reward_profile) + for slot in range(size)) diff --git a/tests/harborrl/test_native_lifecycle.py b/tests/harborrl/test_native_lifecycle.py index 81c76ee56..0c1ecd743 100644 --- a/tests/harborrl/test_native_lifecycle.py +++ b/tests/harborrl/test_native_lifecycle.py @@ -124,11 +124,15 @@ async def fake_spawn(*args, **kwargs): assert registry.attempts[identity().attempt_id]["closed"] assert (root / "failure.json").is_file() if mode != "wrong-event": - assert (root / "cleanup-required.json").is_file() + marker = json.loads((root / "cleanup-required.json").read_text()) + assert marker["identity"] == identity().to_dict() + assert marker["trial_id"] == "trial" + assert marker["runner_host"] is None assert json.loads((root / "runner-events.json").read_text())[0]["type"] == "prepared" -def test_cleanup_timeout_preserves_runner_marker_and_original_error(tmp_path, monkeypatch): +@pytest.mark.parametrize("runner_host", [None, "runner-host"]) +def test_cleanup_timeout_preserves_runner_marker_and_original_error(tmp_path, monkeypatch, runner_host): worker = tmp_path / "worker.py" worker.write_text('''import json, sys, time from pathlib import Path @@ -152,10 +156,12 @@ async def fake_spawn(*args, **kwargs): asyncio.run(run_attempt(identity(), task_path=tmp_path, profile=PROFILE, gateway_url="http://gateway", runner_python=sys.executable, root=root, registry=TraceRegistry(tmp_path / "traces"), reward_profile=RewardProfile(), - timeout=2, cleanup_timeout=.1)) + timeout=2, cleanup_timeout=.1, runner_host=runner_host)) marker = json.loads((root / "cleanup-required.json").read_text()) assert marker["reason"] == "Inspect runner/cleanup-required.json" - assert marker["runner_host"] is None + assert marker["identity"] == identity().to_dict() + assert marker["trial_id"] == "trial" + assert marker["runner_host"] == runner_host assert json.loads((root / "failure.json").read_text())["error_type"] == "ValueError" assert (root / "runner-events.json").is_file() diff --git a/tests/harborrl/test_rollout_runtime_lifecycle.py b/tests/harborrl/test_rollout_runtime_lifecycle.py index b94f33d01..6d5620058 100644 --- a/tests/harborrl/test_rollout_runtime_lifecycle.py +++ b/tests/harborrl/test_rollout_runtime_lifecycle.py @@ -110,7 +110,8 @@ async def launch(): assert runtime._RUNTIME is None and runtime._EMBEDDED_LOOP is None -def test_embedded_failure_waits_for_other_groups_and_releases_loop(monkeypatch): +@pytest.mark.parametrize("error_type", [ValueError, asyncio.CancelledError]) +def test_embedded_failure_waits_for_other_groups_and_releases_loop(monkeypatch, error_type): from harborrl.backends.slime_v032 import rollout runtime.close_runtime() @@ -119,7 +120,7 @@ def test_embedded_failure_waits_for_other_groups_and_releases_loop(monkeypatch): async def generate(args, group, params, *, rollout_id): if group == "failed": await asyncio.sleep(.01) - raise ValueError("original group error") + raise error_type("original group error") try: await asyncio.sleep(60) finally: @@ -128,7 +129,7 @@ async def generate(args, group, params, *, rollout_id): monkeypatch.delenv("HARBORRL_ROLLOUT_SERVICE_URL", raising=False) monkeypatch.setattr(rollout, "generate_group", generate) - with pytest.raises(ValueError, match="original group error"): + with pytest.raises(error_type, match="original group error"): runtime.run_embedded(rollout.run_native_groups( SimpleNamespace(rollout_max_response_len=128), ["failed", "pending"], None, 0)) assert events == ["sibling cleaned"] From b456f14fa7a89d1fa67284c1f05654a0b91025d2 Mon Sep 17 00:00:00 2001 From: luyd Date: Wed, 28 Oct 2026 16:53:58 +0800 Subject: [PATCH 3/6] Fix native example, sampling locks, and Claude controls --- README.md | 6 +++- README_zh.md | 6 +++- examples/native/catalog.json | 2 +- examples/native/catalog.jsonl | 2 +- examples/native/claude-code-profile.json | 8 +++--- examples/native/tasks/hello_world/task.toml | 8 +++--- .../native/tasks/hello_world/tests/test.sh | 8 ++++-- harborrl/backends/slime_v032/rollout.py | 9 +++--- harborrl/gateway/messages.py | 8 +++--- harborrl/rollout/harbor_job/bindings.py | 3 ++ harborrl/rollout/service.py | 4 +-- harborrl/trajectories/native.py | 10 +++++++ tests/harborrl/test_native_gateway.py | 19 +++++++++++++ tests/harborrl/test_native_lifecycle.py | 11 ++++++-- tests/harborrl/test_rollout_service.py | 28 +++++++++++++++++++ 15 files changed, 105 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index bdb9f0e3c..71ca79609 100644 --- a/README.md +++ b/README.md @@ -165,7 +165,11 @@ task/ Here, `task.toml` declares resources, the agent, and the verifier; `instruction.md` is the agent-facing instruction; `environment/Dockerfile` builds the isolated environment; and `tests/test.sh` verifies the result and writes `reward.json` or `reward.txt`. For training, tasks are locked into a catalog with their task ID, upstream revision, local path, content digest, and reward parsing rules; see [examples/native/catalog.json](examples/native/catalog.json). -The bundled runnable example is `hello_world`. Real training tasks are not repackaged by this repository: fetch Harbor-native datasets or tasks from Harbor Hub, and use the [compatibility summary](configs/harbor_hub/public_compatibility_summary.md) to understand validated coverage. After fetching a task root, validate it locally with `harborrl hub inspect` and lock it into your own catalog. +The bundled runnable example is `hello_world`. It allows internet access so Harbor can install the pinned Claude Code CLI, reserves 4 GiB each for memory and storage, and writes a 0/1 reward to `/logs/verifier/reward.json`. Its profile allows four turns so the agent can observe a tool result and finish. Offline deployments need the exact CLI version preinstalled and a reachable policy gateway. After changing a task, regenerate its catalog digest. + +The example pins the runtime-validated Claude Code `2.1.141`. The adapter disables prompt caching and cache warmup, compaction, and nonessential traffic, and always disallows `Agent` and `Task` even with a custom tool denylist. CLI `2.1.0` still makes an independent file-path-extraction model request after Bash and is incompatible with the single-context rollout contract. Probe other CLI versions before use; native session audit rejects auxiliary contexts, summaries, and unconsumed model responses. + +Real training tasks are not repackaged by this repository: fetch Harbor-native datasets or tasks from Harbor Hub, and use the [compatibility summary](configs/harbor_hub/public_compatibility_summary.md) to understand validated coverage. After fetching a task root, validate it locally with `harborrl hub inspect` and lock it into your own catalog. ## License diff --git a/README_zh.md b/README_zh.md index 22cfcedd9..347b1f1e4 100644 --- a/README_zh.md +++ b/README_zh.md @@ -154,7 +154,11 @@ task/ 其中 `task.toml` 声明资源、agent 与 verifier，`instruction.md` 是给 agent 的任务说明，`environment/Dockerfile` 构建隔离执行环境，`tests/test.sh` 负责验证并输出 `reward.json` 或 `reward.txt`。训练时，这些任务会被锁定到 catalog 中，记录任务 ID、上游 revision、本地路径、内容 digest 和 reward 解析规则；示例见 [examples/native/catalog.json](examples/native/catalog.json)。 -仓库内置的可直接运行示例是 `hello_world`。实际训练任务不由本 repo 重新打包，而是从 Harbor Hub 获取符合 Harbor native task 格式的 dataset/task；可先查看[兼容性汇总](configs/harbor_hub/public_compatibility_summary_zh.md)了解已验证覆盖情况。取回 task root 后，先用 `harborrl hub inspect` 做本地静态校验，再锁定进自己的 catalog。 +仓库内置的可直接运行示例是 `hello_world`。示例允许联网，以便 Harbor 安装固定版本的 Claude Code CLI；内存与存储各配置 4 GiB，verifier 将 0/1 奖励写入 `/logs/verifier/reward.json`。profile 允许四轮交互，供 agent 执行工具、读取结果并正常结束。离线部署需要预装完全一致的 CLI 版本，并确保能够访问 policy gateway。修改任务内容后，需要重新生成 catalog 中的内容摘要。 + +示例固定使用经过运行验证的 Claude Code `2.1.141`。adapter 默认关闭 prompt caching 及缓存预热、上下文压缩和非必要流量；自定义工具禁用列表也会保留 `Agent` 和 `Task`。CLI `2.1.0` 在 Bash 执行后仍会发起独立的文件路径提取请求，不符合单上下文 rollout 契约。使用其他 CLI 版本前需要重新验证；原生会话审计会拒绝辅助上下文、摘要和未消费的模型响应。 + +实际训练任务不由本 repo 重新打包，而是从 Harbor Hub 获取符合 Harbor native task 格式的 dataset/task；可先查看[兼容性汇总](configs/harbor_hub/public_compatibility_summary_zh.md)了解已验证覆盖情况。取回 task root 后，先用 `harborrl hub inspect` 做本地静态校验，再锁定进自己的 catalog。 ## License diff --git a/examples/native/catalog.json b/examples/native/catalog.json index b8b090ada..70854815e 100644 --- a/examples/native/catalog.json +++ b/examples/native/catalog.json @@ -12,6 +12,6 @@ "scale": 1.0 }, "revision": "v0.1-example", - "task_digest": "905f496d2852cca5b1e86e9de72395b41c291d2b8f6a80807e0ed24ca19541b3" + "task_digest": "2293cd4682597d9c6c5a4bb35d2ded9978873a8deb48ee07fae4e58121c79691" } ] diff --git a/examples/native/catalog.jsonl b/examples/native/catalog.jsonl index eba16a31c..0afb50238 100644 --- a/examples/native/catalog.jsonl +++ b/examples/native/catalog.jsonl @@ -1 +1 @@ -{"id": "hello_world", "path": "tasks/hello_world", "revision": "v0.1-example", "reward_profile": {"key": "reward", "offset": 0.0, "raw_range": [0.0, 1.0], "scale": 1.0}, "task_digest": "905f496d2852cca5b1e86e9de72395b41c291d2b8f6a80807e0ed24ca19541b3"} +{"id": "hello_world", "path": "tasks/hello_world", "revision": "v0.1-example", "reward_profile": {"key": "reward", "offset": 0.0, "raw_range": [0.0, 1.0], "scale": 1.0}, "task_digest": "2293cd4682597d9c6c5a4bb35d2ded9978873a8deb48ee07fae4e58121c79691"} diff --git a/examples/native/claude-code-profile.json b/examples/native/claude-code-profile.json index bbc2804a9..0434f902a 100644 --- a/examples/native/claude-code-profile.json +++ b/examples/native/claude-code-profile.json @@ -1,8 +1,8 @@ { - "agent_timeout_sec": 60, - "cli_version": "2.1.0", - "max_turns": 1, + "agent_timeout_sec": 120, + "cli_version": "2.1.141", + "max_turns": 4, "model": "policy", - "setup_timeout_sec": 60, + "setup_timeout_sec": 300, "verifier_timeout_sec": 60 } diff --git a/examples/native/tasks/hello_world/task.toml b/examples/native/tasks/hello_world/task.toml index f3534bd83..8323405d0 100644 --- a/examples/native/tasks/hello_world/task.toml +++ b/examples/native/tasks/hello_world/task.toml @@ -3,14 +3,14 @@ version = "1.0" [environment] os = "linux" cpus = 1 -memory_mb = 512 -storage_mb = 256 -allow_internet = false +memory_mb = 4096 +storage_mb = 4096 +allow_internet = true gpus = 0 [agent] name = "claude-code" -max_turns = 1 +max_turns = 4 [verifier] type = "shared" diff --git a/examples/native/tasks/hello_world/tests/test.sh b/examples/native/tasks/hello_world/tests/test.sh index d86585c80..99dd9cd76 100755 --- a/examples/native/tasks/hello_world/tests/test.sh +++ b/examples/native/tasks/hello_world/tests/test.sh @@ -1,5 +1,9 @@ #!/bin/sh set -eu -test "$(cat /hello 2>/dev/null || true)" = "ok" -printf '{"reward":1}\n' >reward.json +reward=0 +if [ "$(cat /hello 2>/dev/null || true)" = "ok" ]; then + reward=1 +fi +mkdir -p /logs/verifier +printf '{"reward":%s}\n' "$reward" >/logs/verifier/reward.json diff --git a/harborrl/backends/slime_v032/rollout.py b/harborrl/backends/slime_v032/rollout.py index fd00f1d82..c8b6ffd9d 100644 --- a/harborrl/backends/slime_v032/rollout.py +++ b/harborrl/backends/slime_v032/rollout.py @@ -24,19 +24,20 @@ runtime, task_entry, ) +from harborrl.trajectories.native import canonical_sampling def sampling_params(args): max_new_tokens = getattr(args, "rollout_max_response_len", None) if type(max_new_tokens) is not int or max_new_tokens < 1: raise ValueError("native rollout requires a positive rollout_max_response_len") - return { + return canonical_sampling({ "temperature": getattr(args, "rollout_temperature", 1.0), "top_p": getattr(args, "rollout_top_p", 1.0), "top_k": getattr(args, "rollout_top_k", -1), "max_new_tokens": max_new_tokens, "stop": list(getattr(args, "rollout_stop", None) or []), - } + }) def runtime_config_from_args(args): @@ -92,13 +93,13 @@ async def generate_group(args, group, sampling_params_, rollout_id, evaluation=F if report["status"] != "NEEDS_PROBE": raise ValueError(f"native task preflight failed: {report['reasons']}") stop = list(sampling_params_.get("stop") or []) - sampling_digest = digest({ + sampling_digest = digest(canonical_sampling({ "temperature": sampling_params_.get("temperature"), "top_p": sampling_params_.get("top_p"), "top_k": sampling_params_.get("top_k"), "max_new_tokens": sampling_params_.get("max_new_tokens"), "stop": stop, - }) + })) config = runtime_config_from_args(args) runtime_ = await runtime(config, tokenizer_loader=slime_tokenizer_loader(args)) policy_version = current_policy_version(runtime_) diff --git a/harborrl/gateway/messages.py b/harborrl/gateway/messages.py index 9c474817b..08578c2a3 100644 --- a/harborrl/gateway/messages.py +++ b/harborrl/gateway/messages.py @@ -2,7 +2,7 @@ import json import re -from harborrl.trajectories.native import finite +from harborrl.trajectories.native import canonical_sampling, finite class ProtocolError(ValueError): @@ -40,7 +40,7 @@ def convert(payload, *, model, max_output_tokens=8192, count_only=False): if type(max_tokens) is not int or not 0 < max_tokens <= max_output_tokens: raise ProtocolError("max_tokens outside profile budget") temperature, top_p, top_k = payload.get("temperature", 1.0), payload.get("top_p", 1.0), payload.get("top_k", -1) - # Initial raw-model recipe fixes the distribution; no silent normalization. + # Initial raw-model recipe fixes the distribution, regardless of JSON numeric type. if any(not finite(v) for v in (temperature, top_p)) or temperature != 1 or top_p != 1 or type(top_k) is not int or top_k != -1: raise ProtocolError("initial profile requires temperature=1, top_p=1, top_k=-1") stop = payload.get("stop_sequences", []) @@ -129,9 +129,9 @@ def convert(payload, *, model, max_output_tokens=8192, count_only=False): messages.append(msg) if pending: raise ProtocolError("missing tool results in generation context") - return {"messages": messages, "tools": tools, "sampling": { + return {"messages": messages, "tools": tools, "sampling": canonical_sampling({ "temperature": temperature, "top_p": top_p, "top_k": top_k, - "max_new_tokens": max_tokens, "stop": stop}} + "max_new_tokens": max_tokens, "stop": stop})} def parse_qwen_output(text, *, response_id, tools): diff --git a/harborrl/rollout/harbor_job/bindings.py b/harborrl/rollout/harbor_job/bindings.py index e0e553e85..d0da81703 100644 --- a/harborrl/rollout/harbor_job/bindings.py +++ b/harborrl/rollout/harbor_job/bindings.py @@ -19,6 +19,7 @@ def claude_config(profile, gateway_url, credential, max_output_tokens=8192): if not isinstance(disallowed_tools, str) or not re.fullmatch( r"[A-Za-z][A-Za-z0-9_]*(?:,[A-Za-z][A-Za-z0-9_]*)*", disallowed_tools): raise ValueError("disallowed tools must be a comma-separated identifier list") + disallowed_tools = ",".join(dict.fromkeys(["Agent", "Task", *disallowed_tools.split(",")])) url = urlsplit(gateway_url) if url.scheme not in ("http", "https") or not url.hostname or url.username or url.password or url.query or url.fragment or url.path not in ("", "/"): raise ValueError("Anthropic base URL must be a plain origin (no /v1 suffix)") @@ -43,9 +44,11 @@ def claude_config(profile, gateway_url, credential, max_output_tokens=8192): def client_controls(): """Pass controls into the Agent environment, not only the Runner host.""" + # Disabling prompt caching also prevents cache-warming model requests. return {"CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1", "CLAUDE_CODE_DISABLE_ADAPTIVE_THINKING": "1", "CLAUDE_CODE_DISABLE_AUTO_MEMORY": "1", + "DISABLE_PROMPT_CACHING": "1", "DISABLE_COMPACT": "1", "DISABLE_AUTOUPDATER": "1", "DISABLE_AUTO_COMPACT": "1"} diff --git a/harborrl/rollout/service.py b/harborrl/rollout/service.py index e223ab05a..c71cff61a 100644 --- a/harborrl/rollout/service.py +++ b/harborrl/rollout/service.py @@ -29,7 +29,7 @@ from harborrl.rollout.exporters.native import export_group from harborrl.rollout.runtime import run_group -from harborrl.trajectories.native import digest, finite, publish, read_json +from harborrl.trajectories.native import canonical_sampling, digest, finite, publish, read_json SERVICE_SCHEMA = "harborrl-rollout-service-v1" GROUP_REQUEST_KEYS = {"group_id", "rollout_id", "group_index", "policy_version", @@ -147,7 +147,7 @@ async def default_executor(config, runtime_, request): """The production executor: the framework-free rollout core.""" return await run_group( config, runtime_, entry=request["task"], group_index=request["group_index"], - rollout_id=request["rollout_id"], sampling_digest=digest(request["sampling"]), + rollout_id=request["rollout_id"], sampling_digest=digest(canonical_sampling(request["sampling"])), policy_version=request["policy_version"], group_size=len(request["slots"])) diff --git a/harborrl/trajectories/native.py b/harborrl/trajectories/native.py index 5de91de56..8043c670d 100644 --- a/harborrl/trajectories/native.py +++ b/harborrl/trajectories/native.py @@ -14,6 +14,16 @@ def finite(value): return type(value) in (int, float) and math.isfinite(value) +def canonical_sampling(sampling): + """Give equivalent JSON probabilities one representation before hashing.""" + result = dict(sampling) + for key in ("temperature", "top_p"): + if not finite(result[key]): + raise ValueError(f"sampling {key} must be a finite number") + result[key] = float(result[key]) + return result + + def encode(value): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() diff --git a/tests/harborrl/test_native_gateway.py b/tests/harborrl/test_native_gateway.py index 17a6dbfbf..0055fe515 100644 --- a/tests/harborrl/test_native_gateway.py +++ b/tests/harborrl/test_native_gateway.py @@ -15,6 +15,7 @@ from harborrl.gateway.server import Gateway, make_server from harborrl.gateway.trace import TraceRegistry from harborrl.rollout.harbor_job.audit import audit_session +from harborrl.trajectories.native import digest def payload(**changes): @@ -42,6 +43,8 @@ def setup(tmp_path): @pytest.mark.parametrize("change", [{"model": "external"}, {"thinking": {"type": "enabled"}}, {"temperature": .7}, {"top_p": .9}, {"top_k": 1}, {"max_tokens": True}, {"stream": 1}, + {"temperature": True}, {"top_p": True}, {"temperature": "1"}, {"top_p": "1"}, + {"temperature": float("nan")}, {"top_p": float("inf")}, {"messages": []}, {"messages": [{"role": "assistant", "content": [{"type": "image"}]}]}, {"tool_choice": {"type": "any"}}]) def test_unsupported_request(change): @@ -49,6 +52,22 @@ def test_unsupported_request(change): convert(payload(**change), model="policy") +@pytest.mark.parametrize("temperature", [1, 1.0], ids=["int", "float"]) +@pytest.mark.parametrize("top_p", [1, 1.0], ids=["int", "float"]) +def test_equivalent_sampling_numbers_share_attempt_lock(tmp_path, temperature, top_p): + registry, ident, token, gateway = setup(tmp_path) + aid, message = gateway.generate(token, payload(temperature=temperature, top_p=top_p)) + registry.delivery(aid, message["id"], "sent") + registry.consume(aid, message["id"], message["content"], "session") + registry.close(aid) + turns, seal = registry.seal(aid) + assert seal["errors"] == [] + assert digest(turns[0]["sampling"]) == ident.sampling_digest + for key, value in (("temperature", temperature), ("top_p", top_p)): + assert type(turns[0]["sampling"][key]) is float + assert type(turns[0]["client_request"][key]) is type(value) + + def test_tool_roundtrip(): tools = [{"name": "bash", "input_schema": {"type": "object"}}] p = payload(tools=tools, messages=[ diff --git a/tests/harborrl/test_native_lifecycle.py b/tests/harborrl/test_native_lifecycle.py index 0c1ecd743..6813b1548 100644 --- a/tests/harborrl/test_native_lifecycle.py +++ b/tests/harborrl/test_native_lifecycle.py @@ -12,7 +12,7 @@ from harborrl.rollout.harbor_job.coordinator import GroupSlots, run_attempt from harborrl.trajectories.native import RewardProfile -PROFILE = {"cli_version": "2.1.0", "model": "policy", "max_turns": 4, +PROFILE = {"cli_version": "2.1.141", "model": "policy", "max_turns": 4, "agent_timeout_sec": 60, "setup_timeout_sec": 60, "verifier_timeout_sec": 60} @@ -39,11 +39,16 @@ def test_profile_controls_reach_agent_and_host(): disabled = claude_config({**PROFILE, "disallowed_tools": "Agent,Task,WebSearch,WebFetch"}, "http://gateway:8000", "token") assert disabled["kwargs"]["disallowed_tools"] == "Agent,Task,WebSearch,WebFetch" - assert config["env"]["DISABLE_AUTO_COMPACT"] == "1" + extended = claude_config({**PROFILE, "disallowed_tools": "WebSearch"}, + "http://gateway:8000", "token") + assert extended["kwargs"]["disallowed_tools"] == "Agent,Task,WebSearch" + for key in ("DISABLE_PROMPT_CACHING", "DISABLE_COMPACT", "DISABLE_AUTO_COMPACT"): + assert config["env"][key] == "1" + assert sanitized_runner_env({key: "0"})[key] == "1" assert config["env"]["ANTHROPIC_MODEL"] == "policy" assert config["env"]["NO_PROXY"] == "gateway" assert config["env"]["no_proxy"] == "gateway" - assert config["kwargs"]["version"] == "2.1.0" + assert config["kwargs"]["version"] == "2.1.141" assert sanitized_runner_env({"PATH": "/bin", "ANTHROPIC_API_KEY": "external", "HTTPS_PROXY": "external"})["PATH"] == "/bin" assert "ANTHROPIC_API_KEY" not in sanitized_runner_env({"ANTHROPIC_API_KEY": "external"}) with pytest.raises(ValueError): diff --git a/tests/harborrl/test_rollout_service.py b/tests/harborrl/test_rollout_service.py index f12994feb..f8d101459 100644 --- a/tests/harborrl/test_rollout_service.py +++ b/tests/harborrl/test_rollout_service.py @@ -108,6 +108,34 @@ def group_request(group_index=0, policy_version="1", slots=2, task=TASK): } +@pytest.mark.parametrize("probability", [1, 1.0], ids=["int", "float"]) +def test_service_executor_normalizes_sampling_lock(monkeypatch, probability): + from harborrl.rollout import service + + request = group_request() + request["sampling"].update(temperature=probability, top_p=probability) + + async def run_group(config, current, **kwargs): + assert kwargs["sampling_digest"] == digest(SAMPLING) + return [] + + monkeypatch.setattr(service, "run_group", run_group) + assert asyncio.run(service.default_executor(None, None, request)) == [] + assert type(request["sampling"]["temperature"]) is type(probability) + + +@pytest.mark.parametrize("probability", [1, 1.0], ids=["int", "float"]) +def test_slime_sampling_uses_the_same_numeric_lock(probability): + from harborrl.backends.slime_v032.rollout import sampling_params + + args = SimpleNamespace(rollout_max_response_len=SAMPLING["max_new_tokens"], + rollout_temperature=probability, rollout_top_p=probability) + params = sampling_params(args) + assert digest(params) == digest(SAMPLING) + assert type(params["temperature"]) is float + assert type(params["top_p"]) is float + + class ServiceHarness: def __init__(self, tmp_path, executor, **kwargs): self.tmp_path = tmp_path From 78734737a28839470ce798071872bfe150b31ff1 Mon Sep 17 00:00:00 2001 From: luyd Date: Thu, 29 Oct 2026 17:36:05 +0800 Subject: [PATCH 4/6] Allow rollout lifecycle guide in public tree audit --- scripts/audit_public_tree.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/audit_public_tree.sh b/scripts/audit_public_tree.sh index b9c044203..8f10531c0 100755 --- a/scripts/audit_public_tree.sh +++ b/scripts/audit_public_tree.sh @@ -19,7 +19,7 @@ public_documents = { "configs/harbor_hub/public_compatibility_summary.md", "configs/harbor_hub/public_compatibility_summary_zh.md", "docs/agents/environment-check.md", "docs/configuration-wizard.md", - "docs/deployment.md", "docs/sandbox-rollout.md", + "docs/deployment.md", "docs/rollout-runtime-lifecycle.md", "docs/sandbox-rollout.md", "docs/train-inference-consistency.md", "docs/web-console.md", "practices/h_cluster/README.md", "harborrl/webui/static/lucide-LICENSE.txt", } From b76acae4802811393aaf027658a7cb748c195d81 Mon Sep 17 00:00:00 2001 From: luyd Date: Thu, 29 Oct 2026 17:42:13 +0800 Subject: [PATCH 5/6] Fix lint formatting in rollout service and tests --- harborrl/rollout/service.py | 8 +++++++- tests/harborrl/test_backend_bootstrap.py | 4 ++-- tests/harborrl/test_slime_v032_exporter.py | 1 + 3 files changed, 10 insertions(+), 3 deletions(-) diff --git a/harborrl/rollout/service.py b/harborrl/rollout/service.py index c71cff61a..81d2ed273 100644 --- a/harborrl/rollout/service.py +++ b/harborrl/rollout/service.py @@ -29,7 +29,13 @@ from harborrl.rollout.exporters.native import export_group from harborrl.rollout.runtime import run_group -from harborrl.trajectories.native import canonical_sampling, digest, finite, publish, read_json +from harborrl.trajectories.native import ( + canonical_sampling, + digest, + finite, + publish, + read_json, +) SERVICE_SCHEMA = "harborrl-rollout-service-v1" GROUP_REQUEST_KEYS = {"group_id", "rollout_id", "group_index", "policy_version", diff --git a/tests/harborrl/test_backend_bootstrap.py b/tests/harborrl/test_backend_bootstrap.py index 48607d2bd..2e16413ae 100644 --- a/tests/harborrl/test_backend_bootstrap.py +++ b/tests/harborrl/test_backend_bootstrap.py @@ -40,8 +40,8 @@ def test_backend_env_exports_quoted_absolute_paths(tmp_path): child = subprocess.run( ["bash", "-c", '. "$1"\nexec "$2" -c "$3"', "bootstrap-test", str(env_file), sys.executable, - 'import json, os; print(json.dumps({k: os.environ.get(k) for k in ' - '("SLIME_DIR", "MEGATRON_DIR", "SGLANG_IMAGE")}))'], + ('import json, os; print(json.dumps({k: os.environ.get(k) for k in ' + '("SLIME_DIR", "MEGATRON_DIR", "SGLANG_IMAGE")}))')], cwd=bin_dir, env={key: value for key, value in env.items() if key != "SGLANG_IMAGE"}, check=True, capture_output=True, text=True, timeout=10) assert json.loads(child.stdout) == { diff --git a/tests/harborrl/test_slime_v032_exporter.py b/tests/harborrl/test_slime_v032_exporter.py index 6597a579d..5fb0e406f 100644 --- a/tests/harborrl/test_slime_v032_exporter.py +++ b/tests/harborrl/test_slime_v032_exporter.py @@ -357,6 +357,7 @@ def test_launcher_hooks_resolve_and_dry_run(): def test_rollout_hook_locks_policy_version_and_keeps_groups_atomic(tmp_path, monkeypatch, request): import asyncio + from harborrl.rollout.runtime import close_runtime close_runtime() From 4e83d02ad8010d463c7af4a4b9de4b55eb0f41d4 Mon Sep 17 00:00:00 2001 From: luyd Date: Thu, 29 Oct 2026 18:18:48 +0800 Subject: [PATCH 6/6] Fix cancellation assertion for Python 3.10 --- tests/harborrl/test_rollout_runtime_lifecycle.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/harborrl/test_rollout_runtime_lifecycle.py b/tests/harborrl/test_rollout_runtime_lifecycle.py index 6d5620058..29ec250ec 100644 --- a/tests/harborrl/test_rollout_runtime_lifecycle.py +++ b/tests/harborrl/test_rollout_runtime_lifecycle.py @@ -129,7 +129,9 @@ async def generate(args, group, params, *, rollout_id): monkeypatch.delenv("HARBORRL_ROLLOUT_SERVICE_URL", raising=False) monkeypatch.setattr(rollout, "generate_group", generate) - with pytest.raises(error_type, match="original group error"): + # Python 3.10's asyncio.gather can discard CancelledError messages. + match = "original group error" if error_type is ValueError else None + with pytest.raises(error_type, match=match): runtime.run_embedded(rollout.run_native_groups( SimpleNamespace(rollout_max_response_len=128), ["failed", "pending"], None, 0)) assert events == ["sibling cleaned"]