Debate, Consensus & Multi-Agent Operations

# script_id: day_067_debate_multi_agent_ops/orbit_dupes
# orbit_dupes.py: 20 duplicate-decision items from Orbit (12 real duplicates + 8 "none").
# Install: nothing (stdlib only) · Env: ORBIT_DATA (default datasets/orbit) · Services: none
from __future__ import annotations
import os, sqlite3
from dataclasses import dataclass
from pathlib import Path

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

@dataclass(frozen=True)
class Candidate:
    id: str
    text: str

@dataclass(frozen=True)
class DupItem:
    issue_id: int
    text: str
    candidates: tuple[Candidate, ...]
    gold: str                                  # a candidate id, or "none"

    @property
    def options(self) -> list[str]:
        return [c.id for c in self.candidates] + ["none"]

# Near-misses: EARLIER issues sharing a component label, outside the true duplicate cluster.
NEAR_MISS_SQL = """
SELECT c.id FROM issue_labels a
JOIN issue_labels b ON b.label_id = a.label_id
JOIN labels l ON l.id = a.label_id
JOIN issues c ON c.id = b.issue_id
WHERE a.issue_id = :x AND c.id < :x AND c.id != :gold
  AND (c.duplicate_of IS NULL OR c.duplicate_of != :gold)
GROUP BY c.id HAVING SUM(l.kind = 'comp') > 0
ORDER BY COUNT(*) DESC, c.id DESC LIMIT :k
"""

# script_id: day_067_debate_multi_agent_ops/orbit_dupes
def _text(con: sqlite3.Connection, issue_id: int, limit: int = 600) -> str:
    title, body = con.execute("SELECT title, body FROM issues WHERE id = ?", (issue_id,)).fetchone()
    return f"#{issue_id} {title}\n{body[:limit]}"

def _item(con: sqlite3.Connection, x: int, gold: int | None) -> DupItem | None:
    k = 2 if gold else 3
    near = [r[0] for r in con.execute(NEAR_MISS_SQL, {"x": x, "gold": gold or -1, "k": k})]
    if len(near) < k:
        return None                            # not enough near-misses: skip this issue
    ids = sorted(near + ([gold] if gold else []))
    cands = tuple(Candidate(str(i), _text(con, i, 400)) for i in ids)
    return DupItem(x, _text(con, x), cands, str(gold) if gold else "none")

def load_items(n_dup: int = 12, n_none: int = 8) -> list[DupItem]:
    con = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)      # read-only: P7 tools never write
    firsts = con.execute("SELECT MIN(id), duplicate_of FROM issues WHERE duplicate_of IS NOT NULL "
                         "GROUP BY duplicate_of ORDER BY duplicate_of").fetchall()
    roots = {g for _, g in firsts}
    singles = con.execute("SELECT id FROM issues WHERE duplicate_of IS NULL AND id >= 100 "
                          "ORDER BY id").fetchall()
    items = [it for x, g in firsts if (it := _item(con, x, g))][:n_dup]
    items += [it for (x,) in singles if x not in roots and (it := _item(con, x, None))][:n_none]
    con.close()
    return items

if __name__ == "__main__":
    items = load_items()
    print(len(items), "items,", sum(i.gold != "none" for i in items), "real duplicates")
    first = items[0]
    print(f"issue #{first.issue_id}: options={first.options} gold={first.gold}")

20 items, 12 real duplicates
issue #22: options=['4', '15', '16', 'none'] gold=4

# script_id: day_067_debate_multi_agent_ops/vote_by_hand
# Majority-of-3 by hand: independent vs correlated voters, then round-2 flips. Stdlib only.
from collections import Counter
from math import comb

def majority(votes: list[str]) -> str:
    return Counter(votes).most_common(1)[0][0]

def p_majority(p: float, n: int = 3) -> float:
    # P(more than half of n independent voters are right), each right with probability p
    return sum(comb(n, k) * p**k * (1 - p) ** (n - k) for k in range(n // 2 + 1, n + 1))

print("votes ['12', '12', '40'] ->", majority(["12", "12", "40"]))
p = 0.7
for rho in (0.0, 0.5, 1.0):   # rho = share of items where the 3 voters act as ONE voter
    print(f"correlation {rho:.1f}: majority accuracy {rho * p + (1 - rho) * p_majority(p):.3f}")

gold = "12"
for r1, r2 in ((["12", "40", "40"], ["12", "12", "40"]), (["12", "12", "40"], ["40", "12", "40"])):
    good = sum(a != gold and b == gold for a, b in zip(r1, r2))
    bad = sum(a == gold and b != gold for a, b in zip(r1, r2))
    print(f"{r1} -> {r2}: flips +{good} / -{bad}, majority {majority(r1)} -> {majority(r2)}")

votes ['12', '12', '40'] -> 12
correlation 0.0: majority accuracy 0.784
correlation 0.5: majority accuracy 0.742
correlation 1.0: majority accuracy 0.700
['12', '40', '40'] -> ['12', '12', '40']: flips +1 / -0, majority 40 -> 12
['12', '12', '40'] -> ['40', '12', '40']: flips +0 / -1, majority 12 -> 40

# script_id: day_067_debate_multi_agent_ops/vote_minimal
# Recap of Day 41 voting: 3 independent votes + majority. The baseline debate must beat.
# Install: uv add "openai>=3.46,<4" pydantic · Env: OPENAI_API_KEY, ORBIT_DATA · Needs orbit_dupes.py
import os
from collections import Counter
from typing import Literal
from openai import BadRequestError, OpenAI
from pydantic import BaseModel, create_model
from orbit_dupes import Candidate, load_items

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
client = OpenAI()

class Verdict(BaseModel):
    answer: str
    reason: str
    confidence: Literal["low", "med", "high"]

item = load_items()[0]
ItemVerdict = create_model("ItemVerdict", __base__=Verdict, answer=(Literal[tuple(item.options)], ...))

def vote(order: list[Candidate]) -> str:
    cands = "\n\n".join(f"[{c.id}] {c.text}" for c in order)
    r = client.responses.parse(model=MODEL, reasoning={"effort": "low"}, text_format=ItemVerdict,
        instructions="Is the NEW issue a duplicate of one CANDIDATE? Answer a candidate id or none.",
        input=f"NEW:\n{item.text}\n\nCANDIDATES:\n{cands}")
    return r.output_parsed.answer if r.output_parsed else "none"   # None = refusal/incomplete

c = list(item.candidates)
votes = [vote(c[i:] + c[:i]) for i in range(3)]       # rotate the evidence order per voter
print(f"votes: {votes} -> {Counter(votes).most_common(1)[0][0]} (gold {item.gold})")

votes: ['4', '4', '16'] -> 4 (gold 4)     <- illustrative; your votes will differ

# script_id: day_067_debate_multi_agent_ops/vote_minimal
same = [vote(c) for _ in range(3)]                                   # (a) no diversity at all
print("identical prompts:", same, "agreement", max(Counter(same).values()), "/ 3")
try:                                                                 # (b) temperature + effort
    client.responses.create(model=MODEL, input="hi", reasoning={"effort": "low"}, temperature=1.2)
except BadRequestError as e:
    print("rejected:", e.status_code, str(e)[:160])
else:                                                                # the rule may change: check
    print("accepted: temperature + effort 'low' was not rejected on", MODEL)

identical prompts: ['4', '4', '4'] agreement 3 / 3          <- illustrative
rejected: 400 ...                                           <- expected; the exact text is unverified

# script_id: day_067_debate_multi_agent_ops/atlas/llm
# atlas/llm.py: per-agent $ budgets with a hard stop, plus one OTel span per agent per round.
# add these to your atlas/llm.py from Day 12
# Install: uv add opentelemetry-sdk · Env: none · Services: none (prints one line per span)
# Compact self-contained copy of Day 12's prices and Day 42's setup_tracing so it runs on its own.
from __future__ import annotations
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Any
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter, SpanExportResult

# USD per 1M tokens (input, output), standard tier, as of 2026-11; verify at the pricing page
PRICES = {"gpt-6-luna": (0.10, 0.50), "gpt-6.1-terra": (1.00, 5.00)}

def cost_usd(model: str, usage: Any) -> float:
    # Reasoning tokens are billed as output tokens; cached-input discounts are ignored here
    p_in, p_out = PRICES[model]
    return (usage.input_tokens * p_in + usage.output_tokens * p_out) / 1e6

def estimate_usd(model: str, input_tokens: int, max_output_tokens: int) -> float:
    # Worst case for one call: every allowed output token gets used
    p_in, p_out = PRICES[model]
    return (input_tokens * p_in + max_output_tokens * p_out) / 1e6

class BudgetExceeded(Exception):
    """Raised BEFORE a call that would break a per-agent or total $ cap."""

# script_id: day_067_debate_multi_agent_ops/atlas/llm
@dataclass
class AgentBudget:
    per_agent_usd: float
    total_usd: float
    spent: dict[str, float] = field(default_factory=dict)
    held: dict[str, float] = field(default_factory=dict)      # reserved for calls in flight

    def _committed(self, agent: str | None = None) -> float:
        if agent is None:
            return sum(self.spent.values()) + sum(self.held.values())
        return self.spent.get(agent, 0.0) + self.held.get(agent, 0.0)

    def reserve(self, agent: str, est_usd: float) -> None:
        if self._committed(agent) + est_usd > self.per_agent_usd:
            raise BudgetExceeded(f"{agent}: per-agent cap ${self.per_agent_usd:.4f} reached")
        if self._committed() + est_usd > self.total_usd:
            raise BudgetExceeded(f"{agent}: total cap ${self.total_usd:.4f} reached")
        self.held[agent] = self.held.get(agent, 0.0) + est_usd

    def release(self, agent: str, est_usd: float) -> None:
        self.held[agent] = self.held.get(agent, 0.0) - est_usd

    def charge(self, agent: str, usage: Any, model: str, est_usd: float) -> float:
        self.release(agent, est_usd)
        cost = est_usd if usage is None else cost_usd(model, usage)   # unknown usage: bill the estimate
        self.spent[agent] = self.spent.get(agent, 0.0) + cost
        return cost

    def spent_by_agent(self) -> dict[str, float]:
        return dict(sorted(self.spent.items()))

# script_id: day_067_debate_multi_agent_ops/atlas/llm
class LineExporter(SpanExporter):
    """Prints one line per finished span; swap in Day 42's Phoenix exporter for a UI."""
    def export(self, spans) -> SpanExportResult:
        for s in spans:
            ms = (s.end_time - s.start_time) / 1e6
            print(f"span {s.name:<26} {ms:6.1f} ms usd={(s.attributes or {}).get('atlas.cost_usd', '-')}")
        return SpanExportResult.SUCCESS

def setup_tracing(exporter: SpanExporter | None = None) -> TracerProvider:
    provider = TracerProvider()
    provider.add_span_processor(SimpleSpanProcessor(exporter or LineExporter()))
    trace.set_tracer_provider(provider)          # global; OTel lets you set it once per process
    return provider

@contextmanager
def agent_span(name: str, round: int, **attrs: Any) -> Iterator[trace.Span]:
    # An exception inside the block is recorded on the span, then re-raised
    tracer = trace.get_tracer("atlas.debate")
    with tracer.start_as_current_span(f"invoke_agent {name}", attributes={
            "gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": name,
            "atlas.debate.round": round, **attrs}) as span:
        yield span

# script_id: day_067_debate_multi_agent_ops/atlas/llm
if __name__ == "__main__":
    from types import SimpleNamespace as Usage          # fake usage: runs with no API key
    setup_tracing()
    budget = AgentBudget(per_agent_usd=0.002, total_usd=0.004)
    est = estimate_usd("gpt-6-luna", 2_500, 1_500)       # $0.001 worst case per call
    for rnd in (1, 2, 3):
        for agent in ("debater_0", "judge"):
            try:
                with agent_span(agent, rnd) as span:
                    budget.reserve(agent, est)
                    fake = Usage(input_tokens=2_400, output_tokens=900)
                    span.set_attribute("atlas.cost_usd", round(budget.charge(agent, fake, "gpt-6-luna", est), 6))
            except BudgetExceeded as e:
                print("BudgetExceeded:", e)
    print({a: round(v, 5) for a, v in budget.spent_by_agent().items()})

span invoke_agent debater_0        0.1 ms usd=0.00069
span invoke_agent judge            0.0 ms usd=0.00069
span invoke_agent debater_0        0.0 ms usd=0.00069
span invoke_agent judge            0.0 ms usd=0.00069
span invoke_agent debater_0        2.6 ms usd=-
BudgetExceeded: debater_0: per-agent cap $0.0020 reached
span invoke_agent judge            0.2 ms usd=-
BudgetExceeded: judge: per-agent cap $0.0020 reached
{'debater_0': 0.00138, 'judge': 0.00138}

# script_id: day_067_debate_multi_agent_ops/atlas/debate
# atlas/debate.py: N debaters x R rounds, majority or judge, $ budget, spans.
# Install: uv add pydantic opentelemetry-sdk · Env: none · Needs atlas/llm.py, orbit_dupes.py
from __future__ import annotations
import asyncio, time
from collections import Counter
from dataclasses import dataclass, field
from typing import Any, Literal, Protocol
from opentelemetry import trace
from pydantic import BaseModel, create_model
from atlas.llm import AgentBudget, BudgetExceeded, agent_span, cost_usd
from orbit_dupes import DupItem

class Verdict(BaseModel):
    answer: str                                   # a candidate id, or "none"
    reason: str
    confidence: Literal["low", "med", "high"]

def verdict_model(item: DupItem) -> type[Verdict]:
    # Per-item Literal -> strict JSON-schema enum: hallucinated ids become impossible
    return create_model("ItemVerdict", __base__=Verdict, answer=(Literal[tuple(item.options)], ...))

class Ask(Protocol):
    model: str
    est_usd: float                                # worst-case $ per call, reserved up front
    async def __call__(self, agent: str, instructions: str, prompt: str,
                       schema: type[Verdict]) -> tuple[Verdict, Any]: ...   # (verdict, usage)

@dataclass
class Call:
    agent: str
    round: int
    usd: float
    seconds: float

# script_id: day_067_debate_multi_agent_ops/atlas/debate
@dataclass
class DebateResult:
    final: Verdict
    rounds: list[list[Verdict]]                   # rounds[r][i] = debater i in round r + 1
    calls: list[Call] = field(default_factory=list)
    stopped: str | None = None                    # BudgetExceeded message if the hard stop fired

    def majority_at(self, r: int) -> Verdict:
        return majority(self.rounds[min(r, len(self.rounds)) - 1])

    def flips(self, gold: str) -> tuple[int, int]:
        """(wrong -> right, right -> wrong) between round 1 and the last round."""
        pairs = list(zip(self.rounds[0], self.rounds[-1])) if len(self.rounds) > 1 else []
        good = sum(a.answer != gold and b.answer == gold for a, b in pairs)
        bad = sum(a.answer == gold and b.answer != gold for a, b in pairs)
        return good, bad

    def _upto(self, r: int | None) -> list[Call]:
        return [c for c in self.calls if r is None or c.round <= r]

    def usd(self, upto_round: int | None = None) -> float:
        return sum(c.usd for c in self._upto(upto_round))

    def n_calls(self, upto_round: int | None = None) -> int:
        return len(self._upto(upto_round))

    def seconds(self, upto_round: int | None = None) -> float:
        slowest: dict[int, float] = {}
        for c in self._upto(upto_round):
            slowest[c.round] = max(slowest.get(c.round, 0.0), c.seconds)
        return sum(slowest.values())

    def usage_by_agent(self) -> dict[str, float]:
        out: dict[str, float] = {}
        for c in self.calls:
            out[c.agent] = out.get(c.agent, 0.0) + c.usd
        return out

# script_id: day_067_debate_multi_agent_ops/atlas/debate
RANK = {"low": 0, "med": 1, "high": 2}
LENSES = ["Focus on error messages and ORB-xxxx error codes.",
          "Focus on the component, version and configuration involved.",
          "Focus on the symptom and the repro steps."]
SYSTEM = ("You decide whether a NEW Orbit issue duplicates one CANDIDATE issue. A duplicate has the "
          "same root cause, not only the same component. Answer a candidate id, or none.")

def majority(votes: list[Verdict]) -> Verdict:
    if len(votes) == 1:
        return votes[0]                           # a single call keeps its own reason/confidence
    counts = Counter(v.answer for v in votes)
    # Ties: most total confidence wins, then "none" (the safe default for a duplicate check)
    best = max(counts, key=lambda a: (counts[a], sum(RANK[v.confidence] for v in votes
                                                     if v.answer == a), a == "none"))
    k = counts[best]
    conf = (min((v.confidence for v in votes if v.answer == best), key=RANK.__getitem__)
            if k == len(votes) else "med" if k > len(votes) / 2 else "low")
    return Verdict(answer=best, reason=f"{k}/{len(votes)} votes", confidence=conf)

def _cands(c: list) -> str:
    return "\n\n".join(f"[{x.id}] {x.text}" for x in c)

def debater_prompt(item: DupItem, i: int, peers: list[Verdict] | None) -> str:
    c = list(item.candidates)
    c = c[i % len(c):] + c[:i % len(c)]          # rotate candidate order per debater
    text = f"{LENSES[i % len(LENSES)]}\n\nNEW:\n{item.text}\n\nCANDIDATES:\n{_cands(c)}"
    if peers:
        notes = "\n".join(f"- {'You' if j == i else 'Reviewer ' + chr(65 + j)}: {p.answer} ({p.reason})"
                          for j, p in enumerate(peers))
        text += (f"\n\nRound-1 answers:\n{notes}\nKeep your answer unless a reviewer points to "
                 "evidence you missed. Do not change it only to agree.")
    return text

def judge_prompt(item: DupItem, finals: list[Verdict]) -> str:
    notes = "\n".join(f"- Reviewer {chr(65 + j)}: {v.answer} ({v.reason})" for j, v in enumerate(finals))
    return (f"NEW:\n{item.text}\n\nCANDIDATES:\n{_cands(list(item.candidates))}\n\n"
            f"Reviewer answers:\n{notes}\nPick the best-supported answer.")

# script_id: day_067_debate_multi_agent_ops/atlas/debate
def _reserve_all(budget: AgentBudget | None, wants: list[tuple[str, Ask]]) -> None:
    done: list[tuple[str, Ask]] = []
    try:
        for name, ask in wants:
            if budget:
                budget.reserve(name, ask.est_usd)
                done.append((name, ask))
    except BudgetExceeded:
        for name, ask in done:
            budget.release(name, ask.est_usd)
        raise

async def _call(ask: Ask, agent: str, rnd: int, prompt: str, schema: type[Verdict],
                budget: AgentBudget | None, calls: list[Call], instructions: str = SYSTEM) -> Verdict:
    with agent_span(agent, rnd, **{"gen_ai.request.model": ask.model}) as span:
        t0 = time.perf_counter()
        try:
            verdict, use = await ask(agent, instructions, prompt, schema)
        except BaseException:
            if budget:
                budget.release(agent, ask.est_usd)   # don't leak the hold on errors
            raise
        if budget:
            cost = budget.charge(agent, use, ask.model, ask.est_usd)
        else:
            cost = cost_usd(ask.model, use) if use is not None else ask.est_usd
        span.set_attributes({"atlas.cost_usd": round(cost, 6), "atlas.answer": verdict.answer})
        calls.append(Call(agent, rnd, cost, time.perf_counter() - t0))
        return verdict

# script_id: day_067_debate_multi_agent_ops/atlas/debate
async def _judge(item: DupItem, ask: Ask, finals: list[Verdict], rnd: int, schema: type[Verdict],
                 budget: AgentBudget | None, calls: list[Call], fallback: Verdict) -> Verdict:
    _reserve_all(budget, [("judge", ask), ("judge", ask)])
    instr = "You are the judge of a review panel. " + SYSTEM
    a, b = await asyncio.gather(
        _call(ask, "judge", rnd, judge_prompt(item, finals), schema, budget, calls, instr),
        _call(ask, "judge", rnd, judge_prompt(item, finals[::-1]), schema, budget, calls, instr))
    return a if a.answer == b.answer else fallback

async def debate(item: DupItem, ask: Ask | list[Ask], n_agents: int = 3, rounds: int = 2,
                 judge: bool = False, budget: AgentBudget | None = None,
                 judge_ask: Ask | None = None) -> DebateResult:
    asks = ask if isinstance(ask, list) else [ask] * n_agents   # a list = a mixed panel
    names = [f"debater_{i}" for i in range(len(asks))]
    schema, history, calls, stopped = verdict_model(item), [], [], None
    tracer = trace.get_tracer("atlas.debate")
    with tracer.start_as_current_span(f"debate issue {item.issue_id}") as root:
        try:
            for r in range(1, rounds + 1):
                _reserve_all(budget, list(zip(names, asks)))      # before anything launches
                peers = history[-1] if history else None
                with tracer.start_as_current_span(f"round {r}"):
                    history.append(list(await asyncio.gather(*(
                        _call(a, n, r, debater_prompt(item, i, peers), schema, budget, calls)
                        for i, (n, a) in enumerate(zip(names, asks))))))
            final = majority(history[-1])
            if judge:
                final = await _judge(item, judge_ask or asks[0], history[-1], rounds + 1,
                                     schema, budget, calls, final)
        except BudgetExceeded as e:
            stopped = str(e)
            root.add_event("budget_exceeded", {"reason": stopped})
            final = majority(history[-1]) if history else Verdict(
                answer="none", reason="budget exhausted before any vote", confidence="low")
    return DebateResult(final, history, calls, stopped)

# script_id: day_067_debate_multi_agent_ops/test_debate
# test_debate.py: offline process evals (flips, hard stop, judge swap, spans). No API key.
# Install: uv add pytest pydantic opentelemetry-sdk · Run: pytest test_debate.py (or python test_debate.py)
import asyncio
from dataclasses import dataclass, field
from types import SimpleNamespace
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from atlas.llm import AgentBudget, estimate_usd, setup_tracing
from atlas.debate import debate
from orbit_dupes import load_items

SPANS = InMemorySpanExporter()
setup_tracing(SPANS)
ITEM = load_items()[0]
WRONG = next(c.id for c in ITEM.candidates if c.id != ITEM.gold)

@dataclass
class FakeAsk:
    """Scripted panel: answers[agent][k] is that agent's k-th answer. Same shape as the real Ask."""
    answers: dict[str, list[str]]
    model: str = "gpt-6-luna"
    est_usd: float = estimate_usd("gpt-6-luna", 2_500, 1_500)
    seen: list[str] = field(default_factory=list)

    async def __call__(self, agent, instructions, prompt, schema):
        k = self.seen.count(agent)
        self.seen.append(agent)
        script = self.answers[agent]
        v = schema(answer=script[min(k, len(script) - 1)], reason="scripted", confidence="med")
        return v, SimpleNamespace(input_tokens=2_400, output_tokens=900)

# script_id: day_067_debate_multi_agent_ops/test_debate
def test_good_flip_and_one_span_per_agent_per_round():
    SPANS.clear()
    ask = FakeAsk({"debater_0": [ITEM.gold], "debater_1": [WRONG, ITEM.gold], "debater_2": [WRONG]})
    res = asyncio.run(debate(ITEM, ask, n_agents=3, rounds=2))
    assert res.majority_at(1).answer == WRONG and res.final.answer == ITEM.gold
    assert res.flips(ITEM.gold) == (1, 0)
    spans = [s for s in SPANS.get_finished_spans() if s.name.startswith("invoke_agent")]
    assert len(spans) == res.n_calls() == 6
    assert {(s.attributes["gen_ai.agent.name"], s.attributes["atlas.debate.round"]) for s in spans} \
        == {(f"debater_{i}", r) for i in range(3) for r in (1, 2)}

def test_per_agent_cap_returns_best_so_far():
    ask = FakeAsk({"debater_0": [ITEM.gold], "debater_1": [WRONG, ITEM.gold], "debater_2": [WRONG]})
    budget = AgentBudget(per_agent_usd=1.5 * ask.est_usd, total_usd=1.0)   # room for 1 call each
    res = asyncio.run(debate(ITEM, ask, rounds=2, budget=budget))
    assert res.stopped and "per-agent cap" in res.stopped
    assert len(res.rounds) == 1 and res.final.answer == WRONG              # round-1 majority
    assert all(v <= budget.per_agent_usd for v in budget.spent_by_agent().values())

def test_judge_swap_disagreement_keeps_majority():
    ask = FakeAsk({f"debater_{i}": [ITEM.gold] for i in range(3)} | {"judge": [WRONG, ITEM.gold]})
    res = asyncio.run(debate(ITEM, ask, rounds=2, judge=True))
    assert res.final.answer == ITEM.gold and res.n_calls() == 8

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

PASS test_good_flip_and_one_span_per_agent_per_round
PASS test_per_agent_cap_returns_best_so_far
PASS test_judge_swap_disagreement_keeps_majority

# script_id: day_067_debate_multi_agent_ops/debate_eval
# debate_eval.py: matched-cost eval of 5 configs on 20 Orbit duplicate decisions.
# Install: uv add "openai>=3.46,<4" pydantic opentelemetry-sdk · Env: OPENAI_API_KEY, ORBIT_DATA
# Needs orbit_dupes.py, atlas/llm.py, atlas/debate.py · Cost ≈ $0.15 on gpt-6-luna (as of 2026-11, verify)
from __future__ import annotations
import asyncio, os, statistics
from dataclasses import dataclass
from openai import AsyncOpenAI
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from pydantic import ValidationError
from atlas.llm import AgentBudget, estimate_usd, setup_tracing
from atlas.debate import debate
from orbit_dupes import DupItem, load_items

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
MAX_OUT = 8_000   # caps reasoning + answer; the reasoning guide suggests ~25k headroom for hard tasks

@dataclass
class OpenAIAsk:
    client: AsyncOpenAI
    model: str = MODEL
    effort: str = "low"

    @property
    def est_usd(self) -> float:
        return estimate_usd(self.model, 3_000, MAX_OUT)   # 3k input is generous here; measure yours

    async def __call__(self, agent, instructions, prompt, schema):
        try:
            r = await self.client.responses.parse(model=self.model, instructions=instructions,
                input=prompt, text_format=schema, reasoning={"effort": self.effort},
                max_output_tokens=MAX_OUT)
        except ValidationError:                  # truncated or malformed JSON
            return schema(answer="none", reason="unparseable", confidence="low"), None
        v = r.output_parsed or schema(answer="none", reason=f"no answer ({r.status})", confidence="low")
        return v, r.usage

# script_id: day_067_debate_multi_agent_ops/debate_eval
CONFIGS = ["single_low", "single_high", "vote3", "debate", "debate_judge"]

async def run_item(client: AsyncOpenAI, item: DupItem) -> dict:
    a = await debate(item, OpenAIAsk(client), n_agents=1, rounds=1)
    b = await debate(item, OpenAIAsk(client, effort="high"), n_agents=1, rounds=1)
    d = await debate(item, OpenAIAsk(client), n_agents=3, rounds=2, judge=True,
                     budget=AgentBudget(per_agent_usd=0.03, total_usd=0.10))

    def row(res, verdict, upto: int | None) -> dict:
        return {"ok": verdict.answer == item.gold, "usd": res.usd(upto),
                "calls": res.n_calls(upto), "s": res.seconds(upto)}

    return {"single_low": row(a, a.final, None), "single_high": row(b, b.final, None),
            "vote3": row(d, d.majority_at(1), 1), "debate": row(d, d.majority_at(2), 2),
            "debate_judge": row(d, d.final, None), "_runs": (a, b, d)}

async def evaluate(items: list[DupItem]) -> list[dict]:
    sem = asyncio.Semaphore(4)                   # a few items at once; backoff is Day 84's topic
    async with AsyncOpenAI() as client:
        async def one(it: DupItem) -> dict:
            async with sem:
                return await run_item(client, it)
        return list(await asyncio.gather(*(one(it) for it in items)))

# script_id: day_067_debate_multi_agent_ops/debate_eval
def report(items: list[DupItem], results: list[dict], spans: InMemorySpanExporter) -> None:
    print(f"{'config':<13} {'acc':>5} {'$/item':>9} {'calls':>6} {'p50 s':>6}")
    for cfg in CONFIGS:
        rows = [r[cfg] for r in results]
        print(f"{cfg:<13} {sum(r['ok'] for r in rows) / len(rows):5.2f} "
              f"{statistics.mean(r['usd'] for r in rows):9.5f} "
              f"{statistics.mean(r['calls'] for r in rows):6.1f} {statistics.median(r['s'] for r in rows):6.1f}")
    debates = [r["_runs"][2] for r in results]
    flips = [d.flips(it.gold) for d, it in zip(debates, items)]
    print(f"round-2 flips: +{sum(f[0] for f in flips)} wrong->right, -{sum(f[1] for f in flips)} right->wrong")
    spend: dict[str, float] = {}
    for d in debates:
        for agent, v in d.usage_by_agent().items():
            spend[agent] = spend.get(agent, 0.0) + v
    print("debate spend by agent:", {a: round(v, 4) for a, v in sorted(spend.items())})
    print("hard stops:", sum(d.stopped is not None for d in debates))
    agent_spans = [s for s in spans.get_finished_spans() if s.name.startswith("invoke_agent")]
    assert len(agent_spans) == sum(run.n_calls() for r in results for run in r["_runs"])
    assert all(d.usd() <= 0.10 for d in debates), "a debate broke its total budget"

if __name__ == "__main__":
    spans = InMemorySpanExporter()
    setup_tracing(spans)                          # or Day 42's Phoenix setup to browse the trees
    items = load_items()
    report(items, asyncio.run(evaluate(items)), spans)

config          acc    $/item  calls  p50 s
single_low     x.xx   0.000xx    1.0    x.x
single_high    x.xx   0.000xx    1.0    x.x
vote3          x.xx   0.00xxx    3.0    x.x
debate         x.xx   0.00xxx    6.0    x.x
debate_judge   x.xx   0.00xxx    8.0    x.x
round-2 flips: +n wrong->right, -n right->wrong
debate spend by agent: {'debater_0': ..., 'debater_1': ..., 'debater_2': ..., 'judge': ...}
hard stops: 0