Improve bounded RSI startup observability

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)

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<Record<string, unknown>>; invalidReceipt?: boolean; finishedPastDeadline?: boolean; idleActiveCampaign?: boolean; armEpoch?: string; armState?: string; armPhase?: string; armIds?: string[]; attemptRows?: Array<Record<string, unknown>>; delayMs?: number } = {}) {
+function fixturePool(options: { malformedWorker?: boolean; throwOnQuery?: boolean; missingCampaign?: boolean; queueRows?: Array<Record<string, unknown>>; invalidReceipt?: boolean; finishedPastDeadline?: boolean; idleActiveCampaign?: boolean; armEpoch?: string; armState?: string; armPhase?: string; armIds?: string[]; attemptRows?: Array<Record<string, unknown>>; 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 })

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,

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)