codex-chief-of-staff/test/monitor.test.ts

import assert from "node:assert/strict";
import test from "node:test";

import { setTimeout as delay } from "node:timers/promises";

import { AppServerClient } from "../src/app-server-client.js";
import { ThreadMonitor } from "../src/monitor.js";
import { FakeAppServer, rawThread, successfulHandshake } from "./fake-app-server.js";

async function until(predicate: () => boolean, timeoutMs = 500): Promise<void> {
  const deadline = Date.now() + timeoutMs;
  while (!predicate() && Date.now() < deadline) await delay(1);
  assert.equal(predicate(), true, "condition was not reached before timeout");
}

function appClient(server: FakeAppServer, timeoutMs = 50): AppServerClient {
  return new AppServerClient({
    url: "ws://127.0.0.1:4500",
    requestTimeoutMs: timeoutMs,
    connectTimeoutMs: timeoutMs,
    webSocketFactory: server.factory,
  });
}

function inventoryServer(id: string, status = "notLoaded"): FakeAppServer {
  return new FakeAppServer(successfulHandshake((request, server) => {
    if (request.method === "thread/list") {
      server.respond(request, {
        data: [rawThread(id, { status: { type: status } })],
        nextCursor: null,
        backwardsCursor: null,
      });
    } else if (request.method === "thread/loaded/list") {
      server.respond(request, { data: status === "notLoaded" ? [] : [id], nextCursor: null });
    }
  }));
}

test("startup snapshot stays unavailable until both inventories complete", async () => {
  let loadedRequest: Record<string, unknown> | null = null;
  const server = new FakeAppServer(successfulHandshake((request, fake) => {
    if (request.method === "thread/list") {
      fake.respond(request, { data: [rawThread("a")], nextCursor: null, backwardsCursor: null });
    } else if (request.method === "thread/loaded/list") {
      loadedRequest = request;
    }
  }));
  const monitor = new ThreadMonitor({ clientFactory: () => appClient(server), reconnectBaseMs: 1 });
  monitor.start();
  await until(() => loadedRequest !== null);
  assert.equal(monitor.snapshot().snapshotAvailable, false);
  server.respond(loadedRequest as unknown as Record<string, unknown>, { data: [], nextCursor: null });
  assert.deepEqual((await monitor.waitUntilSynchronized()).threads.map((thread) => thread.id), ["a"]);
  await monitor.stop();
});

test("startup failure retries and recovers with a current snapshot", async () => {
  const unavailable = new FakeAppServer(() => undefined, false);
  const recovered = inventoryServer("recovered", "idle");
  const servers = [unavailable, recovered];
  const monitor = new ThreadMonitor({
    clientFactory: () => appClient(servers.shift() ?? recovered, 10),
    reconnectBaseMs: 1,
    reconnectMaxMs: 2,
    sleep: async () => undefined,
  });
  monitor.start();
  const snapshot = await monitor.waitUntilSynchronized(500);
  assert.equal(snapshot.freshness, "current");
  assert.equal(snapshot.threads[0]?.id, "recovered");
  await monitor.stop();
});

test("disconnect retains stale cache and reconnect atomically replaces it", async () => {
  const first = inventoryServer("old", "idle");
  const second = inventoryServer("new", "active");
  const servers = [first, second];
  let releaseBackoff: (() => void) | null = null;
  const backoff = new Promise<void>((resolve) => { releaseBackoff = resolve; });
  let slept = false;
  const monitor = new ThreadMonitor({
    clientFactory: () => appClient(servers.shift() ?? second),
    reconnectBaseMs: 1,
    sleep: async () => {
      if (!slept) {
        slept = true;
        await backoff;
      }
    },
  });
  monitor.start();
  await monitor.waitUntilSynchronized();
  first.disconnect();
  await until(() => monitor.snapshot().connectionState === "disconnected");
  assert.equal(monitor.snapshot().freshness, "stale");
  assert.deepEqual(monitor.snapshot().threads.map((thread) => thread.id), ["old"]);
  (releaseBackoff as unknown as () => void)();
  await until(() => monitor.snapshot().freshness === "current"
    && monitor.snapshot().threads[0]?.id === "new");
  assert.deepEqual(monitor.snapshot().threads.map((thread) => thread.id), ["new"]);
  await monitor.stop();
});

test("bounded wait refresh detects a missed status notification", async () => {
  let status = "idle";
  let listCalls = 0;
  const server = new FakeAppServer(successfulHandshake((request, fake) => {
    if (request.method === "thread/list") {
      listCalls += 1;
      fake.respond(request, {
        data: [rawThread("a", { status: { type: status } })],
        nextCursor: null,
        backwardsCursor: null,
      });
    } else if (request.method === "thread/loaded/list") {
      fake.respond(request, { data: ["a"], nextCursor: null });
    }
  }));
  const monitor = new ThreadMonitor({
    clientFactory: () => appClient(server),
    waitRefreshIntervalMs: 5,
    refreshTimeoutMs: 50,
  });
  monitor.start();
  await monitor.waitUntilSynchronized();
  const revision = monitor.snapshot().revision;
  status = "active";
  const result = await monitor.waitForChange(revision, 100);
  assert.equal(result.timedOut, false);
  assert.equal(monitor.snapshot().threads[0]?.status, "active");
  assert.equal(listCalls, 2);
  await monitor.stop();
});