diff --git a/scripts/monitor-pilot/live-runner.ts b/scripts/monitor-pilot/live-runner.ts index 7ba74a4..c1346e0 100644 --- a/scripts/monitor-pilot/live-runner.ts +++ b/scripts/monitor-pilot/live-runner.ts @@ -779,6 +779,16 @@ function completedRecords(journal: PilotJournal): Map { return done } +function latestEpochByRun(journal: PilotJournal): Map { + const epochs = new Map() + for (const record of journal.list()) { + if (record.runId !== undefined && record.epoch !== undefined) { + epochs.set(record.runId, Math.max(epochs.get(record.runId) ?? record.epoch, record.epoch)) + } + } + return epochs +} + function billingTotals(journal: PilotJournal): LiveBillingTotals { const records = journal.list() const accepted = records.filter((record) => record.type === "llm_response_billed") @@ -862,6 +872,8 @@ export async function collectLiveDevelopment(options: LiveRunnerOptions = {}): P await fs.chmod(global.filePath, 0o600).catch(async (error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") throw error }) const client = options.client ?? await createProductionClient() const done = completedRecords(journal) + const globalEpochByRun = latestEpochByRun(global) + const localEpochByRun = latestEpochByRun(journal) const allRows: LabeledWindow[] = [] let completedRuns = 0 let stopReason: string | undefined @@ -880,8 +892,8 @@ export async function collectLiveDevelopment(options: LiveRunnerOptions = {}): P stopReason = `STOP: ${slot.slotKey} has an aborted/uncertain attempt; explicit retry is required.` break } - const epoch = priorAttempts.length + 1 const runId = slot.slotKey + const epoch = Math.max(priorAttempts.length, globalEpochByRun.get(runId) ?? 0, localEpochByRun.get(runId) ?? 0) + 1 const attemptId = randomUUID() const reservation: BudgetReservation = { attemptId, runId, maxTokens: LIVE_BUDGET.maxTokens, maxWallMs: LIVE_BUDGET.maxWallMs } const used = globalBudgetUsage(global) @@ -1097,6 +1109,8 @@ export async function collectLiveHeldout(options: LiveRunnerOptions = {}): Promi await assertRegularFileNoSymlink(global.filePath) await global.open() const done = completedRecords(journal) + const globalEpochByRun = latestEpochByRun(global) + const localEpochByRun = latestEpochByRun(journal) const outstanding = slots.filter((slot) => !done.has(slot.slotKey)).length const used = globalBudgetUsage(global) let stopReason: string | undefined @@ -1117,8 +1131,8 @@ export async function collectLiveHeldout(options: LiveRunnerOptions = {}): Promi stopReason = `STOP: held-out slot ${slot.slotKey} has an aborted/uncertain attempt; explicit retry is required.` break } - const epoch = priorAttempts.length + 1 const runId = `heldout-${slot.fixture.id}-${slot.seed}-${slot.arm}` + const epoch = Math.max(priorAttempts.length, globalEpochByRun.get(runId) ?? 0, localEpochByRun.get(runId) ?? 0) + 1 const attemptId = randomUUID() const reservation: BudgetReservation = { attemptId, runId, maxTokens: LIVE_BUDGET.maxTokens, maxWallMs: LIVE_BUDGET.maxWallMs } const current = globalBudgetUsage(global) diff --git a/src/monitoring/__tests__/live-runner.test.ts b/src/monitoring/__tests__/live-runner.test.ts index 5f549ed..c901a55 100644 --- a/src/monitoring/__tests__/live-runner.test.ts +++ b/src/monitoring/__tests__/live-runner.test.ts @@ -280,6 +280,27 @@ async function testUncertainRequestIsNotAutomaticallyReplayed(): Promise { } finally { await fs.rm(directory, { recursive: true, force: true }) } } +async function testCopiedGlobalLedgerEpochsDoNotBlockResume(): Promise { + const directory = await scratch() + try { + const firstSlot = LIVE_DEVELOPMENT_SLOTS[0]! + const global = new PilotJournal(path.join(directory, "global-budget.jsonl")) + await global.open() + for (const epoch of [1, 2]) { + const attemptId = `prior-attempt-${epoch}` + await global.append("budget_reserved", LIVE_EXPERIMENT_ID, { attemptId, runId: firstSlot.slotKey, maxTokens: 20000, maxWallMs: 300000 }, { runId: firstSlot.slotKey, attemptId, epoch }) + await global.append("budget_settled", LIVE_EXPERIMENT_ID, { attemptId, actualTokens: 100, actualWallMs: 100 }, { runId: firstSlot.slotKey, attemptId, epoch }) + } + const client = new MockBillingClient([{ name: "attempt_completion", args: JSON.stringify({ result: "done" }) }]) + const summary = await collectLiveDevelopment({ outputDir: directory, client }) + assert.equal(summary.completedRuns, 32) + const journal = new PilotJournal(summary.journalPath) + await journal.open() + const first = journal.list().find((record) => record.type === "attempt_started" && record.payload.slotKey === firstSlot.slotKey)! + assert.equal(first.epoch, 3, "resume advances past epochs already recorded in the copied shared ledger") + } finally { await fs.rm(directory, { recursive: true, force: true }) } +} + async function testKnownPerRunAdmissionStopCompletesAndSettlesTrajectory(): Promise { const directory = await scratch() try { @@ -485,6 +506,7 @@ await testFiniteBillingIsCountedWhenAdmissionAborts() await testAdmittedRequestWithoutResponseIsShownAsUnresolved() await testResponseWithoutFiniteGenerationCostIsUnbilled() await testUncertainRequestIsNotAutomaticallyReplayed() +await testCopiedGlobalLedgerEpochsDoNotBlockResume() await testKnownPerRunAdmissionStopCompletesAndSettlesTrajectory() await testCompletionOverrunIsJournaledBeforeAnyToolDispatch() await testHeldoutCannotStartWithoutQualifiedDevelopment()