Your First Agent

# script_id: day_048_first_agent/first_agent
# The smallest agent: a while loop, two read-only Orbit tools, one brake (max_steps).
# Install: uv add openai   Env: OPENAI_API_KEY, ATLAS_MODEL (default gpt-6-luna), ORBIT_DATA
# Services: none. Reads datasets/orbit/issues/orbit.db (read-only) and docs/*.md.
import json
import os
import sqlite3
from contextlib import closing
from pathlib import Path

from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))


def search_issues(query: str) -> list[dict]:  # simplified from Day 35: keyword-ranked, no state/limit
    words = [w for w in query.lower().split() if len(w) > 2]
    uri = f"file:{DATA_DIR / 'issues' / 'orbit.db'}?mode=ro"  # read-only: tools can't write
    with closing(sqlite3.connect(uri, uri=True)) as db:
        rows = db.execute("SELECT id, title, lower(title || ' ' || body) FROM issues").fetchall()
    ranked = sorted(rows, key=lambda r: -sum(w in r[2] for w in words))[:5]
    return [{"id": i, "title": t} for i, t, text in ranked if any(w in text for w in words)]


def get_doc_section(doc_id: str) -> str:  # simplified from Day 35: whole page, capped, no heading
    path = DATA_DIR / "docs" / f"{Path(doc_id).name}.md"  # .name blocks ../ traversal
    return path.read_text(encoding="utf-8")[:3000] if path.exists() else f"unknown doc_id {doc_id!r}"


def schema(name: str, description: str, arg: str) -> dict:  # flat, strict tool schema (Day 36)
    return {"type": "function", "name": name, "description": description, "strict": True,
            "parameters": {"type": "object", "properties": {arg: {"type": "string"}},
                           "required": [arg], "additionalProperties": False}}


TOOLS = [schema("search_issues", "Keyword search over past Orbit issues.", "query"),
         schema("get_doc_section", "Read an Orbit doc page by id, e.g. 'retries-and-backoff'.", "doc_id")]
REGISTRY = {"search_issues": search_issues, "get_doc_section": get_doc_section}

# script_id: day_048_first_agent/agent
# atlas/agent.py: run_agent() with five stop conditions, budgets and a labelled partial result.
# Install: uv add openai opentelemetry-api   Env: OPENAI_API_KEY, ATLAS_MODEL, ORBIT_DATA
# Services: none (orbit.db read-only, docs/*.md). Spans are no-ops until Day 42's setup_tracing().
from __future__ import annotations

import json
import os
import time
from dataclasses import dataclass, field
from typing import Literal

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
StopReason = Literal["done", "max_steps", "max_seconds", "max_tokens", "max_errors", "interrupted"]


@dataclass(frozen=True)
class AgentConfig:
    max_steps: int = 8               # starting point: re-set to p95 steps + 2 after the 5-set run
    max_seconds: float = 60          # wall clock, checked between steps
    max_tokens: int = 100_000        # cumulative input + output tokens across all steps
    max_errors: int = 3              # consecutive steps that had any failure
    effort: str | None = None        # None = the model's default ("medium" on gpt-6-luna)
    request_timeout: float = 30.0    # seconds per model call; raise it if you use effort="high"
    max_output_tokens: int = 25_000  # reasoning + visible output; the reasoning guide says reserve 25k


@dataclass
class RunState:
    started: float
    steps: int = 0
    tokens: int = 0
    consecutive_errors: int = 0
    done: bool = False
    calls: list[str] = field(default_factory=list)  # every "name(args)" so far, for debugging

# script_id: day_048_first_agent/agent
class StopChecker:
    """Pure: the same state and clock reading always give the same answer."""

    def __init__(self, config: AgentConfig) -> None:
        self.c = config

    def check(self, s: RunState, now: float) -> StopReason | None:
        if s.done:
            return "done"
        if s.steps >= self.c.max_steps:
            return "max_steps"
        if s.tokens >= self.c.max_tokens:
            return "max_tokens"
        if s.consecutive_errors >= self.c.max_errors:
            return "max_errors"
        if now - s.started >= self.c.max_seconds:
            return "max_seconds"
        return None

# script_id: day_048_first_agent/test_stop_checker
# Offline StopChecker tests with a fake clock: no API key, no network, no sleeping.
# Install: uv add --dev pytest   Run: uv run pytest test_stop_checker.py -q  (or: python test_stop_checker.py)
# Needs agent.py (the `agent` script) in the same folder.
from agent import AgentConfig, RunState, StopChecker

CFG = AgentConfig(max_steps=3, max_seconds=10, max_tokens=1_000, max_errors=2)


def check(now: float = 0.0, **state) -> str | None:
    return StopChecker(CFG).check(RunState(started=0.0, **state), now)


def test_keeps_going_under_every_limit() -> None:
    assert check(now=9.9, steps=2, tokens=999, consecutive_errors=1) is None


def test_each_limit_trips_at_its_boundary() -> None:
    assert check(steps=3) == "max_steps"
    assert check(tokens=1_000) == "max_tokens"
    assert check(consecutive_errors=2) == "max_errors"
    assert check(now=10.0) == "max_seconds"  # the fake clock jumps; nobody waits 10 s


def test_done_wins_over_a_limit_crossed_on_the_same_step() -> None:
    assert check(steps=3, tokens=5_000, done=True) == "done"


def test_first_limit_in_check_order_is_reported() -> None:
    assert check(now=99.0, steps=3, tokens=5_000) == "max_steps"


if __name__ == "__main__":
    for name, fn in list(globals().items()):
        if name.startswith("test_"):
            fn()
            print("PASS", name)

PASS test_keeps_going_under_every_limit
PASS test_each_limit_trips_at_its_boundary
PASS test_done_wins_over_a_limit_crossed_on_the_same_step
PASS test_first_limit_in_check_order_is_reported

# script_id: day_048_first_agent/agent
@dataclass
class StepRecord:
    index: int
    tool_calls: list[str] = field(default_factory=list)  # "name(args)" for each call
    errors: int = 0              # failed tool calls, failed or incomplete model calls
    input_tokens: int = 0
    output_tokens: int = 0
    ms: int = 0


@dataclass
class AgentResult:
    text: str
    stop_reason: StopReason
    steps: list[StepRecord]
    usage: dict[str, int]        # {"input_tokens": n, "output_tokens": n}, wrap-up call included
    cost_usd: float
    partial: bool                # True whenever stop_reason != "done"

# script_id: day_048_first_agent/agent
import difflib
import sqlite3
from contextlib import closing
from functools import cache
from pathlib import Path

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))


def _rows(sql: str, *args) -> list[tuple]:  # built on Day 33: read-only connection
    with closing(sqlite3.connect(f"file:{DATA_DIR / 'issues' / 'orbit.db'}?mode=ro", uri=True)) as db:
        return db.execute(sql, args).fetchall()


@cache
def _docs() -> dict[str, str]:
    return {p.stem: p.read_text(encoding="utf-8") for p in sorted((DATA_DIR / "docs").glob("*.md"))}


def _rank(texts: dict, query: str, k: int) -> list:  # keyword stand-in for R2 search (Day 34)
    words = [w for w in query.lower().split() if len(w) > 2]
    scored = sorted(((sum(w in t.lower() for w in words), key) for key, t in texts.items()), reverse=True)
    return [key for score, key in scored[:k] if score > 0]


def search_issues(query: str) -> list[dict]:
    rows = {r[0]: r for r in _rows("SELECT id, title, state, duplicate_of, title || ' ' || body FROM issues")}
    keys = ("id", "title", "state", "duplicate_of")
    return [dict(zip(keys, rows[i][:4])) for i in _rank({i: r[4] for i, r in rows.items()}, query, 5)]


def get_issue(issue_id: int) -> dict:
    found = _rows("SELECT title, state, body, duplicate_of FROM issues WHERE id = ?", issue_id)
    if not found:
        raise LookupError(f"No issue #{issue_id}. Use ids returned by search_issues.")
    title, state, body, dup = found[0]
    labels = [r[0] for r in _rows("SELECT l.name FROM issue_labels il JOIN labels l ON l.id = il.label_id "
                                  "WHERE il.issue_id = ?", issue_id)]
    comments = [r[0][:300] for r in _rows("SELECT body FROM comments WHERE issue_id = ? ORDER BY position "
                                          "LIMIT 3", issue_id)]
    return {"id": issue_id, "title": title, "state": state, "labels": labels, "duplicate_of": dup,
            "body": body[:1500], "comments": comments}

# script_id: day_048_first_agent/agent
def search_docs(query: str) -> list[dict]:
    meta = lambda text: {k: v.strip('"') for k, v in  # front matter lines look like "title: ..."
                         (ln.split(": ", 1) for ln in text.split("---")[1].strip().splitlines() if ": " in ln)}
    return [{"doc_id": d, "title": meta(_docs()[d]).get("title"), "applies_to": meta(_docs()[d]).get("applies_to")}
            for d in _rank(_docs(), query, 4)]


def get_doc_section(doc_id: str, heading: str | None) -> dict:
    if doc_id not in _docs():
        near = difflib.get_close_matches(doc_id, list(_docs()), n=3)
        raise LookupError(f"Unknown doc_id {doc_id!r}. Closest: {near}. Call search_docs first.")
    text = _docs()[doc_id]
    if heading:  # one "## " section, or the whole page if the heading doesn't exist
        parts = [p for p in text.split("\n## ") if p.lower().startswith(heading.lower())]
        text = "## " + parts[0] if parts else text
    return {"doc_id": doc_id, "text": text[:3000]}


def fn_tool(name: str, description: str, **props: dict) -> dict:  # flat strict schema (Day 36)
    return {"type": "function", "name": name, "description": description, "strict": True,
            "parameters": {"type": "object", "properties": props, "required": list(props),
                           "additionalProperties": False}}


STR, INT, OPT_STR = {"type": "string"}, {"type": "integer"}, {"type": ["string", "null"]}
TOOLS = [
    fn_tool("search_issues", "Keyword search over past Orbit issues; returns id, title, duplicate_of.", query=STR),
    fn_tool("get_issue", "Read one issue's body, labels and first comments. Use ids from search_issues.",
            issue_id=INT),
    fn_tool("search_docs", "Keyword search over Orbit docs; returns doc_id, title, applies_to (2.x/3.x).",
            query=STR),
    fn_tool("get_doc_section", "Read a doc page, or only its '## heading' section when heading is set.",
            doc_id=STR, heading=OPT_STR),
]
REGISTRY = {"search_issues": search_issues, "get_issue": get_issue, "search_docs": search_docs,
            "get_doc_section": get_doc_section}

# script_id: day_048_first_agent/agent
HINTS = {"unknown_tool": "That tool does not exist. Use only the tools you were given.",  # Day 37's kinds
         "bad_args": "The arguments did not match the schema. Fix them using the message.",
         "permanent": "Repeating this exact call will fail again. Fix the arguments or use another tool.",
         "unknown": "An unexpected error. Do not retry; tell the user it failed."}


def dispatch(name: str, arguments: str) -> tuple[str, bool]:
    """Run one call and never raise. Returns (envelope JSON, ok); envelope shape from Day 37."""
    if name not in REGISTRY:
        kind, msg = "unknown_tool", f"No tool {name!r}. Available: {sorted(REGISTRY)}"
    else:
        try:
            return json.dumps({"ok": True, "data": REGISTRY[name](**json.loads(arguments))}, default=str), True
        except (json.JSONDecodeError, TypeError) as e:  # bad JSON or wrong kwargs (before ValueError)
            kind, msg = "bad_args", str(e)
        except (LookupError, ValueError) as e:  # the same call will fail again
            kind, msg = "permanent", str(e)
        except Exception as e:  # the model gets a message it can act on, not a stack trace
            kind, msg = "unknown", f"{type(e).__name__}: {e}"
    err = {"kind": kind, "message": msg[:300], "retryable": False, "hint": HINTS[kind]}
    return json.dumps({"ok": False, "error": err}), False

# script_id: day_048_first_agent/agent
from contextlib import contextmanager

import openai
from openai import OpenAI
from opentelemetry import trace

# USD per 1M input / output tokens; as of 2026-11, verify at developers.openai.com/api/docs/pricing
PRICES = {"gpt-6-luna": (0.10, 0.50)}


def cost_usd(model: str, input_tokens: int, output_tokens: int) -> float:  # built on Day 12
    p_in, p_out = PRICES.get(model, (0.0, 0.0))  # unknown model -> $0, so add its row first
    return (input_tokens * p_in + output_tokens * p_out) / 1_000_000  # ignores cache discounts


@contextmanager
def step_span(name: str, kind: str, **attrs):  # built on Day 42: no-op until setup_tracing()
    with trace.get_tracer("atlas").start_as_current_span(name, attributes={"atlas.kind": kind, **attrs}) as span:
        yield span


@cache
def get_client() -> OpenAI:
    return OpenAI()


def call_model(*, timeout: float, **kwargs):
    # max_retries=1 keeps one step's worst case near 2 x timeout (the SDK default is 2 retries)
    return get_client().with_options(timeout=timeout, max_retries=1).responses.create(**kwargs)

# script_id: day_048_first_agent/agent
AGENT_SYSTEM = """You are Atlas, investigating issues for the maintainers of Orbit, a Python jobs library.
Goal: answer the maintainer's question from tool results; never state facts the tools didn't show.
Constraints: tools are read-only. Prefer docs whose applies_to matches the issue's version (2.x or 3.x).
If a tool returns ok=false, read the hint and change approach instead of repeating the call.
Output contract, once you stop calling tools: three short sections. Found (cite issue #ids and doc ids),
Not found, Likely cause (or "unknown"). "Not found" is a valid, useful answer."""


def run_step(items: list, i: int, tools: list[dict], system: str, config: AgentConfig) -> tuple[StepRecord, str | None]:
    """One step = one model call + every tool call it asks for. Returns (record, final text or None)."""
    rec, t0, final = StepRecord(index=i), time.monotonic(), None
    extra = {"reasoning": {"effort": config.effort}} if config.effort else {}
    with step_span(f"agent.step {i}", "agent_step", **{"atlas.step": i}) as span:
        try:
            resp = call_model(timeout=config.request_timeout, model=MODEL, instructions=system, input=items,
                              tools=tools, max_output_tokens=config.max_output_tokens, **extra)
        except openai.APIError as e:  # timeouts and 5xx after SDK retries: an error step, not a crash
            rec.errors, rec.tool_calls = 1, [f"model_error({type(e).__name__})"]
            resp = None
        if resp is not None:
            rec.input_tokens, rec.output_tokens = resp.usage.input_tokens, resp.usage.output_tokens
            if resp.status == "incomplete":  # e.g. reasoning used every max_output_tokens
                rec.errors = 1
            else:
                items.extend(resp.output)  # calls + reasoning items first, or the next call gets a 400
                calls = [o for o in resp.output if o.type == "function_call"]
                for call in calls:
                    envelope, ok = dispatch(call.name, call.arguments)
                    rec.tool_calls.append(f"{call.name}({call.arguments})")
                    rec.errors += not ok
                    items.append({"type": "function_call_output", "call_id": call.call_id, "output": envelope})
                final = None if calls else resp.output_text
        span.set_attributes({"gen_ai.usage.input_tokens": rec.input_tokens,
                             "gen_ai.usage.output_tokens": rec.output_tokens, "atlas.errors": rec.errors})
    rec.ms = int((time.monotonic() - t0) * 1000)
    return rec, final

# script_id: day_048_first_agent/agent
WRAP_UP = ("You are out of budget and may not call tools. Using only the tool results above, state "
           "what you established (cite ids), what is still missing, and the next thing you would check.")


def wrap_up(items: list, tools: list[dict], system: str, config: AgentConfig, reason: str) -> tuple[str, int, int]:
    try:
        resp = call_model(timeout=config.request_timeout, model=MODEL, instructions=system,
                          input=[*items, {"role": "user", "content": WRAP_UP}], tools=tools,
                          tool_choice="none", max_output_tokens=config.max_output_tokens)
        return f"[partial: {reason}]\n{resp.output_text}", resp.usage.input_tokens, resp.usage.output_tokens
    except openai.APIError:
        return f"[partial: {reason}] The summary call failed too; see steps for what was tried.", 0, 0


def run_agent(task: str, *, tools: list[dict] = TOOLS, system: str = AGENT_SYSTEM,
              config: AgentConfig = AgentConfig()) -> AgentResult:
    items: list = [{"role": "user", "content": task}]
    state, checker, steps, text = RunState(started=time.monotonic()), StopChecker(config), [], ""
    try:
        while (reason := checker.check(state, time.monotonic())) is None:
            rec, final = run_step(items, state.steps + 1, tools, system, config)
            steps.append(rec)
            state.steps, state.tokens = state.steps + 1, state.tokens + rec.input_tokens + rec.output_tokens
            state.calls += rec.tool_calls
            state.consecutive_errors = state.consecutive_errors + 1 if rec.errors else 0  # a clean step resets
            if final is not None:
                text, state.done = final, True
    except KeyboardInterrupt:  # Ctrl-C: no extra API call, hand over what we have
        reason = "interrupted"
        text = f"[partial: interrupted after {len(steps)} steps] Tool calls so far: {state.calls}"
    usage = {"input_tokens": sum(s.input_tokens for s in steps), "output_tokens": sum(s.output_tokens for s in steps)}
    if reason not in ("done", "interrupted"):
        text, w_in, w_out = wrap_up(items, tools, system, config, reason)
        usage["input_tokens"] += w_in
        usage["output_tokens"] += w_out
    return AgentResult(text=text, stop_reason=reason, steps=steps, usage=usage,
                       cost_usd=cost_usd(MODEL, usage["input_tokens"], usage["output_tokens"]),
                       partial=reason != "done")

# script_id: day_048_first_agent/agent
def render(result: AgentResult) -> str:
    rows = ["step | tools | in tok | out tok | ms", "---|---|---|---|---"]
    for s in result.steps:
        names = ", ".join(c.split("(")[0] for c in s.tool_calls) or "(answer)"
        rows.append(f"{s.index} | {names}{' !' * s.errors} | {s.input_tokens} | {s.output_tokens} | {s.ms}")
    rows.append(f"stop_reason={result.stop_reason} partial={result.partial} steps={len(result.steps)} "
                f"cost=${result.cost_usd:.4f}")
    return "\n".join(rows) + "\n\n" + result.text


if __name__ == "__main__":
    import sys

    task = " ".join(sys.argv[1:]) or ("A worker re-runs a long job while it is still running. Is this a "
                                      "known Orbit issue, what do the docs say, and what is the likely cause?")
    print(render(run_agent(task)))

step | tools | in tok | out tok | ms
---|---|---|---|---
1 | search_issues, search_docs | 1180 | 214 | 2410
2 | get_issue, get_doc_section | 3020 | 188 | 2950
3 | search_issues | 4890 | 96 | 1730
4 | (answer) | 5410 | 402 | 3880
stop_reason=done partial=False steps=4 cost=$0.0019

# script_id: day_048_first_agent/test_agent_offline
# Replays scripted model responses through run_agent: real loop and tools, fake model, no key.
# Run: uv run pytest test_agent_offline.py -q  (or: python test_agent_offline.py). Needs agent.py, ORBIT_DATA.
from types import SimpleNamespace as NS

import agent


def call(name: str, args: str, cid: str) -> NS:
    return NS(type="function_call", name=name, arguments=args, call_id=cid)


def reply(*output: NS, text: str = "") -> NS:
    return NS(output=list(output), output_text=text, status="completed",
              usage=NS(input_tokens=1_000, output_tokens=100))


def script(*responses: NS) -> list[dict]:
    queue, seen = list(responses), []
    agent.call_model = lambda **kw: seen.append(kw) or queue.pop(0)  # record each request, replay
    return seen

# script_id: day_048_first_agent/test_agent_offline
def test_done_after_two_tool_steps() -> None:
    script(reply(call("search_docs", '{"query": "lease heartbeat"}', "c1")),
           reply(call("get_doc_section", '{"doc_id": "leases-and-heartbeats", "heading": null}', "c2")),
           reply(NS(type="message"), text="Found: lease expiry (doc leases-and-heartbeats)"))
    r = agent.run_agent("Why does my long job run twice?")
    assert (r.stop_reason, r.partial, len(r.steps)) == ("done", False, 3)
    assert r.usage == {"input_tokens": 3_000, "output_tokens": 300}


def test_budget_stop_returns_a_labelled_partial() -> None:
    seen = script(*[reply(call("search_issues", '{"query": "kafka broker"}', f"c{i}")) for i in range(2)],
                  reply(NS(type="message"), text="Established: no Kafka issues. Missing: the review."))
    r = agent.run_agent("Who approved the Kafka broker?", config=agent.AgentConfig(max_steps=2))
    assert (r.stop_reason, r.partial, len(r.steps)) == ("max_steps", True, 2)
    assert seen[-1]["tool_choice"] == "none" and r.text.startswith("[partial: max_steps]")


def test_repeated_tool_errors_stop_with_max_errors() -> None:
    script(*[reply(call("get_issue", '{"issue_id": 999999}', f"c{i}")) for i in range(3)],
           reply(NS(type="message"), text="Established: issue 999999 does not exist."))
    r = agent.run_agent("Summarise issue 999999", config=agent.AgentConfig(max_errors=3))
    assert r.stop_reason == "max_errors" and [s.errors for s in r.steps] == [1, 1, 1]


if __name__ == "__main__":
    for name, fn in list(globals().items()):
        if name.startswith("test_"):
            fn()
            print("PASS", name)

PASS test_done_after_two_tool_steps
PASS test_budget_stop_returns_a_labelled_partial
PASS test_repeated_tool_errors_stop_with_max_errors

# script_id: day_048_first_agent/agent_set
# Pick the fixed 5-issue agent set (reused Days 49-59), then print scoreboard row D48.
# Env: ORBIT_DATA; OPENAI_API_KEY to run the agent (without a key it only writes agent_set.json).
import json
import os
from pathlib import Path
from statistics import mean

from agent import DATA_DIR, AgentConfig, run_agent

TASK = ("Investigate Orbit issue #{id} ({title!r}): is it a duplicate of an earlier issue, what do "
        "the docs say, and what is the likely cause?")
WANTS = {  # kind -> predicate on (gold record, issue); None = vaguest (shortest body) left
    "duplicate": lambda g, i: g["duplicate_of"] is not None,
    "docs": lambda g, i: "type:docs" in g["labels"],
    "question": lambda g, i: "type:question" in g["labels"],
    "bug with logs": lambda g, i: "type:bug" in g["labels"] and "```" in i["body"],
    "vague": None,
}


def _jsonl(path: Path) -> list[dict]:
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


def pick_agent_set(path: Path | None = Path("agent_set.json")) -> list[str]:
    issues = {i["id"]: i for i in _jsonl(DATA_DIR / "issues" / "issues.jsonl")}
    gold = sorted(_jsonl(DATA_DIR / "eval" / "triage_gold.jsonl"), key=lambda g: g["issue_id"])
    picked: list[dict] = []
    for kind, want in WANTS.items():
        pool = [g for g in gold if g["issue_id"] not in {p["issue_id"] for p in picked}
                and (want is None or want(g, issues[g["issue_id"]]))]
        if want is None:
            pool.sort(key=lambda g: len(issues[g["issue_id"]]["body"]))
        if pool:
            iss = issues[pool[0]["issue_id"]]
            picked.append({"issue_id": iss["id"], "kind": kind, "task": TASK.format(**iss)})
    if path:
        path.write_text(json.dumps(picked, indent=2), encoding="utf-8")
    return [p["task"] for p in picked]

# script_id: day_048_first_agent/agent_set
def scoreboard(tasks: list[str], config: AgentConfig = AgentConfig()) -> list[dict]:
    rows = []
    for task in tasks:
        r = run_agent(task, config=config)
        rows.append({"steps": len(r.steps), "calls": sum(len(s.tool_calls) for s in r.steps),
                     "stop_reason": r.stop_reason, "cost_usd": r.cost_usd, "text": r.text})
    return rows


if __name__ == "__main__":
    tasks = pick_agent_set()
    for t in tasks:
        print(t[:90])
    if not os.environ.get("OPENAI_API_KEY"):
        print("wrote agent_set.json; set OPENAI_API_KEY to run the scoreboard")
        raise SystemExit(0)
    rows = scoreboard(tasks)
    print("\ntask | steps | calls | stop_reason | $")
    for n, r in enumerate(rows, 1):
        print(f"{n} | {r['steps']} | {r['calls']} | {r['stop_reason']} | {r['cost_usd']:.4f}")
    print(f"D48 | mean steps {mean(r['steps'] for r in rows):.1f} | done "
          f"{sum(r['stop_reason'] == 'done' for r in rows)}/5 | ${mean(r['cost_usd'] for r in rows):.4f}/task")

Investigate Orbit issue #22 ('Same job runs on two workers at once (video transcode)'): is
Investigate Orbit issue #204 ('Typo in cli-reference: "Comma-seperated" in the orbit worke
Investigate Orbit issue #10 ('worker not picking up jobs'): is it a duplicate of an earlie
Investigate Orbit issue #13 ('ORB-1002 BrokerAuthError even though the password is correct
Investigate Orbit issue #186 ('Postgres broker: every job waits ~5 seconds before it start
wrote agent_set.json; set OPENAI_API_KEY to run the scoreboard

task | steps | calls | stop_reason | $
1 | 4 | 6 | done | 0.0016
...
D48 | mean steps 4.6 | done 5/5 | $0.0019/task