The bot's browser keeps up with your mouse: input on one socket, the newest move wins

cloud/browser-probe.ts

@@ -0,0 +1,11 @@
+// Opt-in timings for the local browser latency probe. Never log input text or credentials.
+export function probeTiming(phase: string, detail: Record<string, unknown>) {
+  if (process.env.BOPS_BROWSER_PROBE === "1")
+    console.log(`[browser-probe] ${JSON.stringify({ phase, at: performance.timeOrigin + performance.now(), ...detail })}`);
+}
+
+export function probeInput(phase: string, method: unknown, params: unknown) {
+  if (process.env.BOPS_BROWSER_PROBE !== "1" || method !== "Input.dispatchMouseEvent" || !params || typeof params !== "object") return;
+  const p = params as Record<string, unknown>;
+  probeTiming(phase, { type: p.type, x: p.x, y: p.y, deltaY: p.deltaY });
+}

cloud/http.ts

@@ -71,8 +71,9 @@ export class HttpError extends Error {
   }
 }
 
-/** Refuse a WebSocket upgrade with an HTTP status. */
-export function refuseUpgrade(socket: Duplex, status: number, message: string) {
-  socket.write(`HTTP/1.1 ${status} ${message}\r\nConnection: close\r\nContent-Length: 0\r\n\r\n`);
+/** Refuse a WebSocket upgrade with an HTTP status (and `headers`: names and values of letters, digits and dashes). */
+export function refuseUpgrade(socket: Duplex, status: number, message: string, headers: Record<string, string> = {}) {
+  const extra = Object.entries(headers).map(([k, v]) => `${k}: ${v}\r\n`).join("");
+  socket.write(`HTTP/1.1 ${status} ${message}\r\nConnection: close\r\n${extra}Content-Length: 0\r\n\r\n`);
   socket.destroy();
 }

cloud/live.ts

@@ -555,6 +555,17 @@ function chatsOf(blob: Record<string, unknown>): Map<string, string> {
   return chats;
 }
 
+/**
+ * What else hears of each write of an account's state, with the blob as written, before anything below: the bot's
+ * browser asks again who may use it (browser.ts), as a per-bot level or a bot's computer lives only in the owner's
+ * state and orgo-web sends no notice for it. Kept here, not imported, so nothing that writes the state imports browser.ts.
+ */
+const writeHooks = new Set<(account: string, blob: Record<string, unknown>) => void>();
+export function onStateWritten(f: (account: string, blob: Record<string, unknown>) => void): () => void {
+  writeHooks.add(f);
+  return () => void writeHooks.delete(f);
+}
+
 /**
  * The user's state was written (`by` the device that wrote it): every watched workspace whose part of it
  * changed hears "state", with the fingerprint of what it shows (sliceHash, and each member's stream its
@@ -564,6 +575,13 @@ function chatsOf(blob: Record<string, unknown>): Map<string, string> {
  */
 export function stateWritten(account: string, blob: Record<string, unknown>, by: string | null, h: LiveHub = hub) {
   h.written++;
+  for (const f of writeHooks) {
+    try {
+      f(account, blob);
+    } catch (e) {
+      console.warn(`[live] after a state write: ${(e as Error)?.message ?? e}`);
+    }
+  }
   const watched = h.watched(account);
   if (!watched.length) return void h.seen.delete(account);
   const prev = h.seen.get(account);

cloud/protocol.ts

@@ -1279,11 +1279,13 @@ export type BrowserActivity = { step: "browsing" | "idle"; waiting: boolean; sig
 
 /**
  * What Bops Cloud tells a runner on its socket (WS /v1/browser/runner): who watches each of its bots' browsers now
- * ("watch", the whole list each time; empty: nobody), and what a viewer asked ("take", "release", "input"). A take
+ * ("watch", the whole list each time; empty: nobody; `joined`: a stream just opened, maybe for someone already watching
+ * on another Mac or window, who sees the page at once as someone new does), and what a viewer asked ("take", "release",
+ * "input"). A take
  * by the owner (`over`) takes control from a member who has it; a member's never takes it from anyone.
  */
 export type BrowserToRunner =
-  | { t: "watch"; bot: string; viewers: BrowserPerson[] }
+  | { t: "watch"; bot: string; viewers: BrowserPerson[]; joined?: true }
   | { t: "take"; bot: string; who: BrowserPerson; over?: true }
   | { t: "release"; bot: string; who: BrowserPerson }
   | { t: "input"; bot: string; who: BrowserPerson; method: string; params: unknown };
@@ -1319,6 +1321,20 @@ export type BrowserEvent =
 /** POST /v1/browser/act: take control, hand it back, or (in control) one CDP input. `project`: a shared workspace's Orgo id; none for the caller's own bots. */
 export type BrowserAct = { bot: string; project?: string; act: "take" | "release" | "input"; method?: string; params?: unknown };
 
+/**
+ * WS /v1/browser/input?bot=<id>[&project=<Orgo workspace>]: the acts of one viewer's Bops for one bot, in order, on one
+ * socket (lib/server/browser-view.ts), each message a BrowserAct without `bot` and `project` (the address says them), at
+ * most 64 KiB. Checked as the watch stream is: at the upgrade (refused with mayUseBrowser's status), every ROLE_MS while
+ * open and at each write of the owner's state, and ended by orgo-web's notice; each act checked as POST /v1/browser/act
+ * checks it, then sent to the runner of that computer at that moment. Bops Cloud answers only what didn't go
+ * (BrowserInputReply), and pings it; each end pings the other, and one that doesn't answer by the next ping is taken for
+ * gone. Its refusals of the upgrade carry BROWSER_INPUT_HEADER, so a 404 without it is a Bops Cloud from before this socket.
+ */
+export const BROWSER_INPUT_PATH = "/v1/browser/input";
+export const BROWSER_INPUT_HEADER = "x-bops-browser-input";
+export type BrowserInputAct = Omit<BrowserAct, "bot" | "project">;
+export type BrowserInputReply = { t: "error"; code: string; why: string };
+
 /** Why Bops Cloud won't let someone watch a bot's browser (403, with the words to show them). */
 export const BROWSER_NEEDS_WRITE = "browser_needs_write";
 export const BROWSER_NEEDS_COMPUTER = "browser_needs_computer";