AgentHub/server/__tests__/read-pool.test.js

/**
 * @file Tests server/lib/read-pool.js: a read job runs in the worker on its own
 * connection, identical calls in flight share one run, a job over the timeout
 * is refused with EREADTIMEOUT and the next one gets a fresh worker,
 * AGENTHUB_READ_WORKER=0 runs the job inline, `maxAgeMs` reuses a young result,
 * the analytics events job matches the main connection, and the WAL checkpoint
 * runs on the worker's writable connection.
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

const { describe, it, before, after, afterEach } = require("node:test");
const assert = require("node:assert/strict");
const fs = require("fs");
const os = require("os");
const path = require("path");

const TEST_DB = path.join(os.tmpdir(), `dashboard-readpool-test-${Date.now()}-${process.pid}.db`);
process.env.DASHBOARD_DB_PATH = TEST_DB;

const { db } = require("../db");
const { runRead } = require("../lib/read-pool");

// The pool unrefs its worker (a server must be able to exit while one idles),
// so nothing else would keep this file's event loop alive while a job runs:
// node:test then cancels the awaiting test ("Promise resolution is still
// pending but the event loop has already resolved"). A server never sees this.
let keepAlive;

before(() => {
  keepAlive = setInterval(() => {}, 1000);
  db.prepare("INSERT INTO sessions (id, status) VALUES ('rp-1', 'completed')").run();
  const ins = db.prepare(
    "INSERT INTO events (session_id, event_type, tool_name) VALUES ('rp-1', 'PreToolUse', ?)"
  );
  for (const t of ["Read", "Edit", "Read", "Edit"]) ins.run(t);
});

afterEach(() => {
  delete process.env.AGENTHUB_READ_TIMEOUT_MS;
  delete process.env.AGENTHUB_READ_WORKER;
});

after(() => {
  clearInterval(keepAlive);
  try {
    db.close();
  } catch {}
  for (const s of ["", "-wal", "-shm"]) fs.rmSync(TEST_DB + s, { force: true });
});

describe("read-pool", () => {
  it("runs a job in the worker and sees the main connection's committed rows", async () => {
    const data = await runRead("workflows", { sections: ["toolFlow"] });
    assert.deepEqual(data.toolFlow.transitions.map((t) => [t.source, t.target, t.value]).sort(), [
      ["Edit", "Read", 1],
      ["Read", "Edit", 2],
    ]);
  });

  it("shares one run between identical calls in flight", async () => {
    const a = runRead("workflows", { sections: ["stats"] });
    const b = runRead("workflows", { sections: ["stats"] });
    assert.equal(a, b);
    assert.notEqual(a, runRead("workflows", { sections: ["toolFlow"] }));
    await Promise.all([a, b]);
  });

  it("refuses a job over the timeout, then recovers with a fresh worker", async () => {
    process.env.AGENTHUB_READ_TIMEOUT_MS = "1";
    await assert.rejects(runRead("workflows", { tag: "slow" }), { code: "EREADTIMEOUT" });
    delete process.env.AGENTHUB_READ_TIMEOUT_MS;
    const data = await runRead("workflows", { sections: ["stats"] });
    assert.equal(data.stats.totalSessions, 1);
  });

  it("reuses a result younger than maxAgeMs, and only for the same args", async () => {
    const first = await runRead(
      "workflows",
      { sections: ["stats"], tag: "memo" },
      { maxAgeMs: 60_000 }
    );
    db.prepare("INSERT INTO sessions (id, status) VALUES ('rp-2', 'completed')").run();
    const again = await runRead(
      "workflows",
      { sections: ["stats"], tag: "memo" },
      { maxAgeMs: 60_000 }
    );
    assert.equal(again, first); // the cached object, not a new run
    const fresh = await runRead("workflows", { sections: ["stats"], tag: "memo" });
    assert.equal(fresh.stats.totalSessions, first.stats.totalSessions + 1);
    db.prepare("DELETE FROM sessions WHERE id = 'rp-2'").run();
  });

  it("computes the analytics events numbers in the worker like the main connection", async () => {
    const { eventAggregates } = require("../lib/scoped-stats");
    const args = { scope: null, range: null, tzModifier: "+0 minutes" };
    const inWorker = await runRead("analyticsEvents", args);
    assert.deepEqual(inWorker, eventAggregates(db, null, null, "+0 minutes"));
    assert.deepEqual(inWorker.toolUsage.map((r) => [r.tool_name, r.count]).sort(), [
      ["Edit", 2],
      ["Read", 2],
    ]);
  });

  it("checkpoints the WAL from the worker", async () => {
    const [row] = await runRead("walCheckpoint");
    assert.equal(row.busy, 0);
    assert.ok(row.log >= row.checkpointed);
  });

  it("runs inline with AGENTHUB_READ_WORKER=0, and rejects an unknown job", async () => {
    process.env.AGENTHUB_READ_WORKER = "0";
    const data = await runRead("workflows", { sections: ["stats"], tag: "inline" });
    assert.equal(data.stats.topFlow.source, "Read");
    await assert.rejects(runRead("nope"), { code: "EBADJOB" });
  });
});