Open computers from the chat, let bots ask for a take-over, show task progress

- Serve each computer to its owner under /computers/<id>/ (HTTP and
  websockets proxied to the AIO sandbox) and show it in a shared
  ComputerView with browser, terminal and code modes, opened from the
  computers modal or a take-over button in the thread.
- ask_user_to_take_over lets a bot hand a computer to the user for a
  login or captcha; the message gets a take-over button and needs-you,
  and the viewer hands it back with a chat message.
- While a task runs the thread shows "<bot> is working..." with the
  task's latest step; steps read as plain words, not tool names.
- A coordinator with no workers no longer claims to delegate, and bots
  describe started tasks in the first person.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

src/functions/server/computers/computer-proxy-socket.ts

@@ -0,0 +1,69 @@
+import type { ServerWebSocket } from "bun";
+import type { WebSocketSession } from "@/src/websocket/types";
+
+type Socket = ServerWebSocket<WebSocketSession>;
+
+/**
+ * Bridges a user's websocket on `/computers/<id>/...` to the same path on
+ * the computer (a terminal shell, the browser's CDP, the code editor).
+ * Messages sent before the computer's side opens are held, and either
+ * side closing closes the other.
+ */
+export const computerProxySocket = {
+    open(ws: Socket) {
+        const proxy = ws.data.proxy;
+
+        if (!proxy) return;
+
+        const upstream = new WebSocket(proxy.url, {
+            headers: proxy.headers,
+            protocols: proxy.protocols,
+        } as any);
+
+        upstream.binaryType = "arraybuffer";
+        proxy.upstream = upstream;
+        proxy.pending = [];
+
+        upstream.onopen = () => {
+            for (const message of proxy.pending || []) upstream.send(message as string | Uint8Array<ArrayBuffer>);
+            proxy.pending = undefined;
+        };
+
+        upstream.onmessage = (event) => {
+            ws.send(typeof event.data == "string" ? event.data : new Uint8Array(event.data));
+        };
+
+        upstream.onclose = (event) => {
+            ws.close(closeCode({ code: event.code }), event.reason);
+        };
+
+        upstream.onerror = () => {
+            ws.close(1011, "The computer isn't answering");
+        };
+    },
+
+    message(ws: Socket, message: string | Buffer) {
+        const proxy = ws.data.proxy;
+
+        if (!proxy?.upstream) return;
+
+        if (proxy.pending) proxy.pending.push(message);
+        else if (proxy.upstream.readyState == WebSocket.OPEN) {
+            proxy.upstream.send(message as string | Uint8Array<ArrayBuffer>);
+        }
+    },
+
+    close(ws: Socket) {
+        const upstream = ws.data.proxy?.upstream;
+
+        if (upstream && upstream.readyState < WebSocket.CLOSING) upstream.close();
+    },
+};
+
+/**
+ * Codes a server may not send itself (no status, abnormal) become a
+ * normal close.
+ */
+function closeCode({ code }: { code: number }) {
+    return code == 1005 || code == 1006 || code == 1015 ? 1000 : code;
+}

src/functions/server/computers/computer-proxy-target.ts

@@ -0,0 +1,46 @@
+import grabComputerMcpUrl from "@/src/functions/backend/db/computers/grab-computer-mcp-url";
+import grabUserComputer from "@/src/functions/backend/db/computers/grab-user-computer";
+import userAuth from "@/src/functions/backend/auth/user-auth";
+import parseComputerProxyPath from "@/src/utils/parse-computer-proxy-path";
+
+type Params = {
+    req: Request;
+};
+
+/**
+ * Where a `/computers/<id>/...` request goes: the computer's own address
+ * for that path, only for the computer's owner. An error response when
+ * the user isn't logged in, the computer isn't theirs, or it has no
+ * container.
+ */
+export default async function computerProxyTarget({ req }: Params) {
+    const url = new URL(req.url);
+    const parsed = parseComputerProxyPath({ pathname: url.pathname });
+
+    if (!parsed) return { error: new Response("Not Found", { status: 404 }) };
+
+    const session = await userAuth({ req });
+
+    if (!session.user?.id) return { error: new Response("Unauthorized", { status: 401 }) };
+
+    const computer = await grabUserComputer({
+        computer_id: parsed.computer_id,
+        user_id: session.user.id,
+    });
+
+    if (!computer?.id) return { error: new Response("Not Found", { status: 404 }) };
+
+    const address = await grabComputerMcpUrl({ computer_id: computer.id });
+
+    if (!address) {
+        return { error: new Response("This computer has no container.", { status: 502 }) };
+    }
+
+    return {
+        session,
+        url,
+        prefix: parsed.prefix,
+        base_url: address.base_url,
+        target: `${address.base_url}/${parsed.rest}${url.search}`,
+    };
+}

src/functions/server/computers/handle-computer-proxy.ts

@@ -0,0 +1,70 @@
+import type { Server } from "bun";
+import computerProxyRequestHeaders from "@/src/utils/computer-proxy-request-headers";
+import computerProxyResponseHeaders from "@/src/utils/computer-proxy-response-headers";
+import type { WebSocketSession } from "@/src/websocket/types";
+import computerProxyTarget from "./computer-proxy-target";
+
+type Params = {
+    req: Request;
+    server: Server<WebSocketSession>;
+};
+
+/**
+ * Serves a computer's own apps (browser, terminal, code) to its owner
+ * under `/computers/<id>/`, so the chat can show them in a frame without
+ * the sandbox being reachable from the user's browser. Plain requests are
+ * fetched and streamed back; websockets are bridged
+ * (`computerProxySocket`).
+ */
+export default async function handleComputerProxy({ req, server }: Params) {
+    const resolved = await computerProxyTarget({ req });
+
+    if (resolved.error) return resolved.error;
+
+    const { session, url, prefix, base_url, target } = resolved;
+    const headers = computerProxyRequestHeaders({ headers: req.headers, url, prefix });
+
+    if (req.headers.get("upgrade")?.toLowerCase() == "websocket") {
+        const protocols = (req.headers.get("sec-websocket-protocol") || "")
+            .split(",")
+            .map((p) => p.trim())
+            .filter(Boolean);
+
+        for (const name of [...headers.keys()]) {
+            if (name.startsWith("sec-websocket-")) headers.delete(name);
+        }
+
+        const upgraded = server.upgrade(req, {
+            headers: protocols[0] ? { "Sec-WebSocket-Protocol": protocols[0] } : undefined,
+            data: {
+                user: session.user!,
+                user_types: session.user_types ?? [],
+                proxy: {
+                    url: target.replace(/^http/, "ws"),
+                    headers: Object.fromEntries(headers.entries()),
+                    protocols,
+                },
+            } satisfies WebSocketSession,
+        });
+
+        return upgraded ? undefined : new Response("Upgrade failed", { status: 400 });
+    }
+
+    const has_body = req.method != "GET" && req.method != "HEAD";
+
+    const res = await fetch(target, {
+        method: req.method,
+        headers,
+        body: has_body ? req.body : undefined,
+        redirect: "manual",
+        signal: req.signal,
+    }).catch(() => null);
+
+    if (!res) return new Response("This computer isn't answering.", { status: 502 });
+
+    return new Response(res.body, {
+        status: res.status,
+        statusText: res.statusText,
+        headers: computerProxyResponseHeaders({ headers: res.headers, base_url, prefix }),
+    });
+}