chore: v2.8.16 — Oh My Pi login sharing: a pane's key is listed before it starts

omp's gateway re-reads its clients file only every 10 s, and an omp that starts while its key is refused fails model discovery once and hangs for good. omp-share.js now keeps four keys queued in the clients file (hashes only; the keys in <data dir>/omp-share/tokens.json, 0600): a new pane takes one already listed (AGENTHUB_OMP_KEY), a fresh one joins the queue, and a gone pane's key leaves at the next sync.
2026-10-22 10:29:55 +07:00
parent 750d74a2af
commit 4eb7734ab9
8 changed files with 175 additions and 100 deletions
@@ -2,9 +2,10 @@
 * @file omp-share.test.js
 * @description Oh My Pi login sharing (server/lib/omp-share.js): a member's
 * `models.yml` gets one `agenthub` provider pointing at the loopback gateway
- * with the pane token's env var NAME as its key (the rest of the file kept, a
- * non-mapping refused); the gateway's clients file holds only LIVE omp panes'
- * token hashes; a spawn is untouched while sharing is off, for a host pane, or
+ * with the key's env var NAME as its key (the rest of the file kept, a
+ * non-mapping refused); a pane takes a key the clients file listed BEFORE it
+ * started (the gateway re-reads it only every 10 s), and a gone pane's key
+ * leaves it; a spawn is untouched while sharing is off, for a host pane, or
 * without S3; and the route refuses outside S3. No omp daemon is started.
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */
@@ -23,13 +24,15 @@ process.env.CLAUDE_HOME = path.join(TMP, ".claude");
 process.env.DASHBOARD_DATA_DIR = path.join(TMP, "data");
 process.env.AGENTHUB_OMP_GATEWAY_PORT = "4999";

+const crypto = require("node:crypto");
 const yaml = require("js-yaml");
+
+const sha = (t) => crypto.createHash("sha256").update(t).digest("hex");
 const { createApp, startServer } = require("../index");
 const share = require("../lib/omp-share");
 const linux = require("../lib/linux-accounts");
 const runAs = require("../lib/run-as");
 const tmux = require("../lib/tmux");
-const { db } = require("../db");

 const realS3 = linux.s3Enabled;
 const realLive = tmux.liveSessions;
@@ -57,7 +60,7 @@ describe("mergedModels", () => {
      providers: {
        agenthub: {
          baseUrl: "http://127.0.0.1:4999",
-          apiKey: "AGENTHUB_PANE_TOKEN",
+          apiKey: "AGENTHUB_OMP_KEY",
          api: "anthropic-messages",
          authHeader: true,
          discovery: { type: "proxy" },
@@ -72,7 +75,7 @@ describe("mergedModels", () => {
    const doc = yaml.load(share.mergedModels(text));
    assert.equal(doc.providers.mine.baseUrl, "http://h/v1");
    assert.deepEqual(doc.equivalence, { x: "y" });
-    assert.equal(doc.providers.agenthub.apiKey, "AGENTHUB_PANE_TOKEN");
+    assert.equal(doc.providers.agenthub.apiKey, "AGENTHUB_OMP_KEY");
  });

  it("writes nothing when the provider is already there as is", () => {
@@ -113,17 +116,18 @@ describe("forSpawn", () => {
  const member = { login: "ah-mia", home: path.join(TMP, "home-mia") };

  it("is null while sharing is off, for a host pane, or without S3", () => {
-    assert.equal(share.forSpawn(member, "agenthub-lane-1", "h"), null); // off
+    assert.equal(share.forSpawn(member, "agenthub-lane-1"), null); // off
    fs.mkdirSync(path.dirname(flag), { recursive: true });
    fs.writeFileSync(flag, "");
-    assert.equal(share.forSpawn(member, "agenthub-lane-1", "h"), null); // no S3
+    assert.equal(share.forSpawn(member, "agenthub-lane-1"), null); // no S3
    linux.s3Enabled = () => true;
-    assert.equal(share.forSpawn(null, "agenthub-lane-1", "h"), null); // the host's
-    assert.equal(share.forSpawn(member, "agenthub-lane-1", null), null); // no token
+    assert.equal(share.forSpawn(null, "agenthub-lane-1"), null); // the host's
  });

-  it("writes the member's models.yml AS them and lists the pane's token", () => {
+  it("hands out a key the gateway already listed, and writes models.yml AS the member", () => {
    turnOn();
+    share.syncClients(); // the queue, listed before any pane asks
+    const listedBefore = JSON.parse(fs.readFileSync(clientsFile, "utf8"));
    const writes = [];
    const real = { read: runAs.readFileSafe, mkdir: runAs.mkdirAs, write: runAs.writeAs };
    runAs.readFileSafe = () => {
@@ -135,17 +139,27 @@ describe("forSpawn", () => {
    runAs.writeAs = (acct, file, mode, content) => writes.push([mode, acct.login, file, content]);
    tmux.liveSessions = () => new Set();
    try {
-      const s = share.forSpawn(member, "agenthub-lane-7", "a".repeat(64));
-      assert.deepEqual(s.env, { OMP_SKIP_SETUP: "1" });
+      const s = share.forSpawn(member, "agenthub-lane-7");
+      const key = s.env.AGENTHUB_OMP_KEY;
+      assert.match(key, /^ahomp_/);
+      assert.equal(s.env.OMP_SKIP_SETUP, "1");
+      const hash = sha(key);
+      assert.ok(
+        Object.values(listedBefore).some((c) => c.tokenSha256 === hash),
+        "the key was listed before the pane started"
+      );
+      const after = JSON.parse(fs.readFileSync(clientsFile, "utf8"));
+      assert.deepEqual(after["agenthub-lane-7"], { tokenSha256: hash });
+      // its place in the queue is filled again, and no plaintext reaches the clients file
+      assert.equal(Object.keys(after).filter((k) => k.startsWith("standby-")).length, 4);
+      assert.ok(!fs.readFileSync(clientsFile, "utf8").includes("ahomp_"));
+      assert.equal((fs.statSync(clientsFile).mode & 0o777).toString(8), "600");
      s.place();
      const agentDir = path.join(member.home, ".omp", "agent");
      assert.deepEqual(writes[0], ["mkdir", "ah-mia", agentDir]);
      assert.equal(writes[1][0], "replace");
      assert.equal(writes[1][2], path.join(agentDir, "models.yml"));
-      assert.equal(yaml.load(writes[1][3]).providers.agenthub.apiKey, "AGENTHUB_PANE_TOKEN");
-      const clients = JSON.parse(fs.readFileSync(clientsFile, "utf8"));
-      assert.deepEqual(clients["agenthub-lane-7"], { tokenSha256: "a".repeat(64) });
-      assert.equal((fs.statSync(clientsFile).mode & 0o777).toString(8), "600");
+      assert.equal(yaml.load(writes[1][3]).providers.agenthub.apiKey, "AGENTHUB_OMP_KEY");
    } finally {
      runAs.readFileSafe = real.read;
      runAs.mkdirAs = real.mkdir;
@@ -154,29 +168,26 @@ describe("forSpawn", () => {
  });
 });

-describe("liveClients", () => {
-  before(() => {
-    const ins = db.prepare(
-      `INSERT INTO dashboard_runs (id, cwd, status, tool, pane_token_hash) VALUES (?, '/x', ?, ?, ?)`
-    );
-    ins.run("agenthub-lane-21", "running", "omp", "b".repeat(64));
-    ins.run("agenthub-lane-22", "running", "omp", "c".repeat(64)); // pane gone
-    ins.run("agenthub-lane-23", "ended", "omp", "d".repeat(64));
-    ins.run("agenthub-lane-24", "running", "claude", "e".repeat(64));
+describe("pruneTokens", () => {
+  const now = 1_000_000;
+  const tokens = () => ({
+    standby: [],
+    used: {
+      "agenthub-lane-21": { hash: "b", at: now - 120_000 }, // live
+      "agenthub-lane-22": { hash: "c", at: now - 120_000 }, // pane gone
+      "agenthub-lane-23": { hash: "d", at: now - 5_000 }, // just started
+    },
  });

-  it("lists only live omp panes", () => {
-    tmux.liveSessions = () => new Set(["agenthub-lane-21", "agenthub-lane-23", "agenthub-lane-24"]);
-    const c = share.liveClients();
-    assert.deepEqual(c["agenthub-lane-21"], { tokenSha256: "b".repeat(64) });
-    for (const id of ["agenthub-lane-22", "agenthub-lane-23", "agenthub-lane-24"]) {
-      assert.equal(c[id], undefined, id);
-    }
+  it("drops only a gone pane's key past the grace, and keeps the queue full", () => {
+    const t = share.pruneTokens(tokens(), new Set(["agenthub-lane-21"]), now);
+    assert.deepEqual(Object.keys(t.used).sort(), ["agenthub-lane-21", "agenthub-lane-23"]);
+    assert.equal(t.standby.length, 4);
  });

-  it("says nothing when tmux could not be asked", () => {
-    tmux.liveSessions = () => null;
-    assert.equal(share.liveClients(), null);
+  it("drops nothing when tmux could not be asked", () => {
+    const t = share.pruneTokens(tokens(), null, now);
+    assert.equal(Object.keys(t.used).length, 3);
  });
 });

@@ -14,13 +14,14 @@
 *    `/v1/messages`, `/v1/chat/completions`, `/v1/responses`, `/v1/models`
 *    with the broker's credentials put in upstream. Its own unscoped token
 *    stays in the host's home too.
- * A member's omp pane reaches the gateway as a SCOPED client: the clients file
- * lists the SHA-256 of each live omp pane's own pane token
- * (`dashboard_runs.pane_token_hash`, the same hash `auth.js` keeps), and its
- * `~/.omp/agent/models.yml` gets one `agenthub` provider whose `apiKey` is the
- * env var NAME `AGENTHUB_PANE_TOKEN`. A scoped client can call models but
- * never read a credential or the gateway's usage, and its token stops working
- * within the gateway's 10 s re-read once its pane is gone (`syncClients`) — so
+ * A member's omp pane reaches the gateway as a SCOPED client: it is handed one
+ * key (`AGENTHUB_OMP_KEY`, through tmux -e / its launch file) that the
+ * clients file had already listed — by SHA-256 only — before the pane started
+ * (the gateway re-reads that file only every 10 s; see `takeToken`), and its
+ * `~/.omp/agent/models.yml` gets one `agenthub` provider whose `apiKey` is
+ * that env var's NAME. A scoped client can call models but never read a
+ * credential or the gateway's usage, and its key stops working within the
+ * next sync + the gateway's re-read once its pane is gone (`syncClients`) — so
 * an agent can spend through the login while its pane lives, never carry it
 * away. omp's gateway serves only its built-in providers' models (a custom
 * `models.yml` provider of the host, such as a 9Router, is not shared).
@@ -30,6 +31,7 @@
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

+const crypto = require("node:crypto");
 const fs = require("node:fs");
 const os = require("node:os");
 const path = require("node:path");
@@ -37,10 +39,8 @@ const net = require("node:net");
 const { spawn } = require("node:child_process");

 const PROVIDER = "agenthub";
-const TOKEN_ENV = "AGENTHUB_PANE_TOKEN";
+const TOKEN_ENV = "AGENTHUB_OMP_KEY";
 const SYNC_MS = 15_000;
-/** A spawn's token counts before its run row exists (recordRun runs after). */
-const PENDING_MS = 60_000;

 const port = (name, fallback) => {
  const n = parseInt(process.env[name] || "", 10);
@@ -53,6 +53,7 @@ const dir = () => path.join(require("./claude-home").getDataDir(), "omp-share");
 const onFlag = () => path.join(dir(), "enabled");
 const clientsFile = () => path.join(dir(), "clients.json");
 const pidsFile = () => path.join(dir(), "pids.json");
+const tokensFile = () => path.join(dir(), "tokens.json");
 const logFile = (name) => path.join(dir(), `${name}.log`);
 /** The host's own omp config dir, where the broker keeps its token. */
 const ompDir = () => process.env.AGENTHUB_OMP_DIR || path.join(os.homedir(), ".omp");
@@ -67,52 +68,110 @@ function ensureDir() {
  fs.mkdirSync(dir(), { recursive: true, mode: 0o700 });
 }

-// ── the clients file ──────────────────────────────────────────────────────
+// ── tokens and the clients file ───────────────────────────────────────────
+//
+// The gateway re-reads its clients file only every 10 s, and an omp that
+// starts while its key is still refused hangs for good (its model discovery
+// fails once and is never retried). So a pane never gets a key minted for
+// it: it takes one of STANDBY keys that have been listed for a while, and a
+// fresh one takes that key's place in the queue. `tokens.json` (0600, the
+// host's data dir) keeps the queue and which run holds which key, so both
+// survive a restart; a key whose pane is gone leaves the file at the next sync.

-const pending = new Map(); // run id → { hash, at }
-let lastClients = null;
+const STANDBY = 4;
+/** A run's key stays listed this long before its pane must show up in tmux. */
+const GRACE_MS = 60_000;

-/** `{ <run id>: { tokenSha256 } }` for every live omp pane, or null when tmux could not say. */
-function liveClients() {
-  const live = require("./tmux").liveSessions();
-  if (!live) return null;
-  const rows = require("../db")
-    .db.prepare(
-      `SELECT id, pane_token_hash FROM dashboard_runs
-        WHERE tool = 'omp' AND status IN ('running', 'spawning') AND pane_token_hash IS NOT NULL`
-    )
-    .all();
+const sha = (t) => crypto.createHash("sha256").update(t).digest("hex");
+const newToken = () => `ahomp_${crypto.randomBytes(24).toString("base64url")}`;
+
+function readTokens() {
+  try {
+    const j = JSON.parse(fs.readFileSync(tokensFile(), "utf8"));
+    return {
+      standby: Array.isArray(j.standby) ? j.standby.filter((t) => typeof t === "string") : [],
+      used: j.used && typeof j.used === "object" ? j.used : {},
+    };
+  } catch {
+    return { standby: [], used: {} };
+  }
+}
+
+function writeAtomic(file, text) {
+  ensureDir();
+  const tmp = `${file}.${process.pid}.tmp`;
+  fs.writeFileSync(tmp, text, { mode: 0o600 });
+  fs.renameSync(tmp, file);
+}
+
+function refill(t) {
+  while (t.standby.length < STANDBY) t.standby.push(newToken());
+}
+
+/** The gateway's view: every queued key and every key a run holds — hashes only. */
+function clientsFor(t) {
  const out = {};
-  for (const r of rows) if (live.has(r.id)) out[r.id] = { tokenSha256: r.pane_token_hash };
-  const now = Date.now();
-  for (const [id, p] of pending) {
-    if (now - p.at > PENDING_MS) pending.delete(id);
-    else out[id] = { tokenSha256: p.hash };
+  for (const s of t.standby) out[`standby-${sha(s).slice(0, 16)}`] = { tokenSha256: sha(s) };
+  for (const [id, u] of Object.entries(t.used)) {
+    if (u && typeof u.hash === "string") out[id] = { tokenSha256: u.hash };
  }
  return out;
 }

-function writeClients(clients) {
-  const text = `${JSON.stringify(clients, null, 2)}\n`;
-  if (text === lastClients) return;
-  ensureDir();
-  const tmp = `${clientsFile()}.${process.pid}.tmp`;
-  fs.writeFileSync(tmp, text, { mode: 0o600 });
-  fs.renameSync(tmp, clientsFile());
+let lastClients = null;
+
+function save(t) {
+  writeAtomic(tokensFile(), `${JSON.stringify(t)}\n`);
+  const text = `${JSON.stringify(clientsFor(t), null, 2)}\n`;
+  if (text === lastClients && fs.existsSync(clientsFile())) return;
+  writeAtomic(clientsFile(), text);
  lastClients = text;
 }

-/** Rewrite the clients file from the live omp panes (a gone pane's token drops out). */
+/** A key for run `runId`, already listed with the gateway; its place in the queue is refilled. */
+function takeToken(runId) {
+  const t = readTokens();
+  refill(t);
+  const token = t.standby.shift();
+  t.used[runId] = { hash: sha(token), at: Date.now() };
+  refill(t);
+  save(t);
+  return token;
+}
+
+/**
+ * Drop the keys of runs whose pane is gone (after GRACE_MS), keep the queue
+ * full, and rewrite both files. `live`: the tmux session names, or null when
+ * tmux could not say — then nothing is dropped.
+ */
+function pruneTokens(t, live, now = Date.now()) {
+  if (live) {
+    for (const [id, u] of Object.entries(t.used)) {
+      if (!live.has(id) && now - (Number(u?.at) || 0) > GRACE_MS) delete t.used[id];
+    }
+  }
+  refill(t);
+  return t;
+}
+
+/** Rewrite the clients file from the live omp panes (a gone pane's key drops out). */
 function syncClients() {
  if (!enabled() || !available()) return;
  try {
-    const clients = liveClients();
-    if (clients) writeClients(clients);
+    save(pruneTokens(readTokens(), require("./tmux").liveSessions()));
  } catch (err) {
    console.warn("omp share: clients sync failed:", err.message);
  }
 }

+/** Sharing turned off: no key stays listed, queued or held. */
+function revokeAll() {
+  lastClients = null;
+  if (fs.existsSync(tokensFile()) || fs.existsSync(clientsFile())) {
+    save({ standby: [], used: {} });
+  }
+}
+
 // ── a member's models.yml ─────────────────────────────────────────────────

 function providerEntry() {
@@ -200,10 +259,12 @@ function shareModel(requested, hostDefault = hostDefaultModel()) {
 * null when sharing is off (the pane then runs on the member's own login).
 * Only a MEMBER's pane — the host's omp already holds the login itself.
 */
-function forSpawn(member, runId, tokenHash) {
-  if (!member || !tokenHash || !enabled() || !available()) return null;
+function forSpawn(member, runId) {
+  if (!member || !enabled() || !available()) return null;
+  const token = takeToken(runId);
  return {
-    env: { OMP_SKIP_SETUP: "1" },
+    // The key itself, never in argv: tmux -e, or a member's launch file.
+    env: { OMP_SKIP_SETUP: "1", [TOKEN_ENV]: token },
    model: (requested) => shareModel(requested),
    place() {
      try {
@@ -211,8 +272,6 @@ function forSpawn(member, runId, tokenHash) {
      } catch (err) {
        console.warn(`omp share: ${member.login}'s models.yml left as is:`, err.message);
      }
-      pending.set(runId, { hash: tokenHash, at: Date.now() });
-      syncClients();
    },
  };
 }
@@ -328,8 +387,9 @@ async function startDaemons() {
    return;
  }
  killStale();
-  // The gateway refuses to start without the clients file it names.
-  if (!fs.existsSync(clientsFile())) writeClients({});
+  // The gateway refuses to start without the clients file it names, and
+  // must list the queued keys from its first read.
+  save(pruneTokens(readTokens(), null));
  const bport = brokerPort();
  const gport = gatewayPort();
  for (const p of [bport, gport]) {
@@ -422,9 +482,7 @@ async function setEnabled(on) {
    fs.rmSync(onFlag(), { force: true });
    stop();
    // No pane keeps a way in once sharing is off.
-    lastClients = null;
-    pending.clear();
-    if (fs.existsSync(clientsFile())) writeClients({});
+    revokeAll();
  }
  return state();
 }
@@ -458,7 +516,10 @@ module.exports = {
  mergedModels,
  providerEntry,
  shareModel,
-  liveClients,
+  takeToken,
+  pruneTokens,
+  clientsFor,
+  readTokens,
  PROVIDER,
  TOKEN_ENV,
 };
@@ -490,13 +490,12 @@ function spawnRun(args) {
      : null;
  const paneCred = require("./auth").newPaneToken();
  // A member's omp on the host's shared omp login (omp-share.js): its
-  // models.yml points at the loopback gateway, keyed by this pane's token.
+  // models.yml points at the loopback gateway, with a key already listed there.
  const ompShare =
    tool === "omp"
      ? require("./omp-share").forSpawn(
          require("./run-as").accountForLane(laneId),
-          runIdForLane(laneId, slot),
-          paneCred?.hash
+          runIdForLane(laneId, slot)
        )
      : null;
  if (forkSession === true) {