Failure Recovery, Context Discipline & Tracing

# Colab cell 1 — pure Python: runs with no key and no client.
import json
from collections import Counter

class SafeExecutor:
    def __init__(self, impl: dict, max_failures_per_tool: int = 3):
        self.impl = impl
        self.max_failures = max_failures_per_tool
        self.failures = Counter()        # per tool name
        self.failed_calls = set()        # exact (tool, args) repeats

    def execute(self, name: str, args: dict) -> tuple[str, bool]:
        """Returns (content, is_error). Never raises."""
        key = (name, json.dumps(args, sort_keys=True))

        if self.failures[name] >= self.max_failures:
            return (f"Tool '{name}' is disabled after "
                    f"{self.failures[name]} failures this run. "
                    "Use a different tool or approach.", True)
        if key in self.failed_calls:
            return ("You already tried this exact call and it failed. "
                    "Do not repeat it; change the arguments or approach.",
                    True)
        try:
            return (self.impl[name](**args), False)
        except Exception as e:
            self.failures[name] += 1
            self.failed_calls.add(key)
            return (f"{type(e).__name__}: {e}", True)

# in the loop:
#   content, is_error = executor.execute(block.name, block.input)
#   results.append({"type": "tool_result", "tool_use_id": block.id,
#                   "content": content, "is_error": is_error})

# demo: all three defenses fire, no API key needed
flaky = SafeExecutor({"read_file": lambda path: open(path).read()})
for path in ["/no/a", "/no/a", "/no/b", "/no/c", "/no/d"]:
    content, _ = flaky.execute("read_file", {"path": path})
    print(f"{path}: {content[:58]}")

# Colab cell 2 — pure Python: runs with no key and no client.
MAX_TOOL_OUTPUT_CHARS = 4000

def truncate(output: str, limit: int = MAX_TOOL_OUTPUT_CHARS) -> str:
    if len(output) <= limit:
        return output
    dropped = len(output) - limit
    return (output[:limit] +
            f"\n\n[TRUNCATED: {dropped} more characters not shown. "
            "Narrow your grep pattern, or call read_file with an offset "
            "to view a specific region.]")

def compact_old_results(messages: list, keep_last: int = 2,
                        stub_over: int = 1000) -> list:
    """Replace big tool results from old iterations with short stubs.
    The model already extracted what it needed; the bytes are just rent."""
    compacted = []
    cutoff = len(messages) - keep_last * 2   # each iteration = 2 messages
    for idx, msg in enumerate(messages):
        if idx >= cutoff or msg["role"] != "user" or isinstance(msg["content"], str):
            compacted.append(msg)
            continue
        new_content = []
        for part in msg["content"]:
            if (isinstance(part, dict) and part.get("type") == "tool_result"
                    and len(str(part.get("content", ""))) > stub_over):
                new_content.append({**part, "content":
                    "[old tool result elided to save context - "
                    "re-run the tool if you need it again]"})
            else:
                new_content.append(part)
        compacted.append({**msg, "content": new_content})
    return compacted

# demo: six fake iterations shrink; the two most recent stay intact
print(truncate("x" * 9000)[-90:])
history = []
for i in range(6):
    history.append({"role": "assistant", "content": f"(tool call {i})"})
    history.append({"role": "user", "content": [
        {"type": "tool_result", "tool_use_id": f"t{i}", "content": "y" * 5000}]})
before = sum(len(str(m)) for m in history)
after = sum(len(str(m)) for m in compact_old_results(history))
print(f"history: {before:,} chars -> {after:,} chars")

# Colab cell 3 — run cells 1 and 2 first (SafeExecutor, truncate). This
# cell adds the key, the client, and a fake repo, then traces a full run.
!pip install -q anthropic

import os
try:
    from google.colab import userdata
    os.environ["ANTHROPIC_API_KEY"] = userdata.get("ANTHROPIC_API_KEY")
except Exception:
    from getpass import getpass
    os.environ.setdefault("ANTHROPIC_API_KEY", getpass("Anthropic API key: "))

import json, time, uuid
import anthropic

client = anthropic.Anthropic()
MODEL = "claude-sonnet-5"

class Tracer:
    def __init__(self, path: str = "trace.jsonl"):
        self.path = path
        self.run_id = uuid.uuid4().hex[:8]

    def log(self, event: str, **fields) -> None:
        record = {"run_id": self.run_id, "ts": round(time.time(), 3),
                  "event": event, **fields}
        with open(self.path, "a") as f:
            f.write(json.dumps(record, default=str) + "\n")

# the same tiny "repo" as lesson 2, so the tools have something to find
REPO = {
    "README.md": "Sample service. Retry policy is configured in config/app.yaml.",
    "config/app.yaml": "retries: 5\nbackoff: exponential\ntimeout_s: 30\n",
    "src/retry.py": "def backoff(attempt):\n    return min(2 ** attempt, 60)\n",
}

def list_dir(path: str = "") -> str:
    hits = [p for p in REPO if p.startswith(path)]
    return "\n".join(sorted(hits)) if hits else f"Nothing under {path!r}."

def grep(pattern: str) -> str:
    hits = [f"{p}: {line}" for p, body in REPO.items()
            for line in body.splitlines() if pattern.lower() in line.lower()]
    return "\n".join(hits) if hits else f"No lines match {pattern!r}."

def read_file(path: str) -> str:
    return REPO.get(path, f"No file at {path!r}.")

IMPL = {"list_dir": list_dir, "grep": grep, "read_file": read_file}
TOOLS = [
    {"name": "list_dir", "description": "List repo paths under a prefix.",
     "input_schema": {"type": "object",
                      "properties": {"path": {"type": "string"}}, "required": []}},
    {"name": "grep", "description": "Find lines matching a substring.",
     "input_schema": {"type": "object",
                      "properties": {"pattern": {"type": "string"}},
                      "required": ["pattern"]}},
    {"name": "read_file", "description": "Read one file in full, by path.",
     "input_schema": {"type": "object",
                      "properties": {"path": {"type": "string"}},
                      "required": ["path"]}},
]

def run_traced(question: str, max_iterations: int = 10) -> str:
    tracer = Tracer()
    executor = SafeExecutor(IMPL)                     # cell 1
    messages = [{"role": "user", "content": question}]
    for i in range(max_iterations):
        t0 = time.monotonic()
        resp = client.messages.create(model=MODEL, max_tokens=1024,
                                      tools=TOOLS, messages=messages)
        tracer.log("llm_call", iteration=i,
                   input_tokens=resp.usage.input_tokens,
                   output_tokens=resp.usage.output_tokens,
                   stop_reason=resp.stop_reason,
                   latency_ms=round((time.monotonic() - t0) * 1000))
        if resp.stop_reason != "tool_use":
            tracer.log("terminate", reason="natural stop", complete=True,
                       iterations=i)
            return next(b.text for b in resp.content if b.type == "text")
        messages.append({"role": "assistant", "content": resp.content})
        results = []
        for block in resp.content:
            if block.type == "tool_use":
                content, is_error = executor.execute(block.name, block.input)
                content = truncate(content)           # cell 2
                tracer.log("tool_call", iteration=i, tool=block.name,
                           args=block.input, result_chars=len(content),
                           is_error=is_error)
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": content, "is_error": is_error})
        messages.append({"role": "user", "content": results})
    tracer.log("terminate", reason="max iterations", complete=False,
               iterations=max_iterations)
    return "(incomplete: max iterations exceeded)"

print(run_traced("How does this service configure retries?"))
print("--- trace.jsonl ---")
print(open("trace.jsonl").read())