Files

T

Mortdecai 28dc693781 docs: workbench v2 implementation plan — 8 tasks

2026-03-30 07:19:20 -04:00

38 KiB

Raw Permalink Blame History

Workbench Server v2 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Restructure workbench-server into a pip-installable package with server persistence (survives AI CLI restarts), simplified desktop-only layout, and AI-guided setup.

Architecture: The existing monolithic server.py gets split into a proper Python package under src/workbench/. A .server.json file per project tracks running HTTP servers. On MCP startup, the server scans for live servers and reattaches. The scaffold HTML drops the terminal panel for a full-width display surface.

Tech Stack: Python 3.10+, mcp (FastMCP), aiohttp (HTTP/WebSocket), pytest/pytest-asyncio for tests.

Spec: docs/superpowers/specs/2026-03-30-workbench-v2-design.md

Existing code: The current server.py (293 lines) and scaffold.html (142 lines) are the starting point. We're restructuring, not rewriting from scratch — the core logic (MCP tools, HTTP server, WebSocket broadcast) is preserved.

File Structure

workbench-server/
  pyproject.toml                    # Package config, entry point
  README.md                         # Public-facing docs (rewritten)
  INSTALL.md                        # AI-readable setup instructions
  LICENSE                           # MIT
  src/
    workbench/
      __init__.py                   # Version string
      __main__.py                   # python -m workbench
      cli.py                        # CLI: mcp, serve, list, help
      server.py                     # MCP tools + HTTP/WS server management
      project.py                    # Project dir management, logging, persistence
      scaffold.html                 # HTML template (moved from root, simplified)
  tests/
    conftest.py                     # Shared fixtures
    test_project.py                 # Project creation, logging, persistence
    test_server.py                  # MCP tool tests

# Old files to remove:
  server.py                         # replaced by src/workbench/server.py
  scaffold.html                     # replaced by src/workbench/scaffold.html
  requirements.txt                  # replaced by pyproject.toml

Task 1: Package Scaffold + Move Files

Files:

Create: pyproject.toml
Create: LICENSE
Create: src/workbench/__init__.py
Create: src/workbench/__main__.py
Remove: requirements.txt
Move: server.py → will be replaced in later tasks
Move: scaffold.html → will be replaced in later tasks
Step 1: Create directory structure

mkdir -p ~/bin/workbench-server/src/workbench
mkdir -p ~/bin/workbench-server/tests

Step 2: Write pyproject.toml

[build-system]
requires = ["setuptools>=68.0"]
build-backend = "setuptools.build_meta"

[project]
name = "workbench-server"
version = "0.2.0"
description = "MCP server that lets AI CLIs build interactive web pages served over LAN"
readme = "README.md"
license = "MIT"
requires-python = ">=3.10"
dependencies = [
    "mcp>=1.26.0",
    "aiohttp>=3.9.0",
]

[project.scripts]
workbench = "workbench.cli:main"

Step 3: Write __init__.py

"""Workbench — MCP server for AI-driven interactive web pages."""

__version__ = "0.2.0"

Step 4: Write __main__.py

"""Allow running as `python -m workbench`."""

from workbench.cli import main

main()

Step 5: Write LICENSE (MIT)
Step 6: Back up old files, remove requirements.txt

mkdir -p ~/bin/workbench-server/.backup
cp ~/bin/workbench-server/server.py ~/bin/workbench-server/.backup/server.py.v1
cp ~/bin/workbench-server/scaffold.html ~/bin/workbench-server/.backup/scaffold.html.v1
rm ~/bin/workbench-server/requirements.txt

Step 7: Commit

cd ~/bin/workbench-server
git add pyproject.toml LICENSE src/ .backup/
git rm requirements.txt
git commit -m "feat: package scaffold — pyproject.toml, src/workbench/ layout"

Task 2: Project Management Module

Files:

Create: src/workbench/project.py
Create: tests/conftest.py
Create: tests/test_project.py

Handles project directory creation, session logging, log reading, listing, and the new .server.json persistence.

Step 1: Write test file

# tests/test_project.py
import json
from workbench.project import (
    create_project, project_exists, project_path,
    append_log, read_log, list_projects, log_session_event,
    write_server_info, read_server_info, clear_server_info,
    WORKBENCH_DIR,
)


def test_create_project(tmp_workbench):
    path = create_project("io102", "Heathkit IO-102", workbench_dir=tmp_workbench)
    assert path.exists()
    assert (path / "index.html").exists()
    assert (path / "session.md").exists()
    assert (path / "session.jsonl").exists()
    assert (path / "cost-log.jsonl").exists()
    assert (path / "state.json").exists()
    assert (path / "assets").is_dir()
    md = (path / "session.md").read_text()
    assert "Heathkit IO-102" in md


def test_create_project_idempotent(tmp_workbench):
    create_project("io102", "First", workbench_dir=tmp_workbench)
    append_log("io102", "test entry", workbench_dir=tmp_workbench)
    create_project("io102", "Second", workbench_dir=tmp_workbench)
    entries = read_log("io102", workbench_dir=tmp_workbench)
    assert len(entries) == 1


def test_project_exists(tmp_workbench):
    assert not project_exists("io102", workbench_dir=tmp_workbench)
    create_project("io102", "Test", workbench_dir=tmp_workbench)
    assert project_exists("io102", workbench_dir=tmp_workbench)


def test_append_and_read_log(tmp_workbench):
    create_project("io102", "Test", workbench_dir=tmp_workbench)
    append_log("io102", "R412 measured 1.05M", data={"ohms": 1050000}, workbench_dir=tmp_workbench)
    append_log("io102", "R413 OK", workbench_dir=tmp_workbench)
    entries = read_log("io102", tail=10, workbench_dir=tmp_workbench)
    assert len(entries) == 2
    assert entries[0]["entry"] == "R412 measured 1.05M"
    assert entries[0]["ohms"] == 1050000


def test_read_log_tail(tmp_workbench):
    create_project("io102", "Test", workbench_dir=tmp_workbench)
    for i in range(30):
        append_log("io102", f"Entry {i}", workbench_dir=tmp_workbench)
    entries = read_log("io102", tail=5, workbench_dir=tmp_workbench)
    assert len(entries) == 5
    assert entries[0]["entry"] == "Entry 25"


def test_list_projects(tmp_workbench):
    assert list_projects(workbench_dir=tmp_workbench) == []
    create_project("io102", "First", workbench_dir=tmp_workbench)
    create_project("psu-rebuild", "Second", workbench_dir=tmp_workbench)
    projects = list_projects(workbench_dir=tmp_workbench)
    assert [p["name"] for p in projects] == ["io102", "psu-rebuild"]


def test_log_session_event(tmp_workbench):
    create_project("io102", "Test", workbench_dir=tmp_workbench)
    log_session_event("io102", "session_start", workbench_dir=tmp_workbench)
    cost_log = (tmp_workbench / "io102" / "cost-log.jsonl").read_text().strip()
    entry = json.loads(cost_log.split("\n")[-1])
    assert entry["event"] == "session_start"


def test_server_info_write_read_clear(tmp_workbench):
    create_project("io102", "Test", workbench_dir=tmp_workbench)
    write_server_info("io102", pid=12345, port=8070, workbench_dir=tmp_workbench)
    info = read_server_info("io102", workbench_dir=tmp_workbench)
    assert info is not None
    assert info["pid"] == 12345
    assert info["port"] == 8070
    assert "started" in info

    clear_server_info("io102", workbench_dir=tmp_workbench)
    assert read_server_info("io102", workbench_dir=tmp_workbench) is None

Step 2: Write conftest.py

# tests/conftest.py
import pytest


@pytest.fixture
def tmp_workbench(tmp_path):
    """Provide a temporary workbench directory."""
    wb = tmp_path / "workbench"
    wb.mkdir()
    return wb

Step 3: Run tests to verify they fail

cd ~/bin/workbench-server
pip install --break-system-packages -e .
pip install --break-system-packages pytest pytest-asyncio
pytest tests/test_project.py -v

Expected: ImportError.

Step 4: Write project.py

"""Project directory management — create, log, read, list, server persistence."""

from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

WORKBENCH_DIR = Path.home() / "workbench"
SCAFFOLD_HTML = Path(__file__).parent / "scaffold.html"


def _now_iso() -> str:
    return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


def project_path(name: str, workbench_dir: Path = WORKBENCH_DIR) -> Path:
    return workbench_dir / name


def project_exists(name: str, workbench_dir: Path = WORKBENCH_DIR) -> bool:
    return project_path(name, workbench_dir).is_dir()


def create_project(
    name: str,
    title: str,
    description: str = "",
    workbench_dir: Path = WORKBENCH_DIR,
) -> Path:
    """Create a project directory with scaffold HTML and log files. Idempotent."""
    pdir = project_path(name, workbench_dir)
    pdir.mkdir(parents=True, exist_ok=True)
    (pdir / "assets").mkdir(exist_ok=True)

    # Write scaffold HTML (always overwrite — it's a template, not user content)
    if SCAFFOLD_HTML.exists():
        template = SCAFFOLD_HTML.read_text()
        html = template.replace("{{TITLE}}", title)
        html = html.replace("{{DESCRIPTION}}", description)
        (pdir / "index.html").write_text(html)

    # Only write log files if they don't exist (preserve existing session data)
    md_path = pdir / "session.md"
    if not md_path.exists():
        md_path.write_text(f"# {title} — Session Log\n\nStarted: {_now_iso()}\n\n")

    jsonl_path = pdir / "session.jsonl"
    if not jsonl_path.exists():
        jsonl_path.write_text("")

    cost_path = pdir / "cost-log.jsonl"
    if not cost_path.exists():
        cost_path.write_text("")

    state_path = pdir / "state.json"
    if not state_path.exists():
        state_path.write_text("{}")

    return pdir


def append_log(
    name: str,
    entry: str,
    data: Optional[dict] = None,
    workbench_dir: Path = WORKBENCH_DIR,
) -> dict:
    """Append to session.md and session.jsonl."""
    pdir = project_path(name, workbench_dir)
    ts = _now_iso()

    with open(pdir / "session.md", "a") as f:
        f.write(f"\n### {ts}\n{entry}\n")

    obj = {"ts": ts, "entry": entry}
    if data:
        obj.update(data)
    with open(pdir / "session.jsonl", "a") as f:
        f.write(json.dumps(obj) + "\n")

    return obj


def read_log(
    name: str, tail: int = 20, workbench_dir: Path = WORKBENCH_DIR
) -> list[dict]:
    """Read recent log entries from session.jsonl."""
    jsonl_path = project_path(name, workbench_dir) / "session.jsonl"
    if not jsonl_path.exists():
        return []

    lines = jsonl_path.read_text().strip().split("\n")
    lines = [l for l in lines if l.strip()]
    recent = lines[-tail:] if len(lines) > tail else lines

    entries = []
    for line in recent:
        try:
            entries.append(json.loads(line))
        except json.JSONDecodeError:
            entries.append({"raw": line})
    return entries


def list_projects(workbench_dir: Path = WORKBENCH_DIR) -> list[dict]:
    """List all project directories."""
    if not workbench_dir.exists():
        return []
    projects = []
    for d in sorted(workbench_dir.iterdir()):
        if d.is_dir() and not d.name.startswith("."):
            projects.append({"name": d.name})
    return projects


def log_session_event(
    name: str, event: str, workbench_dir: Path = WORKBENCH_DIR, **extra
) -> None:
    """Append to cost-log.jsonl."""
    pdir = project_path(name, workbench_dir)
    obj = {"ts": _now_iso(), "event": event, "project": name}
    obj.update(extra)
    with open(pdir / "cost-log.jsonl", "a") as f:
        f.write(json.dumps(obj) + "\n")


# --- Server persistence ---

def write_server_info(
    name: str, pid: int, port: int, workbench_dir: Path = WORKBENCH_DIR
) -> None:
    """Write .server.json with running server info."""
    pdir = project_path(name, workbench_dir)
    info = {"pid": pid, "port": port, "started": _now_iso()}
    (pdir / ".server.json").write_text(json.dumps(info))


def read_server_info(
    name: str, workbench_dir: Path = WORKBENCH_DIR
) -> Optional[dict]:
    """Read .server.json. Returns None if not found."""
    server_file = project_path(name, workbench_dir) / ".server.json"
    if not server_file.exists():
        return None
    try:
        return json.loads(server_file.read_text())
    except (json.JSONDecodeError, OSError):
        return None


def clear_server_info(
    name: str, workbench_dir: Path = WORKBENCH_DIR
) -> None:
    """Delete .server.json."""
    server_file = project_path(name, workbench_dir) / ".server.json"
    if server_file.exists():
        server_file.unlink()

Step 5: Copy scaffold.html to src/workbench/

The scaffold HTML needs to exist at src/workbench/scaffold.html for create_project to find it. We'll write the v2 version in Task 3. For now, create a minimal placeholder so tests pass:

<!DOCTYPE html>
<html><head><title>{{TITLE}}</title></head>
<body><h1>{{TITLE}}</h1><p>{{DESCRIPTION}}</p></body></html>

Step 6: Run tests to verify they pass

pytest tests/test_project.py -v

Expected: All 9 tests pass.

Step 7: Commit

git add src/workbench/project.py src/workbench/scaffold.html tests/conftest.py tests/test_project.py
git commit -m "feat: project management module — create, log, read, list, server persistence"

Task 3: Scaffold HTML v2

Files:

Create: src/workbench/scaffold.html (overwrite placeholder from Task 2)
Step 1: Write the v2 scaffold HTML

Full-width desktop layout. No split pane, no iframe, no terminal embed. Same WebSocket reconnect and state handling as v1.

<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{{TITLE}}</title>
<style>
  :root {
    --bg: #0a0f0c;
    --panel-bg: #111a15;
    --border: #2a3a2e;
    --text: #aaccaa;
    --text-dim: #557755;
    --accent: #33ff66;
  }
  * { margin: 0; padding: 0; box-sizing: border-box; }
  body {
    background: var(--bg);
    color: var(--text);
    font-family: monospace;
    font-size: 14px;
    height: 100vh;
    display: flex;
    flex-direction: column;
  }
  #content {
    flex: 1;
    overflow-y: auto;
    padding: 16px;
  }
  #log-feed {
    border-top: 1px solid var(--border);
    padding: 8px 12px;
    max-height: 200px;
    overflow-y: auto;
  }
  #log-feed h3 {
    color: var(--text-dim);
    font-size: 11px;
    letter-spacing: 2px;
    text-transform: uppercase;
    margin-bottom: 4px;
  }
  .log-entry {
    font-size: 12px;
    color: var(--text-dim);
    padding: 2px 0;
  }
  .log-entry .log-time {
    color: var(--accent);
    margin-right: 8px;
  }
  #status {
    background: var(--panel-bg);
    border-top: 1px solid var(--border);
    padding: 4px 12px;
    font-size: 11px;
    color: var(--text-dim);
  }
  #status.connected { color: var(--accent); }
</style>
</head>
<body>
  <div id="content">
    <h1 style="color: var(--accent); font-size: 18px;">{{TITLE}}</h1>
    <p style="color: var(--text-dim); margin-top: 8px;">{{DESCRIPTION}}</p>
    <p style="color: var(--text-dim); margin-top: 16px;">Waiting for content...</p>
  </div>
  <div id="log-feed">
    <h3>Session Log</h3>
    <div id="log-entries"></div>
  </div>
  <div id="status">Connecting...</div>

<script>
const WS_URL = `ws://${location.host}/ws`;
let ws;

function connect() {
  ws = new WebSocket(WS_URL);
  ws.onopen = () => {
    document.getElementById('status').textContent = 'Connected';
    document.getElementById('status').className = 'connected';
  };
  ws.onclose = () => {
    document.getElementById('status').textContent = 'Disconnected — reconnecting...';
    document.getElementById('status').className = '';
    setTimeout(connect, 2000);
  };
  ws.onmessage = (e) => {
    const msg = JSON.parse(e.data);
    if (msg.type === 'state') handleState(msg.state);
    if (msg.type === 'log') handleLog(msg.entry);
  };
}

function handleState(state) {
  if (state.template) {
    document.getElementById('content').innerHTML = state.template;
  }
  if (state.styles) {
    let el = document.getElementById('dynamic-styles');
    if (!el) { el = document.createElement('style'); el.id = 'dynamic-styles'; document.head.appendChild(el); }
    el.textContent = state.styles;
  }
  if (state.script) {
    try { new Function(state.script)(); } catch(e) { console.error('Script error:', e); }
  }
  window.__workbench_state = state;
  try { localStorage.setItem('workbench-state', JSON.stringify(state)); } catch(e) {}
}

function handleLog(entry) {
  const div = document.createElement('div');
  div.className = 'log-entry';
  const time = new Date().toLocaleTimeString();
  div.innerHTML = `<span class="log-time">${time}</span>${entry}`;
  const feed = document.getElementById('log-entries');
  feed.appendChild(div);
  feed.scrollTop = feed.scrollHeight;
}

try {
  const saved = JSON.parse(localStorage.getItem('workbench-state'));
  if (saved) handleState(saved);
} catch(e) {}

connect();
</script>
</body>
</html>

Step 2: Commit

git add src/workbench/scaffold.html
git commit -m "feat: scaffold HTML v2 — full-width desktop layout, no terminal panel"

Task 4: MCP Server with Persistence

Files:

Create: src/workbench/server.py
Create: tests/test_server.py

The core MCP server — same 6 tools as v1, but with persistence logic. Adapted from the original server.py with the .server.json check-and-reattach flow.

Step 1: Write test file

# tests/test_server.py
import asyncio
import json
import os
from pathlib import Path
from unittest.mock import patch, AsyncMock, MagicMock

import pytest

from workbench.server import WorkbenchServer


@pytest.fixture
def server(tmp_workbench):
    return WorkbenchServer(workbench_dir=tmp_workbench)


@pytest.mark.asyncio
async def test_scaffold_creates_project(server, tmp_workbench):
    with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
        result = await server.workbench_scaffold("test-proj", "Test Project")
    data = json.loads(result)
    assert "url" in data
    assert (tmp_workbench / "test-proj" / "index.html").exists()


@pytest.mark.asyncio
async def test_scaffold_reattaches_to_running_server(server, tmp_workbench):
    """If .server.json exists and server is alive, don't start a new one."""
    from workbench.project import create_project, write_server_info
    create_project("test-proj", "Test", workbench_dir=tmp_workbench)
    write_server_info("test-proj", pid=os.getpid(), port=9999, workbench_dir=tmp_workbench)

    with patch.object(server, "_is_server_alive", return_value=True):
        with patch.object(server, "_start_http_server", new_callable=AsyncMock) as mock_start:
            result = await server.workbench_scaffold("test-proj", "Test")
            mock_start.assert_not_called()  # Should NOT start a new server
    data = json.loads(result)
    assert "9999" in data["url"]


@pytest.mark.asyncio
async def test_scaffold_replaces_dead_server(server, tmp_workbench):
    """If .server.json exists but server is dead, start a new one."""
    from workbench.project import create_project, write_server_info
    create_project("test-proj", "Test", workbench_dir=tmp_workbench)
    write_server_info("test-proj", pid=99999, port=9999, workbench_dir=tmp_workbench)

    with patch.object(server, "_is_server_alive", return_value=False):
        with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
            result = await server.workbench_scaffold("test-proj", "Test")
    data = json.loads(result)
    assert "8070" in data["url"]


@pytest.mark.asyncio
async def test_list_empty(server):
    result = await server.workbench_list()
    data = json.loads(result)
    assert data["projects"] == []


@pytest.mark.asyncio
async def test_log_writes_to_disk(server, tmp_workbench):
    with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
        await server.workbench_scaffold("test-proj", "Test")
    result = await server.workbench_log("test-proj", "R412 measured 1.05M")
    data = json.loads(result)
    assert data["ok"] is True
    jsonl = (tmp_workbench / "test-proj" / "session.jsonl").read_text().strip()
    assert "R412 measured 1.05M" in jsonl


@pytest.mark.asyncio
async def test_state_saves_to_disk(server, tmp_workbench):
    with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
        await server.workbench_scaffold("test-proj", "Test")
    state = json.dumps({"template": "<h1>Hello</h1>"})
    result = await server.workbench_state("test-proj", state)
    data = json.loads(result)
    assert data["ok"] is True
    saved = json.loads((tmp_workbench / "test-proj" / "state.json").read_text())
    assert saved["template"] == "<h1>Hello</h1>"


@pytest.mark.asyncio
async def test_stop_cleans_up(server, tmp_workbench):
    with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
        await server.workbench_scaffold("test-proj", "Test")
    server._runners["test-proj"] = AsyncMock()
    result = await server.workbench_stop("test-proj")
    data = json.loads(result)
    assert data["ok"] is True
    assert not (tmp_workbench / "test-proj" / ".server.json").exists()


@pytest.mark.asyncio
async def test_reconnect_on_startup(tmp_workbench):
    """On init, server should find live servers from .server.json files."""
    from workbench.project import create_project, write_server_info
    create_project("live-proj", "Live", workbench_dir=tmp_workbench)
    write_server_info("live-proj", pid=os.getpid(), port=8070, workbench_dir=tmp_workbench)
    create_project("dead-proj", "Dead", workbench_dir=tmp_workbench)
    write_server_info("dead-proj", pid=99999, port=8071, workbench_dir=tmp_workbench)

    srv = WorkbenchServer(workbench_dir=tmp_workbench)
    with patch.object(srv, "_is_server_alive", side_effect=lambda port: port == 8070):
        await srv.reconnect_existing_servers()

    assert "live-proj" in srv._active
    assert srv._active["live-proj"]["port"] == 8070
    assert "dead-proj" not in srv._active
    # Dead server's .server.json should be cleaned up
    assert not (tmp_workbench / "dead-proj" / ".server.json").exists()

Step 2: Run tests to verify they fail

pytest tests/test_server.py -v

Expected: ImportError.

Step 3: Write server.py

"""MCP server — 6 workbench tools with HTTP/WS server management and persistence."""

from __future__ import annotations

import asyncio
import json
import os
import socket
from pathlib import Path
from typing import Optional

from aiohttp import web
from mcp.server.fastmcp import FastMCP

from workbench.project import (
    create_project, project_exists, project_path,
    append_log, read_log, list_projects, log_session_event,
    write_server_info, read_server_info, clear_server_info,
    WORKBENCH_DIR,
)


def _get_lan_ip() -> str:
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("192.168.0.1", 80))
        ip = s.getsockname()[0]
        s.close()
        return ip
    except Exception:
        return "127.0.0.1"


def _find_free_port(start: int = 8070) -> int:
    for port in range(start, start + 100):
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.bind(("0.0.0.0", port))
            s.close()
            return port
        except OSError:
            continue
    raise RuntimeError(f"No free port found in range {start}-{start + 99}")


class WorkbenchServer:
    """Core server logic — testable without MCP transport."""

    def __init__(self, workbench_dir: Path = WORKBENCH_DIR):
        self.workbench_dir = Path(workbench_dir)
        self._active: dict[str, dict] = {}  # name -> {"port": int, "runner": AppRunner, "ws_clients": set}
        self._runners: dict[str, web.AppRunner] = {}

    def _is_server_alive(self, port: int) -> bool:
        """Check if an HTTP server is responding on the given port."""
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(1)
            s.connect(("127.0.0.1", port))
            s.close()
            return True
        except (OSError, ConnectionRefusedError):
            return False

    async def reconnect_existing_servers(self) -> None:
        """Scan for running servers from .server.json files. Called on MCP startup."""
        if not self.workbench_dir.exists():
            return
        for d in self.workbench_dir.iterdir():
            if not d.is_dir():
                continue
            name = d.name
            info = read_server_info(name, workbench_dir=self.workbench_dir)
            if info is None:
                continue
            port = info["port"]
            if self._is_server_alive(port):
                self._active[name] = {"port": port, "ws_clients": set()}
            else:
                clear_server_info(name, workbench_dir=self.workbench_dir)

    async def _start_http_server(self, name: str) -> int:
        """Start an HTTP + WebSocket server for a project. Returns port."""
        port = _find_free_port()
        pdir = project_path(name, self.workbench_dir)

        app = web.Application()
        app["project_name"] = name
        app["workbench_server"] = self

        async def ws_handler(request):
            ws = web.WebSocketResponse()
            await ws.prepare(request)
            proj = request.app["project_name"]
            if proj in self._active:
                self._active[proj]["ws_clients"].add(ws)
            try:
                async for msg in ws:
                    pass
            finally:
                if proj in self._active:
                    self._active[proj]["ws_clients"].discard(ws)
            return ws

        async def static_handler(request):
            proj = request.app["project_name"]
            path = request.match_info.get("path", "index.html") or "index.html"
            file_path = project_path(proj, self.workbench_dir) / path
            if not file_path.exists():
                return web.Response(status=404, text="Not found")
            return web.FileResponse(file_path)

        app.router.add_get("/ws", ws_handler)
        app.router.add_get("/{path:.*}", static_handler)
        app.router.add_get("/", static_handler)

        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, "0.0.0.0", port)
        await site.start()

        self._runners[name] = runner
        self._active[name] = {"port": port, "ws_clients": set()}
        write_server_info(name, pid=os.getpid(), port=port, workbench_dir=self.workbench_dir)

        return port

    async def _broadcast_ws(self, name: str, message: dict) -> None:
        if name not in self._active:
            return
        clients = self._active[name]["ws_clients"]
        dead = set()
        for ws in clients:
            try:
                await ws.send_json(message)
            except Exception:
                dead.add(ws)
        clients -= dead

    # --- MCP Tool implementations ---

    async def workbench_scaffold(self, name: str, title: str, description: str = "") -> str:
        pdir = project_path(name, self.workbench_dir)
        create_project(name, title, description, workbench_dir=self.workbench_dir)

        # Check for existing running server
        info = read_server_info(name, workbench_dir=self.workbench_dir)
        if info and self._is_server_alive(info["port"]):
            port = info["port"]
            if name not in self._active:
                self._active[name] = {"port": port, "ws_clients": set()}
        else:
            if info:
                clear_server_info(name, workbench_dir=self.workbench_dir)
            port = await self._start_http_server(name)

        ip = _get_lan_ip()
        log_session_event(name, "session_start", workbench_dir=self.workbench_dir)
        return json.dumps({"path": str(pdir), "url": f"http://{ip}:{port}"})

    async def workbench_state(self, project: str, state: str) -> str:
        if not project_exists(project, workbench_dir=self.workbench_dir):
            return json.dumps({"error": f"Project '{project}' not found. Run workbench_scaffold first."})

        state_obj = json.loads(state)
        pdir = project_path(project, self.workbench_dir)
        (pdir / "state.json").write_text(json.dumps(state_obj, indent=2))

        await self._broadcast_ws(project, {"type": "state", "state": state_obj})
        return json.dumps({"ok": True})

    async def workbench_log(self, project: str, entry: str, data: str = "{}") -> str:
        if not project_exists(project, workbench_dir=self.workbench_dir):
            return json.dumps({"error": f"Project '{project}' not found."})

        data_obj = json.loads(data) if data and data != "{}" else None
        append_log(project, entry, data=data_obj, workbench_dir=self.workbench_dir)
        await self._broadcast_ws(project, {"type": "log", "entry": entry})
        return json.dumps({"ok": True})

    async def workbench_read_log(self, project: str, tail: int = 20) -> str:
        if not project_exists(project, workbench_dir=self.workbench_dir):
            return json.dumps({"error": f"Project '{project}' not found."})
        entries = read_log(project, tail=tail, workbench_dir=self.workbench_dir)
        return json.dumps({"entries": entries})

    async def workbench_list(self) -> str:
        projects = list_projects(workbench_dir=self.workbench_dir)
        for p in projects:
            p["active"] = p["name"] in self._active
            if p["active"]:
                ip = _get_lan_ip()
                port = self._active[p["name"]]["port"]
                p["url"] = f"http://{ip}:{port}"
        return json.dumps({"projects": projects})

    async def workbench_stop(self, project: str) -> str:
        if project not in self._active:
            return json.dumps({"error": f"Project '{project}' is not running."})

        # Log session end
        if project_exists(project, workbench_dir=self.workbench_dir):
            entries = read_log(project, tail=999999, workbench_dir=self.workbench_dir)
            log_session_event(
                project, "session_end",
                workbench_dir=self.workbench_dir,
                log_entries=len(entries),
            )

        # Shutdown HTTP server
        if project in self._runners:
            await self._runners[project].cleanup()
            del self._runners[project]

        clear_server_info(project, workbench_dir=self.workbench_dir)
        del self._active[project]
        return json.dumps({"ok": True})


def create_mcp_server(workbench_dir: Path = WORKBENCH_DIR) -> FastMCP:
    """Create the FastMCP instance with all workbench tools registered."""
    srv = WorkbenchServer(workbench_dir=workbench_dir)

    mcp = FastMCP(
        "workbench",
        instructions="Workbench — build interactive web pages served over LAN. Call workbench_scaffold first.",
    )

    @mcp.tool()
    async def workbench_scaffold(name: str, title: str, description: str = "") -> str:
        """Create a workbench project and start the HTTP server. Returns the LAN URL to open in a browser. If the project already exists and its server is running, reattaches without starting a duplicate. Always safe to call — never creates duplicates."""
        return await srv.workbench_scaffold(name, title, description)

    @mcp.tool()
    async def workbench_state(project: str, state: str) -> str:
        """Push a state update to the browser via WebSocket. The state is a JSON string — include 'template' (HTML string) to replace the page content, 'styles' (CSS string) to inject styles, and 'script' (JS string) to execute code. The AI has full control over the page."""
        return await srv.workbench_state(project, state)

    @mcp.tool()
    async def workbench_log(project: str, entry: str, data: str = "{}") -> str:
        """Append a log entry to the session log. Shows in the browser log feed. entry: human-readable markdown string. data: optional JSON for the machine-readable log."""
        return await srv.workbench_log(project, entry, data)

    @mcp.tool()
    async def workbench_read_log(project: str, tail: int = 20) -> str:
        """Read recent session log entries so the AI can resume a previous session."""
        return await srv.workbench_read_log(project, tail)

    @mcp.tool()
    async def workbench_list() -> str:
        """List all workbench projects and whether their HTTP server is running."""
        return await srv.workbench_list()

    @mcp.tool()
    async def workbench_stop(project: str) -> str:
        """Stop the HTTP server for a project and end the session."""
        return await srv.workbench_stop(project)

    return mcp

Step 4: Run tests

pytest tests/test_server.py -v

Expected: All 9 tests pass.

Step 5: Run full suite

pytest tests/ -v

Expected: All 18 tests pass (9 project + 9 server).

Step 6: Commit

git add src/workbench/server.py tests/test_server.py
git commit -m "feat: MCP server with persistence — reattach to running servers on restart"

Task 5: CLI Entry Point

Files:

Create: src/workbench/cli.py
Step 1: Write cli.py

"""CLI entry point for workbench."""

from __future__ import annotations

import argparse
import asyncio
import sys

from workbench import __version__


def cmd_mcp(args):
    """Start the MCP server on stdio."""
    from workbench.server import create_mcp_server
    mcp = create_mcp_server()
    mcp.run(transport="stdio")


def cmd_serve(args):
    """Serve a project without MCP."""
    from workbench.server import WorkbenchServer

    srv = WorkbenchServer()

    async def run():
        result = await srv.workbench_scaffold(args.name, args.name)
        import json
        data = json.loads(result)
        print(f"Serving: {data['url']}")
        try:
            while True:
                await asyncio.sleep(1)
        except (KeyboardInterrupt, asyncio.CancelledError):
            await srv.workbench_stop(args.name)

    asyncio.run(run())


def cmd_list(args):
    """List all projects."""
    from workbench.project import list_projects, read_server_info
    projects = list_projects()
    if not projects:
        print("No projects found in ~/workbench/")
        return
    for p in projects:
        info = read_server_info(p["name"])
        status = f" (running on port {info['port']})" if info else ""
        print(f"  {p['name']}{status}")


def main():
    parser = argparse.ArgumentParser(
        prog="workbench",
        description="MCP server for AI-driven interactive web pages",
    )
    parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
    sub = parser.add_subparsers(dest="command")

    sub.add_parser("mcp", help="Start MCP server (stdio transport)")

    serve_parser = sub.add_parser("serve", help="Serve a project (standalone)")
    serve_parser.add_argument("name", help="Project name")

    sub.add_parser("list", help="List all projects")
    sub.add_parser("help", help="Show help")

    args = parser.parse_args()

    commands = {
        "mcp": cmd_mcp,
        "serve": cmd_serve,
        "list": cmd_list,
        "help": lambda a: parser.print_help(),
    }

    if args.command is None:
        parser.print_help()
        sys.exit(0)

    commands[args.command](args)

Step 2: Install and verify

cd ~/bin/workbench-server
pip install --break-system-packages -e .
workbench --version
workbench help
workbench list

Expected: version "workbench 0.2.0", help shows subcommands, list says "No projects found".

Step 3: Commit

git add src/workbench/cli.py
git commit -m "feat: CLI entry point — mcp, serve, list, help"

Task 6: Clean Up Old Files

Files:

Remove: server.py (root level — replaced by src/workbench/server.py)
Remove: scaffold.html (root level — replaced by src/workbench/scaffold.html)
Step 1: Remove old files

cd ~/bin/workbench-server
git rm server.py scaffold.html

Step 2: Run full test suite to confirm nothing breaks

pytest tests/ -v

Expected: All 18 tests pass.

Step 3: Commit

git commit -m "refactor: remove old root-level server.py and scaffold.html"

Task 7: Docs — README, INSTALL.md, LICENSE

Files:

Rewrite: README.md
Create: INSTALL.md
Step 1: Rewrite README.md

Public-facing. Desktop browser as display surface. Setup is "paste URL" or "read INSTALL.md". No mention of kitty or terminal splitting. Adapted from the kitty-workbench README pattern but for browser-based display.

Key sections: what it does (with ASCII art diagram), setup (clone + "tell AI to read INSTALL.md"), tools table, usage examples, project structure, requirements, FAQ.

Step 2: Write INSTALL.md

AI-readable setup instructions. Simpler than kitty-workbench (no terminal detection needed):

Clone + pip install
Detect environment: Python version, platform, browser availability
Detect SSH usage: $SSH_CONNECTION, ~/.ssh/known_hosts, ControlMaster config
Present options: SSH ControlMaster if applicable
Configure MCP for user's AI CLI (detect Claude Code, Gemini CLI, etc.)
Smoke test: workbench list
Write ~/workbench/START.md

Include SSH ControlMaster setup (same as kitty-workbench INSTALL.md — ControlMaster auto, ControlPersist yes, sockets dir).

Step 3: Commit

git add README.md INSTALL.md
git commit -m "docs: README and INSTALL.md — AI-guided setup"

Task 8: Push to Gitea

Step 1: Run full test suite

cd ~/bin/workbench-server
pytest tests/ -v

Expected: All 18 tests pass.

Step 2: Push

cd ~/bin/workbench-server
gitea push

Summary

Task	What	Key files
1	Package scaffold + move files	pyproject.toml, __init__.py, __main__.py, LICENSE
2	Project management module	project.py, test_project.py
3	Scaffold HTML v2	scaffold.html
4	MCP server with persistence	server.py, test_server.py
5	CLI entry point	cli.py
6	Clean up old files	remove server.py, scaffold.html from root
7	Docs	README.md, INSTALL.md
8	Push to Gitea	—

38 KiB Raw Permalink Blame History

Workbench Server v2 Implementation Plan

File Structure

Task 1: Package Scaffold + Move Files

Task 2: Project Management Module

Task 3: Scaffold HTML v2

Task 4: MCP Server with Persistence

Task 5: CLI Entry Point

Task 6: Clean Up Old Files

Task 7: Docs — README, INSTALL.md, LICENSE

Task 8: Push to Gitea

Summary

38 KiB

Raw Permalink Blame History