fix(runtime-host): serve revision-consistent Usage snapshots

apps/desktop/src/main/__tests__/usage-settings-view.test.ts

@@ -270,6 +270,128 @@ describe('Usage feature scope', () => {
     await act(async () => root.unmount());
   });
 
+  it('runs at most one load at a time and skips superseded queued ranges', async () => {
+    const { container, root } = setupDom();
+    const base: AppSettings = mergeSettings(createDefaultSettings(), {
+      usage: { range: '24h', activeTab: 'providers' },
+    });
+    const loads = new Map<UsageRange, Deferred<UsageStats | null>>();
+    const calls: UsageRange[] = [];
+    let activeLoads = 0;
+    let maxActiveLoads = 0;
+    const services: UsageServices = {
+      loadUsageStats: (range) => {
+        calls.push(range);
+        activeLoads += 1;
+        maxActiveLoads = Math.max(maxActiveLoads, activeLoads);
+        const d = deferred<UsageStats | null>();
+        loads.set(range, d);
+        return d.promise.finally(() => {
+          activeLoads -= 1;
+        });
+      },
+      updateUsageSettings: async (patch) => mergeSettings(base, { usage: patch }).usage,
+    };
+
+    await act(async () => {
+      root.render(tree({ active: true, settings: base, targetKey: 'hostA:1', services }));
+      await Promise.resolve();
+    });
+    for (const range of ['7d', '30d', 'all'] as const) {
+      const settings = mergeSettings(base, { usage: { range } });
+      await act(async () => {
+        root.render(tree({ active: true, settings, targetKey: 'hostA:1', services }));
+        await Promise.resolve();
+      });
+    }
+
+    assert.deepEqual(
+      calls,
+      ['24h'],
+      'range changes must queue behind the current load instead of consuming more snapshot slots',
+    );
+
+    await act(async () => {
+      loads.get('24h')!.resolve(statsWithRequests(111));
+      await flush();
+    });
+    assert.deepEqual(calls, ['24h', 'all'], 'only the latest queued range should load next');
+    assert.equal(maxActiveLoads, 1, 'a Usage scope must never overlap loads for one Host generation');
+
+    await act(async () => {
+      loads.get('all')!.resolve(statsWithRequests(444));
+      await flush();
+    });
+    assert.match(container.textContent ?? '', /444/, 'the latest range load should land');
+
+    await act(async () => root.unmount());
+  });
+
+  it('reuses a Host lane when switching A to B to A while the first A load is active', async () => {
+    const { container, root } = setupDom();
+    const base: AppSettings = mergeSettings(createDefaultSettings(), {
+      usage: { range: '24h', activeTab: 'providers' },
+    });
+    const loads = new Map<string, Deferred<UsageStats | null>[]>();
+    const calls: string[] = [];
+    const activeLoads = new Map<string, number>();
+    const maxActiveLoads = new Map<string, number>();
+    const servicesFor = (host: string): UsageServices => ({
+      loadUsageStats: (range) => {
+        calls.push(`${host}:${range}`);
+        const active = (activeLoads.get(host) ?? 0) + 1;
+        activeLoads.set(host, active);
+        maxActiveLoads.set(host, Math.max(maxActiveLoads.get(host) ?? 0, active));
+        const d = deferred<UsageStats | null>();
+        loads.set(host, [...(loads.get(host) ?? []), d]);
+        return d.promise.finally(() => {
+          activeLoads.set(host, (activeLoads.get(host) ?? 1) - 1);
+        });
+      },
+      updateUsageSettings: async (patch) => mergeSettings(base, { usage: patch }).usage,
+    });
+    const servicesA = servicesFor('A');
+    const servicesB = servicesFor('B');
+
+    await act(async () => {
+      root.render(tree({ active: true, settings: base, targetKey: 'hostA:1', services: servicesA }));
+      await Promise.resolve();
+    });
+    await act(async () => {
+      root.render(tree({ active: true, settings: base, targetKey: 'hostB:1', services: servicesB }));
+      await flush();
+    });
+    await act(async () => {
+      root.render(tree({ active: true, settings: base, targetKey: 'hostA:1', services: servicesA }));
+      await flush();
+    });
+
+    assert.deepEqual(
+      calls,
+      ['A:24h', 'B:24h'],
+      'returning to A must queue behind A while B remains independent',
+    );
+    assert.equal(maxActiveLoads.get('A'), 1, 'one connection must not overlap its own loads');
+    assert.equal(maxActiveLoads.get('B'), 1, 'a different connection may load independently');
+
+    await act(async () => {
+      loads.get('A')![0].resolve(statsWithRequests(111));
+      await flush();
+    });
+    assert.deepEqual(calls, ['A:24h', 'B:24h', 'A:24h']);
+
+    await act(async () => {
+      loads.get('A')![1].resolve(statsWithRequests(333));
+      loads.get('B')![0].resolve(statsWithRequests(222));
+      await flush();
+    });
+    assert.equal(maxActiveLoads.get('A'), 1, 'the revisited A load starts only after release');
+    assert.match(container.textContent ?? '', /333/, 'the current A load should land');
+    assert.doesNotMatch(container.textContent ?? '', /222/, 'the superseded B load must not land');
+
+    await act(async () => root.unmount());
+  });
+
   it('discards the previous Host generation snapshot when targetKey changes', async () => {
     const { container, root } = setupDom();
     const base: AppSettings = mergeSettings(createDefaultSettings(), {

apps/desktop/src/main/runtime-host-client.ts

@@ -30,6 +30,7 @@ import {
   projectSessionTodoItemsForDisplay,
   type SessionTodoItem,
 } from "@maka/core/session-todo";
+import type { UsageProvenance } from "@maka/core/usage-ledger-merge";
 
 import type {
   ConnectionVersionBasis,
@@ -42,7 +43,7 @@ import {
   canonicalPricingConfigsEqual,
   comparePricingModelKeys,
 } from "@maka/core/usage-stats/pricing";
-import type { PricingConfig } from "@maka/core/usage-stats/types";
+import type { PricingConfig, TimeRange, UsageSummaryV2 } from "@maka/core/usage-stats/types";
 import {
   type ClientCapabilityProvider,
   type DecodedSessionTranscriptPage,
@@ -151,6 +152,11 @@ import {
   type TurnInterruptResult,
   type TurnMessageSubmitInput,
   type TurnMessageSubmitResult,
+  type LlmUsageLogProjection,
+  type ToolUsageLogProjection,
+  PRICING_PAGE_MAX_ITEMS,
+  USAGE_PAGE_MAX_ITEMS,
+  USAGE_SNAPSHOT_ACTIVITY_MAX_ITEMS,
   type WorkspaceProjection,
 } from "@maka/runtime-host/protocol";
 
@@ -159,6 +165,8 @@ const decodeStoredMessage = (value: unknown): StoredMessage =>
 const MAX_OPTIMISTIC_ATTEMPTS = 3;
 const MAX_SESSION_REVISION_ATTEMPTS = 8;
 const MAX_PRICING_SNAPSHOT_ATTEMPTS = 3;
+const MAX_USAGE_SNAPSHOT_ATTEMPTS = 3;
+const USAGE_SNAPSHOT_RETRY_DELAY_MS = 50;
 const RUNTIME_HOST_RETIREMENT_TIMEOUT_MS = 5_000;
 
 export type DesktopSessionConfigurationPatch = SessionConfigurationPatch;
@@ -189,6 +197,7 @@ export type DesktopRuntimeHostClientErrorCode =
   | "revision_conflict"
   | "session_not_found"
   | "skill_catalog_unstable"
+  | "usage_unstable"
   | "unsupported_session";
 
 export class DesktopRuntimeHostClientError extends Error {
@@ -231,6 +240,17 @@ export interface DesktopPricingSnapshot {
   readonly entries: readonly EffectivePricingEntry[];
 }
 
+export interface DesktopUsageSnapshot {
+  readonly revision: string;
+  readonly summary: UsageSummaryV2;
+  readonly provenance: UsageProvenance;
+  readonly llmLogs: readonly LlmUsageLogProjection[];
+  readonly toolLogs: readonly ToolUsageLogProjection[];
+  readonly pricingEntries: readonly EffectivePricingEntry[];
+  readonly llmLogsTruncated: boolean;
+  readonly toolLogsTruncated: boolean;
+}
+
 export interface DesktopSkillCatalogSnapshot {
   readonly revision: SkillCatalogRevision;
   readonly view: SkillCatalogView;
@@ -1378,6 +1398,30 @@ export class DesktopRuntimeHostClient {
     return this.request("usage.query", input);
   }
 
+  async loadUsageSnapshot(range: TimeRange): Promise<DesktopUsageSnapshot> {
+    for (let attempt = 0; attempt < MAX_USAGE_SNAPSHOT_ATTEMPTS; attempt += 1) {
+      try {
+        const snapshot = await this.#readUsageSnapshot(range);
+        if (snapshot) return snapshot;
+      } catch (error) {
+        if (
+          !(error instanceof RuntimeHostOperationError) ||
+          error.operation !== "usage.query" ||
+          error.code !== "operation_conflict"
+        ) {
+          throw error;
+        }
+      }
+      if (attempt + 1 < MAX_USAGE_SNAPSHOT_ATTEMPTS) {
+        await new Promise<void>((resolve) => setTimeout(resolve, USAGE_SNAPSHOT_RETRY_DELAY_MS));
+      }
+    }
+    throw new DesktopRuntimeHostClientError(
+      "usage_unstable",
+      "Usage snapshot stayed unavailable across bounded retries",
+    );
+  }
+
   queryGoal(sessionId: string): Promise<OperationOutput<"goal.query">> {
     return this.request("goal.query", { sessionId });
   }
@@ -1719,6 +1763,157 @@ export class DesktopRuntimeHostClient {
     };
   }
 
+  async #readUsageSnapshot(range: TimeRange): Promise<DesktopUsageSnapshot | undefined> {
+    this.#assertOpen();
+    const started = await this.request("usage.query", { kind: "snapshot_start", range });
+    if (started.kind !== "snapshot_started") {
+      throw invalidProjection("Usage snapshot start");
+    }
+    try {
+      if (
+        typeof range === "object" &&
+        (started.summary.range.from !== range.from || started.summary.range.to !== range.to)
+      ) {
+        throw invalidProjection("Usage snapshot start");
+      }
+      const [llm, tool, pricing] = await Promise.all([
+        this.#readUsageSnapshotLogs(started.revision, "llm"),
+        this.#readUsageSnapshotLogs(started.revision, "tool"),
+        this.#readUsageSnapshotPricing(started.revision),
+      ]);
+      if (!llm || !tool || !pricing) return undefined;
+      return {
+        revision: started.revision,
+        summary: started.summary,
+        provenance: started.provenance,
+        llmLogs: llm.rows,
+        toolLogs: tool.rows,
+        pricingEntries: pricing,
+        llmLogsTruncated: llm.truncated,
+        toolLogsTruncated: tool.truncated,
+      };
+    } finally {
+      try {
+        await this.request("usage.snapshot.release", { revision: started.revision });
+      } catch {
+        // Usage snapshot release is best-effort cleanup.
+      }
+    }
+  }
+
+  async #readUsageSnapshotLogs(
+    revision: string,
+    source: "llm",
+  ): Promise<{ readonly rows: readonly LlmUsageLogProjection[]; readonly truncated: boolean } | undefined>;
+  async #readUsageSnapshotLogs(
+    revision: string,
+    source: "tool",
+  ): Promise<{ readonly rows: readonly ToolUsageLogProjection[]; readonly truncated: boolean } | undefined>;
+  async #readUsageSnapshotLogs(
+    revision: string,
+    source: "llm" | "tool",
+  ): Promise<
+    | {
+        readonly rows: readonly (LlmUsageLogProjection | ToolUsageLogProjection)[];
+        readonly truncated: boolean;
+      }
+    | undefined
+  > {
+    const rows: Array<LlmUsageLogProjection | ToolUsageLogProjection> = [];
+    let offset = 0;
+    let total: number | undefined;
+    let truncated: boolean | undefined;
+    while (true) {
+      const page = await this.request("usage.query", {
+        kind: "snapshot_logs",
+        revision,
+        source,
+        offset,
+        limit: USAGE_PAGE_MAX_ITEMS,
+      });
+      if (page.kind === "revision_changed") {
+        if (page.expectedRevision !== revision) throw invalidProjection("Usage snapshot revision");
+        return undefined;
+      }
+      if (
+        page.kind !== "snapshot_logs" ||
+        page.revision !== revision ||
+        page.source !== source ||
+        page.offset !== offset ||
+        page.rows.length > USAGE_PAGE_MAX_ITEMS ||
+        page.total > USAGE_SNAPSHOT_ACTIVITY_MAX_ITEMS
+      ) {
+        throw invalidProjection("Usage snapshot logs");
+      }
+      total ??= page.total;
+      truncated ??= page.truncated;
+      if (page.total !== total || page.truncated !== truncated || rows.length !== offset) {
+        throw invalidProjection("Usage snapshot logs");
+      }
+      rows.push(...page.rows);
+      if (rows.length > total) throw invalidProjection("Usage snapshot logs");
+      if (page.nextOffset === null) {
+        if (rows.length !== total) throw invalidProjection("Usage snapshot logs");
+        return { rows, truncated };
+      }
+      if (
+        page.rows.length === 0 ||
+        page.nextOffset !== offset + page.rows.length ||
+        page.nextOffset >= total
+      ) {
+        throw invalidProjection("Usage snapshot logs");
+      }
+      offset = page.nextOffset;
+    }
+  }
+
+  async #readUsageSnapshotPricing(
+    revision: string,
+  ): Promise<readonly EffectivePricingEntry[] | undefined> {
+    const entries: EffectivePricingEntry[] = [];
+    let offset = 0;
+    let total: number | undefined;
+    while (true) {
+      const page = await this.request("usage.query", {
+        kind: "snapshot_pricing",
+        revision,
+        offset,
+        limit: PRICING_PAGE_MAX_ITEMS,
+      });
+      if (page.kind === "revision_changed") {
+        if (page.expectedRevision !== revision) throw invalidProjection("Usage snapshot revision");
+        return undefined;
+      }
+      if (
+        page.kind !== "snapshot_pricing" ||
+        page.revision !== revision ||
+        page.offset !== offset ||
+        page.entries.length > PRICING_PAGE_MAX_ITEMS ||
+        entries.length !== offset
+      ) {
+        throw invalidProjection("Usage snapshot pricing");
+      }
+      total ??= page.total;
+      if (page.total !== total) throw invalidProjection("Usage snapshot pricing");
+      entries.push(...page.entries);
+      if (entries.length > total) throw invalidProjection("Usage snapshot pricing");
+      if (page.nextOffset === null) {
+        if (entries.length !== total || !pricingEntriesAreCanonical(entries)) {
+          throw invalidProjection("Usage snapshot pricing");
+        }
+        return entries;
+      }
+      if (
+        page.entries.length === 0 ||
+        page.nextOffset !== offset + page.entries.length ||
+        page.nextOffset >= total
+      ) {
+        throw invalidProjection("Usage snapshot pricing");
+      }
+      offset = page.nextOffset;
+    }
+  }
+
   async #reconcilePricingMutation(
     target: PricingReconciliationTarget,
     reason: "revision_conflict" | "outcome_unknown",