The Complete Tour

from pathlib import Path
import sys

# Resolve the repo root whether this cell is running from ./notebooks
# or from the repo root — mirrors the existing 01–36 notebook series.
ROOT = (
    Path.cwd().resolve().parent
    if Path.cwd().name == "notebooks"
    else Path.cwd().resolve()
)
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from examples.run_multi_tool_agent import build_llm_from_env

# Bedrock Llama 4 Scout is the default model this library ships with
# (see `shipit_agent/config.py`). No model name required — the helper
# reads `AWS_REGION` / credentials from env and returns an adapter that
# works with every Autopilot / Agent class.
llm = build_llm_from_env("bedrock")
print("Bedrock LLM ready:", type(llm).__name__)

from shipit_agent.agents import AgentRegistry

registry = AgentRegistry.default()
agents = registry.all()
print(f"{len(agents)} agents available")

by_cat: dict[str, list[str]] = {}
for a in agents:
    by_cat.setdefault(a.category, []).append(a.id)

for cat in sorted(by_cat):
    print(f"\n{cat} ({len(by_cat[cat])})")
    for aid in by_cat[cat][:8]:
        print(f"  · {aid}")
    if len(by_cat[cat]) > 8:
        print(f"  … +{len(by_cat[cat]) - 8} more")

from shipit_agent.autopilot import Autopilot, BudgetPolicy, default_heartbeat_stderr
from shipit_agent.deep import Goal

basic = Autopilot(
    llm=llm,
    goal=Goal(
        objective="Explain list vs tuple in Python with a 3-line example.",
        success_criteria=[
            "Mentions mutability",
            "Shows a tuple literal",
            "Shows a list literal",
        ],
    ),
    budget=BudgetPolicy(max_iterations=4, max_seconds=180),
    on_heartbeat=default_heartbeat_stderr,
)
r = basic.run(run_id="tour-basics")
print(f"status={r.status} iters={r.iterations} halt={r.halt_reason}")
print(r.output[:500])

autopilot_stream = Autopilot(
    llm=llm,
    goal=Goal(
        objective="Write a two-line haiku about shipit_agent.",
        success_criteria=["Two lines"],
    ),
    budget=BudgetPolicy(max_iterations=3, max_seconds=120),
)
for ev in autopilot_stream.stream(run_id="tour-stream"):
    kind = ev["kind"]
    if kind == "autopilot.iteration":
        met = ev["criteria_met"]
        u = ev["usage"]
        print(
            f"  iter {ev['iteration']} · {sum(met)}/{len(met)} met · {u['seconds']:.0f}s"
        )
    elif kind in (
        "autopilot.run_started",
        "autopilot.result",
        "autopilot.criteria_satisfied",
    ):
        print(
            f"  [{kind}] {ev.get('status') or ev.get('goal', {}).get('objective', '')}"
        )

from shipit_agent.live_renderer import render_stream

autopilot_r = Autopilot(
    llm=llm,
    goal=Goal(objective="2+2?", success_criteria=["Answer is 4"]),
    budget=BudgetPolicy(max_iterations=2, max_seconds=60),
)
# `plain` is notebook-friendly — no ANSI escapes.
render_stream(autopilot_r.stream(run_id="tour-render"), fmt="plain")

# First attempt with a deliberately tiny budget — halts partial.
first = Autopilot(
    llm=llm,
    goal=Goal(
        objective="Describe SOLID principles with one sentence each.",
        success_criteria=[f"Describes {p}" for p in ("S", "O", "L", "I", "D")],
    ),
    budget=BudgetPolicy(max_iterations=1),
).run(run_id="tour-resume")
print(f"first: {first.status} iters={first.iterations}")

# Resume with a larger budget — keeps going.
second = Autopilot(
    llm=llm,
    goal=Goal(
        objective="Describe SOLID principles with one sentence each.",
        success_criteria=[f"Describes {p}" for p in ("S", "O", "L", "I", "D")],
    ),
    budget=BudgetPolicy(max_iterations=6),
).resume("tour-resume")
print(f"second: {second.status} iters={second.iterations}")

from shipit_agent.scheduler_daemon import SchedulerDaemon

daemon = SchedulerDaemon(llm_factory=lambda: llm, tick_seconds=3)
daemon.enqueue(
    run_id="tour-q-1",
    objective="Write one sentence about Python's walrus operator.",
    success_criteria=["Mentions :="],
    budget={"max_iterations": 3, "max_seconds": 120},
)
for entry in daemon.list_queue():
    print(f"{entry.run_id:<20} [{entry.status}] {entry.objective[:50]}")

autopilot_c = Autopilot(
    llm=llm,
    goal=Goal(
        objective="Name three HTTP status codes and what they mean.",
        success_criteria=["At least 3 codes", "Each code has a one-line meaning"],
    ),
    budget=BudgetPolicy(max_iterations=5, max_seconds=180),
    critic=True,
)
r = autopilot_c.run(run_id="tour-critic")
print(f"status={r.status} halt={r.halt_reason}")
print("verdict:", {k: r.critic_verdict.get(k) for k in ("confidence", "reasoning")})

from shipit_agent.autopilot import ArtifactCollector

collector = ArtifactCollector()
autopilot_a = Autopilot(
    llm=llm,
    goal=Goal(
        objective="Write a Python function `fib(n)` that returns the n-th Fibonacci number.",
        success_criteria=["A `def fib(n)` appears in a fenced code block"],
    ),
    budget=BudgetPolicy(max_iterations=4, max_seconds=180),
    artifacts=collector,
    critic=True,
)
r = autopilot_a.run(run_id="tour-artifacts")
print(f"artifacts extracted: {len(r.artifacts)}")
for a in r.artifacts:
    print(f"  [{a['kind']:<9}] {a['name']} — {len(a['content'])} chars")

parent = Autopilot(
    llm=llm,
    goal=Goal(objective="batch-tour", success_criteria=["done"]),
    budget=BudgetPolicy(
        max_seconds=600, max_iterations=6, max_tool_calls=20, max_dollars=3.0
    ),
)
r = parent.fanout(
    items=["threading", "asyncio", "multiprocessing"],
    objective_template="Write a one-paragraph overview of Python's {item} module.",
    criteria_template=["Mentions typical use case"],
    max_parallel=3,
    child_budget_frac=0.3,
)
print(f"fan-out: {r.status} · wall: {r.wall_seconds:.1f}s")
for c in r.children:
    print(f"  {c['run_id']:<44} {c['status']:<10} {c['iterations']} iters")

events = list(
    parent.fanout_stream(
        items=["A", "B"],
        objective_template="Summarize module {item}",
        max_parallel=2,
        child_budget_frac=0.2,
    )
)
for ev in events:
    if ev["kind"] == "autopilot.fanout_child":
        print(f"  child {ev['item_index']}: {ev['status']} ({ev['iterations']} iters)")
    elif ev["kind"] == "autopilot.fanout_result":
        print(f"  final: {ev['status']}")

from shipit_agent.tools.computer_use import ComputerUseTool
from shipit_agent.tools.hubspot import HubspotTool
from shipit_agent.tools.research_brief import ResearchBriefTool

cu = ComputerUseTool()
hs = HubspotTool()
rb = ResearchBriefTool()

for tool in (cu, hs, rb):
    actions = (
        tool.schema()["function"]["parameters"]["properties"]
        .get("action", {})
        .get("enum", [])
    )
    print(f"{tool.name:<20} actions={len(actions)} — {actions[:6]}…")

from shipit_agent.tools.base import ToolContext

out = rb.run(
    ToolContext(prompt="demo"), query="postgres connection pooling 2026", max_sources=3
)
print(out.text[:900])

out = cu.run(ToolContext(prompt="demo"), action="screenshot")
print(out.text)

from shipit_agent import Agent
from shipit_agent.builtins import get_builtin_tools

dev = Agent(
    llm=llm,
    prompt=registry.get("generalist-developer").prompt,
    tools=get_builtin_tools(project_root="."),
    max_iterations=20,
    name="Generalist Developer",
)
result = dev.run(
    "Add a one-line docstring to the function `BudgetPolicy.exceeded` in shipit_agent/autopilot/budget.py. Do not change behavior."
)
print(result.output[:600])

from shipit_agent.agents import AgentRegistry

researcher = AgentRegistry.default().get("researcher")

research_pilot = Autopilot(
    llm=llm,
    goal=Goal(
        objective="Compile a one-page digest on Python type checkers (mypy / pyright / pyre).",
        success_criteria=[
            "At least 3 type checkers covered",
            "Each has one-line tradeoff",
            "Ends with a recommendation paragraph",
        ],
    ),
    tools=[rb],
    budget=BudgetPolicy(max_iterations=5, max_seconds=480, max_tool_calls=15),
    critic=True,
    artifacts=True,
    prompt=researcher.prompt,  # use the researcher's persona as the inner-agent prompt
)
print("Ready. Call .run(run_id='type-checker-digest') when you want it to execute.")

mega_pilot = Autopilot(
    llm=llm,
    goal=Goal(objective="mega-tour", success_criteria=["done"]),
    budget=BudgetPolicy(
        max_seconds=600, max_iterations=6, max_tool_calls=30, max_dollars=3.0
    ),
    critic=True,
    artifacts=True,
    tools=[rb],
    prompt=registry.get("generalist-developer").prompt,
)
# Fan out three small jobs under the same critic+artifact umbrella.
mega_result = mega_pilot.fanout(
    items=["dataclasses", "enum", "pathlib"],
    objective_template="Give a 50-word description of Python's `{item}` module with one code snippet.",
    criteria_template=["Under 100 words of prose", "One fenced code block"],
    max_parallel=3,
    child_budget_frac=0.25,
)
print(
    f"mega result: {mega_result.status}, {len(mega_result.children)} children, {len(mega_result.failed)} failed"
)