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

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

import { MonitorStore } from "../src/store.js";
import { normalizeThread } from "../src/domain.js";
import { rawThread } from "./fake-app-server.js";

function summary(id: string, status = "notLoaded") {
  return normalizeThread(rawThread(id, { status: { type: status } }));
}

test("snapshot replacement is atomic and advances revision only for changed inventory", () => {
  const store = new MonitorStore();
  store.replaceSnapshot([summary("a")], new Set(["a"]));
  const firstRevision = store.revision;
  assert.deepEqual(store.snapshot().threads.map((thread) => thread.id), ["a"]);
  store.replaceSnapshot([summary("b")], new Set(["b"]));
  assert.deepEqual(store.snapshot().threads.map((thread) => thread.id), ["b"]);
  assert.equal(store.revision, firstRevision + 1);
  store.replaceSnapshot([summary("b")], new Set(["b"]));
  assert.equal(store.revision, firstRevision + 1);
});

test("status events update known and newly observed threads", () => {
  const store = new MonitorStore();
  store.replaceSnapshot([summary("known")], new Set());
  store.updateStatus("known", { type: "active", activeFlags: ["waitingOnApproval"] });
  store.updateStatus("new", { type: "idle", activeFlags: [] });
  const byId = new Map(store.snapshot().threads.map((thread) => [thread.id, thread]));
  assert.equal(byId.get("known")?.status, "active");
  assert.deepEqual(byId.get("known")?.activeFlags, ["waitingOnApproval"]);
  assert.equal(byId.get("new")?.status, "idle");
});

test("change ring stays bounded and reports lost history", () => {
  const store = new MonitorStore({ changeRingSize: 2 });
  store.setConnection("connecting");
  store.setConnection("disconnected", "one");
  store.setConnection("connecting");
  const result = store.changesSince(0, 10);
  assert.equal(result.changes.length, 2);
  assert.equal(result.truncated, true);
});

test("connection loss retains cached snapshot and marks it stale", () => {
  const store = new MonitorStore();
  store.replaceSnapshot([summary("a")], new Set());
  store.setConnection("disconnected", "socket closed");
  const snapshot = store.snapshot();
  assert.equal(snapshot.freshness, "stale");
  assert.equal(snapshot.threads.length, 1);
  assert.equal(snapshot.connectionError, "socket closed");
});

test("waiting observes changes and normal timeouts", async () => {
  const store = new MonitorStore();
  const wait = store.waitForChange(store.revision, 100);
  store.setConnection("connecting");
  const changed = await wait;
  assert.equal(changed.timedOut, false);
  assert.equal(changed.changes[0]?.kind, "connection");
  const timedOut = await store.waitForChange(store.revision, 5);
  assert.equal(timedOut.timedOut, true);
});

test("concurrent waiters are all released and shutdown cancels remaining waits", async () => {
  const store = new MonitorStore();
  const first = store.waitForChange(0, 100);
  const second = store.waitForChange(0, 100);
  store.setConnection("connecting");
  assert.deepEqual((await Promise.all([first, second])).map((result) => result.revision), [1, 1]);
  const pending = store.waitForChange(store.revision, 10_000);
  store.shutdown();
  assert.equal((await pending).cancelled, true);
});