EvoHarnessBench

def release_schedule(ann, stages=5, first_frac=0.45, min_uses=2, min_new=5):
    """Give every capability the stage it arrives at: {capability: stage}.

    Hands back the annotation as well, because dropping a capability drops the
    tasks that needed it -- the two have to stay in step.
    """
    ann = dict(ann)
    while True:                     # a rare capability is noise, and so is its task
        freq = Counter(c for caps in ann.values() for c in caps)
        rare = {c for c, n in freq.items() if n < min_uses}
        if not rare:
            break
        ann = {i: caps for i, caps in ann.items() if not (caps & rare)}
    if not freq:      # the third precondition failing, not a bug: this seed has no core
        raise ValueError("nothing survives min_uses: no capability repeats across tasks")

    # Core first, long tail last; the name breaks ties so the order can never
    # depend on the iteration order of a dict.
    order = sorted(freq, key=lambda c: (-freq[c], c))
    head = max(1, round(len(order) * first_frac))
    tail = order[head:]
    per = max(1, ceil(len(tail) / max(stages - 1, 1)))
    buckets = [order[:head]] + [tail[i:i + per] for i in range(0, len(tail), per)]
    release = {c: t for t, bucket in enumerate(buckets, 1) for c in bucket}

    return ann, _merge_thin(ann, release, min_new)


def _merge_thin(ann, release, min_new):
    """Fold away a stage too few tasks land on: it is a rerun of the one before it.

    The floor gives way to the stage count, since three stages is the least that is
    still a stream -- but a stage nothing at all lands on is folded whatever the
    count, because zero is not a judgment call. Folding the first stage forward is
    how the initial anchor grows when too little turns out to be covered by it.
    """
    while True:
        landed = Counter(date_tasks(ann, release).values())   # step 3, same module
        live = sorted(set(release.values()))
        floor = min_new if len(live) > 3 else 1
        thin = next((t for t in live if landed[t] < floor), None)
        if thin is None or len(live) == 1:
            break
        into = next((t for t in live if t > thin), live[live.index(thin) - 1])
        release = {c: (into if r == thin else r) for c, r in release.items()}

    renumber = {r: t for t, r in enumerate(sorted(set(release.values())), 1)}
    return {c: renumber[r] for c, r in release.items()}

def date_tasks(ann, release):
    """Each task lands at the first stage that covers it, which is the release
    stage of its latest-arriving capability."""
    return {i: max(release[c] for c in caps) for i, caps in ann.items() if caps}


def harness(release):
    """What a stage offers: everything released so far, distractors included. This
    is what the agent is handed -- never the oracle set, which makes growth free."""
    return {t: frozenset(c for c, r in release.items() if r <= t)
            for t in range(1, max(release.values()) + 1)}


def stream(ann, release, domain, adapt_ratio=0.3, seed=0, min_adapt=1, min_test=3):
    """One row per task, in the layout the published datasets use."""
    H, at, rng = harness(release), date_tasks(ann, release), random.Random(seed)
    rows = []
    for t in sorted(H):
        here = sorted(i for i, s in at.items() if s == t)
        rng.shuffle(here)
        # Floors on both sides: a stage scored on two tasks says nothing, and a
        # stage with nothing to adapt on cannot be adapted to.
        k = min(max(round(len(here) * adapt_ratio), min_adapt),
                max(len(here) - min_test, 0))
        for n, i in enumerate(here):
            assert ann[i] <= H[t]                               # solvable on arrival
            assert ann[i] & (H[t] - H.get(t - 1, frozenset()))  # needs what arrived
            rows.append({"domain": domain, "version": f"v{t}",
                         "split": "train" if n < k else "test", "task_id": i,
                         "oracle": sorted(ann[i]), "cumulative": sorted(H[t])})
    return rows


def manifest(ann, release, rows, **provenance):
    """What lets anyone check your stream without rerunning your build: what arrived
    when, what landed where, and the knobs you used. If a model is any part of your
    grader, pin its exact snapshot in here as provenance."""
    return {"stages": max(release.values()), "release": dict(sorted(release.items())),
            "dated": dict(sorted(date_tasks(ann, release).items())),
            "splits": {v: sorted(r["task_id"] for r in rows if r["version"] == v)
                       for v in sorted({r["version"] for r in rows})}, **provenance}


def write(rows, out: Path, meta=None):
    """v_k/train.jsonl, v_k/test.jsonl and the manifest beside them. That is the
    whole of the layout."""
    for row in rows:
        f = out / row["version"] / f"{row['split']}.jsonl"
        f.parent.mkdir(parents=True, exist_ok=True)
        with f.open("a") as fh:
            fh.write(json.dumps(row) + "\n")
    if meta:
        (out / "manifest.json").write_text(json.dumps(meta, indent=2, default=list))

import re

BOOKKEEPING = {"id", "uuid", "created_at", "updated_at", "created_by", "status"}


def mine_skills(prompt, min_chars=400):
    """Route A. Cut a shared policy on its own section structure, and hand back the
    prompt with what you cut REMOVED -- a procedure still sitting in front of the
    agent is not one it has to retrieve.
    """
    parts = re.split(r"\n(?=#{2,3} )", prompt)     # your seed's own headings
    general, skills = [parts[0]], {}
    for part in parts[1:]:
        title = part.splitlines()[0].lstrip("# ").strip()
        if len(part) >= min_chars:      # thinner than this is not a procedure
            skills[title.lower().replace(" ", "_")] = part
        else:
            general.append(part)
    return skills, "\n".join(general)


def skill_capabilities(task, skills, grader_reads):
    """Which mined skills a task cannot pass without -- decided by the verifier.

    `grader_reads(task)` is the one function you write: a parse of your own grader
    into the fields it asserts on and the literal values it asserts. The three rules
    above are the three lines below.
    """
    checked = grader_reads(task)
    need = set()
    for sid, body in skills.items():
        text = body.lower()
        hits = [v for v in checked["values"] if len(str(v)) > 3 and str(v).lower() in text]
        fields = [f for f in checked["fields"]
                  if f.lower() not in BOOKKEEPING and f.lower() in text]
        if hits or len(fields) > 1:     # a value decides; one bare field does not
            need.add(sid)
    return need

def owner_of(tool):
    """A tool namespace already encodes ownership -- split it. Where a seed has no
    namespace, a flat list of software say, canonicalize into families first so a
    stack, its CLI and its library all land in one place.
    """
    return tool.split(".", 1)[0]              # salesforce.create_case -> salesforce


def induce_agents(tool_ann):
    """{owner: its tools} and {task: the owners it needs}. The second one is the
    agent-axis annotation, and it goes through the same steps 2-4 as any other."""
    bundles = {}
    for caps in tool_ann.values():
        for tool in caps:
            bundles.setdefault(owner_of(tool), set()).add(tool)
    agent_ann = {i: frozenset(owner_of(t) for t in caps) for i, caps in tool_ann.items()}

    # Ownership has to be a partition and not a tagging, and the induced labels have
    # to cover the axis they came from -- otherwise the annotation means nothing.
    assert sum(map(len, bundles.values())) == len(set().union(*bundles.values()))
    for i, owners in agent_ann.items():
        assert tool_ann[i] <= set().union(*(bundles[o] for o in owners))
    return bundles, agent_ann

from pathlib import Path
from datasets import load_dataset
from evolve import annotate, manifest, release_schedule, stream, write

JUDGE = "pin-the-exact-snapshot-here"   # a judge that moves moves every score with it

# A closed catalog of the applications a world exposes, each with the cues that
# identify it. Lexical and fixed: no model decides what a task needs.
APPS = {
    "mail":          ("email", "e-mail", "inbox", "reply to", "sends a message"),
    "spreadsheets":  ("spreadsheet", "xlsx", "worksheet", "cell", "formula", "tab named"),
    "documents":     ("document", "docx", "memo", "write-up", "drafts a"),
    "presentations": ("presentation", "slide", "deck", "pptx"),
    "calendar":      ("calendar", "meeting", "invite", "schedules a"),
    "chat":          ("chat", "channel", "direct message"),
    "code":          ("script", "python", "notebook", "code execution"),
    "pdfs":          ("pdf", "scanned", "filing"),
}


def apex_tasks():
    """Take the dataset card's own advice and print ds[0].keys() first -- field names
    move. What this reads is the id, the world, the prompt and the rubric criteria.
    """
    for row in load_dataset("mercor/apex-agents", split="train"):   # gated: accept terms
        rubric = row.get("rubric") or row.get("rubric_criteria") or []
        criteria = [c if isinstance(c, str) else c.get("criterion", str(c)) for c in rubric]
        yield {"task_id": row["task_id"], "world": row["world_id"],
               "text": " ".join([row["prompt"], *criteria]).lower()}


def apex_capabilities(task):
    """A criterion that says "the email states X" needs Mail. Grade-relevant text is
    the right thing to read: the rubric is what the task is actually about.
    """
    return {app for app, cues in APPS.items() if any(c in task["text"] for c in cues)}


ann = annotate(apex_tasks(), apex_capabilities)
ann, release = release_schedule(ann, stages=4, first_frac=0.5, min_new=12)
rows = stream(ann, release, domain="apex_agents")
write(rows, Path("data/apex_agents"), manifest(ann, release, rows, judge=JUDGE))

# The agent axis then comes for free: one specialist per application, since an
# application owns its own actions. induce_agents(ann), then steps 2-4 again.