diff --git a/packages/runtime-host/src/__tests__/daily-review-coordinator.test.ts b/packages/runtime-host/src/__tests__/daily-review-coordinator.test.ts index c781604f41..9df8437004 100644 --- a/packages/runtime-host/src/__tests__/daily-review-coordinator.test.ts +++ b/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((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[0]['sessions'] = { + list: async () => [], + }, ): Promise { 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: () => { diff --git a/packages/runtime-host/src/__tests__/owned-candidate.test.ts b/packages/runtime-host/src/__tests__/owned-candidate.test.ts index 83f9e212e7..3c264f145d 100644 --- a/packages/runtime-host/src/__tests__/owned-candidate.test.ts +++ b/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', { diff --git a/packages/runtime-host/src/server/daily-review-coordinator.ts b/packages/runtime-host/src/server/daily-review-coordinator.ts index 864e507b01..60f0564664 100644 --- a/packages/runtime-host/src/server/daily-review-coordinator.ts +++ b/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; } >(); @@ -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 { - 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 { 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 { - 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 { + 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);