feat(server-mode): pane presence and typing audit (sharing phase 4)

A pane's tab shows the other people who have it open (pane-presence.js,
in memory, fed by the /ws-pty attach; GET /api/lanes/:id/presence and the
pane_presence broadcast to the owner and full collaborators) and says
"Billed to <name>" for a pane a collaborator opened (startedBy on the run).
The first real keystroke of each attach into a shared workspace's pane by
anyone but its owner is audited as `type into <run id>` (mouse, focus and
terminal replies to tmux do not count). The pty socket now pings, so a
vanished browser is ended at the next re-check and leaves the presence list.

server/__tests__/pane-presence.test.js

@@ -0,0 +1,128 @@
+/**
+ * @file pane-presence.test.js
+ * @description Who has which pane open (server/lib/pane-presence.js, sharing
+ * phase 4): a user counted once however many attaches they hold, gone with
+ * their last one, listed per workspace; the first keystroke into a SHARED
+ * workspace's pane is audited (who and where, never what) and an unshared
+ * one's is not; `pane_presence` reaches the workspace's owner and the users
+ * it is shared with, nobody else.
+ * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
+ */
+
+const { describe, it, before, after, beforeEach } = require("node:test");
+const assert = require("node:assert/strict");
+const path = require("node:path");
+const fs = require("node:fs");
+const os = require("node:os");
+
+const TMP = fs.mkdtempSync(path.join(os.tmpdir(), "pane-presence-test-"));
+process.env.DASHBOARD_DB_PATH = path.join(TMP, "dashboard.db");
+
+const { db } = require("../db");
+const presence = require("../lib/pane-presence");
+const scope = require("../lib/scope");
+
+let alice;
+let bob;
+let carol;
+let lane;
+let lonely;
+
+before(() => {
+  const user = (name) =>
+    Number(
+      db
+        .prepare("INSERT INTO users (username, password_hash, role) VALUES (?, 'x', 'developer')")
+        .run(name).lastInsertRowid
+    );
+  alice = user("alice");
+  bob = user("bob");
+  carol = user("carol");
+  const mk = () =>
+    Number(
+      db
+        .prepare("INSERT INTO lanes (cwd, tool, owner_id) VALUES (?, 'claude', ?)")
+        .run(`/w/${Math.random()}`, alice).lastInsertRowid
+    );
+  lane = mk();
+  lonely = mk();
+  db.prepare("INSERT INTO lane_shares (lane_id, user_id) VALUES (?, ?)").run(lane, bob);
+});
+
+after(() => fs.rmSync(TMP, { recursive: true, force: true }));
+beforeEach(() => presence.__reset());
+
+describe("pane presence", () => {
+  it("counts a user once per pane, until their last attach closes", () => {
+    const run = `agenthub-lane-${lane}-1`;
+    presence.join(run, { id: bob, username: "bob", displayName: "Bob B" });
+    presence.join(run, { id: bob, username: "bob" });
+    presence.join(run, { id: alice, username: "alice" });
+    assert.deepEqual(presence.forLane(lane)[run], [
+      { id: bob, name: "Bob B" },
+      { id: alice, name: "alice" },
+    ]);
+    presence.leave(run, { id: bob });
+    assert.equal(presence.forLane(lane)[run].length, 2, "one of bob's attaches is still open");
+    presence.leave(run, { id: bob });
+    presence.leave(run, { id: alice });
+    assert.deepEqual(presence.forLane(lane), {});
+    presence.join("not-a-run", { id: bob });
+    assert.deepEqual(presence.users("not-a-run"), []);
+  });
+
+  it("audits the first keystroke only in a shared workspace", () => {
+    const count = () =>
+      db.prepare("SELECT COUNT(*) AS n FROM audit_log WHERE path LIKE 'type into %'").get().n;
+    presence.auditInput(`agenthub-lane-${lonely}`, { id: alice });
+    assert.equal(count(), 0);
+    presence.auditInput(`agenthub-lane-${lane}-2`, { id: bob });
+    const row = db
+      .prepare(
+        "SELECT user_id, owner_id, method, path FROM audit_log WHERE path LIKE 'type into %'"
+      )
+      .get();
+    assert.deepEqual(row, {
+      user_id: bob,
+      owner_id: alice,
+      method: "PTY",
+      path: `type into agenthub-lane-${lane}-2`,
+    });
+  });
+
+  it("is news for the workspace's owner and collaborators only", () => {
+    const aud = scope.broadcastAudience("pane_presence", { laneId: lane, runId: "x", users: [] });
+    const dev = (id) => ({ id, role: "developer" });
+    assert.equal(scope.audienceIncludes(aud, dev(alice)), true);
+    assert.equal(scope.audienceIncludes(aud, dev(bob)), true);
+    assert.equal(scope.audienceIncludes(aud, dev(carol)), false);
+  });
+});
+
+describe("what counts as typing", () => {
+  const { isTyping } = require("../lib/pty-attach");
+  it("ignores mouse, focus and the terminal answering tmux", () => {
+    for (const s of [
+      "\x1b[<0;10;5M",
+      "\x1b[I",
+      "\x1b[>0;276;0c",
+      "\x1b[?1;2c",
+      "\x1b[12;40R",
+      "\x1b[0n",
+      "\x1b[?2004;1$y",
+      "\x1b]11;rgb:1e1e/1e1e/1e1e\x1b\\",
+      "\x1b]10;rgb:ffff/ffff/ffff\x07",
+      "\x1bP>|xterm.js(6.0.0)\x1b\\",
+    ]) {
+      assert.equal(isTyping(s), false, JSON.stringify(s));
+    }
+    assert.equal(isTyping("ls\r"), true);
+    assert.equal(isTyping("\x1b[A"), true, "an arrow key is typing");
+  });
+
+  it("never audits the owner's own typing", () => {
+    const before = db.prepare("SELECT COUNT(*) AS n FROM audit_log").get().n;
+    presence.auditInput(`agenthub-lane-${lane}`, { id: alice });
+    assert.equal(db.prepare("SELECT COUNT(*) AS n FROM audit_log").get().n, before);
+  });
+});

server/lib/dashboard-runs.js

@@ -53,7 +53,7 @@ const updateStmt = db.prepare(`
 
 const RUN_COLUMNS = `id, session_id, mode, cwd, model, permission_mode, effort,
          resume_session_id, prompt_preview, status, exit_code,
-         started_at, ended_at, lane_id, tool, worktree, proxy, account`;
+         started_at, ended_at, lane_id, tool, worktree, proxy, account, started_by`;
 
 const listStmt = db.prepare(`
   SELECT ${RUN_COLUMNS}

server/lib/pane-presence.js

@@ -0,0 +1,111 @@
+/**
+ * @file pane-presence.js
+ * @description Who has which pane open right now (server mode, workspace
+ * sharing phase 4): every terminal attach (`/ws-pty/:runId`, websocket.js)
+ * adds its logged-in user to its pane, and its close takes them off. In
+ * memory only — a restart drops every attach anyway. A change is broadcast as
+ * `pane_presence` `{laneId, runId, users}` to the workspace's owner and the
+ * users it is shared with; `GET /api/lanes/:id/presence` gives the current
+ * state. The first keystroke of an attach into a SHARED workspace's pane by
+ * anyone but its owner is written to `audit_log` (`type into <run id>`) — who
+ * typed, never what.
+ * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
+ */
+
+/** runId → Map(userId → {id, name, n}) — `n` attaches of that user. */
+const panes = new Map();
+
+const laneOf = (runId) => {
+  const m = /^agenthub-lane-(\d+)(?:-\d+)?$/.exec(String(runId || ""));
+  return m ? Number(m[1]) : null;
+};
+
+const nameOf = (user) => user.displayName || user.display_name || user.username;
+
+function users(runId) {
+  return [...(panes.get(runId)?.values() || [])].map(({ id, name }) => ({ id, name }));
+}
+
+function announce(runId) {
+  try {
+    require("../websocket").broadcast("pane_presence", {
+      laneId: laneOf(runId),
+      runId,
+      users: users(runId),
+    });
+  } catch {
+    /* no websocket server (tests) */
+  }
+}
+
+/** `user` opened pane `runId`. */
+function join(runId, user) {
+  if (!user || laneOf(runId) == null) return;
+  if (!panes.has(runId)) panes.set(runId, new Map());
+  const here = panes.get(runId);
+  const me = here.get(user.id);
+  if (me) {
+    me.n += 1;
+    return;
+  }
+  here.set(user.id, { id: user.id, name: nameOf(user), n: 1 });
+  announce(runId);
+}
+
+/** `user` closed one attach of pane `runId`. */
+function leave(runId, user) {
+  const here = user && panes.get(runId);
+  const me = here?.get(user.id);
+  if (!me) return;
+  me.n -= 1;
+  if (me.n > 0) return;
+  here.delete(user.id);
+  if (!here.size) panes.delete(runId);
+  announce(runId);
+}
+
+/** `{runId: [{id, name}]}` for workspace `laneId`'s panes. */
+function forLane(laneId) {
+  const out = {};
+  for (const runId of panes.keys()) {
+    if (laneOf(runId) === Number(laneId)) out[runId] = users(runId);
+  }
+  return out;
+}
+
+/** Whether workspace `laneId` is shared with anyone, whole or a session of it. */
+function isShared(laneId) {
+  const { db } = require("../db");
+  return Boolean(
+    db
+      .prepare(
+        `SELECT 1 FROM lane_shares WHERE lane_id = ?
+         UNION ALL SELECT 1 FROM session_shares
+          WHERE session_id IN (SELECT id FROM sessions WHERE lane_id = ?) LIMIT 1`
+      )
+      .get(laneId, laneId)
+  );
+}
+
+/**
+ * The first keystroke of an attach: in a shared workspace, write who typed
+ * into which pane (`audit_log`, owner = the workspace's owner). Best effort.
+ */
+function auditInput(runId, user) {
+  const laneId = laneOf(runId);
+  if (!user || laneId == null) return;
+  try {
+    if (!isShared(laneId)) return;
+    const { db } = require("../db");
+    const owner = db.prepare("SELECT owner_id FROM lanes WHERE id = ?").get(laneId)?.owner_id;
+    if (owner === user.id) return; // the owner in their own workspace
+    require("./audit").recordAccountChange(user.id, owner ?? null, "PTY", `type into ${runId}`);
+  } catch (err) {
+    console.warn("[presence] could not audit:", err.message);
+  }
+}
+
+/** Test seam. */
+const __reset = () => panes.clear();
+
+module.exports = { join, leave, forLane, auditInput, users, __reset };

server/lib/pty-attach.js

@@ -82,6 +82,17 @@ const PASTE_HOLD_MS = 700;
 // keystroke first leaves copy-mode (`tmux.leaveCopyMode`), then is written.
 const NOT_TYPING_RE = /\x1b\[<\d+;\d+;\d+[Mm]|\x1b\[M[\s\S]{3}|\x1b\[[IO]/g;
 
+// What the browser's terminal sends back on its own when tmux asks it
+// something on attach (device attributes, cursor/status reports, mode
+// reports, colour queries): no one typed it.
+const TERMINAL_REPLY_RE =
+  /\x1b\[[?>=]?[\d;]*[cRn]|\x1b\[\?[\d;]*\$y|\x1bP[\s\S]*?\x1b\\|\x1b\][\s\S]*?(?:\x07|\x1b\\)/g;
+
+/** Whether `text` from a browser is someone typing (not a mouse, focus or terminal reply). */
+function isTyping(text) {
+  return text.replace(NOT_TYPING_RE, "").replace(TERMINAL_REPLY_RE, "") !== "";
+}
+
 // runId -> { driver, viewers: Set<{ ws, pty, cols, rows }>, scrolled }
 // (`scrolled`: a mouse report reached the pane since the last keystroke — per
 // pane, since one browser may scroll and another type)
@@ -333,4 +344,11 @@ __setSpawnImpl((...args) => require("@lydell/node-pty").spawn(...args));
 
 const __resetGroups = () => groups.clear();
 
-module.exports = { attach, attachBinary, validateRunId, __setSpawnImpl, __resetGroups };
+module.exports = {
+  attach,
+  attachBinary,
+  validateRunId,
+  isTyping,
+  __setSpawnImpl,
+  __resetGroups,
+};

server/lib/pty-run.js

@@ -934,9 +934,25 @@ function publicRun(id, live = tmux.hasSession(id)) {
     worktree: !!row?.worktree,
     startedAt: row?.started_at || null,
     promptPreview: row?.prompt_preview || null,
+    // Server mode: a pane a user the workspace is shared with opened — whose
+    // money it is (session-payers.js). Null = the workspace's owner.
+    startedBy: row?.started_by != null ? starterOf(row.started_by) : null,
   };
 }
 
+function starterOf(userId) {
+  try {
+    const u = require("../db")
+      .db.prepare(
+        "SELECT id, COALESCE(NULLIF(display_name, ''), username) AS name FROM users WHERE id = ?"
+      )
+      .get(userId);
+    return u ? { id: u.id, name: u.name } : null;
+  } catch {
+    return null;
+  }
+}
+
 function getRun(id) {
   if (laneIdFromRunId(id) === null) return null;
   return publicRun(id);

server/lib/scope.js

@@ -514,6 +514,7 @@ function broadcastAudience(type, data) {
         owners: laneOwners(d.lane?.id),
         viewers: laneViewers(d.lane?.id, { sessions: true }),
       };
+    case "pane_presence":
     case "lane_hook_output":
     case "lane_hook_result":
     case "lane_runtime":

server/openapi-extra/lanes-more.js

@@ -738,6 +738,17 @@ const paths = {
       },
     },
   },
+  "/api/lanes/{id}/presence": {
+    get: {
+      tags: ["Lanes"],
+      summary: "Who has which of a workspace's panes open",
+      description:
+        "Server mode (local mode: `{panes: {}}`). `{panes: {<run id>: [{id, name}]}}` — the users with a terminal attached to each pane, from memory; changes arrive as the WebSocket message `pane_presence` `{laneId, runId, users}` (the workspace's owner and the users it is shared with).",
+      operationId: "getLanePresence",
+      parameters: [idParam],
+      responses: { 200: { description: "`{panes}`.", ...json(obj) }, 404: notFound },
+    },
+  },
   "/api/lanes/{id}/sessions/{sid}/shares": {
     get: {
       tags: ["Lanes"],

server/routes/lanes.js

@@ -1521,6 +1521,16 @@ router.put("/:id/shares", sameOriginGuard, (req, res) => {
   if (shares) res.json({ shares });
 });
 
+// Who has which of this workspace's panes open (pane-presence.js): server
+// mode's `{panes: {runId: [{id, name}]}}`; changes come as `pane_presence`.
+router.get("/:id/presence", (req, res) => {
+  const lane = lanesLib.getLane(req.params.id);
+  if (!lane)
+    return res.status(404).json({ error: { code: "ENOLANE", message: "workspace not found" } });
+  if (!require("../lib/auth").isServerMode()) return res.json({ panes: {} });
+  res.json({ panes: require("../lib/pane-presence").forLane(lane.id) });
+});
+
 /** A session of this workspace (`sessions.lane_id`), or null after sending 404. */
 function laneSession(req, res, lane) {
   const row = require("../db")

server/websocket.js

@@ -159,6 +159,33 @@ function initPtyWebSocket(server) {
     if (cookie !== undefined) {
       ws.cookie = cookie;
       ws.runId = runId;
+      // A browser that vanished without a close (a sleeping laptop) answers
+      // no ping: the beat below ends it, and its presence with it.
+      ws.isAlive = true;
+      ws.on("pong", () => {
+        ws.isAlive = true;
+      });
+    }
+    // Server mode: who has this pane open (pane-presence.js), and the first
+    // keystroke of a collaborator-era attach written to the audit log.
+    if (cookie !== undefined) {
+      const presence = require("./lib/pane-presence");
+      const user = userFromRequest({ headers: { cookie } });
+      if (user) {
+        presence.join(runId, user);
+        ws.on("close", () => presence.leave(runId, user));
+        let typed = false;
+        ws.on("message", (data, isBinary) => {
+          if (typed) return;
+          const binary = typeof isBinary === "boolean" ? isBinary : false;
+          const text = data.toString("utf8");
+          if (!binary && text.startsWith('{"type"')) return; // a control frame
+          // A mouse report, a focus change or the terminal answering tmux.
+          if (!require("./lib/pty-attach").isTyping(text)) return;
+          typed = true;
+          presence.auditInput(runId, user);
+        });
+      }
     }
     try {
       attach(ws, runId, size || { cols: 80, rows: 24 });
@@ -221,13 +248,20 @@ function initPtyWebSocket(server) {
     ptyWss.clients.forEach((ws) => {
       if (ws.cookie === undefined) return;
       const user = userFromRequest({ headers: { cookie: ws.cookie } });
-      if (!user || !canSeeRun({ user, method: "POST" }, ws.runId)) {
+      if (!user || !canSeeRun({ user, method: "POST" }, ws.runId) || ws.isAlive === false) {
         try {
           ws.close(1008, "log in again");
         } catch {
           /* already gone */
         }
         ws.terminate();
+        return;
+      }
+      ws.isAlive = false;
+      try {
+        ws.ping();
+      } catch {
+        /* closing */
       }
     });
   }, PTY_RECHECK_MS);