From df633ca7dc179c8ba6fb1b27cf47e77d8324bf7b Mon Sep 17 00:00:00 2001 From: w4ffl35 <25737761+w4ffl35@users.noreply.github.com> Date: Mon, 12 Oct 2026 10:54:03 -0600 Subject: [PATCH 1/4] Report sequential RSI proof arm registration (#119) --- src/rsi/__tests__/operations-monitor.test.ts | 39 +++++++++++++++++++- src/rsi/operations-monitor.ts | 21 +++++++---- 2 files changed, 51 insertions(+), 9 deletions(-) diff --git a/src/rsi/__tests__/operations-monitor.test.ts b/src/rsi/__tests__/operations-monitor.test.ts index ea0650d..8eaf3bf 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; attemptRows?: Array>; delayMs?: number } = {}) { +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 } = {}) { 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") diff --git a/src/rsi/operations-monitor.ts b/src/rsi/operations-monitor.ts index 687efce..5c49cbe 100644 --- a/src/rsi/operations-monitor.ts +++ b/src/rsi/operations-monitor.ts @@ -13,6 +13,7 @@ const MAX_QUERY_TIMEOUT_MS = 30_000 const MAX_POLL_MS = 60_000 const MAX_MONITOR_RUNTIME_MS = 86_400_000 const MAX_STALL_MS = 604_800_000 +const SUPPORTED_PROOF_ARMS = ["fixed-control", "self-hosted"] as const const REQUIRED_TABLES = [ "headlesscode_rsi_queue_policy", "headlesscode_rsi_workers", @@ -291,16 +292,22 @@ export async function readRsiOperationsStatus(pool: RsiMonitorPool, options: Rsi const rootCeiling = safeInteger(campaign.max_spend_microusd, "root budget") const rootReserved = safeInteger(campaign.reserved_spend_microusd, "root reserved budget") const rootSpent = safeInteger(campaign.used_spend_microusd, "root spent budget") - if (armResult.rows.length !== 2) throw new Error("campaign must have exactly two arms for a trustworthy status") - const arms = armResult.rows.map((row) => { + const registeredArms = new Map>() + for (const row of armResult.rows) { + const armId = safeEnum(row.arm_id, SUPPORTED_PROOF_ARMS, "proof arm") + if (registeredArms.has(armId)) throw new Error("duplicate proof arm registration") if (typeof row.lease_fresh !== "boolean") throw new Error("invalid arm lease state") - return { - armId: identifier(row.arm_id), epoch: safeInteger(row.epoch, "arm epoch"), + registeredArms.set(armId, { + registered: true, armId, epoch: safeInteger(row.epoch, "arm epoch"), state: safeEnum(row.state, ["planned", "running", "evaluated", "selected", "rejected", "confirmed", "finished"] as const, "arm state"), phase: safeEnum(row.phase, ["development", "confirmation"] as const, "arm phase"), coordinatorId: row.coordinator_id === null ? null : identifier(row.coordinator_id), leaseExpiresAt: row.lease_expires_at === null ? null : iso(row.lease_expires_at, "arm lease expiry"), leaseFresh: row.lease_fresh, - } + }) + } + const arms = SUPPORTED_PROOF_ARMS.map((armId) => registeredArms.get(armId) ?? { + registered: false, armId, epoch: null, state: "unregistered", phase: null, + coordinatorId: null, leaseExpiresAt: null, leaseFresh: null, }) const attemptCounts = countRows(attemptResult.rows, "status", "campaign attempt", ["admitted", "dispatched", "completed", "uncertain", "failed", "incomplete", "skipped-after-hard-gate", "not-selected"]) const unsettledAttempts = safeSum(["admitted", "dispatched", "uncertain", "incomplete"].map((key) => attemptCounts[key] ?? 0), "unsettled attempt count") @@ -320,7 +327,7 @@ export async function readRsiOperationsStatus(pool: RsiMonitorPool, options: Rsi return { role, recorded: true, ceiling, spent, reserved, unsettled: reserved, blocked: row.blocked, calls } }) const brokerReserved = safeSum(inferenceBudgetsMicrousd.map((budget) => budget.reserved ?? 0), "inference reserved budget sum") - const armsFinished = arms.length === 2 && arms.every((arm) => arm.state === "finished") + const armsFinished = SUPPORTED_PROOF_ARMS.every((armId) => registeredArms.get(armId)?.state === "finished") const terminal = campaignState === "finished" && armsFinished const armLeaseExpired = arms.some((arm) => arm.state === "running" && arm.leaseFresh === false) const assessment = healthCondition({ @@ -334,7 +341,7 @@ export async function readRsiOperationsStatus(pool: RsiMonitorPool, options: Rsi } const progressFingerprint = createHash("sha256").update(JSON.stringify({ jobCounts, attemptsTotal, campaignState, sequence, attemptCounts, rootCeiling, rootSpent, rootReserved, - arms: arms.map((arm) => ({ armId: arm.armId, epoch: arm.epoch, state: arm.state, phase: arm.phase })), + arms: arms.map((arm) => ({ registered: arm.registered, armId: arm.armId, epoch: arm.epoch, state: arm.state, phase: arm.phase })), inference: inferenceBudgetsMicrousd.map(({ role, recorded, ceiling, spent, reserved, calls }) => ({ role, recorded, ceiling, spent, reserved, calls })), })).digest("hex") await client.query("COMMIT") From 47ba933cd8f19d931795d68285d644ed058ec9aa Mon Sep 17 00:00:00 2001 From: w4ffl35 <25737761+w4ffl35@users.noreply.github.com> Date: Mon, 12 Oct 2026 10:55:47 -0600 Subject: [PATCH 2/4] Cover sequential arm registration in PostgreSQL monitor test --- .../operations-monitor-postgres.test.ts | 30 +++++++++++++++++-- 1 file changed, 27 insertions(+), 3 deletions(-) diff --git a/src/rsi/__tests__/operations-monitor-postgres.test.ts b/src/rsi/__tests__/operations-monitor-postgres.test.ts index 15bbcd5..1b39be8 100644 --- a/src/rsi/__tests__/operations-monitor-postgres.test.ts +++ b/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) From 751128a60873ba0f4b8ee87ce7207823c80c3c5a Mon Sep 17 00:00:00 2001 From: w4ffl35 <25737761+w4ffl35@users.noreply.github.com> Date: Mon, 12 Oct 2026 10:55:37 -0600 Subject: [PATCH 3/4] Check RSI fleet resource coverage before admission Require fresh workers for mutation and CPU evaluation classes before baseline, campaign, or inference admission. Document separate and combined cloud worker configuration and cover stale registrations and missing evaluator capacity. Refs #120 --- docs/recursive-self-improvement.md | 40 ++++++++++++++++++++-- src/rsi/__tests__/controller-fleet.test.ts | 22 ++++++++++++ src/rsi/__tests__/postgres-queue.test.ts | 34 +++++++++++++++++- src/rsi/controller.ts | 3 +- src/rsi/postgres-queue.ts | 36 +++++++++++++++++++ 5 files changed, 130 insertions(+), 5 deletions(-) diff --git a/docs/recursive-self-improvement.md b/docs/recursive-self-improvement.md index b2307fb..6ce9f07 100644 --- a/docs/recursive-self-improvement.md +++ b/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 +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 diff --git a/src/rsi/__tests__/controller-fleet.test.ts b/src/rsi/__tests__/controller-fleet.test.ts index 4fed517..70dd285 100644 --- a/src/rsi/__tests__/controller-fleet.test.ts +++ b/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 {} async activeWorkerCount(): Promise { return 2 } + async assertResourceClassCoverage(required: readonly string[]): Promise { + 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 { 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) diff --git a/src/rsi/__tests__/postgres-queue.test.ts b/src/rsi/__tests__/postgres-queue.test.ts index 0fdb19b..dd4cb46 100644 --- a/src/rsi/__tests__/postgres-queue.test.ts +++ b/src/rsi/__tests__/postgres-queue.test.ts @@ -6,7 +6,7 @@ import test from "node:test" import { S3RsiArtifactStore } from "../artifact-store.js" import { FileRsiArtifactStore } from "../artifact-store.js" import type { OpenRouterBrokerGrant, OpenRouterCallReservation, OpenRouterBrokerLimits, OpenRouterEndpointPriceSnapshot } from "../inference-broker.js" -import { fleetJobMatches, fleetQueuePolicy, PostgresRsiJobQueue, requiredFleetJobLeaseMs, type FleetJob, type FleetJobIdentity, type FleetJobPayload, type FleetQueuePolicy } from "../postgres-queue.js" +import { fleetJobMatches, fleetQueuePolicy, missingFleetResourceClasses, PostgresRsiJobQueue, requiredFleetJobLeaseMs, type FleetJob, type FleetJobIdentity, type FleetJobPayload, type FleetQueuePolicy } from "../postgres-queue.js" test("fleet job replay identity rejects stale snapshots, tasks, base commits, evaluation config, and routing", () => { const expected: FleetJobIdentity = { @@ -62,6 +62,38 @@ test("evaluation lease budget scales with every configured visible command", () assert.ok(threeCommands > oneCommand) }) +test("fleet resource coverage requires every fresh worker class under the persisted capacity policy", () => { + const policy = fleetQueuePolicy("rsi-default", 2, { HEADLESSCODE_RSI_CLASS_CAPACITY: '{"CPU":1,"REMOTE_API":1}' }) + assert.deepEqual(missingFleetResourceClasses(["REMOTE_API", "CPU"], { REMOTE_API: 1, CPU: 1 }, policy), [], "separate or combined live workers cover the two required classes") + assert.deepEqual(missingFleetResourceClasses(["REMOTE_API", "CPU"], { REMOTE_API: 1 }, policy), ["CPU"], "a stale CPU worker is absent from live eligible capacity") + assert.deepEqual(missingFleetResourceClasses(["REMOTE_API", "CPU"], { REMOTE_API: 1, CPU: 1 }, { ...policy, classCapacity: { CPU: 0, REMOTE_API: 1 } }), ["CPU"], "a persisted class policy with no available capacity fails closed") +}) + +test("PostgreSQL worker coverage accepts separate or combined workers and ignores stale registrations", { skip: !databaseUrl }, async () => { + const schema = `rsi_coverage_${randomUUID().replaceAll("-", "")}` + const root = await import("node:fs/promises").then(({ mkdtemp }) => mkdtemp(`${process.env.TMPDIR ?? "/tmp"}/rsi-coverage-test-`)) + const admin = new Pool({ connectionString: databaseUrl }) + await admin.query(`CREATE SCHEMA ${schema}`) + const pool = new Pool({ connectionString: databaseUrl, options: `-c search_path=${schema}` }) + const policy = testPolicy(`coverage-${randomUUID().slice(0, 8)}`, { maxInFlight: 2, classCapacity: { REMOTE_API: 1, CPU: 1 } }) + const queue = new PostgresRsiJobQueue(pool, policy, new FileRsiArtifactStore(root)) + try { + await queue.migrate() + await queue.registerWorker({ workerId: "remote", gatewayId: "remote-gateway", hostname: "remote-host", maxActive: 1, resourceClasses: ["REMOTE_API"] }) + await queue.registerWorker({ workerId: "eval", gatewayId: "eval-gateway", hostname: "eval-host", maxActive: 1, resourceClasses: ["CPU"] }) + await queue.assertResourceClassCoverage(["REMOTE_API", "CPU"]) + await pool.query("UPDATE headlesscode_rsi_workers SET heartbeat_at=clock_timestamp()-interval '5 minutes' WHERE worker_id='eval'") + await assert.rejects(queue.assertResourceClassCoverage(["REMOTE_API", "CPU"]), /CPU/) + await queue.registerWorker({ workerId: "combined", gatewayId: "combined-gateway", hostname: "combined-host", maxActive: 1, resourceClasses: ["REMOTE_API", "CPU"] }) + await queue.assertResourceClassCoverage(["REMOTE_API", "CPU"]) + } finally { + await queue.close().catch(() => undefined) + await admin.query(`DROP SCHEMA ${schema} CASCADE`) + await admin.end() + await import("node:fs/promises").then(({ rm }) => rm(root, { recursive: true, force: true })) + } +}) + test("PostgreSQL migrations upgrade an earlier artifact and lease schema", { skip: !databaseUrl }, async () => { const schema = `rsi_migrate_${randomUUID().replaceAll("-", "")}` const root = await import("node:fs/promises").then(({ mkdtemp }) => mkdtemp(`${process.env.TMPDIR ?? "/tmp"}/rsi-migration-test-`)) diff --git a/src/rsi/controller.ts b/src/rsi/controller.ts index 08da7f1..ad2c1dd 100644 --- a/src/rsi/controller.ts +++ b/src/rsi/controller.ts @@ -1329,7 +1329,8 @@ export async function runRsi(config: RsiConfig, hooks: RsiHooks = {}, queueOverr if (fleetQueue) { if (fleetQueue.artifactBackend !== "s3") throw new Error("RSI production runs require a shared S3-compatible artifact store") await fleetQueue.migrate() - if (await fleetQueue.activeWorkerCount() < 1) throw new Error("no live OpenShell RSI worker is registered for the configured queue") + const requiredResourceClasses = ["CPU", workerProvider === "openshell-openrouter" ? "REMOTE_API" : "LOCAL_GPU"] as const + await fleetQueue.assertResourceClassCoverage(requiredResourceClasses) } 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") diff --git a/src/rsi/postgres-queue.ts b/src/rsi/postgres-queue.ts index 77dbeee..5a2453e 100644 --- a/src/rsi/postgres-queue.ts +++ b/src/rsi/postgres-queue.ts @@ -85,6 +85,19 @@ export function fleetQueuePolicy(queueName: string, maxInFlight: number, env: No }) } +/** Return required classes that have no live registered worker capacity under this queue policy. */ +export function missingFleetResourceClasses( + required: readonly ResourceClass[], + eligibleWorkerSlots: Partial>, + policy: FleetQueuePolicy, +): ResourceClass[] { + return [...new Set(required)].filter((resourceClass) => { + const slots = eligibleWorkerSlots[resourceClass] ?? 0 + const classCapacity = policy.classCapacity?.[resourceClass] ?? policy.maxInFlight + return !Number.isSafeInteger(slots) || slots < 1 || classCapacity < 1 + }) +} + export interface FleetWorkerRegistration { workerId: string gatewayId: string @@ -355,6 +368,29 @@ export class PostgresRsiJobQueue { return Number(result.rows[0]?.count ?? 0) } + async assertResourceClassCoverage(required: readonly ResourceClass[]): Promise { + const classes = [...new Set(required)] + if (classes.length === 0) return + const result = await this.pool.query( + `SELECT required.resource_class, + COALESCE(SUM(worker.max_active), 0)::int AS worker_slots + FROM unnest($2::text[]) AS required(resource_class) + LEFT JOIN headlesscode_rsi_workers AS worker + ON worker.queue_name=$1 AND worker.active=true + AND worker.heartbeat_at > clock_timestamp()-($3::text || ' milliseconds')::interval + AND worker.resource_classes @> jsonb_build_array(required.resource_class) + GROUP BY required.resource_class`, + [this.policy.queueName, classes, this.policy.leaseMs * 2], + ) + const eligibleWorkerSlots: Partial> = {} + for (const row of result.rows) eligibleWorkerSlots[row.resource_class as ResourceClass] = Number(row.worker_slots) + const missing = missingFleetResourceClasses(classes, eligibleWorkerSlots, this.policy) + if (missing.length > 0) { + const details = missing.map((resourceClass) => `${resourceClass} (no fresh worker slot or queue capacity)`).join(", ") + throw new Error(`RSI fleet admission requires live worker capacity for ${details}; start a worker registered with HEADLESSCODE_RSI_WORKER_RESOURCE_CLASSES including each missing class and verify its shared queue policy`) + } + } + async stopWorker(workerId: string): Promise { await this.pool.query("UPDATE headlesscode_rsi_workers SET active=false, heartbeat_at=clock_timestamp() WHERE queue_name=$1 AND worker_id=$2", [this.policy.queueName, workerId]) } From 4dae7286166734b496c755e2f358019adcff0859 Mon Sep 17 00:00:00 2001 From: w4ffl35 <25737761+w4ffl35@users.noreply.github.com> Date: Mon, 12 Oct 2026 11:00:50 -0600 Subject: [PATCH 4/4] Prepare v1.9.9 sequential monitoring and cloud admission --- docs/recursive-self-improvement.md | 4 ++-- package-lock.json | 4 ++-- package.json | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/recursive-self-improvement.md b/docs/recursive-self-improvement.md index 6ce9f07..c1189a3 100644 --- a/docs/recursive-self-improvement.md +++ b/docs/recursive-self-improvement.md @@ -3,8 +3,8 @@ `headlesscode improve --dry-run` plans candidate worktrees. A real run requires 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 +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 diff --git a/package-lock.json b/package-lock.json index ad4fc0a..e49c258 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "headlesscode", - "version": "1.9.8", + "version": "1.9.9", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "headlesscode", - "version": "1.9.8", + "version": "1.9.9", "license": "Apache-2.0", "dependencies": { "@aws-sdk/client-s3": "^3.1142.0", diff --git a/package.json b/package.json index 140575d..2e6f44d 100644 --- a/package.json +++ b/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": {