Super Agent - high-depth live validation

import sys
from pathlib import Path


def repo_root(start: Path) -> Path:
    for candidate in [start.resolve(), *start.resolve().parents]:
        if (candidate / "shipit_agent" / "__init__.py").exists():
            return candidate
    raise RuntimeError(f"Cannot find shipit_agent from {start}")


REPO = repo_root(Path.cwd())
if str(REPO) not in sys.path:
    sys.path.insert(0, str(REPO))
print("repo:", REPO)

import tempfile
from typing import Any

from shipit_agent import (
    Agent,
    FunctionTool,
    Guardrails,
    PermissionEngine,
    RouterPolicy,
    ToolOutputChunk,
)
from shipit_agent.llms import LLMResponse, SimpleEchoLLM
from shipit_agent.mcp import MCPServer, MCPTool
from shipit_agent.models import ToolCall
from shipit_agent.multimodal import MediaParser, build_multimodal_message
from shipit_agent.permissions import PermissionDecision


class ScenarioLLM:
    model = "test-model-128k"

    def __init__(self, calls):
        self.calls = list(calls)
        self.index = 0

    def complete(self, **kwargs: Any):
        if self.index < len(self.calls):
            call = self.calls[self.index]
            self.index += 1
            return LLMResponse(
                content="",
                tool_calls=[ToolCall(name=call[0], arguments=call[1])],
                usage={
                    "prompt_tokens": 40,
                    "completion_tokens": 8,
                    "cache_read_input_tokens": 30,
                    "cache_creation_input_tokens": 4,
                },
            )
        return LLMResponse(
            content="acceptance complete",
            usage={
                "prompt_tokens": 20,
                "completion_tokens": 5,
                "cache_read_input_tokens": 15,
            },
        )


def stream_probe():
    yield ToolOutputChunk("phase-1\n", {"phase": 1})
    yield ToolOutputChunk("phase-2\n", {"phase": 2})
    yield "phase-3\n"


workspace = Path(tempfile.mkdtemp(prefix="shipit-super-agent-"))
policy = PermissionEngine(
    allow=["stream_probe", "write_file", "read_file"],
    default_decision=PermissionDecision.DENY,
)
model = ScenarioLLM(
    [
        ("stream_probe", {}),
        ("write_file", {"path": "report.txt", "content": "verified\n"}),
        ("read_file", {"path": "report.txt"}),
    ]
)
agent = Agent.for_project(
    llm=model,
    project_root=workspace,
    optimized=True,
    tools=[FunctionTool.from_callable(stream_probe)],
    permissions=policy,
    guardrails=Guardrails(),
    router_policy=RouterPolicy(auto_plan=False),
    auto_use_skills=False,
)
events = list(agent.stream("Run the full deterministic acceptance flow."))
deltas = [e for e in events if e.type == "tool_output_delta"]
assert (workspace / "report.txt").read_text() == "verified\n"
assert "".join(e.payload["chunk"] for e in deltas[:1]) == "phase-1\nphase-2\nphase-3\n"
assert all(e.payload["buffered"] for e in events if e.type == "tool_output_started")
assert agent.code_mode and agent.max_iterations == 8
assert agent.max_tool_output_chars == 16_000
completed = next(e for e in reversed(events) if e.type == "run_completed")
assert completed.payload["usage"]["cache_read_input_tokens"] == 105

# Permission denial is checked without relying on model behavior.
assert policy.check("bash", {"command": "rm -rf /"}).denied

# Local MCP tools retain server provenance and use the same stream.
mcp = MCPServer(name="knowledge").register(
    MCPTool(
        name="lookup",
        description="Lookup facts",
        handler=lambda **kwargs: "mcp-fact",
    )
)
mcp_agent = Agent(
    llm=ScenarioLLM([("lookup", {})]),
    mcps=[mcp],
    auto_use_skills=False,
    router_policy=RouterPolicy(auto_plan=False),
)
mcp_events = list(mcp_agent.stream("lookup"))
mcp_delta = next(e for e in mcp_events if e.type == "tool_output_delta")
assert mcp_delta.payload["chunk"] == "mcp-fact"
assert mcp_delta.payload["chunk_metadata"]["server"] == "knowledge"

# Durable sessions survive agent reconstruction.
first = Agent.for_project(llm=SimpleEchoLLM(), project_root=workspace, optimized=True)
first.chat_session(session_id="main").send("remember this checkpoint")
second = Agent.for_project(llm=SimpleEchoLLM(), project_root=workspace, optimized=True)
assert second.chat_session(session_id="main").history()
assert (workspace / ".shipit" / "sessions" / "main.json").exists()

# Screenshot/media references preserve text-image-text ordering.
parsed = MediaParser(allowlist_domains=["cdn.example.com"]).parse(
    "Inspect ![error screenshot](https://cdn.example.com/error.png) carefully."
)
media_message = build_multimodal_message(parsed)
assert [block["type"] for block in media_message["content"]] == [
    "text",
    "image",
    "text",
]

doctor = agent.doctor()
assert not any(check.status == "fail" for check in doctor.checks)
print("PASS: optimized project agent")
print("PASS: live tool chunks and canonical results")
print("PASS: permissions, MCP, sessions, cache, media, doctor")
print("workspace:", workspace)

for event in events:
    if event.type == "tool_output_delta":
        p = event.payload
        print(f"[{p['call_id']} #{p['sequence']}] {p['chunk']}", end="")
    elif event.type in {"tool_called", "tool_completed", "tool_denied"}:
        print(f"\n{event.type}: {event.payload.get('tool')}")

report = agent.doctor()
for check in report.checks:
    print(f"{check.status.upper():4} {check.name:14} {check.message}")
    if check.name in {"tools", "skills", "mcps", "connections", "efficiency"}:
        print("    ", check.details)

import os

RUN_LIVE = os.getenv("SHIPIT_RUN_LIVE", "").lower() in {"1", "true", "yes"}
PROVIDER = Path(
    "~/Documents/MYWORK/AFTDRK/CACHE/DRK_CACHE_BACK"
    "/drk_cache/llm/bedrock_mantle_provider.py"
)
MODEL = "bedrock-mantle/google.gemma-4-26b-a4b"


def configured_llm():
    import importlib.util

    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

    return LiteLLMChatLLM(model=MODEL)


print("live enabled:", RUN_LIVE)
print("provider exists:", PROVIDER.exists())

if RUN_LIVE:
    from shipit_agent.builtins import get_builtin_tool_map

    audit_map = get_builtin_tool_map(llm=None, project_root=str(REPO))
    audit_tools = [audit_map[name] for name in ("read_file", "grep_files")]
    live_policy = PermissionEngine(
        allow=["read_file", "grep_files"],
        default_decision=PermissionDecision.DENY,
    )
    live_agent = Agent(
        llm=configured_llm(),
        project_root=REPO,
        tools=audit_tools,
        permissions=live_policy,
        max_iterations=16,
        context_window_tokens=128_000,
        max_tool_output_chars=16_000,
        progress_summaries=True,
        auto_use_skills=False,
    )
    deep_prompt = """
Perform a rigorous architecture audit of this coding agent. Use tools; do not guess.
Do not inspect build/, dist/, site/, or .venv/. First call read_file on these exact
source paths: shipit_agent/agent.py, shipit_agent/runtime_core.py,
shipit_agent/runtime.py, shipit_agent/async_runtime.py, shipit_agent/mcp.py,
shipit_agent/permissions.py, shipit_agent/tool_runner.py,
shipit_agent/tools/helpers.py, and shipit_agent/streaming.py. Use grep_files
inside shipit_agent only for symbols you must verify. Cite exact source paths.
Answer with: (1) evidence table with file paths, (2) exact tool/skill/MCP
lifecycle, (3) token and prompt-cache controls, (4) long-session behavior,
(5) permission and secret boundaries, (6) sync/async parity risks, and
(7) five highest-value improvements. Separate verified facts from inference.
"""
    live_events = []
    for event in live_agent.stream(deep_prompt):
        live_events.append(event)
        if event.type in {"text_delta", "tool_output_delta"}:
            print(event.payload.get("chunk", ""), end="", flush=True)
        elif event.type == "tool_called":
            print(f"\n\nCALL {event.payload['call_id']}: {event.payload['tool']}")
    read_calls = [
        e
        for e in live_events
        if e.type == "tool_called" and e.payload.get("tool") == "read_file"
    ]
    assert len(read_calls) >= 5, f"not enough source reads: {len(read_calls)}"
    assert any(e.type == "tool_output_delta" for e in live_events)
    assert not any(
        e.type == "tool_denied"
        and e.payload.get("tool") in {"read_file", "grep_files", "glob_files"}
        for e in live_events
    )
else:
    print("Skipped paid live audit. Set RUN_LIVE=True to execute it.")

if RUN_LIVE:
    import tempfile

    challenge = Path(tempfile.mkdtemp(prefix="shipit-live-project-"))
    (challenge / "SPEC.md").write_text(
        "# Task\nCreate stats.py with summarize(values). Return count, min, max, mean. "
        "Reject an empty list with ValueError. Add test_stats.py.\n"
    )
    edit_policy = PermissionEngine(
        allow=[
            "read_file",
            "write_file",
            "edit_file",
            "grep_files",
            "glob_files",
            "bash",
            "tool_search",
            "describe_binding",
            "execute_code",
        ],
        default_decision=PermissionDecision.DENY,
    )
    coder = Agent.for_project(
        llm=configured_llm(),
        project_root=challenge,
        optimized=True,
        permissions=edit_policy,
        max_iterations=12,
    )
    edit_events = list(
        coder.stream(
            "Read SPEC.md, implement it, add strong tests, run pytest, and fix failures."
        )
    )
    assert (challenge / "stats.py").exists()
    assert (challenge / "test_stats.py").exists()
    edit_output = "\n".join(
        str(e.payload.get("output", ""))
        for e in edit_events
        if e.type == "tool_completed"
    )
    assert "passed" in edit_output.lower(), edit_output[-2000:]
    assert not any(e.type == "tool_failed" for e in edit_events)
    denied = [e for e in edit_events if e.type == "tool_denied"]
    assert all(e.payload.get("decision") == "deny" for e in denied)
    assert any(e.type == "tool_output_delta" for e in edit_events)
    print("challenge:", challenge)
    print("tool calls:", sum(e.type == "tool_called" for e in edit_events))
    print("safely denied:", [e.payload.get("tool") for e in denied])
else:
    print("Skipped paid edit challenge. Set RUN_LIVE=True to execute it.")