EvoHarnessBench

SDK_BASE="https://educator-marrow-cultural.ngrok-free.dev"
: "${EVAL_SERVICE_API_KEY:?Required — get it from MyAuthtoken first}" &&
WHEEL=$(curl -sSL -H "Authorization: Bearer $EVAL_SERVICE_API_KEY" "$SDK_BASE/sdk" \
  | python -c 'import json,sys; d=json.load(sys.stdin); print(d["path"]) if "path" in d else sys.exit("SDK download denied: " + d.get("detail", "unknown error"))') &&
curl -fsSL -H "Authorization: Bearer $EVAL_SERVICE_API_KEY" \
  "$SDK_BASE$WHEEL" -o "/tmp/${WHEEL##*/}" &&
python -m pip install "/tmp/${WHEEL##*/}"

from simple_agentic_evals import (CommandAgent, EvalClient, ServiceError,
                                  catalog_targets, resolve_target,
                                  run_benchmark, run_leaderboard,
                                  react_agent, acp_codex_agent, to_openai_tools)

client    = EvalClient()      # endpoint and auth resolved for you -- no URL to paste
LLM_MODEL = "gpt-4o-mini"     # whatever YOUR method thinks with

# Every /v1/* route needs the key, health included, so this one call both connects
# and proves the key -- a bad one fails HERE rather than three sections later.
print("connected ->", client.health())

print(client.health())
for target in catalog_targets(client):
    print(target["dataset"], target["benchmark"],
          "runnable=", target.get("runnable"),
          "health=", target.get("health"),
          "actions=", target.get("action_types"))

# Hosted is selected only when every requested capability is healthy.
choice = resolve_target("eog", dataset="evovling_tools",
                        mode="deployment", harness="react", action_type="mcp",
                        client=client)
assert choice["location"] == "hosted", choice
print(choice["reason"])

def solve_eog(task, max_steps=6):
    """EOG: act on the gym through MCP tools. Replace the loop with your method."""
    used = 0
    mcp = task.mcp_session(task.mcp_servers[0])
    try:
        tools = to_openai_tools(mcp.list_tools())      # MCP schemas -> OpenAI specs
        msgs = [{"role": "system", "content": task.system_prompt or ""},
                {"role": "user",   "content": task.user_prompt or ""}]
        for _ in range(max_steps):
            r = oai.chat.completions.create(model=LLM_MODEL, messages=msgs, tools=tools)
            used += r.usage.total_tokens
            m = r.choices[0].message
            msgs.append(m.model_dump(exclude_none=True))
            if not m.tool_calls:                       # no call -> the agent is done
                break
            for tc in m.tool_calls:
                out = mcp.call_tool(tc.function.name,
                                    json.loads(tc.function.arguments or "{}"))
                msgs.append({"role": "tool", "tool_call_id": tc.id,
                             "content": json.dumps(out)[:4000]})
    finally:
        mcp.close()                                    # rule 2 -- even if we raised
    return {"total_tokens": used}


def solve_ale(task):
    """ALE: read the staged inputs, write the deliverable. One shot, no tools."""
    files = task.inputs()
    head = task.fetch_input(files[0]["path"])[:4000] if files else ""
    r = oai.chat.completions.create(model=LLM_MODEL, messages=[
        {"role": "system", "content": task.system_prompt or ""},
        {"role": "user", "content": f"{task.user_prompt}\n\n{head}"}])
    task.submit_text(task.output_path, r.choices[0].message.content or "")
    return {"total_tokens": r.usage.total_tokens}


def my_agent(task):
    """One entry point for callable MCP and artifact adapters."""
    if task.action_type == "mcp":
        return solve_eog(task)
    if task.action_type == "sandbox":
        return solve_ale(task)
    raise RuntimeError(f"Unsupported hosted action type: {task.action_type}")

server  = task.mcp_servers[0]
url     = task.mcp_url(server)          # hand these two to your own MCP client
headers = {"Authorization": f"Bearer {EVAL_SERVICE_API_KEY}", **server.headers}

# Or use the bundled minimal client. These two calls are the entire seam --
# everything else in the SDK is a convenience on top of them.
mcp   = task.mcp_session(server)
tools = mcp.list_tools()                # 1. WHAT YOUR POLICY CAN DO   (plain JSON Schema)
mcp.call_tool(tools[0]["name"], {})     # 2. HOW YOUR POLICY ACTS      (a plain function)
mcp.close()

import subprocess, tempfile
from pathlib import Path


def containerized_agent(task):
    """An existing container, unchanged, scored on an ALE task.

    It never talks to the service: it reads one directory and writes another,
    exactly as it does on your laptop. We fill the first, we collect the second.
    """
    with tempfile.TemporaryDirectory() as tmp:
        work   = Path(tmp)
        indir  = work / "input"                       # we fill this one
        outdir = work / "output"                      # your system fills this one
        outdir.mkdir()

        task.fetch_inputs_to(indir)                   # 1. service -> your disk
        subprocess.run([                              # 2. your system runs, however it likes
            "docker", "run", "--rm", "-v", f"{work}:/work", "my-system:latest",
            "--input", "/work/input", "--output", "/work/output",
        ], check=True, timeout=1800)
        task.submit_dir(outdir)                       # 3. your disk -> service, then grading

    # Returning 0 would be worse than saying nothing: it reports a measurement of zero.
    return None


# Nothing to do with Docker, really. For a plain CLI, steps 1 and 3 are identical and
# step 2 becomes: subprocess.run(["my-system", "--in", indir, "--out", outdir])

report = run_benchmark(my_agent, mode="deployment_eval", client=client,
                       benchmark="eog", limit=2, progress=True)
print(report)                                # the paper-style block

print(report.success_rate, report.accuracy)  # Pass % (strict), Score (partial credit)
print(report.agent_s, report.total_tokens)   # agent time, not wall clock; None if unmeasured
print(report.n_errors, report.complete)      # a non-zero n_errors did not measure what you think


def floor_agent(task):
    """Acts on nothing. Whatever this scores, the environment was giving away."""
    return {"total_tokens": 0}


floor = run_benchmark(floor_agent, "deployment_eval", client=client,
                      benchmark="eog", limit=2, progress=False)
print(f"floor {floor.accuracy:.3f}   mine {report.accuracy:.3f}")
# ^ if these two match, your agent is not acting -- check that it reaches mcp.call_tool

# The SDK runs both benchmarks, aggregates repeats, and returns leaderboard JSON.
# A limited rehearsal is stamped `partial`; token telemetry stays None if unreported.
row = run_leaderboard(my_agent, "My Method", client=client,
                      benchmarks=["eog", "ale"], cat="memory", seeds=1, limit=2)
print(json.dumps(row, indent=2))
print(row["benchmark_results"])       # results keyed by hosted benchmark id

#   Pass %    mean(r.success_rate)          strict -- every verifier on the task passed
#   Score     mean(r.accuracy)              partial credit -- mean verifier pass rate
#   h         mean(r.agent_s) / 3600        agent time; r.latency_s is wall clock, not this
#   Tok M     mean(r.total_tokens) / 1e6    None rather than 0 when nothing reported usage
#   Overall   sum(n_success) / sum(n_tasks) pooled across both, over tasks not percentages
#   +/-       pstdev over the `seeds` repeats

# A publishable no-limit run is expensive and requires explicit acknowledgement:
# row = run_leaderboard(my_agent, "My Method", client=client, seeds=3,
#                       benchmarks=["eog", "ale"],
#                       confirm_full_cost=True)

# Two switches worth knowing before you commit:
#   dataset=  the evolving axis -- "evovling_tools" | "evovling_skills" | "evovling_agents",
#             each a separate leaderboard.
#   mode=     the question. "deployment_eval" is a fixed method under a growing harness and
#             is the right default; a method that LEARNS between stages belongs in
#             "self_evolving_adapt_eval" -- the next tab.

report = run_benchmark(my_agent, "deployment_eval", client=client,
                       domain="calendar", matrix=True)    # costs roughly T x the last row

#                  cohort v1   cohort v2   cohort v3
#   under H_v1          0.31           .           .    <- adaptation: v1 tasks, v1 tools
#   under H_v2          0.28        0.40           .    <- retention:  v1 tasks, v2 tools exist
#   under H_v3          0.25        0.36        0.22    <- the last row is what ACC pools
#
# Down a column is one cohort's story as the world grows around it. 0.31 -> 0.28 -> 0.25 is
# FORGETTING: nothing about those tasks changed, only the harness did. That column delta is
# what BWT summarizes.

print(report.bwt, report.fwt)     # cohort-size-weighted means, exactly as the appendix defines
print(report.per_stage_acc)       # the diagonal, by stage

class EvolvingAgent:
    """A policy with state: `adapt` writes it, `__call__` reads it.

    run_benchmark picks up `.adapt` automatically, so this object is the whole
    integration -- no registration, no adapt= argument needed.
    """

    def __init__(self):
        self.notes = []

    def adapt(self, stage, tasks):      # once per stage, with that stage's TRAIN split
        for t in tasks:
            with t:                     # prompts exist only once a task is provisioned
                self.notes.append(f"[v{stage}] {(t.user_prompt or '')[:140]}")

    def __call__(self, task):
        # `act` is the loop from "Wrap your method", with room in the system prompt
        # for whatever the policy has learned so far.
        return act(task, notes="\n".join(self.notes[-8:]))


evolve = run_benchmark(EvolvingAgent(), "self_evolving_adapt_eval", client=client,
                       domain="calendar", matrix=True)
print(evolve.bwt, evolve.fwt)           # FWT is None in deployment mode; here it is real