diff --git a/docs/recursive-self-improvement.md b/docs/recursive-self-improvement.md index b2307fb..c1189a3 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 -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 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": { 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__/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) 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/__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/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") 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]) }