Datagen Job Cleanup

datagen-job-cleanup

After a datagen (trace-generation) job terminates, follow these steps to get the traces onto HF and free disk. Unlike RL/SFT there is no model checkpoint — the artifact is the trace dataset.

0. Recognize the common case: TIMEOUT stranded the traces

Datagen jobs run to a wall-clock --time_limit; on TIMEOUT, Harbor's terminal trace-upload step is killed mid-run, leaving completed trials on disk but no upload. A clean COMPLETED exit usually uploads automatically; a TIMEOUT almost never does. Either way, do NOT assume the upload happened — verify (step 4).

sacct -j <jobid> -X --format=JobID,JobName%55,State,Elapsed,End -n

1. Locate the trace_jobs dir (ONE-level nesting for datagen)

Datagen writes to <run_dir>/trace_jobs/<inner_run_name>/<task>__<id>/ — a SINGLE level of nesting under trace_jobs/, NOT the double-nest <run>/<run>/trace_jobs of the RL path. The --job_dir you pass to the uploader must be the dir that DIRECTLY CONTAINS the <task>__<id> trial dirs, i.e. trace_jobs/<inner_run_name>:

# Direct-ssh launches land under experiments/; --experiments_dir launches under ot-baf.
RUN=/e/scratch/jureap59/feuer1/OpenThoughts-Agent/experiments/<job_name>   # or /e/data1/.../ot-baf/<job_name>
INNER=$(ls -d $RUN/trace_jobs/*/ 2>/dev/null | head -1); echo "$INNER"

Iris/CoreWeave durable S3 layout

--s3-output-dir is a campaign/root prefix, not a per-job destination. The shared Iris output resolver appends the Iris job name and routes Harbor to:

<s3-output-root>/<iris-job>/trace_jobs/<harbor-job>/<trial>/result.json

For example, launch with --s3-output-dir s3://marin-us-east-02a/iris/<campaign>, not with a URI ending in $JOB. Eval and datagen both use the IrisOutputPaths plan in hpc/iris/outputs.py; do not reproduce this path construction in a launcher.

Run S3 cleanup inside a cw-rno2a Iris worker so the transfer uses cluster credentials and bandwidth:

python scripts/harbor/cleanup_coreweave_datagen_s3.py \
  --target '<job>|s3://marin-us-east-02a/iris/<campaign>/<job>|penfever/<repo>'

The downloader lists the prefix once, downloads with bounded parallelism and adaptive S3 retries, applies the realness gate, uploads in one Hub commit, reloads the published train split, and removes only its worker-local temporary copy. It never deletes the durable S3 source. Repeat --target to process several completed jobs serially and idempotently.

Consolidate named relaunches without dropping retries

When a dataset row has several independently launched -rN attempts, give their durable trial_uri values as a comma-separated prefix list in one target. The cleanup tool creates a separate local run directory for each prefix, retains each run's config.json, then exports their trials into the one requested Hub repository. Repeated task IDs are intentional independent samples and are retained; only exact copied artifacts must not be counted twice. Do not point the tool at only the newest retry.

python scripts/harbor/cleanup_coreweave_datagen_s3.py \
  --target '<dataset>|s3://.../first-attempt,s3://.../retry-r2|penfever/<dataset>-traces'

--no_literal_tokens, --single_commit, and --skip_register are cleanup tool defaults; pass only --target arguments to this wrapper. Its reusable, git-tracked implementation is scripts/harbor/cleanup_coreweave_datagen_s3.py.

Jobs launched before the shared-path fix may have the malformed rescue layout <root>/<iris-job>/<harbor-job>/<trial> with no trace_jobs component. The cleanup tool recognizes this existing layout from result.json parents, but new launchers must only write the canonical layout above.

2. Sanity-check the trials are REAL before uploading

A served /v1/models healthcheck does NOT mean generation worked. Compute avg turn count + exception rate: if avg turns ≈ 1.0, the run is near-total failure (e.g. endpoint healthy + many result.json, but every trial a 1-turn InternalServerError from a dead EngineCore). A real run has multi-step trajectories (turns > 1) + a tolerable exception rate (tezos datagen ~20-25% AgentTimeout is normal).

# trajectory.json is a dict with a "steps" list; turns ≈ len(steps).
$OTAGENT_PY - <<'PY'
import json, glob, os, statistics
inner = os.environ["INNER"]
dirs = glob.glob(os.path.join(inner, "*__*/"))
turns, exc = [], 0
for d in dirs:
    tj = os.path.join(d, "agent", "trajectory.json")
    if os.path.exists(tj):
        t = json.load(open(tj))
        turns.append(len(t.get("steps", [])) if isinstance(t, dict) else len(t))
    r = os.path.join(d, "result.json")
    if os.path.exists(r):
        if (json.load(open(r)).get("exception_info") or {}).get("exception_type"): exc += 1
print(f"trials={len(dirs)} avg_turns={statistics.mean(turns):.2f} exceptions={exc}" if turns else "no trajectories")
PY

If avg_turns ≈ 1.0 → the run failed; do NOT upload. Diagnose instead (read a trial's exception.txt + the _vllm.log for the engine-side error) and write an agent_log; the traces are not worth keeping.

3. Upload the traces to HF penfever org

From the otagent conda env — the uploader needs google.cloud.storage + matplotlib, which envs/rl lacks:

# otagent env, source secrets first. $INNER = the dir DIRECTLY containing the <task>__<id> trial dirs.
python -m scripts.harbor.make_and_upload_trace_dataset \
  --job_dir "$INNER" \
  --repo_id penfever/<descriptive-name> \
  --episodes last \
  --served_model <the exact served model ref> \   # required for decodable literals
  --include_literal_tokens \                       # REQUIRE literals (fail loud if 0 bind)
  --single_commit                                  # ONE HF commit — avoids the 128-commit/hr 429 on big sets

Default to the penfever/ org + --episodes last. For a chunked launch, upload each chunk's trace_jobs to its own _chunk{i} repo. Public by default (feedback_hf_public_default).

Literals are AUTO-INCLUDED when present (no flag needed). The uploader favors the durable literal.jsonl: it auto-discovers the sibling <experiments_dir>/logs/*_literal.jsonl (searching --job_dir and a few parents) and correlates the RecordProxy records into the trajectory step metrics so the exported dataset carries the trainable prompt_token_ids / completion_token_ids / logprobs columns. Requirements + knobs:

• --job_dir must sit inside the experiments-dir tree so the parent-walk reaches …/logs/. $INNER qualifies as long as the local run dir still has its sibling logs/ (local runs do).
• ⚠ Resolve the recorded output bucket — NEVER hardcode gs://marin-models-{us,eu}. Jobs pin to a co-located single-region bucket (gs://marin-us-east5, …); older jobs are on the multi-region mirror. Resolve each job's ACTUAL recorded OUTER prefix (registry-first, iris-fallback):

  JOB=<job_name>
  OUT=$(/Users/benjaminfeuer/miniconda3/envs/otagent/bin/python -m hpc.iris.job_output_resolver "$JOB" \
        --cluster /Users/benjaminfeuer/Documents/marin/lib/iris/config/marin.yaml)   # e.g. gs://marin-us-east5/ot-agent/<job>
  mkdir -p /tmp/${JOB}_traces
  gsutil -m rsync -r "$OUT/" /tmp/${JOB}_traces/     # OUTER <job>/ — carries logs/*_literal.jsonl

  The resolver returns the correct bucket for both single- and multi-region jobs (--cluster is only consulted for the iris fallback when the job isn't in the local registry).
• ⚠ gs:// iris rescue — the INNER <job>/ is the --job_dir, NOT the outer rescue root. You rsync the OUTER $OUT/ → /tmp/<job>_traces so logs/*_literal.jsonl rides along, but the trial dirs sit one level down at /tmp/<job>_traces/<job>/<trial>, so pass --job_dir **/tmp/<job>_traces/<job>** (the inner subdir that DIRECTLY contains the <task>__<id> dirs). Passing the outer /tmp/<job>_traces still uploads TEXT rows, but the extra <job>/ prefix makes 0 trials bind → silent text-only dataset (with --include_literal_tokens it fails loud instead). Also: gsutil rsync needs the local dest to EXIST — mkdir -p before the rsync.
• ⚠ Big datasets → add --single_commit (HF 128-commits/hour cap). ~1 commit per shard; a multi-thousand-row set + any re-attempt blows past HF's 128 repo-commits/hour limit → 429 mid-push → a partial dataset. --single_commit pushes the whole dataset in ONE commit → never trips the limit. (If you already 429'd, the repo may be partial/empty; wait ~1h for the window to reset, then re-run with --single_commit.)
• Correlator is RAM-heavy — one upload at a time on a RAM-limited host. Correlation holds ALL parsed records (token ids) in memory for chain-reconstruction; a big literal file (e.g. 1.5 GB) can flood a 36 GB Mac. Run ONE upload at a time, never two concurrent. If one-at-a-time OOMs, offload to a bigger box or drop messages from records after reconstruct_chains.
• A job with no literal.jsonl exports text-only, byte-identical to before (parity).
• --no_literal_tokens forces text-only even when a literal.jsonl is present.
• --include_literal_tokens means REQUIRE: fails loud if literals are expected but none are found.
• The uploader FAILS LOUD if a literal.jsonl is present but 0 trials bind (a regression, not a valid dataset).
• Pass --served_model on any literal upload. Token-id columns are only decodable with the EXACT tokenizer the engine served (a same-family tokenizer decodes word tokens to garbage); the uploader stamps the model ref into tokenizer_provenance.json + the dataset-card README. For the opencode-131k campaign: --served_model Qwen/Qwen3.5-122B-A10B-FP8 (or gs://marin-models-us/ot-agent/models/Qwen/Qwen3.5-122B-A10B-FP8/ — storage root via marin_prefix(), see .agents/ops/iris/ops.md §rendezvous, don't hardcode the region bucket).
• Schema-pin (OT-Agent 7c978b78): the exporter pins the literal token columns to an explicit nested type per shard. Datasets uploaded BEFORE 7c978b78 are degraded (under-populated literal yield, shards with no token columns, load_dataset CastError) — re-rescue them to recover full yield. Full reference (decoding, tokenizer provenance, re-rescue, literals→SFT): .agents/projects/harbor/ops.md (§ Literal-token trace datasets).