Architecture & Codebase Exploration

import subprocess, pathlib

REPO = pathlib.Path("/sandbox/repo")

def search_symbol(pattern: str, max_results: int = 40) -> str:
    """Grep the repo for a symbol or error string. Ripgrep if available."""
    try:
        out = subprocess.run(
            ["rg", "-n", "--max-count", "3", pattern, str(REPO)],
            capture_output=True, text=True, timeout=20,
        ).stdout
    except FileNotFoundError:
        out = subprocess.run(
            ["grep", "-rn", pattern, str(REPO)],
            capture_output=True, text=True, timeout=20,
        ).stdout
    lines = out.splitlines()[:max_results]
    return "\n".join(lines) or "no matches"

def list_dir(rel: str = ".") -> str:
    target = (REPO / rel).resolve()
    # Constrain to the repo — never let the agent wander the filesystem.
    if REPO not in target.parents and target != REPO:
        return "error: path escapes the repo"
    entries = sorted(p.name + ("/" if p.is_dir() else "") for p in target.iterdir())
    return "\n".join(entries)

def read_file(rel: str, start: int = 1, end: int = 400) -> str:
    target = (REPO / rel).resolve()
    if REPO not in target.parents and target != REPO:
        return "error: path escapes the repo"
    if not target.is_file():
        return f"error: {rel} is not a file"
    text = target.read_text(errors="replace").splitlines()
    window = text[start - 1:end]
    # Line numbers help the model reference and later edit precisely.
    return "\n".join(f"{i + start:>5}  {ln}" for i, ln in enumerate(window))

import json, pathlib

PLAN_PATH = pathlib.Path("/sandbox/plan.json")

def explore_and_plan(issue_text: str) -> dict:
    system = (
        "You are a senior engineer triaging a bug. Use search_symbol, "
        "list_dir, and read_file to locate the relevant code. Do NOT guess "
        "at file contents — read them. When confident, call record_plan with "
        "the files you will edit, your understanding of the bug, and the fix "
        "approach. Read only what you need; the repo is large."
    )
    messages = [{"role": "user", "content": f"Issue:\n{issue_text}"}]
    while True:
        resp = call_model(system, EXPLORE_TOOLS, messages)   # search/list/read/record_plan
        if resp.stop_reason != "tool_use":
            continue
        messages.append({"role": "assistant", "content": resp.content})
        results = []
        plan = None
        for block in resp.content:
            if block.type != "tool_use":
                continue
            if block.name == "record_plan":
                plan = block.input          # {"files":[...], "bug":"...", "fix":"..."}
            else:
                results.append(run_explore_tool(block))
        if plan is not None:
            # Checkpoint: survives a crash so W22's implement stage can resume.
            PLAN_PATH.write_text(json.dumps(plan, indent=2))
            return plan
        messages.append({"role": "user", "content": results})