diff --git a/claude_agent_pipe_sandboxed.py b/claude_agent_pipe_sandboxed.py
new file mode 100644
index 0000000..669d588
--- /dev/null
+++ b/claude_agent_pipe_sandboxed.py
@@ -0,0 +1,868 @@
+"""
+title: Claude Code (Sandboxed)
+description: Run Claude Code inside an open-webui/open-terminal sandbox, one
+  Linux account per OpenWebUI user. Isolates the agent's file/process reach
+  from the Open WebUI backend host.
+author: Thomas Friedel
+version: 0.1
+license: MIT
+requirements: httpx>=0.27
+"""
+
+# Why this pipe exists
+# --------------------
+# The sibling `claude_agent_pipe.py` runs Claude Code's agent loop *in-process*
+# via the Claude Agent SDK. That means Bash/Read/Write/Edit calls hit the
+# Open WebUI backend's filesystem with whatever UID that process has — fine
+# for a solo dev setup, dangerous in any shared deployment.
+#
+# This variant shells the `claude` CLI inside a per-user Linux account on an
+# open-webui/open-terminal container. Isolation is by standard Unix
+# permissions (shared kernel). Ship this when you have multiple users who
+# you'd rather keep from stomping on each other's workspaces.
+#
+# Not yet ported from the SDK pipe:
+#   - Knowledge-base MCP tool (Workspace Model KB attachments). The SDK pipe
+#     registers an in-process MCP tool that calls OpenWebUI's retrieval layer
+#     directly. In the sandbox the `claude` CLI can't reach Python functions;
+#     we'd need to expose the KB as an HTTP MCP server from the OWUI backend
+#     and `claude mcp add` it inside the container. Left as a follow-up.
+#   - Partial-message token streaming for thinking blocks. stream-json emits
+#     full `assistant` events, which is fine for text but slightly chunkier
+#     UX than the SDK's `include_partial_messages`. Can be upgraded later
+#     via `claude --verbose` stream_event passthrough.
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import mimetypes
+import re
+import shlex
+import time
+import uuid
+from dataclasses import dataclass
+from typing import Any, AsyncGenerator, AsyncIterator, Callable, Dict, List, Optional
+
+import httpx
+from pydantic import BaseModel, Field
+
+log = logging.getLogger(__name__)
+
+_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"}
+_DOWNLOAD_EXTENSIONS = {
+    ".pdf", ".csv", ".tsv", ".txt", ".md", ".json", ".yaml", ".yml",
+    ".html", ".xml", ".xlsx", ".docx", ".pptx", ".zip",
+}
+_ARTIFACT_EXTENSIONS = _IMAGE_EXTENSIONS | _DOWNLOAD_EXTENSIONS
+_MAX_ARTIFACT_BYTES = 50 * 1024 * 1024
+
+# OpenWebUI calls pipe() fresh for each chat turn. Same pattern as the SDK
+# pipe — in-process map of chat_id → claude session_id so `claude --resume`
+# picks up prior turns' state.
+_chat_sessions: Dict[str, str] = {}
+
+
+# ============================================================================
+# open-terminal client
+# ============================================================================
+
+_USER_ID_SAFE = re.compile(r"[^a-zA-Z0-9_-]")
+_CHAT_ID_SAFE = re.compile(r"[^a-zA-Z0-9_-]")
+
+
+def _sanitize_user_id(raw: str) -> str:
+    """Match open-terminal's server-side sanitation so we can predict the
+    Linux username it provisions (useful for logging)."""
+    return _USER_ID_SAFE.sub("", raw)[:32] or "anon"
+
+
+def _safe_chat_id(raw: str) -> str:
+    """Strip everything except alphanumerics, dash, and underscore. OWUI's
+    chat_ids are UUIDs so this is a no-op in practice, but it lets us embed
+    the id directly in shell commands without shlex.quote — which is
+    necessary because shlex.quote wraps in single quotes and single quotes
+    suppress the `~` and `$HOME` expansion our workdir templates rely on."""
+    return _CHAT_ID_SAFE.sub("", raw) or "default"
+
+
+@dataclass
+class _ProcessHandle:
+    process_id: str
+    user_id: str
+
+
+class OpenTerminalError(RuntimeError):
+    pass
+
+
+class OpenTerminalClient:
+    def __init__(self, base_url: str, api_key: str, *, timeout: float = 30.0):
+        self._base = base_url.rstrip("/")
+        self._headers = {"Authorization": f"Bearer {api_key}"}
+        self._timeout = timeout
+        self._client: Optional[httpx.AsyncClient] = None
+
+    async def __aenter__(self):
+        self._client = httpx.AsyncClient(
+            base_url=self._base, headers=self._headers, timeout=self._timeout
+        )
+        return self
+
+    async def __aexit__(self, *_exc):
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
+
+    def _uh(self, user_id: str, session_id: Optional[str] = None) -> dict:
+        h = {"X-User-Id": _sanitize_user_id(user_id)}
+        if session_id:
+            h["X-Session-Id"] = session_id
+        return h
+
+    async def start(
+        self,
+        user_id: str,
+        command: str,
+        *,
+        cwd: Optional[str] = None,
+        session_id: Optional[str] = None,
+    ) -> _ProcessHandle:
+        assert self._client is not None
+        payload: dict = {"command": command}
+        if cwd:
+            payload["cwd"] = cwd
+        r = await self._client.post(
+            "/execute", json=payload, headers=self._uh(user_id, session_id)
+        )
+        if r.status_code >= 400:
+            raise OpenTerminalError(f"POST /execute {r.status_code}: {r.text}")
+        # open-terminal returns {id, command, status, output, next_offset, ...}.
+        # `id` here is the process identifier (not related to X-User-Id).
+        return _ProcessHandle(process_id=r.json()["id"], user_id=user_id)
+
+    async def stream_output(
+        self,
+        handle: _ProcessHandle,
+        *,
+        session_id: Optional[str] = None,
+        wait: float = 10.0,
+    ) -> AsyncIterator[tuple[str, str]]:
+        """Yield (stream, chunk) tuples until the process exits.
+
+        Uses the long-poll pattern: `GET /execute/{id}/status?wait=N&offset=K`
+        returns as soon as new output appears or the process exits, up to N
+        seconds. `offset` is advanced by the cumulative number of entries
+        we've already consumed (from the server's `next_offset`).
+
+        Stream values: 'output' for merged stdout/stderr (open-terminal runs
+        commands through a PTY so both streams are combined), and 'exit' as
+        a synthetic terminator carrying the exit code.
+        """
+        assert self._client is not None
+        headers = self._uh(handle.user_id, session_id)
+        offset = 0
+        # Per-request timeout must exceed `wait`, or httpx aborts while the
+        # server is still happily long-polling.
+        read_timeout = wait + 10.0
+        while True:
+            r = await self._client.get(
+                f"/execute/{handle.process_id}/status",
+                params={"wait": wait, "offset": offset},
+                headers=headers,
+                timeout=read_timeout,
+            )
+            if r.status_code >= 400:
+                raise OpenTerminalError(f"status: {r.status_code} {r.text}")
+            data = r.json()
+            for entry in data.get("output") or []:
+                # All entries currently arrive with type='output'. Strip the
+                # PTY's CR so downstream splitting on '\n' works cleanly.
+                text = (entry.get("data") or "").replace("\r\n", "\n").replace("\r", "\n")
+                if text:
+                    yield "output", text
+            offset = data.get("next_offset", offset)
+            if data.get("status") != "running":
+                yield "exit", str(data.get("exit_code") or 0)
+                return
+
+    async def kill(self, handle: _ProcessHandle, *, force: bool = False) -> None:
+        assert self._client is not None
+        try:
+            await self._client.delete(
+                f"/execute/{handle.process_id}",
+                params={"force": str(force).lower()},
+                headers=self._uh(handle.user_id),
+            )
+        except Exception:
+            pass  # best-effort cleanup
+
+    async def list_files(self, user_id: str, path: str) -> List[dict]:
+        assert self._client is not None
+        r = await self._client.get(
+            "/files/list", params={"path": path}, headers=self._uh(user_id)
+        )
+        if r.status_code >= 400:
+            return []
+        return r.json().get("entries", [])
+
+    async def read_file(self, user_id: str, path: str) -> bytes:
+        """Download a file's raw bytes. Uses `/files/view` (not `/files/read`)
+        — /read returns a JSON envelope `{path, content}` for text and 415
+        for binary, so it's unusable for PNG/PDF/ZIP artifacts."""
+        assert self._client is not None
+        r = await self._client.get(
+            "/files/view", params={"path": path}, headers=self._uh(user_id)
+        )
+        if r.status_code >= 400:
+            raise OpenTerminalError(f"view {path}: {r.status_code}")
+        return r.content
+
+    async def ensure_dir(self, user_id: str, path: str) -> None:
+        """Create a directory in the user's home if missing. `path` must be
+        pre-sanitized (alphanumeric + `/._-` + `~` / `$HOME`): it's embedded
+        unquoted so bash can expand `~` and variables. We can't shlex.quote
+        here — single quotes suppress the very expansion we need."""
+        handle = await self.start(user_id, f"mkdir -p {path}")
+        async for _ in self.stream_output(handle):
+            pass
+
+
+# ============================================================================
+# claude CLI invocation + stream-json parsing
+# ============================================================================
+
+
+@dataclass
+class _ClaudeRunConfig:
+    model: str
+    permission_mode: str
+    allowed_tools: List[str]
+    max_turns: int
+    resume_session_id: Optional[str]
+    oauth_token: Optional[str]
+    api_key: Optional[str]
+    workdir: str
+    system_prompt_append: Optional[str]
+
+
+# Baked into the system prompt on every turn so Claude knows the sandbox's
+# quirks without the user having to phrase requests carefully. Kept short —
+# it's prepended to whatever the Workspace Model's system prompt says.
+_SANDBOX_ENV_NOTE = (
+    "You are running in a headless Linux sandbox (no display, no audio). "
+    "For any plot or chart, ALWAYS save it to a file in the current working "
+    "directory with an explicit extension (e.g. matplotlib → "
+    "plt.savefig('name.png'); pandas/plotly → .to_html or kaleido png). "
+    "plt.show() is a no-op here. The host automatically inlines any new "
+    "PNG/PDF/CSV/HTML/… files from the working directory into the chat "
+    "after each turn, so the user will see saved files without you needing "
+    "to read them back."
+)
+
+
+def _claude_command(prompt: str, cfg: _ClaudeRunConfig) -> str:
+    env_parts: List[str] = []
+    if cfg.oauth_token:
+        env_parts.append(f"CLAUDE_CODE_OAUTH_TOKEN={shlex.quote(cfg.oauth_token)}")
+    elif cfg.api_key:
+        env_parts.append(f"ANTHROPIC_API_KEY={shlex.quote(cfg.api_key)}")
+    # `claude --dangerously-skip-permissions` still refuses root unless told
+    # it's sandboxed. open-terminal runs processes as the per-user UID so this
+    # rarely matters, but setting it is harmless.
+    env_parts.append("IS_SANDBOX=1")
+    # Force matplotlib's non-interactive backend. Without this, any stray
+    # `plt.show()` can trigger a DISPLAY-not-found exception that aborts the
+    # script before savefig runs.
+    env_parts.append("MPLBACKEND=Agg")
+
+    cli_parts = [
+        "claude", "--print",
+        "--output-format", "stream-json",
+        "--verbose",  # required for stream-json
+        "--model", shlex.quote(cfg.model),
+        "--permission-mode", shlex.quote(cfg.permission_mode),
+        "--allowed-tools", shlex.quote(",".join(cfg.allowed_tools)),
+    ]
+    if cfg.max_turns and cfg.max_turns > 0:
+        cli_parts += ["--max-turns", str(cfg.max_turns)]
+    if cfg.resume_session_id:
+        cli_parts += ["--resume", shlex.quote(cfg.resume_session_id)]
+
+    # Always prepend the sandbox environment note; merge in whatever the
+    # Workspace Model configured. One --append-system-prompt flag carries both.
+    system_append = _SANDBOX_ENV_NOTE
+    if cfg.system_prompt_append:
+        system_append = f"{_SANDBOX_ENV_NOTE}\n\n{cfg.system_prompt_append}"
+    cli_parts += ["--append-system-prompt", shlex.quote(system_append)]
+
+    cli_parts.append(shlex.quote(prompt))
+
+    return " ".join(env_parts + cli_parts)
+
+
+async def _stream_claude_events(
+    client: OpenTerminalClient,
+    user_id: str,
+    prompt: str,
+    cfg: _ClaudeRunConfig,
+    *,
+    session_id: Optional[str],
+) -> AsyncIterator[dict]:
+    """Start claude in the sandbox and yield parsed stream-json events.
+    Synthetic event types (underscore-prefixed) surface transport-level
+    signals the caller may want (stderr diagnostics, exit code)."""
+    await client.ensure_dir(user_id, cfg.workdir)
+    cmd = _claude_command(prompt, cfg)
+    # workdir is unquoted on purpose — see ensure_dir docstring. cfg.workdir
+    # has already been rendered through _safe_chat_id so it's shell-safe.
+    full = f"cd {cfg.workdir} && {cmd}"
+    handle = await client.start(user_id, full, session_id=session_id)
+
+    # open-terminal merges stdout+stderr through a PTY, so anything the CLI
+    # prints on stderr (auth warnings, config notices) will interleave with
+    # the stream-json lines on stdout. We parse every line as JSON; those
+    # that fail become synthetic _raw events so the caller can log them
+    # without disrupting the agent loop.
+    buf = ""
+    try:
+        async for stream, chunk in client.stream_output(handle, session_id=session_id):
+            if stream == "exit":
+                if buf.strip():
+                    # Flush any trailing unterminated line.
+                    try:
+                        yield json.loads(buf)
+                    except json.JSONDecodeError:
+                        yield {"type": "_raw", "text": buf.strip()}
+                    buf = ""
+                yield {"type": "_exit", "code": int(chunk)}
+                return
+            buf += chunk
+            while "\n" in buf:
+                line, buf = buf.split("\n", 1)
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    yield json.loads(line)
+                except json.JSONDecodeError:
+                    yield {"type": "_raw", "text": line}
+    except asyncio.CancelledError:
+        await client.kill(handle, force=True)
+        raise
+
+
+# ============================================================================
+# Rendering helpers (kept in sync with claude_agent_pipe.py)
+# ============================================================================
+
+_TOOL_PREVIEW_FIELDS = {
+    "Bash": "command", "Read": "file_path", "Write": "file_path",
+    "Edit": "file_path", "Glob": "pattern", "Grep": "pattern",
+    "WebSearch": "query", "WebFetch": "url", "Task": "description",
+}
+_FENCE_LANG_PER_TOOL = {
+    "Bash": "bash", "Glob": "text", "Grep": "text",
+    "WebSearch": "text", "WebFetch": "text",
+}
+
+
+def _tool_preview(name: str, tool_input: Dict[str, Any]) -> str:
+    key = _TOOL_PREVIEW_FIELDS.get(name)
+    if key and key in tool_input:
+        raw = str(tool_input[key])
+    elif tool_input:
+        raw = ", ".join(f"{k}={str(v)[:40]}" for k, v in list(tool_input.items())[:2])
+    else:
+        return ""
+    first = raw.split("\n", 1)[0]
+    truncated = first if len(first) <= 120 else first[:117] + "…"
+    return truncated + (" …" if "\n" in raw and not truncated.endswith("…") else "")
+
+
+def _tool_input_block(name: str, tool_input: Dict[str, Any]) -> str:
+    if not tool_input:
+        return "```\n(no input)\n```"
+    primary = _TOOL_PREVIEW_FIELDS.get(name)
+    if primary and primary in tool_input:
+        lang = _FENCE_LANG_PER_TOOL.get(name, "text")
+        parts = [f"```{lang}\n{tool_input[primary]}\n```"]
+        others = {k: v for k, v in tool_input.items() if k != primary}
+        if others:
+            parts.append(
+                f"```json\n{json.dumps(others, indent=2, ensure_ascii=False)}\n```"
+            )
+        return "\n\n".join(parts)
+    return f"```json\n{json.dumps(tool_input, indent=2, ensure_ascii=False)}\n```"
+
+
+def _format_tool_result(content: Any) -> str:
+    if content is None:
+        return ""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts: List[str] = []
+        for item in content:
+            if isinstance(item, dict):
+                text = item.get("text")
+                parts.append(text if isinstance(text, str) else repr(item))
+            else:
+                parts.append(str(item))
+        return "\n".join(parts)
+    return str(content)
+
+
+# ============================================================================
+# body / metadata extraction (ported verbatim from claude_agent_pipe.py)
+# ============================================================================
+
+
+def _extract_latest_user_prompt(body: Dict[str, Any]) -> str:
+    messages = body.get("messages") or []
+    for message in reversed(messages):
+        if message.get("role") != "user":
+            continue
+        content = message.get("content")
+        if isinstance(content, str):
+            return content
+        if isinstance(content, list):
+            texts = [
+                part.get("text", "")
+                for part in content
+                if isinstance(part, dict) and part.get("type") == "text"
+            ]
+            if texts:
+                return "\n".join(texts)
+    return ""
+
+
+def _extract_system_prompt(body: Dict[str, Any]) -> Optional[str]:
+    parts: List[str] = []
+    for msg in body.get("messages") or []:
+        if msg.get("role") != "system":
+            continue
+        content = msg.get("content")
+        if isinstance(content, str):
+            parts.append(content)
+        elif isinstance(content, list):
+            for piece in content:
+                if isinstance(piece, dict) and piece.get("type") == "text":
+                    parts.append(piece.get("text", ""))
+    merged = "\n\n".join(p for p in parts if p and p.strip())
+    return merged or None
+
+
+# ============================================================================
+# Artifact upload (walks the sandbox workspace, uploads new files to OWUI)
+# ============================================================================
+
+
+async def _snapshot_workspace(
+    client: OpenTerminalClient, user_id: str, workdir: str
+) -> Dict[str, int]:
+    """Take a name→mtime_ns snapshot of artifact-like files in the workspace.
+    We compare before/after to detect what the agent produced this turn."""
+    # Use find with -printf to get path + mtime in one call, avoiding N+1
+    # HTTP hits against /files/list for every subdirectory.
+    # workdir unquoted so `~` / `$HOME` expand; see OpenTerminalClient.ensure_dir.
+    cmd = (
+        f"find {workdir} -type f "
+        rf"\( -iname '*.png' -o -iname '*.jpg' -o -iname '*.jpeg' "
+        rf"-o -iname '*.gif' -o -iname '*.svg' -o -iname '*.webp' "
+        rf"-o -iname '*.pdf' -o -iname '*.csv' -o -iname '*.tsv' "
+        rf"-o -iname '*.txt' -o -iname '*.md' -o -iname '*.json' "
+        rf"-o -iname '*.yaml' -o -iname '*.yml' -o -iname '*.html' "
+        rf"-o -iname '*.xml' -o -iname '*.xlsx' -o -iname '*.docx' "
+        rf"-o -iname '*.pptx' -o -iname '*.zip' \) "
+        rf"-printf '%T@\t%p\n' 2>/dev/null"
+    )
+    handle = await client.start(user_id, cmd)
+    out = ""
+    async for stream, chunk in client.stream_output(handle):
+        if stream == "output":
+            out += chunk
+    snapshot: Dict[str, int] = {}
+    for line in out.splitlines():
+        if "\t" not in line:
+            continue
+        mtime_str, path = line.split("\t", 1)
+        try:
+            snapshot[path] = int(float(mtime_str) * 1e9)
+        except ValueError:
+            continue
+    return snapshot
+
+
+async def _inline_new_artifacts(
+    client: OpenTerminalClient,
+    user_id: str,
+    workdir: str,
+    before: Dict[str, int],
+    owui_user_id: Optional[str],
+) -> List[str]:
+    """Detect files created/modified since `before`, download them from the
+    sandbox, and register them as OpenWebUI artifacts. See the SDK pipe for
+    why we serve via /api/v1/files/{id}/content instead of data URIs."""
+    if not owui_user_id:
+        return ["\n\n_(Can't save artifacts: no user context.)_\n"]
+
+    try:
+        from open_webui.models.files import FileForm, Files
+        from open_webui.storage.provider import Storage
+    except Exception as exc:
+        log.warning("artifact inline: OWUI file store unavailable: %s", exc)
+        return [f"\n\n_(File store unavailable: {exc})_\n"]
+
+    after = await _snapshot_workspace(client, user_id, workdir)
+    new_paths = [p for p, m in after.items() if before.get(p) != m]
+    new_paths.sort()
+    log.info(
+        "artifact inline: workdir=%s before=%d after=%d new=%d paths=%s",
+        workdir, len(before), len(after), len(new_paths), new_paths[:10],
+    )
+
+    chunks: List[str] = []
+    for remote_path in new_paths:
+        name = remote_path.rsplit("/", 1)[-1]
+        ext = ("." + name.rsplit(".", 1)[-1]).lower() if "." in name else ""
+        if ext not in _ARTIFACT_EXTENSIONS:
+            continue
+
+        try:
+            content = await client.read_file(user_id, remote_path)
+        except Exception as exc:
+            chunks.append(f"\n\n_(Couldn't fetch {name}: {exc})_\n")
+            continue
+
+        size = len(content)
+        if size > _MAX_ARTIFACT_BYTES:
+            chunks.append(
+                f"\n\n_(Skipped {name}: {size // 1024 // 1024} MiB exceeds "
+                f"{_MAX_ARTIFACT_BYTES // 1024 // 1024} MiB limit.)_\n"
+            )
+            continue
+
+        is_image = ext in _IMAGE_EXTENSIONS
+        mime = mimetypes.guess_type(name)[0] or (
+            "image/png" if is_image else "application/octet-stream"
+        )
+        file_id = str(uuid.uuid4())
+        storage_filename = f"{file_id}_{name}"
+
+        # Write through a temp buffer OWUI's Storage layer can consume.
+        # It expects a file-like with .read(); BytesIO fits.
+        import io
+        try:
+            stored, storage_path = Storage.upload_file(
+                io.BytesIO(content),
+                storage_filename,
+                {"OpenWebUI-User-Id": owui_user_id, "OpenWebUI-File-Id": file_id},
+            )
+        except Exception as exc:
+            log.exception("Artifact upload failed: %s", remote_path)
+            chunks.append(f"\n\n_(Failed to save {name}: {exc})_\n")
+            continue
+
+        try:
+            Files.insert_new_file(
+                owui_user_id,
+                FileForm(
+                    id=file_id,
+                    filename=name,
+                    path=storage_path,
+                    data={},
+                    meta={"name": name, "content_type": mime, "size": len(stored)},
+                ),
+            )
+        except Exception as exc:
+            log.exception("Artifact DB row failed: %s", remote_path)
+            chunks.append(f"\n\n_(Saved but not linkable: {name}: {exc})_\n")
+            continue
+
+        if is_image:
+            chunks.append(f"\n\n![{name}](/api/v1/files/{file_id}/content)\n")
+        else:
+            kib = size // 1024
+            chunks.append(
+                f"\n\n📎 [{name}](/api/v1/files/{file_id}/content) · {kib} KiB\n"
+            )
+    return chunks
+
+
+# ============================================================================
+# Pipe
+# ============================================================================
+
+
+class Pipe:
+    class Valves(BaseModel):
+        OPEN_TERMINAL_URL: str = Field(
+            default="http://open-terminal:8000",
+            description=(
+                "Base URL of the open-terminal sandbox. Use the docker-compose "
+                "service name when both run in the same network."
+            ),
+        )
+        OPEN_TERMINAL_API_KEY: str = Field(
+            default="",
+            description=(
+                "Shared secret matching OPEN_TERMINAL_API_KEY on the sandbox "
+                "container. Authenticates this pipe as a privileged caller; "
+                "the end user is then conveyed via the X-User-Id header."
+            ),
+        )
+        ANTHROPIC_API_KEY: str = Field(
+            default="",
+            description="Anthropic API key (pay-per-token). Forwarded into the sandbox as an env var.",
+        )
+        CLAUDE_CODE_OAUTH_TOKEN: str = Field(
+            default="",
+            description=(
+                "Claude subscription OAuth token. Takes priority over "
+                "ANTHROPIC_API_KEY — don't share with untrusted users (per "
+                "Anthropic's subscription terms)."
+            ),
+        )
+        MODEL: str = Field(default="claude-haiku-4-5")
+        PERMISSION_MODE: str = Field(default="bypassPermissions")
+        ALLOWED_TOOLS: str = Field(
+            default="Read,Write,Edit,Bash,Glob,Grep,WebSearch,WebFetch"
+        )
+        MAX_TURNS: int = Field(default=30)
+        WORKDIR_TEMPLATE: str = Field(
+            default="~/chat-{chat_id}",
+            description=(
+                "Per-chat workspace path inside the user's sandbox home. "
+                "{chat_id} is substituted. Artifacts persist across turns."
+            ),
+        )
+
+    def __init__(self) -> None:
+        self.valves = self.Valves()
+
+    def pipes(self) -> List[Dict[str, str]]:
+        return [{"id": "claude-code-sandboxed", "name": "Claude Code (Sandboxed)"}]
+
+    async def pipe(
+        self,
+        body: Dict[str, Any],
+        __chat_id__: Optional[str] = None,
+        __event_emitter__: Optional[Callable] = None,
+        __user__: Optional[Dict[str, Any]] = None,
+        __metadata__: Optional[Dict[str, Any]] = None,
+        __files__: Optional[List[Dict[str, Any]]] = None,
+    ) -> AsyncGenerator[str, None]:
+        if not self.valves.OPEN_TERMINAL_API_KEY:
+            yield "_Sandbox not configured: set the OPEN_TERMINAL_API_KEY valve._"
+            return
+
+        prompt = _extract_latest_user_prompt(body)
+        if not prompt:
+            yield "_No user message to send to Claude Code._"
+            return
+
+        chat_id = __chat_id__ or "default"
+        owui_user_id = (__user__ or {}).get("id") or "anon"
+        # The sandbox's X-User-Id drives Linux account provisioning. Tie it
+        # to OWUI's stable user.id (not email/name — those can change).
+        sandbox_user = owui_user_id
+        # Render the workdir through _safe_chat_id so it's safe to embed
+        # unquoted in shell commands. Unquoted is required for `~` and
+        # `$HOME` in the template to expand.
+        workdir = self.valves.WORKDIR_TEMPLATE.format(
+            chat_id=_safe_chat_id(chat_id)
+        )
+
+        allowed_tools = [
+            t.strip() for t in self.valves.ALLOWED_TOOLS.split(",") if t.strip()
+        ]
+        cfg = _ClaudeRunConfig(
+            model=self.valves.MODEL,
+            permission_mode=self.valves.PERMISSION_MODE,
+            allowed_tools=allowed_tools,
+            max_turns=self.valves.MAX_TURNS,
+            resume_session_id=_chat_sessions.get(chat_id),
+            oauth_token=self.valves.CLAUDE_CODE_OAUTH_TOKEN or None,
+            api_key=self.valves.ANTHROPIC_API_KEY or None,
+            workdir=workdir,
+            system_prompt_append=_extract_system_prompt(body),
+        )
+
+        async def emit_status(description: str, done: bool = False) -> None:
+            if __event_emitter__ is None:
+                return
+            await __event_emitter__(
+                {"type": "status", "data": {"description": description, "done": done}}
+            )
+
+        active_tools: Dict[str, Dict[str, Any]] = {}
+        heartbeat_task: Optional[asyncio.Task] = None
+
+        async def _heartbeat() -> None:
+            try:
+                while active_tools:
+                    await asyncio.sleep(2)
+                    if not active_tools:
+                        return
+                    oldest = min(active_tools.values(), key=lambda t: t["started"])
+                    elapsed = int(time.monotonic() - oldest["started"])
+                    count = len(active_tools)
+                    label = (
+                        oldest["label"]
+                        if count == 1
+                        else f"{count} tools · longest {oldest['label']}"
+                    )
+                    await emit_status(f"⏳ {label} · running {elapsed}s…")
+            except asyncio.CancelledError:
+                pass
+
+        def _ensure_heartbeat() -> None:
+            nonlocal heartbeat_task
+            if heartbeat_task is None or heartbeat_task.done():
+                heartbeat_task = asyncio.create_task(_heartbeat())
+
+        await emit_status(f"Starting Claude Code in sandbox ({_sanitize_user_id(sandbox_user)})…")
+
+        try:
+            async with OpenTerminalClient(
+                self.valves.OPEN_TERMINAL_URL,
+                self.valves.OPEN_TERMINAL_API_KEY,
+            ) as client:
+                await client.ensure_dir(sandbox_user, workdir)
+                before_snapshot = await _snapshot_workspace(
+                    client, sandbox_user, workdir
+                )
+
+                async for event in _stream_claude_events(
+                    client, sandbox_user, prompt, cfg, session_id=chat_id
+                ):
+                    async for out in _handle_event(
+                        event,
+                        chat_id=chat_id,
+                        active_tools=active_tools,
+                        ensure_heartbeat=_ensure_heartbeat,
+                        emit_status=emit_status,
+                    ):
+                        yield out
+
+                await emit_status("Done.", done=True)
+                for chunk in await _inline_new_artifacts(
+                    client, sandbox_user, workdir, before_snapshot, owui_user_id
+                ):
+                    yield chunk
+
+        except Exception as exc:
+            log.exception("Sandboxed Claude Code pipe failed")
+            await emit_status(f"Error: {exc}", done=True)
+            yield f"\n\n**Claude Code error:** `{type(exc).__name__}: {exc}`\n"
+        finally:
+            active_tools.clear()
+            if heartbeat_task is not None and not heartbeat_task.done():
+                heartbeat_task.cancel()
+                try:
+                    await heartbeat_task
+                except asyncio.CancelledError:
+                    pass
+
+
+async def _handle_event(
+    event: dict,
+    *,
+    chat_id: str,
+    active_tools: Dict[str, Dict[str, Any]],
+    ensure_heartbeat: Callable[[], None],
+    emit_status: Callable[..., Any],
+) -> AsyncGenerator[str, None]:
+    """Translate one stream-json event into UI output.
+
+    Event shapes (from `claude --output-format stream-json --verbose`):
+      system    — {type:"system", subtype:"init", session_id:"..."}
+      assistant — {type:"assistant", message:{content:[{type:"text"|"tool_use"|"thinking",...}]}}
+      user      — {type:"user", message:{content:[{type:"tool_result", tool_use_id, content, is_error}]}}
+      result    — {type:"result", subtype:"success"|..., total_cost_usd, duration_ms}
+      _stderr/_exit/_raw — synthetic transport-layer signals from our runner.
+    """
+    etype = event.get("type")
+
+    if etype == "system":
+        if event.get("subtype") == "init":
+            sid = event.get("session_id")
+            if sid:
+                _chat_sessions[chat_id] = sid
+        return
+
+    if etype == "assistant":
+        content = ((event.get("message") or {}).get("content")) or []
+        for block in content:
+            btype = block.get("type")
+            if btype == "text":
+                yield block.get("text", "")
+            elif btype == "thinking":
+                text = (block.get("thinking") or "").strip()
+                if text:
+                    await emit_status("💭 Thinking…")
+                    yield (
+                        "\n\n<details>\n<summary>💭 Thinking</summary>\n\n"
+                        f"{text}\n\n</details>\n\n"
+                    )
+            elif btype == "tool_use":
+                name = block.get("name", "tool")
+                tool_input = block.get("input") or {}
+                tool_id = block.get("id", "")
+                preview = _tool_preview(name, tool_input)
+                label = f"{name}: {preview}" if preview else name
+                await emit_status(f"🔧 {label}")
+                active_tools[tool_id] = {"label": label, "started": time.monotonic()}
+                ensure_heartbeat()
+                summary_text = f"🔧 {name}" + (f" · {preview}" if preview else "")
+                body_block = _tool_input_block(name, tool_input)
+                yield (
+                    "\n\n<details>\n"
+                    f"<summary>{summary_text}</summary>\n\n"
+                    f"{body_block}\n\n</details>\n\n"
+                )
+        return
+
+    if etype == "user":
+        content = ((event.get("message") or {}).get("content")) or []
+        for block in content:
+            if block.get("type") != "tool_result":
+                continue
+            active_tools.pop(block.get("tool_use_id", ""), None)
+            if block.get("is_error"):
+                err = _format_tool_result(block.get("content"))[:800]
+                yield (
+                    "\n\n<details>\n<summary>"
+                    "<sub>⚙️ tool hiccup (retrying)</sub>"
+                    "</summary>\n\n"
+                    f"```\n{err}\n```\n\n</details>\n\n"
+                )
+        return
+
+    if etype == "result":
+        subtype = event.get("subtype")
+        if subtype and subtype != "success":
+            yield f"\n\n_Agent stopped: {subtype}_\n"
+        cost = event.get("total_cost_usd")
+        dur = event.get("duration_ms")
+        if cost is not None and dur is not None:
+            yield f"\n\n_Cost: ${float(cost):.4f} · {dur}ms_\n"
+        return
+
+    if etype == "_stderr":
+        # claude CLI prints auth/config warnings on stderr. Log but don't surface
+        # to the chat unless they're fatal — the _exit event will carry that.
+        log.debug("claude stderr: %s", event.get("text", "").strip())
+        return
+
+    if etype == "_exit":
+        code = event.get("code", 0)
+        if code != 0:
+            yield f"\n\n_Sandbox process exited with code {code}._\n"
+        return
+
+    if etype == "_raw":
+        log.debug("unparsed line from claude: %s", event.get("text", ""))
+        return
diff --git a/sandbox/Dockerfile b/sandbox/Dockerfile
new file mode 100644
index 0000000..49c8b81
--- /dev/null
+++ b/sandbox/Dockerfile
@@ -0,0 +1,9 @@
+FROM ghcr.io/open-webui/open-terminal:latest
+
+# Base image runs as uid=1000(user) with npm prefix /usr, which requires
+# root to write. Switch to root for the global install, then back so the
+# container's runtime user and entrypoint stay identical to upstream.
+USER root
+RUN npm install -g @anthropic-ai/claude-code@latest \
+    && claude --version
+USER user
diff --git a/sandbox/README.md b/sandbox/README.md
new file mode 100644
index 0000000..b66aabf
--- /dev/null
+++ b/sandbox/README.md
@@ -0,0 +1,75 @@
+# Sandboxed Claude Code via open-terminal
+
+Runs the Claude Code agent inside [open-webui/open-terminal](https://github.com/open-webui/open-terminal) instead of directly in the Open WebUI backend process. Each Open WebUI user gets a dedicated Linux account (via `OPEN_TERMINAL_MULTI_USER=true`), so files, processes, and commands are isolated by standard Unix permissions.
+
+> **Threat model:** small, trusted groups. One shared kernel, no hard multi-tenant boundaries. Good enough to stop accidental cross-user damage and to keep the agent out of the Open WebUI host's filesystem. Not a substitute for microVMs if you're exposing this to untrusted users.
+
+## Components
+
+| File | Purpose |
+| --- | --- |
+| `Dockerfile` | Extends `ghcr.io/open-webui/open-terminal` with `@anthropic-ai/claude-code` preinstalled. |
+| `docker-compose.yml` | Runs the sandbox on `:8000` with multi-user mode and a named volume for `/home`. |
+| `open_terminal_client.py` | Async HTTP client: `start()`, `stream_output()`, `read_file()`, `write_file()`. |
+| `claude_runner.py` | Builds + invokes the `claude --output-format stream-json` command in the user's account and yields parsed events. |
+
+## Why this architecture
+
+The existing `claude_agent_pipe.py` uses the Claude Agent SDK in-process — which means Bash/Read/Write tool calls hit the Open WebUI host's filesystem with whatever permissions that process has. That's fine for a solo dev setup, dangerous in any shared deployment.
+
+The Agent SDK has no "run on a remote host" hook: it always spawns `claude` locally. So instead of using the SDK, we invoke the `claude` CLI directly with `--output-format stream-json` inside the sandbox. The event stream that comes back is identical to what the SDK surfaces (system/assistant/user/result messages), just newline-delimited JSON — so the existing pipe's renderer can stay largely unchanged.
+
+## Bringing up the sandbox
+
+```bash
+cd sandbox
+echo "OPEN_TERMINAL_API_KEY=$(openssl rand -hex 32)" > .env
+docker compose up -d --build
+curl -s http://localhost:8000/health
+```
+
+Quick smoke test, impersonating two different OWUI users to verify isolation:
+
+```bash
+API_KEY=$(grep OPEN_TERMINAL_API_KEY .env | cut -d= -f2)
+
+# User A writes a secret
+curl -s http://localhost:8000/execute \
+  -H "Authorization: Bearer $API_KEY" \
+  -H "X-User-Id: alice" \
+  -H "Content-Type: application/json" \
+  -d '{"command": "echo hunter2 > ~/secret.txt && ls -la ~"}'
+
+# User B tries to read it (should fail — different /home/owui_bob)
+curl -s http://localhost:8000/execute \
+  -H "Authorization: Bearer $API_KEY" \
+  -H "X-User-Id: bob" \
+  -H "Content-Type: application/json" \
+  -d '{"command": "cat /home/owui_alice/secret.txt || echo DENIED"}'
+```
+
+## Wiring into the pipe
+
+In `claude_agent_pipe.py`, replace the SDK-backed code path with a call into `claude_runner.run_claude()`, passing:
+
+- `user_id = __user__["id"]` — Open WebUI injects the user object into `pipe()`; its stable `id` becomes the sandbox account prefix.
+- `cfg.resume_session_id` — look up the `chat_id → claude_session_id` map the same way the existing pipe does. The first event from the stream (`type: "system"`, `subtype: "init"`) carries the new session id; stash it.
+- `cfg.workdir = f"~/chat-{chat_id}"` — per-chat subdirectory inside the user's home. Carries artifacts across turns without leaking between chats.
+
+Event shape differences from the SDK:
+
+| SDK type | stream-json equivalent |
+| --- | --- |
+| `AssistantMessage.content[ToolUseBlock]` | `{"type":"assistant","message":{"content":[{"type":"tool_use",...}]}}` |
+| `UserMessage.content[ToolResultBlock]` | `{"type":"user","message":{"content":[{"type":"tool_result",...}]}}` |
+| `ResultMessage` | `{"type":"result","subtype":"success","total_cost_usd":...}` |
+| `SystemMessage(subtype="init")` | `{"type":"system","subtype":"init","session_id":"..."}` |
+
+The renderer's `_tool_preview` / `_tool_input_block` helpers already work on the `tool_use.input` dict, so they port over directly.
+
+## Open questions / next steps
+
+- **Artifacts:** the existing pipe scans `cwd` post-run for generated PDFs/CSVs/images and uploads them as OWUI artifacts. In the sandbox version, swap the filesystem scan for `client.list_files(user, workdir)` + `client.read_file()`.
+- **Image context:** when the user attaches images in the chat, `client.write_file()` them into the workdir before invoking claude, then reference by path in the prompt.
+- **Session resume:** store `claude_session_id` per `chat_id` in-process (same as the current pipe). Needs testing that `claude --resume` works cleanly across separate `POST /execute` calls — each call is a fresh process, but claude persists session state to `~/.claude/` inside the user's home.
+- **Cold start:** first request per user spawns `useradd`; measure and decide whether to pre-warm on Open WebUI login.
diff --git a/sandbox/claude_runner.py b/sandbox/claude_runner.py
new file mode 100644
index 0000000..dac1830
--- /dev/null
+++ b/sandbox/claude_runner.py
@@ -0,0 +1,132 @@
+"""Run Claude Code inside an open-terminal sandbox.
+
+Replaces the Claude Agent SDK's in-process invocation with a remote one: we
+shell the `claude` CLI inside a per-user Linux account on the open-terminal
+container, and consume its stream-json output.
+
+Why stream-json over the SDK: the SDK runs `claude` in the *local* process;
+there's no hook for "spawn it on a remote machine." Shelling `claude` with
+`--output-format stream-json` gives us the same event stream (assistant
+messages, tool_use, tool_result, result) that the SDK surfaces, just as
+newline-delimited JSON we can forward straight to the pipe's UI renderer.
+"""
+
+from __future__ import annotations
+
+import json
+import shlex
+from dataclasses import dataclass
+from typing import AsyncIterator, Optional
+
+from open_terminal_client import OpenTerminalClient, ProcessHandle
+
+
+@dataclass
+class ClaudeRunConfig:
+    model: str = "claude-haiku-4-5"
+    permission_mode: str = "bypassPermissions"
+    allowed_tools: tuple[str, ...] = (
+        "Read", "Write", "Edit", "Bash", "Glob", "Grep", "WebSearch", "WebFetch",
+    )
+    max_turns: int = 30
+    # Persisted across turns: `claude --resume <session_id>` picks up
+    # conversation state. We store the id per OWUI chat on the pipe side.
+    resume_session_id: Optional[str] = None
+    # OAuth token or API key. Passed via env so it never hits argv.
+    oauth_token: Optional[str] = None
+    api_key: Optional[str] = None
+    # Working dir inside the user's home. The pipe should stick to one
+    # directory per OWUI chat so artifacts persist across turns.
+    workdir: str = "~/workspace"
+
+
+def _build_env(cfg: ClaudeRunConfig) -> dict[str, str]:
+    env: dict[str, str] = {}
+    if cfg.oauth_token:
+        # OAuth takes priority — matches the existing pipe's contract.
+        env["CLAUDE_CODE_OAUTH_TOKEN"] = cfg.oauth_token
+    elif cfg.api_key:
+        env["ANTHROPIC_API_KEY"] = cfg.api_key
+    return env
+
+
+def _env_prefix(env: dict[str, str]) -> str:
+    """Render env assignments as a safely-quoted shell prefix.
+    shlex.quote guarantees no injection via the values."""
+    return " ".join(f"{k}={shlex.quote(v)}" for k, v in env.items())
+
+
+def _build_claude_command(prompt: str, cfg: ClaudeRunConfig) -> str:
+    parts = [
+        "claude",
+        "--print",
+        "--output-format", "stream-json",
+        "--verbose",
+        "--model", shlex.quote(cfg.model),
+        "--permission-mode", shlex.quote(cfg.permission_mode),
+        "--allowed-tools", shlex.quote(",".join(cfg.allowed_tools)),
+    ]
+    if cfg.max_turns and cfg.max_turns > 0:
+        parts += ["--max-turns", str(cfg.max_turns)]
+    if cfg.resume_session_id:
+        parts += ["--resume", shlex.quote(cfg.resume_session_id)]
+    parts.append(shlex.quote(prompt))
+    return " ".join(parts)
+
+
+async def run_claude(
+    client: OpenTerminalClient,
+    user_id: str,
+    prompt: str,
+    cfg: ClaudeRunConfig,
+    *,
+    session_id: Optional[str] = None,
+) -> AsyncIterator[dict]:
+    """Yield parsed stream-json events from Claude Code running in the sandbox.
+
+    Each event is a dict with a `type` key: 'system', 'assistant', 'user'
+    (tool results), 'result'. The pipe's renderer already understands these
+    shapes from the SDK path — they're the same schema, just transport-wrapped.
+    """
+    env = _build_env(cfg)
+    # Ensure workspace exists, then run claude. A single compound command
+    # keeps it one process the pipe can cancel with one DELETE.
+    mkdir = f"mkdir -p {shlex.quote(cfg.workdir)}"
+    cd = f"cd {shlex.quote(cfg.workdir)}"
+    prefix = _env_prefix(env)
+    cmd = _build_claude_command(prompt, cfg)
+    full = f"{mkdir} && {cd} && {prefix} {cmd}"
+
+    handle: ProcessHandle = await client.start(
+        user_id=user_id,
+        command=full,
+        session_id=session_id,
+    )
+
+    # open-terminal merges stdout/stderr via PTY, so auth warnings printed
+    # by `claude` on stderr will interleave with stream-json on stdout.
+    # Non-JSON lines surface as synthetic _raw events.
+    buf = ""
+    async for stream, chunk in client.stream_output(handle, session_id=session_id):
+        if stream == "exit":
+            if buf.strip():
+                try:
+                    yield json.loads(buf)
+                except json.JSONDecodeError:
+                    yield {"type": "_raw", "text": buf.strip()}
+                buf = ""
+            yield {"type": "_exit", "code": int(chunk)}
+            return
+        buf += chunk
+        # stream-json is newline-delimited; one JSON object per line.
+        while "\n" in buf:
+            line, buf = buf.split("\n", 1)
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                yield json.loads(line)
+            except json.JSONDecodeError:
+                # Claude CLI prints warnings before the stream starts;
+                # expose them rather than swallow silently.
+                yield {"type": "_raw", "text": line}
diff --git a/sandbox/docker-compose.yml b/sandbox/docker-compose.yml
new file mode 100644
index 0000000..539004a
--- /dev/null
+++ b/sandbox/docker-compose.yml
@@ -0,0 +1,20 @@
+services:
+  open-terminal:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    image: helm/open-terminal-claude:latest
+    container_name: open-terminal
+    ports:
+      - "8000:8000"
+    environment:
+      OPEN_TERMINAL_MULTI_USER: "true"
+      OPEN_TERMINAL_API_KEY: ${OPEN_TERMINAL_API_KEY:?set in .env}
+      OPEN_TERMINAL_USER_PREFIX: "owui_"
+      EXECUTE_TIMEOUT: "600"
+    volumes:
+      - open-terminal-home:/home
+    restart: unless-stopped
+
+volumes:
+  open-terminal-home:
diff --git a/sandbox/open_terminal_client.py b/sandbox/open_terminal_client.py
new file mode 100644
index 0000000..48deafb
--- /dev/null
+++ b/sandbox/open_terminal_client.py
@@ -0,0 +1,183 @@
+"""Async client for the open-webui/open-terminal HTTP API.
+
+Scoped to what the Claude Code pipe needs: run a command as a specific
+end-user (via X-User-Id), poll its streaming output, and pull back any
+artifact files the agent produces.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import AsyncIterator, Optional
+
+import httpx
+
+_USER_ID_SAFE = re.compile(r"[^a-zA-Z0-9_-]")
+
+
+def sanitize_user_id(raw: str) -> str:
+    """Mirror open-terminal's own sanitation so we can predict the Linux
+    username it will provision. The server sanitizes server-side too; doing
+    it here lets callers log/reference the resolved identity."""
+    return _USER_ID_SAFE.sub("", raw)[:32] or "anon"
+
+
+@dataclass
+class ProcessHandle:
+    process_id: str
+    user_id: str
+
+
+class OpenTerminalError(RuntimeError):
+    pass
+
+
+class OpenTerminalClient:
+    def __init__(
+        self,
+        base_url: str,
+        api_key: str,
+        *,
+        timeout: float = 30.0,
+    ):
+        self._base = base_url.rstrip("/")
+        self._headers = {"Authorization": f"Bearer {api_key}"}
+        self._timeout = timeout
+        self._client: Optional[httpx.AsyncClient] = None
+
+    async def __aenter__(self):
+        self._client = httpx.AsyncClient(
+            base_url=self._base,
+            headers=self._headers,
+            timeout=self._timeout,
+        )
+        return self
+
+    async def __aexit__(self, *_exc):
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
+
+    def _user_headers(self, user_id: str, session_id: Optional[str] = None) -> dict:
+        h = {"X-User-Id": sanitize_user_id(user_id)}
+        if session_id:
+            h["X-Session-Id"] = session_id
+        return h
+
+    async def start(
+        self,
+        user_id: str,
+        command: str,
+        *,
+        cwd: Optional[str] = None,
+        stdin: Optional[str] = None,
+        session_id: Optional[str] = None,
+    ) -> ProcessHandle:
+        """Start a command in the user's account. Returns immediately with a
+        process handle; use `stream_output` to consume stdout/stderr."""
+        assert self._client is not None, "use as async context manager"
+        payload: dict = {"command": command}
+        if cwd:
+            payload["cwd"] = cwd
+        if stdin is not None:
+            payload["stdin"] = stdin
+        r = await self._client.post(
+            "/execute",
+            json=payload,
+            headers=self._user_headers(user_id, session_id),
+        )
+        if r.status_code >= 400:
+            raise OpenTerminalError(f"POST /execute {r.status_code}: {r.text}")
+        return ProcessHandle(process_id=r.json()["id"], user_id=user_id)
+
+    async def stream_output(
+        self,
+        handle: ProcessHandle,
+        *,
+        session_id: Optional[str] = None,
+        wait: float = 10.0,
+    ) -> AsyncIterator[tuple[str, str]]:
+        """Long-poll status with `wait=N&offset=K` to yield new output
+        entries as they arrive. Yields ('output', text) for each chunk and
+        a final ('exit', exit_code_str). Stdout and stderr are merged by
+        the server's PTY — we strip the resulting CRLFs."""
+        assert self._client is not None
+        headers = self._user_headers(handle.user_id, session_id)
+        offset = 0
+        read_timeout = wait + 10.0
+        while True:
+            r = await self._client.get(
+                f"/execute/{handle.process_id}/status",
+                params={"wait": wait, "offset": offset},
+                headers=headers,
+                timeout=read_timeout,
+            )
+            if r.status_code >= 400:
+                raise OpenTerminalError(
+                    f"status {handle.process_id}: {r.status_code} {r.text}"
+                )
+            data = r.json()
+            for entry in data.get("output") or []:
+                text = (entry.get("data") or "").replace("\r\n", "\n").replace("\r", "\n")
+                if text:
+                    yield "output", text
+            offset = data.get("next_offset", offset)
+            if data.get("status") != "running":
+                yield "exit", str(data.get("exit_code") or 0)
+                return
+
+    async def kill(self, handle: ProcessHandle, *, force: bool = False) -> None:
+        assert self._client is not None
+        await self._client.delete(
+            f"/execute/{handle.process_id}",
+            params={"force": str(force).lower()},
+            headers=self._user_headers(handle.user_id),
+        )
+
+    async def send_stdin(self, handle: ProcessHandle, data: str) -> None:
+        assert self._client is not None
+        r = await self._client.post(
+            f"/execute/{handle.process_id}/input",
+            json={"input": data},
+            headers=self._user_headers(handle.user_id),
+        )
+        if r.status_code >= 400:
+            raise OpenTerminalError(f"stdin: {r.status_code} {r.text}")
+
+    # ---- file ops (used to pull artifacts Claude produces in $HOME) ----
+
+    async def read_file(self, user_id: str, path: str) -> bytes:
+        assert self._client is not None
+        r = await self._client.get(
+            "/files/read",
+            params={"path": path},
+            headers=self._user_headers(user_id),
+        )
+        if r.status_code >= 400:
+            raise OpenTerminalError(f"read {path}: {r.status_code}")
+        return r.content
+
+    async def list_files(self, user_id: str, path: str = "~") -> list[dict]:
+        assert self._client is not None
+        r = await self._client.get(
+            "/files/list",
+            params={"path": path},
+            headers=self._user_headers(user_id),
+        )
+        if r.status_code >= 400:
+            raise OpenTerminalError(f"list {path}: {r.status_code}")
+        return r.json().get("entries", [])
+
+    async def write_file(self, user_id: str, path: str, content: bytes | str) -> None:
+        assert self._client is not None
+        if isinstance(content, str):
+            content = content.encode("utf-8")
+        r = await self._client.post(
+            "/files/write",
+            params={"path": path},
+            content=content,
+            headers=self._user_headers(user_id),
+        )
+        if r.status_code >= 400:
+            raise OpenTerminalError(f"write {path}: {r.status_code} {r.text}")
diff --git a/sandbox/smoke_test.py b/sandbox/smoke_test.py
new file mode 100644
index 0000000..24ca819
--- /dev/null
+++ b/sandbox/smoke_test.py
@@ -0,0 +1,60 @@
+"""Smoke test: run Claude Code inside the sandbox as a fake OWUI user.
+
+Run from the `sandbox/` directory after `docker compose up -d`:
+
+    OPEN_TERMINAL_API_KEY=$(grep OPEN_TERMINAL_API_KEY .env | cut -d= -f2) \
+    ANTHROPIC_API_KEY=... \
+    python -m smoke_test
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import sys
+
+from open_terminal_client import OpenTerminalClient
+from claude_runner import ClaudeRunConfig, run_claude
+
+
+async def main() -> int:
+    base_url = os.environ.get("OPEN_TERMINAL_URL", "http://localhost:8000")
+    api_key = os.environ["OPEN_TERMINAL_API_KEY"]
+    anthropic_key = os.environ.get("ANTHROPIC_API_KEY")
+    oauth_token = os.environ.get("CLAUDE_CODE_OAUTH_TOKEN")
+
+    if not (anthropic_key or oauth_token):
+        print("Set ANTHROPIC_API_KEY or CLAUDE_CODE_OAUTH_TOKEN", file=sys.stderr)
+        return 2
+
+    cfg = ClaudeRunConfig(
+        model=os.environ.get("CLAUDE_MODEL", "claude-haiku-4-5"),
+        api_key=anthropic_key,
+        oauth_token=oauth_token,
+        workdir="~/chat-smoke",
+        max_turns=5,
+    )
+
+    prompt = (
+        "Create a file called hello.py that prints 'hi from sandbox' and "
+        "then run it. Report the output."
+    )
+
+    async with OpenTerminalClient(base_url, api_key) as client:
+        async for event in run_claude(client, user_id="smoke", prompt=prompt, cfg=cfg):
+            etype = event.get("type", "?")
+            if etype == "_stderr":
+                print(f"[stderr] {event['text']}", end="", file=sys.stderr)
+            elif etype == "_exit":
+                print(f"\n[exit code: {event['code']}]")
+            elif etype == "_raw":
+                print(f"[raw] {event['text']}")
+            else:
+                # Compact dump so you can eyeball the event stream.
+                print(json.dumps(event)[:200])
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))
diff --git a/sync_pipe.py b/sync_pipe.py
new file mode 100644
index 0000000..5e6679c
--- /dev/null
+++ b/sync_pipe.py
@@ -0,0 +1,145 @@
+"""Push the local claude_agent_pipe_sandboxed.py to a running OpenWebUI.
+
+Usage (typical, from this directory):
+    set -a; source /home/thomas/projects/helm_assistant/.env; set +a
+    python3 sync_pipe.py
+
+Behaviour:
+    1. Reads OPENWEBUI_URL (default http://localhost:${OPENWEBUI_PORT:-13000})
+       and OPENWEBUI_API_KEY from the environment. The key never enters this
+       script's logs or output.
+    2. GETs /api/v1/functions/ to find the sandboxed pipe's id. Match
+       priority: explicit `--id`, then exact id == SANDBOXED_PIPE_ID,
+       then a unique function whose content starts with the pipe's
+       `title: Claude Code (Sandboxed)` marker.
+    3. POSTs /api/v1/functions/id/{id}/update with the current file
+       contents, preserving the function's existing `name` / `meta`.
+
+    Existing valves and the `is_active` flag are untouched — the update
+    endpoint only rewrites `content` + `meta.manifest` + `name`.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+SANDBOXED_PIPE_ID = "claude-code-sandboxed"
+SANDBOXED_PIPE_MARKER = "title: Claude Code (Sandboxed)"
+PIPE_FILE = Path(__file__).parent / "claude_agent_pipe_sandboxed.py"
+
+
+def api(method: str, url: str, key: str, body: dict | None = None) -> dict:
+    req = urllib.request.Request(
+        url,
+        method=method,
+        headers={
+            "Authorization": f"Bearer {key}",
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        },
+        data=json.dumps(body).encode("utf-8") if body is not None else None,
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=30) as r:
+            raw = r.read().decode("utf-8")
+            return json.loads(raw) if raw else {}
+    except urllib.error.HTTPError as e:
+        detail = e.read().decode("utf-8", errors="replace")
+        raise SystemExit(f"HTTP {e.code} {method} {url}\n  → {detail[:500]}")
+
+
+def find_function_id(base: str, key: str, explicit_id: str | None) -> tuple[str, dict]:
+    """Return (id, current_function_record)."""
+    functions = api("GET", f"{base}/api/v1/functions/", key)
+    by_id = {f["id"]: f for f in functions}
+
+    if explicit_id:
+        if explicit_id not in by_id:
+            raise SystemExit(
+                f"No function with id={explicit_id!r}. "
+                f"Available: {sorted(by_id)}"
+            )
+        return explicit_id, by_id[explicit_id]
+
+    if SANDBOXED_PIPE_ID in by_id:
+        return SANDBOXED_PIPE_ID, by_id[SANDBOXED_PIPE_ID]
+
+    # Fall back: find functions whose content contains the title marker.
+    # The list endpoint doesn't include `content`, so fetch each candidate
+    # individually. Keep the search narrow — only unloaded pipes need it.
+    matches: list[tuple[str, dict]] = []
+    for fid, meta in by_id.items():
+        if meta.get("type") != "pipe":
+            continue
+        detail = api("GET", f"{base}/api/v1/functions/id/{fid}", key)
+        if SANDBOXED_PIPE_MARKER in (detail.get("content") or ""):
+            matches.append((fid, detail))
+    if len(matches) == 1:
+        return matches[0]
+    if not matches:
+        raise SystemExit(
+            f"No function matches id={SANDBOXED_PIPE_ID!r} and no pipe "
+            f"contains '{SANDBOXED_PIPE_MARKER}'. "
+            f"Create it once in the OWUI UI, or pass --id <existing_id>."
+        )
+    raise SystemExit(
+        f"Multiple pipes contain '{SANDBOXED_PIPE_MARKER}': "
+        f"{[m[0] for m in matches]}. Pass --id explicitly."
+    )
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--id", help="Override function id (default: auto-detect).")
+    ap.add_argument(
+        "--url",
+        default=os.environ.get(
+            "OPENWEBUI_URL",
+            f"http://localhost:{os.environ.get('OPENWEBUI_PORT', '13000')}",
+        ),
+        help="OpenWebUI base URL.",
+    )
+    args = ap.parse_args()
+
+    key = os.environ.get("OPENWEBUI_API_KEY")
+    if not key:
+        print(
+            "OPENWEBUI_API_KEY not set. Source the helm_assistant .env first:\n"
+            "  set -a; source /home/thomas/projects/helm_assistant/.env; set +a",
+            file=sys.stderr,
+        )
+        return 2
+
+    if not PIPE_FILE.exists():
+        print(f"Pipe file not found: {PIPE_FILE}", file=sys.stderr)
+        return 2
+
+    content = PIPE_FILE.read_text(encoding="utf-8")
+    # Pull the pipe's own name from its docstring frontmatter so we don't
+    # overwrite whatever the user titled it. Fall back to a sensible default.
+    m = re.search(r"^title:\s*(.+)$", content, re.M)
+    title = m.group(1).strip() if m else "Claude Code (Sandboxed)"
+
+    base = args.url.rstrip("/")
+    fid, current = find_function_id(base, key, args.id)
+
+    form = {
+        "id": fid,
+        "name": current.get("name") or title,
+        "content": content,
+        "meta": current.get("meta") or {"description": title, "manifest": {}},
+    }
+    api("POST", f"{base}/api/v1/functions/id/{fid}/update", key, form)
+    print(f"✓ Synced {PIPE_FILE.name} → {base} (id={fid}, {len(content)} bytes)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())