Fix rollout lifecycle, GPU colocation, and Claude Code integration

.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

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 <Path to HarborRL backends>
-. <Path to HarborRL backends>/harborrl-backend.env
+bash scripts/bootstrap_backends.sh "<Path to HarborRL backends>"
+. "<Path to HarborRL backends>/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
 

README_zh.md

@@ -108,10 +108,14 @@ uv run pytest
 如需原生训练，请把固定版本的 native 训练栈安装到可写位置：
 
 ```bash
-bash scripts/bootstrap_backends.sh <Path to HarborRL backends>
-. <Path to HarborRL backends>/harborrl-backend.env
+bash scripts/bootstrap_backends.sh "<Path to HarborRL backends>"
+. "<Path to HarborRL backends>/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
 

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.

examples/native/catalog.json

@@ -12,6 +12,6 @@
       "scale": 1.0
     },
     "revision": "v0.1-example",
-    "task_digest": "905f496d2852cca5b1e86e9de72395b41c291d2b8f6a80807e0ed24ca19541b3"
+    "task_digest": "2293cd4682597d9c6c5a4bb35d2ded9978873a8deb48ee07fae4e58121c79691"
   }
 ]

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"}

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
 }

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"

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

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(

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):

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):

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