fix(codex): serve the last catalog reading while the Windows probe refreshes (rebase of #2542)

src/codex/app-server-processes.ts

@@ -740,6 +740,28 @@ const CATALOG_STATE_TTL_MS = 5_000;
  */
 const CATALOG_STATE_UNKNOWN_TTL_MS = 250;
 
+/**
+ * How long a real observation may still be SERVED after it expires, while a refresh
+ * runs behind it (#2499).
+ *
+ * The probe is advisory and, on Windows, slow: `Invoke-CimMethod GetOwner` costs
+ * ~0.4s per candidate process, so a cold probe routinely outlives the 5s TTL and
+ * every turn that misses the cache pays for it on the request path. Serving the
+ * previous reading immediately keeps that cost off the turn without pretending it is
+ * fresh -- the refresh it triggers is what makes the next reading current.
+ *
+ * Measured from expiry rather than from when the reading was taken, so a `fresh`
+ * entry stays servable for its own TTL plus this bound. Anchoring it to expiry keeps
+ * the stale window independent of the TTL -- a cap on total age would quietly turn
+ * this path off if the TTL were ever raised past it.
+ *
+ * Bounded rather than unlimited: if the refresh keeps failing, an observation this
+ * old stops being evidence about the machine and it is better to wait for a real one.
+ * `unknown` is never served this way -- it is a failure to observe, not an
+ * observation, and it already has its own short window for exactly that reason.
+ */
+export const CATALOG_STATE_MAX_STALE_MS = 60_000;
+
 export function catalogStateTtlMs(state: CodexAppServerCatalogState): number {
   return state === "unknown" ? CATALOG_STATE_UNKNOWN_TTL_MS : CATALOG_STATE_TTL_MS;
 }
@@ -835,6 +857,14 @@ export function collectCodexAppServerCatalogState(
  * - 선택한 방식: retain the synchronous API and use async PowerShell plus an identity-scoped in-flight refresh, short cache, and invalidation generation only for Windows requests.
  * - 다른 대안 대신 이 방식을 선택한 이유: it fixes unrelated `/healthz` starvation without widening the process-matching or restart contract.
  * - 장점, 단점 및 영향: concurrent turns share one CIM walk, invalidated pre-write results cannot repopulate the cache, and the event loop stays responsive; a cold v2 turn can still await the bounded advisory probe.
+ *
+ * [Decision Log · #2499]
+ * - 목적과 의도: a cold probe outlives its own 5s TTL on Windows (~435ms per candidate process for `Invoke-CimMethod GetOwner`), so the cache expires before it can serve and the miss lands on a turn.
+ * - 기존 구현 및 제약 조건: the reading is advisory, and only `fresh` authorizes positive guidance (`src/server/responses/collaboration.ts`); `unknown` is a failure to observe rather than an observation.
+ * - 검토한 주요 대안: drop the per-process GetOwner fan-out (issue suggestion 1), or widen the TTL past the probe duration (suggestion 3).
+ * - 선택한 방식: serve an expired reading immediately when its generation still matches, bounded by `CATALOG_STATE_MAX_STALE_MS`, never for `unknown`, and refresh behind it; a failed refresh no longer evicts the reading it was refreshing.
+ * - 다른 대안 대신 이 방식을 선택한 이유: the fan-out change alters what "could not verify the owner" means for the current-user scoping contract and needs its own ground-truth comparison; a wider TTL still pays the probe on every human-paced turn.
+ * - 장점, 단점 및 영향: after the first probe the request path never waits; a server that stopped between readings can be described as `fresh` for up to the stale bound; a catalog write still invalidates immediately through the generation, and the cold path is unchanged.
  */
 export async function collectCodexAppServerCatalogStateForRequest(
   io: CodexAppServerProcessIo = {},
@@ -853,16 +883,30 @@ export async function collectCodexAppServerCatalogStateForRequest(
     catalogMtimeMs: io.catalogMtimeMs,
     now: io.now,
   };
-  if (requestCatalogStateCache
+  const cached = requestCatalogStateCache
     && requestCatalogStateCache.generation === generation
     && sameRequestCatalogStateIdentity(requestCatalogStateCache.identity, identity)
-    && now - requestCatalogStateCache.atMs < catalogStateTtlMs(requestCatalogStateCache.status.state)) {
-    return requestCatalogStateCache.status;
+    ? requestCatalogStateCache
+    : null;
+  if (cached && now - cached.atMs < catalogStateTtlMs(cached.status.state)) {
+    return cached.status;
   }
+  // An expired real reading is still worth handing back while the refresh runs. It
+  // cannot have been invalidated by an ocx catalog write: every such write calls
+  // `resetCodexAppServerCatalogStateCache`, which advances the generation and drops
+  // this entry, so a generation match means no write has landed since it was taken.
+  // What it can miss is an app-server that started or stopped meanwhile -- and a
+  // server started after the reading is newer than the catalog, which is the `fresh`
+  // this entry already says.
+  const servableStale = cached
+    && cached.status.state !== "unknown"
+    && now - cached.atMs < catalogStateTtlMs(cached.status.state) + CATALOG_STATE_MAX_STALE_MS
+    ? cached.status
+    : null;
   if (requestCatalogStateFlight
     && requestCatalogStateFlight.generation === generation
     && sameRequestCatalogStateIdentity(requestCatalogStateFlight.identity, identity)) {
-    return requestCatalogStateFlight.promise;
+    return servableStale ?? requestCatalogStateFlight.promise;
   }
 
   const refresh = async (): Promise<CodexAppServerCatalogStatus> => {
@@ -920,7 +964,18 @@ export async function collectCodexAppServerCatalogStateForRequest(
     if (requestCatalogStateGeneration !== generation) {
       return { state: "unknown" as const, processes: [], catalogMtimeMs: null };
     }
-    if (requestCatalogStateFlight === flight) {
+    // A refresh that failed must not evict a real reading. Before this function
+    // served stale entries, caching `unknown` cost at most the 250ms that state
+    // is allowed to live. Now that an expired observation is what callers are
+    // handed, overwriting one with `unknown` would take the answer AWAY from
+    // them on a transient failure -- `unknown` is not servable, so the next
+    // caller waits for a probe instead of getting the reading it would have had.
+    // Keep the observation and let its own age retire it.
+    const wouldEvictAnObservation = status.state === "unknown"
+      && requestCatalogStateCache?.generation === generation
+      && requestCatalogStateCache.status.state !== "unknown"
+      && sameRequestCatalogStateIdentity(requestCatalogStateCache.identity, identity);
+    if (requestCatalogStateFlight === flight && !wouldEvictAnObservation) {
       requestCatalogStateCache = {
         generation,
         identity,
@@ -934,7 +989,9 @@ export async function collectCodexAppServerCatalogStateForRequest(
   });
   flight = { generation, identity, promise };
   requestCatalogStateFlight = flight;
-  return flight.promise;
+  // `promise` already absorbs its own failures, so leaving it unawaited here cannot
+  // surface as an unhandled rejection; the next caller picks up whatever it stored.
+  return servableStale ?? flight.promise;
 }
 
 /**

tests/codex-app-server-processes.test.ts

@@ -7,6 +7,7 @@ import { setTrustedWindowsElevationExecutablesForTests } from "../src/lib/window
 import {
   afterCatalogWriteHandleAppServers,
   attachStaleAppServerHint,
+  CATALOG_STATE_MAX_STALE_MS,
   catalogStateTtlMs,
   collectCodexAppServerCatalogState,
   collectCodexAppServerCatalogStateForRequest,
@@ -1063,3 +1064,208 @@ describe("platform termination ladder", () => {
     expect(signals).toEqual([{ pid: 4242, signal: "SIGTERM" }]);
   });
 });
+
+describe("request-path catalog state serves stale while revalidating (#2499)", () => {
+  // The suite's other command-line fixture is scoped to its own describe block.
+  const APP_SERVER_COMMAND_LINE = "/usr/local/bin/codex app-server";
+  const APP_SERVER = [{ pid: 42, commandLine: APP_SERVER_COMMAND_LINE }];
+
+  /**
+   * A probe that never settles on its own. Every test here needs to observe what a
+   * caller gets back WHILE an enumeration is still running, which is the whole point:
+   * on Windows this probe costs ~0.4s per candidate process and routinely outlives
+   * the 5s TTL, so before this change one turn per window paid for it on the request
+   * path.
+   */
+  function makeIo(clock: { ms: number }) {
+    const releases: Array<(snapshots: Array<{ pid: number; commandLine: string }>) => void> = [];
+    const io = {
+      platform: "win32" as const,
+      now: () => clock.ms,
+      listSnapshotsAsync: () =>
+        new Promise<Array<{ pid: number; commandLine: string }>>(resolve => {
+          releases.push(resolve);
+        }),
+      readStartMsBatchAsync: async (pids: number[]) => new Map(pids.map(pid => [pid, 2_000] as const)),
+      catalogMtimeMs: () => 1_000,
+    };
+    return { io, releases, enumerations: () => releases.length };
+  }
+
+  /**
+   * Yield until the released refresh has stored its result, or give up loudly.
+   *
+   * A fixed sleep would be a bet on how many turns the refresh's continuation
+   * needs, and losing that bet on a slow runner looks like a product bug rather
+   * than a slow machine. This waits for the condition instead of for a duration.
+   */
+  async function drainUntil(predicate: () => boolean, what: string) {
+    for (let attempt = 0; attempt < 1_000; attempt += 1) {
+      if (predicate()) return;
+      await new Promise(resolve => setTimeout(resolve, 0));
+    }
+    throw new Error(`timed out waiting for ${what}`);
+  }
+  /** Did *promise* settle without the pending probe being released? */
+  async function settledWithoutTheProbe<T>(promise: Promise<T> | T): Promise<T | "waited"> {
+    return Promise.race([
+      Promise.resolve(promise),
+      new Promise<"waited">(resolve => setTimeout(() => resolve("waited"), 25)),
+    ]);
+  }
+
+  beforeEach(() => {
+    resetCodexAppServerCatalogStateCache();
+  });
+
+  test("the first turn waits, later turns do not, and N turns cost one enumeration", async () => {
+    const clock = { ms: 1_000_000 };
+    const { io, releases, enumerations } = makeIo(clock);
+
+    // Cold: nothing cached, so this one has to wait for the probe.
+    const cold = collectCodexAppServerCatalogStateForRequest(io);
+    expect(await settledWithoutTheProbe(cold)).toBe("waited");
+    releases[0]([...APP_SERVER]);
+    await expect(cold).resolves.toMatchObject({ state: "fresh" });
+    expect(enumerations()).toBe(1);
+
+    // Past the TTL. The reading is expired, and the refresh it triggers is still
+    // running -- the caller must get the previous reading now, not wait for it.
+    clock.ms += catalogStateTtlMs("fresh") + 1;
+    const warm = collectCodexAppServerCatalogStateForRequest(io);
+    expect(await settledWithoutTheProbe(warm)).toMatchObject({ state: "fresh" });
+    expect(enumerations()).toBe(2);
+
+    // Three more turns while that refresh is still in flight: still served, and the
+    // in-flight dedup means none of them starts another enumeration.
+    for (let turn = 0; turn < 3; turn += 1) {
+      clock.ms += 10;
+      const next = collectCodexAppServerCatalogStateForRequest(io);
+      expect(await settledWithoutTheProbe(next)).toMatchObject({ state: "fresh" });
+    }
+    expect(enumerations()).toBe(2);
+  });
+
+  test("the refresh it triggers is what makes the next reading current", async () => {
+    const clock = { ms: 2_000_000 };
+    const { io, releases } = makeIo(clock);
+
+    const cold = collectCodexAppServerCatalogStateForRequest(io);
+    releases[0]([...APP_SERVER]);
+    await expect(cold).resolves.toMatchObject({ state: "fresh" });
+
+    clock.ms += catalogStateTtlMs("fresh") + 1;
+    // Served from the expired entry, and the refresh that call started is what the
+    // rest of this test is about.
+    const served = collectCodexAppServerCatalogStateForRequest(io);
+    expect(await settledWithoutTheProbe(served)).toMatchObject({ state: "fresh" });
+
+    // The background refresh finds the machine empty now.
+    let refreshed = false;
+    served.then(() => {}).catch(() => {});
+    releases[1]([]);
+    // The refresh has landed once a read at the same instant reports the new
+    // state; before that it still answers from the entry it is replacing.
+    await drainUntil(() => {
+      void collectCodexAppServerCatalogStateForRequest(io).then(seen => {
+        if (seen.state === "not_running") refreshed = true;
+      });
+      return refreshed;
+    }, "the background refresh to store not_running");
+
+    // Served from the refreshed entry, still without waiting.
+    clock.ms += 1;
+    const after = collectCodexAppServerCatalogStateForRequest(io);
+    expect(await settledWithoutTheProbe(after)).toMatchObject({ state: "not_running" });
+  });
+
+  test("a reading older than the bound is not served -- the caller waits for a real one", async () => {
+    const clock = { ms: 3_000_000 };
+    const { io, releases } = makeIo(clock);
+
+    const cold = collectCodexAppServerCatalogStateForRequest(io);
+    releases[0]([...APP_SERVER]);
+    await expect(cold).resolves.toMatchObject({ state: "fresh" });
+
+    // One tick inside the bound, which runs from expiry rather than from when the
+    // reading was taken: still served, still without waiting.
+    clock.ms += catalogStateTtlMs("fresh") + CATALOG_STATE_MAX_STALE_MS - 1;
+    const last = collectCodexAppServerCatalogStateForRequest(io);
+    expect(await settledWithoutTheProbe(last)).toMatchObject({ state: "fresh" });
+
+    // One tick past it, where the reading stops being evidence about the machine.
+    // That call joins the refresh the previous one started rather than waiting on a
+    // second enumeration, so releasing that one probe is what settles it.
+    clock.ms += 1;
+    const tooOld = collectCodexAppServerCatalogStateForRequest(io);
+    expect(await settledWithoutTheProbe(tooOld)).toBe("waited");
+    releases[1]([...APP_SERVER]);
+    await expect(tooOld).resolves.toMatchObject({ state: "fresh" });
+  });
+
+  test("a failed refresh does not evict the reading it was refreshing", async () => {
+    const clock = { ms: 5_000_000 };
+    const releases: Array<(snapshots: Array<{ pid: number; commandLine: string }>) => void> = [];
+    const rejects: Array<(reason: Error) => void> = [];
+    const io = {
+      platform: "win32" as const,
+      now: () => clock.ms,
+      listSnapshotsAsync: () =>
+        new Promise<Array<{ pid: number; commandLine: string }>>((resolve, reject) => {
+          releases.push(resolve);
+          rejects.push(reject);
+        }),
+      readStartMsBatchAsync: async (pids: number[]) => new Map(pids.map(pid => [pid, 2_000] as const)),
+      catalogMtimeMs: () => 1_000,
+    };
+
+    const cold = collectCodexAppServerCatalogStateForRequest(io);
+    releases[0]([{ pid: 42, commandLine: APP_SERVER_COMMAND_LINE }]);
+    await expect(cold).resolves.toMatchObject({ state: "fresh" });
+
+    // Expire it, take the stale answer, and let the refresh it started fail.
+    clock.ms += catalogStateTtlMs("fresh") + 1;
+    const served = collectCodexAppServerCatalogStateForRequest(io);
+    expect(await settledWithoutTheProbe(served)).toMatchObject({ state: "fresh" });
+    rejects[1](new Error("windows_enum_incomplete"));
+    await drainUntil(() => rejects.length === 2, "the failing refresh to be started");
+    await new Promise(resolve => setTimeout(resolve, 0));
+
+    // The transient failure must not have replaced the observation. Caching
+    // `unknown` over it would take the answer away from the next caller: `unknown`
+    // is not servable, so that caller would wait for a probe instead of being
+    // handed the reading it would otherwise have had.
+    clock.ms += 1;
+    const after = collectCodexAppServerCatalogStateForRequest(io);
+    expect(await settledWithoutTheProbe(after)).toMatchObject({ state: "fresh" });
+  });
+
+  test("`unknown` is never served stale -- a failure to observe is not an observation", async () => {
+    const clock = { ms: 4_000_000 };
+    const releases: Array<(value: never[]) => void> = [];
+    const rejects: Array<(reason: Error) => void> = [];
+    const io = {
+      platform: "win32" as const,
+      now: () => clock.ms,
+      listSnapshotsAsync: () =>
+        new Promise<never[]>((resolve, reject) => {
+          releases.push(resolve);
+          rejects.push(reject);
+        }),
+      readStartMsBatchAsync: async (pids: number[]) => new Map(pids.map(pid => [pid, 2_000] as const)),
+      catalogMtimeMs: () => 1_000,
+    };
+
+    const cold = collectCodexAppServerCatalogStateForRequest(io);
+    rejects[0](new Error("windows_enum_incomplete"));
+    await expect(cold).resolves.toMatchObject({ state: "unknown" });
+
+    // Its own short window has passed. Handing `unknown` back again would keep
+    // guidance suppressed on the strength of a reading that never observed anything.
+    clock.ms += catalogStateTtlMs("unknown") + 1;
+    const next = collectCodexAppServerCatalogStateForRequest(io);
+    expect(await settledWithoutTheProbe(next)).toBe("waited");
+    releases[1]([]);
+    await expect(next).resolves.toMatchObject({ state: "not_running" });
+  });
+});