research-workbench/docs/plans/2026-05-26-research-workbench-phase1.md

Research Workbench — Phase 1: Infrastructure

Execution: Use the subagent-driven-development workflow to implement this plan.

Goal: Build the Python backend (FastAPI + auth + AG-UI adapter + artifact API) and Docker container infrastructure for the research workbench.

Architecture: A single Docker container runs amplifierd (session daemon), FastAPI (serves frontend + proxies amplifierd via AG-UI protocol), playwright-cli (browser control), and Xvfb/noVNC (virtual display). FastAPI translates between amplifierd's SSE event format and the AG-UI protocol that CopilotKit expects. Tool results with _meta.ui.resourceUri (MCP Apps) are passed through so the frontend can render interactive HTML visualizations in sandboxed iframes. Auth is simple email + bcrypt password hash with session cookies.

Tech Stack: Python 3.13, FastAPI, uvicorn, httpx, bcrypt, ag-ui-protocol, Docker, Xvfb, x11vnc, websockify, noVNC

---

Prerequisites

• The repo is at /home/ken/workspace/research-workbench
• It currently contains old car-help code (search.py, guide.py, sites.yaml, results/, static/, app.py) that will be replaced
• Keep: DESIGN.md, README.md, .gitignore, .python-version, .git/

---

Task 1: Clean out old car-help files

Files:

• Delete: search.py, guide.py, sites.yaml, app.py
• Delete: static/ directory, results/ directory
• Delete: Dockerfile, entrypoint.sh, pyproject.toml, uv.lock
• Modify: .gitignore

Step 1: Remove old files

Run:

cd /home/ken/workspace/research-workbench
rm -f search.py guide.py sites.yaml app.py
rm -rf static/ results/
rm -f Dockerfile entrypoint.sh pyproject.toml uv.lock

Step 2: Update .gitignore for new project structure

Replace .gitignore with:

# Python
.venv/
__pycache__/
*.pyc
*.egg-info/

# Node / Frontend
frontend/node_modules/
frontend/dist/

# Runtime
artifacts/

# IDE
.idea/
.vscode/
*.swp

Step 3: Create directory structure

Run:

cd /home/ken/workspace/research-workbench
mkdir -p backend tests bundle/behaviors bundle/agents bundle/context frontend/src artifacts docs/plans

Step 4: Commit

Run:

cd /home/ken/workspace/research-workbench
git add -A && git commit -m "chore: remove old car-help files, prepare new structure"

---

Task 2: Create backend pyproject.toml

Files:

• Create: backend/pyproject.toml

Step 1: Create the pyproject.toml

Create backend/pyproject.toml:

[project]
name = "research-workbench"
version = "0.1.0"
description = "AI research workbench backend - FastAPI + AG-UI adapter for amplifierd"
requires-python = ">=3.13"
dependencies = [
    "fastapi>=0.115.0",
    "uvicorn[standard]>=0.34.0",
    "httpx>=0.28.0",
    "bcrypt>=4.2.0",
    "python-multipart>=0.0.18",
    "ag-ui-protocol>=0.1.18",
]

[dependency-groups]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.24",
    "httpx>=0.28.0",
]

Step 2: Initialize the venv and install deps

Run:

cd /home/ken/workspace/research-workbench/backend
uv sync

Expected: Dependencies install successfully. A uv.lock file is created.

Step 3: Verify ag-ui-protocol is importable

Run:

cd /home/ken/workspace/research-workbench/backend
uv run python -c "from ag_ui.core import RunAgentInput, EventType; print('ag-ui-protocol OK')"

Expected: ag-ui-protocol OK

Step 4: Commit

Run:

cd /home/ken/workspace/research-workbench
git add backend/pyproject.toml backend/uv.lock && git commit -m "feat: add backend pyproject.toml with dependencies"

---

Task 3: Auth module (email + bcrypt + session cookie)

Files:

• Create: backend/auth.py
• Create: tests/test_auth.py

Step 1: Write the failing test

Create tests/test_auth.py:

"""Tests for auth module."""

import os
import sys
import pytest
from unittest.mock import patch

# Add backend to path so tests can import modules
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))

from auth import verify_password, create_session_token, validate_session_token


class TestVerifyPassword:
    """Test bcrypt password verification."""

    def test_correct_password(self):
        import bcrypt
        hashed = bcrypt.hashpw(b"testpass123", bcrypt.gensalt()).decode()
        with patch.dict(os.environ, {"AUTH_PASS_HASH": hashed}):
            assert verify_password("testpass123") is True

    def test_wrong_password(self):
        import bcrypt
        hashed = bcrypt.hashpw(b"testpass123", bcrypt.gensalt()).decode()
        with patch.dict(os.environ, {"AUTH_PASS_HASH": hashed}):
            assert verify_password("wrongpassword") is False

    def test_empty_password(self):
        import bcrypt
        hashed = bcrypt.hashpw(b"testpass123", bcrypt.gensalt()).decode()
        with patch.dict(os.environ, {"AUTH_PASS_HASH": hashed}):
            assert verify_password("") is False


class TestSessionToken:
    """Test session token creation and validation."""

    def test_create_and_validate(self):
        token = create_session_token("ken@example.com")
        assert token is not None
        assert len(token) > 20

    def test_validate_valid_token(self):
        token = create_session_token("ken@example.com")
        email = validate_session_token(token)
        assert email == "ken@example.com"

    def test_validate_invalid_token(self):
        email = validate_session_token("bogus-token-value")
        assert email is None

    def test_validate_empty_token(self):
        email = validate_session_token("")
        assert email is None

Step 2: Run test to verify it fails

Run:

cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_auth.py -v 2>&1 | head -20

Expected: FAIL with ModuleNotFoundError: No module named 'auth'

Step 3: Write the implementation

Create backend/auth.py:

"""
Simple email + password authentication with session cookies.

Auth credentials come from environment variables:
  AUTH_USER      - the allowed email address
  AUTH_PASS_HASH - bcrypt hash of the password

Session tokens are stored in-memory (single container, single user).
Token format: random hex string mapped to email in a dict.
"""

import os
import secrets

import bcrypt

AUTH_USER: str = os.environ.get("AUTH_USER", "admin@localhost")
AUTH_PASS_HASH: str = os.environ.get("AUTH_PASS_HASH", "")

# In-memory session store: token -> email
_sessions: dict[str, str] = {}


def verify_password(password: str) -> bool:
    """Check a plaintext password against the AUTH_PASS_HASH env var."""
    stored_hash = os.environ.get("AUTH_PASS_HASH", AUTH_PASS_HASH)
    if not stored_hash or not password:
        return False
    try:
        return bcrypt.checkpw(password.encode("utf-8"), stored_hash.encode("utf-8"))
    except (ValueError, TypeError):
        return False


def create_session_token(email: str) -> str:
    """Create a new session token for the given email."""
    token = secrets.token_hex(32)
    _sessions[token] = email
    return token


def validate_session_token(token: str) -> str | None:
    """Return the email for a valid token, or None."""
    if not token:
        return None
    return _sessions.get(token)


def invalidate_session_token(token: str) -> None:
    """Remove a session token."""
    _sessions.pop(token, None)

Step 4: Run tests to verify they pass

Run:

cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_auth.py -v

Expected: All 6 tests PASS.

Step 5: Commit

Run:

cd /home/ken/workspace/research-workbench
git add backend/auth.py tests/test_auth.py && git commit -m "feat: auth module with bcrypt password verification and session tokens"

---

Task 4: Artifacts module (per-session CRUD)

Files:

• Create: backend/artifacts.py
• Create: tests/test_artifacts.py

Step 1: Write the failing test

Create tests/test_artifacts.py:

"""Tests for artifacts module."""

import os
import sys
import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))

from artifacts import list_artifacts, get_artifact, save_artifact


@pytest.fixture(autouse=True)
def temp_artifacts(tmp_path, monkeypatch):
    """Use a temp directory for artifacts during tests."""
    monkeypatch.setattr("artifacts.ARTIFACTS_DIR", tmp_path)
    return tmp_path


class TestArtifacts:
    def test_list_empty_session(self):
        result = list_artifacts("session-1")
        assert result == []

    def test_save_and_list(self, temp_artifacts):
        save_artifact("session-1", "guide.md", "# My Guide\n\nContent here.")
        result = list_artifacts("session-1")
        assert len(result) == 1
        assert result[0]["name"] == "guide.md"
        assert result[0]["session_id"] == "session-1"

    def test_save_and_get(self, temp_artifacts):
        save_artifact("session-1", "report.md", "# Report\n\nFindings.")
        content = get_artifact("session-1", "report.md")
        assert content == "# Report\n\nFindings."

    def test_get_nonexistent(self):
        content = get_artifact("no-session", "no-file.md")
        assert content is None

    def test_multiple_artifacts(self, temp_artifacts):
        save_artifact("s1", "a.md", "A")
        save_artifact("s1", "b.md", "B")
        save_artifact("s2", "c.md", "C")
        assert len(list_artifacts("s1")) == 2
        assert len(list_artifacts("s2")) == 1

    def test_overwrite_artifact(self, temp_artifacts):
        save_artifact("s1", "doc.md", "v1")
        save_artifact("s1", "doc.md", "v2")
        assert get_artifact("s1", "doc.md") == "v2"
        assert len(list_artifacts("s1")) == 1

Step 2: Run test to verify it fails

Run:

cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_artifacts.py -v 2>&1 | head -10

Expected: FAIL with ModuleNotFoundError: No module named 'artifacts'

Step 3: Write the implementation

Create backend/artifacts.py:

"""
Per-session artifact storage.

Artifacts are markdown files stored at artifacts/<session_id>/<name>.md
Written by the researcher agent via tool-filesystem, served by FastAPI.
"""

from pathlib import Path

ARTIFACTS_DIR = Path("/app/artifacts")


def list_artifacts(session_id: str) -> list[dict]:
    """List all artifacts for a session."""
    session_dir = ARTIFACTS_DIR / session_id
    if not session_dir.is_dir():
        return []
    artifacts = []
    for f in sorted(session_dir.iterdir()):
        if f.is_file():
            artifacts.append({
                "name": f.name,
                "session_id": session_id,
                "size": f.stat().st_size,
            })
    return artifacts


def get_artifact(session_id: str, name: str) -> str | None:
    """Get the content of a specific artifact."""
    path = ARTIFACTS_DIR / session_id / name
    if not path.is_file():
        return None
    return path.read_text(encoding="utf-8")


def save_artifact(session_id: str, name: str, content: str) -> Path:
    """Save (or overwrite) an artifact."""
    session_dir = ARTIFACTS_DIR / session_id
    session_dir.mkdir(parents=True, exist_ok=True)
    path = session_dir / name
    path.write_text(content, encoding="utf-8")
    return path

Step 4: Run tests to verify they pass

Run:

cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_artifacts.py -v

Expected: All 6 tests PASS.

Step 5: Commit

Run:

cd /home/ken/workspace/research-workbench
git add backend/artifacts.py tests/test_artifacts.py && git commit -m "feat: artifact storage module with per-session CRUD"

---

Task 5: AG-UI adapter (amplifierd SSE -> AG-UI events, MCP App metadata passthrough)

This is the critical integration piece. It translates amplifierd's SSE event format into AG-UI protocol events that CopilotKit understands. It also passes through _meta.ui and structuredContent from tool results so the frontend can render MCP App iframes for interactive visualizations.

amplifierd SSE events (from GET /events?session=X):

• content_block:start — {block_type: "text"|"tool_use", block_index, tool_id?, tool_name?}
• content_block:delta — {token?, block_type?, content?, block_index}
• content_block:end — {block_index}
• tool:result — {tool_call_id, tool_name, content}
• orchestrator:complete — {}

AG-UI events (what CopilotKit frontend expects):

• RUN_STARTED, TEXT_MESSAGE_START/CONTENT/END, TOOL_CALL_START/ARGS/END/RESULT, RUN_FINISHED

Files:

• Create: backend/agui_adapter.py
• Create: tests/test_agui_adapter.py

Step 1: Write the failing test

Create tests/test_agui_adapter.py:

"""Tests for the AG-UI adapter that translates amplifierd SSE to AG-UI events."""

import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))

from agui_adapter import AmplifierdEventTranslator


class TestTextBlocks:
    """Test text content block translation."""

    def test_text_block_start(self):
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        event = {
            "event_type": "content_block:start",
            "data": {"block_type": "text", "block_index": 0},
        }
        ag_events = list(translator.translate(event))
        assert len(ag_events) == 1
        assert ag_events[0]["type"] == "TEXT_MESSAGE_START"
        assert ag_events[0]["role"] == "assistant"
        assert "messageId" in ag_events[0]

    def test_text_delta(self):
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        # Start text block first
        translator.translate({
            "event_type": "content_block:start",
            "data": {"block_type": "text", "block_index": 0},
        })
        ag_events = list(translator.translate({
            "event_type": "content_block:delta",
            "data": {"token": "Hello ", "block_index": 0},
        }))
        assert len(ag_events) == 1
        assert ag_events[0]["type"] == "TEXT_MESSAGE_CONTENT"
        assert ag_events[0]["delta"] == "Hello "

    def test_text_block_end(self):
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        translator.translate({
            "event_type": "content_block:start",
            "data": {"block_type": "text", "block_index": 0},
        })
        ag_events = list(translator.translate({
            "event_type": "content_block:end",
            "data": {"block_index": 0},
        }))
        assert len(ag_events) == 1
        assert ag_events[0]["type"] == "TEXT_MESSAGE_END"


class TestToolBlocks:
    """Test tool call translation."""

    def test_tool_use_start(self):
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        ag_events = list(translator.translate({
            "event_type": "content_block:start",
            "data": {
                "block_type": "tool_use",
                "block_index": 1,
                "tool_id": "tool_123",
                "tool_name": "bash",
            },
        }))
        assert len(ag_events) == 1
        assert ag_events[0]["type"] == "TOOL_CALL_START"
        assert ag_events[0]["toolCallName"] == "bash"
        assert ag_events[0]["toolCallId"] == "tool_123"

    def test_tool_use_delta(self):
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        translator.translate({
            "event_type": "content_block:start",
            "data": {
                "block_type": "tool_use",
                "block_index": 1,
                "tool_id": "tool_123",
                "tool_name": "bash",
            },
        })
        ag_events = list(translator.translate({
            "event_type": "content_block:delta",
            "data": {
                "block_type": "tool_use",
                "content": '{"command": "ls"}',
                "block_index": 1,
            },
        }))
        assert len(ag_events) == 1
        assert ag_events[0]["type"] == "TOOL_CALL_ARGS"
        assert ag_events[0]["delta"] == '{"command": "ls"}'

    def test_tool_use_end(self):
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        translator.translate({
            "event_type": "content_block:start",
            "data": {
                "block_type": "tool_use",
                "block_index": 1,
                "tool_id": "tool_123",
                "tool_name": "bash",
            },
        })
        ag_events = list(translator.translate({
            "event_type": "content_block:end",
            "data": {"block_index": 1},
        }))
        assert len(ag_events) == 1
        assert ag_events[0]["type"] == "TOOL_CALL_END"

    def test_tool_result(self):
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        ag_events = list(translator.translate({
            "event_type": "tool:result",
            "data": {
                "tool_call_id": "tool_123",
                "tool_name": "bash",
                "content": "file1.txt\nfile2.txt",
            },
        }))
        assert len(ag_events) == 1
        assert ag_events[0]["type"] == "TOOL_CALL_RESULT"
        assert ag_events[0]["toolCallId"] == "tool_123"

    def test_tool_result_with_mcp_app_metadata(self):
        """Tool results with _meta.ui.resourceUri pass through UI metadata for MCP Apps."""
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        ag_events = list(translator.translate({
            "event_type": "tool:result",
            "data": {
                "tool_call_id": "tool_456",
                "tool_name": "show_chart",
                "content": "Chart rendered with 5 data points",
                "_meta": {
                    "ui": {
                        "resourceUri": "app://chart",
                        "csp": "default-src 'self'; script-src 'unsafe-inline' cdn.jsdelivr.net",
                    }
                },
                "structuredContent": {
                    "chartType": "bar",
                    "labels": ["Q1", "Q2", "Q3"],
                    "datasets": [{"label": "Sales", "data": [10, 20, 30]}],
                },
            },
        }))
        assert len(ag_events) == 1
        assert ag_events[0]["type"] == "TOOL_CALL_RESULT"
        assert ag_events[0]["toolCallId"] == "tool_456"
        # _meta.ui must be passed through for the frontend MCPAppRenderer
        assert ag_events[0]["_meta"]["ui"]["resourceUri"] == "app://chart"
        # structuredContent must be passed through for the iframe data
        assert ag_events[0]["structuredContent"]["chartType"] == "bar"

    def test_tool_result_without_mcp_app_metadata(self):
        """Tool results without _meta.ui should NOT have _meta or structuredContent keys."""
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        ag_events = list(translator.translate({
            "event_type": "tool:result",
            "data": {
                "tool_call_id": "tool_789",
                "tool_name": "bash",
                "content": "hello world",
            },
        }))
        assert len(ag_events) == 1
        assert "_meta" not in ag_events[0]
        assert "structuredContent" not in ag_events[0]


class TestLifecycle:
    """Test lifecycle event translation."""

    def test_orchestrator_complete(self):
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        ag_events = list(translator.translate({
            "event_type": "orchestrator:complete",
            "data": {},
        }))
        assert len(ag_events) == 1
        assert ag_events[0]["type"] == "RUN_FINISHED"
        assert ag_events[0]["threadId"] == "thread-1"
        assert ag_events[0]["runId"] == "run-1"

    def test_unknown_event_ignored(self):
        translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
        ag_events = list(translator.translate({
            "event_type": "some:unknown:event",
            "data": {"foo": "bar"},
        }))
        assert ag_events == []

Step 2: Run test to verify it fails

Run:

cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_agui_adapter.py -v 2>&1 | head -10

Expected: FAIL with ModuleNotFoundError: No module named 'agui_adapter'

Step 3: Write the implementation

Create backend/agui_adapter.py:

"""
AG-UI protocol adapter for amplifierd.

Translates amplifierd's SSE events into AG-UI protocol events that CopilotKit
can consume. This is the critical bridge between the two systems.

amplifierd events (from GET /events?session=X):
  content_block:start  {block_type: "text"|"tool_use", block_index, tool_id?, tool_name?}
  content_block:delta  {token?, block_type?, content?, block_index}
  content_block:end    {block_index}
  tool:result          {tool_call_id, tool_name, content}
  orchestrator:complete {}

AG-UI events (sent to CopilotKit frontend):
  RUN_STARTED          {threadId, runId}
  TEXT_MESSAGE_START    {messageId, role}
  TEXT_MESSAGE_CONTENT  {messageId, delta}
  TEXT_MESSAGE_END      {messageId}
  TOOL_CALL_START       {toolCallId, toolCallName, parentMessageId?}
  TOOL_CALL_ARGS        {toolCallId, delta}
  TOOL_CALL_END         {toolCallId}
  TOOL_CALL_RESULT      {toolCallId, content, role, _meta?, structuredContent?}
  RUN_FINISHED          {threadId, runId}

MCP App passthrough:
  When a tool:result includes _meta.ui.resourceUri, the TOOL_CALL_RESULT event
  preserves _meta and structuredContent so the frontend can render an MCP App
  iframe instead of a plain ToolCallCard.
"""

import uuid


class AmplifierdEventTranslator:
    """Stateful translator from amplifierd SSE events to AG-UI events.

    Tracks active content blocks to correctly pair start/end events and
    assign consistent message IDs across the stream.
    """

    def __init__(self, run_id: str, thread_id: str):
        self.run_id = run_id
        self.thread_id = thread_id
        # Track active blocks: block_index -> {type, message_id, tool_id}
        self._blocks: dict[int, dict] = {}
        # Current assistant message ID (for text blocks)
        self._current_message_id: str | None = None

    def translate(self, event: dict) -> list[dict]:
        """Translate a single amplifierd event into zero or more AG-UI events."""
        event_type = event.get("event_type", "")
        data = event.get("data", {})

        if event_type == "content_block:start":
            return self._handle_block_start(data)
        elif event_type == "content_block:delta":
            return self._handle_block_delta(data)
        elif event_type == "content_block:end":
            return self._handle_block_end(data)
        elif event_type == "tool:result":
            return self._handle_tool_result(data)
        elif event_type == "orchestrator:complete":
            return self._handle_complete()
        else:
            return []

    def _handle_block_start(self, data: dict) -> list[dict]:
        block_type = data.get("block_type", "text")
        block_index = data.get("block_index", 0)

        if block_type == "tool_use":
            tool_id = data.get("tool_id", str(uuid.uuid4()))
            tool_name = data.get("tool_name", "unknown")
            self._blocks[block_index] = {
                "type": "tool_use",
                "tool_id": tool_id,
                "tool_name": tool_name,
            }
            return [{
                "type": "TOOL_CALL_START",
                "toolCallId": tool_id,
                "toolCallName": tool_name,
                "parentMessageId": self._current_message_id,
            }]
        else:
            message_id = str(uuid.uuid4())
            self._current_message_id = message_id
            self._blocks[block_index] = {
                "type": "text",
                "message_id": message_id,
            }
            return [{
                "type": "TEXT_MESSAGE_START",
                "messageId": message_id,
                "role": "assistant",
            }]

    def _handle_block_delta(self, data: dict) -> list[dict]:
        block_index = data.get("block_index", 0)
        block_info = self._blocks.get(block_index)

        if block_info is None:
            # Delta without a matching start - handle gracefully
            if data.get("block_type") == "tool_use":
                tool_id = data.get("tool_id", "unknown")
                return [{
                    "type": "TOOL_CALL_ARGS",
                    "toolCallId": tool_id,
                    "delta": data.get("content", ""),
                }]
            return []

        if block_info["type"] == "tool_use":
            return [{
                "type": "TOOL_CALL_ARGS",
                "toolCallId": block_info["tool_id"],
                "delta": data.get("content", ""),
            }]
        else:
            token = data.get("token", "")
            if not token:
                return []
            return [{
                "type": "TEXT_MESSAGE_CONTENT",
                "messageId": block_info["message_id"],
                "delta": token,
            }]

    def _handle_block_end(self, data: dict) -> list[dict]:
        block_index = data.get("block_index", 0)
        block_info = self._blocks.pop(block_index, None)

        if block_info is None:
            return []

        if block_info["type"] == "tool_use":
            return [{
                "type": "TOOL_CALL_END",
                "toolCallId": block_info["tool_id"],
            }]
        else:
            return [{
                "type": "TEXT_MESSAGE_END",
                "messageId": block_info["message_id"],
            }]

    def _handle_tool_result(self, data: dict) -> list[dict]:
        event: dict = {
            "type": "TOOL_CALL_RESULT",
            "toolCallId": data.get("tool_call_id", "unknown"),
            "content": data.get("content", ""),
            "role": "tool",
        }
        # Pass through MCP App metadata if present.
        # When a tool result includes _meta.ui.resourceUri, the frontend
        # renders an MCPAppRenderer (sandboxed iframe) instead of a ToolCallCard.
        if "_meta" in data:
            event["_meta"] = data["_meta"]
        if "structuredContent" in data:
            event["structuredContent"] = data["structuredContent"]
        return [event]

    def _handle_complete(self) -> list[dict]:
        return [{
            "type": "RUN_FINISHED",
            "threadId": self.thread_id,
            "runId": self.run_id,
        }]

Step 4: Run tests to verify they pass

Run:

cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_agui_adapter.py -v

Expected: All 10 tests PASS.

Step 5: Commit

Run:

cd /home/ken/workspace/research-workbench
git add backend/agui_adapter.py tests/test_agui_adapter.py && git commit -m "feat: AG-UI adapter translating amplifierd SSE events to AG-UI protocol (with MCP App metadata passthrough)"

---

Task 6: FastAPI application (app.py)

The main FastAPI app: auth endpoints, AG-UI/CopilotKit SSE endpoint, artifact API, MCP App template serving, session proxy to amplifierd, and static file serving.

Files:

• Create: backend/app.py
• Create: tests/test_app.py

Step 1: Write the failing test

Create tests/test_app.py:

"""Integration tests for the FastAPI app."""

import os
import sys

import bcrypt
import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))

# Set auth env vars before importing app
TEST_HASH = bcrypt.hashpw(b"testpass", bcrypt.gensalt()).decode()
os.environ["AUTH_USER"] = "test@example.com"
os.environ["AUTH_PASS_HASH"] = TEST_HASH

from fastapi.testclient import TestClient

from app import app

client = TestClient(app)


class TestHealthEndpoint:
    def test_health(self):
        resp = client.get("/api/health")
        assert resp.status_code == 200
        assert resp.json()["status"] == "ok"


class TestAuthEndpoints:
    def test_login_success(self):
        resp = client.post("/api/auth/login", json={
            "email": "test@example.com",
            "password": "testpass",
        })
        assert resp.status_code == 200
        data = resp.json()
        assert "token" in data
        assert "session" in resp.cookies

    def test_login_wrong_password(self):
        resp = client.post("/api/auth/login", json={
            "email": "test@example.com",
            "password": "wrongpass",
        })
        assert resp.status_code == 401

    def test_login_wrong_email(self):
        resp = client.post("/api/auth/login", json={
            "email": "wrong@example.com",
            "password": "testpass",
        })
        assert resp.status_code == 401

    def test_me_unauthenticated(self):
        resp = client.get("/api/auth/me")
        assert resp.status_code == 401

    def test_me_authenticated(self):
        login_resp = client.post("/api/auth/login", json={
            "email": "test@example.com",
            "password": "testpass",
        })
        token = login_resp.json()["token"]
        resp = client.get("/api/auth/me", cookies={"session": token})
        assert resp.status_code == 200
        assert resp.json()["email"] == "test@example.com"

    def test_logout(self):
        login_resp = client.post("/api/auth/login", json={
            "email": "test@example.com",
            "password": "testpass",
        })
        token = login_resp.json()["token"]
        resp = client.post("/api/auth/logout", cookies={"session": token})
        assert resp.status_code == 200
        # Token should now be invalid
        resp2 = client.get("/api/auth/me", cookies={"session": token})
        assert resp2.status_code == 401


class TestArtifactEndpoints:
    def _auth_cookies(self) -> dict:
        resp = client.post("/api/auth/login", json={
            "email": "test@example.com",
            "password": "testpass",
        })
        return {"session": resp.json()["token"]}

    def test_list_artifacts_empty(self, tmp_path, monkeypatch):
        monkeypatch.setattr("artifacts.ARTIFACTS_DIR", tmp_path)
        cookies = self._auth_cookies()
        resp = client.get("/api/artifacts/test-session", cookies=cookies)
        assert resp.status_code == 200
        assert resp.json() == []

    def test_list_artifacts_unauthenticated(self):
        resp = client.get("/api/artifacts/test-session")
        assert resp.status_code == 401


class TestMCPAppsEndpoint:
    """Test serving MCP App HTML templates."""

    def _auth_cookies(self) -> dict:
        resp = client.post("/api/auth/login", json={
            "email": "test@example.com",
            "password": "testpass",
        })
        return {"session": resp.json()["token"]}

    def test_get_app_not_found(self):
        cookies = self._auth_cookies()
        resp = client.get("/api/apps/nonexistent", cookies=cookies)
        assert resp.status_code == 404

    def test_get_app_unauthenticated(self):
        resp = client.get("/api/apps/chart")
        assert resp.status_code == 401

Step 2: Run test to verify it fails

Run:

cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_app.py -v 2>&1 | head -10

Expected: FAIL with ModuleNotFoundError: No module named 'app'

Step 3: Write the implementation

Create backend/app.py with the full FastAPI application. This is the longest file. Key sections:

• Auth endpoints (/api/auth/login, /api/auth/me, /api/auth/logout)
• Session proxy to amplifierd (/api/sessions/*)
• AG-UI endpoint for CopilotKit (/api/copilotkit) - receives RunAgentInput, streams AG-UI events
• Artifact API (/api/artifacts/*)
• Static file serving for built frontend

"""
Research Workbench - FastAPI application.

Serves:
  /api/health            - health check
  /api/auth/*            - login, logout, me
  /api/sessions/*        - proxy to amplifierd session API
  /api/copilotkit        - AG-UI protocol endpoint for CopilotKit
  /api/artifacts/*       - per-session artifact CRUD
  /api/apps/{name}       - MCP App HTML templates (for sandboxed iframes)
  /                      - built frontend static files
"""

import json
import os
import uuid
from contextlib import asynccontextmanager
from pathlib import Path

import httpx
from fastapi import Cookie, FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles

from agui_adapter import AmplifierdEventTranslator
from artifacts import get_artifact, list_artifacts
from auth import (
    create_session_token,
    invalidate_session_token,
    validate_session_token,
    verify_password,
)

AMPLIFIERD_URL = os.environ.get("AMPLIFIERD_URL", "http://localhost:8410")
AUTH_USER = os.environ.get("AUTH_USER", "admin@localhost")
BUNDLE_NAME = os.environ.get("BUNDLE_NAME", "research-workbench")


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http_client = httpx.AsyncClient(base_url=AMPLIFIERD_URL, timeout=30.0)
    yield
    await app.state.http_client.aclose()


app = FastAPI(title="research-workbench", version="0.1.0", lifespan=lifespan)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
    allow_credentials=True,
)


# --- Auth helpers ---

def _require_auth(session: str | None) -> str:
    email = validate_session_token(session) if session else None
    if email is None:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return email


# --- Health ---

@app.get("/api/health")
async def health():
    return {"status": "ok"}


# --- Auth endpoints ---

@app.post("/api/auth/login")
async def login(request: Request):
    body = await request.json()
    email = body.get("email", "")
    password = body.get("password", "")
    if email != AUTH_USER or not verify_password(password):
        return JSONResponse({"error": "Invalid credentials"}, status_code=401)
    token = create_session_token(email)
    response = JSONResponse({"token": token, "email": email})
    response.set_cookie(key="session", value=token, httponly=True,
                        samesite="lax", max_age=86400 * 7)
    return response


@app.get("/api/auth/me")
async def me(session: str | None = Cookie(default=None)):
    email = validate_session_token(session) if session else None
    if email is None:
        return JSONResponse({"error": "Not authenticated"}, status_code=401)
    return {"email": email}


@app.post("/api/auth/logout")
async def logout(session: str | None = Cookie(default=None)):
    if session:
        invalidate_session_token(session)
    response = JSONResponse({"status": "ok"})
    response.delete_cookie("session")
    return response


# --- Session proxy (to amplifierd) ---

@app.get("/api/sessions")
async def list_sessions(session: str | None = Cookie(default=None)):
    _require_auth(session)
    resp = await app.state.http_client.get("/sessions")
    return JSONResponse(resp.json(), status_code=resp.status_code)


@app.post("/api/sessions")
async def create_session(request: Request, session: str | None = Cookie(default=None)):
    _require_auth(session)
    body = await request.json()
    if "bundle_name" not in body:
        body["bundle_name"] = BUNDLE_NAME
    resp = await app.state.http_client.post("/sessions", json=body)
    return JSONResponse(resp.json(), status_code=resp.status_code)


@app.get("/api/sessions/{session_id}")
async def get_session(session_id: str, session: str | None = Cookie(default=None)):
    _require_auth(session)
    resp = await app.state.http_client.get(f"/sessions/{session_id}")
    return JSONResponse(resp.json(), status_code=resp.status_code)


@app.delete("/api/sessions/{session_id}")
async def delete_session(session_id: str, session: str | None = Cookie(default=None)):
    _require_auth(session)
    resp = await app.state.http_client.delete(f"/sessions/{session_id}")
    return JSONResponse({"status": "deleted"}, status_code=resp.status_code)


@app.get("/api/sessions/{session_id}/transcript")
async def get_transcript(session_id: str, session: str | None = Cookie(default=None)):
    _require_auth(session)
    resp = await app.state.http_client.get(f"/sessions/{session_id}/transcript")
    return JSONResponse(resp.json(), status_code=resp.status_code)


# --- AG-UI endpoint (CopilotKit connects here) ---

@app.post("/api/copilotkit")
async def copilotkit_endpoint(request: Request, session: str | None = Cookie(default=None)):
    """AG-UI protocol endpoint.

    CopilotKit sends RunAgentInput (messages, threadId, runId).
    We map threadId to an amplifierd session, execute the prompt,
    and stream back AG-UI events translated from amplifierd's SSE.
    """
    _require_auth(session)
    body = await request.json()

    thread_id = body.get("threadId", str(uuid.uuid4()))
    run_id = body.get("runId", str(uuid.uuid4()))
    messages = body.get("messages", [])

    # Extract latest user message as prompt
    prompt = ""
    for msg in reversed(messages):
        if msg.get("role") == "user" and msg.get("content"):
            prompt = msg["content"]
            break
    if not prompt:
        return JSONResponse({"error": "No user message found"}, status_code=400)

    # Use threadId as amplifierd session_id
    amp_session_id = thread_id
    client = app.state.http_client

    # Create session if it doesn't exist
    check = await client.get(f"/sessions/{amp_session_id}")
    if check.status_code == 404:
        await client.post("/sessions", json={
            "session_id": amp_session_id,
            "bundle_name": BUNDLE_NAME,
        })

    # Fire execute/stream
    exec_resp = await client.post(
        f"/sessions/{amp_session_id}/execute/stream",
        json={"prompt": prompt},
    )
    if exec_resp.status_code != 202:
        return JSONResponse(
            {"error": "Failed to start execution", "detail": exec_resp.text},
            status_code=502,
        )

    async def event_generator():
        translator = AmplifierdEventTranslator(run_id=run_id, thread_id=thread_id)
        yield _sse({"type": "RUN_STARTED", "threadId": thread_id, "runId": run_id})

        async with httpx.AsyncClient(base_url=AMPLIFIERD_URL, timeout=None) as sse_client:
            async with sse_client.stream(
                "GET", "/events", params={"session": amp_session_id},
            ) as response:
                buffer = ""
                async for chunk in response.aiter_text():
                    buffer += chunk
                    while "\n\n" in buffer:
                        raw_event, buffer = buffer.split("\n\n", 1)
                        parsed = _parse_sse(raw_event)
                        if parsed is None:
                            continue
                        for ag_event in translator.translate(parsed):
                            yield _sse(ag_event)
                        if parsed.get("event_type") == "orchestrator:complete":
                            return

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


def _sse(event: dict) -> str:
    return f"data: {json.dumps(event)}\n\n"


def _parse_sse(raw: str) -> dict | None:
    event_type = None
    data_lines = []
    for line in raw.strip().split("\n"):
        if line.startswith("event:"):
            event_type = line[6:].strip()
        elif line.startswith("data:"):
            data_lines.append(line[5:].strip())
    if not data_lines:
        return None
    try:
        data = json.loads("\n".join(data_lines))
    except json.JSONDecodeError:
        return None
    return {"event_type": event_type or data.get("event_type", ""), "data": data}


# --- Artifacts ---

@app.get("/api/artifacts/{session_id}")
async def api_list_artifacts(session_id: str, session: str | None = Cookie(default=None)):
    _require_auth(session)
    return JSONResponse(list_artifacts(session_id))


@app.get("/api/artifacts/{session_id}/{name}")
async def api_get_artifact(session_id: str, name: str,
                           session: str | None = Cookie(default=None)):
    _require_auth(session)
    content = get_artifact(session_id, name)
    if content is None:
        return JSONResponse({"error": "Not found"}, status_code=404)
    return JSONResponse({"name": name, "session_id": session_id, "content": content})


# --- MCP App templates (served to sandboxed iframes) ---

APPS_DIR = Path(__file__).parent.parent / "bundle" / "apps"


@app.get("/api/apps/{app_name}")
async def get_mcp_app(app_name: str, session: str | None = Cookie(default=None)):
    """Serve MCP App HTML templates for inline interactive visualizations.

    MCP Apps are self-contained HTML files rendered in sandboxed iframes.
    Tool results with _meta.ui.resourceUri reference these apps.
    """
    _require_auth(session)
    # Normalize: allow "chart" or "chart.html"
    if not app_name.endswith(".html"):
        app_name += ".html"
    app_path = APPS_DIR / app_name
    if not app_path.is_file():
        return JSONResponse({"error": f"App '{app_name}' not found"}, status_code=404)
    html = app_path.read_text(encoding="utf-8")
    return HTMLResponse(content=html)


# --- Static files (built frontend) - MUST be last ---

_static_dir = Path(__file__).parent.parent / "frontend" / "dist"
if _static_dir.is_dir():
    app.mount("/", StaticFiles(directory=str(_static_dir), html=True), name="frontend")

Step 4: Run tests to verify they pass

Run:

cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_app.py -v

Expected: All 11 tests PASS. (Session proxy tests that call amplifierd will fail without amplifierd running -- that's expected. The tests above only cover auth, health, and artifact endpoints.)

Step 5: Commit

Run:

cd /home/ken/workspace/research-workbench
git add backend/app.py tests/test_app.py && git commit -m "feat: FastAPI app with auth, AG-UI endpoint, session proxy, artifact API"

---

Task 7: Dockerfile

Files:

• Create: Dockerfile

Step 1: Create the Dockerfile

Create Dockerfile:

FROM python:3.13-slim

# System deps: display, VNC, browser libs, Node.js for playwright-cli
RUN apt-get update -qq && apt-get install -y -qq --no-install-recommends \
    xvfb x11vnc websockify novnc curl \
    libnss3 libatk1.0-0 libatk-bridge2.0-0 libcups2 libdrm2 \
    libxkbcommon0 libxcomposite1 libxdamage1 libxrandr2 libgbm1 \
    libpango-1.0-0 libcairo2 libasound2 libxshmfence1 libgtk-3-0 \
    fonts-liberation && \
    rm -rf /var/lib/apt/lists/*

# Install Node.js (for playwright-cli)
RUN curl -fsSL https://deb.nodesource.com/setup_22.x | bash - && \
    apt-get install -y nodejs && \
    rm -rf /var/lib/apt/lists/*

# Install playwright-cli globally
RUN npm install -g @anthropic-ai/playwright-cli

# Install uv
RUN pip install uv

# Backend deps
COPY backend/pyproject.toml backend/uv.lock /app/backend/
WORKDIR /app/backend
RUN uv sync --frozen

# Install Playwright browser
RUN uv run playwright install chromium

# Install amplifierd
RUN uv pip install amplifierd

# Copy backend source
COPY backend/*.py /app/backend/

# Copy bundle
COPY bundle/ /app/bundle/

# Copy built frontend (build before docker build)
COPY frontend/dist/ /app/frontend/dist/

# Copy entrypoint
COPY entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh

# Create artifacts directory
RUN mkdir -p /app/artifacts

WORKDIR /app

EXPOSE 8080 6080

ENV DISPLAY=:99
ENV AMPLIFIERD_URL=http://localhost:8410
ENV BUNDLE_NAME=research-workbench

ENTRYPOINT ["/entrypoint.sh"]

Step 2: Commit

Run:

cd /home/ken/workspace/research-workbench
git add Dockerfile && git commit -m "feat: Dockerfile with all-in-one container"

---

Task 8: Entrypoint script

Files:

• Create: entrypoint.sh

Step 1: Create the entrypoint

Create entrypoint.sh:

#!/bin/bash
set -e

export DISPLAY=:99

echo "=== Starting Research Workbench ==="

# 1. Virtual display
Xvfb :99 -screen 0 1280x720x24 -ac &
sleep 1
echo "[OK] Xvfb display :99"

# 2. VNC server
x11vnc -display :99 -forever -nopw -listen 0.0.0.0 -rfbport 5900 -shared 2>/dev/null &
sleep 1
echo "[OK] x11vnc on :5900"

# 3. noVNC (web VNC client)
websockify --web=/usr/share/novnc/ 6080 localhost:5900 &
sleep 1
echo "[OK] noVNC on :6080"

# 4. amplifierd (session daemon)
cd /app
amplifierd --port 8410 --bundle /app/bundle/bundle.md &
AMPLIFIERD_PID=$!

# Wait for amplifierd to be ready
echo "Waiting for amplifierd..."
for i in $(seq 1 30); do
    if curl -s http://localhost:8410/health > /dev/null 2>&1; then
        echo "[OK] amplifierd on :8410"
        break
    fi
    sleep 1
done

# 5. FastAPI (web frontend + API)
cd /app/backend
uv run python -m uvicorn app:app --host 0.0.0.0 --port 8080 --log-level info &
sleep 2
echo "[OK] FastAPI on :8080"

echo "============================================="
echo "  Research Workbench: http://localhost:8080"
echo "  Browser VNC:        http://localhost:6080"
echo "============================================="

# Keep container running
if [ $# -eq 0 ]; then
    wait $AMPLIFIERD_PID
else
    exec "$@"
fi

Step 2: Commit

Run:

cd /home/ken/workspace/research-workbench
git add entrypoint.sh && git commit -m "feat: entrypoint.sh starting all services"

---

Phase 1 Complete

At this point you have:

• Clean repo with new directory structure
• Backend with auth, artifacts, AG-UI adapter (with MCP App metadata passthrough), and FastAPI app
• MCP App template serving endpoint (/api/apps/{name})
• Tests for auth, artifacts, AG-UI adapter (including MCP App passthrough), and app integration
• Dockerfile and entrypoint for the all-in-one container
• No frontend yet (that's Phase 2)

Verify everything passes:

Run:

cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/ -v

Expected: All tests pass.