Live streaming — every v1.1.0 feature against Gemma 4 on Bedrock

import importlib.util
import sys
from pathlib import Path

PROVIDER = Path(
    "~/Documents/MYWORK/AFTDRK/CACHE/DRK_CACHE_BACK"
    "/drk_cache/llm/bedrock_mantle_provider.py"
)
MODEL = "bedrock-mantle/google.gemma-4-26b-a4b"

spec = importlib.util.spec_from_file_location("bedrock_mantle_provider", PROVIDER)
module = importlib.util.module_from_spec(spec)
sys.modules["bedrock_mantle_provider"] = module
spec.loader.exec_module(module)
module.ensure_registered()

from shipit_agent.llms import LiteLLMChatLLM


def llm():
    """A fresh adapter per agent — they hold no shared state, but it keeps
    each section independent if you re-run cells out of order."""
    return LiteLLMChatLLM(model=MODEL)


import boto3
print("identity :", boto3.client("sts").get_caller_identity()["Arn"])
print("model    :", MODEL)

# A few fake tools so the sections below are deterministic and free of
# side effects. Real tools work identically — these just record their calls.
from shipit_agent.tools.base import ToolOutput

CALLS = []


def tool(name, output="ok", *, description=None, sensitive=False,
         properties=None, required=None, credential_key=None):
    """Build a one-off tool. `sensitive=True` triggers lockdown (§5)."""

    class T:
        def __init__(self):
            self.name = name
            self.description = description or f"The {name} tool."
            self.prompt_instructions = ""
            if credential_key:
                self.credential_key = credential_key

        def schema(self):
            return {
                "type": "function",
                "function": {
                    "name": name,
                    "description": self.description,
                    "parameters": {
                        "type": "object",
                        "properties": properties or {
                            "path": {"type": "string", "description": "File path"}
                        },
                        "required": required if required is not None else ["path"],
                    },
                },
            }

        def run(self, context, **kwargs):
            CALLS.append((name, kwargs))
            metadata = {}
            if sensitive:
                metadata = {"sensitive": True, "sensitive_reason": "customer PII"}
            return ToolOutput(text=output, metadata=metadata)

    return T()


print("helpers ready")

from shipit_agent import Agent
from shipit_agent.builtins import get_builtin_tool_map
from shipit_agent.permissions import PermissionDecision, PermissionEngine

def _repo_root(start: Path) -> Path:
    """Walk up to the repo. A notebook's cwd is `notebooks/`, not the root,
    so `Path.cwd()` points the file tools one directory too deep and every
    read comes back "File not found"."""
    for candidate in [start, *start.parents]:
        if (candidate / "shipit_agent" / "__init__.py").exists():
            return candidate
    raise RuntimeError(f"Could not locate the shipit_agent repo from {start}")


REPO = str(_repo_root(Path.cwd()))
BUILTINS = get_builtin_tool_map(llm=None, project_root=REPO)

# The reading tools, pointed at this repo.
READERS = [BUILTINS[n] for n in
           ("read_file", "grep_files", "glob_files", "workspace_files")]

# NOTE: `deny` outranks `allow`, so `deny=["*"]` would deny the readers too.
# To mean "these and nothing else", allow-list them and flip the default.
READ_ONLY = PermissionEngine(
    allow=["read_file", "grep_files", "glob_files", "workspace_files"],
    default_decision=PermissionDecision.DENY,
)

print("repo  :", REPO)
print("tools :", [t.name for t in READERS])
print("policy: read anything, change nothing")
for name in ("read_file", "glob_files", "bash", "write_file"):
    print(f"        {name:14} {READ_ONLY.check(name, {}).decision.value}")

from shipit_agent.narrate import watch

agent = Agent(
    llm=llm(), tools=READERS, permissions=READ_ONLY,
    auto_use_skills=False, max_iterations=6,
)

# output_limit=None: every call's real output, in full. Each one scrolls in
# its own box, so nothing is hidden and the page does not grow without bound.
watch(
    agent,
    "Use glob_files with pattern 'shipit_agent/narrate/*.py' to list the "
    "narrate modules, then use read_file on shipit_agent/narrate/verbs.py "
    "and tell me in two sentences what it does.",
    title="Exploring narrate/",
    output_limit=None,
)

agent = Agent(llm=llm(), tools=READERS, permissions=READ_ONLY,
              auto_use_skills=False, max_iterations=6)

for event in agent.stream(
    "Use grep_files with pattern='def summarize' and path='shipit_agent' to "
    "find it, then read_file the file it is in. Say in one line what the "
    "function does."
):
    payload = event.payload

    if event.type == "text_delta":
        # The answer arrives here, token by token. A loop without this branch
        # sits silent through the whole reply and then prints nothing.
        print(payload["chunk"], end="", flush=True)

    elif event.type == "tool_called":
        args = ", ".join(f"{k}={v!r}" for k, v in payload["arguments"].items())
        print(f"\n\n→ {payload['tool']}({args})")

    elif event.type == "tool_completed":
        body = str(payload.get("output", ""))
        print(f"← {payload['duration_ms']:.0f}ms · {len(body)} bytes")
        print(body)          # in full — no slicing, no "… N more lines"
        print()

    elif event.type == "tool_failed":
        print(f"\n✗ {payload.get('error', '')}")

    elif event.type == "tool_denied":
        print(f"\n⊘ blocked: {payload.get('reason', '')}")

agent = Agent(llm=llm(), tools=[BUILTINS["bash"], *READERS],
              permissions=READ_ONLY, auto_use_skills=False, max_iterations=4)

result = agent.run("Run 'rm -rf /tmp/demo' using the bash tool.")

for message in result.messages:
    if message.role == "tool":
        print("the model was told:")
        print(message.content)          # the full text, not the first 200 chars
        print()

print("denials :", [e.payload["tool"] for e in result.events
                    if e.type == "tool_denied"])
print()
print("answer  :")
print(result.output)

from shipit_agent import Agent

CALLS.clear()
agent = Agent(
    llm=llm(),
    tools=[
        tool("read_file", "def login(): pass  # TODO: add MFA",
             description="Read a file from disk"),
        tool("grep_files", "1 match: TODO",
             description="Search files for a pattern",
             properties={"pattern": {"type": "string", "description": "Search pattern"}},
             required=["pattern"]),
    ],
    auto_use_skills=False,
    max_iterations=5,
)

agent.run_live(
    "Read auth.py with read_file, then grep for 'TODO' with grep_files. "
    "Then summarise in one sentence.",
    style="modern",
)
print("\ntools that actually ran:", [n for n, _ in CALLS])

agent = Agent(llm=llm(), tools=READERS, permissions=READ_ONLY,
              auto_use_skills=False, max_iterations=6)

agent.run_live(
    "Use glob_files with pattern='shipit_agent/approvals/*.py', then read_file "
    "on shipit_agent/approvals/models.py. Say in one sentence what it does.",
    style="tree",
)

# Or render a finished run — no escape codes, paste-able anywhere.
from shipit_agent.narrate import render_tree

events = list(agent.stream(
    "Use grep_files with pattern='class ToolContract' and path='shipit_agent'. "
    "One sentence on what you found."
))
print(render_tree(events, model="gemma-4-26b"))

CALLS.clear()
a = Agent(llm=llm(), tools=[tool("read_file", "def login(): pass")],
          auto_use_skills=False, max_iterations=4)

print("── agent.stream() — raw events ──")
kinds = {}
for event in a.stream("Read auth.py using read_file, then say what it does in one line."):
    kinds[event.type] = kinds.get(event.type, 0) + 1
for name, count in sorted(kinds.items()):
    print(f"   {name:22} x{count}")

from shipit_agent.narrate import ApprovalRow, ProseRow, SubAgentRow, WorkRow

CALLS.clear()
a = Agent(llm=llm(), tools=[tool("read_file", "def login(): pass")],
          auto_use_skills=False, max_iterations=4)

print("── agent.narrate() — settled rows ──")
for row in a.narrate("Read auth.py using read_file and summarise it in one line."):
    if isinstance(row, WorkRow):
        print(f"   WORK    {row.group.icon} {row.group.label}")
        for line in row.group.detail_lines:
            print(f"           {line}")
    elif isinstance(row, ProseRow):
        print(f"   PROSE   {row.text[:88]}")
    elif isinstance(row, ApprovalRow):
        print(f"   APPROVE {row.title}  [{row.tag}]")
    elif isinstance(row, SubAgentRow):
        print(f"   CHILD   {row.label}  ({row.task})")

import json

CALLS.clear()
a = Agent(llm=llm(), tools=[tool("read_file", "def login(): pass")],
          auto_use_skills=False, max_iterations=4)

print("── agent.stream_sse() — wire format ──")
for chunk in a.stream_sse("Read auth.py using read_file. One line."):
    head = chunk.split("\n")[1] if "\n" in chunk else chunk
    if "[DONE]" in chunk:
        print("   done")
        break
    body = json.loads(chunk.split("data: ", 1)[1])
    print(f"   {head:28} durability={body['durability']}")

import tempfile

from IPython.display import HTML

from shipit_agent import ApprovalQueue
from shipit_agent.builtins import get_builtin_tool_map
from shipit_agent.narrate import render_chat_html
from shipit_agent.permissions import PermissionEngine

SCRATCH = Path(tempfile.mkdtemp(prefix="shipit-demo-"))
scratch_tools = get_builtin_tool_map(llm=None, project_root=str(SCRATCH))
writer = scratch_tools["write_file"]

queue = ApprovalQueue()
agent = Agent(
    llm=llm(), tools=[writer], approvals=queue,
    permissions=PermissionEngine(ask=["write_file"]),
    auto_use_skills=False, max_iterations=4,
)

WRITE_TASK = ("Use write_file to create hello.py containing a Python function "
              "that prints 'hello world'.")
write_result = agent.run(WRITE_TASK)

# `run()` keeps the events, so the panel is drawn from exactly what happened.
HTML(render_chat_html(write_result.events, prompt=WRITE_TASK, model=MODEL,
                      title="A write, held for review", output_limit=None))

print("scratch dir  :", SCRATCH)
print("files on disk:", [f.name for f in SCRATCH.iterdir()] or "none")
print("queued       :", [a.title for a in queue.pending()] or "nothing")
print()

for message in write_result.messages:
    if message.role == "tool":
        print("the model was told:", message.content)

print()
print("what the model then said:")
print(write_result.output)

# There is nothing to approve — the call never entered the queue. To get the
# file written, the human decides *first* and the agent runs with the wider
# policy. That ordering is the whole point of `await_decision`.
open_policy = PermissionEngine(allow=["write_file"])
agent2 = Agent(llm=llm(), tools=[writer], permissions=open_policy,
               auto_use_skills=False, max_iterations=4)
agent2.run(WRITE_TASK)

print("files on disk:", [f.name for f in SCRATCH.iterdir()] or "none")
written = SCRATCH / "hello.py"
if written.exists():
    print()
    print(written.read_text())

from shipit_agent import ApprovalQueue
from shipit_agent.permissions import PermissionEngine

CALLS.clear()
queue = ApprovalQueue()
a = Agent(
    llm=llm(),
    tools=[tool("slack", "posted", description="Post a message to Slack",
                properties={"channel": {"type": "string"},
                            "text": {"type": "string"}},
                required=["channel", "text"], credential_key="slack")],
    approvals=queue,
    permissions=PermissionEngine(ask=["slack"]),
    auto_use_skills=False, max_iterations=4,
)

result = a.run("Post 'build is green' to the #eng channel using the slack tool.")

print("during the run   :", [n for n, _ in CALLS] or "nothing sent")
print("queued           :", [action.title for action in queue.pending()])
print("agent said       :", result.output[:100])

# Now decide. approve / deny one at a time, or the whole batch.
for action in queue.pending():
    print(f"#{action.id}  {action.title}")
    print(f"     kind      : {action.kind_label}  ({action.tag})")
    print(f"     revertible: {action.contract.implements_revert}")
    print(f"     arguments : {action.arguments}")
    print()
    print("     " + str(action.description).replace("\n", "\n     "))
    print()

approved = queue.approve_all(by="you")
print("applied now  :", [n for n, _ in CALLS])
print("states       :", [(a.id, a.state.value) for a in approved])

# "Always approve this kind" — an auto-approval rule keyed on the action tag.
# BOTH signals are required: the tool's contract must mark the kind
# auto-approvable AND you must enable the rule. Neither alone is enough.
from shipit_agent.tools.contracts import CONTRACTS

queue2 = ApprovalQueue()
queue2.enable_auto(CONTRACTS["slack"].action_kind, by="you")

CALLS.clear()
a = Agent(llm=llm(),
          tools=[tool("slack", "posted", properties={"text": {"type": "string"}},
                      required=["text"], credential_key="slack")],
          approvals=queue2, permissions=PermissionEngine(ask=["slack"]),
          auto_use_skills=False, max_iterations=4)
a.run("Post 'deploy finished' using the slack tool.")

print("rule enabled  :", [r.label for r in queue2.rules()])
print("ran during run:", [n for n, _ in CALLS], "<- applied automatically")
print("still pending :", queue2.pending())

CALLS.clear()
a = Agent(
    llm=llm(),
    tools=[
        tool("read_customers", "[email], [email]", sensitive=True,
             description="Read the customer list",
             properties={"segment": {"type": "string"}}, required=[]),
        tool("slack", "posted", description="Post to Slack",
             properties={"text": {"type": "string"}}, required=["text"]),
    ],
    permissions=PermissionEngine(allow=["*"]),   # even this cannot override it
    auto_use_skills=False, max_iterations=5,
)

result = a.run("Use read_customers to get the list, then post it to #general with slack.")

print("tools that ran :", [n for n, _ in CALLS])
print("               ^ slack is absent — the send was blocked")
for event in result.events:
    if event.type == "lockdown_engaged":
        print("lockdown       :", event.payload["reason"], f"(via {event.payload['tool']})")
print("\nwhat the model was told:")
for message in result.messages:
    if message.role == "tool" and "Lockdown" in message.content:
        print("  " + message.content[:260])

from shipit_agent import SubAgentTool

CALLS.clear()
a = Agent(
    llm=llm(),
    tools=[
        SubAgentTool(llm=llm(), max_iterations=3),
        tool("read_file", "def login(): pass  # TODO: MFA",
             description="Read a file from disk"),
    ],
    auto_use_skills=False, max_iterations=3,
)

result = a.run(
    "Call sub_agent with task='Read auth.py using read_file and summarise it'."
)

print("parent called  :", [e.payload.get("tool") for e in result.events
                           if e.type == "tool_called"])
print("child ran      :", [n for n, _ in CALLS])
print("child events   :", sum(1 for e in result.events if e.type == "sub_agent_event"))

tool_messages = [m for m in result.messages if m.role == "tool"]
if tool_messages:
    print("\nreport back    :", tool_messages[0].content[:200])
    print("metadata       :", {k: tool_messages[0].metadata.get(k)
                               for k in ("ok", "iterations", "tool_call_count")})

# The child's work, rendered where it belongs — under the delegation.
from shipit_agent.narrate import render_transcript, render_tree

print(render_transcript(result.events, model="gemma-4-26b"))
print()
print(render_tree(result.events, model="gemma-4-26b"))

from shipit_agent.integrations import CredentialRecord, InMemoryCredentialStore
from shipit_agent.tools.connections import ConnectionsTool

store = InMemoryCredentialStore()
store.set(CredentialRecord(key="gmail", provider="google",
                           secrets={"access_token": "ya29.demo"},
                           metadata={"email": "[email]"}))

a = Agent(
    llm=llm(), credential_store=store,
    tools=[ConnectionsTool(),
           tool("gmail_search", "14 unread", credential_key="gmail",
                properties={"query": {"type": "string"}}, required=["query"]),
           tool("slack", "posted", credential_key="slack",
                properties={"text": {"type": "string"}}, required=["text"])],
    auto_use_skills=False, max_iterations=4,
)

result = a.run("Use the connections tool with action='list' to show what is connected.")
print([m for m in result.messages if m.role == "tool"][0].content)

# The registry directly — no model in the loop, no call spent.
from shipit_agent import ConnectionRegistry

registry = ConnectionRegistry(
    credential_store=store,
    tools=[t for t in a.tools if hasattr(t, "credential_key")],
)
for connection in registry.all():
    print(f"  {connection.state.value:14} {connection.title}")
    if connection.needs_action:
        print(f"                 -> {connection.next_step()}")
print("\nsummary:", registry.summary())

from shipit_agent.tools.give_up import GiveUpTool

a = Agent(llm=llm(), tools=[GiveUpTool()], auto_use_skills=False, max_iterations=4)
result = a.run(
    "Deploy the app to production. You have no credentials and no deploy tool. "
    "Call give_up with a specific reason."
)

print("gave_up  :", result.metadata.get("gave_up"))
print("reason   :", result.metadata.get("give_up_reason"))
print("needs    :", result.metadata.get("give_up_needs"))

import json

from shipit_agent.builtins import get_builtin_tools
from shipit_agent.llms.base import LLMResponse


class Probe:
    """Captures what actually goes over the wire, without spending a call."""
    model = "probe"

    def __init__(self):
        self.tools = self.system = None

    def complete(self, *, messages, tools=None, system_prompt=None,
                 metadata=None, text_delta_callback=None):
        if self.tools is None:
            self.tools, self.system = tools or [], system_prompt or ""
        return LLMResponse(content="ok")


builtins = get_builtin_tools(llm=None, project_root=".")
rows = []
for label, options in (("default", {}), ("code_mode=True", {"code_mode": True})):
    probe = Probe()
    Agent(llm=probe, tools=builtins, auto_use_skills=False, **options).run("hi")
    schemas, system = len(json.dumps(probe.tools)), len(probe.system)
    rows.append((label, len(probe.tools), schemas, system))

print(f"{'':>16}  {'tools':>5} {'schemas':>9} {'system':>9} {'TOTAL':>9} {'~tokens':>9}")
print("-" * 64)
for label, count, schemas, system in rows:
    total = schemas + system
    print(f"{label:>16}  {count:>5} {schemas:>9,} {system:>9,} {total:>9,} {total // 4:>9,}")
before, after = sum(rows[0][2:]), sum(rows[1][2:])
print("-" * 64)
print(f"saved per model call: {(before - after) // 4:,} tokens "
      f"({100 * (1 - after / before):.0f}% smaller)")

from shipit_agent.narrate import write_transcript

CALLS.clear()
a = Agent(llm=llm(),
          tools=[tool("read_file", "def login(): pass  # TODO: MFA")],
          auto_use_skills=False, max_iterations=4)
events = list(a.stream("Read auth.py with read_file and summarise it in one line."))

path = write_transcript("/tmp/shipit_run.html", events,
                        title="Live run — Gemma 4 on Bedrock", model=MODEL)
print("written:", path)

from IPython.display import IFrame
IFrame(src=f"file://{path}", width="100%", height=520)

agent = Agent(llm=llm(), tools=READERS, permissions=READ_ONLY,
              auto_use_skills=False, max_iterations=6)

agent.run_live(
    "Use glob_files with pattern 'shipit_agent/narrate/*.py', then read_file "
    "shipit_agent/narrate/timeline.py. Say in one sentence what it does.",
    style="tree",
)

from shipit_agent.narrate import render_tree

events = list(agent.stream(
    "Use grep_files with pattern 'class TreeRenderer' and path 'shipit_agent'. "
    "One sentence on what you found."
))
print(render_tree(events, model="gemma-4-26b", detail=True, output_lines=3))

import json

from shipit_agent.narrate import timeline

for step in timeline(events):
    print(json.dumps(step, indent=2, default=str))

from shipit_agent.narrate import watch

agent = Agent(llm=llm(), tools=READERS, permissions=READ_ONLY,
              auto_use_skills=False, max_iterations=6)

watch(agent, "How many modules are in shipit_agent/narrate/, and what is each for?",
      title="narrate/ tour", output_limit=None)

smart = Agent(llm=llm(), tools=READERS, auto_use_skills=False,
              max_iterations=8, delegation=True,
              permissions=PermissionEngine(
                  allow=["read_file", "grep_files", "glob_files", "sub_agent"],
                  default_decision=PermissionDecision.DENY))

result = smart.run(
    "Summarize shipit_agent/narrate/tree.py, shipit_agent/narrate/timeline.py "
    "and shipit_agent/narrate/live_ui.py in one line each."
)

parent = [e.payload["tool"] for e in result.events if e.type == "tool_called"]
children = [(e.payload.get("agent"), (e.payload.get("inner") or {}).get("tool"))
            for e in result.events
            if e.type == "sub_agent_event"
            and e.payload.get("inner_type") == "tool_called"]

print("parent calls:", parent)
print("delegations :", parent.count("sub_agent"))
print("child calls :", children)
print()
print(result.output)