Buckets:
| """Write the cleaned trajectories as Codex rollout files, one file per trajectory row. | |
| Reads : trajectories.train.jsonl, trajectories.eval.jsonl (cleaned output of build.py; raw logs are NOT read) | |
| Writes : traces/train/<id>.jsonl, traces/test/<id>.jsonl | |
| Why : the Hugging Face Hub decodes raw Codex rollouts natively (harness "codex", tool calls, results) and lists them | |
| as agent traces. These files use the same record types as the original logs, so the Hub reads them the same way: | |
| session_meta id, cwd, originator, model_provider, base_instructions | |
| event_msg/task_started, turn_context turn id, model, effort, cwd | |
| response_item/message (user | assistant) input_text / output_text content parts | |
| response_item/function_call name, arguments (JSON string), call_id | |
| response_item/function_call_output call_id, output | |
| event_msg/user_message, event_msg/task_complete | |
| What is NOT here: reasoning, token counts, rate limits, injected developer/permission messages and per-line timing. | |
| They are absent from the cleaned data, and nothing is invented to fill them: every line carries the turn's start | |
| timestamp. These are reconstructions of the cleaned trajectories, not the original rollout files. Trajectory metadata | |
| that has no Codex field (segment, flags, task_source, ...) is kept under session_meta.payload["trajectory"]. | |
| Every file is read back and compared with its source row (system prompt, then user text, assistant text, tool calls | |
| and tool outputs in order); any difference raises and the script exits non-zero. Idempotent: traces/ is recreated. | |
| Usage : python3 -I scripts/to_codex.py [--out DIR] [--ids ID[,ID...]] | |
| """ | |
| import argparse | |
| import calendar | |
| import json | |
| import os | |
| import shutil | |
| import sys | |
| import time | |
| ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| SPLITS = {"train": "trajectories.train.jsonl", "test": "trajectories.eval.jsonl"} | |
| META = ["segment", "segments_total", "compaction_summary_available", "thread_source", "agent_role", "task_source", | |
| "n_tool_calls", "n_failed_calls", "flags"] | |
| MSG_KEYS = {"role", "content", "tool_calls", "tool_call_id"} | |
| def epoch(ts): | |
| return calendar.timegm(time.strptime(ts.split(".")[0].rstrip("Z"), "%Y-%m-%dT%H:%M:%S")) | |
| def convert_row(row): | |
| ts, turn_id, msgs = row["timestamp"], row["turn_id"], row["messages"] | |
| if not msgs or msgs[0]["role"] != "system": | |
| raise ValueError("row must start with a system message: " + row["id"]) | |
| lines = [] | |
| def rec(kind, payload): | |
| lines.append({"timestamp": ts, "type": kind, "payload": payload}) | |
| meta = {"id": row["session_id"], "timestamp": ts, "cwd": row["cwd"], "originator": "Codex Desktop", | |
| "model_provider": "openai", "base_instructions": {"text": msgs[0]["content"]}, | |
| "trajectory": {"id": row["id"], "turn_id": turn_id, **{k: row[k] for k in META}}} | |
| rec("session_meta", meta) | |
| rec("event_msg", {"type": "task_started", "turn_id": turn_id, "started_at": epoch(ts)}) | |
| ctx = {"turn_id": turn_id, "cwd": row["cwd"]} | |
| if row["model"]: | |
| ctx["model"] = row["model"] | |
| if row["effort"]: | |
| ctx["effort"] = row["effort"] | |
| rec("turn_context", ctx) | |
| last_assistant = None | |
| for m in msgs[1:]: | |
| extra = set(m) - MSG_KEYS | |
| if extra: | |
| raise ValueError(f"unexpected message keys {sorted(extra)}") | |
| role, text = m["role"], m.get("content") or "" | |
| if role == "user": | |
| rec("response_item", {"type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}) | |
| rec("event_msg", {"type": "user_message", "message": text, "images": [], "local_images": [], "text_elements": []}) | |
| elif role == "assistant": | |
| if text: | |
| rec("response_item", {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": text}]}) | |
| last_assistant = text | |
| for tc in m.get("tool_calls") or []: | |
| if set(tc) != {"id", "type", "function"} or tc["type"] != "function" or set(tc["function"]) != {"name", "arguments"}: | |
| raise ValueError(f"unexpected tool call shape {sorted(tc)}") | |
| rec("response_item", {"type": "function_call", "name": tc["function"]["name"], | |
| "arguments": tc["function"]["arguments"], "call_id": tc["id"]}) | |
| elif role == "tool": | |
| rec("response_item", {"type": "function_call_output", "call_id": m["tool_call_id"], "output": text}) | |
| else: | |
| raise ValueError("unexpected role " + role) | |
| rec("event_msg", {"type": "task_complete", "turn_id": turn_id, "last_agent_message": last_assistant}) | |
| return lines | |
| def atoms_from_row(row): | |
| out = [("system", row["messages"][0]["content"])] | |
| for m in row["messages"][1:]: | |
| text = m.get("content") or "" | |
| if m["role"] == "user": | |
| out.append(("user", text)) | |
| elif m["role"] == "assistant": | |
| if text: | |
| out.append(("assistant", text)) | |
| for tc in m.get("tool_calls") or []: | |
| out.append(("call", tc["id"], tc["function"]["name"], tc["function"]["arguments"])) | |
| else: | |
| out.append(("output", m["tool_call_id"], text)) | |
| return out | |
| def atoms_from_lines(lines): | |
| out = [("system", lines[0]["payload"]["base_instructions"]["text"])] | |
| for r in lines: | |
| p = r["payload"] | |
| if r["type"] != "response_item": | |
| continue | |
| t = p["type"] | |
| if t == "message": | |
| (part,) = p["content"] | |
| out.append((p["role"], part["text"])) | |
| elif t == "function_call": | |
| out.append(("call", p["call_id"], p["name"], p["arguments"])) | |
| elif t == "function_call_output": | |
| out.append(("output", p["call_id"], p["output"])) | |
| else: | |
| raise ValueError("unexpected response_item " + t) | |
| return out | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--out", default=os.path.join(ROOT, "traces")) | |
| ap.add_argument("--ids", default="", help="comma-separated row ids; default: every row") | |
| args = ap.parse_args() | |
| out = os.path.abspath(args.out) | |
| only = {i for i in args.ids.split(",") if i} | |
| if os.path.exists(out): | |
| if os.path.basename(out) != "traces" and os.listdir(out): | |
| sys.exit("refusing to wipe a non-empty directory that is not named 'traces': " + out) | |
| shutil.rmtree(out) | |
| names, counts, total, biggest, n_calls = set(), {}, 0, 0, 0 | |
| for split, fn in SPLITS.items(): | |
| os.makedirs(os.path.join(out, split)) | |
| counts[split] = 0 | |
| with open(os.path.join(ROOT, fn), encoding="utf-8") as fh: | |
| for raw in fh: | |
| row = json.loads(raw) | |
| if only and row["id"] not in only: | |
| continue | |
| lines = convert_row(row) | |
| name = row["id"].replace("#", "_") + ".jsonl" | |
| if name in names: | |
| raise ValueError("duplicate file name " + name) | |
| names.add(name) | |
| path = os.path.join(out, split, name) | |
| body = "".join(json.dumps(x, ensure_ascii=False, separators=(",", ":")) + "\n" for x in lines) | |
| with open(path, "w", encoding="utf-8", newline="\n") as w: | |
| w.write(body) | |
| size = len(body.encode("utf-8")) | |
| total, biggest = total + size, max(biggest, size) | |
| counts[split] += 1 | |
| with open(path, encoding="utf-8") as r: | |
| back = [json.loads(x) for x in r] | |
| assert back[0]["type"] == "session_meta" and back[0]["payload"]["id"] == row["session_id"], path | |
| assert atoms_from_lines(back) == atoms_from_row(row), "content differs: " + path | |
| n_calls += sum(len(m.get("tool_calls") or []) for m in row["messages"]) | |
| print(json.dumps({"files": counts, "bytes_total": total, "bytes_largest_file": biggest, | |
| "tool_calls_verified": n_calls, "verified_round_trip": True}, indent=1)) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 8.24 kB
- Xet hash:
- 88b7ccf29c25537936b85fe1f96e8c7365d1d063e2631a08ed31c40a2d1e0d41
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.