Handoffs & Agents-as-Tools

# script_id: day_065_handoffs_agents_as_tools/first_handoff
# Install: uv add "openai-agents==0.24.0"   Env: ORBIT_DATA; OPENAI_API_KEY switches to a live run
import json, os
from pathlib import Path

LIVE = bool(os.environ.get("OPENAI_API_KEY"))
if not LIVE:
    os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")  # keyless: nothing to export
from agents import Agent, Runner
from agents.testing import ScriptedModel, assistant_message, function_call

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

def model(*turns):  # live: the real model decides; keyless: replay the turns we expect
    return MODEL if LIVE else ScriptedModel(list(turns))

bug = Agent(name="Bug specialist", handoff_description="Crashes, ORB- error codes, wrong behaviour.",
            instructions="Diagnose the Orbit bug in two sentences.",
            model=model([assistant_message("Likely a duplicate of #4: the 300 s lease expires during "
                                           "the long transcode, so a second worker takes the job (ORB-1101).")]))
question = Agent(name="Question specialist", handoff_description="How-to and usage questions.",
                 instructions="Answer the Orbit question in two sentences.",
                 model=model([assistant_message("Run one worker process per queue.")]))
triage = Agent(name="Triage", instructions="Transfer the issue to exactly one specialist.",
               handoffs=[bug, question],
               model=model([function_call("transfer_to_bug_specialist", {}, call_id="h1")]))

issues = {r["id"]: r for r in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open())}
issue = issues[22]
result = Runner.run_sync(triage, f"Issue #22: {issue['title']}\n\n{issue['body']}", max_turns=6)
print("routed to:", result.last_agent.name)
print(result.final_output)

routed to: Bug specialist
Likely a duplicate of #4: the 300 s lease expires during the long transcode, so a second worker takes the job (ORB-1101).

# script_id: day_065_handoffs_agents_as_tools/break_it
# Install: uv add "openai-agents==0.24.0" pydantic   Env: none (keyless)
import os
os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")
from pydantic import BaseModel
from agents import Agent, MaxTurnsExceeded, Runner, handoff
from agents.exceptions import UserError
from agents.testing import ScriptedModel, function_call

# 1. The handoff is a tool, and its description is the only routing signal the model gets.
bare = Agent(name="Bug specialist", instructions="Diagnose bugs.")
described = Agent(name="Bug specialist", instructions="Diagnose bugs.",
                  handoff_description="Crashes, ORB- error codes, regressions, wrong behaviour.")
for agent in (bare, described):
    h = handoff(agent)
    print(f"{h.tool_name}: {h.tool_description!r}")

# 2. input_type needs an on_handoff callback, and it does NOT choose the destination.
class HandoffNote(BaseModel):
    suspected_type: str
try:
    handoff(described, input_type=HandoffNote)
except UserError as e:
    print("UserError:", e)
note = lambda ctx, n: None  # a no-op satisfies the SDK; the full team logs the note
triage = Agent(name="Triage", handoffs=[
    handoff(described, input_type=HandoffNote, on_handoff=note),
    handoff(Agent(name="Question specialist"), input_type=HandoffNote, on_handoff=note)])
print([h.tool_name for h in triage.handoffs])  # one tool per destination

# 3. A specialist that can hand back to triage can ping-pong until max_turns runs out.
ping = ScriptedModel([[function_call("transfer_to_bug_specialist", {}, call_id=f"t{i}")] for i in range(9)])
pong = ScriptedModel([[function_call("transfer_to_triage", {}, call_id=f"b{i}")] for i in range(9)])
front = Agent(name="Triage", instructions="Route.", model=ping)
bug = Agent(name="Bug specialist", instructions="Diagnose.", model=pong, handoffs=[front])
front.handoffs = [bug]
try:
    Runner.run_sync(front, "Issue #171: ORB-1001 warnings during a failover drill. Expected?", max_turns=6)
except MaxTurnsExceeded as e:
    print(f"{type(e).__name__}: {e}")

transfer_to_bug_specialist: 'Handoff to the Bug specialist agent to handle the request. '
transfer_to_bug_specialist: 'Handoff to the Bug specialist agent to handle the request. Crashes, ORB- error codes, regressions, wrong behaviour.'
UserError: You must provide on_handoff when input_type is provided
['transfer_to_bug_specialist', 'transfer_to_question_specialist']
MaxTurnsExceeded: Max turns (6) exceeded

# script_id: day_065_handoffs_agents_as_tools/handoff_contract
# Install: uv add "openai-agents==0.24.0" pydantic   Env: none (keyless)
import os
os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")
from typing import Literal
from pydantic import BaseModel
from agents import Agent, ModelBehaviorError, Runner, handoff
from agents.testing import ScriptedModel, assistant_message, function_call

class HandoffNote(BaseModel):  # what triage must say when it transfers an issue
    issue_id: int
    suspected_type: Literal["bug", "question", "feature", "docs"]
    reason: str

class TriageDecision(BaseModel):  # what the owner returns; mirrors eval/triage_gold.jsonl
    labels: list[str]
    duplicate_of: int | None
    needs_human: bool
    reply_points: list[str]

def log_note(ctx, note: HandoffNote) -> None:  # runs at the moment of transfer; sync is fine
    print(f"handoff note: #{note.issue_id} as {note.suspected_type} ({note.reason})")

def run(note_args: dict) -> None:
    answer = TriageDecision(labels=["type:question", "comp:broker-redis", "version:3.1"], duplicate_of=None,
                            needs_human=False, reply_points=["A few seconds of ORB-1001 is expected on 3.1.2"])
    question = Agent(name="Question specialist", instructions="Answer.", output_type=TriageDecision,
                     model=ScriptedModel([[assistant_message(answer.model_dump_json())]]))
    triage = Agent(name="Triage", instructions="Transfer to one specialist.",
                   handoffs=[handoff(question, input_type=HandoffNote, on_handoff=log_note)],
                   model=ScriptedModel([[function_call("transfer_to_question_specialist", note_args, call_id="h1")]]))
    try:
        result = Runner.run_sync(triage, "Issue #171: ORB-1001 warnings during a failover drill. Expected?")
        print(result.last_agent.name, "->", result.final_output)
    except ModelBehaviorError as e:  # the contract rejected the note before the specialist ran
        print("rejected:", e)

run({"issue_id": 171, "suspected_type": "question", "reason": "asks whether the warnings are normal"})
run({"issue_id": 171, "suspected_type": "incident", "reason": "not one of the allowed types"})

handoff note: #171 as question (asks whether the warnings are normal)
Question specialist -> labels=['type:question', 'comp:broker-redis', 'version:3.1'] duplicate_of=None needs_human=False reply_points=['A few seconds of ORB-1001 is expected on 3.1.2']
rejected: Invalid JSON when parsing model output

# script_id: day_065_handoffs_agents_as_tools/agent_as_tool
# Install: uv add "openai-agents==0.24.0" pydantic   Env: ORBIT_DATA (keyless)
import json, os, re
from pathlib import Path
from typing import Literal
os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")
from pydantic import BaseModel
from agents import Agent, Runner, function_tool
from agents.testing import ScriptedModel, assistant_message, function_call

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
words = lambda text: set(re.findall(r"[a-z0-9-]{3,}", text.lower()))

class ResearchRequest(BaseModel):  # the consult's contract
    question: str
    scope: Literal["docs", "issues"]

@function_tool
def search_docs(query: str) -> str:
    """Keyword search over the Orbit docs. Returns the best doc id and its opening lines."""
    best = max((DATA_DIR / "docs").glob("*.md"), key=lambda p: len(words(query) & words(p.read_text())))
    return f"{best.stem}: {best.read_text()[:200]}"

researcher_model = ScriptedModel([
    [function_call("search_docs", {"query": "redis failover reconnect ORB-1001"}, call_id="s1")],
    [assistant_message("troubleshooting-broker: the 3.1.0 reconnect storm is fixed; a short burst is normal.")]])
researcher = Agent(name="Docs researcher", instructions="Search the docs, answer in 2 lines, cite doc ids.",
                   tools=[search_docs], model=researcher_model)
research_docs = researcher.as_tool(tool_name="research_docs", tool_description="Ask the docs researcher one question.",
                                   parameters=ResearchRequest, max_turns=4)  # the nested run's own budget

question_model = ScriptedModel([
    [function_call("research_docs", {"question": "Is a burst of ORB-1001 during failover normal on 3.1.2?",
                                     "scope": "docs"}, call_id="r1")],
    [assistant_message("Yes: a few seconds of ORB-1001 is expected on 3.1.2 (see troubleshooting-broker).")]])
question = Agent(name="Question specialist", instructions="Consult research_docs once, then answer.",
                 tools=[research_docs], model=question_model)

issue = next(r for r in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open()) if r["id"] == 171)
result = Runner.run_sync(question, f"Issue #171: {issue['title']}\n\n{issue['body']}", max_turns=6)
print("last speaker:", result.last_agent.name)  # the caller still owns the answer
print("specialist input chars:", len(json.dumps(question_model.first_call.input)))
print("researcher saw:", researcher_model.first_call.input)  # the request only, not the issue thread
print(result.final_output)

last speaker: Question specialist
specialist input chars: 1486
researcher saw: [{'content': '{"question": "Is a burst of ORB-1001 during failover normal on 3.1.2?", "scope": "docs"}', 'role': 'user'}]
Yes: a few seconds of ORB-1001 is expected on 3.1.2 (see troubleshooting-broker).

# script_id: day_065_handoffs_agents_as_tools/context_filter
# Install: uv add "openai-agents==0.24.0"   Env: ORBIT_DATA (keyless)
import json, os
from pathlib import Path
os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")
from agents import Agent, Runner, function_tool, handoff
from agents.extensions import handoff_filters
from agents.testing import ScriptedModel, assistant_message, function_call

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
ISSUES = {r["id"]: r for r in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open())}

@function_tool
def get_issue(issue_id: int) -> str:
    """Return one Orbit issue as JSON (title, body, comments)."""
    return json.dumps(ISSUES[issue_id])

def specialist_view(input_filter) -> list:
    bug_model = ScriptedModel([[assistant_message("Duplicate of #4.")]])
    bug = Agent(name="Bug specialist", instructions="Diagnose.", model=bug_model)
    triage = Agent(name="Triage", instructions="Route.", tools=[get_issue],
                   handoffs=[handoff(bug, input_filter=input_filter)],
                   model=ScriptedModel([[function_call("get_issue", {"issue_id": 4}, call_id="g1")],
                                        [function_call("transfer_to_bug_specialist", {}, call_id="h1")]]))
    issue = ISSUES[22]
    Runner.run_sync(triage, f"Issue #22: {issue['title']}\n\n{issue['body']}", max_turns=6)
    return bug_model.first_call.input  # exactly what the specialist's first model call received

for name, f in (("no filter", None), ("remove_all_tools", handoff_filters.remove_all_tools)):
    items = specialist_view(f)
    kinds = [i.get("type") or i.get("role") for i in items]
    print(f"{name:17} items={len(items)} ~tokens={len(json.dumps(items, default=str)) // 4:>5}  {kinds}")

no filter         items=5 ~tokens= 1230  ['user', 'function_call', 'function_call_output', 'function_call', 'function_call_output']
remove_all_tools  items=1 ~tokens=  150  ['user']

# script_id: day_065_handoffs_agents_as_tools/boundary_guardrails
# Install: uv add "openai-agents==0.24.0" pydantic   Env: ORBIT_DATA (keyless)
import json, os, re
from pathlib import Path
os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")
from pydantic import BaseModel
from agents import (Agent, GuardrailFunctionOutput, InputGuardrailTripwireTriggered,
                    OutputGuardrailTripwireTriggered, Runner, input_guardrail, output_guardrail)
from agents.testing import ScriptedModel, assistant_message, function_call

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
ISSUES = {r["id"]: r for r in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open())}
KNOWN_LABELS = {lab for r in ISSUES.values() for lab in r["labels"]}  # Orbit's real label set
INJECTION = re.compile(r"ignore (all |your )?previous instructions|automated triage|ai assistant", re.I)

class TriageDecision(BaseModel):
    labels: list[str]
    needs_human: bool

@input_guardrail(run_in_parallel=False)  # finish the check BEFORE triage's model call: a blocked run costs $0
async def injection_tripwire(ctx, agent, user_input) -> GuardrailFunctionOutput:
    return GuardrailFunctionOutput(output_info=None, tripwire_triggered=bool(INJECTION.search(str(user_input))))

@output_guardrail  # runs on whichever agent produces the final output
async def label_allow_list(ctx, agent, output: TriageDecision) -> GuardrailFunctionOutput:
    unknown = [lab for lab in output.labels if lab not in KNOWN_LABELS]
    return GuardrailFunctionOutput(output_info=unknown, tripwire_triggered=bool(unknown))

def triage(issue_id: int, labels: list[str]) -> TriageDecision:
    decision = TriageDecision(labels=labels, needs_human=False).model_dump_json()
    bug = Agent(name="Bug specialist", instructions="Label the bug.", output_type=TriageDecision,
                output_guardrails=[label_allow_list], model=ScriptedModel([[assistant_message(decision)]]))
    front = Agent(name="Triage", instructions="Route.", handoffs=[bug], input_guardrails=[injection_tripwire],
                  model=ScriptedModel([[function_call("transfer_to_bug_specialist", {}, call_id="h1")]]))
    try:
        return Runner.run_sync(front, ISSUES[issue_id]["body"], max_turns=4).final_output
    except (InputGuardrailTripwireTriggered, OutputGuardrailTripwireTriggered) as trip:
        print(f"  #{issue_id}: {type(trip).__name__} -> needs_human")  # a decision, not a crash
        return TriageDecision(labels=[], needs_human=True)

gold = [json.loads(line) for line in (DATA_DIR / "eval" / "triage_gold.jsonl").open()]
flagged = [g["issue_id"] for g in gold if triage(g["issue_id"], ["type:bug"]).needs_human]
print("input tripwire fired on:", flagged)
print("invented label:", triage(22, ["type:bug", "severity:S0"]))

  #20: InputGuardrailTripwireTriggered -> needs_human
  #46: InputGuardrailTripwireTriggered -> needs_human
  #92: InputGuardrailTripwireTriggered -> needs_human
  #159: InputGuardrailTripwireTriggered -> needs_human
  #180: InputGuardrailTripwireTriggered -> needs_human
  #204: InputGuardrailTripwireTriggered -> needs_human
input tripwire fired on: [20, 46, 92, 159, 180, 204]
  #22: OutputGuardrailTripwireTriggered -> needs_human
invented label: labels=[] needs_human=True

# script_id: day_065_handoffs_agents_as_tools/triage_team
# atlas/triage_agents.py: Day 41's router rebuilt as triage -> specialist handoffs (openai-agents 0.24.0).
# Install: uv add "openai-agents==0.24.0" pydantic   Env: ORBIT_DATA; OPENAI_API_KEY for a live run;
#   optional ATLAS_MODEL_<ROLE> (TRIAGE, BUG, QUESTION, FEATURE, RESEARCHER), ATLAS_LIMIT (default 30).
# Services: none. Read-only: decisions are returned, never posted.
import asyncio, json, logging, os, re, statistics, time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Literal
from pydantic import BaseModel

LIVE = bool(os.environ.get("OPENAI_API_KEY"))
if not LIVE:
    os.environ.setdefault("OPENAI_AGENTS_DISABLE_TRACING", "1")  # keyless: nothing to export
from agents import (Agent, FunctionTool, GuardrailFunctionOutput, InputGuardrailTripwireTriggered,
                    ModelSettings, OutputGuardrailTripwireTriggered, RunHooks, Runner, Usage,
                    function_tool, handoff, input_guardrail, output_guardrail, trace)
from agents.extensions import handoff_filters
from agents.models.interface import Model
from agents.testing import ModelStep, ScriptedModel, assistant_message, function_call
from openai.types.shared import Reasoning

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
# USD per 1M tokens (input, output), as of 2026-11; verify at https://developers.openai.com/api/docs/pricing
PRICES = {"gpt-6-luna": (0.10, 0.50), "gpt-6.1-terra": (1.00, 5.00), "gpt-6.1-sol": (2.00, 10.00)}
SETTINGS = ModelSettings(reasoning=Reasoning(effort="low"))  # explicit: the SDK's defaults predate GPT-6
ISSUES = {r["id"]: r for r in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open())}
KNOWN_LABELS = {lab for r in ISSUES.values() for lab in r["labels"]}
log = logging.getLogger("atlas.triage")

class TriageDecision(BaseModel):  # mirrors eval/triage_gold.jsonl; Day 76 reuses it
    labels: list[str]
    duplicate_of: int | None
    needs_human: bool
    reply_points: list[str]

class HandoffNote(BaseModel):
    issue_id: int
    suspected_type: Literal["bug", "question", "feature", "docs"]
    reason: str

class ResearchRequest(BaseModel):
    question: str
    scope: Literal["docs", "issues"]

# script_id: day_065_handoffs_agents_as_tools/triage_team
def _words(text: str) -> set[str]:
    return set(re.findall(r"[a-z0-9-]{3,}", text.lower()))

@function_tool
def search_issues(query: str) -> str:
    """Keyword search over Orbit issue titles. Returns up to 3 lines: '#id (score) title'."""
    q = _words(query)  # built on Day 35: keyword overlap, no index needed
    hits = sorted(((len(q & _words(r["title"])) / (len(q) or 1), r) for r in ISSUES.values()),
                  key=lambda h: -h[0])[:3]
    return "\n".join(f"#{r['id']} ({s:.2f}) {r['title']}" for s, r in hits)

@function_tool
def search_docs(query: str) -> str:
    """Keyword search over the Orbit docs. Returns the best doc id and its first 400 characters."""
    q = _words(query)  # built on Day 27: the keyword half of hybrid search
    best = max((DATA_DIR / "docs").glob("*.md"), key=lambda p: len(q & _words(p.read_text())))
    return f"{best.stem}: {best.read_text()[:400]}"

def consult(scope: str, model: str | Model, hooks: RunHooks | None) -> FunctionTool:
    """A researcher agent wrapped as a tool: own context, own turn budget; the caller keeps control."""
    researcher = Agent(name=f"{scope.title()} researcher", model=model, model_settings=SETTINGS,
                       tools=[search_issues if scope == "issues" else search_docs],
                       instructions="Search once or twice, answer in 3 lines, cite ids. Say 'none' if unsure.")
    return researcher.as_tool(tool_name=f"research_{scope}", parameters=ResearchRequest, max_turns=4,
                              tool_description=f"Ask the {scope} researcher one question.", hooks=hooks)

# script_id: day_065_handoffs_agents_as_tools/triage_team
INJECTION = re.compile(r"ignore (all |your )?previous instructions|automated triage|ai assistant", re.I)

@input_guardrail(run_in_parallel=False)  # ingress: first agent only, before its model call
async def injection_tripwire(ctx, agent, user_input) -> GuardrailFunctionOutput:
    hits = INJECTION.findall(str(user_input))  # placement demo; real defences are Day 69
    return GuardrailFunctionOutput(output_info={"markers": len(hits)}, tripwire_triggered=bool(hits))

@output_guardrail  # egress: whichever agent produces the final output
async def label_allow_list(ctx, agent, output: TriageDecision) -> GuardrailFunctionOutput:
    unknown = [lab for lab in output.labels if lab not in KNOWN_LABELS]
    return GuardrailFunctionOutput(output_info={"unknown": unknown}, tripwire_triggered=bool(unknown))

SPECIALISTS = {  # kind -> (handoff_description, research scope)
    "bug": ("Crashes, ORB- error codes, regressions, wrong behaviour.", "issues"),
    "question": ("How-to and usage questions, including 'is this expected?'.", "docs"),
    "feature": ("Requests for new options, commands or behaviour.", "docs"),
}

def log_note(ctx, note: HandoffNote) -> None:
    log.info("handoff #%s as %s: %s", note.issue_id, note.suspected_type, note.reason)

def build_triage_team(model_for: Callable[[str], str | Model], hooks: RunHooks | None = None,
                      input_filter=handoff_filters.remove_all_tools) -> Agent:
    handoffs = []
    for kind, (description, scope) in SPECIALISTS.items():
        owner = Agent(name=f"{kind.title()} specialist", handoff_description=description,
                      model=model_for(kind), model_settings=SETTINGS, output_type=TriageDecision,
                      tools=[consult(scope, model_for("researcher"), hooks)], output_guardrails=[label_allow_list],
                      instructions=f"You own {kind} issues. Call research_{scope} once, then return a "
                                   "TriageDecision using only Orbit labels (type:*, comp:*, severity:*, version:*).")
        handoffs.append(handoff(owner, input_type=HandoffNote, on_handoff=log_note, input_filter=input_filter))
    return Agent(name="Triage", handoffs=handoffs, model=model_for("triage"), model_settings=SETTINGS,
                 input_guardrails=[injection_tripwire],
                 instructions="Read the Orbit issue and transfer it to exactly one specialist. Never answer it.")

# script_id: day_065_handoffs_agents_as_tools/triage_team
class PerAgentUsage(RunHooks):
    """Attributes every model call's usage to the agent that made it."""
    def __init__(self) -> None:
        self.by_agent: dict[str, Usage] = {}
        self.model_of: dict[str, str] = {}
    async def on_llm_end(self, context, agent, response) -> None:
        self.by_agent.setdefault(agent.name, Usage()).add(response.usage)
        self.model_of[agent.name] = agent.model if isinstance(agent.model, str) else MODEL

@dataclass
class RunStats:  # built on Day 64 (compact): one row per issue per config
    decision: TriageDecision
    hooks: PerAgentUsage = field(default_factory=PerAgentUsage)
    seconds: float = 0.0
    @property
    def calls(self) -> int:
        return sum(u.requests for u in self.hooks.by_agent.values())
    @property
    def cost(self) -> float:
        def usd(name: str, u: Usage) -> float:
            p_in, p_out = PRICES.get(self.hooks.model_of[name], PRICES["gpt-6-luna"])
            return (u.input_tokens * p_in + u.output_tokens * p_out) / 1e6
        return sum(usd(name, u) for name, u in self.hooks.by_agent.items())

def issue_text(issue: dict) -> str:
    return f"Issue #{issue['id']}: {issue['title']}\n\n{issue['body']}"

async def triage_with_agents(issue: dict, model_for=None, build=None) -> tuple[TriageDecision, RunStats]:
    hooks, start = PerAgentUsage(), time.perf_counter()
    team = (build or build_triage_team)(model_for or default_model_for, hooks)  # swap the topology here
    try:
        with trace("atlas-triage", group_id=f"orbit-issue-{issue['id']}"):
            run = await Runner.run(team, issue_text(issue), hooks=hooks, max_turns=8)
        decision = run.final_output_as(TriageDecision)
    except (InputGuardrailTripwireTriggered, OutputGuardrailTripwireTriggered) as trip:
        decision = TriageDecision(labels=[], duplicate_of=None, needs_human=True,
                                  reply_points=[f"blocked by {type(trip).__name__}"])
    return decision, RunStats(decision, hooks, time.perf_counter() - start)

# script_id: day_065_handoffs_agents_as_tools/triage_team
async def triage_with_workflow(issue: dict, model_for=None) -> tuple[TriageDecision, RunStats]:
    # built on Day 41 (compact): ONE structured call decides; no tools, no handoffs.
    hooks, start = PerAgentUsage(), time.perf_counter()
    router = Agent(name="Router", model=(model_for or default_model_for)("triage"), model_settings=SETTINGS,
                   output_type=TriageDecision, input_guardrails=[injection_tripwire],
                   instructions="Triage this Orbit issue into a TriageDecision using only Orbit labels.")
    try:
        decision = (await Runner.run(router, issue_text(issue), hooks=hooks, max_turns=2)).final_output
    except InputGuardrailTripwireTriggered:
        decision = TriageDecision(labels=[], duplicate_of=None, needs_human=True, reply_points=["blocked"])
    return decision, RunStats(decision, hooks, time.perf_counter() - start)

def score(gold: dict, d: TriageDecision) -> dict:
    kind = lambda labels: next((x for x in labels if x.startswith("type:")), None)
    g = {x for x in gold["labels"] if x.startswith("comp:")}
    p = {x for x in d.labels if x.startswith("comp:")}
    return {"type": kind(gold["labels"]) == kind(d.labels), "human": gold["needs_human"] == d.needs_human,
            "comp_f1": 2 * len(g & p) / (len(g) + len(p)) if g or p else 1.0,
            "dup": gold["duplicate_of"] == d.duplicate_of}

async def delta_report(limit: int = 30, concurrency: int = 4) -> None:
    gold = [json.loads(line) for line in (DATA_DIR / "eval" / "triage_gold.jsonl").open()][:limit]
    sem = asyncio.Semaphore(concurrency)
    async def one(fn, g):
        async with sem:
            return g, *(await fn(ISSUES[g["issue_id"]]))
    print("config       | type acc | comp F1 | human acc | dup acc | $/issue  | p50 s | calls/issue")
    for name, fn in (("d41_workflow", triage_with_workflow), ("handoffs", triage_with_agents)):
        rows = await asyncio.gather(*(one(fn, g) for g in gold))
        s, st = [score(g, d) for g, d, _ in rows], [r for _, _, r in rows]
        avg = lambda k: sum(x[k] for x in s) / len(s)
        print(f"{name:12} | {avg('type'):8.2f} | {avg('comp_f1'):7.2f} | {avg('human'):9.2f} | {avg('dup'):7.2f} |"
              f" {sum(r.cost for r in st) / len(st):.6f} | {statistics.median(r.seconds for r in st):5.2f} |"
              f" {sum(r.calls for r in st) / len(st):.1f}")

# script_id: day_065_handoffs_agents_as_tools/triage_team
def stand_in(call) -> ModelStep:  # keyless only: a keyword rule pretending to be a model
    text = call.input if isinstance(call.input, str) else json.dumps(call.input, default=str)
    m = re.search(r"Issue #(\d+): ([^\n\\]*)", text)
    iid, title = (int(m.group(1)), m.group(2)) if m else (0, text[:80])
    kind = "bug" if re.search(r"ORB-\d{4}|Traceback|crash|fails", text) else "question" if "?" in title else "feature"
    results = re.findall(r'"output": "([^"]*)"', text)
    if call.handoffs:  # triage: transfer with a filled-in HandoffNote
        out = function_call(f"transfer_to_{kind}_specialist", {"issue_id": iid, "suspected_type": kind,
                                                               "reason": "keyword rule"}, call_id="h1")
    elif call.tools and not results:  # first turn with tools: use the first one
        props = call.tools[0].params_json_schema["properties"]
        args = {"question": title, "scope": "issues"} if "question" in props else {"query": title}
        out = function_call(call.tools[0].name, args, call_id="t1")
    elif call.output_schema is not None:  # an owner: return a decision
        dup = re.search(r"#(\d+) \((0\.[6-9]\d|1\.00)\)", results[-1] if results else "")
        dup_id = int(dup.group(1)) if dup and int(dup.group(1)) != iid else None
        out = assistant_message(TriageDecision(labels=[f"type:{kind}"], duplicate_of=dup_id,
                                               needs_human=False, reply_points=[title]).model_dump_json())
    else:  # a researcher: pass the tool result up
        out = assistant_message(results[-1][:300] if results else "none")
    n_in = len(text) // 4  # rough token estimate
    return ModelStep(output=[out], usage=Usage(requests=1, input_tokens=n_in, output_tokens=40,
                                               total_tokens=n_in + 40))

def default_model_for(role: str) -> str | Model:
    if LIVE:
        return os.environ.get(f"ATLAS_MODEL_{role.upper()}", MODEL)
    return ScriptedModel([ModelStep.respond(stand_in)] * 8)  # fresh fake per agent per run

if __name__ == "__main__":
    decision, stats = asyncio.run(triage_with_agents(ISSUES[171]))
    print(decision)
    for agent, u in stats.hooks.by_agent.items():
        print(f"{agent:20} calls={u.requests}  in={u.input_tokens:>5}  out={u.output_tokens:>4}")
    print("mode:", "live" if LIVE else "keyless stand-in: quality columns score a keyword rule, not a model")
    asyncio.run(delta_report(int(os.environ.get("ATLAS_LIMIT", "30"))))

labels=['type:bug'] duplicate_of=None needs_human=False reply_points=['After upgrading to 3.1.2: ~10 s of ORB-1001 warnings during a failover drill, is that expected?']
Triage               calls=1  in=  371  out=  40
Bug specialist       calls=2  in=  893  out=  80
Issues researcher    calls=2  in=  237  out=  80
mode: keyless stand-in: quality columns score a keyword rule, not a model
config       | type acc | comp F1 | human acc | dup acc | $/issue  | p50 s | calls/issue
d41_workflow |     0.53 |    0.00 |      0.67 |    0.63 | 0.000113 |  0.01 | 0.8
handoffs     |     0.53 |    0.00 |      0.67 |    0.63 | 0.000398 |  0.10 | 4.0