fix(runtime-host): claim Daily Review archives before reading and widen the owned settle bound

packages/runtime-host/src/__tests__/daily-review-coordinator.test.ts

@@ -22,7 +22,12 @@ import { mkdtemp, rm } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { test } from 'node:test';
-import { localDayBoundsAt, type DailyReviewArchive } from '@maka/core/daily-review';
+import {
+  dailyReviewArchiveId,
+  localDayBoundsAt,
+  localDayBoundsForInstant,
+  type DailyReviewArchive,
+} from '@maka/core/daily-review';
 import { openInteractiveDailyReviewAuthorityForWrite } from '@maka/storage/daily-review-authority';
 import { acquireOperationalStateDatabase } from '@maka/storage/operational-state-store';
 import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority';
@@ -153,6 +158,124 @@ test('Daily Review conflicts rather than coalescing different generation options
   );
 });
 
+test('Daily Review joins an in-flight generation even when its own reads land later', async () => {
+  let releaseLaggingRead: (() => void) | undefined;
+  const laggingRead = new Promise<void>((resolve) => {
+    releaseLaggingRead = resolve;
+  });
+  let listCalls = 0;
+  let modelCalls = 0;
+
+  await withCoordinator(
+    async ({ coordinator, usage }) => {
+      const now = Date.now();
+      await usage.telemetry.recordLlmCall({
+        id: 'daily-review-join-source',
+        callKind: 'main',
+        callId: 'daily-review-join-source',
+        connectionSlug: 'test',
+        providerId: 'test',
+        modelId: 'test',
+        inputTokens: 1,
+        outputTokens: 1,
+        cacheHitInputTokens: 0,
+        cacheMissInputTokens: 1,
+        cachedInputTokens: 0,
+        cacheWriteInputTokens: 0,
+        reasoningTokens: 0,
+        totalTokens: 2,
+        costUsd: 0,
+        latencyMs: 1,
+        status: 'success',
+        startedAt: now,
+        date: new Date(now).toISOString().slice(0, 10),
+        ts: now,
+      });
+      const run = {
+        kind: 'run' as const,
+        range: 1 as const,
+        offsetDays: 0,
+        modelKeyOverride: 'provider::model-a',
+        replaceExisting: true,
+      };
+      // Two Clients ask for the same archive at once. Whichever settles first
+      // releases the other's lagging read, so the laggard only reaches its own
+      // bookkeeping after the leader has already published. A rejected leader
+      // releases it too; otherwise close() would wait on the laggard forever.
+      const first = coordinator.handlers['daily-review.mutate'](run, CONTEXT);
+      const second = coordinator.handlers['daily-review.mutate'](run, CONTEXT);
+      const release = () => releaseLaggingRead?.();
+      void Promise.race([first, second]).then(release, release);
+      const [firstResult, secondResult] = await Promise.all([first, second]);
+      assert.equal(modelCalls, 1);
+      assert.deepEqual(secondResult, firstResult);
+    },
+    {
+      generate: async () => {
+        modelCalls += 1;
+        return { ok: false, errorClass: 'configuration' };
+      },
+    },
+    true,
+    {
+      list: async () => {
+        listCalls += 1;
+        if (listCalls === 2) await laggingRead;
+        return [];
+      },
+    },
+  );
+});
+
+test('Daily Review regenerates for a replace that arrives during a non-replacing run', async () => {
+  await withCoordinator(async ({ coordinator, store }) => {
+    const archiveId = dailyReviewArchiveId(localDayBoundsForInstant(Date.now()), 1);
+    const existing = await store.publishArchive(archive(archiveId), 180);
+    const run = {
+      kind: 'run' as const,
+      range: 1 as const,
+      offsetDays: 0,
+      modelKeyOverride: '',
+      replaceExisting: false,
+    };
+    const keep = coordinator.handlers['daily-review.mutate'](run, CONTEXT);
+    const replace = coordinator.handlers['daily-review.mutate'](
+      { ...run, replaceExisting: true },
+      CONTEXT,
+    );
+    const [kept, replaced] = await Promise.all([keep, replace]);
+    assert.deepEqual(kept, { ok: true, result: { kind: 'archive', archive: existing } });
+    assert.equal(replaced.ok, true);
+    if (!replaced.ok || replaced.result.kind !== 'archive') return;
+    assert.equal(replaced.result.archive.status, 'no_data');
+    assert.deepEqual(await store.getArchive(archiveId), replaced.result.archive);
+  });
+});
+
+test('Daily Review lets a non-replacing run join a replace already in flight', async () => {
+  await withCoordinator(async ({ coordinator, store }) => {
+    const archiveId = dailyReviewArchiveId(localDayBoundsForInstant(Date.now()), 1);
+    await store.publishArchive(archive(archiveId), 180);
+    const run = {
+      kind: 'run' as const,
+      range: 1 as const,
+      offsetDays: 0,
+      modelKeyOverride: '',
+      replaceExisting: true,
+    };
+    const replace = coordinator.handlers['daily-review.mutate'](run, CONTEXT);
+    const keep = coordinator.handlers['daily-review.mutate'](
+      { ...run, replaceExisting: false },
+      CONTEXT,
+    );
+    const [replaced, kept] = await Promise.all([replace, keep]);
+    assert.equal(replaced.ok, true);
+    if (!replaced.ok || replaced.result.kind !== 'archive') return;
+    assert.equal(replaced.result.archive.status, 'no_data');
+    assert.deepEqual(kept, replaced);
+  });
+});
+
 test('Daily Review does not coalesce cron and manual archive provenance', async () => {
   let releaseModel: (() => void) | undefined;
   let notifyModelStarted: (() => void) | undefined;
@@ -275,6 +398,9 @@ async function withCoordinator(
     generate: async () => ({ ok: false, errorClass: 'configuration' }),
   },
   recoverBeforeRun = true,
+  sessions: ConstructorParameters<typeof HostDailyReviewCoordinator>[0]['sessions'] = {
+    list: async () => [],
+  },
 ): Promise<void> {
   const base = await mkdtemp(join(tmpdir(), 'maka-daily-review-coordinator-'));
   const root = join(base, 'interactive');
@@ -291,7 +417,7 @@ async function withCoordinator(
   const coordinator = new HostDailyReviewCoordinator({
     store,
     usage,
-    sessions: { list: async () => [] },
+    sessions,
     model,
     acquireResidency: () => ({ release: () => undefined }),
     requestDrain: () => {

packages/runtime-host/src/__tests__/owned-candidate.test.ts

@@ -258,7 +258,11 @@ test('owned Host exits promptly after its first connection closes', async () =>
   assert.equal(result.kind, 'connected', connectFailure(result));
   if (result.kind !== 'connected') return;
   await result.connection.close();
-  assert.equal(await result.host.settle(500), true);
+  // Prompt means the owned launch's idleGraceMs of 0, as opposed to the 30 s
+  // default grace, so the bound only has to sit well below that. Shutdown takes
+  // about 30 ms on an idle machine and stretches past 500 ms under a full CI
+  // suite while still exiting cleanly: the Host is starved, not stuck.
+  assert.equal(await result.host.settle(5_000), true);
 });
 
 test('an exited owned Candidate permits one real successor in the same election', {

packages/runtime-host/src/server/daily-review-coordinator.ts

@@ -30,6 +30,7 @@ import {
   type DailyReviewArchiveSectionContent,
   type DailyReviewRange,
   type DailyReviewSummary,
+  type DayRangeMs,
 } from '@maka/core/daily-review';
 import { collapseSessionRevisions } from '@maka/core/session-revisions';
 import { mergeUsageBuckets, mergeUsageSummary } from '@maka/core/usage-ledger-merge';
@@ -90,8 +91,9 @@ export class HostDailyReviewCoordinator {
   readonly #inFlight = new Map<
     string,
     {
-      readonly modelKey: string;
+      readonly modelKeyOverride: string;
       readonly trigger: 'cron' | 'manual';
+      readonly replaceExisting: boolean;
       readonly promise: Promise<DailyReviewArchive>;
     }
   >();
@@ -170,11 +172,13 @@ export class HostDailyReviewCoordinator {
           const snapshot = await this.#store.readConfig();
           return querySuccess({ kind: 'config', ...snapshot });
         }
-        case 'summary':
+        case 'summary': {
+          const now = this.#now();
           return querySuccess({
             kind: 'summary',
-            summary: await this.#buildSummary(input.offsetDays, input.daySpan),
+            summary: await this.#buildSummary(dayRange(now, input.offsetDays, input.daySpan), now),
           });
+        }
         case 'archives': {
           const beforeArchiveId = input.beforeArchiveId;
           const page = await this.#store.listArchivePage(beforeArchiveId, input.limit);
@@ -247,13 +251,7 @@ export class HostDailyReviewCoordinator {
     }
   }
 
-  async #buildSummary(offsetDays: number, daySpan: number): Promise<DailyReviewSummary> {
-    const offset = Math.trunc(offsetDays);
-    const span = Math.max(1, Math.min(30, Math.trunc(daySpan)));
-    const now = this.#now();
-    const endDay = offset === 0 ? localDayBoundsForInstant(now) : localDayBoundsAt(now, offset);
-    const startDay = localDayBoundsAt(endDay.fromMs, -(span - 1));
-    const range = { fromMs: startDay.fromMs, toMs: endDay.toMs };
+  async #buildSummary(range: DayRangeMs, now: number): Promise<DailyReviewSummary> {
     const query = dailyUsageQuery(range);
     const canonical = await readCompleteCanonicalUsage(this.#usage, query, now);
     const [usageSummary, toolBuckets, modelBuckets, sessions] = await Promise.all([
@@ -285,21 +283,35 @@ export class HostDailyReviewCoordinator {
     readonly trigger: 'cron' | 'manual';
     readonly replaceExisting: boolean;
   }): Promise<DailyReviewArchive> {
-    const summary = await this.#buildSummary(input.offsetDays, input.range);
-    const archiveId = dailyReviewArchiveId(summary.day, input.range);
-    const existing = await this.#store.getArchive(archiveId);
-    if (existing && !input.replaceExisting) return existing;
-    const config = await this.#store.readConfig();
-    const modelKey = input.modelKeyOverride.trim() || config.config.modelKey;
+    const now = this.#now();
+    const day = dayRange(now, input.offsetDays, input.range);
+    const archiveId = dailyReviewArchiveId(day, input.range);
+    const modelKeyOverride = input.modelKeyOverride.trim();
+    // Claim the archive before the first await. Two Clients asking for the
+    // same archive at once share one generation; a claim taken only after the
+    // reads let the second request slip past a first that had already
+    // published, and each Client then saw its own archive. Requests match on
+    // the override they asked for, not the resolved model key: resolving it
+    // needs the config read, which would put the claim back after an await.
     const inFlight = this.#inFlight.get(archiveId);
     if (inFlight) {
-      if (inFlight.modelKey === modelKey && inFlight.trigger === input.trigger) {
-        return inFlight.promise;
+      if (inFlight.modelKeyOverride !== modelKeyOverride || inFlight.trigger !== input.trigger) {
+        throw new DailyReviewRunConflictError(archiveId);
       }
-      throw new DailyReviewRunConflictError(archiveId);
+      if (!input.replaceExisting || inFlight.replaceExisting) return inFlight.promise;
+      // A non-replacing leader may hand back an archive it merely found. A
+      // replace must not inherit that, so it waits its turn and claims for
+      // itself.
+      await inFlight.promise.catch(() => undefined);
+      return this.#run(input);
     }
-    const pending = this.#generateArchive(archiveId, summary, modelKey, input);
-    const entry = { modelKey, trigger: input.trigger, promise: pending };
+    const pending = this.#generateArchive(archiveId, day, now, modelKeyOverride, input);
+    const entry = {
+      modelKeyOverride,
+      trigger: input.trigger,
+      replaceExisting: input.replaceExisting,
+      promise: pending,
+    };
     this.#inFlight.set(archiveId, entry);
     try {
       return await pending;
@@ -310,13 +322,20 @@ export class HostDailyReviewCoordinator {
 
   async #generateArchive(
     archiveId: string,
-    summary: DailyReviewSummary,
-    modelKey: string,
+    day: DayRangeMs,
+    now: number,
+    modelKeyOverride: string,
     input: {
       readonly range: DailyReviewRange;
       readonly trigger: 'cron' | 'manual';
+      readonly replaceExisting: boolean;
     },
   ): Promise<DailyReviewArchive> {
+    const summary = await this.#buildSummary(day, now);
+    const existing = await this.#store.getArchive(archiveId);
+    if (existing && !input.replaceExisting) return existing;
+    const config = await this.#store.readConfig();
+    const modelKey = modelKeyOverride || config.config.modelKey;
     const base = {
       id: archiveId,
       day: summary.day,
@@ -457,6 +476,14 @@ export class HostDailyReviewCoordinator {
   }
 }
 
+function dayRange(nowMs: number, offsetDays: number, daySpan: number): DayRangeMs {
+  const offset = Math.trunc(offsetDays);
+  const span = Math.max(1, Math.min(30, Math.trunc(daySpan)));
+  const endDay = offset === 0 ? localDayBoundsForInstant(nowMs) : localDayBoundsAt(nowMs, offset);
+  const startDay = localDayBoundsAt(endDay.fromMs, -(span - 1));
+  return { fromMs: startDay.fromMs, toMs: endDay.toMs };
+}
+
 function scheduledTimeHasPassed(nowMs: number, executeTime: string): boolean {
   const now = new Date(nowMs);
   const [hours, minutes] = executeTime.split(':').map(Number);