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. The same folder contains a Union-backend variant that additionally uses reusable containers, copy-on-write volumes, and run forking.

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 trials and steps from the record instead of re-executing them, and resumes from the last completed step’s checkpointed weights
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 training step is a pure, deterministic function of its inputs, so re-execution recomputes the identical update
New rubric, old rollouts	Rollout generation is cached with `cache="auto"`; a new run with a new rubric reuses every rollout whose inputs are unchanged and re-runs only judging and training

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_oss.py
                
# Separate environments so each tier's resources, retries, and timeouts are its own.
sandbox_env = flyte.TaskEnvironment(
    name="agentic-rl-oss-sandbox",
    image=image,
    resources=flyte.Resources(cpu="500m", memory="1Gi"),
)

judge_env = flyte.TaskEnvironment(
    name="agentic-rl-oss-judge", image=image, resources=flyte.Resources(cpu="500m", memory="512Mi")
)

trainer_env = flyte.TaskEnvironment(
    name="agentic-rl-oss-trainer", image=image, resources=flyte.Resources(cpu="500m", memory="512Mi")
)

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

The separation is the point: a trial pod that OOMs takes down one trial, not the step; the judge tier’s retry policy is independent of the sandbox tier’s; and the driver is a cheap CPU pod whose only job is orchestration — all the expensive work happens in child actions that outlive it.

Weights travel as inputs (pull-based sync)

agentic_rl_durable_oss.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

class Rubric(BaseModel):
    """Judge configuration is an INPUT, so changing it changes only judging + training."""

    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

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_oss.py
                
@sandbox_env.task(cache="auto")
async def build_world(world_id: str) -> Dir:
    """Populate the world's filesystem once; uploaded as a Dir every trial downloads.

    Cached, so every run (including a re-scoring run) reuses the same Dir — which is what makes
    downstream `generate` cache keys line up across runs.
    """
    secret_doc, secret = _world_secret(world_id)
    root = Path(tempfile.mkdtemp(prefix=f"world-{world_id}-")) / "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 Dir.from_local(root)

The world is built once (the task is cached) and uploaded as a flyte.io.Dir. Each trial downloads its own private copy:

agentic_rl_durable_oss.py
                
                    
                
            
                
            

                
                    
                
            
@sandbox_env.task(cache="auto", retries=1, timeout=flyte.Timeout(max_runtime=300))
async def generate(spec: TrialSpec, weights: Weights, world: Dir, flaky_rate: float = 0.0) -> Trajectory:
    """One trial: download a private copy of the world, run the agent loop, return what happened.

    Idempotent by construction — re-running produces the same trajectory (seeded by trial_id),
    so a retry is a clean regeneration. Cached, so a re-scoring run with identical inputs
    (same spec, same weights, same world) reuses the rollout instead of regenerating it.
    The pod's own filesystem is the sandbox: writes are private and vanish with the pod.
    """
    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

    # private copy of the world for this trial (downloaded into this pod)
    ws = Path(await world.download())
    docs = ws if (ws / "doc_0.txt").exists() else ws / "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),
    )

The pod’s own filesystem is the sandbox: the downloaded copy is private, writes never leak between trials, and everything vanishes with the pod. (On a Union backend, Volume.fork() replaces the download with a sub-second copy-on-write fork — worth knowing about when worlds get large.)

The task 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). It’s also cached with cache="auto" — that’s what powers rubric re-scoring later. 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_oss.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
    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 is a pure function: current weights and rewards in, new weights out.

agentic_rl_durable_oss.py
                
                    
                
            
                
            

                
                    
                
            
@trainer_env.task(retries=1, timeout=flyte.Timeout(max_runtime=600))
async def train_step(step: int, weights: Weights, rewards: list[Reward], lr: float = 2.0) -> StepResult:
    """One optimizer step as a pure function: (weights in, rewards in) -> new weights out.

    The returned weights ARE the checkpoint — every step's output is durably recorded, so a
    driver crash resumes from the last completed step. Determinism makes re-execution safe:
    a replayed step recomputes the identical update, so there is no double-apply to guard
    against (the stateful-actor version needs an explicit exactly-once check instead).
    """
    new = weights.model_copy(deep=True)
    for r in rewards:
        if r.found_doc is not None and r.reward > 0:
            prior = new.doc_prior[r.world_id]
            prior[r.found_doc] += lr * r.reward  # bandit update toward the rewarded doc
    new.version += 1
    return StepResult(
        step=step,
        weights=new,
        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),
    )

The returned weights are the checkpoint. Every step’s output is durably recorded as the action’s result, so a driver crash resumes from the last completed step with no checkpoint-restore code at all. And because the function is deterministic, re-execution is harmless: a replayed step recomputes the identical update, so there’s no double-apply to guard against.

This is the trade against keeping trainer state resident in a long-lived process (as the Union-backend variant does with a detached Ray actor on a reusable cluster): weights must be small enough to serialize every step — the LoRA/small-model shape — in exchange for a trainer with no state to lose.

The driver: a durable log of the training run

agentic_rl_durable_oss.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; initial weights are deterministic, so step-0 generate inputs
    # hash identically across runs (which is what lets a re-scoring run reuse them)
    worlds = dict(zip(world_ids, await asyncio.gather(*[build_world(w) for w in world_ids])))
    weights = Weights(version=0, doc_prior={w: [1.0] * N_DOCS for w in 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
            rewards = sorted((r for r in results if isinstance(r, Reward)), key=lambda r: r.trial_id)

            # one durable step; its output weights are the checkpoint
            result = await train_step(step, weights, rewards)
            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"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_oss.py
                
                    
                
            
                
            

                
                    
                
            
@driver_env.task(report=True)
async def rescore(
    n_steps: int = 3,
    n_worlds: int = 3,
    prompts_per_step: int = 3,
    n_samples: int = 2,
) -> list[StepResult]:
    """Re-run training with a new rubric. Cached `build_world` and `generate` actions are reused
    wherever their inputs are unchanged; judging and training re-run everywhere.

    Step 0 reuses every rollout (same worlds, same v0 weights). The new rewards change the
    weights after step 0, so later steps' `generate` inputs differ and regenerate — cache-key
    identity finds that invalidation frontier by itself; nothing here knows about "forks".
    """
    new = Rubric(correctness_weight=1.0, efficiency_weight=1.0, tidiness_weight=0.5, name="v2-efficiency")
    return await train(
        n_steps=n_steps, n_worlds=n_worlds, prompts_per_step=prompts_per_step, n_samples=n_samples, rubric=new
    )

Here the reuse comes from caching rather than run recovery: build_world and generate are cached with cache="auto", worlds and initial weights are deterministic, so a new run with a new rubric hits the cache for every step-0 rollout — identical spec, weights, and world — and reuses it. Judging re-runs (its rubric input changed), the new rewards change the weights after step 0, so later steps’ generate inputs differ, miss the cache, and regenerate. Same invalidation frontier, found by cache-key identity instead of fork recovery.

(A Union backend adds run forking — flyte.rerun(run_name, recover=True, rubric=new) — which reuses non-cached actions too and links the fork to its parent run in the UI.)

Run it

        
    
# clean run
flyte --config <your-config> run agentic_rl_durable_oss.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_oss.py train \
    --crash_driver_at_step 1 --flaky_trial_rate 0.25 --judge_flake_rate 0.3

# re-score with a new rubric (caching reuses worlds + step-0 rollouts)
flyte --config <your-config> run agentic_rl_durable_oss.py rescore

In the failure-injected run, watch three things in the UI:

Retried generate actions — each injected sandbox failure retries once and succeeds; the step completes with all trials.
Retried verify_and_judge actions — their attempt-0 logs show JUDGE CALLED; their retry logs don’t. The trace replayed.
The driver’s second attempt — completed trials and steps replay from the durable record; only work after the crash point actually executes.

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.
- Checkpoint-as-output scales down gracefully. Threading weights through step tasks costs a serialization per step but makes the trainer stateless — the strongest possible recovery story. When trainer state gets too big to serialize per step, keep it in the training framework’s own checkpoint format and pass a reference (flyte.io.File/Dir) between steps instead.
- Caching is the experiment multiplier. Cache generation on its true inputs and any re-run — new rubric, new judge, fixed bug downstream — reuses every rollout 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.