diff --git a/scripts/rsi-proof-demonstration.ts b/scripts/rsi-proof-demonstration.ts index af4cda1..1134b62 100644 --- a/scripts/rsi-proof-demonstration.ts +++ b/scripts/rsi-proof-demonstration.ts @@ -16,7 +16,7 @@ import { analyzeProofDemonstration, verifyProofDemonstrationArtifactBytes } from import { workerHarnessRuntimeIdentityFromEnvironment } from "../src/rsi/worker-harness.js" import { proofCampaignManifestDigest } from "../src/rsi/proof-campaign.js" import { resolveRoles } from "../src/rsi/roles.js" -import type { CandidateRecord, ProofBenchmarkFixtureBinding, RsiConfig, RsiRunRecord } from "../src/rsi/types.js" +import type { CandidateRecord, ProofBenchmarkFixtureBinding, RsiConfig, RsiPreparationReceipt, RsiRunRecord } from "../src/rsi/types.js" interface Options { command: "estimate" | "plan" | "verify" | "analyze" | "engineering" | "execute" @@ -440,7 +440,7 @@ export function formatAnalysisReport(report: ReturnType { +export async function buildProofPlanPreview(options: Options, env: NodeJS.ProcessEnv = process.env, fetcher: typeof fetch = fetch, onPreparationReceipt?: (receipt: RsiPreparationReceipt) => void | Promise): Promise { if (options.command !== "estimate" && options.command !== "plan" && options.command !== "engineering") throw new Error("plan options are required") if (options.command === "engineering" && options.seeds.length !== 1) throw new Error("engineering requires exactly one explicit --seed") if (options.command !== "engineering" && options.seeds.length !== 4) throw new Error("provide one --seed for engineering and exactly three --campaign-seed values") @@ -448,47 +448,91 @@ export async function buildProofPlanPreview(options: Options, env: NodeJS.Proces if (new Set(seeds).size !== 4) throw new Error("engineering and campaign seeds must be distinct") const repo = path.resolve(options.repo) const rootCommit = await ensureCleanRoot(repo) - const runtime = workerHarnessRuntimeIdentityFromEnvironment(env) - try { execFileSync("git", ["cat-file", "-e", `${runtime.harnessCommit}^{commit}`], { cwd: repo, stdio: "ignore" }) } - catch { throw new Error("fixed-control runtime harness commit must be present in the root repository") } - const fixtureRoot = options.benchmarkFixturesRoot ?? path.join(repo, "fixtures", "rsi-benchmark") - const { binding: benchmarkFixtures, loaded } = await loadProofBenchmarkFixtures(fixtureRoot, repo) - const proofConfig = proofConfigFromOptions({ ...options, repo, seeds }, rootCommit, benchmarkFixtures) - const profile = openRouterProfileById(options.openRouterProfile ?? RSI_DEFAULT_OPENROUTER_PROFILE_ID) - const baseConfig = withEvaluatorIntegrityPin({ - ...proofConfig, - archiveDir: options.archiveDir ?? proofConfig.archiveDir, - model: profile.model, - roles: { worker: { provider: "openrouter" as const, model: profile.model, ...(profile.id !== RSI_DEFAULT_OPENROUTER_PROFILE_ID ? { openRouterProfile: profile.id } : {}) } }, - mutationTask: options.mutationTask, - workerHarnessRuntime: runtime, - }, options.evaluatorTrustedKeySha256) - assertProofWorkerProfile(baseConfig, profile, env) - const configs = seeds.map((seed, index) => ({ ...baseConfig, generations: index === 0 ? 2 : 3, seed })) as [typeof baseConfig, typeof baseConfig, typeof baseConfig, typeof baseConfig] - const config = configs[0] - const modelKey = env.HEADLESSCODE_OPENROUTER_API_KEY?.trim() ?? "" - const price = await fetchOpenRouterEndpointPrice(profile, modelKey, fetcher) - const mutationEnv = openRouterAllocatedEnvironment(env, "mutation") - const benchmarkEnv = openRouterAllocatedEnvironment(env, "benchmark") - const mutationLimits = openRouterBrokerLimitsFromEnvironment(mutationEnv, "plan-mutation", { maxCalls: config.maxIterations, ...(config.maxTokens ? { maxOutputTokensPerCall: config.maxTokens } : {}), maxDurationMs: config.commandTimeoutMs }) - const benchmarkLimits = openRouterBrokerLimitsFromEnvironment(benchmarkEnv, "plan-benchmark") - if (mutationLimits.maxInputPricePerMillionUsd < price.inputPricePerMillionUsd || mutationLimits.maxOutputPricePerMillionUsd < price.outputPricePerMillionUsd || benchmarkLimits.maxInputPricePerMillionUsd < price.inputPricePerMillionUsd || benchmarkLimits.maxOutputPricePerMillionUsd < price.outputPricePerMillionUsd) throw new Error("configured broker price ceiling is below the live pinned endpoint rate") - const plan = planProofDemonstration({ config, manifest: loaded.manifest, mutationLimits, benchmarkLimits }) - return { - createdAt: new Date().toISOString(), rootCommit, benchmarkFixtures, benchmarkManifestSha256: loaded.manifestDigest, - model: { provider: "openshell-openrouter", requestedId: profile.model, ...(profile.id !== RSI_DEFAULT_OPENROUTER_PROFILE_ID ? { profileId: profile.id } : {}), sampling: { temperature: 0, think: false, seed: "provider-default" } }, - endpointPrice: price, - priceCeilingsPerMillionUsd: { input: benchmarkLimits.maxInputPricePerMillionUsd, output: benchmarkLimits.maxOutputPricePerMillionUsd }, - runtime, - execution: { - configs, configSha256: frozenProofConfigDigest(configs), analysisSourceSha256: await analysisDigest(repo), - inferenceLimits: { mutation: mutationLimits, benchmark: benchmarkLimits }, - taskIdentities: loaded.manifest.tasks.map((task) => ({ id: task.id, split: task.split, family: task.family, seed: task.seed, startDigest: task.startDigest, verifierDigest: task.verifierDigest, knownGoodDigest: task.knownGoodDigest, knownBadDigest: task.knownBadDigest, ceilings: task.ceilings })), - }, - seeds: { engineering: seeds[0]!, campaigns: [seeds[1]!, seeds[2]!, seeds[3]!] }, - campaignIds: { engineering: campaignId(seeds[0]!), campaigns: [campaignId(seeds[1]!), campaignId(seeds[2]!), campaignId(seeds[3]!)] }, - analysis: { bootstrapSeed: "rsi-proof-task-cluster-bootstrap-v1", bootstrapReplicates: 10_000, confirmationRetentionTaskIds: [...RSI_CONFIRMATION_RETENTION_TASK_IDS] }, - plan, + const preparationStartedAt = process.hrtime.bigint() + const preparationPhaseStartedAt = new Map() + let previewFinalStateEmitted = false + let previewManifestSha256: string | undefined + const emitPreparationReceipt = async (phase: RsiPreparationReceipt["phase"], state: RsiPreparationReceipt["state"], manifestSha256?: string): Promise => { + if (!onPreparationReceipt) return + const timestamp = process.hrtime.bigint() + if (state === "started") preparationPhaseStartedAt.set(phase, timestamp) + const phaseStart = preparationPhaseStartedAt.get(phase) ?? timestamp + const elapsedMs = (value: bigint) => Math.min(Number.MAX_SAFE_INTEGER, Math.max(0, Math.round(Number(value) / 1_000_000))) + const receipt: RsiPreparationReceipt = { + schemaVersion: 1, kind: "rsi-preparation", phase, state, sourceCommit: rootCommit, + seedSha256: sha256(options.seeds[0]!), campaignId: campaignId(options.seeds[0]!), + ...(manifestSha256 ? { manifestSha256 } : {}), phaseElapsedMs: state === "started" ? 0 : elapsedMs(timestamp - phaseStart), + elapsedSinceStartMs: elapsedMs(timestamp - preparationStartedAt), readiness: "preparing", inferenceAdmittedByReceipt: false, + } + await onPreparationReceipt(receipt) + if (phase === "proof-plan-preview" && (state === "completed" || state === "failed")) previewFinalStateEmitted = true + } + await emitPreparationReceipt("proof-plan-preview", "started") + try { + const runtime = workerHarnessRuntimeIdentityFromEnvironment(env) + try { execFileSync("git", ["cat-file", "-e", `${runtime.harnessCommit}^{commit}`], { cwd: repo, stdio: "ignore" }) } + catch { throw new Error("fixed-control runtime harness commit must be present in the root repository") } + const fixtureRoot = options.benchmarkFixturesRoot ?? path.join(repo, "fixtures", "rsi-benchmark") + await emitPreparationReceipt("fixture-validation", "started") + let fixtureLoad: Awaited> + try { fixtureLoad = await loadProofBenchmarkFixtures(fixtureRoot, repo) } + catch (error) { + try { await emitPreparationReceipt("fixture-validation", "failed") } catch { /* preserve the original validation error */ } + throw error + } + const { binding: benchmarkFixtures, loaded } = fixtureLoad + previewManifestSha256 = loaded.manifestDigest + await emitPreparationReceipt("fixture-validation", "completed", loaded.manifestDigest) + const proofConfig = proofConfigFromOptions({ ...options, repo, seeds }, rootCommit, benchmarkFixtures) + const profile = openRouterProfileById(options.openRouterProfile ?? RSI_DEFAULT_OPENROUTER_PROFILE_ID) + const baseConfig = withEvaluatorIntegrityPin({ + ...proofConfig, + archiveDir: options.archiveDir ?? proofConfig.archiveDir, + model: profile.model, + roles: { worker: { provider: "openrouter" as const, model: profile.model, ...(profile.id !== RSI_DEFAULT_OPENROUTER_PROFILE_ID ? { openRouterProfile: profile.id } : {}) } }, + mutationTask: options.mutationTask, + workerHarnessRuntime: runtime, + }, options.evaluatorTrustedKeySha256) + assertProofWorkerProfile(baseConfig, profile, env) + const configs = seeds.map((seed, index) => ({ ...baseConfig, generations: index === 0 ? 2 : 3, seed })) as [typeof baseConfig, typeof baseConfig, typeof baseConfig, typeof baseConfig] + const config = configs[0] + const modelKey = env.HEADLESSCODE_OPENROUTER_API_KEY?.trim() ?? "" + await emitPreparationReceipt("pricing-admission", "started", loaded.manifestDigest) + let price: Awaited> + try { price = await fetchOpenRouterEndpointPrice(profile, modelKey, fetcher) } + catch (error) { + try { await emitPreparationReceipt("pricing-admission", "failed", loaded.manifestDigest) } catch { /* preserve the original pricing error */ } + throw error + } + await emitPreparationReceipt("pricing-admission", "completed", loaded.manifestDigest) + const mutationEnv = openRouterAllocatedEnvironment(env, "mutation") + const benchmarkEnv = openRouterAllocatedEnvironment(env, "benchmark") + const mutationLimits = openRouterBrokerLimitsFromEnvironment(mutationEnv, "plan-mutation", { maxCalls: config.maxIterations, ...(config.maxTokens ? { maxOutputTokensPerCall: config.maxTokens } : {}), maxDurationMs: config.commandTimeoutMs }) + const benchmarkLimits = openRouterBrokerLimitsFromEnvironment(benchmarkEnv, "plan-benchmark") + if (mutationLimits.maxInputPricePerMillionUsd < price.inputPricePerMillionUsd || mutationLimits.maxOutputPricePerMillionUsd < price.outputPricePerMillionUsd || benchmarkLimits.maxInputPricePerMillionUsd < price.inputPricePerMillionUsd || benchmarkLimits.maxOutputPricePerMillionUsd < price.outputPricePerMillionUsd) throw new Error("configured broker price ceiling is below the live pinned endpoint rate") + const plan = planProofDemonstration({ config, manifest: loaded.manifest, mutationLimits, benchmarkLimits }) + await emitPreparationReceipt("proof-plan-preview", "completed", loaded.manifestDigest) + return { + createdAt: new Date().toISOString(), rootCommit, benchmarkFixtures, benchmarkManifestSha256: loaded.manifestDigest, + model: { provider: "openshell-openrouter", requestedId: profile.model, ...(profile.id !== RSI_DEFAULT_OPENROUTER_PROFILE_ID ? { profileId: profile.id } : {}), sampling: { temperature: 0, think: false, seed: "provider-default" } }, + endpointPrice: price, + priceCeilingsPerMillionUsd: { input: benchmarkLimits.maxInputPricePerMillionUsd, output: benchmarkLimits.maxOutputPricePerMillionUsd }, + runtime, + execution: { + configs, configSha256: frozenProofConfigDigest(configs), analysisSourceSha256: await analysisDigest(repo), + inferenceLimits: { mutation: mutationLimits, benchmark: benchmarkLimits }, + taskIdentities: loaded.manifest.tasks.map((task) => ({ id: task.id, split: task.split, family: task.family, seed: task.seed, startDigest: task.startDigest, verifierDigest: task.verifierDigest, knownGoodDigest: task.knownGoodDigest, knownBadDigest: task.knownBadDigest, ceilings: task.ceilings })), + }, + seeds: { engineering: seeds[0]!, campaigns: [seeds[1]!, seeds[2]!, seeds[3]!] }, + campaignIds: { engineering: campaignId(seeds[0]!), campaigns: [campaignId(seeds[1]!), campaignId(seeds[2]!), campaignId(seeds[3]!)] }, + analysis: { bootstrapSeed: "rsi-proof-task-cluster-bootstrap-v1", bootstrapReplicates: 10_000, confirmationRetentionTaskIds: [...RSI_CONFIRMATION_RETENTION_TASK_IDS] }, + plan, + } + } catch (error) { + if (!previewFinalStateEmitted) { + try { await emitPreparationReceipt("proof-plan-preview", "failed", previewManifestSha256) } catch { /* progress reporting must not replace the original preparation error */ } + } + throw error } } @@ -579,7 +623,8 @@ export async function writePriorIncompleteEngineeringStopReport( export async function executeEngineeringPair(options: Options, env: NodeJS.ProcessEnv = process.env, runner: typeof runRsi = runRsi): Promise { if (options.command !== "engineering") throw new Error("engineering command is required") if (options.execute) requireEvaluatorIntegrityOptions(options) - const preview = await buildProofPlanPreview(options, env) + const receiptSink = (receipt: RsiPreparationReceipt): void => { process.stdout.write(`RSI_PREPARATION_RECEIPT ${JSON.stringify(receipt)}\n`) } + const preview = await buildProofPlanPreview(options, env, fetch, receiptSink) const required = preview.plan.engineering.rootSpendCeilingUsd if (!options.execute) { if (options.operatorCapUsd !== undefined && parseUsd(options.operatorCapUsd) + 1e-9 < required) throw new Error(`operator cap $${options.operatorCapUsd.toFixed(6)} is below engineering-only root reservation $${required.toFixed(6)}`) @@ -593,7 +638,21 @@ export async function executeEngineeringPair(options: Options, env: NodeJS.Proce const benchmarkAllocation = preview.execution.inferenceLimits.benchmark.maxCampaignSpendUsd if (mutationAllocation + benchmarkAllocation > cap + 1e-9 || mutationAllocation + benchmarkAllocation + 1e-9 < required) throw new Error("configured mutation and benchmark allocations must fit the operator cap and fund the engineering-only root reservation") if (await ensureCleanRoot(options.repo) !== preview.rootCommit) throw new Error("engineering execution root changed after its estimate") - const config = await validateRuntimeEvaluatorIntegrity(preview.execution.configs[0], options) + const evaluatorStartedAt = process.hrtime.bigint() + const evaluatorReceipt = (state: RsiPreparationReceipt["state"]): RsiPreparationReceipt => { + const elapsed = Math.min(Number.MAX_SAFE_INTEGER, Math.max(0, Math.round(Number(process.hrtime.bigint() - evaluatorStartedAt) / 1_000_000))) + return { + schemaVersion: 1, kind: "rsi-preparation", phase: "evaluator-admission", state, + sourceCommit: preview.rootCommit, seedSha256: sha256(preview.seeds.engineering), campaignId: preview.campaignIds.engineering, + manifestSha256: preview.benchmarkManifestSha256, phaseElapsedMs: state === "started" ? 0 : elapsed, + elapsedSinceStartMs: elapsed, readiness: "preparing", inferenceAdmittedByReceipt: false, + } + } + receiptSink(evaluatorReceipt("started")) + let config: RsiConfig + try { config = await validateRuntimeEvaluatorIntegrity(preview.execution.configs[0], options) } + catch (error) { receiptSink(evaluatorReceipt("failed")); throw error } + receiptSink(evaluatorReceipt("completed")) assertProofWorkerProfile(config, openRouterProfileById(preview.model.profileId ?? RSI_DEFAULT_OPENROUTER_PROFILE_ID), process.env) const campaign = campaignId(preview.seeds.engineering) const priorIncomplete = await writePriorIncompleteEngineeringStopReport(options, config, campaign, preview.rootCommit, preview.seeds.engineering, preview.runtime) diff --git a/src/rsi/__tests__/controller-fleet.test.ts b/src/rsi/__tests__/controller-fleet.test.ts index 70dd285..1e797e4 100644 --- a/src/rsi/__tests__/controller-fleet.test.ts +++ b/src/rsi/__tests__/controller-fleet.test.ts @@ -12,7 +12,7 @@ import { FileRsiArtifactStore } from "../artifact-store.js" import { PostgresProofCampaign } from "../proof-campaign.js" import { PostgresRsiJobQueue, type FleetJob, type FleetJobPayload } from "../postgres-queue.js" import { newCandidate } from "../workspace.js" -import type { CandidateRecord, RsiConfig, RsiRunRecord, RsiHooks, TrialResult } from "../types.js" +import type { CandidateRecord, RsiConfig, RsiPreparationReceipt, RsiRunRecord, RsiHooks, TrialResult } from "../types.js" import { fakeDevelopmentBenchmarkHook } from "./development-proof-fixture.js" import { fixtureDevelopmentProofProtocol } from "./development-proof-fixture.js" @@ -314,6 +314,70 @@ async function syntheticEvaluatorTrust(config: RsiConfig): Promise<{ config: Rsi } } +test("paired startup reports slow preparation and stops before admission when startup cannot proceed", async () => { + const { root, config: baseConfig } = await setup() + const { config: trustedConfig, cleanup } = await syntheticEvaluatorTrust(baseConfig) + const budgetNames = [ + "HEADLESSCODE_RSI_OPENROUTER_MAX_INPUT_PRICE_PER_MILLION_USD", "HEADLESSCODE_RSI_OPENROUTER_MAX_OUTPUT_PRICE_PER_MILLION_USD", + "HEADLESSCODE_RSI_OPENROUTER_MAX_CALLS", "HEADLESSCODE_RSI_OPENROUTER_MAX_INPUT_TOKENS_PER_CALL", "HEADLESSCODE_RSI_OPENROUTER_MAX_OUTPUT_TOKENS_PER_CALL", "HEADLESSCODE_RSI_OPENROUTER_MAX_TOTAL_TOKENS", + "HEADLESSCODE_RSI_OPENROUTER_MAX_CALL_SPEND_USD", "HEADLESSCODE_RSI_OPENROUTER_MAX_JOB_SPEND_USD", "HEADLESSCODE_RSI_OPENROUTER_MAX_CAMPAIGN_SPEND_USD", + "HEADLESSCODE_RSI_OPENROUTER_MUTATION_BUDGET_USD", "HEADLESSCODE_RSI_OPENROUTER_BENCHMARK_BUDGET_USD", + "HEADLESSCODE_RSI_OPENROUTER_MUTATION_MAX_TOTAL_TOKENS", "HEADLESSCODE_RSI_OPENROUTER_MUTATION_MAX_JOB_SPEND_USD", + "HEADLESSCODE_RSI_OPENROUTER_BENCHMARK_MAX_TOTAL_TOKENS", "HEADLESSCODE_RSI_OPENROUTER_BENCHMARK_MAX_JOB_SPEND_USD", + ] + const saved = Object.fromEntries(budgetNames.map((name) => [name, process.env[name]])) + const values = { + HEADLESSCODE_RSI_OPENROUTER_MAX_INPUT_PRICE_PER_MILLION_USD: "0.1", HEADLESSCODE_RSI_OPENROUTER_MAX_OUTPUT_PRICE_PER_MILLION_USD: "0.2", + HEADLESSCODE_RSI_OPENROUTER_MAX_CALLS: "40", HEADLESSCODE_RSI_OPENROUTER_MAX_INPUT_TOKENS_PER_CALL: "148000", HEADLESSCODE_RSI_OPENROUTER_MAX_OUTPUT_TOKENS_PER_CALL: "7000", HEADLESSCODE_RSI_OPENROUTER_MAX_TOTAL_TOKENS: "8000000", + HEADLESSCODE_RSI_OPENROUTER_MAX_CALL_SPEND_USD: "0.031", HEADLESSCODE_RSI_OPENROUTER_MAX_JOB_SPEND_USD: "0.694", HEADLESSCODE_RSI_OPENROUTER_MAX_CAMPAIGN_SPEND_USD: "19.291206", + HEADLESSCODE_RSI_OPENROUTER_MUTATION_BUDGET_USD: "5.28", HEADLESSCODE_RSI_OPENROUTER_BENCHMARK_BUDGET_USD: "13.44", + HEADLESSCODE_RSI_OPENROUTER_MUTATION_MAX_TOTAL_TOKENS: "3300000", HEADLESSCODE_RSI_OPENROUTER_MUTATION_MAX_JOB_SPEND_USD: "0.66", + HEADLESSCODE_RSI_OPENROUTER_BENCHMARK_MAX_TOTAL_TOKENS: "170000", HEADLESSCODE_RSI_OPENROUTER_BENCHMARK_MAX_JOB_SPEND_USD: "0.034", + } + Object.assign(process.env, values) + const campaign = `rsi-proof-${randomUUID().replaceAll("-", "").slice(0, 20)}` + const runtime = { imageName: "synthetic-test-image", imageDigest: `sha256:${"0".repeat(64)}`, harnessCommit: "1".repeat(40), dependencyLockDigest: "2".repeat(64) } + const sourceCommit = execFileSync("git", ["rev-parse", "HEAD"], { cwd: root, encoding: "utf8" }).trim() + const receipts: RsiPreparationReceipt[] = [] + let inferenceCalls = 0 + const fakeQueue = { + artifactBackend: "s3", + async migrate() {}, async assertResourceClassCoverage() {}, async close() {}, + proofCampaignCoordinator() { return { async snapshotIfExists() { await new Promise((resolve) => setTimeout(resolve, 20)); return undefined } } }, + async enqueue() { inferenceCalls++; throw new Error("inference must not run before acquisition") }, + } + try { + const model = "meta/muse-spark-1.3-contributor" + const candidateConfig: RsiConfig = { + ...trustedConfig, model, roles: { worker: { provider: "openrouter", model, openRouterProfile: "muse-spark-contributor" } }, + population: 1, generations: 2, maxIterations: 1, maxTokens: 7000, maxRuntimeMs: 48 * 60 * 60_000, + workerHarnessPairId: campaign, workerHarnessRunId: `${campaign}-fixed`, workerHarnessMode: "fixed-control", workerHarnessPhase: "development", workerHarnessEngineeringOnly: true, workerHarnessRuntime: runtime, + } + await assert.rejects(() => runRsiImplementation(candidateConfig, { + syntheticRuntimeIdentity: true, log: () => undefined, + async runDevelopmentBenchmark() { return [] }, + async onPreparationReceipt(receipt) { + receipts.push(receipt) + if (receipt.phase === "fixture-validation" && receipt.state === "started") await new Promise((resolve) => setTimeout(resolve, 20)) + }, + }, fakeQueue as unknown as PostgresRsiJobQueue), /paired proof campaigns cannot use an injected benchmark hook/) + assert.equal(inferenceCalls, 0, "no inference enqueue occurs before successful durable acquisition") + assert.deepEqual(receipts.map(({ phase, state }) => `${phase}:${state}`), [ + "fixture-validation:started", "fixture-validation:completed", "proof-protocol:started", "proof-protocol:completed", + "campaign-manifest:started", "campaign-manifest:completed", "campaign-observation:started", "campaign-observation:completed", + ]) + assert.ok(receipts.every((receipt) => receipt.sourceCommit === sourceCommit && receipt.seedSha256.length === 64 && receipt.campaignId === campaign)) + assert.ok(receipts.slice(1).every((receipt) => /^[a-f0-9]{64}$/.test(receipt.manifestSha256 ?? ""))) + assert.ok(receipts.every((receipt) => receipt.inferenceAdmittedByReceipt === false && receipt.readiness === "preparing" && receipt.elapsedSinceStartMs >= 0 && receipt.phaseElapsedMs >= 0)) + assert.ok((receipts[1]?.phaseElapsedMs ?? 0) >= 15, "slow startup work is visible in the completed phase receipt") + assert.ok((receipts.at(-1)?.phaseElapsedMs ?? 0) >= 15, "slow campaign snapshot observation is visible before the rejected test hook reaches admission") + } finally { + for (const [name, value] of Object.entries(saved)) if (value === undefined) delete process.env[name]; else process.env[name] = value + await cleanup() + await fs.rm(root, { recursive: true, force: true }) + } +}) + test("fleet controller queues baseline and all candidate evaluations before waiting", async () => { const { root, config } = await setup() const queue = new FakeFleetQueue(2) diff --git a/src/rsi/__tests__/operations-monitor.test.ts b/src/rsi/__tests__/operations-monitor.test.ts index 8eaf3bf..c980a6f 100644 --- a/src/rsi/__tests__/operations-monitor.test.ts +++ b/src/rsi/__tests__/operations-monitor.test.ts @@ -20,7 +20,7 @@ const campaign = { deadline_expired: false, max_spend_microusd: "1000000", reserved_spend_microusd: "100", used_spend_microusd: "50", } -function fixturePool(options: { malformedWorker?: boolean; throwOnQuery?: boolean; queueRows?: Array>; invalidReceipt?: boolean; finishedPastDeadline?: boolean; idleActiveCampaign?: boolean; armEpoch?: string; armState?: string; armPhase?: string; armIds?: string[]; attemptRows?: Array>; delayMs?: number } = {}) { +function fixturePool(options: { malformedWorker?: boolean; throwOnQuery?: boolean; missingCampaign?: boolean; queueRows?: Array>; invalidReceipt?: boolean; finishedPastDeadline?: boolean; idleActiveCampaign?: boolean; armEpoch?: string; armState?: string; armPhase?: string; armIds?: string[]; attemptRows?: Array>; delayMs?: number } = {}) { const statements: Array<{ sql: string; values?: unknown[] }> = [] let released = 0 const client = { @@ -35,7 +35,7 @@ function fixturePool(options: { malformedWorker?: boolean; throwOnQuery?: boolea : sql.includes("FROM headlesscode_rsi_workers") ? [{ worker_id: "worker-a", active: true, resource_classes: options.malformedWorker ? ["ALIEN"] : ["CPU"], registered_at: new Date(), heartbeat_at: new Date(), heartbeat_age_ms: "2", heartbeat_fresh: true }] : sql.includes("FROM headlesscode_rsi_jobs") && sql.includes("GROUP BY status") ? options.queueRows ?? (options.idleActiveCampaign ? [] : [{ status: "queued", count: "1", expired_leases: "0", pending_teardown: "0", attempts_total: "0" }]) : sql.includes("FROM headlesscode_rsi_jobs") ? [] - : sql.includes("FROM headlesscode_rsi_proof_campaigns") ? [{ ...campaign, state: options.finishedPastDeadline ? "finished" : campaign.state, deadline_expired: options.finishedPastDeadline ?? false, deadline_at: options.finishedPastDeadline ? new Date(Date.now() - 60_000) : campaign.deadline_at, reserved_spend_microusd: options.finishedPastDeadline || options.idleActiveCampaign ? "0" : campaign.reserved_spend_microusd, used_spend_microusd: options.finishedPastDeadline ? "100" : options.idleActiveCampaign ? "50" : campaign.used_spend_microusd }] + : sql.includes("FROM headlesscode_rsi_proof_campaigns") ? options.missingCampaign ? [] : [{ ...campaign, state: options.finishedPastDeadline ? "finished" : campaign.state, deadline_expired: options.finishedPastDeadline ?? false, deadline_at: options.finishedPastDeadline ? new Date(Date.now() - 60_000) : campaign.deadline_at, reserved_spend_microusd: options.finishedPastDeadline || options.idleActiveCampaign ? "0" : campaign.reserved_spend_microusd, used_spend_microusd: options.finishedPastDeadline ? "100" : options.idleActiveCampaign ? "50" : campaign.used_spend_microusd }] : sql.includes("FROM headlesscode_rsi_proof_campaign_arms") ? (options.armIds ?? ["fixed-control", "self-hosted"]).map((arm_id) => ({ arm_id, epoch: options.armEpoch ?? "1", state: options.finishedPastDeadline ? "finished" : options.armState ?? "running", phase: options.armPhase ?? "development", coordinator_id: null, lease_expires_at: new Date(Date.now() + 30_000), lease_fresh: true })) : sql.includes("FROM headlesscode_rsi_proof_campaign_attempts") ? options.finishedPastDeadline || options.idleActiveCampaign ? [] : options.attemptRows ?? [{ status: "dispatched", count: "1" }] : sql.includes("FROM headlesscode_rsi_inference_campaigns AS campaigns") ? [options.invalidReceipt @@ -158,6 +158,9 @@ test("monitor reports zero, sequential one, and both supported arm registrations }) test("database outage and malformed operational rows produce unknown, never healthy", async () => { + const campaignPending = await readRsiOperationsStatus(fixturePool({ missingCampaign: true }).pool, statusOptions) + assert.equal(campaignPending.health, "unknown", "preparation progress cannot substitute for a durable campaign observation") + assert.equal(campaignPending.condition, "campaign-not-found") const outage = fixturePool({ throwOnQuery: true }) assert.equal((await readRsiOperationsStatus(outage.pool, statusOptions)).health, "unknown") const malformed = fixturePool({ malformedWorker: true }) diff --git a/src/rsi/__tests__/proof-fixture-preview.test.ts b/src/rsi/__tests__/proof-fixture-preview.test.ts index 8036ada..8742700 100644 --- a/src/rsi/__tests__/proof-fixture-preview.test.ts +++ b/src/rsi/__tests__/proof-fixture-preview.test.ts @@ -10,6 +10,7 @@ import { runRsi } from "../controller.js" import { benchmarkTreeDigest, materializeBenchmarkSnapshot } from "../benchmark/runner.js" import { buildSnapshotPackage } from "../benchmark/snapshot-package.js" import type { BenchmarkTask } from "../benchmark/types.js" +import type { RsiPreparationReceipt } from "../types.js" const sha256 = (value: Buffer | string) => createHash("sha256").update(value).digest("hex") const git = (cwd: string, args: string[]) => execFileSync("git", args, { cwd, stdio: ["ignore", "pipe", "pipe"] }) @@ -122,19 +123,56 @@ test("proof preview binds an external package root and rejects missing fixtures }] } }), { status: 200 }) }) as typeof fetch - const preview = await buildProofPlanPreview(options, env, fetcher) + const previewReceipts: RsiPreparationReceipt[] = [] + const preview = await buildProofPlanPreview(options, env, fetcher, (receipt) => { previewReceipts.push(receipt) }) assert.equal(metadataRequests, 1, "valid synthetic preview performs only the mocked pricing metadata GET") + assert.deepEqual(previewReceipts.map(({ phase, state }) => `${phase}:${state}`), [ + "proof-plan-preview:started", "fixture-validation:started", "fixture-validation:completed", + "pricing-admission:started", "pricing-admission:completed", "proof-plan-preview:completed", + ]) + assert.ok(previewReceipts.every((receipt) => receipt.sourceCommit === rootCommit && receipt.seedSha256 === sha256("synthetic-eng") && receipt.inferenceAdmittedByReceipt === false && receipt.readiness === "preparing")) + const previewManifestSha256 = sha256(await fs.readFile(manifestPath)) + assert.ok(previewReceipts.slice(2).every((receipt) => receipt.manifestSha256 === previewManifestSha256)) assert.deepEqual(preview.benchmarkFixtures, { root: await fs.realpath(fixtures), manifestSha256: sha256(await fs.readFile(manifestPath)), snapshotPackage: { path: "snapshots", digest: indexDigest } }) assert.deepEqual(preview.execution.configs.map((config) => config.proofBenchmarkFixtures), Array(4).fill(preview.benchmarkFixtures)) assert.equal(preview.benchmarkManifestSha256, preview.benchmarkFixtures.manifestSha256) assert.equal(gitText(repo, ["status", "--porcelain=v1", "--untracked-files=all"]), "", "preview leaves the synthetic source root clean") + const runtimeFailureReceipts: RsiPreparationReceipt[] = [] + const invalidRuntimeEnv = { ...env } + delete invalidRuntimeEnv.HEADLESSCODE_RSI_RUNTIME_IMAGE_DIGEST + await assert.rejects(buildProofPlanPreview(options, invalidRuntimeEnv, fetcher, (receipt) => { runtimeFailureReceipts.push(receipt) }), /HEADLESSCODE_RSI_RUNTIME_IMAGE_DIGEST/) + assert.deepEqual(runtimeFailureReceipts.map(({ phase, state }) => `${phase}:${state}`), ["proof-plan-preview:started", "proof-plan-preview:failed"]) + + const priceCeilingFailureReceipts: RsiPreparationReceipt[] = [] + await assert.rejects(buildProofPlanPreview(options, { + ...env, HEADLESSCODE_RSI_OPENROUTER_MAX_INPUT_PRICE_PER_MILLION_USD: "0.00000001", + }, fetcher, (receipt) => { priceCeilingFailureReceipts.push(receipt) }), /price ceiling is below the live pinned endpoint rate/) + assert.deepEqual(priceCeilingFailureReceipts.map(({ phase, state }) => `${phase}:${state}`), [ + "proof-plan-preview:started", "fixture-validation:started", "fixture-validation:completed", + "pricing-admission:started", "pricing-admission:completed", "proof-plan-preview:failed", + ]) + + const plannerFailureReceipts: RsiPreparationReceipt[] = [] + await assert.rejects(buildProofPlanPreview(options, { + ...env, + HEADLESSCODE_RSI_OPENROUTER_MAX_CALL_SPEND_USD: "0.000001", + HEADLESSCODE_RSI_OPENROUTER_MAX_JOB_SPEND_USD: "0.000002", + HEADLESSCODE_RSI_OPENROUTER_MUTATION_BUDGET_USD: "0.000003", + HEADLESSCODE_RSI_OPENROUTER_BENCHMARK_BUDGET_USD: "0.000003", + }, fetcher, (receipt) => { plannerFailureReceipts.push(receipt) }), /per-call spend cap is below the conservative runtime admission ceiling/) + assert.equal(plannerFailureReceipts.at(-1)?.phase, "proof-plan-preview") + assert.equal(plannerFailureReceipts.at(-1)?.state, "failed", "planner rejection terminates the preview receipt instead of leaving a stale started state") + assert.ok(plannerFailureReceipts.every((receipt) => receipt.inferenceAdmittedByReceipt === false && receipt.readiness === "preparing")) + const indexPath = path.join(packagePath, "index.json") const originalIndex = await fs.readFile(indexPath) await fs.writeFile(indexPath, Buffer.concat([originalIndex, Buffer.from(" ")])) metadataRequests = 0 - await assert.rejects(buildProofPlanPreview(options, env, fetcher), /pinned snapshot package digest mismatch/) + const tamperedReceipts: RsiPreparationReceipt[] = [] + await assert.rejects(buildProofPlanPreview(options, env, fetcher, (receipt) => { tamperedReceipts.push(receipt) }), /pinned snapshot package digest mismatch/) assert.equal(metadataRequests, 0, "changed package bytes fail before metadata pricing fetch") + assert.deepEqual(tamperedReceipts.map(({ phase, state }) => `${phase}:${state}`), ["proof-plan-preview:started", "fixture-validation:started", "fixture-validation:failed", "proof-plan-preview:failed"]) const controllerConfig = { ...proofConfigFromOptions({ repo, seeds: ["synthetic-eng"], mutationTask: "synthetic objective" }, rootCommit, preview.benchmarkFixtures), archiveDir: path.join(root, "controller-archive"), worktreeDir: path.join(root, "controller-worktrees"), dryRun: true, diff --git a/src/rsi/benchmark/__tests__/snapshot-package.test.ts b/src/rsi/benchmark/__tests__/snapshot-package.test.ts index c247844..496a2fc 100644 --- a/src/rsi/benchmark/__tests__/snapshot-package.test.ts +++ b/src/rsi/benchmark/__tests__/snapshot-package.test.ts @@ -104,6 +104,7 @@ try { const loaded = await loadBenchmark(fixtureRoot, emptyRepo) assert.ok(loaded.snapshotPackage, "new runner loads its explicitly pinned local package") assert.equal(loaded.snapshotPackage.digest, descriptor.digest) + assert.ok(loaded.snapshotPackage.blobContents.size > 0, "package loading retains the exact bytes whose hashes and Git object identities were validated") const fixtureBinding = (await loadProofBenchmarkFixtures(fixtureRoot, emptyRepo)).binding assert.deepEqual(fixtureBinding.snapshotPackage, { path: "snapshots", digest: descriptor.digest }) assert.equal((await loadBoundProofBenchmarkFixtures(fixtureBinding, emptyRepo)).snapshotPackage?.digest, descriptor.digest, "external manifest and package index match the frozen supervisor binding") @@ -124,6 +125,15 @@ try { assert.equal(await fs.access(path.join(candidateRoot, "verifiers")).then(() => true).catch(() => false), false, "package extraction contains only the requested source tree") assert.equal(await fs.access(path.join(candidateRoot, "snapshots")).then(() => true).catch(() => false), false, "operator package files are not copied into the candidate") await fs.rm(candidateRoot, { recursive: true, force: true }) + const mutatedBlob = path.join(packagePath, "blobs", firstBlob) + const pinnedBlob = await fs.readFile(mutatedBlob) + await fs.writeFile(mutatedBlob, Buffer.concat([pinnedBlob, Buffer.from("changed after this invocation loaded the package")])) + await assert.rejects(loadBenchmark(fixtureRoot, emptyRepo), /blob digest mismatch/, "a new loading invocation rejects changed package bytes") + const cachedCandidate = path.join(tempRoot, "candidate-pinned-cache") + await materializeBenchmarkSnapshot(tasks[0]!, emptyRepo, cachedCandidate, loaded.snapshotPackage) + assert.equal(await fs.readFile(path.join(cachedCandidate, "value.txt"), "utf8"), "one\n", "materialization uses only the already-verified invocation-local bytes") + await fs.rm(cachedCandidate, { recursive: true, force: true }) + await fs.writeFile(mutatedBlob, pinnedBlob) const blobPath = path.join(packagePath, "blobs", firstBlob) const originalBlob = await fs.readFile(blobPath) diff --git a/src/rsi/benchmark/snapshot-package.ts b/src/rsi/benchmark/snapshot-package.ts index 4881f41..d18e9f1 100644 --- a/src/rsi/benchmark/snapshot-package.ts +++ b/src/rsi/benchmark/snapshot-package.ts @@ -43,6 +43,8 @@ export interface LoadedSnapshotPackage { path: string digest: string snapshots: Map + /** Exact blob bytes verified while loading this package invocation. */ + blobContents: ReadonlyMap } export interface BuiltSnapshotPackage { @@ -194,6 +196,7 @@ export async function loadSnapshotPackage(root: string, descriptor: BenchmarkSna let priorCommit = "" const blobSizes = new Map() const blobGitOids = new Map>() + const blobContents = new Map() for (const snapshot of index.snapshots) { if (!snapshot || !/^[a-f0-9]{40}$/.test(snapshot.commit) || snapshot.commit <= priorCommit || !/^(?:[a-f0-9]{40}|[a-f0-9]{64})$/.test(snapshot.gitTreeOid) || !/^[a-f0-9]{64}$/.test(snapshot.treeDigest) || !Array.isArray(snapshot.entries) || snapshot.entries.length > MAX_TREE_ENTRIES) throw new Error("snapshot package contains malformed or unordered snapshot metadata") if (!hasExactlyKeys(snapshot, ["commit", "gitTreeOid", "treeDigest", "entries"])) throw new Error("snapshot package snapshot has unexpected fields") @@ -235,8 +238,9 @@ export async function loadSnapshotPackage(root: string, descriptor: BenchmarkSna const content = await readRegularFile(fixtureRoot, `${descriptor.path}/blobs/${digest}`, MAX_BLOB_BYTES) if (content.length !== size || sha256(content) !== digest) throw new Error(`snapshot package blob digest mismatch: ${digest}`) if ([...(blobGitOids.get(digest) ?? [])].some((oid) => gitBlobOid(content, oid.length) !== oid)) throw new Error(`snapshot package blob identity mismatch: ${digest}`) + blobContents.set(digest, content) } - return { root: fixtureRoot, path: descriptor.path, digest: descriptor.digest, snapshots } + return { root: fixtureRoot, path: descriptor.path, digest: descriptor.digest, snapshots, blobContents } } export async function materializePackagedSnapshot(snapshotPackage: LoadedSnapshotPackage, commit: string, destination: string): Promise { @@ -247,8 +251,8 @@ export async function materializePackagedSnapshot(snapshotPackage: LoadedSnapsho const digestEntries: Array<{ path: string; mode: 0o644 | 0o755; content: Buffer }> = [] for (const entry of snapshot.entries) { if ((totalBytes += entry.size) > MAX_TREE_BYTES) throw new Error(`snapshot package snapshot exceeds materialization limits: ${commit}`) - const content = await readRegularFile(snapshotPackage.root, `${snapshotPackage.path}/blobs/${entry.blobSha256}`, MAX_BLOB_BYTES) - if (content.length !== entry.size || sha256(content) !== entry.blobSha256 || gitBlobOid(content, entry.gitBlobOid.length) !== entry.gitBlobOid) throw new Error(`snapshot package blob identity mismatch: ${commit}`) + const content = snapshotPackage.blobContents.get(entry.blobSha256) + if (!content || content.length !== entry.size || sha256(content) !== entry.blobSha256 || gitBlobOid(content, entry.gitBlobOid.length) !== entry.gitBlobOid) throw new Error(`snapshot package blob identity mismatch: ${commit}`) const target = path.resolve(destination, entry.path) if (!target.startsWith(destination + path.sep)) throw new Error(`snapshot package path escaped workspace: ${entry.path}`) await fs.mkdir(path.dirname(target), { recursive: true, mode: 0o700 }) diff --git a/src/rsi/controller.ts b/src/rsi/controller.ts index ad2c1dd..8d91aa3 100644 --- a/src/rsi/controller.ts +++ b/src/rsi/controller.ts @@ -22,7 +22,7 @@ import type { EvaluatorRuntimeIdentity } from "./evaluator-runtime.js" import { adversarialEvaluationArtifactIdentity, createAdversarialEvaluationSnapshotBundle, createEvaluationSnapshotBundle, createMutationSnapshotBundle, createMutationSnapshotIdentity, applyFleetMutationPatch } from "./openshell.js" import { ADVERSARIAL_COMMAND, ADVERSARIAL_PROMPT_VERSION, ADVERSARIAL_TESTS_PATH, buildAdversarialPrompt, parseAdversarialReview, requestAdversarialReview, sha256 } from "./adversarial.js" import { fleetJobMatches, fleetPayloadDigest, PostgresRsiJobQueue, fleetQueuePolicy, requiredFleetJobLeaseMs, type FleetJob, type FleetJobIdentity, type FleetJobPayload } from "./postgres-queue.js" -import type { AdversarialReviewRecord, CandidateRecord, EvaluationSummary, ExperimentJob, RsiArchive, RsiConfig, RsiHooks, RsiRunRecord, TrajectoryRecord, TrialResult, WorkerHarnessExecution } from "./types.js" +import type { AdversarialReviewRecord, CandidateRecord, EvaluationSummary, ExperimentJob, RsiArchive, RsiConfig, RsiHooks, RsiPreparationReceipt, RsiRunRecord, TrajectoryRecord, TrialResult, WorkerHarnessExecution } from "./types.js" import { candidateCommits, changedFiles, @@ -1167,17 +1167,49 @@ export async function runRsi(config: RsiConfig, hooks: RsiHooks = {}, queueOverr jobId, }) : undefined const proofFixturesRoot = config.proofBenchmarkFixtures?.root ?? path.join(packageRoot, "fixtures", "rsi-benchmark") - const loadedProofBenchmark = config.proofBenchmarkFixtures - ? await loadBoundProofBenchmarkFixtures(config.proofBenchmarkFixtures, config.repoRoot) - : hooks.runDevelopmentBenchmark - ? await fs.readFile(path.join(proofFixturesRoot, "manifest.json"), "utf8").then((rawManifest) => ({ manifest: JSON.parse(rawManifest) as BenchmarkManifest, manifestDigest: createHash("sha256").update(rawManifest).digest("hex") })) - : await loadBenchmark(proofFixturesRoot, config.repoRoot) + const preparationStartedAt = process.hrtime.bigint() + const preparationPhaseStartedAt = new Map() + const emitPreparationReceipt = async (phase: RsiPreparationReceipt["phase"], state: RsiPreparationReceipt["state"], manifestSha256?: string): Promise => { + if (!config.workerHarnessPairId) return + const timestamp = process.hrtime.bigint() + if (state === "started") preparationPhaseStartedAt.set(phase, timestamp) + const phaseStart = preparationPhaseStartedAt.get(phase) ?? timestamp + const elapsedMs = (value: bigint) => Math.min(Number.MAX_SAFE_INTEGER, Math.max(0, Math.round(Number(value) / 1_000_000))) + const receipt: RsiPreparationReceipt = { + schemaVersion: 1, kind: "rsi-preparation", phase, state, sourceCommit: baseCommit, + seedSha256: createHash("sha256").update(config.seed).digest("hex"), campaignId: config.workerHarnessPairId, + ...((manifestSha256 ?? config.proofBenchmarkFixtures?.manifestSha256) ? { manifestSha256: manifestSha256 ?? config.proofBenchmarkFixtures!.manifestSha256 } : {}), phaseElapsedMs: state === "started" ? 0 : elapsedMs(timestamp - phaseStart), + elapsedSinceStartMs: elapsedMs(timestamp - preparationStartedAt), + readiness: phase === "campaign-acquire" && state === "completed" ? "campaign-acquired" : "preparing", + inferenceAdmittedByReceipt: false, + } + if (hooks.onPreparationReceipt) await hooks.onPreparationReceipt(receipt) + else log(hooks, `RSI_PREPARATION_RECEIPT ${JSON.stringify(receipt)}`) + } + const emitFailureReceipt = async (phase: RsiPreparationReceipt["phase"], manifestSha256?: string): Promise => { + try { await emitPreparationReceipt(phase, "failed", manifestSha256) } catch { /* preserve the original preparation error */ } + } + await emitPreparationReceipt("fixture-validation", "started") + let loadedProofBenchmark: { manifest: BenchmarkManifest; manifestDigest: string } + try { + loadedProofBenchmark = config.proofBenchmarkFixtures + ? await loadBoundProofBenchmarkFixtures(config.proofBenchmarkFixtures, config.repoRoot) + : hooks.runDevelopmentBenchmark + ? await fs.readFile(path.join(proofFixturesRoot, "manifest.json"), "utf8").then((rawManifest) => ({ manifest: JSON.parse(rawManifest) as BenchmarkManifest, manifestDigest: createHash("sha256").update(rawManifest).digest("hex") })) + : await loadBenchmark(proofFixturesRoot, config.repoRoot) + } catch (error) { + await emitFailureReceipt("fixture-validation") + throw error + } + await emitPreparationReceipt("fixture-validation", "completed", loadedProofBenchmark.manifestDigest) + await emitPreparationReceipt("proof-protocol", "started", loadedProofBenchmark.manifestDigest) const proofProtocol = freezeDevelopmentProofProtocol( loadedProofBenchmark.manifest, loadedProofBenchmark.manifestDigest, { provider: workerProvider, id: workerModel, sampling: { temperature: 0, think: false, seed: "provider-default" }, ...(workerOpenRouterProfile && workerOpenRouterProfile.id !== RSI_DEFAULT_OPENROUTER_PROFILE_ID ? { profileId: workerOpenRouterProfile.id } : {}) }, config.developmentProofRule, ) + await emitPreparationReceipt("proof-protocol", "completed", loadedProofBenchmark.manifestDigest) const plannedGenerations = resumed?.workerHarnessPlannedGenerations ?? stageCount if (resumed && (config.workerHarnessMode !== undefined || run.workerHarnessPairId !== undefined) && stageCount !== plannedGenerations) { throw new Error("worker harness planned generation count differs from the frozen paired experiment") @@ -1335,10 +1367,20 @@ export async function runRsi(config: RsiConfig, hooks: RsiHooks = {}, queueOverr if (config.workerHarnessPairId && !config.dryRun) { if (!config.workerHarnessMode || !workerHarnessRuntime || !fleetQueue) throw new Error("restart-safe paired worker-harness campaigns require an explicit arm, frozen runtime identity, and PostgreSQL fleet queue") if (config.workerHarnessEngineeringOnly && (requestedPhase !== "development" || config.resumeRunId && resumed?.proofCampaign?.manifest.phaseMode !== "engineering-only")) throw new Error("engineering-only execution cannot resume or enter a confirmation campaign") + await emitPreparationReceipt("campaign-manifest", "started", loadedProofBenchmark.manifestDigest) const manifest = createProofCampaignManifest({ config: runConfig, campaignId: config.workerHarnessPairId, baseCommit, model: workerModel, provider: workerProvider, ...(workerOpenRouterProfile && workerOpenRouterProfile.id !== RSI_DEFAULT_OPENROUTER_PROFILE_ID ? { openRouterProfileId: workerOpenRouterProfile.id } : {}), runtime: workerHarnessRuntime, developmentManifest: loadedProofBenchmark.manifest, developmentManifestDigest: loadedProofBenchmark.manifestDigest, plannedGenerations, proofProtocol }) - campaignStore = fleetQueue.proofCampaignCoordinator() + await emitPreparationReceipt("campaign-manifest", "completed", loadedProofBenchmark.manifestDigest) + await emitPreparationReceipt("campaign-observation", "started", loadedProofBenchmark.manifestDigest) + let priorSnapshot: Record | undefined + try { + campaignStore = fleetQueue.proofCampaignCoordinator() + priorSnapshot = await campaignStore.snapshotIfExists(config.workerHarnessPairId) + } catch (error) { + await emitFailureReceipt("campaign-observation", loadedProofBenchmark.manifestDigest) + throw error + } + await emitPreparationReceipt("campaign-observation", "completed", loadedProofBenchmark.manifestDigest) if (hooks.runDevelopmentBenchmark) throw new Error("paired proof campaigns cannot use an injected benchmark hook that bypasses durable campaign admission and settlement") - const priorSnapshot = await campaignStore.snapshotIfExists(config.workerHarnessPairId) const priorArm = (priorSnapshot?.arms as Array> | undefined)?.find((entry) => entry.arm_id === workerHarnessMode) if (priorArm?.state === "finished") { return rebuildFinishedProofCampaignArchive({ run, config, snapshot: priorSnapshot!, manifest, armId: workerHarnessMode, now }) @@ -1346,7 +1388,14 @@ export async function runRsi(config: RsiConfig, hooks: RsiHooks = {}, queueOverr if (priorArm?.phase === "confirmation" && requestedPhase !== "confirmation") { return rebuildEvaluatedProofCampaignArchive({ run, config, snapshot: priorSnapshot!, manifest, armId: workerHarnessMode, now }) } - campaignLease = await campaignStore.acquire(manifest, workerHarnessMode, run.runId, `${process.pid}:${randomUUID()}`) + await emitPreparationReceipt("campaign-acquire", "started", loadedProofBenchmark.manifestDigest) + try { + campaignLease = await campaignStore.acquire(manifest, workerHarnessMode, run.runId, `${process.pid}:${randomUUID()}`) + } catch (error) { + await emitPreparationReceipt("campaign-acquire", "failed", loadedProofBenchmark.manifestDigest) + throw error + } + await emitPreparationReceipt("campaign-acquire", "completed", loadedProofBenchmark.manifestDigest) if (campaignLease.phase !== requestedPhase) throw new Error(`proof campaign lease is in ${campaignLease.phase}; requested ${requestedPhase}`) run.proofCampaign = { manifest, manifestSha256: campaignLease.manifestSha256, campaignId: campaignLease.campaignId, armId: campaignLease.armId, epoch: campaignLease.epoch, state: "running" } campaignHeartbeat = setInterval(() => { diff --git a/src/rsi/types.ts b/src/rsi/types.ts index acb78bb..e5b4f8a 100644 --- a/src/rsi/types.ts +++ b/src/rsi/types.ts @@ -536,6 +536,21 @@ export interface MutationRunner { (candidate: CandidateRecord, config: RsiConfig): Promise<{ ok: boolean; result?: TrialResult; error?: string; workerHarnessExecution?: WorkerHarnessExecution }> } +export interface RsiPreparationReceipt { + schemaVersion: 1 + kind: "rsi-preparation" + phase: "proof-plan-preview" | "pricing-admission" | "evaluator-admission" | "fixture-validation" | "proof-protocol" | "campaign-manifest" | "campaign-observation" | "campaign-acquire" + state: "started" | "completed" | "failed" + sourceCommit: string + seedSha256: string + campaignId: string + manifestSha256?: string + phaseElapsedMs: number + elapsedSinceStartMs: number + readiness: "preparing" | "campaign-acquired" + inferenceAdmittedByReceipt: false +} + export interface RsiHooks { runCommand?: CommandRunner runMutation?: MutationRunner @@ -543,6 +558,8 @@ export interface RsiHooks { runDevelopmentBenchmark?: (request: { harnessCommit: string; model: string; outputRoot: string; protocol: DevelopmentProofProtocol }) => Promise now?: () => string log?: (line: string) => void + /** Bounded diagnostics only; the receipt never authorizes evaluator or inference work. */ + onPreparationReceipt?: (receipt: RsiPreparationReceipt) => void | Promise /** Marks test-only injected runtimes so integrity records do not imply a production image identity. */ syntheticRuntimeIdentity?: true }