From 20dfd61bc284ff6bddd8bcfc924512ce76903389 Mon Sep 17 00:00:00 2001 From: RSI smoke Date: Mon, 12 Oct 2026 15:47:13 -0600 Subject: [PATCH] Fix live pilot resume after shared-ledger copy (#11) collectLiveDevelopment/ Heldout computed the next attempt epoch from the fresh experiment journal alone, so resuming in a new output directory after copying global-budget.jsonl (the documented frozen-protocol step) aborted with 'stale attempt epoch'. The next epoch now advances past the highest epoch recorded for the run in either the shared global ledger or the experiment journal. Adds a regression test seeding a copied ledger with epochs 1-2 and asserting the resumed attempt starts at epoch 3. --- scripts/monitor-pilot/live-runner.ts | 18 ++++++++++++++-- src/monitoring/__tests__/live-runner.test.ts | 22 ++++++++++++++++++++ 2 files changed, 38 insertions(+), 2 deletions(-) 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()