Durable agentic RL with SkyRL and Harbor

Reinforcement learning for LLM agents has a different shape than classic RLHF. In frameworks like SkyRL and agentic environment suites like Harbor, each training sample is a trial: the policy is dropped into a sandboxed environment with a filesystem and tools, takes many turns to attempt a task, and is then scored by a deterministic verifier plus an LLM judge. A training step fans out dozens or hundreds of these trials, collects the rewards, and updates the policy.

This inverts where the wall-clock time and the money go. SkyRL’s own benchmarks show that rollout scheduling, not gradient computation, is the lever on agentic workloads: trials are long (minutes of tool calls), heterogeneous (some finish in 3 turns, some in 30), and flaky (sandboxes die, judge APIs rate-limit and time out). And the failure economics are brutal:

A driver crash at step 40 of 50 loses every completed rollout in flight — hours of GPU inference regenerated from scratch.
A flaky sandbox that kills one trial shouldn’t kill the step, but in a monolithic training loop it often does.
A judge retry that re-calls the LLM re-bills you for tokens you already paid for.
Wanting to re-score old rollouts with a new rubric — the cheapest experiment in agentic RL — usually means regenerating everything, because trajectories and scores are entangled in the trainer’s process.

SkyRL itself has no fault-tolerance or Kubernetes story; that’s the layer this tutorial builds. We take a Harbor-shaped agentic RL loop — per-trial sandboxes, verifier + LLM judge, a bandit trainer standing in for the GPU trainer — and structure it so that every trial, judgment, and training step is a durable, independently retryable unit. The training here is a toy multi-armed bandit so the whole thing runs on CPU, but every structural decision is the real one.

Full code available on GitHub, alongside the design document and measured experiments behind it.

Overview

The loop is the standard agentic RL loop:

Generate: for each trial, the policy explores a private copy of a “world” (a filesystem of documents, one of which contains a secret) using its current weights.
Verify and judge: a deterministic verifier checks correctness; a (simulated) LLM judge scores the trajectory against a rubric — correctness, efficiency, tidiness.
Train: rewards update the policy, and the next step’s trials use the new weights.

What makes it durable is how the loop maps onto Flyte’s execution model:

Failure mode	What handles it
Driver crash mid-run	Every completed action is durably recorded; the retried driver replays completed steps in ~0.07 s each and resumes where it left off — on the same warm Ray cluster, because the cluster’s lifetime is scoped to the run, not the driver attempt
Flaky sandbox kills a trial	Each trial is its own task with `retries=1` and a timeout; a trial that fails after retries is skipped, never fatal to the step
Judge task fails after the LLM call	The judge call is wrapped in `@flyte.trace`, so the retry replays the recorded result instead of re-billing the API
Training step re-executes	The trainer actor tracks applied steps, so an update is applied exactly once
New rubric, old rollouts	`flyte.rerun(recover=True, rubric=new)` forks the finished run: every rollout whose inputs are unchanged is recovered; only judging and training re-run

Three of these lean on Union-backend capabilities — reusable containers for the warm Ray cluster, volumes for instant per-trial world copies, and run forking for rubric re-scoring. The rest is core Flyte 2.

The result: in a run with a driver crash injected after step 1, 25% of sandboxes failing, and 30% of judge tasks failing after the judge call, all 18 trials completed, the judge was called exactly once per trial, and the bandit still learned (mean turns to find the secret dropped from 3.83 to 1.5, mean reward rose from 1.62 to 2.36).

Implementation

Environments: one failure domain per tier

Each tier of the loop — sandboxed rollouts, judging, training, and the driver — gets its own flyte.TaskEnvironment, so its resources, retries, and timeouts are independent.

agentic_rl_durable.py
                
                    
                
            
                
            

                
                    
                
            
# [L1+] one long-lived Ray cluster for the whole run; each step is a task on it.
trainer_env = flyte.TaskEnvironment(
    name="agentic-rl-trainer",
    plugin_config=ray_config,
    image=image,
    resources=flyte.Resources(cpu=(1, 2), memory=("2000Mi", "4000Mi")),
    reusable=flyte.ReusePolicy(replicas=1, idle_ttl=600, scope="run"),
)

# [S2] pod-as-sandbox: FUSE for the forked world volume; one fresh pod per trial.
sandbox_env = flyte.TaskEnvironment(
    name="agentic-rl-sandbox",
    image=image,
    pod_template=flyte.PodTemplate().allow_fuse(),
    resources=flyte.Resources(cpu="500m", memory="1Gi"),
)

# [J] judging is cheap CPU; separate env so its retries/timeouts are its own.
judge_env = flyte.TaskEnvironment(
    name="agentic-rl-judge", image=image, resources=flyte.Resources(cpu="500m", memory="512Mi")
)

driver_env = flyte.TaskEnvironment(
    name="agentic-rl-driver",
    image=image,
    resources=flyte.Resources(cpu=1, memory="1Gi"),
    depends_on=[trainer_env, sandbox_env, judge_env],
)

Two things to notice:

The trainer runs on one long-lived Ray cluster. reusable=flyte.ReusePolicy(replicas=1, idle_ttl=600, scope="run") keeps the same Ray head alive for the whole run. Every training step is a separate Flyte task submitted to that same cluster — so the step is a durable boundary, but the expensive state (in real training: FSDP shards and optimizer state; here: the bandit’s priors) never leaves GPU/memory between steps. Critically, the cluster’s lifetime is scoped to the run, not the driver attempt: a driver crash does not tear it down.
The sandbox environment allows FUSE (flyte.PodTemplate().allow_fuse()) so each trial pod can mount a forked volume — more on that below.

Weights travel as inputs (pull-based sync)

agentic_rl_durable.py
                
                    
                
            
                
            

                
                    
                
            
class Weights(BaseModel):
    """[L2b] The whole 'policy' — small enough to travel as an input to every rollout."""

    version: int = 0
    # per world: prior over which document holds the secret (the bandit's arm weights)
    doc_prior: dict[str, list[float]] = {}

class TrialSpec(BaseModel):
    trial_id: str
    step: int
    world_id: str
    prompt: str
    max_turns: int = 6

class Trajectory(BaseModel):
    trial_id: str
    world_id: str
    weights_version: int
    turns: list[dict]  # [{doc, found}]
    answer: Optional[str]
    files_written: list[str]
    elapsed_s: float
    fork_name: str

class Rubric(BaseModel):
    """[F] Judge configuration is an INPUT, so changing it changes only the judging actions."""

    correctness_weight: float = 1.0
    efficiency_weight: float = 0.3
    tidiness_weight: float = 0.1
    name: str = "v1"

class Reward(BaseModel):
    trial_id: str
    world_id: str
    verified_correct: bool
    n_turns: int
    judge_score: float
    reward: float
    found_doc: Optional[int]

class StepResult(BaseModel):
    step: int
    weights: Weights
    mean_reward: float
    mean_turns: float
    n_trials: int
    actor_pid: int

The Weights model is the policy, and it flows from the trainer to every rollout as an ordinary task input. This is the pull-based weight-sync shape that fits LoRA adapters and small models: the rollout tier needs no NCCL group, no engine control plane, and no connection back to the trainer — which is exactly what lets a rollout be retried on a fresh pod minutes later with no coordination.

Just as important: Rubric is an input to judging, not a constant baked into the code. Every input is part of an action’s content-hashed identity, and that identity is what makes selective re-execution possible later.

Worlds: built once, private copy per trial

Each world is a small filesystem of documents; one document contains a (hashed) secret the agent must find. Agent trials mutate their environment, so every trial needs its own copy — stale state from a previous attempt is how you get silently corrupted training data.

agentic_rl_durable.py
                
@sandbox_env.task(cache="auto")
async def build_world(world_id: str) -> ROVolume:
    """Populate the world's filesystem once; sealed as a read-only volume every trial forks."""
    secret_doc, secret = _world_secret(world_id)
    vol = Volume.new(name=f"world-{world_id}-{flyte.ctx().action.run_name}")
    await vol.mount()
    root = Path(vol.mount_path) / "docs"
    root.mkdir(parents=True)
    for i in range(N_DOCS):
        lines = [f"title: report {i} for {world_id}", f"owner: team-{i % 3}", f"pages: {10 + i}"]
        if i == secret_doc:
            lines.append(f"secret_sha256: {hashlib.sha256(secret.encode()).hexdigest()}")
        (root / f"doc_{i}.txt").write_text("\n".join(lines) + "\n")
    return await vol.finalize(message=f"world {world_id}")

The world is built once per run (the task is cached) and sealed as a read-only volume. Each trial then forks it:

agentic_rl_durable.py
                
                    
                
            
                
            

                
                    
                
            
@sandbox_env.task(retries=1, timeout=flyte.Timeout(max_runtime=300))
async def generate(spec: TrialSpec, weights: Weights, world: ROVolume, flaky_rate: float = 0.0) -> Trajectory:
    """One trial: fork the world, run the agent loop against it, return what happened.

    Idempotent by construction — re-running produces the same trajectory (seeded by trial_id),
    so a retry is a clean regeneration, not a replay of stale sandbox state.
    """
    t0 = time.monotonic()
    attempt = int(os.environ.get("FLYTE_ATTEMPT_NUMBER", "0"))
    rng = random.Random(_seed("trial", spec.trial_id))
    if attempt == 0 and rng.random() < flaky_rate:
        raise RuntimeError(f"simulated sandbox failure for {spec.trial_id} (attempt 0)")  # -> retried

    # [S2] instant private copy of the world for this trial
    fork_name = f"{world.name}-{spec.trial_id}"
    ws = await world.fork(name=fork_name)
    await ws.mount()
    docs = Path(ws.mount_path) / "docs"

    # the "agent": pick documents to read in an order driven by the policy's prior for this world
    prior = list(weights.doc_prior.get(spec.world_id) or [1.0] * N_DOCS)
    order: list[int] = []
    remaining = list(range(N_DOCS))
    while remaining:
        w = [prior[i] for i in remaining]
        pick = rng.choices(remaining, weights=w, k=1)[0]
        order.append(pick)
        remaining.remove(pick)

    turns, answer = [], None
    for doc in order[: spec.max_turns]:
        text = (docs / f"doc_{doc}.txt").read_text()
        found = "secret_sha256:" in text
        turns.append({"doc": doc, "found": found})
        await asyncio.sleep(0.2)  # a tool call
        if found:
            answer = text.split("secret_sha256:")[1].strip()
            break

    written = []
    (docs.parent / "answer.txt").write_text(answer or "")
    written.append("answer.txt")
    if rng.random() < 0.3:  # untidy agents leave scratch files behind (the judge cares)
        (docs.parent / "scratch.tmp").write_text("notes")
        written.append("scratch.tmp")

    return Trajectory(
        trial_id=spec.trial_id,
        world_id=spec.world_id,
        weights_version=weights.version,
        turns=turns,
        answer=answer,
        files_written=written,
        elapsed_s=round(time.monotonic() - t0, 2),
        fork_name=fork_name,
    )

world.fork(...) is a copy-on-write fork of the volume — about 0.1 s to fork plus 0.55 s to mount for a 1 GB world, versus 6+ s to download the same world from blob storage, and the gap widens with world size. The trial writes freely into its private copy; reset is free because the parent volume never changes. The pod is the sandbox, the forked volume is the world.

The task itself carries the trial-level durability contract: retries=1, a 300 s timeout, and idempotency by construction (the trajectory is seeded by trial_id, so a retry is a clean regeneration, not a replay of half-mutated sandbox state). The flaky_rate parameter injects sandbox failures on attempt 0 so you can watch the retries work.

The judge is memoized: retried, never re-billed

Scoring is two things fused into one task: a deterministic verifier (in Harbor terms, the environment’s test.sh) and an LLM judge that scores the trajectory against the rubric.

agentic_rl_durable.py
                
                    
                
            
                
            

                
                    
                
            
@flyte.trace
async def call_judge(traj: Trajectory, rubric: Rubric) -> float:
    """Stand-in for the vendor LLM judge. Traced: replayed, not re-called, on a task retry."""
    print(f"JUDGE CALLED trial={traj.trial_id} rubric={rubric.name}", flush=True)  # count these in logs
    await asyncio.sleep(0.5)  # network
    rng = random.Random(_seed("judge", traj.trial_id, rubric.name))
    correct = 1.0 if traj.answer == _world_secret_digest(traj.world_id) else 0.0
    efficiency = max(0.0, 1.0 - (len(traj.turns) - 1) / N_DOCS)
    tidy = 1.0 if traj.files_written == ["answer.txt"] else 0.0
    score = (
        rubric.correctness_weight * correct + rubric.efficiency_weight * efficiency + rubric.tidiness_weight * tidy
    ) + rng.uniform(-0.05, 0.05)
    return round(score, 3)

@judge_env.task(retries=2)
async def verify_and_judge(traj: Trajectory, rubric: Rubric, judge_flake_rate: float = 0.0) -> Reward:
    verified = traj.answer == _world_secret_digest(traj.world_id)  # the deterministic verifier (tests/test.sh in Harbor)
    score = await call_judge(traj, rubric)  # [J] memoized across retries

    attempt = int(os.environ.get("FLYTE_ATTEMPT_NUMBER", "0"))
    if attempt == 0 and random.Random(_seed("flake", traj.trial_id)).random() < judge_flake_rate:
        raise RuntimeError("simulated post-judge failure (e.g. upload) — retry must NOT re-call the judge")

    found_doc = next((t["doc"] for t in traj.turns if t["found"]), None)
    return Reward(
        trial_id=traj.trial_id,
        world_id=traj.world_id,
        verified_correct=verified,
        n_turns=len(traj.turns),
        judge_score=score,
        reward=round((1.0 if verified else 0.0) + score, 3),
        found_doc=found_doc,
    )

The @flyte.trace decorator is the important line. The judge call’s result is durably recorded the moment it returns. If the surrounding task then fails — the injected judge_flake_rate failure simulates a post-judge upload error — the retry replays the recorded result instead of calling the judge again. In the failure-injected runs, retried judge tasks show JUDGE CALLED exactly once in their attempt-0 logs and zero times on the retry.

At scale this is real money: a judge pass over hundreds of long trajectories is a substantial LLM bill, and infrastructure flakiness shouldn’t multiply it. Keep your vendor judge; just never re-bill it.

Training steps that can’t double-apply

The trainer’s state lives in a detached, namespaced Ray actor on the reusable cluster — the stand-in for GPU-resident FSDP shards and optimizer state. Each training step is a Flyte task that re-attaches to the actor by name:

agentic_rl_durable.py
                
                    
                
            
                
            

                
                    
                
            
def _policy_actor():
    import ray

    @ray.remote
    class PolicyTrainer:
        """Holds the trainable state (think: FSDP shards + optimizer) across steps."""

        def __init__(self, world_ids: list[str]):
            self.weights = Weights(version=0, doc_prior={w: [1.0] * N_DOCS for w in world_ids})
            self.pid = os.getpid()
            self.applied: dict[int, dict] = {}  # step -> weights after that step (exactly-once)

        def update(self, step: int, rewards: list[dict], lr: float) -> dict:
            # A re-executed step task (retry after the actor already applied it, or a driver
            # replay with a differently-ordered batch) must not apply the update twice.
            if step in self.applied:
                return self.applied[step]
            for r in rewards:
                if r["found_doc"] is not None and r["reward"] > 0:
                    prior = self.weights.doc_prior[r["world_id"]]
                    prior[r["found_doc"]] += lr * r["reward"]  # bandit update toward the rewarded doc
            self.weights.version += 1
            self.applied[step] = self.weights.model_dump()
            return self.applied[step]

        def info(self) -> dict:
            return {"pid": self.pid, "version": self.weights.version}

    return PolicyTrainer

def _attach_policy(world_ids: list[str]):
    import ray

    try:
        return ray.get_actor(POLICY_ACTOR, namespace=RAY_NAMESPACE)
    except ValueError:
        return (
            _policy_actor()
            .options(name=POLICY_ACTOR, namespace=RAY_NAMESPACE, lifetime="detached", get_if_exists=True)
            .remote(world_ids)
        )

@trainer_env.task
async def setup_trainer(world_ids: list[str]) -> Weights:
    import ray

    actor = _attach_policy(world_ids)
    info = ray.get(actor.info.remote())
    print(f"trainer actor pid={info['pid']} version={info['version']}", flush=True)
    return Weights(version=info["version"], doc_prior={w: [1.0] * N_DOCS for w in world_ids})

@trainer_env.task(retries=1, timeout=flyte.Timeout(max_runtime=600))
async def train_step(step: int, rewards: list[Reward], world_ids: list[str], lr: float = 2.0) -> StepResult:
    """One optimizer step. Re-attaches to the warm actor; returns the new (small) weights."""
    import ray

    actor = _attach_policy(world_ids)
    new_weights = ray.get(actor.update.remote(step, [r.model_dump() for r in rewards], lr))
    info = ray.get(actor.info.remote())
    return StepResult(
        step=step,
        weights=Weights(**new_weights),
        mean_reward=round(sum(r.reward for r in rewards) / max(1, len(rewards)), 3),
        mean_turns=round(sum(r.n_turns for r in rewards) / max(1, len(rewards)), 2),
        n_trials=len(rewards),
        actor_pid=info["pid"],
    )

Two details make this crash-safe:

The actor is detached and namespaced. Each Flyte task on the reusable cluster is a new Ray job in a new process; module globals don’t survive between steps. A detached, namespaced actor does — across steps, across step retries, and across driver crashes. In the failure-injected run, all three training steps report the same actor PID even though the driver died between steps 1 and 2.
Updates are exactly-once by construction. The actor records which step indices it has applied (self.applied). A re-executed step task — a retry after the actor already applied the update, or a driver replay — gets the recorded result back instead of applying the gradient twice.

The driver: a durable log of the training run

agentic_rl_durable.py
                
                    
                
            
                
            

                
                    
                
            
@driver_env.task(report=True)
async def train(
    n_steps: int = 3,
    n_worlds: int = 3,
    prompts_per_step: int = 3,
    n_samples: int = 2,
    rubric: Rubric = Rubric(),
    crash_driver_at_step: int = -1,
    flaky_trial_rate: float = 0.0,
    judge_flake_rate: float = 0.0,
) -> list[StepResult]:
    attempt = int(os.environ.get("FLYTE_ATTEMPT_NUMBER", "0"))
    started = flyte.durable.now()  # recorded once; replayed on retry
    world_ids = [f"w{i}" for i in range(n_worlds)]

    # worlds are cached; the trainer actor is created once per run and re-attached afterwards
    worlds = dict(zip(world_ids, await asyncio.gather(*[build_world(w) for w in world_ids])))
    weights = await setup_trainer(world_ids)

    history: list[StepResult] = []
    tab = flyte.report.get_tab("training")
    for step in range(n_steps):
        with flyte.group(f"step-{step}"):
            # [L3] fan out trials; [L2b] every trial carries the current weights as an input
            specs = [
                TrialSpec(
                    trial_id=f"s{step}-p{p}-n{s}", step=step, world_id=world_ids[p % n_worlds], prompt="find the secret"
                )
                for p in range(prompts_per_step)
                for s in range(n_samples)
            ]
            gens = [asyncio.create_task(generate(sp, weights, worlds[sp.world_id], flaky_trial_rate)) for sp in specs]

            # pipeline: judge each trajectory the moment it lands; a failed trial is skipped, not fatal
            judges = []
            for fut in asyncio.as_completed(gens):
                try:
                    traj = await fut
                except Exception as e:  # noqa: BLE001 — skip_failed_rollouts
                    print(f"trial failed after retries, skipping: {e}", flush=True)
                    continue
                judges.append(asyncio.create_task(verify_and_judge(traj, rubric, judge_flake_rate)))
            results = await asyncio.gather(*judges, return_exceptions=True)
            # canonical order: as_completed order differs between attempts, and the step's inputs
            # must hash identically for the replayed step to be reused (lesson from run upq5qwpv)
            rewards = sorted((r for r in results if isinstance(r, Reward)), key=lambda r: r.trial_id)

            # [L1+] one durable step on the warm cluster
            result = await train_step(step, rewards, world_ids)
            weights = result.weights
            history.append(result)

            tab.log(
                f"<p><b>step {step}</b> · trials {result.n_trials}/{len(specs)} · mean reward "
                f"{result.mean_reward} · mean turns {result.mean_turns} · weights v{weights.version} · "
                f"actor pid {result.actor_pid} · driver attempt {attempt}</p>"
            )
            await flyte.report.flush.aio()

            if attempt == 0 and step == crash_driver_at_step:
                raise flyte.errors.RuntimeSystemError("simulated", f"driver crash after step {step} on attempt 0")

    tab.log(f"<p>started {started.isoformat()} · finished on attempt {attempt}</p>")
    await flyte.report.flush.aio()
    return history

The driver reads as a plain Python training loop, but every call in it is a durable action. That has three consequences worth spelling out:

Pipelined, not batched. Trials are launched concurrently and judged the moment each one lands (asyncio.as_completed), so a 30-turn straggler doesn’t hold up scoring for the trials that finished in 3. A trial that fails after its retries is logged and skipped — skip_failed_rollouts, as SkyRL calls it — never fatal to the step.
A driver crash is a replay, not a restart. When the driver task is retried (the example injects a RuntimeSystemError after a chosen step to demonstrate), completed child actions are not re-executed: the retry replays them from the durable record and picks up at the first incomplete action. Notice there is no checkpointing code, no state file, no resume flag anywhere in the loop.
Canonicalize step inputs. The one subtle line is rewards = sorted(..., key=lambda r: r.trial_id). The first version of this example didn’t sort, and a driver replay collected the same rewards in a different as_completed order — so the training step’s inputs hashed differently, the step re-ran instead of replaying, and (in the stateful-trainer version) the update applied twice. If a replayed step’s inputs must hash identically, make them order-independent. This bug cost a debugging session; the fix is one sorted().

flyte.durable.now() records the wall-clock start once and replays it on retry — so even the report’s timestamps are stable across driver attempts. Progress streams to a live report tab in the UI as the run executes.

Re-scoring old rollouts with a new rubric

Rubric iteration is the cheapest experiment in agentic RL — if you can re-judge existing trajectories without regenerating them. Because the rubric is an input to judging (not to generation), changing it should invalidate only judging and everything downstream.

agentic_rl_durable.py
                
def fork_with_new_rubric(run_name: str) -> None:
    """Every `generate` is reused; `verify_and_judge` (new rubric input) and `train_step` re-run."""
    new = Rubric(correctness_weight=1.0, efficiency_weight=1.0, tidiness_weight=0.5, name="v2-efficiency")
    r = flyte.rerun(run_name, recover=True, rubric=new)
    print("forked run:", r.url)

flyte.rerun(run_name, recover=True, rubric=new) forks the finished run with one changed input. Recovery walks the action graph by content-hashed identity: worlds, trainer setup, and every step-0 generate action are recovered (reused as-is); verify_and_judge re-runs everywhere because its rubric input changed; and — correctly — steps 1 and 2 regenerate too, because the new rewards changed the policy after step 0, so later rollouts depend on different weights. Nothing in the driver knows about forks; content-hashed identity finds that invalidation frontier by itself.

In the measured fork, step-0 rollouts were re-scored from 1.62 to 2.41 mean reward on identical trajectories — the new rubric valued efficiency more, and the judging cost was the only cost paid for them.

Run it

The example needs a Union backend with the FUSE device plugin (for volumes) and KubeRay (for the Ray environment):

        
    
# clean run
flyte --config <your-config> run agentic_rl_durable.py train

# same run with every failure path exercised: driver crash after step 1,
# 25% flaky sandboxes, 30% judge tasks that fail after the judge call
flyte --config <your-config> run agentic_rl_durable.py train \
    --crash_driver_at_step 1 --flaky_trial_rate 0.25 --judge_flake_rate 0.3

# fork a finished run with a new rubric
FLYTE_CONFIG=<your-config> python agentic_rl_durable.py fork <run-name>

In the fully failure-injected run (3 steps × 3 worlds × 2 samples):

	count	note
`generate` succeeded / retried	18 / 6	6 injected sandbox failures, each retried once, none fatal
`verify_and_judge` succeeded / retried	18 / 6	retried actions log `JUDGE CALLED` once on attempt 0 and never on the retry
`train_step`	3	all on the same actor PID across the driver crash
driver	2 attempts	crashed after step 1; attempt 2 replayed steps 0–1 (~0.07 s each) and ran step 2
learning	✓	mean turns 3.83 → 1.33 → 1.50; mean reward 1.62 → 2.35 → 2.36

Despite a driver crash, a third of sandboxes flaking, and a third of judge tasks failing, the run completed with zero lost rollouts and zero re-billed judge calls.

What this means for real training

The bandit stands in for a GPU trainer, but the structure transfers directly to a SkyRL/Harbor-scale stack:

The trial is the unit of durability. Generation and judging are separate actions per trial, so the blast radius of any single failure — sandbox, network, judge API — is one trial, and the recovery cost is one retry.
Weights-as-inputs is the LoRA/small-model sync path. For full-parameter training with NCCL-based weight broadcast, the inner loop stays inside the training framework; the durable boundary moves up to the step level.
- Steps-as-tasks on a reusable cluster gives you a durable training log without paying cluster spin-up per step (~18 s of Ray job submission overhead per step is the price; a fresh cluster per step would be minutes). The expensive in-memory state lives in the cluster; the record of training lives in Flyte.
- Fork is the experiment multiplier. Any finished (or crashed) run can be forked with changed inputs — a new rubric, a new judge model, a fixed bug — and recovery reuses everything the change doesn’t invalidate.

The pattern generalizes beyond RL: any loop of expensive, flaky, independently-scoreable work items — evaluation harnesses, synthetic data generation, agent benchmarking — gets the same durability for the same restructuring.