fix(server): usage no longer waits on Muse home discovery

apps/server/src/usage/UsageService.test.ts

@@ -12,6 +12,7 @@ import { vi } from "vite-plus/test";
 import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import * as HostProcess from "@t3tools/shared/HostProcess";
+import { SpawnExecutableResolution } from "@t3tools/shared/shell";
 import { mergeUsage } from "@t3tools/shared/usageMerge";
 import {
   EnvironmentId,
@@ -1120,7 +1121,7 @@ describe("UsageService", () => {
           };
           const muse = (summary: UsageSummary) =>
             summary.buckets.filter((bucket) => bucket.provider === "muse");
-          const first = yield* service.readSummary(window);
+          const first = yield* service.readSummary({ ...window, awaitRefresh: true });
           const buckets = muse(first);
           assert.strictEqual(
             buckets.reduce((sum, b) => sum + b.records, 0),
@@ -1202,6 +1203,7 @@ describe("UsageService", () => {
           assert.closeTo(repriced?.costUsd ?? NaN, 0.000486, 1e-12);
         }).pipe(
           Effect.scoped,
+          Effect.provideService(SpawnExecutableResolution, (command: string) => command),
           Effect.provide(
             layerService({
               prefix: "usage-muse",
@@ -1235,6 +1237,66 @@ describe("UsageService", () => {
       }).pipe(Effect.scoped),
   );
 
+  it.live("returns other providers' usage while Muse home discovery is still pending", () =>
+    Effect.gen(function* () {
+      const { transcript, settings, home } = yield* setup;
+      yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 5)));
+      const museStarted = yield* Deferred.make<void>();
+      const museReady = yield* Deferred.make<MuseSdk.MuseSdkHost>();
+      const museHome = NodePath.join(home, "muse");
+      yield* Effect.promise(() =>
+        NodeFSP.mkdir(NodePath.join(museHome, "sessions"), { recursive: true }),
+      );
+      using _create = vi
+        .spyOn(MuseSdk, "createMuseSdkHostEffect")
+        .mockImplementation(() =>
+          Deferred.succeed(museStarted, undefined).pipe(Effect.andThen(Deferred.await(museReady))),
+        );
+      const museSources = (summary: UsageSummary) =>
+        summary.sources
+          .filter((source) => source.fingerprint.provider === "muse")
+          .map((source) => [source.fingerprint.resolvedHomePath, source.status, source.refreshing]);
+      const [pending, refreshed] = yield* Effect.gen(function* () {
+        const service = yield* UsageService.make;
+        const read = yield* Effect.forkChild(service.readSummary(WINDOW));
+        yield* Deferred.await(museStarted);
+        const pending = yield* Fiber.join(read);
+        const refresh = yield* Effect.forkChild(
+          service.readSummary({ ...WINDOW, awaitRefresh: true }),
+        );
+        yield* Deferred.succeed(museReady, {
+          initializeResult: { grantedCapabilities: [], museHome },
+          close: async () => {},
+        } as unknown as MuseSdk.MuseSdkHost);
+        return [pending, yield* Fiber.join(refresh)] as const;
+      }).pipe(
+        Effect.scoped,
+        Effect.provideService(SpawnExecutableResolution, (command: string) => command),
+        Effect.provide(
+          layerService({
+            prefix: "usage-muse-pending",
+            home,
+            settings: {
+              ...settings,
+              providerInstances: {
+                ...settings.providerInstances,
+                [ProviderInstanceId.make("muse")]: {
+                  driver: ProviderDriverKind.make("muse"),
+                  enabled: false,
+                },
+              },
+            },
+          }),
+        ),
+      );
+      assert.strictEqual(totalOutputTokens(pending), 5);
+      assert.deepStrictEqual(museSources(pending), [["muse", "ok", true]]);
+      assert.deepStrictEqual(museSources(refreshed), [
+        [NodePath.join(museHome, "sessions"), "ok", undefined],
+      ]);
+    }).pipe(Effect.scoped),
+  );
+
   it.live("reports missing Pi history and skips malformed records without losing valid usage", () =>
     Effect.gen(function* () {
       const { home, settings } = yield* setup;

apps/server/src/usage/UsageService.ts

@@ -330,16 +330,25 @@ export const make = Effect.gen(function* () {
   const resolveTranscriptDirs = Effect.fn("UsageService.resolveTranscriptDirs")(function* (
     settings: ServerSettingsValue,
     retentionCutoffMs: number,
+    awaitRefresh: boolean,
   ) {
     const dirs: Array<TranscriptSource> = [];
+    const pending: Array<ScannedDir> = [];
     const seen = new Set<string>();
     for (const { driver, reader } of transcriptReaders) {
       const { provider, format } = reader;
       const directories = yield* Effect.forEach(
         yield* usageInstances(driver, settings),
-        (instance) => reader.directories(instance),
+        (instance) => reader.directories(instance, { awaitRefresh }),
       );
-      for (const { dir: directory, fileName } of directories.flat()) {
+      for (const { dir: directory, fileName, refreshing } of directories.flat()) {
+        if (refreshing) {
+          const key = `${provider}\0${directory}`;
+          if (seen.has(key)) continue;
+          seen.add(key);
+          pending.push({ provider, dir: directory, volumeId: "", files: [], refreshing });
+          continue;
+        }
         const sourceKey = provider + "\0" + directory;
         const previous = sourceCache.get(sourceKey);
         const dir = yield* fileSystem
@@ -375,7 +384,7 @@ export const make = Effect.gen(function* () {
         });
       }
     }
-    return dirs;
+    return { dirs, pending };
   });
 
   // Share first loads so concurrent readers cannot scan against a still-empty cache.
@@ -654,9 +663,11 @@ export const make = Effect.gen(function* () {
     retentionCutoffMs: number,
     awaitRefresh: boolean,
   ) {
-    const dirs = yield* resolveTranscriptDirs(settings, retentionCutoffMs).pipe(
-      Effect.provideContext(readerContext),
-    );
+    const { dirs, pending } = yield* resolveTranscriptDirs(
+      settings,
+      retentionCutoffMs,
+      awaitRefresh,
+    ).pipe(Effect.provideContext(readerContext));
 
     const scans = Effect.forEach(
       scanReaders,
@@ -690,7 +701,7 @@ export const make = Effect.gen(function* () {
       [Effect.forEach(dirs, (dir) => scanTranscriptDir(dir, windowStartMs)), scans],
       { concurrency: "unbounded" },
     );
-    const scanned: readonly ScannedDir[] = [...transcripts, ...scanDirs.flat()];
+    const scanned: readonly ScannedDir[] = [...transcripts, ...pending, ...scanDirs.flat()];
     return scanned;
   });
 

packages/provider-core/src/server/usage.ts

@@ -111,10 +111,16 @@ export type ProviderUsageReader<Config, R> =
       readonly resolveServiceTiers?: (
         records: readonly UsageRecord[],
       ) => Effect.Effect<ReadonlyMap<UsageRecord, UsageSpeed>, never, R>;
+      /** A `refreshing` entry is not scanned; it reports its source as still being located. */
       readonly directories: (
         instance: ProviderUsageInstance<Config>,
+        options: { readonly awaitRefresh: boolean },
       ) => Effect.Effect<
-        ReadonlyArray<{ readonly dir: string; readonly fileName?: string }>,
+        ReadonlyArray<{
+          readonly dir: string;
+          readonly fileName?: string;
+          readonly refreshing?: true;
+        }>,
         never,
         R
       >;

packages/provider-muse/src/server/MuseUsageHome.ts

@@ -4,24 +4,32 @@ import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
 import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
+
+import * as HostProcess from "@t3tools/shared/HostProcess";
+import { SpawnExecutableResolution } from "@t3tools/shared/shell";
 
 import { createMuseSdkHostEffect, MuseSdkError } from "./sdk.ts";
 
 export class MuseUsageHome extends Context.Service<
   MuseUsageHome,
   {
+    /** None while discovery is still running and `awaitRefresh` is false. */
     readonly resolve: (input: {
       readonly binaryPath: string;
       readonly environment: NodeJS.ProcessEnv;
-    }) => Effect.Effect<string | undefined>;
+      readonly awaitRefresh: boolean;
+    }) => Effect.Effect<Option.Option<string | undefined>>;
   }
 >()("@t3tools/provider-muse/server/MuseUsageHome") {}
 
 const make = Effect.gen(function* () {
+  const scope = yield* Effect.scope;
   const cache = yield* Cache.makeWith(
-    Effect.fn("MuseUsageHome.initialize")(function* (
-      input: Parameters<MuseUsageHome["Service"]["resolve"]>[0],
-    ) {
+    Effect.fn("MuseUsageHome.initialize")(function* (input: {
+      readonly binaryPath: string;
+      readonly environment: NodeJS.ProcessEnv;
+    }) {
       return yield* Effect.scoped(
         Effect.gen(function* () {
           const host = yield* Effect.acquireRelease(
@@ -35,7 +43,7 @@ const make = Effect.gen(function* () {
           );
           return host.initializeResult.museHome;
         }),
-      );
+      ).pipe(Effect.orElseSucceed(() => undefined));
     }),
     {
       capacity: 32,
@@ -45,15 +53,27 @@ const make = Effect.gen(function* () {
     },
   );
   const resolve: MuseUsageHome["Service"]["resolve"] = Effect.fn("MuseUsageHome.resolve")(
-    function* ({ binaryPath, environment }) {
+    function* ({ binaryPath, environment, awaitRefresh }) {
       // Cache keys compare structurally. Omitted and undefined variables both
       // disappear from the child process environment and must share an entry.
-      return yield* Cache.get(cache, {
+      const key = {
         binaryPath,
         environment: Object.fromEntries(
           Object.entries(environment).filter(([, value]) => value !== undefined),
         ),
-      }).pipe(Effect.orElseSucceed(() => undefined));
+      };
+      const settled = yield* Cache.getSuccess(cache, key);
+      if (Option.isSome(settled)) return settled;
+      // Spawn failures surface asynchronously; without this a missing CLI would report as pending.
+      const resolveExecutable = yield* SpawnExecutableResolution;
+      if (resolveExecutable(binaryPath, yield* HostProcess.Platform, environment) === undefined) {
+        return Option.some(undefined);
+      }
+      if (awaitRefresh) return Option.some(yield* Cache.get(cache, key));
+      // Awaited in the layer scope so a departing caller cannot cancel discovery.
+      yield* Effect.forkIn(Cache.get(cache, key), scope, { startImmediately: true });
+      // A lookup that settled synchronously needs no refresh.
+      return yield* Cache.getSuccess(cache, key);
     },
   );
   return MuseUsageHome.of({ resolve });