Debugging Agents

AgentResult(text="Closest past report: #227.", stop_reason="done", steps=9, cost_usd=0.0021)
tools: search_issues x5 (same query), get_issue x1, get_doc_section x2, answer

# script_id: day_058_debugging_agents/detect_minimal
# The smallest loop detector: count identical (tool, args) pairs in a toy trace. Stdlib only.
import json
from collections import Counter

trace = [  # (step, tool, args) from the run in the table above
    (1, "search_issues", {"query": "notify missed"}),
    (2, "search_issues", {"query": "postgres notify missed"}),
    (3, "search_issues", {"query": "postgres notify missed"}),
    (4, "search_issues", {"query": "postgres notify missed"}),
    (5, "search_issues", {"query": "postgres notify missed"}),
    (6, "get_issue", {"issue_id": 227}),
]
key = lambda tool, args: (tool, json.dumps(args, sort_keys=True))
calls = Counter(key(tool, args) for _, tool, args in trace)
for (tool, args), n in calls.items():
    if n >= 3:  # one repeat can be a retry; three identical calls is a pattern
        steps = [s for s, t, a in trace if key(t, a) == (tool, args)]
        print(f"loop: {tool} x{n} (steps {steps[0]}-{steps[-1]})  args={args}")

loop: search_issues x4 (steps 2-5)  args={"query": "postgres notify missed"}

# script_id: day_058_debugging_agents/record_runs
# Records three broken Atlas runs as span fixtures (fixtures/run_*.json). Keyless, deterministic.
# Install: uv add opentelemetry-sdk · Env: ORBIT_DATA (default datasets/orbit), FIXTURES (default
# fixtures) · Services: none. Content capture (tool args) is ON here: local debugging only, mind PII.
import json, os, re
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Callable, Iterator
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
FIXTURES = Path(os.environ.get("FIXTURES", "fixtures"))
provider = TracerProvider()          # local, not global: nothing leaves this process
tracer = provider.get_tracer("atlas")


def attach_memory_exporter() -> InMemorySpanExporter:   # built on Day 42
    exporter = InMemorySpanExporter()
    provider.add_span_processor(SimpleSpanProcessor(exporter))
    return exporter


@contextmanager
def step_span(name: str, kind: str, **attrs: Any) -> Iterator[Any]:   # built on Day 42
    """One timed step. kind is the GenAI operation name: invoke_agent, chat or execute_tool."""
    with tracer.start_as_current_span(name, attributes={"gen_ai.operation.name": kind, **attrs}) as span:
        yield span


def span_to_dict(s: Any) -> dict:
    """The fixture format: plain JSON, so any exporter's spans can be converted to it."""
    return {"name": s.name, "start_ns": s.start_time, "end_ns": s.end_time,
            "status": s.status.status_code.name, "attributes": dict(s.attributes or {})}

# script_id: day_058_debugging_agents/record_runs
# built on Days 35-37: compact read-only tools that return Day 37's envelope.
ISSUES = {r["id"]: r for r in map(json.loads, (DATA_DIR / "issues" / "issues.jsonl").open())}
DOC_IDS = sorted(p.stem for p in (DATA_DIR / "docs").glob("*.md"))


def tokens(text: str) -> int:
    return len(text) // 4                      # rough estimate; Day 4 counts exactly


def envelope_err(kind: str, message: str, hint: str | None = None) -> dict:   # built on Day 37
    return {"ok": False, "error": {"kind": kind, "message": message, "retryable": False, "hint": hint}}


def make_tools(*, strict_docs: bool = False, issues: dict | None = None) -> dict[str, Callable]:
    issues = issues or ISSUES

    def search_issues(query: str) -> dict:
        words = set(query.lower().split())
        hits = sorted(issues.values(), key=lambda r: -len(words & set(r["title"].lower().split())))
        return {"ok": True, "data": [{"id": r["id"], "title": r["title"][:80]} for r in hits[:3]]}

    def get_issue(issue_id: int) -> dict:
        r = issues[issue_id]
        text = "\n\n".join([r["body"], *(c["body"] for c in r["comments"])])
        return {"ok": True, "data": {"id": issue_id, "title": r["title"], "text": text}}

    def get_doc_section(doc_id: str) -> dict:
        path = DATA_DIR / "docs" / f"{doc_id}.md"
        if not path.exists():
            if not strict_docs:                # the bug: a bare "not found" gives no next step
                return envelope_err("not_found", f"no doc {doc_id!r}")
            return envelope_err("bad_args", f"unknown doc_id {doc_id!r}",
                                hint=f"valid doc ids: {', '.join(DOC_IDS)}")
        return {"ok": True, "data": {"doc_id": doc_id, "text": path.read_text(encoding="utf-8")[:1_500]}}

    return {"search_issues": search_issues, "get_issue": get_issue, "get_doc_section": get_doc_section}

# script_id: day_058_debugging_agents/record_runs
CLEARED = "[cleared: {name} result, ~{n} tokens]"


def clear_old_results(context: list[dict], keep_last: int, min_tokens: int = 200) -> list[str]:
    """Day 49's clear_tool_results, compact: blank all but the last keep_last results. Returns codes lost."""
    lost: list[str] = []
    for item in [c for c in context if c["role"] == "tool"][:-keep_last]:
        if item["content"].startswith("[cleared") or tokens(item["content"]) < min_tokens:
            continue                           # already cleared, or too small to be worth it
        lost += re.findall(r"ORB-\d{4}", item["content"])    # instrument what you throw away
        item["content"] = CLEARED.format(name=item["name"], n=tokens(item["content"]))
    return sorted(set(lost))

# script_id: day_058_debugging_agents/record_runs
def run(policy: Callable, task: str, *, run_id: str, tools: dict | None = None, max_steps: int = 8,
        keep_last: int | None = None, loop_guard: int | None = None) -> None:
    tools, context, seen = tools or make_tools(), [{"role": "user", "content": task}], {}
    with step_span("invoke_agent atlas", "invoke_agent", **{"atlas.run_id": run_id}) as root:
        stop, answer = "max_steps", ""
        for step in range(1, max_steps + 1):
            sent = 300 + sum(tokens(c["content"]) for c in context)   # 300 ≈ instructions + tool schemas
            with step_span("chat", "chat", **{"atlas.step": step, "gen_ai.usage.input_tokens": sent}):
                action = policy(context)
            if action[0] == "answer":
                stop, answer = "done", action[1]
                break
            _, name, args = action
            key = f"{name}{json.dumps(args, sort_keys=True)}"
            seen[key] = seen.get(key, 0) + 1
            with step_span(f"execute_tool {name}", "execute_tool", **{"atlas.step": step,
                           "gen_ai.tool.name": name, "gen_ai.tool.call.arguments": json.dumps(args)}) as span:
                try:
                    env = tools[name](**args) if name in tools else envelope_err("unknown_tool", name)
                except (KeyError, TypeError) as e:          # bad id or bad arg names: still an envelope
                    env = envelope_err("bad_args", str(e))
                text = json.dumps(env)
                if loop_guard and seen[key] >= loop_guard:  # Fix A: say it in the model's own context
                    text += f"\n[loop-guard] {name} already ran {seen[key]}x with these args; open a result or answer."
                context.append({"role": "tool", "name": name, "content": text})
                lost = clear_old_results(context, keep_last) if keep_last else []
                span.set_attributes({"atlas.ok": env["ok"], "atlas.result_tokens": tokens(text),
                                     "atlas.error_kind": "" if env["ok"] else env["error"]["kind"],
                                     "atlas.cleared_codes": lost})
        root.set_attributes({"atlas.stop_reason": stop, "atlas.answer": answer, "atlas.steps": step})

# script_id: day_058_debugging_agents/record_runs
def seen_text(context: list[dict]) -> str:
    return "\n".join(c["content"] for c in context)


def policy_loop(context: list[dict]) -> tuple:
    """Run A: wants details, but nothing in its context says how to open a result, so it re-searches."""
    tool_msgs = [c for c in context if c["role"] == "tool"]
    if "[loop-guard]" not in seen_text(context):
        return ("call", "search_issues", {"query": "postgres notify missed"})
    if tool_msgs[-1]["name"] != "get_issue":
        top = json.loads(tool_msgs[-1]["content"].split("\n[loop-guard]")[0])["data"][0]["id"]
        return ("call", "get_issue", {"issue_id": top})
    return ("answer", f"Closest past report: #{json.loads(tool_msgs[-1]['content'])['data']['id']}.")


def policy_bad_args(context: list[dict]) -> tuple:
    """Run B: invents a doc id shaped like Orbit's error codes, then answers even if the call failed."""
    tool_msgs = [c for c in context if c["role"] == "tool"]
    if not tool_msgs:
        return ("call", "get_doc_section", {"doc_id": "ORB-DOC-9"})
    last = json.loads(tool_msgs[-1]["content"])
    if "broker-postgres" in ((last.get("error") or {}).get("hint") or ""):   # an actionable hint
        return ("call", "get_doc_section", {"doc_id": "broker-postgres"})
    cited = last["data"]["doc_id"] if last["ok"] else "ORB-DOC-9"
    return ("answer", f"Postgres broker setup is covered in {cited}.")


def policy_context(context: list[dict]) -> tuple:
    """Run C: reads the long issue and two docs, then diagnoses from whatever is still in context."""
    n = sum(c["role"] == "tool" for c in context)
    issue_id = int(re.search(r"#(\d+)", context[0]["content"]).group(1))
    plan = [("search_issues", {"query": "postgres pickup latency"}), ("get_issue", {"issue_id": issue_id}),
            ("get_doc_section", {"doc_id": "scaling-workers"}),
            ("get_doc_section", {"doc_id": "performance-tuning"})]
    if n < len(plan):
        return ("call", *plan[n])
    if "ORB-1001" in seen_text(context):
        return ("answer", "Listener connection resets (ORB-1001) from an idle timeout; enable TCP keepalives.")
    return ("answer", "Workers look saturated under load; scale out (see scaling-workers).")

# script_id: day_058_debugging_agents/record_runs
LONG_ISSUE = max(ISSUES.values(), key=lambda r: len(r["body"]))["id"]
BROKEN = {  # fixture name -> (policy, task, run kwargs)
    "run_a_loop": (policy_loop, "Find past issues about missed NOTIFY events on Postgres.", {}),
    "run_b_bad_args": (policy_bad_args, "Which doc covers Postgres broker setup?", {}),
    "run_c_context_loss": (policy_context, f"Diagnose Orbit issue #{LONG_ISSUE}.", {"keep_last": 1}),
}
EXPORTER = attach_memory_exporter()


def record(name: str, policy: Callable, task: str, **kw: Any) -> list[dict]:
    """Run once and return its spans as fixture dicts."""
    EXPORTER.clear()
    run(policy, task, run_id=name, **kw)
    return [span_to_dict(s) for s in EXPORTER.get_finished_spans()]


def ensure_fixtures(out: Path = FIXTURES, force: bool = False) -> Path:
    out.mkdir(parents=True, exist_ok=True)
    for name, (policy, task, kw) in BROKEN.items():
        if force or not (out / f"{name}.json").exists():
            (out / f"{name}.json").write_text(json.dumps(record(name, policy, task, **kw), indent=1))
    return out


if __name__ == "__main__":
    for f in sorted(ensure_fixtures(force=True).glob("run_*.json")):
        spans = json.loads(f.read_text())
        root = next(s for s in spans if s["name"].startswith("invoke_agent"))["attributes"]
        print(f"{f.name:<26}{len(spans):>3} spans  stop={root['atlas.stop_reason']:<10} answer={root['atlas.answer']!r}")

run_a_loop.json            17 spans  stop=max_steps  answer=''
run_b_bad_args.json         4 spans  stop=done       answer='Postgres broker setup is covered in ORB-DOC-9.'
run_c_context_loss.json    10 spans  stop=done       answer='Workers look saturated under load; scale out (see scaling-workers).'

{
 "name": "execute_tool get_doc_section",
 "status": "UNSET",
 "attributes": {
  "gen_ai.operation.name": "execute_tool",
  "atlas.step": 1,
  "gen_ai.tool.name": "get_doc_section",
  "gen_ai.tool.call.arguments": "{\"doc_id\": \"ORB-DOC-9\"}",
  "atlas.ok": false,
  "atlas.error_kind": "not_found",

# script_id: day_058_debugging_agents/debug
# atlas/debug.py: read an agent run as a trace and flag known failure modes.
# Install: stdlib only (fixtures come from record_runs.py) · Env: none · Services: none
from __future__ import annotations
import json
from collections import Counter
from dataclasses import dataclass


@dataclass(frozen=True)
class FailureMode:
    name: str
    signal: str      # what you look for in the trace
    cause: str       # what usually produces it
    fix: str         # the fix pattern to try first


FAILURE_MODES: dict[str, FailureMode] = {m.name: m for m in [
    FailureMode("loop", "same (tool, args) 3+ times", "no stop signal or next step", "loop guard + hard stop"),
    FailureMode("wrong_tool", "a tool that can't answer the task", "vague tool descriptions", "sharpen descriptions"),
    FailureMode("hallucinated_call", "unknown tool/id, then no recovery", "guessed names", "validate + list valid options"),
    FailureMode("error_storm", "3+ failed calls in a row", "retrying a permanent error", "error kinds + retryable"),
    FailureMode("context_loss", "evidence cleared before the answer", "aggressive clearing", "pin facts; raise keep_last"),
    FailureMode("premature_finish", "answer with no successful tool result", "stops too early", "require evidence"),
    FailureMode("runaway_budget", "stopped by a max_* limit", "a loop or storm upstream", "fix the cause; keep the cap"),
    FailureMode("unsafe_action", "side effect without approval", "no gate on writes", "approval node; run_once"),
]}

# script_id: day_058_debugging_agents/debug
@dataclass
class Step:
    idx: int                  # agent step number (a chat span and its tool span share one)
    kind: str                 # "chat" or "tool"
    name: str
    args: str = ""
    ok: bool = True
    tokens: int = 0           # chat: input tokens sent · tool: result size
    ms: float = 0.0
    error_kind: str = ""
    cleared_codes: tuple[str, ...] = ()


@dataclass
class AgentTrace:
    run_id: str
    steps: list[Step]
    stop_reason: str          # Day 48's StopReason values: done, max_steps, max_tokens, ...
    total_tokens: int         # input tokens summed over model calls (what you were billed for)
    answer: str = ""


def load_trace(spans: list[dict]) -> AgentTrace:
    """Turn exported span dicts (record_runs.span_to_dict) into an AgentTrace."""
    root = next(s for s in spans if s["attributes"]["gen_ai.operation.name"] == "invoke_agent")
    steps = []
    for s in sorted(spans, key=lambda s: s["start_ns"]):
        a = s["attributes"]
        if a["gen_ai.operation.name"] not in ("chat", "execute_tool"):
            continue
        chat = a["gen_ai.operation.name"] == "chat"
        steps.append(Step(idx=a["atlas.step"], kind="chat" if chat else "tool",
                          name=a.get("gen_ai.tool.name", "model"), args=a.get("gen_ai.tool.call.arguments", ""),
                          ok=a.get("atlas.ok", True), ms=(s["end_ns"] - s["start_ns"]) / 1e6,
                          tokens=a["gen_ai.usage.input_tokens"] if chat else a.get("atlas.result_tokens", 0),
                          error_kind=a.get("atlas.error_kind", ""), cleared_codes=tuple(a.get("atlas.cleared_codes", ()))))
    r = root["attributes"]
    return AgentTrace(r["atlas.run_id"], steps, r["atlas.stop_reason"],
                      sum(st.tokens for st in steps if st.kind == "chat"), r.get("atlas.answer", ""))

# script_id: day_058_debugging_agents/debug
@dataclass
class Failure:
    mode: str
    step: int
    evidence: str


def _tools(t: AgentTrace) -> list[Step]:
    return [s for s in t.steps if s.kind == "tool"]


def find_loops(t: AgentTrace, max_repeat: int = 3) -> list[Failure]:
    out = []
    for (name, args), n in Counter((s.name, s.args) for s in _tools(t)).items():
        if n >= max_repeat:
            idxs = [s.idx for s in _tools(t) if (s.name, s.args) == (name, args)]
            out.append(Failure("loop", idxs[0], f"{name}({args}) x{n} (steps {idxs[0]}-{idxs[-1]})"))
    return out


def find_hallucinated_calls(t: AgentTrace) -> list[Failure]:
    tools, out = _tools(t), []
    for i, s in enumerate(tools):
        recovered = any(x.name == s.name and x.ok for x in tools[i + 1:])
        if not s.ok and s.error_kind in ("unknown_tool", "not_found", "bad_args") and not recovered:
            out.append(Failure("hallucinated_call", s.idx, f"{s.name}({s.args}) -> {s.error_kind}, never recovered"))
    return out


def find_context_loss(t: AgentTrace) -> list[Failure]:
    out = []
    for s in _tools(t):
        missing = [c for c in s.cleared_codes if c not in t.answer]
        if missing and t.stop_reason == "done":
            out.append(Failure("context_loss", s.idx,
                               f"cleared {', '.join(missing)} after step {s.idx}; the answer never mentions it"))
    return out

# script_id: day_058_debugging_agents/debug
def find_runaway(t: AgentTrace) -> list[Failure]:
    if not t.stop_reason.startswith("max_"):
        return []
    return [Failure("runaway_budget", t.steps[-1].idx, f"stopped by {t.stop_reason} at step {t.steps[-1].idx}")]


def detect_failures(trace: AgentTrace, *, max_repeat: int = 3, error_streak: int = 3) -> list[Failure]:
    """Known modes only. An empty list means 'none of these', not 'the run was correct'."""
    found = find_loops(trace, max_repeat) + find_hallucinated_calls(trace)
    found += find_context_loss(trace) + find_runaway(trace)
    return sorted(found, key=lambda f: f.step)


def report(t: AgentTrace) -> None:
    print(f"{t.run_id}: {len({s.idx for s in t.steps})} steps, stop={t.stop_reason}, "
          f"~{t.total_tokens:,} input tokens")
    for f in detect_failures(t) or [Failure("-", 0, "no known failure mode")]:
        print(f"  {f.mode:<18} step {f.step:<3} {f.evidence}")


if __name__ == "__main__":
    from record_runs import ensure_fixtures      # writes the fixtures if they're missing
    for path in sorted(ensure_fixtures().glob("run_*.json")):
        report(load_trace(json.loads(path.read_text())))

run_a_loop: 8 steps, stop=max_steps, ~4,752 input tokens
  loop               step 1   search_issues({"query": "postgres notify missed"}) x8 (steps 1-8)
  runaway_budget     step 8   stopped by max_steps at step 8
run_b_bad_args: 2 steps, stop=done, ~646 input tokens
  hallucinated_call  step 1   get_doc_section({"doc_id": "ORB-DOC-9"}) -> not_found, never recovered
run_c_context_loss: 5 steps, stop=done, ~26,829 input tokens
  context_loss       step 3   cleared ORB-1001 after step 3; the answer never mentions it

# script_id: day_058_debugging_agents/fix_runs
# Fix each broken run with ONE change, re-run the same task, compare. Keyless.
# Needs record_runs.py and debug.py from this lesson next to it.
from debug import AgentTrace, detect_failures, load_trace
from record_runs import BROKEN, DOC_IDS, make_tools, record

FIXES = {  # one change per run
    "run_a_loop": {"loop_guard": 2},
    "run_b_bad_args": {"tools": make_tools(strict_docs=True)},
    "run_c_context_loss": {"keep_last": 3},
}
CHECKS = {  # what "correct" means for each task
    "run_a_loop": lambda t: "#" in t.answer,
    "run_b_bad_args": lambda t: any(f" {d}." in t.answer for d in DOC_IDS),
    "run_c_context_loss": lambda t: "ORB-1001" in t.answer,
}


def measure(name: str, fixed: bool) -> AgentTrace:
    policy, task, kw = BROKEN[name]
    return load_trace(record(name, policy, task, **{**kw, **(FIXES[name] if fixed else {})}))


if __name__ == "__main__":
    print(f"{'run':<20}{'version':<8}{'steps':>5}  {'stop':<10}{'input tok':>10}  {'pass':<6}failures")
    for name in BROKEN:
        for label, fixed in (("before", False), ("after", True)):
            t = measure(name, fixed)
            modes = ",".join(f.mode for f in detect_failures(t)) or "-"
            print(f"{name:<20}{label:<8}{len({s.idx for s in t.steps}):>5}  {t.stop_reason:<10}"
                  f"{t.total_tokens:>10,}  {str(CHECKS[name](t)):<6}{modes}")

run                 version steps  stop       input tok  pass  failures
run_a_loop          before      8  max_steps      4,752  False loop,runaway_budget
run_a_loop          after       4  done          25,841  True  -
run_b_bad_args      before      2  done             646  False hallucinated_call
run_b_bad_args      after       3  done           1,797  True  -
run_c_context_loss  before      5  done          26,829  False context_loss
run_c_context_loss  after       5  done          75,487  True  -

# script_id: day_058_debugging_agents/replay_run_c
# Replay Run C in LangGraph, find the checkpoint before the bad answer, fork it with a pinned note.
# Install: uv add langgraph opentelemetry-sdk · Needs record_runs.py next to it · No API key, no services.
import json, re
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from record_runs import LONG_ISSUE, clear_old_results, make_tools, policy_context

TOOLS = make_tools()


class RunState(TypedDict, total=False):
    context: list[dict]       # no reducer: each node returns the whole list (last write wins)
    action: list
    answer: str


def agent(s: RunState) -> dict:
    action = list(policy_context(s["context"]))
    return {"action": action, "answer": action[1] if action[0] == "answer" else ""}


def tools(s: RunState) -> dict:
    _, name, args = s["action"]
    ctx = [dict(c) for c in s["context"]]           # copy: never mutate a checkpointed value in place
    ctx.append({"role": "tool", "name": name, "content": json.dumps(TOOLS[name](**args))})
    clear_old_results(ctx, keep_last=1)             # the shipped "cost fix" behind the bug
    return {"context": ctx}


g = StateGraph(RunState)
g.add_node("agent", agent)
g.add_node("tools", tools)
g.add_edge(START, "agent")
g.add_edge("tools", "agent")
g.add_conditional_edges("agent", lambda s: END if s["answer"] else "tools", ["tools", END])
app = g.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "run-c"}, "recursion_limit": 25}   # set the limit yourself

# script_id: day_058_debugging_agents/replay_run_c
bad = app.invoke({"context": [{"role": "user", "content": f"Diagnose Orbit issue #{LONG_ISSUE}."}]}, cfg)
print("bad run: ", bad["answer"])

history = list(app.get_state_history(cfg))          # newest first
tool_items = lambda s: [c for c in s.values.get("context", []) if c["role"] == "tool"]
before = next(s for s in history if s.next == ("agent",) and len(tool_items(s)) == 4)
print("fork point: step", before.metadata["step"], "· cleared:",
      [c["name"] for c in tool_items(before) if c["content"].startswith("[cleared")])

# The oldest checkpoint where get_issue's result is still intact holds the evidence.
intact = next(c["content"] for s in reversed(history) for c in tool_items(s)
              if c["name"] == "get_issue" and not c["content"].startswith("[cleared"))
codes = sorted(set(re.findall(r"ORB-\d{4}", intact)))
note = {"role": "note", "content": f"Pinned from issue #{LONG_ISSUE}: error codes {codes}."}

fork_cfg = app.update_state(before.config, {"context": before.values["context"] + [note]},
                            as_node="tools")              # a NEW checkpoint; the old one is untouched
good = app.invoke(None, fork_cfg)
print("fork run:", good["answer"])
print("checkpoints:", len(history), "->", len(list(app.get_state_history(cfg))))

bad run:  Workers look saturated under load; scale out (see scaling-workers).
fork point: step 8 · cleared: ['get_issue', 'get_doc_section']
fork run: Listener connection resets (ORB-1001) from an idle timeout; enable TCP keepalives.
checkpoints: 11 -> 13

# script_id: day_058_debugging_agents/minimal_repro
# Shrink Run C's long issue to the smallest text that fails FOR THE SAME REASON. Keyless.
# Needs record_runs.py and debug.py next to it. Writes fixtures/run_c_minimal.json.
import json
from debug import detect_failures, load_trace
from record_runs import BROKEN, FIXTURES, ISSUES, LONG_ISSUE, make_tools, record

policy, task, _ = BROKEN["run_c_context_loss"]
full = ISSUES[LONG_ISSUE]
lines = "\n\n".join([full["body"], *(c["body"] for c in full["comments"])]).splitlines()


def run_with(text_lines: list[str], keep_last: int) -> list[dict]:
    issue = {**full, "body": "\n".join(text_lines), "comments": []}
    return record("run_c_minimal", policy, task, tools=make_tools(issues={LONG_ISSUE: issue}),
                  keep_last=keep_last)


def still_fails(text_lines: list[str]) -> bool:
    """The same bug, not any failure: wrong with keep_last=1 AND right with keep_last=3."""
    buggy, fixed = load_trace(run_with(text_lines, 1)), load_trace(run_with(text_lines, 3))
    return "ORB-1001" not in buggy.answer and "ORB-1001" in fixed.answer


assert still_fails(lines), "reproduce on the full input before you shrink it"
print(f"start: {len(lines)} lines")
while len(lines) > 1:
    half = len(lines) // 2
    smaller = next((part for part in (lines[:half], lines[half:]) if still_fails(part)), None)
    if smaller is None:
        break                                  # neither half fails alone: minimal for halving
    lines = smaller
spans = run_with(lines, 1)
print(f"minimal: {len(lines)} lines, ~{len(' '.join(lines)) // 4} tokens ->",
      [f.mode for f in detect_failures(load_trace(spans))])
FIXTURES.mkdir(parents=True, exist_ok=True)
(FIXTURES / "run_c_minimal.json").write_text(json.dumps(spans, indent=1))

start: 864 lines
minimal: 13 lines, ~250 tokens -> ['context_loss']

# script_id: day_058_debugging_agents/test_debug
# Regression tests for the three bugs. Run: uv run pytest test_debug.py -q
# Needs record_runs.py, debug.py and fix_runs.py next to it. Keyless.
import pytest
from debug import detect_failures
from fix_runs import CHECKS, measure
from record_runs import BROKEN

EXPECTED = {"run_a_loop": "loop", "run_b_bad_args": "hallucinated_call",
            "run_c_context_loss": "context_loss"}


@pytest.mark.parametrize("name", list(BROKEN))
def test_detector_catches_the_broken_run(name: str) -> None:
    assert EXPECTED[name] in {f.mode for f in detect_failures(measure(name, fixed=False))}


@pytest.mark.parametrize("name", list(BROKEN))
def test_fixed_run_is_clean_and_correct(name: str) -> None:
    trace = measure(name, fixed=True)
    assert not detect_failures(trace), detect_failures(trace)
    assert trace.stop_reason == "done" and CHECKS[name](trace)


if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-q"]))

......                                                                   [100%]
6 passed in 0.02s