Mini Vibe: a coding agent in your terminal and your browser

# The patterns read the call's arguments as JSON text, where a newline is the two characters \n.
# A command that could run a second command or write a file asks: a separator, a newline, a
# redirection, an expansion ($), a backtick, or git's --output option.
ASK = ("*;*", "*&*", "*|*", "*\\n*", "*>*", "*<(*", "*$*", "*`*", "*--output*")
# Git's settings can name a program that `git diff` or `git log` runs, so an edit to them asks: a
# path with .git in it (.git/, .gitattributes, .gitconfig), in any case, since macOS disks ignore
# case. `path` matches both write_file's `path` and edit's `file_path`.
GIT_SETTINGS = ('*path": "*[.][gG][iI][tT]*',)
READ_ONLY = (
    "ls",
    "ls *",
    "cat *",
    "head *",
    "tail *",
    "wc *",
    "grep *",
    "git status",
    "git status *",
    "git diff",
    "git diff *",  # with the space: `git difftool` runs a program of its choosing
    "git log",
    "git log *",
    "git show",
    "git show *",
)

coder = agents.Agent(
    model="mistral-large-4",
    name="mini-vibe",
    instructions=(
        "You are a coding agent working in the user's repository, your current directory. Read the "
        "code before you change it, make the smallest change that does the task, then run the "
        "project's tests. End with two or three lines: what you changed and what the tests said."
    ),
    environment=agents.DangerousLocalSandbox(workspace=os.getcwd()),  # your checkout, no container
    permission_policy=agents.PermissionPolicy(
        default="ask",
        rules=[
            agents.PermissionRule(tools=["file_system.read_file"], decision="allow"),
            agents.PermissionRule(
                tools=["file_system.write_file", "file_system.edit"],
                decision="allow",  # like Vibe: edits apply, `git diff` shows them
                sensitive=GIT_SETTINGS,
            ),
            agents.PermissionRule(
                tools=["file_system.bash"],
                decision="ask",
                sensitive=ASK,
                allowlist=[f'*"command": "{command}"*' for command in READ_ONLY],
            ),
        ],
    ),
)

async def show(self) -> None:
    async for event in self.session.iter_events():
        match event:
            case agents.TurnStarted():
                self.turn_id = event.turn_id
            case agents.TextDelta(entry_id=entry_id, text=text):
                self.streamed.add(entry_id)
                self.out(text, end="")
            case agents.EntryCompleted(entry=agents.MessageEntry(role="assistant") as message):
                self.out("" if message.id in self.streamed else message.text)
            case agents.EntryCompleted(entry=agents.ToolCallEntry(name=name, arguments=args)):
                self.out(f"{ORANGE}⏺{RESET} {BOLD}{name.split('.')[-1]}{RESET} {web.summary(args)}")
            case agents.EntryCompleted(entry=agents.ToolResultEntry() as result):
                lines = web.result_text(result).strip().splitlines() or [""]
                more = f" (+{len(lines) - 1} lines)" if len(lines) > 1 else ""
                self.out(f"  {DIM}⎿ {lines[0][:100]}{more}{RESET}")
            case agents.CallbackOpened(callback=agents.OpenApproval() as approval):
                await self.ask(approval)
            case agents.CallbackResolved(callback_id=callback_id):
                for approval in [a for a in self.waiting if a.id == callback_id]:
                    self.waiting.remove(approval)
                    self.out(f"  {DIM}answered in the web view: {web.summary(approval.request.input)}{RESET}")
            case agents.TurnEnded(state=state):
                self.turn_id = None
                if not isinstance(state, agents.Completed):
                    self.out(f"{DIM}[{type(state).__name__}]{RESET}")
                self.out(f"{BOLD}>{RESET} ", end="")
        if self.waiting and not self.prompting and not isinstance(event, agents.TextDelta):
            self.prompt()  # ask again, below the new output

async def events(request: Request) -> StreamingResponse:
    from_the_page(request)
    # A browser that reconnects sends the id of the last event it got: nothing shows twice.
    after = request.headers.get("last-event-id")

    async def stream():
        async for event in session.iter_events(after=after):
            if (data := as_json(event)) is not None:
                yield f"id: {event.id}\ndata: {json.dumps(data)}\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

async def answer(self, approval: agents.OpenApproval, decision: agents.AllowApproval | agents.DenyApproval) -> None:
    match await self.session.turn(approval.turn_id).answer(approval.id, decision):
        case agents.Answered(answer=applied) if applied != decision:
            self.out(f"  {DIM}the web view answered first{RESET}")
        case agents.AlreadySettled():
            self.out(f"  {DIM}the web view answered first{RESET}")

async def answer(request: Request) -> JSONResponse:
    from_the_page(request)
    body = await request.json()
    decision = agents.AllowApproval() if body["allow"] else agents.DenyApproval()
    match await session.turn(body["turn"]).answer(body["id"], decision):
        case agents.Answered(answer=applied) if applied != decision:
            return JSONResponse({"result": "the terminal answered first"})
        case agents.AlreadySettled():
            return JSONResponse({"result": "the terminal answered first"})
    return JSONResponse({"result": "done"})

def from_the_page(request: Request) -> None:
    """Refuse a request without this run's token, or sent by another website's page: it could read
    your code or start a turn that edits your files."""
    token = request.headers.get("x-token") or request.query_params.get("token", "")
    if not secrets.compare_digest(token, TOKEN) or request.headers.get("origin", URL) != URL:
        raise HTTPException(403, "only the web view's own page can send this")

elif line.strip():
    self.prompting = False
    if self.turn_id is not None:
        self.out(f"  {DIM}↳ steering the running turn{RESET}")
    await self.session.send_message(line)  # starts a turn, or steers the running one
    self.prompt()  # an open approval still waits for y/n/a

async def stop(self) -> None:
    if self.turn_id is not None:
        await self.session.turn(self.turn_id).interrupt()

async def main() -> None:
    session = await coder.start_session()
    terminal = Terminal(session)
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, lambda: asyncio.ensure_future(terminal.stop()))  # Ctrl-C stops the turn
    print(
        f"{BOLD}mini-vibe{RESET} {DIM}· {os.path.basename(os.getcwd())} · {coder.model} · web view: {web.LINK}{RESET}"
    )
    print(f"{BOLD}>{RESET} ", end="", flush=True)
    browser = web.server(session)
    tasks = [asyncio.create_task(terminal.show()), asyncio.create_task(browser.serve())]
    try:
        await read_lines(terminal)
    finally:
        browser.should_exit = browser.force_exit = True  # an open browser tab does not hold the exit
        tasks[0].cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
        await session.cancel()

uv sync

export MISTRAL_API_KEY=...
export MINI_VIBE=~/path/to/mini_vibe  # the example's folder
ALLOW_DANGEROUS_LOCAL_HOST=1 uv run --project "$MINI_VIBE" python "$MINI_VIBE/mini_vibe.py"