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..71ca79609 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: @@ -160,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 f75e744a9..347b1f1e4 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)。 + ### 训练 完成安装后，从两种配置方式中选择一种： @@ -150,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/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/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 44c36e2c8..c8b6ffd9d 100644 --- a/harborrl/backends/slime_v032/rollout.py +++ b/harborrl/backends/slime_v032/rollout.py @@ -16,26 +16,28 @@ from harborrl.export.native import export_training_batch from harborrl.rollout.runtime import ( RolloutRuntimeConfig, - close_runtime, current_policy_version, digest, + gather_or_cancel, + run_embedded, run_group, 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): @@ -91,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_) @@ -195,13 +197,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 - try: - results = await asyncio.gather(*( - generate_group(args, group, params, rollout_id=rollout_id) for group in groups - )) - except BaseException: - close_runtime() - 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 = [] @@ -309,7 +306,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/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/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/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/harbor_job/coordinator.py b/harborrl/rollout/harbor_job/coordinator.py index 17047cd40..7316d5bba 100644 --- a/harborrl/rollout/harbor_job/coordinator.py +++ b/harborrl/rollout/harbor_job/coordinator.py @@ -92,8 +92,17 @@ 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 = [] + + 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,7 +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(), '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() @@ -157,6 +166,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 +176,8 @@ 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(): + mark_cleanup("Inspect runner/cleanup-required.json") if registered: registry.close(identity.attempt_id) if process.returncode is None: @@ -181,12 +191,10 @@ 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 (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, - "reason": "Runner exited abnormally; inspect Harbor resources before retry"}) - if registered: + 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) 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..9288ebef8 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,47 @@ 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 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, @@ -337,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/harborrl/rollout/service.py b/harborrl/rollout/service.py index e223ab05a..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 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 +153,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/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", } 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..2e16413ae --- /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..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=[ @@ -93,13 +112,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 +147,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..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): @@ -124,10 +129,48 @@ 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" +@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 +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, runner_host=runner_host)) + marker = json.loads((root / "cleanup-required.json").read_text()) + assert marker["reason"] == "Inspect runner/cleanup-required.json" + 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() + + 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..29ec250ec --- /dev/null +++ b/tests/harborrl/test_rollout_runtime_lifecycle.py @@ -0,0 +1,216 @@ +"""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 + + +@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() + events = [] + + async def generate(args, group, params, *, rollout_id): + if group == "failed": + await asyncio.sleep(.01) + raise error_type("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) + # 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"] + 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_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 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..5fb0e406f 100644 --- a/tests/harborrl/test_slime_v032_exporter.py +++ b/tests/harborrl/test_slime_v032_exporter.py @@ -355,7 +355,13 @@ 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 +374,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 +418,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]