The Full Test Drive: every capability, one by one

from pathlib import Path
import os, sys

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))

LIVE = bool(os.getenv("AWS_BEARER_TOKEN_BEDROCK"))
if not LIVE:
    try:
        from aws_bedrock_token_generator import provide_token
        os.environ["AWS_BEARER_TOKEN_BEDROCK"] = provide_token(region="us-east-1")
        os.environ.setdefault("AWS_REGION_NAME", "us-east-1")
        LIVE = True
        print("✓ short-term Bedrock token generated from AWS credentials")
    except Exception as e:
        print(f"offline mode ({type(e).__name__}) — LIVE sections will be skipped")
else:
    print("✓ using AWS_BEARER_TOKEN_BEDROCK from env")

import shipit_agent
print("shipit_agent", getattr(shipit_agent, "__version__", "(dev)"), "· LIVE =", LIVE)

from shipit_agent import Agent, FunctionTool
from shipit_agent.llms import BedrockChatLLM

def get_time(city: str, **_):
    """Return the local time for a city."""
    return f"It's 3:00 PM in {city}."

def build_agent(model_id: str) -> Agent:
    return Agent(
        llm=BedrockChatLLM(model=model_id, region="us-east-1"),
        tools=[FunctionTool.from_callable(get_time, name="get_time")],
        auto_use_skills=False,
    )

MODELS = ["google.gemma-4-26b-a4b", "bedrock/openai.gpt-oss-120b-1:0"]

if LIVE:
    for model in MODELS:
        print(f"\n════ {model} ════")
        answer = build_agent(model).run_live(
            "What time is it in Tokyo? Use the tool, one sentence."
        )
else:
    print("skipped (offline)")

from shipit_agent import StreamRenderer, format_event_line

if LIVE:
    agent = build_agent("google.gemma-4-26b-a4b")
    print("— raw events —")
    events = []
    for event in agent.stream("What time is it in Paris? Use the tool."):
        events.append(event)
        print(f"  {event.type}")
    print("\n— rendered with StreamRenderer(style='rich') —")
    r = StreamRenderer(style="rich", show_summary=True)
    for e in events:
        r.feed(e)
    r.close()
else:
    print("skipped (offline)")

from shipit_agent import format_activity
from shipit_agent.llms.base import LLMResponse, ToolCall

class ScriptedLLM:
    """Offline: one tool call, then an answer (used in offline sections)."""
    def __init__(self, name="add", args=None, answer="The sum is 5."):
        self.turn, self.name, self.args, self.answer = 0, name, args or {"a":2,"b":3}, answer
    def complete(self, *, messages, tools=None, **_):
        self.turn += 1
        if self.turn == 1:
            return LLMResponse(tool_calls=[ToolCall(name=self.name, arguments=self.args)])
        return LLMResponse(content=self.answer)

def add(a: int, b: int, **_): return str(a + b)

offline_agent = Agent(llm=ScriptedLLM(), tools=[FunctionTool.from_callable(add, name="add")],
                      auto_use_skills=False)
result = offline_agent.run("2+3?")

called = next(e for e in result.events if e.type == "tool_called")
done   = next(e for e in result.events if e.type == "tool_completed")
assert called.payload["call_id"] == done.payload["call_id"]
print("call_id pairing ✓ ·", called.payload["call_id"])
print()
print(format_activity(result))
print()
print("summary():", result.summary())

for role in ("finance-analyst", "marketing-writer", "researcher",
             "figma-designer", "sales-rep", "generalist-developer"):
    a = Agent.for_role(role, llm=ScriptedLLM())
    print(f"{role:<24} {len(a.tools):>2} tools · {a.metadata['category']}")

try:
    Agent.for_role("finance", llm=ScriptedLLM())
except ValueError as e:
    print("\ndid-you-mean ✓ →", e)

if LIVE:
    analyst = Agent.for_role(
        "finance-analyst",
        # 31B is the reliable choice for structured tool arguments;
        # 26B is faster but occasionally mangles nested JSON.
        llm=BedrockChatLLM(model="google.gemma-4-31b", region="us-east-1"),
    )
    live_result = analyst.run(
        "Use build_document: xlsx titled 'Deep Test' with sheet 'P&L', "
        "headers Item, Amount; rows Revenue 124000, Costs -78500, and Net "
        "with formula =B2+B3. Confirm briefly."
    )
    print(format_activity(live_result))

    import openpyxl
    path = [t.metadata["path"] for t in live_result.tool_results if t.metadata.get("path")][0]
    wb = openpyxl.load_workbook(path)
    ws = wb[wb.sheetnames[0]]          # whatever the model named it
    formulas = [cell.value for row in ws.iter_rows() for cell in row
                if isinstance(cell.value, str) and cell.value.startswith("=")]
    print("\nverify:", wb.sheetnames, "· header bold =", ws["A1"].font.bold,
          "· frozen =", ws.freeze_panes, "· live formulas =", formulas)
else:
    print("skipped (offline)")

import tempfile
from shipit_agent.tools import DocumentBuilderTool
from shipit_agent.tools.base import ToolContext

outdir = tempfile.mkdtemp(prefix="shipit_docs_")
tool = DocumentBuilderTool(workspace_root=outdir)
ctx = ToolContext(prompt="", system_prompt="", state={})

SECTIONS = [{"heading": "Highlights", "body": "Revenue grew 24% QoQ.",
             "bullets": ["ARR up 24%", "Churn 1.1%"],
             "table": {"headers": ["Metric", "Q1", "Q2"],
                        "rows": [["Revenue", 100, 124], ["Churn %", 1.4, 1.1]]}}]
SHEETS = [{"name": "P&L", "headers": ["Item", "Amount"],
           "rows": [["Revenue", 124000], ["Costs", -78500], ["Net", "=B2+B3"]]}]

for kind in ("html", "pdf", "xlsx", "docx", "pptx"):
    payload = {"sheets": SHEETS} if kind == "xlsx" else {"sections": SECTIONS}
    out = tool.run(ctx, kind=kind, title=f"Report {kind.upper()}", **payload)
    mark = "✓" if out.metadata.get("ok") else "✗"
    print(f"{mark} {kind:<5} {out.text.splitlines()[0]}")

# error paths, in depth
print()
print("unknown kind →", tool.run(ctx, kind="gif", title="x").text)

from shipit_agent import connect_mcp, list_mcp_catalog

for e in list_mcp_catalog():
    need = f"  (needs {', '.join(e.required_env)})" if e.required_env else ""
    print(f"{e.name:<14} {e.description}{need}")

# fail-fast validation
os.environ.pop("SLACK_BOT_TOKEN", None)
try:
    connect_mcp("slack")
except ValueError as e:
    print("\nfail-fast ✓ →", e)

import shutil

if shutil.which("npx"):
    fs = connect_mcp("filesystem", args=[str(ROOT / "notebooks")])
    try:
        tools = fs.discover_tools()
        print(f"live MCP tools: {len(tools)} —", [t.name for t in tools[:5]], "…")

        # resources API (this server exposes none — graceful empty, no error)
        print("resources:", fs.list_resources() or "(none exposed — graceful)")
        print("prompts:  ", fs.list_prompts() or "(none exposed — graceful)")

        # the model-facing resource browser tool
        rt = fs.resource_tool()
        print("resource_tool:", rt.name, "→", rt.run(ctx).text[:70])

        # call a real MCP tool directly
        list_dir = next(t for t in tools if t.name == "list_directory")
        out = list_dir.run(ctx, path=str(ROOT / "notebooks"))
        print("list_directory →", out.text[:100].replace("\n", " "), "…")
    finally:
        fs.close()
else:
    print("npx not on PATH — skipped")

from shipit_agent import Agent
from shipit_agent.permissions import PermissionDecision, PermissionEngine, PermissionResult

def bash_stub(command: str = "", **_): return f"ran: {command}"

# deny rule blocks the call; the model gets a readable refusal
denier = Agent(
    llm=ScriptedLLM(name="bash", args={"command": "rm -rf /"}, answer="ok"),
    tools=[FunctionTool.from_callable(bash_stub, name="bash")],
    permissions=PermissionEngine(deny=["bash"]),
    auto_use_skills=False,
)
res = denier.run("delete everything")
blocked = [m for m in res.messages if m.metadata.get("error") == "permission_denied"]
print("deny rule ✓ →", blocked[0].content[:70])

# ask + callback = human-in-the-loop (here: auto-approve)
approvals = []
def approve(name, args):
    approvals.append(name)
    return PermissionResult(decision=PermissionDecision.ALLOW, reason="demo approve")

asker = Agent(
    llm=ScriptedLLM(name="bash", args={"command": "ls"}, answer="done"),
    tools=[FunctionTool.from_callable(bash_stub, name="bash")],
    permissions=PermissionEngine(ask=["bash"], callback=approve),
    auto_use_skills=False,
)
res2 = asker.run("list files")
print("ask+callback ✓ → approved:", approvals, "· ran:", res2.tool_results[0].output)

class LoopingLLM:
    def complete(self, *, messages, tools=None, **_):
        return LLMResponse(tool_calls=[ToolCall(name="tick", arguments={})])

agent = Agent(llm=LoopingLLM(), tools=[], auto_use_skills=False, max_iterations=50)

def tick(**_):
    agent.cancel()          # ← ESC
    return "tick"

agent.tools.append(FunctionTool.from_callable(tick, name="tick"))
res = agent.run("loop forever")
print("events:", [e.type for e in res.events])
print("iterations used:", sum(1 for e in res.events if e.type == "step_started"), "of 50")
print("cancelled flag:", res.events[-1].payload.get("cancelled"))

import time as _time
from shipit_agent.tools.edit_file import EditFileTool
from shipit_agent.tools.file_read import FileReadTool

work = Path(tempfile.mkdtemp(prefix="shipit_edit_"))
(work / "app.py").write_text("def greet():\n    return 'hello'\n")
read, edit = FileReadTool(root_dir=str(work)), EditFileTool(root_dir=str(work))
ectx = ToolContext(prompt="", system_prompt="", state={})

print("1) edit without read →", edit.run(ectx, path="app.py", old_text="hello", new_text="hi").text[:60])

read.run(ectx, path="app.py")
out = edit.run(ectx, path="app.py", old_text="'hello'", new_text="'hi there'")
print("\n2) after read — patched with diff:")
print(out.text)

# external change → blocked
(work / "app.py").write_text("def greet():\n    return 'CHANGED OUTSIDE'\n")
os.utime(work / "app.py", ns=(_time.time_ns(), _time.time_ns() + 1_000_000))
print("\n3) external change →", edit.run(ectx, path="app.py", old_text="CHANGED", new_text="x").text[:80])

from shipit_agent.models import Message
from shipit_agent.runtime import AgentRuntime

class SummarizerLLM:
    def complete(self, *, messages, tools=None, metadata=None, **_):
        if metadata and metadata.get("purpose") == "context_compaction":
            return LLMResponse(content="User asked to add numbers; ran add(1,2)=3 twice; awaiting next step.")
        return LLMResponse(content="done")

rt = AgentRuntime(llm=SummarizerLLM(), prompt="p", context_window_tokens=100)
history = [Message(role="system", content="sys"),
           Message(role="user", content="x"*300), Message(role="assistant", content="a"*300),
           Message(role="tool", name="add", content="y"*300), Message(role="user", content="more"),
           Message(role="assistant", content="b"*300), Message(role="tool", name="add", content="z"*300),
           Message(role="user", content="latest")]
compacted, did = rt._compact_messages(history)
print("compacted:", did, f"· {len(history)} → {len(compacted)} messages")
print("model-written summary →", next(m for m in compacted if m.metadata.get("compacted")).content)

from shipit_agent import AgentScheduler, SQLiteJobStore

class Clock:
    def __init__(self): self.now = 1_000_000.0
    def __call__(self): return self.now
    def advance(self, s): self.now += s

class TinyAgent:
    def run(self, prompt): return type("R", (), {"metadata": {}})()

clock, fired = Clock(), []
sched = AgentScheduler(TinyAgent(), clock=clock, sleep=lambda s: None)
sched.add("hourly digest", every=3600, on_result=lambda r: fired.append("hourly"))
sched.add("morning post", at="09:00", on_result=lambda r: fired.append("daily"))
for _ in range(26):
    clock.advance(3600); sched.run_pending()
print(f"26 simulated hours → hourly ×{fired.count('hourly')}, daily ×{fired.count('daily')}")

# durable: run counts + next_run survive a restart
db = str(Path(tempfile.mkdtemp()) / "jobs.db")
c2 = Clock()
s1 = AgentScheduler(TinyAgent(), clock=c2, sleep=lambda s: None, store=SQLiteJobStore(db))
j = s1.add("ping", every=60, name="j1"); c2.advance(60); s1.run_pending()
s2 = AgentScheduler(TinyAgent(), clock=c2, sleep=lambda s: None, store=SQLiteJobStore(db))
restored = s2.add("ping", every=60, name="j1")
print(f"after 'restart': runs={restored.runs} (persisted ✓), next_run resumed ✓")

from shipit_agent.tools.sub_agent import SubAgentTool

class SlowLLM:
    def complete(self, **_):
        _time.sleep(0.1)
        return LLMResponse(content="research finding: agents are eating software")

sub = SubAgentTool(SlowLLM())
started = sub.run(ctx, task="research trends", background=True)
print("started →", started.text[:70])
tid = started.metadata["task_id"]
print("…parent keeps working while the subagent runs…")
print("collect →", sub.run(ctx, collect=tid).text)