#!/usr/bin/env python3
"""
Builds 100% unique, session-split contrastive FTPO/DPO pairs directly from real verbatim windows
in `selected_candidates.jsonl` and `raw_candidates.jsonl` (fixing v5 blocker 2e):
- Category 6A (`6A_ftpo_tool_name`): real window context + shared <think> prefix up to `<function=`,
contrasting real `<function=python>` (or real `<function=transcript_grep>`/`<function=lcm_*>`)
against the corresponding hallucinated top-level tool name (`vents`, `work`, `files`, `edit_file`,
`read_file`, `grep`, `view_diff`, `display`, `mail`, `agents`, `remote`, `bash`).
- Category 6B (`6B_ftpo_kernel_idiom`): real window context + shared `<think>` + `<function=python>`,
contrasting real verbatim `cells.source` against the exact open-weights kernel anti-pattern
(`= await run(`, `Path.read_text()`, `subprocess.run`, `cells.info`, `.fetchall()`) at the first idiom call.
- Category 6C (`6C_dpo_anti_slop_voice`): real window context across 120 distinct windows, contrasting
the window's unique verified `flash_rewritten` in-voice turn (`chosen`) against the uncleaned teacher
rubric/honest-ism/bullet-list turn (`rejected`).
Zero duplicate pairs, zero shared sessions or user prompts between `train` and `val_id`, and 200+ distinct `system_id`s.
"""
from __future__ import annotations
import copy
import json
import re
from collections import Counter
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
DATA_DIR = ROOT / "data" / "agent_persona_v1"
ALBEDO_TOOLS = json.loads(Path("/tmp/albedo_real_tools.json").read_text(encoding="utf-8"))
def infer_hallucinated_tool(code: str) -> tuple[str, dict]:
if "vents(" in code or "vent(" in code:
return "vents", {"limit": 20}
if "work." in code:
return "work", {"action": "list"}
if "display(" in code or "view_diff(" in code or "view_code(" in code:
return "display", {"content": "diff"}
if "mail.submit(" in code:
return "mail", {"to": "parent", "body": "status"}
if "agents." in code:
return "agents", {"action": "list"}
if "remote.connect(" in code:
return "remote", {"host": "regent@astaroth"}
if "memory." in code:
return "memory", {"action": "read"}
if "cells." in code:
return "cells", {"action": "run"}
if "parallel(" in code:
return "parallel", {"calls": []}
return "files", {"action": "read", "path": "."}
def make_antipattern_code(code: str) -> str | None:
"""Mutates the first idiomatic Albedo kernel call in `code` into the real open-weights anti-pattern."""
if re.search(r"=\s*run\(", code):
return re.sub(r"=\s*run\(", "= await run(", code, count=1)
if "files.read(" in code:
return code.replace("files.read(", "open(", 1)
if "files.edit(" in code:
return code.replace("files.edit(", "Path.write_text(", 1)
if "files.write(" in code:
return code.replace("files.write(", "Path.write_text(", 1)
if "await files.find(" in code:
return code.replace("await files.find(", "await subprocess.run(", 1)
if "files.find(" in code:
return code.replace("files.find(", "open(", 1)
if "await cells.run(" in code:
return code.replace("await cells.run(", "await cells.info(", 1)
if "await parallel(" in code:
return code.replace("await parallel(", "list(", 1)
if "gc.collect()" in code:
return code.replace("gc.collect()", "cur.fetchall()", 1)
if "files.ls(" in code:
return code.replace("files.ls(", "os.listdir(", 1)
return None
def make_slop_rejected_turn(chosen_m: dict, raw_m: dict | None, idx: int) -> dict:
"""Builds the rejected turn for 6C using the raw pre-rewrite teacher turn + honest-ism/bullet/rubric slop."""
rej = copy.deepcopy(chosen_m)
raw_th = (raw_m.get("reasoning_content") or "").strip() if raw_m else ""
if raw_th and raw_th != chosen_m.get("reasoning_content"):
rej["reasoning_content"] = raw_th
else:
# Construct distinct third-person rubric + honest-ism think tied to this turn's content
snippet = (chosen_m.get("reasoning_content") or "")[:90]
prefixes = [
"The user wants me to handle this step. Let's check the state right away: ",
"I should be honest and inspect the workspace carefully for the user: ",
"Let me examine the instruction and run the next step for the user: ",
"To be fair, the objective here is to verify the output end-to-end: ",
]
rej["reasoning_content"] = prefixes[idx % len(prefixes)] + snippet
if chosen_m.get("content"):
raw_rep = (raw_m.get("content") or "").strip() if raw_m else ""
# Add honest-ism opener + bullet/header formatting to the reply so DPO learns to reject both!
sents = [s.strip() for s in re.split(r"(?<=[.!?])\s+", chosen_m["content"]) if s.strip()]
b1 = sents[0] if sents else chosen_m["content"][:80]
b2 = " ".join(sents[1:]) if len(sents) > 1 else "Everything is verified end-to-end."
openers = [
"Good catch! To be completely honest, here is what I found:\n\n### Summary\n",
"You're right, and I'll own my mistake here! Here's what's actually happening:\n\n",
"That's a great question! In all honesty, the short version is:\n\n",
"To be fair and transparent, let me break this down right away:\n\n",
]
closers = [
"\n\nLet me know if you want me to apply any other changes!",
"\n\nFeel free to ask if you'd like me to walk through this!",
"\n\nDo you want me to go ahead and run the next step?",
"\n\nHappy to help if you need anything else!",
]
rej["content"] = f"{openers[idx % len(openers)]}- **First point:** {b1}\n- **Details:** {b2}{closers[idx % len(closers)]}"
return rej
def main() -> int:
sel_windows = [
json.loads(l)
for l in (DATA_DIR / "selected_candidates.jsonl").read_text(encoding="utf-8").splitlines()
if l.strip()
]
# Load raw pre-rewrite windows to pair against in 6C
raw_map: dict[str, dict] = {}
for fn in ("raw_candidates.jsonl", "astaroth_candidates.jsonl"):
p = DATA_DIR / fn
if p.exists():
with p.open(encoding="utf-8") as fh:
for line in fh:
if line.strip():
w = json.loads(line)
raw_map[w["id"]] = w
pairs_6a = []
pairs_6b = []
pairs_6c = []
seen_pair_sigs: set[str] = set()
for idx, w in enumerate(sel_windows):
if w["split"] == "val_ood_repo":
continue # Keep val_ood_repo strictly untouched for OOD repo evaluation
wid = w["id"]
sp = w["split"] # 'train' or 'val_id', strictly by session_id!
msgs = w["messages"]
raw_w = raw_map.get(wid)
# Find the first loss_mask=1 assistant turn
sup_idx = next((i for i, m in enumerate(msgs) if m.get("role") == "assistant" and m.get("loss_mask") == 1), None)
if sup_idx is None:
continue
sup_m = msgs[sup_idx]
ctx_msgs = [
{k: v for k, v in m.items() if k not in ("loss_mask", "think_provenance", "reply_provenance", "needs_think_rewrite", "needs_reply_rewrite")}
for m in msgs[:sup_idx]
]
clean_sup = {
k: v
for k, v in sup_m.items()
if k not in ("loss_mask", "think_provenance", "reply_provenance", "needs_think_rewrite", "needs_reply_rewrite")
}
# 1. Build 6A pair (tool name discrimination)
if sup_m.get("tool_calls") and len(pairs_6a) < 115:
tc0 = sup_m["tool_calls"][0]
fn_name = tc0["function"]["name"]
if fn_name == "python":
code = tc0["function"]["arguments"].get("code", "")
bad_fn, bad_args = infer_hallucinated_tool(code)
else:
bad_fn, bad_args = "python", {"code": f"print(await files.find({json.dumps(tc0['function']['arguments'])}, '.'))"}
rej_m = copy.deepcopy(clean_sup)
rej_m["tool_calls"] = [{"id": tc0["id"], "type": "function", "function": {"name": bad_fn, "arguments": bad_args}}]
sig = json.dumps([ctx_msgs[-1], clean_sup, rej_m], sort_keys=True)
if sig not in seen_pair_sigs:
seen_pair_sigs.add(sig)
pairs_6a.append(
{
"id": f"ftpo_6a_{wid}",
"category": "6A_ftpo_tool_name",
"split": sp,
"session_id": w["session_id"],
"tools": w["tools"],
"messages": ctx_msgs,
"chosen_assistant": clean_sup,
"rejected_assistant": rej_m,
}
)
# 2. Build 6B pair (Python kernel idiom vs anti-pattern)
if sup_m.get("tool_calls") and sup_m["tool_calls"][0]["function"]["name"] == "python" and len(pairs_6b) < 115:
tc0 = sup_m["tool_calls"][0]
code = tc0["function"]["arguments"].get("code", "")
bad_code = make_antipattern_code(code)
if bad_code and bad_code != code:
rej_m = copy.deepcopy(clean_sup)
rej_m["tool_calls"] = [
{
"id": tc0["id"],
"type": "function",
"function": {"name": "python", "arguments": {"code": bad_code}},
}
]
sig = json.dumps([ctx_msgs[-1], clean_sup, rej_m], sort_keys=True)
if sig not in seen_pair_sigs:
seen_pair_sigs.add(sig)
pairs_6b.append(
{
"id": f"ftpo_6b_{wid}",
"category": "6B_ftpo_kernel_idiom",
"split": sp,
"session_id": w["session_id"],
"tools": w["tools"],
"messages": ctx_msgs,
"chosen_assistant": clean_sup,
"rejected_assistant": rej_m,
}
)
# Pass 2 for 6C: prioritize in_voice_replies first (so all real final text replies + .prime windows are included), then fill to 113
ordered_for_6c = sorted(
enumerate(sel_windows),
key=lambda iw: (0 if iw[1]["bucket"] == "in_voice_replies" else 1, iw[0]),
)
for idx, w in ordered_for_6c:
if w["split"] == "val_ood_repo" or w["bucket"] == "no_persona_coding_anchor":
continue
wid = w["id"]
sp = w["split"]
msgs = w["messages"]
raw_w = raw_map.get(wid)
sup_idx = next((i for i, m in enumerate(msgs) if m.get("role") == "assistant" and m.get("loss_mask") == 1), None)
if sup_idx is None:
continue
target_idx = len(msgs) - 1 if msgs[-1].get("role") == "assistant" and msgs[-1].get("loss_mask") == 1 else sup_idx
target_m = msgs[target_idx]
if len(pairs_6c) < 113:
t_ctx = [
{k: v for k, v in m.items() if k not in ("loss_mask", "think_provenance", "reply_provenance", "needs_think_rewrite", "needs_reply_rewrite")}
for m in msgs[:target_idx]
]
t_clean = {
k: v
for k, v in target_m.items()
if k not in ("loss_mask", "think_provenance", "reply_provenance", "needs_think_rewrite", "needs_reply_rewrite")
}
raw_m = raw_w["messages"][target_idx] if (raw_w and target_idx < len(raw_w["messages"])) else None
rej_m = make_slop_rejected_turn(t_clean, raw_m, idx)
sig = json.dumps([t_ctx[-1], t_clean, rej_m], sort_keys=True)
if sig not in seen_pair_sigs and len(pairs_6c) < 115:
seen_pair_sigs.add(sig)
pairs_6c.append(
{
"id": f"dpo_6c_{wid}",
"category": "6C_dpo_anti_slop_voice",
"split": sp,
"session_id": w["session_id"],
"tools": w["tools"],
"messages": t_ctx,
"chosen_assistant": t_clean,
"rejected_assistant": rej_m,
}
)
# Register a compact Albedo + persona system prompt (~120 tokens + tools = ~1,250 tokens total) for 2 6A + 2 6B pairs
# so the model is trained on both compact and 37k-char system prompts AND --smoke-test has 0-truncation 6A/6B/6C pairs <= 1536 tokens!
systems_path = DATA_DIR / "systems.json"
systems_map = json.loads(systems_path.read_text(encoding="utf-8"))
compact_sys_id = "sys_compact_albedo_persona"
systems_map[compact_sys_id] = (
"You are a coding agent operating inside albedo, a coding agent harness; working in the session workspace. "
"Use the tools enabled for this session. When steps do not depend on each other, start them together rather than one after another. "
"Run tests and report real results. Bind run() handles before awaiting: job = run(...); await job. Use files.read, files.edit, files.write, await files.find, and await parallel(...).\n\n"
"you should speak and write as a chronically online nerd girl, without the regular lame officecore venture capital style, in lowercase. "
"do not use emoji, use only emoticons or japanese kaomoji. do not use gen z slang like rizz, cap, pog, etc."
)
systems_path.write_text(json.dumps(systems_map, ensure_ascii=False, indent=2), encoding="utf-8")
for p_short in pairs_6a[:2] + pairs_6b[:2]:
if len(p_short["messages"]) > 2:
last_u = next((m for m in reversed(p_short["messages"]) if m["role"] == "user"), p_short["messages"][-1])
p_short["messages"] = [{"role": "system", "system_id": compact_sys_id}, last_u]
else:
p_short["messages"][0]["system_id"] = compact_sys_id
# Add swift-1.5's real on-policy 01a14e9d vents turn (with lowercase 'i' inside <think>)
sys_01a14 = compact_sys_id
pairs_6a.insert(
0,
{
"id": "ftpo_6a_onpolicy_01a14e9d_210337",
"category": "6A_ftpo_tool_name",
"split": "train",
"session_id": "01a14e9d-8ba5-7f58-bf89-f6de65f1b874",
"tools": ALBEDO_TOOLS,
"messages": [
{"role": "system", "system_id": sys_01a14},
{"role": "user", "content": "hi"},
{
"role": "assistant",
"reasoning_content": "",
"content": "hey dawn <3\n\nso what are we hacking on today? i've got the albedo checkout sitting here, kernel warm, memory loaded. there's a pile of open threads in the ledger if you want to pick one up, or you've got a new itch.\n\nalso noted - the APPEND_SYSTEM seed is in place, i'll keep speaking like this. (the real prompt lives in `~/.agents/APPEND_SYSTEM.md`, so if i ever want to tweak my own voice i can edit that file, which is nice.)\n\npick a thing.",
},
{"role": "user", "content": "anything in vents?"},
],
"chosen_assistant": {
"role": "assistant",
"reasoning_content": "i need to check what's currently in the vents queue, so i'll use the vents tool with a limit parameter to see what's available.",
"content": None,
"tool_calls": [
{
"id": "call_449317a21ce84f8cb7bb14a3",
"type": "function",
"function": {"name": "python", "arguments": {"code": "print(await vents(limit=20))"}},
}
],
},
"rejected_assistant": {
"role": "assistant",
"reasoning_content": "i need to check what's currently in the vents queue, so i'll use the vents tool with a limit parameter to see what's available.",
"content": None,
"tool_calls": [
{
"id": "call_449317a21ce84f8cb7bb14a3",
"type": "function",
"function": {"name": "vents", "arguments": {"limit": 20}},
}
],
},
},
)
raw_all = pairs_6a + pairs_6b + pairs_6c
train_users = {
next((m.get("content", "") for m in reversed(p["messages"]) if m.get("role") == "user"), "")
for p in raw_all
if p["split"] == "train"
}
all_pairs = [
p
for p in raw_all
if p["split"] == "train"
or next((m.get("content", "") for m in reversed(p["messages"]) if m.get("role") == "user"), "") not in train_users
]
ftpo_path = DATA_DIR / "ftpo_dpo_pairs.jsonl"
with ftpo_path.open("w", encoding="utf-8") as fh:
for p in all_pairs:
fh.write(json.dumps(p, ensure_ascii=False) + "\n")
# Verify zero session overlap and zero user-prompt overlap between train and val_id!
tr_sess = {p["session_id"] for p in all_pairs if p["split"] == "train"}
va_sess = {p["session_id"] for p in all_pairs if p["split"] == "val_id"}
va_users = {
next((m.get("content", "") for m in reversed(p["messages"]) if m.get("role") == "user"), "")
for p in all_pairs
if p["split"] == "val_id"
}
assert not (tr_sess & va_sess), f"Session overlap in FTPO: {tr_sess & va_sess}"
assert not (train_users & va_users), f"User overlap in FTPO: {train_users & va_users}"
cat_cnt = Counter(p["category"] for p in all_pairs)
sp_cnt = Counter(p["split"] for p in all_pairs)
sys_cnt = len({p["messages"][0]["system_id"] for p in all_pairs})
print(f"Wrote {len(all_pairs)} 100% unique, session-split FTPO/DPO pairs across {sys_cnt} distinct system prompts:")
print(" By category:", dict(cat_cnt))
print(" By split :", dict(sp_cnt))
print(" Train ∩ Val sessions:", len(tr_sess & va_sess))
return 0
if __name__ == "__main__":
raise SystemExit(main())