Streaming helm upgrade to a browser

By late April the platform could provision a real Windows machine on demand, clone it from a golden image, deploy an agent against it and record the result. What it could not do was tell anyone what was going on.

Click “create computer,” get a spinner. Three minutes later, either a machine or an error. If it failed, the only diagnosis available was kubectl.

That is fine for a tool you built. It is not fine for a platform other people use, and the difference between the two is almost entirely feedback.

The shape

React dashboard ──REST──▶ FastAPI API ──▶ MongoDB
       │                       │
       │                       └──▶ helm / kubernetes ──▶ AKS
       │
       └──WebSocket──▶ ws-broker ◀──WebSocket── test-job (K8s Job)
                                 ◀──WebSocket── benchmark-job (K8s Job)

FastAPI, MongoDB, React with Fluent UI, and a small standalone WebSocket broker that turns out to be the interesting piece.

Provisioning as an async generator

The core move: deployment is not a function that returns, it is a generator that yields progress.

async def deploy_computer(computer_name, os_type, hdd_snapshot_name, cpu, ram, disk,
                          benchmark_instance_id, test_instance_id) -> AsyncGenerator[WebsocketMessage, None]:

    yield WebsocketMessage(action="status", payload={
        "type": "step", "status": "inProgress", "message": "Creating a computer..." })

    with tempfile.TemporaryDirectory() as tmpdir:
        yield WebsocketMessage(action="status", payload={
            "type": "step", "status": "inProgress", "message": "Cloning Helm repo..." })
        Repo.clone_from(HELM_REPO_URL, tmpdir)
        yield WebsocketMessage(action="status", payload={
            "type": "step", "status": "done", "message": "Cloning Helm repo..." })
        ...

And Helm’s own stdout is relayed line by line as it is produced:

process = await asyncio.create_subprocess_exec(
    *helm_cmd,
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.STDOUT,
)

while True:
    line = await process.stdout.readline()
    if not line:
        break
    if line.startswith(b"Error: context deadline exceeded"):
        continue
    yield WebsocketMessage(action="status", payload={
        "type": "substep", "status": "info", "message": line.decode().strip() })

So the user watching the dashboard sees what an operator would see in a terminal.

Three message types, which is the whole vocabulary:

• step — a named phase with inProgress / done / success / warning.
• substep — raw output under the current step.
• timer — elapsed seconds, emitted once per second during the wait:

while time.time() < end_time:
    elapsed = int(time.time() - start_time)
    if elapsed != last_timer_sent:
        last_timer_sent = elapsed
        yield WebsocketMessage(action="status", payload={
            "type": "substep", "status": "timer", "message": f"Waiting {elapsed} seconds..." })

    pod_list = core.list_namespaced_pod(namespace=namespace, label_selector=f"app={computer_name}")
    ...

The timer looks like a gimmick and is not. A silent spinner at 150 seconds is indistinguishable from a hung request. A counter that says “waiting 150 seconds” tells the user the system is alive, and — because they learn the normal range — lets them notice themselves when something is wrong. Feedback that builds a mental model beats feedback that just occupies the screen.

There is also this line, which is a small confession:

if line.startswith(b"Error: context deadline exceeded"):
    continue

Helm is called with --wait --timeout 1m0s. A Windows pod frequently takes longer than a minute to become ready, so Helm times out and prints an error while the deployment is in fact proceeding fine. Rather than raise the timeout, we suppress the line and rely on our own readiness poll afterwards, which reports:

yield WebsocketMessage(action="status", payload={ "type": "substep", "status": "warning",
    "message": f"The computer '{computer_name}' might still be starting. Waiting a bit longer..." })

It works. It is also filtering a tool’s error output by string matching, which is the kind of thing that quietly breaks when the tool rewords a message. The honest fix is a timeout that reflects reality, or --wait=false plus our own polling — which is effectively what the code does anyway, just with an extra error to hide.

The broker: 60 lines, path-keyed

The WebSocket broker is deliberately tiny:

connections = defaultdict(set)

async def handler(websocket):
    path = websocket.path
    connections[path].add(websocket)

    try:
        async for message in websocket:
            peers = [ws for ws in connections[path] if ws != websocket and ws.open]
            for ws in peers:
                await ws.send(message)
    finally:
        connections[path].discard(websocket)
        if not connections[path]:
            del connections[path]

That is the entire routing logic. Connect to a path, receive everything anyone else publishes on that path. No topics to declare, no subscriptions to manage, no broker state beyond a dict of sets.

The paths are derived from domain identity:

WS_PATH = os.getenv("WS_PATH", f"/test/{TEST_INSTANCE_ID}")

The browser opens /test/abc123. The Kubernetes Job running that test also opens /test/abc123 and publishes progress. They find each other through a shared name, having never been introduced.

Plus a detail that matters in a cluster:

async with serve(handler, "0.0.0.0", port,
                 ping_interval=30,   # ping every 30 seconds
                 ping_timeout=10):   # close if no pong in 10 seconds

Without heartbeats, idle WebSocket connections through a load balancer or ingress get silently reaped, and the browser sits on a socket that will never deliver anything. Nothing errors; the UI just stops updating. Application-level pings keep the connection warm and detect death quickly.

Why a separate broker

The obvious design is WebSockets on the API. We did not, for one reason.

The publishers are Kubernetes Jobs. A test run is a short-lived Job. It comes into existence, does ten minutes of work, publishes progress throughout, and terminates. It has no stable network identity and no inbound route — nothing can connect to it.

The browser cannot connect to a Job. The Job cannot connect to a browser. Both can connect out to a well-known broker and rendezvous on a shared path.

Putting that in the API would mean the API holds every browser connection and also accepts connections from Jobs, and now the API cannot be restarted or horizontally scaled without dropping live sessions. Separating it means the stateful, connection-holding component is 60 lines with no dependencies, and the API stays a stateless REST service you can roll whenever you like.

The decision I would not repeat

Every deployment does this:

token = os.getenv("GITHUB_TOKEN")
HELM_REPO_URL = f"https://{token}@github.com/420-ai/OpenOperator-Infra.git"

with tempfile.TemporaryDirectory() as tmpdir:
    Repo.clone_from(HELM_REPO_URL, tmpdir)
    chart_path = os.path.join(tmpdir, "k8s-Helm", "OOUser", "computers", os_type.lower())
    ...
    process = await asyncio.create_subprocess_exec("helm", "upgrade", "--install", ...)

The API git-clones its own infrastructure repository into a temp directory and shells out to the helm binary.

In its defence: it was fast to build, charts are always current with main, and there was no chart repository to run. It worked reliably for months.

The problems are structural, and I would fix them before anyone else adopted it:

1. No version pinning. You get whatever is on main at that moment. A provision at 10 and one at 10 can use different charts, and a benchmark result carries no record of which. That is a reproducibility hole in a system whose entire purpose is reproducible measurement.
2. A long-lived PAT in the API’s environment, with read access to a private repository, sitting in a pod that also runs helm against the cluster.
3. Clone latency and failure on every single deploy, coupling every provisioning operation to GitHub’s availability.
4. Shelling out to a binary means the API image needs helm and git installed at compatible versions, and parsing progress means reading stdout.

The right answer is a chart repository — OCI charts in a registry, versioned, pulled by digest, with the version recorded on the run. Beyond that, a Computer CRD and a controller: the API writes a resource describing the desired computer, and a controller reconciles it. That gets you idempotency, retries, observable status and drift correction for free — all things this code implements badly by hand.

We knew. It stayed on the list, because it worked and the queue was long. That is usually why.

What the UI became

A React/TypeScript dashboard with Fluent UI: computers list, test configuration forms, benchmark runs, live desktop view, and a telemetry viewer added in June. The forms are generated from the same scenario schema the config format defines, which is the payoff for keeping scenarios as data — a non-engineer can now compose one in a browser instead of editing JSON.

The part users actually commented on was none of that. It was the step list ticking through “Cloning Helm repo… Deploying computer… Waiting 47 seconds… Pod is ready.”

Three minutes of waiting with visible progress feels shorter than thirty seconds of a spinner. That is not a UI opinion, it is a fact about how people estimate duration, and it cost about a day to implement.

Next: why we deleted every per-VM proxy pod and replaced them with one gateway.