diff --git a/server/computer-control.test.ts b/server/computer-control.test.ts new file mode 100644 index 0000000000..ac22cb069e --- /dev/null +++ b/server/computer-control.test.ts @@ -0,0 +1,130 @@ +// The who-is-driving record. What these tests pin is the authority split: +// the person's three moves (take, release, dismiss) all work, the bot's one +// move (requestHelp) never grants anything, and a release settles the help +// request in the same change the person made. +import { describe, expect, it } from "vitest"; + +import { ComputerControl, type ControlSnapshot } from "./computer-control.ts"; + +function tracked() { + const changes: Array<{ botId: string; snapshot: ControlSnapshot }> = []; + const control = new ComputerControl((botId, snapshot) => changes.push({ botId, snapshot })); + return { control, changes }; +} + +describe("computer control", () => { + it("starts disengaged for an unknown bot", () => { + const { control } = tracked(); + expect(control.snapshot("b1")).toEqual({ held: false, helpReason: null, heldSinceMs: null }); + }); + + it("take → held; release → disengaged, each broadcast once", () => { + const { control, changes } = tracked(); + const held = control.take("b1"); + expect(held.held).toBe(true); + expect(held.heldSinceMs).not.toBeNull(); + expect(control.release("b1").held).toBe(false); + expect(changes.map((c) => c.snapshot.held)).toEqual([true, false]); + }); + + it("a second take does not reset how long the hold has lasted", () => { + let clock = 1000; + const control = new ComputerControl(() => {}, () => clock); + control.take("b1"); + clock = 5000; + expect(control.take("b1").heldSinceMs).toBe(1000); + }); + + it("requestHelp surfaces the plea but never grants control", () => { + const { control } = tracked(); + const snapshot = control.requestHelp("b1", " please log in for me "); + expect(snapshot.held).toBe(false); + expect(snapshot.helpReason).toBe("please log in for me"); + }); + + it("an empty reason still reads as a plea", () => { + const { control } = tracked(); + expect(control.requestHelp("b1", undefined).helpReason).toBe("the bot asked you to take over"); + }); + + it("a shouted second reason cannot clobber the one the person is reading", () => { + const { control } = tracked(); + control.requestHelp("b1", "first"); + expect(control.requestHelp("b1", "second").helpReason).toBe("first"); + }); + + it("expires only the help request that owns the timeout", () => { + const { control, changes } = tracked(); + const first = control.requestHelpLease("b1", "first"); + expect(control.expireHelp("b1", "some-older-request").helpReason).toBe("first"); + expect(changes).toHaveLength(1); + expect(control.expireHelp("b1", first.requestId).helpReason).toBeNull(); + expect(changes).toHaveLength(2); + }); + + it("an old timeout cannot dismiss a newer plea", () => { + const { control } = tracked(); + const first = control.requestHelpLease("b1", "first"); + control.dismissHelp("b1"); + const second = control.requestHelpLease("b1", "second"); + expect(second.requestId).not.toBe(first.requestId); + expect(control.expireHelp("b1", first.requestId).helpReason).toBe("second"); + }); + + it("a novel-length reason is cut to card size", () => { + const { control } = tracked(); + const reason = "x".repeat(2000); + expect(control.requestHelp("b1", reason).helpReason?.length).toBe(280); + }); + + it("release settles an open help request in the same change", () => { + const { control } = tracked(); + control.requestHelp("b1", "stuck on a captcha"); + control.take("b1"); + const after = control.release("b1"); + expect(after).toEqual({ held: false, helpReason: null, heldSinceMs: null }); + }); + + it("dismiss clears the plea without taking control", () => { + const { control } = tracked(); + control.requestHelp("b1", "stuck"); + const after = control.dismissHelp("b1"); + expect(after.helpReason).toBeNull(); + expect(after.held).toBe(false); + }); + + it("dismiss while driving keeps the hold", () => { + const { control } = tracked(); + control.take("b1"); + control.requestHelp("b1", "also this"); + const after = control.dismissHelp("b1"); + expect(after.held).toBe(true); + expect(after.helpReason).toBeNull(); + }); + + it("dismissing nothing is silent — no phantom broadcast", () => { + const { control, changes } = tracked(); + control.dismissHelp("b1"); + expect(changes).toEqual([]); + }); + + it("bots are independent", () => { + const { control } = tracked(); + control.take("b1"); + expect(control.snapshot("b2").held).toBe(false); + }); + + it("forget clears a hold and tells the listeners", () => { + const { control, changes } = tracked(); + control.take("b1"); + control.forget("b1"); + expect(control.snapshot("b1").held).toBe(false); + expect(changes.at(-1)?.snapshot).toEqual({ held: false, helpReason: null, heldSinceMs: null }); + }); + + it("forgetting an unknown bot is silent", () => { + const { control, changes } = tracked(); + control.forget("ghost"); + expect(changes).toEqual([]); + }); +}); diff --git a/server/computer-control.ts b/server/computer-control.ts new file mode 100644 index 0000000000..604d6492c6 --- /dev/null +++ b/server/computer-control.ts @@ -0,0 +1,135 @@ +// Who is driving a bot's computer — the person or the bot. One record per +// bot, held in the harness because every consumer (the panel, the SSE +// stream, and the per-turn computer proxies) already talks to the harness. +// +// The rules this module exists to enforce: +// - A bot can only ASK for hands (`requestHelp`); it can never take +// control, and it cannot clear a hold. Only the person takes and +// releases, from the computer panel. +// - While the person holds control, the bot's computer actions are +// REFUSED by the proxies, not queued. A queued click would land after +// the person has moved on, on whatever happens to be under it. +// - Releasing control also settles any open help request, so a bot +// waiting on `requestHelp` wakes up from the same state change the +// person made — there is no separate "done helping" step to forget. +// +// State is per-boot and in-memory on purpose: a hold is a live fact about +// who is at the screen right now, and surviving a harness restart would +// mean a stale hold silently bricking a bot's computer. + +export interface ControlSnapshot { + /** True while the person is driving; the bot's hands are refused. */ + held: boolean; + /** The bot's open plea for help, verbatim, or null when none is open. */ + helpReason: string | null; + heldSinceMs: number | null; +} + +const NO_CONTROL: ControlSnapshot = { held: false, helpReason: null, heldSinceMs: null }; +/** Keep a shouted help reason card-sized; the transcript has the rest. */ +const MAX_REASON_CHARS = 280; + +interface Entry { + heldSinceMs: number | null; + helpReason: string | null; + helpRequestId: string | null; +} + +export class ComputerControl { + private entries = new Map(); + private onChange: (botId: string, snapshot: ControlSnapshot) => void; + private now: () => number; + private requestSequence = 0; + + constructor( + onChange: (botId: string, snapshot: ControlSnapshot) => void = () => {}, + now: () => number = Date.now, + ) { + this.onChange = onChange; + this.now = now; + } + + snapshot(botId: string): ControlSnapshot { + const entry = this.entries.get(botId); + if (!entry) return NO_CONTROL; + return { + held: entry.heldSinceMs !== null, + helpReason: entry.helpReason, + heldSinceMs: entry.heldSinceMs, + }; + } + + /** The person takes the wheel. Idempotent — a second click must not + * reset `heldSinceMs` and make the hold look newer than it is. */ + take(botId: string): ControlSnapshot { + const entry = this.entries.get(botId); + if (entry?.heldSinceMs != null) return this.snapshot(botId); + this.entries.set(botId, { + heldSinceMs: this.now(), + helpReason: entry?.helpReason ?? null, + helpRequestId: entry?.helpRequestId ?? null, + }); + return this.changed(botId); + } + + /** The person hands the wheel back. Also settles any open help request — + * the waiting bot resumes from this one state change. */ + release(botId: string): ControlSnapshot { + if (!this.entries.has(botId)) return NO_CONTROL; + this.entries.delete(botId); + return this.changed(botId); + } + + /** The bot asks the person to take over. Never grants anything by + * itself — it only surfaces the plea. A reason shouted while the person + * is already driving is kept, but must not clobber an earlier one they + * may still be reading. */ + requestHelp(botId: string, reason: unknown): ControlSnapshot { + return this.requestHelpLease(botId, reason).snapshot; + } + + /** Open a help request and return the lease that owns it. A proxy uses + * this id to expire only its own unanswered plea when its wait ends. */ + requestHelpLease(botId: string, reason: unknown): { snapshot: ControlSnapshot; requestId: string } { + const text = typeof reason === "string" ? reason.trim().slice(0, MAX_REASON_CHARS) : ""; + const entry = this.entries.get(botId) ?? { heldSinceMs: null, helpReason: null, helpRequestId: null }; + if (entry.helpReason === null) { + entry.helpReason = text || "the bot asked you to take over"; + entry.helpRequestId = `${botId}-${++this.requestSequence}`; + } + this.entries.set(botId, entry); + return { snapshot: this.changed(botId), requestId: entry.helpRequestId! }; + } + + /** The person declines without taking over; the waiting bot is told. */ + dismissHelp(botId: string): ControlSnapshot { + const entry = this.entries.get(botId); + if (!entry || entry.helpReason === null) return this.snapshot(botId); + entry.helpReason = null; + entry.helpRequestId = null; + if (entry.heldSinceMs === null) this.entries.delete(botId); + return this.changed(botId); + } + + /** Expire an unanswered plea. The id comparison prevents an old proxy's + * timeout from dismissing a newer request for the same bot. */ + expireHelp(botId: string, requestId: unknown): ControlSnapshot { + const entry = this.entries.get(botId); + if (!entry || entry.helpRequestId !== requestId) return this.snapshot(botId); + entry.helpReason = null; + entry.helpRequestId = null; + if (entry.heldSinceMs === null) this.entries.delete(botId); + return this.changed(botId); + } + + /** The bot is gone; a hold on a deleted bot's computer means nothing. */ + forget(botId: string): void { + if (this.entries.delete(botId)) this.onChange(botId, NO_CONTROL); + } + + private changed(botId: string): ControlSnapshot { + const snapshot = this.snapshot(botId); + this.onChange(botId, snapshot); + return snapshot; + } +} diff --git a/server/computer-proxy.test.ts b/server/computer-proxy.test.ts index 102c7de88f..43700acf38 100644 --- a/server/computer-proxy.test.ts +++ b/server/computer-proxy.test.ts @@ -470,3 +470,197 @@ describe("computer proxy (fake box)", () => { expect(repeated.result.content[0].text).toMatch(/identical/i); }); }); + +describe("computer proxy control gate (fake box + fake control)", () => { + let box: Server; + let controlServer: Server; + let proxy: ChildProcess; + const commands: string[] = []; + let held = false; + let helpOpen = false; + let failHelpPost = false; + const expiredHelpIds: string[] = []; + const authHeaders: Array = []; + + const rpc = (msg: unknown) => proxy.stdin!.write(JSON.stringify(msg) + "\n"); + const results = new Map(); + const waitFor = async (id: number, ms = 8000) => { + const deadline = Date.now() + ms; + while (Date.now() < deadline) { + if (results.has(id)) return results.get(id); + await new Promise((r) => setTimeout(r, 10)); + } + throw new Error(`no response for id ${id}`); + }; + + beforeAll(async () => { + box = createServer((req, res) => { + const url = new URL(req.url ?? "/", "http://x"); + if (url.pathname.endsWith("/commands")) { + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + commands.push(JSON.parse(body || "{}").command ?? ""); + const size = Buffer.from(JPEG, "base64").length; + res.writeHead(200, { "content-type": "application/json" }); + res.end( + JSON.stringify({ + exitCode: 0, + stdout: `GEOM 1920 1080\nHASH h1\nSIZE ${size}\nB64 ${JPEG}\nACT ok\n`, + stderr: "", + }), + ); + }); + return; + } + res.writeHead(404).end("{}"); + }); + await new Promise((r) => box.listen(0, "127.0.0.1", r)); + const boxPort = (box.address() as any).port; + + controlServer = createServer((req, res) => { + authHeaders.push(Array.isArray(req.headers.authorization) ? undefined : req.headers.authorization); + if (req.method === "POST") { + if (failHelpPost) { + req.resume(); + res.writeHead(503, { "content-type": "application/json" }); + res.end(JSON.stringify({ error: "offline" })); + return; + } + helpOpen = true; + req.resume(); + req.on("end", () => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ held, helpOpen, requestId: "help-1" })); + }); + return; + } + if (req.method === "DELETE") { + let body = ""; + req.on("data", (chunk) => (body += chunk)); + req.on("end", () => { + expiredHelpIds.push(JSON.parse(body || "{}").requestId); + helpOpen = false; + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ held, helpOpen })); + }); + return; + } + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ held, helpOpen })); + }); + await new Promise((r) => controlServer.listen(0, "127.0.0.1", r)); + const controlPort = (controlServer.address() as any).port; + + proxy = spawn(process.execPath, ["--experimental-strip-types", PROXY], { + env: { + ...process.env, + OGB_BOX_API: `http://127.0.0.1:${boxPort}`, + OGB_BOX_ID: "box-1", + OGB_BOX_TOKEN: "t", + OMB_CONTROL_URL: `http://127.0.0.1:${controlPort}/api/internal/computer-control?botId=b1`, + OMB_CONTROL_TOKEN: "control-secret", + // fast cadence so the wait tests measure logic, not wall-clock + OMB_CONTROL_POLL_MS: "25", + OMB_CONTROL_WAIT_MS: "1500", + }, + stdio: ["pipe", "pipe", "pipe"], + }); + let buf = ""; + proxy.stdout!.on("data", (c) => { + buf += c; + let nl; + while ((nl = buf.indexOf("\n")) !== -1) { + const line = buf.slice(0, nl); + buf = buf.slice(nl + 1); + if (!line.trim()) continue; + try { + const msg = JSON.parse(line); + if (msg.id != null) results.set(msg.id, msg); + } catch { + /* ignore */ + } + } + }); + rpc({ jsonrpc: "2.0", id: 1, method: "initialize", params: {} }); + await waitFor(1); + }, 20_000); + + afterAll(() => { + proxy?.kill(); + box?.close(); + controlServer?.close(); + }); + + it("acts normally while nobody is driving, sending the boot token along", async () => { + held = false; + rpc({ jsonrpc: "2.0", id: 2, method: "tools/call", params: { name: "click", arguments: { x: 10, y: 10, observe: false } } }); + const result = await waitFor(2); + expect(result.result.isError).toBeUndefined(); + expect(commands.length).toBeGreaterThan(0); + expect(authHeaders.every((h) => h === "Bearer control-secret")).toBe(true); + }); + + it("refuses every action while the person is driving — nothing reaches the box", async () => { + held = true; + const before = commands.length; + rpc({ jsonrpc: "2.0", id: 3, method: "tools/call", params: { name: "click", arguments: { x: 10, y: 10 } } }); + const click = await waitFor(3); + expect(click.result.isError).toBe(true); + expect(click.result.content[0].text).toMatch(/taken control/i); + rpc({ jsonrpc: "2.0", id: 4, method: "tools/call", params: { name: "screenshot", arguments: {} } }); + const shot = await waitFor(4); + expect(shot.result.isError).toBe(true); + expect(commands.length).toBe(before); + }); + + it("computer_request_help waits out the drive and reports the hand-back", async () => { + held = true; + helpOpen = false; + rpc({ jsonrpc: "2.0", id: 5, method: "tools/call", params: { name: "computer_request_help", arguments: {} } }); + await new Promise((r) => setTimeout(r, 120)); + expect(results.has(5)).toBe(false); // still waiting while they drive + held = false; + const result = await waitFor(5); + expect(result.result.isError).toBeUndefined(); + expect(result.result.content[0].text).toMatch(/handed control back/i); + expect(result.result.content[0].text).toMatch(/fresh screenshot/i); + }); + + it("computer_request_help posts the plea and reports a dismissal", async () => { + held = false; + helpOpen = false; + await new Promise((r) => setTimeout(r, 40)); + rpc({ jsonrpc: "2.0", id: 6, method: "tools/call", params: { name: "computer_request_help", arguments: { reason: "please log in" } } }); + await new Promise((r) => setTimeout(r, 120)); + expect(helpOpen).toBe(true); // the POST landed + expect(results.has(6)).toBe(false); // and the bot is waiting + helpOpen = false; // the person dismissed it + const result = await waitFor(6); + expect(result.result.content[0].text).toMatch(/dismissed/i); + }); + + it("times out politely when nobody comes", async () => { + held = false; + helpOpen = false; + await new Promise((r) => setTimeout(r, 40)); + rpc({ jsonrpc: "2.0", id: 7, method: "tools/call", params: { name: "computer_request_help", arguments: {} } }); + helpOpen = true; // plea stays open, nobody answers + const result = await waitFor(7, 4000); + expect(result.result.isError).toBe(true); + expect(result.result.content[0].text).toMatch(/nobody took control/i); + expect(helpOpen).toBe(false); + expect(expiredHelpIds).toContain("help-1"); + }); + + it("returns immediately when the person cannot be paged", async () => { + held = false; + helpOpen = false; + failHelpPost = true; + rpc({ jsonrpc: "2.0", id: 8, method: "tools/call", params: { name: "computer_request_help", arguments: {} } }); + const result = await waitFor(8, 500); + failHelpPost = false; + expect(result.result.isError).toBe(true); + expect(result.result.content[0].text).toMatch(/could not be paged/i); + }); +}); diff --git a/server/computer-proxy.ts b/server/computer-proxy.ts index 4e955b11cc..362d296265 100644 --- a/server/computer-proxy.ts +++ b/server/computer-proxy.ts @@ -35,6 +35,7 @@ import { type BrowserTarget, type CropRegion, } from "./computer-observation.ts"; +import { CONTROL_REFUSAL, createControlClient } from "./control-client.ts"; import { ensureRemoteCuaCommand, REMOTE_CUA_EXECUTABLE, @@ -48,6 +49,18 @@ const BOX_API = process.env.OGB_BOX_API ?? "https://ascii.dev/api/box/v1"; const boxId = process.env.OGB_BOX_ID ?? ""; const token = process.env.OGB_BOX_TOKEN ?? ""; +// Who-is-driving: while the person holds control in the app, every tool +// below is refused (not queued — a queued click lands after they've moved +// on). Even screenshot: the person may be typing a credential, and the +// safest screen for the model to see is the one AFTER the hand-back. +/** Poll cadence while waiting for a hand-back, and the patience ceiling. + * Env-tunable so the contract test doesn't spend wall-clock on it. */ +const CONTROL_POLL_MS = Math.max(Number(process.env.OMB_CONTROL_POLL_MS) || 1_500, 25); +const CONTROL_WAIT_MS = Math.max(Number(process.env.OMB_CONTROL_WAIT_MS) || 600_000, 100); +// The cache must never outlive the poll cadence, or a hand-back would be +// seen a stale cache-window late. +const control = createControlClient({ cacheMs: Math.min(750, CONTROL_POLL_MS) }); + /** The coordinate space the model sees: frames are downscaled to this * width, and clicks are scaled back up to the real display box-side. */ const SHOT_WIDTH = 1280; @@ -518,6 +531,20 @@ const TOOLS = [ description: "Report whether the cloud computer is using Cua Driver or the degraded X11 fallback.", inputSchema: { type: "object", properties: {} }, }, + { + name: "computer_request_help", + description: + "Ask the person to take over this computer (a login, a CAPTCHA, anything you should not do alone) and wait until they hand control back. Also call it with no reason when an action was refused because a person is already driving. You cannot take control yourself — this only asks.", + inputSchema: { + type: "object", + properties: { + reason: { + type: "string", + description: "one short sentence the person will read — what you need their hands for", + }, + }, + }, + }, { name: "click", description: @@ -771,7 +798,47 @@ async function semanticActAndObserve( return observed(id, note, await frameFrom(out)); } +/** The only tools a bot may use while the person is driving: asking to be + * told when they finish, and the two that read nothing from the screen. */ +const OPEN_WHILE_DRIVEN = new Set(["computer_request_help", "computer_status", "observation_metrics"]); + async function call(id: unknown, name: string, args: any) { + if (!OPEN_WHILE_DRIVEN.has(name) && (await control.state(true)).held) { + return text(id, CONTROL_REFUSAL, true); + } + if (name === "computer_request_help") { + if (!control.configured) { + return text(id, "nobody can be paged for this computer right now — carry on carefully", true); + } + const initial = await control.state(true); + // If the person is already driving, don't clobber whatever plea they + // are reading — just wait for the hand-back. + const requestId = initial.held ? null : await control.requestHelp(String(args?.reason ?? "")); + if (!initial.held && requestId === null) { + return text(id, "The person could not be paged for this computer right now. Carry on carefully or tell them in chat.", true); + } + let sawHold = initial.held; + const deadline = Date.now() + CONTROL_WAIT_MS; + while (Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, CONTROL_POLL_MS)); + const state = await control.state(true); + if (state.held) sawHold = true; + if (!state.held && !state.helpOpen) { + return text( + id, + sawHold + ? "The person has finished driving and handed control back. The screen may have changed while they drove — take a fresh screenshot before your next action." + : "The person saw your request and dismissed it without taking control. Carry on yourself.", + ); + } + } + if (requestId) await control.expireHelp(requestId); + return text( + id, + "Nobody took control within the wait window. Carry on carefully, or ask again if you are truly stuck.", + true, + ); + } if (name === "screenshot") { let crop: CropRegion | null = null; if (args.region !== undefined) { diff --git a/server/container-computer.ts b/server/container-computer.ts index 1b6f97c13c..a0cb3da723 100644 --- a/server/container-computer.ts +++ b/server/container-computer.ts @@ -835,11 +835,19 @@ type ContainerMcpLaunch = { env: Record; }; -export function containerComputerMcp(runtime: Runtime): ContainerMcpLaunch { +export function containerComputerMcp( + runtime: Runtime, + control?: { url: string; token: string }, +): ContainerMcpLaunch { return { command: process.execPath, args: [containerMcpPath, runtime, CONTAINER, CUA_SOCKET], - env: { ELECTRON_RUN_AS_NODE: "1" }, + // The control pair rides in env, not argv — argv is world-readable + // through `ps` for the life of the bridge. + env: { + ELECTRON_RUN_AS_NODE: "1", + ...(control ? { OMB_CONTROL_URL: control.url, OMB_CONTROL_TOKEN: control.token } : {}), + }, }; } @@ -897,7 +905,13 @@ export function setupCommands( * bypass it and mount Cua Driver's official MCP server through * containerComputerMcp(). */ export function computerProxyEnv( - computer: { boxId?: string; token?: string }, + computer: { boxId?: string; token?: string; control?: { url: string; token: string } }, ): NodeJS.ProcessEnv { - return { OGB_BOX_ID: computer.boxId ?? "", OGB_BOX_TOKEN: computer.token ?? "" }; + return { + OGB_BOX_ID: computer.boxId ?? "", + OGB_BOX_TOKEN: computer.token ?? "", + ...(computer.control + ? { OMB_CONTROL_URL: computer.control.url, OMB_CONTROL_TOKEN: computer.control.token } + : {}), + }; } diff --git a/server/container-mcp.ts b/server/container-mcp.ts index 4c0562788b..7bc48cd7f8 100644 --- a/server/container-mcp.ts +++ b/server/container-mcp.ts @@ -15,10 +15,16 @@ if (!container || !/^[a-zA-Z0-9_.-]+$/.test(container) || !socket?.startsWith("/ process.exit(2); } +// The who-is-driving pair rides in env, not argv — argv is world-readable +// through `ps`, and the token guards a loopback endpoint. +const controlUrl = process.env.OMB_CONTROL_URL ?? ""; +const controlToken = process.env.OMB_CONTROL_TOKEN ?? ""; + runMcpBridge({ command: runtime, args: cuaExecArgs(["mcp", "--socket", socket], { container, interactive: true }), label: "Cua Driver", // No liveness watchdog: the runtime CLI talks to a local daemon and fails // fast on its own — there is no silent WAN peer to wedge on. + ...(controlUrl && controlToken ? { gate: { url: controlUrl, token: controlToken } } : {}), }); diff --git a/server/contracts.ts b/server/contracts.ts index 0ba6958b26..3fea011bdc 100644 --- a/server/contracts.ts +++ b/server/contracts.ts @@ -155,8 +155,16 @@ export interface SendTurnInput { * bridge harness-controlled lets it turn connection requests into trusted * chat cards consistently across provider CLIs. */ composio?: { command: string; args: string[]; env: Record }; - /** Cloud computer, reached through OpenMausBot's REST-to-MCP adapter. */ - computer?: { kind?: "box"; boxId: string; token: string }; + /** Cloud computer, reached through OpenMausBot's REST-to-MCP adapter. + * `control` is the harness's loopback who-is-driving endpoint: the + * adapter consults it so a person who takes the wheel in the panel + * pauses the bot's hands mid-turn instead of typing over them. */ + computer?: { + kind?: "box"; + boxId: string; + token: string; + control?: { url: string; token: string }; + }; /** Direct stdio connection to a Cua Driver MCP server (host, sandbox, or * VPS). `scope` is set only for the user's host desktop; isolated and * remote computers intentionally omit it so host-only approval rules diff --git a/server/control-client.ts b/server/control-client.ts new file mode 100644 index 0000000000..ee4915eafa --- /dev/null +++ b/server/control-client.ts @@ -0,0 +1,127 @@ +// The proxy-side half of computer control. The harness keeps the record of +// who is driving (server/computer-control.ts); the per-turn computer +// processes consult it through this client before acting, because the +// action paths themselves never traverse the harness — a box click goes +// straight to the box's REST API, and a Local VM / VPS click rides a +// transparent stdio bridge into Cua Driver. +// +// Failure posture: OPEN. Control is cooperation between the person and +// their own bot — "hold my hands while you're driving" — not a security +// boundary against a hostile agent (a hostile agent could reach the same +// REST endpoint without this proxy). Failing closed would mean a harness +// hiccup bricks every computer mid-turn, which costs more than the race +// it would prevent: while the person is driving they are watching the +// screen, and the panel shows the hold either way. +// +// The state is cached briefly so a computer_batch of two dozen actions +// doesn't turn into two dozen loopback round trips. + +export interface ControlState { + /** The person is driving; actions must be refused, not queued. */ + held: boolean; + /** A help request the person has neither answered nor dismissed. */ + helpOpen: boolean; +} + +export interface ControlClient { + /** Current state, cached for `cacheMs`. `fresh` bypasses the cache — + * the wait loop in request-help polls with it so a hand-back is seen + * within one poll interval, not one poll interval plus the cache. */ + state(fresh?: boolean): Promise; + /** Surface the bot's plea in the app. Returns its expiry id, or null when + * the harness could not be told. */ + requestHelp(reason: string): Promise; + /** Close only the unanswered plea opened by this client. */ + expireHelp(requestId: string): Promise; + readonly configured: boolean; +} + +const DISENGAGED: ControlState = { held: false, helpOpen: false }; + +export function createControlClient(options?: { + url?: string; + token?: string; + cacheMs?: number; + fetchImpl?: typeof fetch; +}): ControlClient { + const url = options?.url ?? process.env.OMB_CONTROL_URL ?? ""; + const token = options?.token ?? process.env.OMB_CONTROL_TOKEN ?? ""; + const cacheMs = options?.cacheMs ?? 750; + const fetchImpl = options?.fetchImpl ?? fetch; + const configured = Boolean(url && token); + const headers = { authorization: `Bearer ${token}`, "content-type": "application/json" }; + + let cachedAt = 0; + let cached: ControlState = DISENGAGED; + + async function read(): Promise { + try { + const res = await fetchImpl(url, { headers, signal: AbortSignal.timeout(2_000) }); + if (!res.ok) return DISENGAGED; + const body: any = await res.json().catch(() => null); + return { held: body?.held === true, helpOpen: body?.helpOpen === true }; + } catch { + return DISENGAGED; + } + } + + return { + configured, + async state(fresh = false): Promise { + if (!configured) return DISENGAGED; + const now = Date.now(); + if (!fresh && now - cachedAt < cacheMs) return cached; + cached = await read(); + cachedAt = Date.now(); + return cached; + }, + async requestHelp(reason: string): Promise { + if (!configured) return null; + try { + const res = await fetchImpl(url, { + method: "POST", + headers, + body: JSON.stringify({ reason }), + signal: AbortSignal.timeout(2_000), + }); + if (!res.ok) return null; + const body: any = await res.json().catch(() => null); + return typeof body?.requestId === "string" && body.requestId ? body.requestId : null; + } catch { + return null; + } + }, + async expireHelp(requestId: string): Promise { + if (!configured || !requestId) return; + try { + await fetchImpl(url, { + method: "DELETE", + headers, + body: JSON.stringify({ requestId }), + signal: AbortSignal.timeout(2_000), + }); + } catch { + // Best effort: the harness also clears the in-memory request on + // release/restart, and an unavailable harness cannot show the card. + } + }, + }; +} + +/** The one sentence every refused action gets. Deliberately does not vary + * per tool: the model needs the same three facts every time — nothing + * happened, don't retry blindly, and how to wait properly. */ +export const CONTROL_REFUSAL = + "A person has taken control of this computer, so this call was NOT performed. " + + "Do not retry it — the screen is changing under their hands. " + + "Call computer_request_help (no reason needed) to wait for them to finish, " + + "then take a fresh screenshot before your next action."; + +/** The bridge-gated computers (Local VM, VPS) speak Cua Driver's own tool + * surface, which has no wait tool to point at — so the guidance is to + * pause, not to call anything. */ +export const CONTROL_REFUSAL_PLAIN = + "A person has taken control of this computer, so this call was NOT performed. " + + "Do not retry it — the screen is changing under their hands. " + + "Pause this task, tell the person you are waiting for them to hand control back, " + + "and take a fresh screenshot before your next action once they have."; diff --git a/server/drivers/codex.ts b/server/drivers/codex.ts index fb1bf73772..7acc627493 100644 --- a/server/drivers/codex.ts +++ b/server/drivers/codex.ts @@ -154,6 +154,10 @@ export const CodexDriver: ProviderDriver = { ELECTRON_RUN_AS_NODE: "1", OGB_BOX_ID: proxyEnv.OGB_BOX_ID ?? "", OGB_BOX_TOKEN: proxyEnv.OGB_BOX_TOKEN ?? "", + // who-is-driving endpoint, so a person taking the wheel in the + // panel pauses this bot's hands mid-turn + OMB_CONTROL_URL: proxyEnv.OMB_CONTROL_URL ?? "", + OMB_CONTROL_TOKEN: proxyEnv.OMB_CONTROL_TOKEN ?? "", }, }); } else if (turn.integrations?.localComputer) { diff --git a/server/index.test.ts b/server/index.test.ts index d066b2b66a..5d59d71343 100644 --- a/server/index.test.ts +++ b/server/index.test.ts @@ -1292,3 +1292,58 @@ describe("instance CLI override API", () => { expect((await slowConfigWrite).status).toBe(200); }); }); + +describe("computer control API (who is driving)", () => { + let botId = ""; + + beforeAll(async () => { + const created = await api("POST", "/api/bots", {}); + botId = created.body.bot.id; + }); + + it("starts disengaged", async () => { + const res = await api("GET", `/api/bots/${botId}/computer/control`); + expect(res.status).toBe(200); + expect(res.body).toEqual({ held: false, helpReason: null, heldSinceMs: null }); + }); + + it("take → held, broadcast on the wire, release → disengaged", async () => { + const sse = await openSse(`${BASE}/api/events`); + try { + const took = await api("POST", `/api/bots/${botId}/computer/control`, { action: "take" }); + expect(took.status).toBe(200); + expect(took.body.held).toBe(true); + const frame = await sse.until( + (f) => f.kind === "computer-control" && f.botId === botId && f.held === true, + ); + expect(frame.helpReason).toBeNull(); + const hydrated = await api("GET", "/api/bots"); + expect(hydrated.body.computerControl[botId]).toEqual({ held: true, helpReason: null }); + const released = await api("POST", `/api/bots/${botId}/computer/control`, { action: "release" }); + expect(released.body.held).toBe(false); + } finally { + sse.close(); + } + }); + + it("refuses an unknown action and an unknown bot", async () => { + const bad = await api("POST", `/api/bots/${botId}/computer/control`, { action: "hijack" }); + expect(bad.status).toBe(400); + const ghost = await api("GET", "/api/bots/nope/computer/control"); + expect(ghost.status).toBe(404); + }); + + it("refuses a form-shaped POST — control mutations are JSON-only", async () => { + const res = await fetch(`${BASE}/api/bots/${botId}/computer/control`, { + method: "POST", + headers: { "content-type": "application/x-www-form-urlencoded" }, + body: "action=take", + }); + expect(res.status).toBe(415); + }); + + it("keeps the internal who-is-driving endpoint behind the boot token", async () => { + const res = await fetch(`${BASE}/api/internal/computer-control?botId=${botId}`); + expect(res.status).toBe(401); + }); +}); diff --git a/server/index.ts b/server/index.ts index f6e69936f2..0144d31191 100644 --- a/server/index.ts +++ b/server/index.ts @@ -35,6 +35,7 @@ import { EVENTS_DIR, NATIVE_DIR, } from "./config.ts"; +import { ComputerControl } from "./computer-control.ts"; import { augmentedPath, findCliCandidates, resetPathCache } from "./env-path.ts"; import { describeSpawnFailure, execCli } from "./procs.ts"; import { buildNotification, type Notification } from "./notify.ts"; @@ -166,6 +167,22 @@ function connectedAppsIntegration(botId: string, threadId: string) { }); } +// ── computer control (who is driving) ────────────────────────────────── +// The person can take the wheel of a bot's computer from the panel; while +// they hold it, the bot's computer proxies refuse every action. The record +// lives here; the proxies consult it over loopback with the boot token. +const computerControl = new ComputerControl((botId, snapshot) => { + broadcast({ kind: "computer-control", botId, held: snapshot.held, helpReason: snapshot.helpReason }); +}); + +/** The loopback endpoint a bot's computer proxy polls before acting. */ +function controlIntegration(botId: string) { + return { + url: `http://127.0.0.1:${PORT}/api/internal/computer-control?botId=${encodeURIComponent(botId)}`, + token: COMMS_TOKEN, + }; +} + /** Run a turn on `targetBotId` and resolve with its assistant text — the * synchronous half of ask_bot. Subscribes to the bus, folds assistant_text * for that thread, resolves on turn.completed (or a 4-min ceiling). */ @@ -1305,7 +1322,7 @@ async function startTurn( if (!localVm.ready || !localVm.runtime) { throw new Error(`${localVm.problem ?? "the Local VM is not ready"} (App Settings → Local VM)`); } - integrations.localComputer = containerComputerMcp(localVm.runtime); + integrations.localComputer = containerComputerMcp(localVm.runtime, controlIntegration(bot.id)); computerKind = "vm"; } else if (wants === "local") { if (!shouldMountLocalComputer({ @@ -1334,7 +1351,12 @@ async function startTurn( : await vps.reuseVps(cfg, bot.id); if (remote?.ready && remote.sshAlias) { const targetCfg = { ...cfg, vps: { sshAlias: remote.sshAlias } }; - integrations.localComputer = vps.vpsComputerMcp(targetCfg, bot.id, remote.container_id ?? undefined); + const vpsMcp = vps.vpsComputerMcp(targetCfg, bot.id, remote.container_id ?? undefined); + const vpsControl = controlIntegration(bot.id); + integrations.localComputer = { + ...vpsMcp, + env: { ...vpsMcp.env, OMB_CONTROL_URL: vpsControl.url, OMB_CONTROL_TOKEN: vpsControl.token }, + }; computerKind = "vps"; previewCapture = () => vps.vpsComputerScreenshot(targetCfg, bot.id); } else { @@ -1371,7 +1393,12 @@ async function startTurn( if (b) { previewCapture = () => box.screenshotBox(cfg, bot.id, b!.id); if (mountsCloudComputer) { - integrations.computer = { kind: "box", boxId: b.id, token: cfg.box!.token! }; + integrations.computer = { + kind: "box", + boxId: b.id, + token: cfg.box!.token!, + control: controlIntegration(bot.id), + }; computerKind = "box"; } } @@ -2303,6 +2330,32 @@ const server = createServer(async (req, res) => { res.writeHead(upstream.status, headers); return res.end(Buffer.from(upstream.bytes)); } + // ── computer control: proxies read the hold, bots plead for help ── + if (path === "/api/internal/computer-control") { + const botId = url.searchParams.get("botId") ?? ""; + const bot = store.bot(botId); + if (!bot) return json(res, 404, { error: "no such bot" }); + if (method === "GET") { + const snapshot = computerControl.snapshot(botId); + return json(res, 200, { held: snapshot.held, helpOpen: snapshot.helpReason !== null }); + } + if (method === "POST") { + const body = await readBody(req); + const { snapshot, requestId } = computerControl.requestHelpLease(botId, body.reason); + // worth a buzz: the bot is blocked on the person's hands, which + // is exactly the "blocked on you" rule notify.ts encodes + notify( + buildNotification("takeover", bot, bot.threadId, snapshot.helpReason ?? "asked you to take over"), + ); + return json(res, 200, { held: snapshot.held, helpOpen: snapshot.helpReason !== null, requestId }); + } + if (method === "DELETE") { + const body = await readBody(req); + const snapshot = computerControl.expireHelp(botId, body.requestId); + return json(res, 200, { held: snapshot.held, helpOpen: snapshot.helpReason !== null }); + } + return json(res, 405, { error: "method not allowed" }); + } if (method === "POST" && path === "/api/internal/connectors/request") { const body = await readBody(req); const botId = String(body.botId ?? ""); @@ -2485,6 +2538,12 @@ const server = createServer(async (req, res) => { return json(res, 200, { bots: store.bots.map((bot) => ({ ...publicBot(bot), ...messagePage(bot.threadId, limit) })), groups: store.groups.map((g) => ({ ...g, ...messagePage(g.threadId, limit) })), + computerControl: Object.fromEntries( + store.bots.map((bot) => { + const snapshot = computerControl.snapshot(bot.id); + return [bot.id, { held: snapshot.held, helpReason: snapshot.helpReason }]; + }), + ), }); } @@ -3016,6 +3075,7 @@ const server = createServer(async (req, res) => { // now, and its caller would otherwise wait out the 15-minute timeout cancelPeerApprovalsFor(bot.id); discardDelegations(commsBus, bot.threadId); + computerControl.forget(bot.id); store.deleteBot(bot.id); for (const dir of [EVENTS_DIR, NATIVE_DIR]) { try { @@ -3605,6 +3665,29 @@ const server = createServer(async (req, res) => { ? json(res, 200, { backend: "vps", ...(await vps.vpsComputerStatus(cfg, bot.id)) }) : json(res, 200, { backend: "box", ...(await box.boxStatus(cfg, bot.id)) }); } + // Who is driving this bot's computer. GET is the panel's initial read; + // POST take/release/dismiss-help are the person's three moves. The bot + // has no verb here at all — its only voice is the internal help plea. + m = path.match(/^\/api\/bots\/([\w-]+)\/computer\/control$/); + if (m) { + const bot = store.bot(m[1]); + if (!bot) return json(res, 404, { error: "no such bot" }); + if (method === "GET") return json(res, 200, computerControl.snapshot(bot.id)); + if (method === "POST") { + // JSON-only for the same anti-form-POST reason as every other + // computer mutation below. + if (!String(req.headers["content-type"] ?? "").toLowerCase().startsWith("application/json")) { + return json(res, 415, { error: "content-type must be application/json" }); + } + const body = await readBody(req); + const action = String(body.action ?? ""); + if (action === "take") return json(res, 200, computerControl.take(bot.id)); + if (action === "release") return json(res, 200, computerControl.release(bot.id)); + if (action === "dismiss-help") return json(res, 200, computerControl.dismissHelp(bot.id)); + return json(res, 400, { error: "action must be take, release, or dismiss-help" }); + } + return json(res, 405, { error: "method not allowed" }); + } m = path.match(/^\/api\/bots\/([\w-]+)\/computer\/(provision|join|sleep|exec|screenshot|remove)$/); if (m && method === "POST") { const botId = m[1]; diff --git a/server/mcp-bridge.test.ts b/server/mcp-bridge.test.ts index 33dc8dc5c4..b3f66fbb36 100644 --- a/server/mcp-bridge.test.ts +++ b/server/mcp-bridge.test.ts @@ -4,7 +4,12 @@ // silence PLUS a failed liveness probe does, and traffic always vetoes. import { describe, expect, it, vi } from "vitest"; -import { createInactivityWatchdog, runLivenessProbe } from "./mcp-bridge.ts"; +import { + createGateInterceptor, + createInactivityWatchdog, + createLineSplitter, + runLivenessProbe, +} from "./mcp-bridge.ts"; /** a probe whose answers the test scripts one call at a time */ function scriptedProbe(answers: boolean[]) { @@ -122,3 +127,102 @@ describe("runLivenessProbe", () => { ).resolves.toBe(false); }); }); + +describe("createLineSplitter", () => { + it("reassembles lines across arbitrary chunk boundaries", () => { + const lines: string[] = []; + const splitter = createLineSplitter((line) => lines.push(line)); + splitter.push('{"a"'); + splitter.push(':1}\n{"b":2}\n{"c"'); + expect(lines).toEqual(['{"a":1}', '{"b":2}']); + splitter.flush(); + expect(lines).toEqual(['{"a":1}', '{"b":2}', '{"c"']); + }); + + it("does not corrupt a UTF-8 character split between buffers", () => { + const lines: string[] = []; + const splitter = createLineSplitter((line) => lines.push(line)); + const bytes = Buffer.from('{"text":"mouse 🐭"}\n'); + const splitAt = bytes.indexOf(Buffer.from("🐭")) + 2; + splitter.push(bytes.subarray(0, splitAt)); + splitter.push(bytes.subarray(splitAt)); + splitter.flush(); + expect(lines).toEqual(['{"text":"mouse 🐭"}']); + }); +}); + +describe("createGateInterceptor", () => { + const frame = (method: string, id?: number) => JSON.stringify({ jsonrpc: "2.0", id, method, params: {} }); + const drain = () => new Promise((resolve) => setTimeout(resolve, 0)); + + function harness(isHeld: () => Promise) { + const forwarded: string[] = []; + const refused: string[] = []; + const intercept = createGateInterceptor({ + isHeld, + forward: (line) => forwarded.push(line), + refuse: (line) => refused.push(line), + }); + return { forwarded, refused, intercept }; + } + + it("forwards everything untouched while nobody is driving", async () => { + const { forwarded, refused, intercept } = harness(async () => false); + for (const line of [frame("initialize", 1), frame("tools/list", 2), frame("tools/call", 3), "not json at all"]) { + intercept(line); + } + await drain(); + expect(refused).toEqual([]); + expect(forwarded).toHaveLength(4); + // byte-for-byte: the transparent path must not re-serialize a frame + expect(forwarded[3]).toBe("not json at all"); + }); + + it("refuses only tools/call while the person is driving", async () => { + const { forwarded, refused, intercept } = harness(async () => true); + intercept(frame("tools/list", 1)); + intercept(frame("tools/call", 2)); + await drain(); + expect(forwarded).toEqual([frame("tools/list", 1)]); + expect(refused).toHaveLength(1); + const answer = JSON.parse(refused[0]!); + expect(answer.id).toBe(2); + expect(answer.result.isError).toBe(true); + expect(answer.result.content[0].text).toMatch(/taken control/i); + }); + + it("preserves protocol order even though the held-check is async", async () => { + const order: string[] = []; + let calls = 0; + let releaseFirst!: (held: boolean) => void; + const first = new Promise((resolve) => (releaseFirst = resolve)); + let drained!: () => void; + const allForwarded = new Promise((resolve) => (drained = resolve)); + const intercept = createGateInterceptor({ + isHeld: () => (calls++ === 0 ? first : Promise.resolve(false)), + forward: (line) => { + const parsed = JSON.parse(line); + order.push(`fwd:${parsed.id ?? parsed.marker}`); + if (parsed.marker === "drained") drained(); + }, + refuse: (line) => order.push(`ref:${JSON.parse(line).id}`), + }); + intercept(frame("tools/call", 1)); + intercept(frame("tools/call", 2)); + intercept(JSON.stringify({ marker: "drained" })); + expect(order).toEqual([]); + releaseFirst(false); + await allForwarded; + expect(order).toEqual(["fwd:1", "fwd:2", "fwd:drained"]); + }); + + it("fails open: a broken held-check forwards rather than wedging the computer", async () => { + const { forwarded, refused, intercept } = harness(async () => { + throw new Error("harness went away"); + }); + intercept(frame("tools/call", 1)); + await drain(); + expect(refused).toEqual([]); + expect(forwarded).toHaveLength(1); + }); +}); diff --git a/server/mcp-bridge.ts b/server/mcp-bridge.ts index ee3c9a22af..68ef573fd8 100644 --- a/server/mcp-bridge.ts +++ b/server/mcp-bridge.ts @@ -2,6 +2,15 @@ // (container-mcp.ts for the Local VM, vps-container-mcp.ts for the BYO VPS). // It defines no tools and parses no MCP messages: bytes in, bytes out. // +// The single exception to that transparency is the who-is-driving gate +// (opt-in via `gate`). While the person holds control of this computer in +// the app, a `tools/call` from the agent is answered with a refusal HERE, +// on the near side, and never forwarded — Cua Driver on the far side has +// no concept of a person holding the wheel, so the refusal cannot come +// from anywhere else. Everything that is not a tools/call still passes +// through untouched, and with no gate configured the bridge remains the +// byte-for-byte pipe described above. +// // Two behaviors live here so neither entry point can drift: // 1. Exit without truncation. `process.exit()` in a close/error handler // discards whatever is still buffered on stdout — a final MCP result @@ -12,7 +21,9 @@ // VPS dropping mid-turn leaves the exec silently wedged until the OS // gives up — the harness sees a hung tool call, not an error. import { spawn } from "node:child_process"; +import { StringDecoder } from "node:string_decoder"; +import { CONTROL_REFUSAL_PLAIN, createControlClient } from "./control-client.ts"; import { augmentedPath } from "./env-path.ts"; // 45s of TOTAL silence before the bridge even probes. An MCP session is @@ -119,6 +130,79 @@ export interface BridgeOptions { /** Enables the dead-transport watchdog. Omitted for the Local VM, whose * runtime CLI talks to a local daemon and fails fast on its own. */ liveness?: BridgeLiveness; + /** Enables the who-is-driving gate: the harness's loopback control + * endpoint plus its per-boot token. Absent → fully transparent bridge. */ + gate?: { url: string; token: string }; +} + +/** Collect a byte stream into complete newline-terminated lines. MCP's + * stdio transport is one JSON-RPC frame per line, so line boundaries are + * the only safe place to inspect — or inject — anything. */ +export function createLineSplitter(onLine: (line: string) => void): { + push: (chunk: Buffer | string) => void; + flush: () => void; +} { + let pending = ""; + const decoder = new StringDecoder("utf8"); + return { + push(chunk) { + pending += typeof chunk === "string" ? chunk : decoder.write(chunk); + let newline: number; + while ((newline = pending.indexOf("\n")) !== -1) { + const line = pending.slice(0, newline); + pending = pending.slice(newline + 1); + onLine(line); + } + }, + flush() { + pending += decoder.end(); + if (pending) onLine(pending); + pending = ""; + }, + }; +} + +/** The gate itself, factored free of process wiring so a test can drive it + * with plain strings. Frames are handled on a serialized queue: the + * held-check is async, and answering frame N+1 before frame N would + * reorder the agent's protocol stream. Only a `tools/call` is ever + * refused; every other frame — handshakes, tools/list, notifications, + * lines that are not JSON — passes through untouched. */ +export function createGateInterceptor(options: { + isHeld: () => Promise; + forward: (line: string) => void; + refuse: (line: string) => void; + refusalText?: string; +}): (line: string) => void { + const refusalText = options.refusalText ?? CONTROL_REFUSAL_PLAIN; + let queue: Promise = Promise.resolve(); + return (line: string) => { + queue = queue.then(async () => { + let frame: any = null; + try { + frame = JSON.parse(line); + } catch { + // not a frame this gate understands — never stand between the + // agent and its driver on anything but a recognized tool call + } + if (!frame || frame.method !== "tools/call") { + options.forward(line); + return; + } + const held = await options.isHeld().catch(() => false); + if (!held) { + options.forward(line); + return; + } + options.refuse( + JSON.stringify({ + jsonrpc: "2.0", + id: frame.id ?? null, + result: { content: [{ type: "text", text: refusalText }], isError: true }, + }), + ); + }); + }; } export function runMcpBridge(options: BridgeOptions): void { @@ -130,14 +214,42 @@ export function runMcpBridge(options: BridgeOptions): void { // docker may exit before it drains stdin; pipe() leaves this error unhandled. child.stdin.on("error", () => {}); - process.stdin.pipe(child.stdin); - child.stdout.pipe(process.stdout); child.stderr.pipe(process.stderr); - const detach = () => { - process.stdin.unpipe(child.stdin); - process.stdin.pause(); - }; + let detach: () => void; + if (options.gate) { + const client = createControlClient({ url: options.gate.url, token: options.gate.token }); + const inbound = createLineSplitter( + createGateInterceptor({ + isHeld: async () => (await client.state(true)).held, + forward: (line) => child.stdin.write(line + "\n"), + refuse: (line) => process.stdout.write(line + "\n"), + }), + ); + const onStdin = (chunk: Buffer) => inbound.push(chunk); + process.stdin.on("data", onStdin); + process.stdin.on("end", () => { + inbound.flush(); + child.stdin.end(); + }); + // Injected refusals must never land inside one of the child's + // half-written frames, so the child's stdout is re-emitted at line + // granularity as well. + const outbound = createLineSplitter((line) => process.stdout.write(line + "\n")); + child.stdout.on("data", (chunk) => outbound.push(chunk)); + child.stdout.on("end", () => outbound.flush()); + detach = () => { + process.stdin.off("data", onStdin); + process.stdin.pause(); + }; + } else { + process.stdin.pipe(child.stdin); + child.stdout.pipe(process.stdout); + detach = () => { + process.stdin.unpipe(child.stdin); + process.stdin.pause(); + }; + } let watchdog: WatchdogHandle | null = null; if (options.liveness) { diff --git a/server/notify.ts b/server/notify.ts index c3ba7fb287..c877ff066c 100644 --- a/server/notify.ts +++ b/server/notify.ts @@ -10,7 +10,7 @@ // listening decides what to do with it — a desktop notification today, an // APNs push to a paired phone once that exists. -export type NotifyKind = "approval" | "question" | "done" | "routine-failed"; +export type NotifyKind = "approval" | "question" | "done" | "routine-failed" | "takeover"; export interface Notification { kind: NotifyKind; @@ -56,9 +56,11 @@ export function buildNotification( ? `${bot.name} needs approval` : kind === "question" ? `${bot.name} has a question` - : kind === "routine-failed" - ? `${bot.name}'s routine failed` - : `${bot.name} finished`; + : kind === "takeover" + ? `${bot.name} needs your hands` + : kind === "routine-failed" + ? `${bot.name}'s routine failed` + : `${bot.name} finished`; // A "finished" with nothing to say is not worth a notification — the // badge in the sidebar already carries that much. diff --git a/server/vps-container-mcp.ts b/server/vps-container-mcp.ts index b7808f325a..7da24fe628 100644 --- a/server/vps-container-mcp.ts +++ b/server/vps-container-mcp.ts @@ -16,6 +16,11 @@ try { process.exit(2); } +// The who-is-driving pair rides in env, not argv — argv is world-readable +// through `ps`, and the token guards a loopback endpoint. +const controlUrl = process.env.OMB_CONTROL_URL ?? ""; +const controlToken = process.env.OMB_CONTROL_TOKEN ?? ""; + runMcpBridge({ command: "docker", args, @@ -24,4 +29,5 @@ runMcpBridge({ // driver: a busy desktop mid-tool-call must never look dead, while an // unreachable VPS must, and `docker version` distinguishes exactly that. liveness: { command: "docker", args: vpsDockerArgs(sshAlias, ["version", "--format", "{{.Server.Version}}"]) }, + ...(controlUrl && controlToken ? { gate: { url: controlUrl, token: controlToken } } : {}), }); diff --git a/src/components/ComputerPanel.tsx b/src/components/ComputerPanel.tsx index 18e4fd585f..8349ce757a 100644 --- a/src/components/ComputerPanel.tsx +++ b/src/components/ComputerPanel.tsx @@ -9,6 +9,7 @@ import { CalendarDays, CalendarClock, ExternalLink, + Hand, Loader2, Monitor, Moon, @@ -98,6 +99,7 @@ export function ComputerPanel({ bot }: { bot: Bot }) { const [vmFrame, setVmFrame] = useState(null); const [localFrame, setLocalFrame] = useState(null); const [pending, setPending] = useState<"join" | "sleep" | "provision" | null>(null); + const [controlPending, setControlPending] = useState(false); const [error, setError] = useState(null); const [creatingRoutine, setCreatingRoutine] = useState(false); const [panelView, setPanelView] = useState<"computer" | "android">("computer"); @@ -393,6 +395,42 @@ export function ComputerPanel({ bot }: { bot: Bot }) { ? cloudFrame && `data:${cloudFrame.mime};base64,${cloudFrame.png}` : null; + // who-is-driving: SSE keeps this fresh; the mount fetch covers a panel + // opened after the last frame (e.g. an app reload mid-hold) + const control = state.computerControl[bot.id] ?? { held: false, helpReason: null }; + useEffect(() => { + let alive = true; + api(`/api/bots/${bot.id}/computer/control`) + .then((snap) => { + if (!alive) return; + dispatch({ + type: "computerControl", + botId: bot.id, + held: snap.held === true, + helpReason: typeof snap.helpReason === "string" ? snap.helpReason : null, + }); + }) + .catch(() => {}); + return () => { + alive = false; + }; + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [bot.id]); + const controlAction = (action: "take" | "release" | "dismiss-help") => { + setControlPending(true); + api(`/api/bots/${bot.id}/computer/control`, { method: "POST", body: JSON.stringify({ action }) }) + .then((snap) => + dispatch({ + type: "computerControl", + botId: bot.id, + held: snap.held === true, + helpReason: typeof snap.helpReason === "string" ? snap.helpReason : null, + }), + ) + .catch((e) => setError(e.message)) + .finally(() => setControlPending(false)); + }; + const run = (kind: "join" | "sleep" | "provision") => { setPending(kind); setError(null); @@ -589,9 +627,72 @@ export function ComputerPanel({ bot }: { bot: Bot }) { )} + {/* Who is driving — take the wheel / hand it back */} + {(phase === "ready" || phase === "vm") && control.helpReason && !control.held && ( +
+
+ {bot.name} asked for your hands: {control.helpReason} +
+
+ + +
+
+ )} + {(phase === "ready" || phase === "vm") && control.held && ( +
+
+ You have the wheel — the bot's clicks and keystrokes are refused until you hand it back. + {phase === "ready" && cloudBackend === "box" && " Use Open desktop to drive."} +
+ +
+ )} + {phase === "vm" && !control.held && !control.helpReason && ( + + )} {/* Cloud-only actions */} {phase === "ready" && (
+ {!control.held && !control.helpReason && ( + + )} {cloudBackend === "box" && (