Fix sequential RSI monitoring and require cloud evaluation capacity

docs/recursive-self-improvement.md

@@ -1,15 +1,20 @@
 # Recursive Self-Improvement
 
 `headlesscode improve --dry-run` plans candidate worktrees. A real run requires
-a PostgreSQL queue, a shared S3-compatible artifact store, and at least one
-registered `headlesscode rsi-worker` connected to an OpenShell gateway. The
-coordinator queues a sanitized baseline visible evaluation, candidate mutation
+a PostgreSQL queue, a shared S3-compatible artifact store, and fresh worker
+registrations connected to OpenShell gateways that cover the run's mutation and
+evaluation resource classes. The coordinator queues a sanitized baseline visible
+evaluation, candidate mutation
 snapshots, and candidate visible evaluations; workers run each job in a fresh
 OpenShell guest. The coordinator applies returned patches, calculates fitness,
 compares results with the baseline, and owns the archive and candidate
 selection. Full visible-test output is content-addressed in the object store;
 PostgreSQL keeps bounded previews and artifact references. Hidden evaluation
-remains disabled.
+remains disabled. Before baseline work or campaign admission, the coordinator
+requires fresh worker registrations for every resource class used by the run.
+OpenRouter mutation and proof calls require `REMOTE_API`; baseline and candidate
+evaluation require `CPU`. A single worker may register both classes, or separate
+workers may provide them. A cloud mutation worker alone is insufficient.
 
 ## Candidate guest isolation check
 
@@ -330,6 +335,35 @@ queue must use the same queue policy and artifact-store identity.
 | `HEADLESSCODE_OPENSHELL_IMAGE` | `headlesscode-openshell-rsi:local` | RSI guest image; custom images are rejected. |
 | `HEADLESSCODE_RSI_OLLAMA_HOST` | `host.openshell.internal` | Local Ollama host permitted by the RSI mutation policy. |
 
+Use the same queue name, maximum in-flight value, lease, retry, class-capacity,
+and key-capacity settings on the coordinator and every worker. `fleetQueuePolicy`
+builds this shared policy for both sides, and startup rejects a persisted policy
+that differs from the requested settings. For example, this explicit two-slot
+OpenRouter setup reserves one admission slot for each required class:
+
+```sh
+export HEADLESSCODE_RSI_QUEUE_NAME=rsi-cloud
+export HEADLESSCODE_RSI_MAX_IN_FLIGHT=2
+export HEADLESSCODE_RSI_CLASS_CAPACITY='{"REMOTE_API":1,"CPU":1}'
+export HEADLESSCODE_RSI_ROLE_WORKER_PROVIDER=openrouter
+export HEADLESSCODE_RSI_ROLE_WORKER_MODEL=meta/muse-spark-1.3-contributor
+export HEADLESSCODE_RSI_ROLE_WORKER_OPENROUTER_PROFILE=muse-spark-contributor
+
+# Start each in its own process with the remaining shared database, artifact,
+# gateway, and credential settings configured for that host.
+HEADLESSCODE_RSI_WORKER_ID=mutation-1 HEADLESSCODE_RSI_WORKER_RESOURCE_CLASSES=REMOTE_API headlesscode rsi-worker
+HEADLESSCODE_RSI_WORKER_ID=evaluation-1 HEADLESSCODE_RSI_WORKER_RESOURCE_CLASSES=CPU headlesscode rsi-worker
+
+# Run the coordinator with a matching in-flight cap.
+headlesscode improve --repo . --max-concurrent 2
+```
+
+The `CPU` class here runs regression, typecheck, and visible evaluation
+commands; it does not select a local language model. A worker with a stale
+heartbeat is not eligible. Missing class coverage produces an admission error
+that identifies the class and worker setting to configure, before baseline jobs,
+campaign admission, or inference reservations are created.
+
 ### OpenRouter mutation and development-proof jobs
 
 OpenRouter is available for RSI fleet mutations and the development benchmark

package.json

@@ -1,6 +1,6 @@
 {
 	"name": "headlesscode",
-	"version": "1.9.8",
+	"version": "1.9.9",
 	"description": "Coding-agent harness with CLI and HTTP interfaces, parallel Git worktree orchestration, and bounded recursive self-improvement (RSI).",
 	"license": "Apache-2.0",
 	"repository": {

src/rsi/__tests__/controller-fleet.test.ts

@@ -54,6 +54,7 @@ class FakeFleetQueue {
 	readonly failRegressionForCandidates: boolean
 	readonly staleCurriculumReplayKey: boolean
 	readonly failAdversarial: boolean
+	workerResourceClasses = new Set(["CPU", "LOCAL_GPU", "REMOTE_API"])
 	staleFastPath?: "baseline" | "mutation" | "evaluation"
 	adversarialEvaluationJobCount = 0
 	proofCampaign?: PostgresProofCampaign
@@ -74,6 +75,10 @@ class FakeFleetQueue {
 
 	async migrate(): Promise<void> {}
 	async activeWorkerCount(): Promise<number> { return 2 }
+	async assertResourceClassCoverage(required: readonly string[]): Promise<void> {
+		const missing = required.filter((resourceClass) => !this.workerResourceClasses.has(resourceClass))
+		if (missing.length) throw new Error(`RSI fleet admission requires live worker capacity for ${missing.join(", ")}`)
+	}
 	async close(): Promise<void> { this.closeCount += 1 }
 	setProofCampaign(campaign: PostgresProofCampaign): void {
 		this.proofCampaign = campaign
@@ -323,6 +328,23 @@ test("fleet controller queues baseline and all candidate evaluations before wait
 	}
 })
 
+test("fleet admission rejects a cloud-only worker before baseline work or campaign admission", async () => {
+	const { root, config } = await setup()
+	const queue = new FakeFleetQueue(2)
+	queue.workerResourceClasses = new Set(["REMOTE_API"])
+	try {
+		await assert.rejects(
+			runRsi({ ...config, roles: { worker: { provider: "openrouter", model: "meta/muse-spark-1.3-contributor", openRouterProfile: "muse-spark-contributor" } } }, { log: () => undefined }, queue as unknown as PostgresRsiJobQueue),
+			/live worker capacity for CPU/,
+		)
+		assert.equal(queue.artifacts.size, 0, "missing evaluator capacity is detected before artifacts or baseline jobs are created")
+		assert.equal(queue.evaluationJobCount, 0)
+		assert.equal(queue.mutationJobCount, 0, "no inference mutation is admitted without all required worker classes")
+	} finally {
+		await fs.rm(root, { recursive: true, force: true })
+	}
+})
+
 test("improve reports an incomplete campaign as failure despite a historical selected candidate", () => {
 	const outcome = improveRunOutcome({ runId: "run-incomplete", selected: "candidate-from-earlier-generation", proofCampaign: { state: "incomplete" } }, false)
 	assert.equal(outcome.exitCode, 1)

src/rsi/__tests__/operations-monitor-postgres.test.ts

@@ -33,11 +33,35 @@ test("isolated PostgreSQL monitor reports cancellation teardown and budget state
 			VALUES ($1,'monitor-test','synthetic-job','synthetic-run','synthetic-candidate','evaluation','CPU',$2,'{}'::jsonb,'leased',1,3,'synthetic-worker',$3,clock_timestamp()+interval '30 seconds',clock_timestamp(),'synthetic teardown probe')`, [jobId, digest, randomUUID()])
 		await pool.query(`INSERT INTO headlesscode_rsi_proof_campaigns(campaign_id,schema_version,manifest,manifest_sha256,expected_arms,state,deadline_at,max_calls,max_tokens,max_spend_microusd)
 			VALUES ('monitor-campaign',1,'{}'::jsonb,$1,'[]'::jsonb,'running',clock_timestamp()+interval '1 hour',10,1000,1000000)`, [digest])
-		for (const armId of ["arm-a", "arm-b"]) await pool.query(`INSERT INTO headlesscode_rsi_proof_campaign_arms(campaign_id,arm_id,run_id,state)
-			VALUES ('monitor-campaign',$1,$2,'planned')`, [armId, `synthetic-${armId}`])
 		for (const role of ["mutation", "benchmark"]) await pool.query(`INSERT INTO headlesscode_rsi_inference_campaigns(campaign_id,provider,model,max_spend_microusd)
 			VALUES ($1,'openrouter','synthetic-test-model',500000)`, [`monitor-campaign:${role}`])
-		const status = await readRsiOperationsStatus(pool, { queueName: "monitor-test", campaignId: "monitor-campaign", queryTimeoutMs: 3_000 })
+		const statusOptions = { queueName: "monitor-test", campaignId: "monitor-campaign", queryTimeoutMs: 3_000 }
+		const noArmsStatus = await readRsiOperationsStatus(pool, statusOptions)
+		assert.notEqual(noArmsStatus.health, "unknown", "campaign with no sequential arms registered still has a trustworthy operational status")
+		assert.deepEqual(noArmsStatus.campaign.arms.map((arm) => [arm.armId, arm.registered]), [["fixed-control", false], ["self-hosted", false]])
+		assert.equal(noArmsStatus.campaign.accountingComplete, true)
+		await pool.query(`INSERT INTO headlesscode_rsi_proof_campaign_arms(campaign_id,arm_id,run_id,state)
+			VALUES ('monitor-campaign','fixed-control','synthetic-fixed-control','planned')`)
+		const firstArmStatus = await readRsiOperationsStatus(pool, statusOptions)
+		assert.notEqual(firstArmStatus.health, "unknown", "fixed-control can register before self-hosted without making monitor status unknown")
+		assert.deepEqual(firstArmStatus.campaign.arms.map((arm) => [arm.armId, arm.registered]), [["fixed-control", true], ["self-hosted", false]])
+		assert.deepEqual(firstArmStatus.campaign.arms[1], {
+			registered: false, armId: "self-hosted", epoch: null, state: "unregistered", phase: null,
+			coordinatorId: null, leaseExpiresAt: null, leaseFresh: null,
+		})
+		assert.equal(firstArmStatus.campaign.accountingComplete, true)
+		await pool.query(`INSERT INTO headlesscode_rsi_proof_campaign_arms(campaign_id,arm_id,run_id,state)
+			VALUES ('monitor-campaign','self-hosted','synthetic-self-hosted','planned')`)
+		const pairedArmsStatus = await readRsiOperationsStatus(pool, statusOptions)
+		assert.notEqual(pairedArmsStatus.health, "unknown", "both supported arms are reported after the second registration")
+		assert.deepEqual(pairedArmsStatus.campaign.arms.map((arm) => [arm.armId, arm.registered]), [["fixed-control", true], ["self-hosted", true]])
+		assert.equal(pairedArmsStatus.campaign.accountingComplete, true)
+		await pool.query(`INSERT INTO headlesscode_rsi_proof_campaign_arms(campaign_id,arm_id,run_id,state)
+			VALUES ('monitor-campaign','unsupported-arm','synthetic-unsupported','planned')`)
+		const malformedArmStatus = await readRsiOperationsStatus(pool, statusOptions)
+		assert.equal(malformedArmStatus.health, "unknown", "unsupported persisted arm IDs fail closed")
+		await pool.query("DELETE FROM headlesscode_rsi_proof_campaign_arms WHERE campaign_id='monitor-campaign' AND arm_id='unsupported-arm'")
+		const status = await readRsiOperationsStatus(pool, statusOptions)
 		assert.equal(status.health, "unhealthy")
 		assert.equal(status.condition, "cancellation-teardown-pending")
 		assert.equal(status.queue.pendingTeardown, 1)

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; attemptRows?: Array<Record<string, unknown>>; delayMs?: number } = {}) {
+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 } = {}) {
 	const statements: Array<{ sql: string; values?: unknown[] }> = []
 	let released = 0
 	const client = {
@@ -36,7 +36,7 @@ function fixturePool(options: { malformedWorker?: boolean; throwOnQuery?: boolea
 				: 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_campaign_arms") ? ["arm-a", "arm-b"].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_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
 					? { campaign_id: "campaign-test:mutation", state: "complete", count: "1", reserved_count: "0", unmeasured_count: "0", invalid_receipts: "1", reserved_microusd: "100" }
@@ -68,6 +68,8 @@ test("status reads bounded operational metadata in one read-only snapshot withou
 	assert.equal(fixture.released, 1)
 	assert.ok(fixture.statements.every(({ sql }) => !/\b(payload|manifest|prompt|capability_sha256|artifact_refs|proof)\b/i.test(sql)), "monitor SQL must not read payload or proof contents")
 	assert.ok(fixture.statements.some(({ sql, values }) => sql.includes("heartbeat_at > clock_timestamp()") && values?.[1] === 10_000), "freshness uses twice the persisted lease interval")
+	assert.deepEqual(fixture.statements.find(({ sql }) => sql.includes("FROM headlesscode_rsi_proof_campaign_arms"))?.values, ["campaign-test"])
+	assert.deepEqual(result.campaign.arms.map((arm) => [arm.armId, arm.registered]), [["fixed-control", true], ["self-hosted", true]])
 	const baseFingerprint = await readRsiOperationsStatus(fixturePool({ idleActiveCampaign: true }).pool, statusOptions)
 	const changedEpoch = await readRsiOperationsStatus(fixturePool({ idleActiveCampaign: true, armEpoch: "2" }).pool, statusOptions)
 	const changedPhase = await readRsiOperationsStatus(fixturePool({ idleActiveCampaign: true, armPhase: "confirmation" }).pool, statusOptions)
@@ -122,6 +124,39 @@ test("monitor stalls active work between admissions but leaves settled terminal
 	assert.equal(uncertainCampaign.unsettledAttempts, 1, "stall reporting preserves uncertain campaign attempts")
 })
 
+test("monitor reports zero, sequential one, and both supported arm registrations explicitly", async () => {
+	const stages: Array<{ label: string; armIds: string[]; registered: boolean[] }> = [
+		{ label: "before first arm registration", armIds: [], registered: [false, false] },
+		{ label: "fixed-control registered first", armIds: ["fixed-control"], registered: [true, false] },
+		{ label: "self-hosted registered second", armIds: ["fixed-control", "self-hosted"], registered: [true, true] },
+	]
+	for (const stage of stages) {
+		const status = await readRsiOperationsStatus(fixturePool({ armIds: stage.armIds }).pool, statusOptions)
+		assert.notEqual(status.health, "unknown", stage.label)
+		assert.deepEqual(status.campaign.arms.map((arm) => [arm.armId, arm.registered]), [
+			["fixed-control", stage.registered[0]], ["self-hosted", stage.registered[1]],
+		], stage.label)
+		assert.equal(status.campaign.terminal, false, "nonterminal registration status does not imply completion or acceptance")
+		assert.equal(status.queue.jobCounts.queued, 1, "queue status remains available while arms register sequentially")
+		assert.equal(status.campaign.rootBudgetMicrousd.reserved, 100, "budget accounting remains available while arms register sequentially")
+		if (stage.label === "fixed-control registered first") assert.deepEqual(status.campaign.arms[1], {
+			registered: false, armId: "self-hosted", epoch: null, state: "unregistered", phase: null,
+			coordinatorId: null, leaseExpiresAt: null, leaseFresh: null,
+		}, "the absent paired arm is explicit rather than silently omitted")
+	}
+	const unknownArm = await readRsiOperationsStatus(fixturePool({ armIds: ["fixed-control", "unexpected"] }).pool, statusOptions)
+	assert.equal(unknownArm.health, "unknown", "an unsupported arm ID is not treated as an absent supported arm")
+	const duplicateArm = await readRsiOperationsStatus(fixturePool({ armIds: ["fixed-control", "fixed-control"] }).pool, statusOptions)
+	assert.equal(duplicateArm.health, "unknown", "duplicate arm registrations fail closed")
+	assert.equal((await readRsiOperationsStatus(fixturePool({ armState: "mystery" }).pool, statusOptions)).health, "unknown", "unknown registered arm states fail closed")
+	assert.equal((await readRsiOperationsStatus(fixturePool({ armPhase: "unknown" }).pool, statusOptions)).health, "unknown", "unknown registered arm phases fail closed")
+	assert.equal((await readRsiOperationsStatus(fixturePool({ armEpoch: "9007199254740992" }).pool, statusOptions)).health, "unknown", "unsafe arm epochs fail closed")
+	const terminalMissingArm = await readRsiOperationsStatus(fixturePool({ finishedPastDeadline: true, armIds: ["fixed-control"] }).pool, statusOptions)
+	assert.equal(terminalMissingArm.health, "unhealthy", "a terminal campaign missing a paired arm cannot be healthy")
+	assert.equal(terminalMissingArm.condition, "terminal-campaign-arm-state-mismatch")
+	assert.equal(terminalMissingArm.campaign.terminal, false)
+})
+
 test("database outage and malformed operational rows produce unknown, never healthy", async () => {
 	const outage = fixturePool({ throwOnQuery: true })
 	assert.equal((await readRsiOperationsStatus(outage.pool, statusOptions)).health, "unknown")