From 11256a5a15ce5d572b7e8b1b1754b759e329c251 Mon Sep 17 00:00:00 2001 From: likun Date: Mon, 14 Sep 2026 16:44:22 +0800 Subject: [PATCH] feat(eval): add durable fleet scheduling with deterministic VM simulation Refs #5284. Implement task-group claims, bounded retries, reconciliation, artifact persistence, and mock worker fault replay. Generated-by: Codex --- packages/eval/FLEET.md | 79 ++++ packages/eval/README.md | 5 + packages/eval/package.json | 1 + packages/eval/src/__tests__/fleet.test.ts | 544 ++++++++++++++++++++++ packages/eval/src/fleet-simulation.ts | 284 +++++++++++ packages/eval/src/fleet-store.ts | 273 +++++++++++ packages/eval/src/fleet-worker.ts | 135 ++++++ packages/eval/src/fleet.ts | 459 ++++++++++++++++++ packages/eval/src/index.ts | 3 + 9 files changed, 1783 insertions(+) create mode 100644 packages/eval/FLEET.md create mode 100644 packages/eval/src/__tests__/fleet.test.ts create mode 100644 packages/eval/src/fleet-simulation.ts create mode 100644 packages/eval/src/fleet-store.ts create mode 100644 packages/eval/src/fleet-worker.ts create mode 100644 packages/eval/src/fleet.ts diff --git a/packages/eval/FLEET.md b/packages/eval/FLEET.md new file mode 100644 index 0000000000..11eed1fdc7 --- /dev/null +++ b/packages/eval/FLEET.md @@ -0,0 +1,79 @@ + + +# Fleet scheduling (mock workers) + +This implements the coordinator/worker contract proposed in [RFC #5284](https://github.com/apache/maka/issues/5284), with remote VM execution and transport mocked. The coordinator state machine, durable local storage, artifact checks, and worker outbox are exercised directly by a deterministic simulation. No cloud VM, Kubernetes, Docker, provider credentials, or live inference is needed to run it. + +`maka eval run` continues to use the existing single-host runner. There is no network daemon or fleet CLI deployment command yet. The library exports `FleetCoordinator`, `FileFleetPersistence`, `FleetWorker`, their protocols, and the pure `transitionFleet` function for the next transport integration. + +## Run and replay + +With workspace dependencies built: + +```sh +npm --workspace @maka/eval run simulate:fleet -- 42 +node --test packages/eval/dist/__tests__/fleet.test.js +``` + +The simulation outputs its seed, virtual elapsed ticks, selected results, and event trace as JSON. Running the same seed produces the same trace and state, including attempt identities. Tests replay seed 42 twice and exercise seeds 0–15, checking invariants after every tick. To replay another seed, replace `42` with a uint32 value. CLI output can be redirected to retain a full run trace. + +Three mock VMs dynamically process 40 task groups (120 cells). Faults include request loss, lost replies after a successful commit, a worker network partition, a worker process crash, coordinator disconnection/restart, delayed cell completions, and missing usage on otherwise valid outcomes. After a bounded fault window, connectivity returns and pending work must converge. Separate tests inject storage failure before and after commit, check filesystem recovery and artifact integrity, and exercise admission failures, retry exhaustion, and partial group recovery. This is a deterministic application simulation, not a simulation of a kernel, Docker, or cloud infrastructure. + +## Ownership and capacity + +- `ExperimentSpec.execution.maxConcurrentTaskGroups` is the global active-assignment cap. Each worker declares fixed group slots. A group remains assigned until all cell reports are committed and the worker finishes the group, or ownership expires. +- A group is `task × repetition`, and its pending cells run on one worker. A replacement assignment contains only cells without a selected result and with remaining attempts. A selected zero or subject failure is preserved. +- A worker ID identifies one process incarnation. A VM reboot must use a new ID; continuing to use an ID after losing its in-memory execution/outbox state is unsupported. +- `FleetPolicy.environmentId` names the required, reviewed code/benchmark/toolchain/image manifest. The current mock worker advertises that identity. A real adapter must verify it, not merely echo it. CPU/memory reservations cover a whole group's execution **and verification**, multiplied by worker slots. Deriving requirements and verifying real VM capabilities belong to the future adapter. +- Group retry waits for `retryBackoffMs`. `maxAttemptsPerCell` counts dispatches, including a dispatch lost before execution can be confirmed. Exhaustion remains incomplete evaluation, never a fabricated score. + +## Recovery and evidence + +Workers register, heartbeat with owned assignment IDs, and claim available work. Heartbeat replies also recover assignments whose claim acknowledgement was lost. Cell outputs stay in the worker outbox until artifact upload and result commit are acknowledged. Duplicate submission of the same report is idempotent; conflicting content is rejected. Outboxes currently survive network disconnection only; VM/process death loses uncommitted local state and invokes bounded cell retry. + +Call `pause` when the coordinator's deployment detects its own disconnection or suspends scheduling. It freezes dispatch and lease expiry. Workers continue already assigned work and buffer results. Call `recover` before resuming; opening a coordinator performs this automatically. Recovery marks workers as needing reconciliation and grants outstanding assignments a recovery grace interval. It does not immediately expire every worker after a long local outage. Workers that fail to reconcile eventually expire. The core cannot infer a local network outage from missing worker heartbeats; detecting local connectivity belongs to the transport/deployment layer. + +Each assignment has a new generation and each dispatched cell has a unique attempt ID. Expired assignments cannot supply selected results. Their late reports remain available as evidence and contribute to observed costs. Already selected results are immutable. Assignment expiry permits duplicate physical execution during partitions; it does not kill a remote process or guarantee exactly-once inference. Capacity limits apply to authoritative assignments, not unreachable zombie processes. + +`FleetReport` separates execution, verification, usage, and cleanup evidence. Valid verification following completed/failed subject execution and confirmed cleanup is selectable, even if metering is incomplete. Raw reports are retained; the result summary projects the settled outcome independently of a coarse `indeterminate` status. Missing usage alone does not rerun a cell. Unknown execution or cleanup, or invalid/missing verification, requires a cell retry. Provider request recovery remains the execution adapter/Runtime Host's responsibility. + +A deterministic `environmentFailure` pauses dispatch for this run until an operator issues `repair` with that reason. Already running groups are allowed to settle. This conservative run-wide block avoids repeating the same defect across workers; task-specific routing is not implemented. There is no verification-only recovery, environment snapshot restoration, or within-group straggler optimization. + +## Local persistence + +Use a separate fleet directory; its state format does not replace or import the existing local `FileAttemptStore` format: + +```ts +const persistence = await FileFleetPersistence.open('/absolute/path/to/fleet-run'); +const coordinator = await FleetCoordinator.open({ + persistence, spec, policy, now: Date.now, +}); +// Inject a FleetTransport and FleetGroupExecution into each FleetWorker. +// Poll workers periodically and drive coordinator tick while online. +// Commands are serialized, and replies follow durable state commits. +await coordinator.close(); +await persistence.close(); +``` + +The spec and policy are frozen by run identity: reopening with different values fails. State uses an atomic, checksummed snapshot with fsync before acknowledgement; the parent directory is synced on POSIX. Artifacts use content-addressed files and are checked for length and SHA-256 before accepting their references. Only artifacts listed in `FleetReport.artifacts` have this durability contract; legacy metadata in `EvalResult.artifacts` is not a substitute for uploading bytes. These are evaluation artifacts, not a restorable VM snapshot. + +The directory has one lifetime writer lock. Close the coordinator before its persistence. After a process crash, confirm that the old owner is dead before manually removing a stale `.writer.lock`; the library does not steal locks. Memory persistence provides the same interface for simulation but is not durable across host process death. + +`summarizeFleet` separates selected results from the sum of all reported costs, including retries and late attempts. `observedCostUsd` is only the known subtotal. `usageComplete` is false if any dispatched attempt lacks a complete usage report; VM loss cannot be reported as zero usage. Provider quota scheduling and accounting for unobservable remote work remain outside this mock phase. diff --git a/packages/eval/README.md b/packages/eval/README.md index 548dffbdff..6056133a18 100644 --- a/packages/eval/README.md +++ b/packages/eval/README.md @@ -48,6 +48,11 @@ Maka subjects ask the Runtime Host client to run one owned execution in a dedica The result kernel contains only score, normalized usage, attributable cost, duration, status, and artifacts. Specs carry every semantic setting; environment variables are reserved for credentials and machine-local paths. +The mock-backed multi-VM coordinator and deterministic fault simulator are documented in +[Fleet scheduling](FLEET.md). They expose a separate library API; `maka eval run` retains its +existing local scheduling and file format. Remote VM transport and real fleet execution adapters +are not connected yet. + ## Experiment spec format A spec decodes to the `ExperimentSpec` interface ([`experiment.ts:46-70`](src/experiment.ts)), validated field-by-field by `parseExperimentSpec` ([`spec.ts:22`](src/spec.ts)) — there is no external JSON-schema dependency; the decoder is hand-written and strict (unrecognized top-level keys are rejected). diff --git a/packages/eval/package.json b/packages/eval/package.json index 62104106ec..cbcc53cafe 100644 --- a/packages/eval/package.json +++ b/packages/eval/package.json @@ -33,6 +33,7 @@ "build": "tsc -p tsconfig.json", "typecheck": "tsc -p tsconfig.json --noEmit", "test:dist": "node --test \"dist/**/*.test.js\" && python3 -m unittest discover --start-directory harbor --pattern 'test_*.py'", + "simulate:fleet": "npm run build && node dist/fleet-simulation.js", "test:egress-proxy:live": "python3 harbor/test_egress_filter_live.py" }, "dependencies": { diff --git a/packages/eval/src/__tests__/fleet.test.ts b/packages/eval/src/__tests__/fleet.test.ts new file mode 100644 index 0000000000..99120f95b5 --- /dev/null +++ b/packages/eval/src/__tests__/fleet.test.ts @@ -0,0 +1,544 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import type { ExperimentSpec } from '../experiment.js'; +import { + summarizeFleet, + type FleetAssignment, + type FleetPolicy, + type FleetReport, + type FleetState, +} from '../fleet.js'; +import { simulateFleet, simulationOutput } from '../fleet-simulation.js'; +import { + FileFleetPersistence, + FleetCoordinator, + MemoryFleetPersistence, + type FleetPersistence, + type FleetTransport, +} from '../fleet-store.js'; +import { FleetWorker } from '../fleet-worker.js'; + +function spec(tasks = 3, cap = 2): ExperimentSpec { + return { + schemaVersion: 'maka.eval.v1', + id: 'test', + benchmark: { id: 'test', version: '1', config: {} }, + executor: { kind: 'mock', config: {} }, + execution: { maxConcurrentTaskGroups: cap }, + subjects: ['fast', 'slow'].map((id) => ({ id, kind: 'external', credentials: [], config: {} })), + tasks: Array.from({ length: tasks }, (_, i) => ({ + id: `task-${i}`, + input: 'solve', + config: {}, + })), + repetitions: 1, + budget: {}, + verifier: {}, + }; +} +const policy: FleetPolicy = { + runId: 'test-run', + environmentId: 'reviewed-manifest', + groupCpus: 6, + groupMemoryMb: 1024, + leaseMs: 10, + recoveryGraceMs: 20, + retryBackoffMs: 2, + maxAttemptsPerCell: 2, +}; +function report(overrides: Partial = {}): FleetReport { + return { ...simulationOutput().report, artifacts: [], ...overrides }; +} +function worker(id: string, slots = 1) { + return { + id, + environmentId: policy.environmentId, + cpus: 6 * slots, + memoryMb: 1024 * slots, + groupSlots: slots, + }; +} +async function fixture( + tasks = 3, + cap = 2, + persistence: FleetPersistence = new MemoryFleetPersistence(), +) { + let now = 0; + const experiment = spec(tasks, cap); + let coordinator = await FleetCoordinator.open({ + persistence, + spec: experiment, + policy, + now: () => now, + }); + return { + get coordinator() { + return coordinator; + }, + persistence, + time(value: number) { + now = value; + }, + async restart() { + await coordinator.close(); + coordinator = await FleetCoordinator.open({ + persistence, + spec: experiment, + policy, + now: () => now, + }); + }, + async admit(id: string, slots = 1) { + await coordinator.command({ kind: 'register', worker: worker(id, slots) }); + await coordinator.command({ kind: 'heartbeat', workerId: id, assignmentIds: [] }); + }, + async claim(id: string) { + return (await coordinator.command({ kind: 'claim', workerId: id })).assignment; + }, + async submit(assignment: FleetAssignment, index = 0, value = report()) { + return coordinator.command({ + kind: 'report', + workerId: assignment.workerId, + attemptId: assignment.attemptIds[index], + report: value, + }); + }, + async finish(assignment: FleetAssignment) { + return coordinator.command({ + kind: 'finish', + workerId: assignment.workerId, + assignmentId: assignment.id, + }); + }, + }; +} + +test('atomic claims obey global and per-worker group capacity; a straggler holds only its group', async () => { + const f = await fixture(4, 2); + await f.admit('a'); + await f.admit('b'); + await f.admit('c'); + const claims = await Promise.all([f.claim('a'), f.claim('a'), f.claim('b'), f.claim('c')]); + assert.equal(claims.filter(Boolean).length, 2); + const a = claims[0]!; + const b = claims[2]!; + await f.submit(a); + assert.equal( + (await f.coordinator.snapshot()).attempts.filter((x) => x.disposition === 'selected').length, + 1, + ); + await assert.rejects(f.finish(a), /all cell reports/); + assert.equal(await f.claim('c'), undefined); + await f.submit(b, 0); + await f.submit(b, 1); + await f.finish(b); + const next = await f.claim('b'); + assert.ok(next); + assert.notEqual(next.groupId, a.groupId); + assert.equal( + (await f.coordinator.snapshot()).assignments.find((x) => x.id === a.id)?.status, + 'active', + ); +}); + +test('expired groups skip committed zeroes, fence late results, and retain duplicate execution cost', async () => { + const f = await fixture(1, 1); + await f.admit('a'); + await f.admit('b'); + const first = (await f.claim('a'))!; + const zero = report({ + execution: 'subject_failed', + result: { ...report().result, status: 'subject_failed', score: 0 }, + }); + await f.submit(first, 0, zero); + f.time(10); + await f.coordinator.command({ kind: 'tick' }); + f.time(12); + const replacement = (await f.claim('b'))!; + assert.equal(replacement.generation, 2); + assert.equal(replacement.attemptIds.length, 1); + assert.equal((await f.submit(first, 1)).disposition, 'late'); + assert.equal((await f.submit(first, 1)).disposition, 'late'); + await f.submit(replacement); + await f.finish(replacement); + const state = await f.coordinator.snapshot(); + const summary = summarizeFleet(state); + assert.equal(summary.completedCells, 2); + assert.equal(summary.results.find((r) => r.cellId.endsWith('fast'))?.result.score, 0); + assert.equal( + summary.results.find((r) => r.cellId.endsWith('slow'))?.attemptId, + replacement.attemptIds[0], + ); + assert.equal(summary.observedCostUsd, 0.03); + await assert.rejects(f.submit(first, 1, zero), /conflicting/); +}); + +test('coordinator restart grants reconciliation grace before expiring old ownership', async () => { + const f = await fixture(1, 1); + await f.admit('a'); + await f.admit('b'); + const first = (await f.claim('a'))!; + f.time(1000); + await f.restart(); + assert.equal(await f.claim('b'), undefined); + const recovery = await f.coordinator.command({ + kind: 'heartbeat', + workerId: 'a', + assignmentIds: [first.id], + }); + assert.deepEqual(recovery.activeAssignmentIds, [first.id]); + await f.submit(first, 0); + await f.submit(first, 1); + await f.finish(first); + assert.equal(summarizeFleet(await f.coordinator.snapshot()).completedCells, 2); + assert.equal((await f.coordinator.snapshot()).assignments.length, 1); +}); + +test('workers that never reconcile expire after the recovery grace', async () => { + const f = await fixture(1, 1); + await f.admit('a'); + await f.admit('b'); + await f.claim('a'); + f.time(1000); + await f.restart(); + await f.coordinator.command({ kind: 'heartbeat', workerId: 'b', assignmentIds: [] }); + f.time(1019); + assert.equal(await f.claim('b'), undefined); + f.time(1020); + await f.coordinator.command({ kind: 'tick' }); + f.time(1022); + assert.ok(await f.claim('b')); +}); + +test('deterministic environment failures block dispatch until repair; admission includes verifier capacity', async () => { + const f = await fixture(2, 2); + await assert.rejects( + f.coordinator.command({ kind: 'register', worker: { ...worker('small'), cpus: 4 } }), + /admission/, + ); + await assert.rejects( + f.coordinator.command({ + kind: 'register', + worker: { ...worker('wrong'), environmentId: 'wrong' }, + }), + /admission/, + ); + await f.admit('a'); + await f.admit('b'); + const first = (await f.claim('a'))!; + const failure = report({ + execution: 'not_started', + verification: 'not_run', + environmentFailure: 'verifier certificate missing', + result: { ...report().result, status: 'infra_failed', score: null }, + }); + await f.submit(first, 0, failure); + await f.submit(first, 1, failure); + await f.finish(first); + f.time(2); + assert.equal(await f.claim('b'), undefined); + await f.coordinator.command({ kind: 'repair', reason: 'verifier certificate missing' }); + assert.ok(await f.claim('b')); +}); + +test('retry exhaustion is incomplete evaluation, not a synthetic zero', async () => { + const f = await fixture(1, 1); + await f.admit('a'); + const failure = report({ + execution: 'unknown', + verification: 'invalid', + cleanup: 'unknown', + result: { ...report().result, status: 'indeterminate', score: null }, + }); + for (const at of [0, 2]) { + f.time(at); + const assignment = (await f.claim('a'))!; + await f.submit(assignment, 0, failure); + await f.submit(assignment, 1, failure); + await f.finish(assignment); + } + f.time(4); + assert.equal(await f.claim('a'), undefined); + const summary = summarizeFleet(await f.coordinator.snapshot()); + assert.equal(summary.exhaustedCells, 2); + assert.equal(summary.incompleteCells, 2); + assert.equal(summary.settled, true); + assert.deepEqual(summary.results, []); +}); + +test('selection uses execution/verification/cleanup evidence independently of usage and coarse status', async () => { + const f = await fixture(1, 1); + await f.admit('a'); + const assignment = (await f.claim('a'))!; + const missing = report({ + usage: 'missing', + result: { ...report().result, status: 'indeterminate', usage: null, costUsd: null }, + }); + assert.equal((await f.submit(assignment, 0, missing)).disposition, 'selected'); + const uncertainCleanup = report({ cleanup: 'unknown' }); + assert.equal((await f.submit(assignment, 1, uncertainCleanup)).disposition, 'retryable'); + await f.finish(assignment); + f.time(2); + assert.equal((await f.claim('a'))?.attemptIds.length, 1); + const summary = summarizeFleet(await f.coordinator.snapshot()); + assert.equal(summary.completedCells, 1); + assert.equal(summary.usageComplete, false); + assert.equal(summary.results[0].result.status, 'completed'); + assert.equal((await f.coordinator.snapshot()).attempts[0].report?.result.status, 'indeterminate'); +}); + +test('invalid evidence and foreign reports cannot change durable progress', async () => { + const f = await fixture(1, 1); + await f.admit('a'); + await f.admit('b'); + const assignment = (await f.claim('a'))!; + const before = await f.coordinator.snapshot(); + await assert.rejects( + f.submit(assignment, 0, report({ execution: 'unknown' })), + /verification evidence/, + ); + await assert.rejects( + f.coordinator.command({ + kind: 'report', + workerId: 'b', + attemptId: assignment.attemptIds[0], + report: report(), + }), + /foreign/, + ); + await assert.rejects(f.submit(assignment, 0, report({ usage: 'missing' })), /missing usage/); + assert.deepEqual(await f.coordinator.snapshot(), before); +}); + +test('lost claim and report acknowledgements recover through the real worker outbox', async () => { + const f = await fixture(1, 1); + const lost = new Set(); + let executions = 0; + const transport: FleetTransport = { + async command(command) { + const reply = await f.coordinator.command(command); + if (['claim', 'report', 'finish'].includes(command.kind) && !lost.has(command.kind)) { + lost.add(command.kind); + throw new Error('reply lost'); + } + return reply; + }, + putArtifact: (bytes) => f.coordinator.putArtifact(bytes), + }; + const vm = new FleetWorker(worker('a'), transport, async (work, emit) => { + executions++; + for (const cell of work.cells) emit(cell.attemptId, simulationOutput()); + }); + for (let i = 0; i < 10; i++) { + try { + await vm.poll(); + } catch (error) { + assert.match(String(error), /reply lost/); + } + } + assert.equal(executions, 1); + assert.equal(vm.bufferedReports, 0); + assert.equal(vm.activeGroups, 0); + const state = await f.coordinator.snapshot(); + assert.equal(state.assignments.length, 1); + assert.equal(summarizeFleet(state).completedCells, 2); + for (const attempt of state.attempts) assert.equal(attempt.report?.artifacts.length, 1); +}); + +test('worker buffers completed cells during coordinator outage and resumes without reexecution', async () => { + const f = await fixture(1, 1); + let offline = false; + let executions = 0; + let complete = () => {}; + const vm = new FleetWorker( + worker('a'), + { + command: (command) => + offline ? Promise.reject(new Error('offline')) : f.coordinator.command(command), + putArtifact: (bytes) => + offline ? Promise.reject(new Error('offline')) : f.coordinator.putArtifact(bytes), + }, + (work, emit) => + new Promise((resolve) => { + executions++; + complete = () => { + for (const cell of work.cells) emit(cell.attemptId, simulationOutput()); + resolve(); + }; + }), + ); + await vm.poll(); + offline = true; + complete(); + await assert.rejects(vm.poll(), /offline/); + assert.equal(vm.bufferedReports, 2); + f.time(1000); + await f.restart(); + offline = false; + await vm.poll(); + await vm.poll(); + assert.equal(executions, 1); + assert.equal(summarizeFleet(await f.coordinator.snapshot()).completedCells, 2); +}); + +test('persistence failures before and after commit never expose uncommitted work or lose committed ownership', async () => { + class FaultStore extends MemoryFleetPersistence { + fault: 'before' | 'after' | null = null; + override async save(state: FleetState) { + const fault = this.fault; + this.fault = null; + if (fault === 'before') throw new Error('disk failed before commit'); + await super.save(state); + if (fault === 'after') throw new Error('disk acknowledgement lost'); + } + } + const store = new FaultStore(); + const f = await fixture(1, 1, store); + await f.admit('a'); + store.fault = 'before'; + await assert.rejects(f.claim('a'), /before commit/); + assert.equal((await f.coordinator.snapshot()).assignments.length, 0); + store.fault = 'after'; + await assert.rejects(f.claim('a'), /acknowledgement lost/); + assert.equal((await f.coordinator.snapshot()).assignments.length, 1); + await f.restart(); + const recovered = await f.coordinator.command({ + kind: 'heartbeat', + workerId: 'a', + assignmentIds: [], + }); + assert.equal(recovered.work?.length, 1); + assert.equal(await f.claim('a'), undefined); +}); + +test('file storage reopens committed state/artifacts, enforces one writer and rejects corruption', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-fleet-')); + const stores: FileFleetPersistence[] = []; + t.after(async () => { + for (const store of stores) await store.close(); + await rm(root, { recursive: true, force: true }); + }); + const storage = await FileFleetPersistence.open(root); + stores.push(storage); + await assert.rejects(FileFleetPersistence.open(root), /EEXIST/); + const f = await fixture(1, 1, storage); + await f.admit('a'); + const assignment = (await f.claim('a'))!; + const bytes = new TextEncoder().encode('durable trajectory'); + const ref = await storage.putArtifact(bytes); + await f.submit(assignment, 0, report({ artifacts: [ref] })); + await assert.rejects( + f.submit(assignment, 1, report({ artifacts: [{ sha256: '0'.repeat(64), bytes: 1 }] })), + /ENOENT/, + ); + await f.coordinator.close(); + await storage.close(); + await assert.rejects(storage.load(), /closed/); + const reopened = await FileFleetPersistence.open(root); + stores.push(reopened); + assert.deepEqual(new Uint8Array(await reopened.getArtifact(ref)), bytes); + assert.equal((await reopened.load())?.attempts[0].disposition, 'selected'); + const recovered = await FleetCoordinator.open({ + persistence: reopened, + spec: spec(1, 1), + policy, + now: () => 1000, + }); + const heartbeat = await recovered.command({ + kind: 'heartbeat', + workerId: 'a', + assignmentIds: [assignment.id], + }); + assert.equal(heartbeat.work?.[0].cells.length, 1); + assert.equal(heartbeat.work?.[0].cells[0].attemptId, assignment.attemptIds[1]); + await recovered.close(); + await assert.rejects( + FleetCoordinator.open({ + persistence: reopened, + spec: spec(1, 1), + policy: { ...policy, runId: 'other' }, + now: () => 0, + }), + /identity differs/, + ); + await writeFile(join(root, 'artifacts', ref.sha256), 'corrupt'); + await assert.rejects(reopened.getArtifact(ref), /checksum/); + const path = join(root, 'state.json'); + const envelope = JSON.parse(await readFile(path, 'utf8')); + envelope.state.nextAssignment = 999; + await writeFile(path, JSON.stringify(envelope)); + await assert.rejects(reopened.load(), /checksum/); +}); + +test('closing a coordinator drains accepted commands and rejects old-owner writes', async () => { + const f = await fixture(1, 1); + await f.admit('a'); + await assert.rejects( + FleetCoordinator.open({ persistence: f.persistence, spec: spec(1, 1), policy, now: () => 0 }), + /already has a coordinator/, + ); + const claiming = f.claim('a'); + await f.coordinator.close(); + assert.ok(await claiming); + await assert.rejects(f.claim('a'), /coordinator closed/); + assert.equal((await f.persistence.load())?.assignments.length, 1); +}); + +test('explicit coordinator pause freezes leases and dispatch until recovery', async () => { + const f = await fixture(2, 2); + await f.admit('a'); + await f.admit('b'); + const assignment = (await f.claim('a'))!; + await f.coordinator.command({ kind: 'pause' }); + f.time(1000); + await f.coordinator.command({ kind: 'tick' }); + assert.equal(await f.claim('b'), undefined); + assert.equal((await f.coordinator.snapshot()).assignments[0].status, 'active'); + await f.coordinator.command({ kind: 'recover' }); + await f.coordinator.command({ kind: 'heartbeat', workerId: 'a', assignmentIds: [assignment.id] }); + await f.submit(assignment, 0); + await f.submit(assignment, 1); + await f.finish(assignment); + assert.equal((await f.coordinator.snapshot()).assignments.length, 1); +}); + +test('the same seed reproduces the entire fault trace and selected results byte for byte', async () => { + const first = await simulateFleet(42); + const second = await simulateFleet(42); + assert.deepEqual(first, second); + assert.ok(first.trace.some((line) => line.includes('dropped-before'))); + assert.ok(first.trace.some((line) => line.includes('report dropped-after'))); + assert.ok(first.state.attempts.some((a) => a.disposition === 'late')); + assert.equal(first.summary.completedCells, 120); +}); + +for (let seed = 0; seed < 16; seed++) { + test(`deterministic VM/network/restart simulation converges without duplicate selections: seed=${seed}`, async () => { + const simulation = await simulateFleet(seed); + assert.equal(simulation.summary.completedCells, 120); + assert.equal(simulation.summary.exhaustedCells, 0); + assert.equal(simulation.summary.settled, true); + }); +} diff --git a/packages/eval/src/fleet-simulation.ts b/packages/eval/src/fleet-simulation.ts new file mode 100644 index 0000000000..af667e1d17 --- /dev/null +++ b/packages/eval/src/fleet-simulation.ts @@ -0,0 +1,284 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { pathToFileURL } from 'node:url'; +import type { ExperimentSpec } from './experiment.js'; +import { summarizeFleet, type FleetPolicy, type FleetState } from './fleet.js'; +import { FleetCoordinator, MemoryFleetPersistence, type FleetTransport } from './fleet-store.js'; +import { FleetWorker, type FleetCellOutput } from './fleet-worker.js'; + +/** Replayable virtual-time fault simulation, exercising the real coordinator and worker loop. */ +export async function simulateFleet(seed: number) { + if (!Number.isSafeInteger(seed) || seed < 0 || seed > 0xffffffff) + throw new Error('seed must be uint32'); + let randomState = seed; + const random = () => { + randomState = (Math.imul(1664525, randomState) + 1013904223) >>> 0; + return randomState; + }; + const spec: ExperimentSpec = { + schemaVersion: 'maka.eval.v1', + id: 'fleet-simulation', + benchmark: { id: 'mock-benchmark', version: '1', config: {} }, + executor: { kind: 'mock', config: {} }, + execution: { maxConcurrentTaskGroups: 3 }, + subjects: ['a', 'b', 'c'].map((id) => ({ id, kind: 'external', credentials: [], config: {} })), + tasks: Array.from({ length: 20 }, (_, index) => ({ + id: `task-${index}`, + input: 'mock', + config: {}, + })), + repetitions: 2, + budget: {}, + verifier: {}, + }; + const policy: FleetPolicy = { + runId: `seed-${seed}`, + environmentId: 'mock-manifest-v1', + groupCpus: 6, + groupMemoryMb: 1024, + leaseMs: 12, + recoveryGraceMs: 15, + retryBackoffMs: 2, + maxAttemptsPerCell: 20, + }; + let now = 0; + const persistence = new MemoryFleetPersistence(); + let coordinator = await FleetCoordinator.open({ persistence, spec, policy, now: () => now }); + const trace: string[] = []; + let order = 0; + const events: { at: number; order: number; action: () => void }[] = []; + let crashed = false; + const lost = new Set(); + const committed = new Map(); + const workers = Array.from({ length: 3 }, (_, index) => { + const id = `vm-${index}/boot-1`; + const disconnected = () => (now >= 55 && now < 100) || (index === 0 && now >= 8 && now < 38); + const transport: FleetTransport = { + async command(command) { + if (disconnected()) throw new SimulatedLinkError(); + const roll = random(); + if (now < 240 && roll % 31 === 0) { + trace.push(`${now} ${id} ${command.kind} dropped-before`); + throw new SimulatedLinkError(); + } + const reply = await coordinator.command(command); + const once = `${id}/${command.kind}`; + // Guaranteed lost claim/report acknowledgements, plus seeded background faults. + if ( + now < 240 && + ((['claim', 'report'].includes(command.kind) && !lost.has(once)) || roll % 37 === 0) + ) { + lost.add(once); + trace.push(`${now} ${id} ${command.kind} dropped-after`); + throw new SimulatedLinkError(); + } + return reply; + }, + async putArtifact(bytes) { + if (disconnected()) throw new SimulatedLinkError(); + return coordinator.putArtifact(bytes); + }, + }; + return new FleetWorker( + { + id, + environmentId: policy.environmentId, + cpus: 6, + memoryMb: 1024, + groupSlots: 1, + }, + transport, + (work, emit) => + new Promise((resolve) => { + trace.push(`${now} ${id} start ${work.assignment.id} cells=${work.cells.length}`); + let remaining = work.cells.length; + if (!remaining) { + resolve(); + return; + } + for (const cell of work.cells) { + const delay = 2 + (random() % 19); + events.push({ + at: now + delay, + order: order++, + action: () => { + if (index === 2 && crashed) return; + const output = simulationOutput(); + // Valid outcomes with absent metering must remain selectable. + if (random() % 5 === 0) { + output.report.usage = 'missing'; + output.report.result = { + ...output.report.result, + usage: null, + costUsd: null, + status: 'indeterminate', + }; + } + emit(cell.attemptId, output); + trace.push(`${now} ${id} completed ${cell.attemptId}`); + if (--remaining === 0) resolve(); + }, + }); + } + }), + ); + }); + for (now = 0; now < 3000; now++) { + if (now === 20) { + crashed = true; + trace.push(`${now} vm-2 process-crash`); + } + if (now === 55) { + await coordinator.command({ kind: 'pause' }); + trace.push(`${now} coordinator-offline`); + } + if (now === 100) { + await coordinator.close(); + coordinator = await FleetCoordinator.open({ persistence, spec, policy, now: () => now }); + trace.push(`${now} coordinator-restarted`); + } + events.sort((a, b) => a.at - b.at || a.order - b.order); + while (events[0]?.at <= now) events.shift()!.action(); + // Drain the finite execute/catch/finally chain, without timers or real sleeps. + for (let i = 0; i < 8; i++) await Promise.resolve(); + for (const [index, worker] of workers.entries()) { + if (index === 2 && crashed) continue; + try { + await worker.poll(); + } catch (error) { + if (!(error instanceof SimulatedLinkError)) throw error; + } + } + await coordinator.command({ kind: 'tick' }); + const state = await coordinator.snapshot(); + try { + assertFleetInvariants(state); + for (const [id, evidence] of committed) { + const attempt = state.attempts.find((a) => a.id === id); + if (!attempt || JSON.stringify(attempt) !== evidence) + throw new Error('committed result changed or disappeared'); + } + for (const attempt of state.attempts) { + if (attempt.disposition === 'selected') committed.set(attempt.id, JSON.stringify(attempt)); + } + } catch (error) { + throw new Error( + `simulation invariant failed: seed=${seed} tick=${now}\n${trace.slice(-30).join('\n')}`, + { cause: error }, + ); + } + if (summarizeFleet(state).settled) + return { seed, elapsedTicks: now, trace, state, summary: summarizeFleet(state) }; + } + throw new Error(`simulation did not settle: seed=${seed}\n${trace.slice(-30).join('\n')}`); +} + +/** Cross-cutting safety oracle, checked after every simulated tick. */ +export function assertFleetInvariants(state: FleetState) { + const fail = (message: string): never => { + throw new Error(`fleet invariant: ${message}`); + }; + const active = state.assignments.filter((a) => a.status === 'active'); + if (active.length > state.spec.execution.maxConcurrentTaskGroups) fail('global capacity'); + if (new Set(active.map((a) => a.groupId)).size !== active.length) fail('group ownership'); + for (const worker of state.workers) { + if ( + active.filter((a) => a.workerId === worker.description.id).length > + worker.description.groupSlots + ) + fail('worker capacity'); + } + for (const cellId of state.groups.flatMap((g) => g.cellIds)) { + const attempts = state.attempts.filter((a) => a.cellId === cellId); + if (attempts.filter((a) => a.disposition === 'selected').length > 1) fail('duplicate result'); + if (attempts.length > state.policy.maxAttemptsPerCell) fail('retry budget'); + if (attempts.some((a, i) => a.sequence !== i + 1)) fail('attempt ordering'); + } + if (new Set(state.attempts.map((a) => a.id)).size !== state.attempts.length) + fail('attempt identity'); + for (const assignment of state.assignments) { + const group = state.groups.find((g) => g.id === assignment.groupId); + if ( + !group || + assignment.attemptIds.some((id) => { + const attempt = state.attempts.find((a) => a.id === id); + return ( + !attempt || + attempt.assignmentId !== assignment.id || + !group.cellIds.includes(attempt.cellId) + ); + }) + ) + fail('assignment membership'); + } + for (const attempt of state.attempts.filter((a) => a.disposition === 'selected')) { + if ( + !attempt.report || + attempt.report.verification !== 'valid' || + attempt.report.cleanup !== 'confirmed' || + attempt.report.environmentFailure !== null || + !['completed', 'subject_failed'].includes(attempt.report.execution) + ) + fail('selected evidence'); + } +} + +export function simulationOutput(): FleetCellOutput { + return { + report: { + result: { + status: 'completed', + score: 1, + usage: { + inputTokens: 1, + outputTokens: 1, + cacheReadTokens: 0, + cacheWriteTokens: 0, + reasoningTokens: 0, + totalTokens: 2, + }, + costUsd: 0.01, + durationMs: 1, + failureReason: null, + artifacts: [], + }, + execution: 'completed', + verification: 'valid', + usage: 'complete', + cleanup: 'confirmed', + environmentFailure: null, + }, + artifacts: [new TextEncoder().encode('mock subject artifact')], + }; +} + +class SimulatedLinkError extends Error {} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + const seed = Number(process.argv[2] ?? 1); + const result = await simulateFleet(seed); + process.stdout.write( + `${JSON.stringify( + { seed, elapsedTicks: result.elapsedTicks, ...result.summary, trace: result.trace }, + null, + 2, + )}\n`, + ); +} diff --git a/packages/eval/src/fleet-store.ts b/packages/eval/src/fleet-store.ts new file mode 100644 index 0000000000..4b7226b340 --- /dev/null +++ b/packages/eval/src/fleet-store.ts @@ -0,0 +1,273 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { createHash, randomUUID } from 'node:crypto'; +import { mkdir, open, readFile, rename, unlink } from 'node:fs/promises'; +import { join } from 'node:path'; +import { stableJsonStringify } from '@maka/core/canonical-json'; +import type { ExperimentSpec } from './experiment.js'; +import { + createFleetState, + transitionFleet, + type FleetArtifact, + type FleetCommand, + type FleetPolicy, + type FleetReply, + type FleetState, + type FleetWorkerCommand, +} from './fleet.js'; + +/** Exactly one coordinator owns a persistence instance. save must be atomic and durable. */ +export interface FleetPersistence { + load(): Promise; + save(state: FleetState): Promise; + putArtifact(bytes: Uint8Array): Promise; + getArtifact(ref: FleetArtifact): Promise; +} + +export interface FleetTransport { + command(command: FleetWorkerCommand): Promise; + putArtifact(bytes: Uint8Array): Promise; +} + +export class FleetCoordinator implements FleetTransport { + static #owners = new WeakSet(); + #tail: Promise = Promise.resolve(); + #closed = false; + private constructor( + readonly persistence: FleetPersistence, + private readonly now: () => number, + ) {} + + static async open(input: { + persistence: FleetPersistence; + spec: ExperimentSpec; + policy: FleetPolicy; + now: () => number; + }): Promise { + if (FleetCoordinator.#owners.has(input.persistence)) + throw new Error('fleet persistence already has a coordinator'); + FleetCoordinator.#owners.add(input.persistence); + try { + const initial = createFleetState(input.spec, input.policy); + const existing = await input.persistence.load(); + if ( + existing && + (existing.version !== initial.version || + stableJsonStringify(existing.spec) !== stableJsonStringify(initial.spec) || + stableJsonStringify(existing.policy) !== stableJsonStringify(initial.policy)) + ) { + throw new Error('fleet run identity differs'); + } + if (!existing) await input.persistence.save(initial); + const coordinator = new FleetCoordinator(input.persistence, input.now); + await coordinator.command({ kind: 'recover' }); + return coordinator; + } catch (error) { + FleetCoordinator.#owners.delete(input.persistence); + throw error; + } + } + + command(command: FleetCommand): Promise { + if (this.#closed) return Promise.reject(new Error('coordinator closed')); + const captured = structuredClone(command); + const operation = this.#tail.then(async () => { + const state = await this.persistence.load(); + if (!state) throw new Error('fleet state missing'); + if (captured.kind === 'report') { + for (const artifact of captured.report.artifacts) + await this.persistence.getArtifact(artifact); + } + const next = transitionFleet(state, captured, this.now()); + await this.persistence.save(next.state); + return next.reply; + }); + this.#tail = operation.catch(() => undefined); + return operation; + } + + async snapshot(): Promise { + await this.#tail; + const state = await this.persistence.load(); + if (!state) throw new Error('fleet state missing'); + return state; + } + + putArtifact(bytes: Uint8Array): Promise { + if (this.#closed) return Promise.reject(new Error('coordinator closed')); + return this.persistence.putArtifact(bytes); + } + + /** Stop accepting commands and drain commits before closing persistence or replacing this owner. */ + async close() { + if (this.#closed) { + await this.#tail; + return; + } + this.#closed = true; + await this.#tail; + FleetCoordinator.#owners.delete(this.persistence); + } +} + +export class MemoryFleetPersistence implements FleetPersistence { + #state: FleetState | null = null; + #artifacts = new Map(); + async load() { + return structuredClone(this.#state); + } + async save(state: FleetState) { + this.#state = structuredClone(state); + } + async putArtifact(bytes: Uint8Array) { + const ref = reference(bytes); + this.#artifacts.set(ref.sha256, Uint8Array.from(bytes)); + return ref; + } + async getArtifact(ref: FleetArtifact) { + validateReference(ref); + const bytes = this.#artifacts.get(ref.sha256); + if (!bytes) throw new Error('artifact missing'); + verifyArtifact(ref, bytes); + return Uint8Array.from(bytes); + } +} + +/** Local coordinator storage. A stale writer lock requires explicit operator recovery. */ +export class FileFleetPersistence implements FleetPersistence { + #tail: Promise = Promise.resolve(); + #closing: Promise | undefined; + private constructor( + readonly root: string, + private readonly release: () => Promise, + ) {} + + static async open(root: string): Promise { + await mkdir(root, { recursive: true }); + const lockPath = join(root, '.writer.lock'); + const lock = await open(lockPath, 'wx', 0o600); + let closed = false; + return new FileFleetPersistence(root, async () => { + if (closed) return; + closed = true; + await lock.close(); + await unlink(lockPath); + }); + } + + async close() { + this.#closing ??= this.#tail.then(() => this.release()); + await this.#closing; + } + + #operation(operation: () => Promise): Promise { + if (this.#closing) return Promise.reject(new Error('fleet persistence closed')); + const running = this.#tail.then(operation); + this.#tail = running.catch(() => undefined); + return running; + } + + load(): Promise { + return this.#operation(() => this.#load()); + } + + async #load(): Promise { + let raw: string; + try { + raw = await readFile(join(this.root, 'state.json'), 'utf8'); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return null; + throw error; + } + const envelope = JSON.parse(raw) as { sha256: string; state: FleetState }; + if (digest(Buffer.from(stableJsonStringify(envelope.state))) !== envelope.sha256) { + throw new Error('fleet state checksum mismatch'); + } + if (envelope.state.version !== 'maka.eval.fleet.v1') throw new Error('unsupported fleet state'); + return envelope.state; + } + + async save(state: FleetState) { + const canonical = stableJsonStringify(state); + const bytes = Buffer.from(JSON.stringify({ sha256: digest(Buffer.from(canonical)), state })); + await this.#operation(() => atomicWrite(this.root, 'state.json', bytes)); + } + + async putArtifact(bytes: Uint8Array): Promise { + const captured = Uint8Array.from(bytes); + return this.#operation(async () => { + const ref = reference(captured); + const directory = join(this.root, 'artifacts'); + await mkdir(directory, { recursive: true }); + await atomicWrite(directory, ref.sha256, captured); + return ref; + }); + } + + async getArtifact(ref: FleetArtifact): Promise { + validateReference(ref); + const captured = { ...ref }; + return this.#operation(async () => { + const bytes = await readFile(join(this.root, 'artifacts', captured.sha256)); + verifyArtifact(captured, bytes); + return bytes; + }); + } +} + +async function atomicWrite(directory: string, name: string, bytes: Uint8Array) { + const temporary = join(directory, `.${name}.${randomUUID()}.tmp`); + const file = await open(temporary, 'wx', 0o600); + try { + await file.writeFile(bytes); + await file.sync(); + await file.close(); + await rename(temporary, join(directory, name)); + if (process.platform !== 'win32') { + const parent = await open(directory, 'r'); + try { + await parent.sync(); + } finally { + await parent.close(); + } + } + } finally { + await file.close().catch(() => undefined); + await unlink(temporary).catch((error: NodeJS.ErrnoException) => { + if (error.code !== 'ENOENT') throw error; + }); + } +} + +function digest(bytes: Uint8Array) { + return createHash('sha256').update(bytes).digest('hex'); +} +function reference(bytes: Uint8Array): FleetArtifact { + return { sha256: digest(bytes), bytes: bytes.byteLength }; +} +function validateReference(ref: FleetArtifact) { + if (!/^[a-f0-9]{64}$/.test(ref.sha256) || !Number.isSafeInteger(ref.bytes) || ref.bytes < 0) { + throw new Error('invalid artifact reference'); + } +} +function verifyArtifact(ref: FleetArtifact, bytes: Uint8Array) { + if (bytes.byteLength !== ref.bytes || digest(bytes) !== ref.sha256) + throw new Error('artifact checksum mismatch'); +} diff --git a/packages/eval/src/fleet-worker.ts b/packages/eval/src/fleet-worker.ts new file mode 100644 index 0000000000..c9e5502a5d --- /dev/null +++ b/packages/eval/src/fleet-worker.ts @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import type { FleetReport, FleetWork, FleetWorkerDescription } from './fleet.js'; +import type { FleetTransport } from './fleet-store.js'; + +export interface FleetCellOutput { + report: Omit; + artifacts: Uint8Array[]; +} + +/** Implemented by mock VMs for now. The callback reports cells before the group settles. */ +export type FleetGroupExecution = ( + work: FleetWork, + emit: (attemptId: string, output: FleetCellOutput) => void, +) => Promise; + +/** + * Transport-independent worker loop. Its outbox survives disconnection, not process death. + * The caller drives polling; no wall clock, timers, networking or cloud SDK is hidden here. + */ +export class FleetWorker { + #runs = new Map }>(); + #outbox = new Map(); + #polling: Promise | undefined; + + constructor( + readonly description: FleetWorkerDescription, + private readonly transport: FleetTransport, + private readonly execute: FleetGroupExecution, + ) {} + + poll(): Promise { + this.#polling ??= this.#poll().finally(() => { + this.#polling = undefined; + }); + return this.#polling; + } + + get bufferedReports() { + return this.#outbox.size; + } + get activeGroups() { + return this.#runs.size; + } + + async #poll() { + await this.transport.command({ kind: 'register', worker: this.description }); + const heartbeat = await this.transport.command({ + kind: 'heartbeat', + workerId: this.description.id, + assignmentIds: [...this.#runs.keys()], + }); + for (const work of heartbeat.work ?? []) this.#adopt(work); + for (const [attemptId, output] of this.#outbox) { + const artifacts = []; + for (const bytes of output.artifacts) artifacts.push(await this.transport.putArtifact(bytes)); + await this.transport.command({ + kind: 'report', + workerId: this.description.id, + attemptId, + report: { ...output.report, artifacts }, + }); + this.#outbox.delete(attemptId); + } + for (const [assignmentId, run] of this.#runs) { + if (!run.settled || run.work.cells.some((cell) => this.#outbox.has(cell.attemptId))) continue; + await this.transport.command({ kind: 'finish', workerId: this.description.id, assignmentId }); + this.#runs.delete(assignmentId); + } + while (this.#runs.size < this.description.groupSlots) { + const reply = await this.transport.command({ kind: 'claim', workerId: this.description.id }); + if (!reply.assignment) break; + for (const work of reply.work ?? []) this.#adopt(work); + } + } + + #adopt(work: FleetWork) { + if (this.#runs.has(work.assignment.id)) return; + const run = { work, settled: false, emitted: new Set() }; + this.#runs.set(work.assignment.id, run); + const emit = (attemptId: string, output: FleetCellOutput) => { + if (!work.cells.some((cell) => cell.attemptId === attemptId) || run.emitted.has(attemptId)) { + throw new Error('executor emitted an unknown or duplicate attempt'); + } + run.emitted.add(attemptId); + this.#outbox.set(attemptId, structuredClone(output)); + }; + // A failed local operation is an execution uncertainty, not a provider classification. + void Promise.resolve() + .then(() => this.execute(work, emit)) + .catch(() => undefined) + .finally(() => { + for (const cell of work.cells) { + if (run.emitted.has(cell.attemptId)) continue; + emit(cell.attemptId, { + report: { + result: { + status: 'infra_failed', + score: null, + usage: null, + costUsd: null, + durationMs: 0, + failureReason: 'worker execution ended without a cell report', + artifacts: [], + }, + execution: 'unknown', + verification: 'not_run', + usage: 'missing', + cleanup: 'unknown', + environmentFailure: null, + }, + artifacts: [], + }); + } + run.settled = true; + }); + } +} diff --git a/packages/eval/src/fleet.ts b/packages/eval/src/fleet.ts new file mode 100644 index 0000000000..2f1effcc11 --- /dev/null +++ b/packages/eval/src/fleet.ts @@ -0,0 +1,459 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { stableJsonStringify } from '@maka/core/canonical-json'; +import { expandExperiment, type ExperimentCell, type ExperimentSpec } from './experiment.js'; +import { decodeEvalResult, type EvalResult } from './result.js'; +import { parseExperimentSpec } from './spec.js'; + +/** A worker ID identifies one process incarnation, not a reusable VM hostname. */ +export interface FleetWorkerDescription { + id: string; + environmentId: string; + cpus: number; + memoryMb: number; + groupSlots: number; +} + +export interface FleetPolicy { + runId: string; + /** Identity of the reviewed code, benchmark, toolchain and image manifest. */ + environmentId: string; + /** Must cover both subject execution and verification for one entire group. */ + groupCpus: number; + groupMemoryMb: number; + leaseMs: number; + recoveryGraceMs: number; + retryBackoffMs: number; + maxAttemptsPerCell: number; +} + +export interface FleetArtifact { + sha256: string; + bytes: number; +} + +/** Evidence comes from the execution adapter, never from provider-error parsing here. */ +export interface FleetReport { + result: EvalResult; + execution: 'completed' | 'subject_failed' | 'not_started' | 'unknown'; + verification: 'valid' | 'invalid' | 'not_run'; + usage: 'complete' | 'partial' | 'missing'; + cleanup: 'confirmed' | 'unknown'; + artifacts: FleetArtifact[]; + /** A deterministic environment defect blocks this run until explicitly repaired. */ + environmentFailure: string | null; +} + +export interface FleetAttempt { + id: string; + cellId: string; + sequence: number; + assignmentId: string; + report: FleetReport | null; + disposition: 'pending' | 'selected' | 'retryable' | 'late'; +} + +export interface FleetAssignment { + id: string; + groupId: string; + workerId: string; + generation: number; + attemptIds: string[]; + status: 'active' | 'finished' | 'expired'; + expiresAt: number; +} + +export interface FleetState { + version: 'maka.eval.fleet.v1'; + spec: ExperimentSpec; + policy: FleetPolicy; + clock: number; + nextAssignment: number; + workers: { description: FleetWorkerDescription; reconciled: boolean }[]; + groups: { id: string; cellIds: string[]; generation: number; readyAt: number }[]; + assignments: FleetAssignment[]; + attempts: FleetAttempt[]; + blocked: string[]; + paused: boolean; +} + +export type FleetCommand = + | { kind: 'register'; worker: FleetWorkerDescription } + | { kind: 'claim'; workerId: string } + | { kind: 'heartbeat'; workerId: string; assignmentIds: string[] } + | { kind: 'report'; workerId: string; attemptId: string; report: FleetReport } + | { kind: 'finish'; workerId: string; assignmentId: string } + | { kind: 'recover' } + | { kind: 'pause' } + | { kind: 'tick' } + | { kind: 'repair'; reason: string }; + +export type FleetWorkerCommand = Extract< + FleetCommand, + { kind: 'register' | 'claim' | 'heartbeat' | 'report' | 'finish' } +>; + +export interface FleetReply { + assignment?: FleetAssignment; + disposition?: FleetAttempt['disposition']; + activeAssignmentIds?: string[]; + work?: FleetWork[]; +} + +export interface FleetWork { + assignment: FleetAssignment; + cells: { attemptId: string; cell: ExperimentCell }[]; +} + +export function createFleetState(spec: ExperimentSpec, policy: FleetPolicy): FleetState { + const frozenSpec = parseExperimentSpec(spec); + for (const key of ['runId', 'environmentId'] as const) nonempty(policy[key], key); + for (const key of [ + 'groupCpus', + 'groupMemoryMb', + 'leaseMs', + 'recoveryGraceMs', + 'maxAttemptsPerCell', + ] as const) + positive(policy[key], key); + if (!Number.isSafeInteger(policy.retryBackoffMs) || policy.retryBackoffMs < 0) { + throw new Error('invalid retryBackoffMs'); + } + const groups = new Map(); + const ids = new Set(); + for (const cell of expandExperiment(frozenSpec)) { + if (ids.has(cell.id)) throw new Error('ambiguous experiment cell identity'); + ids.add(cell.id); + const id = JSON.stringify([cell.task.id, cell.repetition]); + const group = groups.get(id) ?? []; + group.push(cell.id); + groups.set(id, group); + } + return { + version: 'maka.eval.fleet.v1', + spec: frozenSpec, + policy: structuredClone(policy), + clock: 0, + nextAssignment: 1, + workers: [], + assignments: [], + attempts: [], + blocked: [], + paused: false, + groups: [...groups].map(([id, cellIds]) => ({ id, cellIds, generation: 0, readyAt: 0 })), + }; +} + +/** Pure deterministic transition. Persist the returned state before delivering its reply. */ +export function transitionFleet( + previous: FleetState, + command: FleetCommand, + now: number, +): { state: FleetState; reply: FleetReply } { + if (!Number.isSafeInteger(now) || now < 0) throw new Error('invalid fleet clock'); + const state = structuredClone(previous); + state.clock = Math.max(state.clock, now); + if (command.kind === 'pause') { + state.paused = true; + return { state, reply: {} }; + } + // Recovery must run BEFORE expiry: a coordinator outage is not evidence of VM loss. + if (command.kind === 'recover') { + state.paused = false; + for (const worker of state.workers) worker.reconciled = false; + for (const assignment of active(state)) { + assignment.expiresAt = Math.max( + assignment.expiresAt, + state.clock + state.policy.recoveryGraceMs, + ); + } + return { state, reply: {} }; + } + for (const assignment of active(state)) { + if (!state.paused && assignment.expiresAt <= state.clock) settle(state, assignment, 'expired'); + } + const reply: FleetReply = {}; + switch (command.kind) { + case 'register': { + const description = command.worker; + nonempty(description.id, 'worker id'); + positive(description.groupSlots, 'groupSlots'); + positive(description.cpus, 'cpus'); + positive(description.memoryMb, 'memoryMb'); + if ( + description.environmentId !== state.policy.environmentId || + description.cpus < description.groupSlots * state.policy.groupCpus || + description.memoryMb < description.groupSlots * state.policy.groupMemoryMb + ) { + throw new Error('worker does not satisfy execution and verifier admission'); + } + const existing = state.workers.find((w) => w.description.id === description.id); + if ( + existing && + stableJsonStringify(existing.description) !== stableJsonStringify(description) + ) { + throw new Error('worker incarnation changed description'); + } + if (!existing) + state.workers.push({ description: structuredClone(description), reconciled: false }); + break; + } + case 'heartbeat': { + const worker = requireWorker(state, command.workerId); + if (new Set(command.assignmentIds).size !== command.assignmentIds.length) { + throw new Error('duplicate heartbeat assignment'); + } + for (const id of command.assignmentIds) { + const assignment = requireAssignment(state, command.workerId, id); + if (assignment.status === 'active') + assignment.expiresAt = state.clock + state.policy.leaseMs; + } + // Omitted assignments are allowed to expire; they are not silently renewed. + worker.reconciled = true; + reply.activeAssignmentIds = active(state) + .filter((a) => a.workerId === command.workerId) + .map((a) => a.id); + // Includes a claim whose reply was lost. The worker can adopt it rather than leak a slot. + reply.work = active(state) + .filter((a) => a.workerId === command.workerId) + .map((a) => workFor(state, a)); + break; + } + case 'claim': { + const worker = requireWorker(state, command.workerId); + const running = active(state); + if ( + state.paused || + !worker.reconciled || + state.blocked.length || + running.length >= state.spec.execution.maxConcurrentTaskGroups || + running.filter((a) => a.workerId === command.workerId).length >= + worker.description.groupSlots + ) + break; + const group = state.groups.find( + (g) => + g.readyAt <= state.clock && + !running.some((a) => a.groupId === g.id) && + pendingCells(state, g.cellIds).length > 0, + ); + if (!group) break; + const id = `${state.policy.runId}/assignment-${state.nextAssignment++}`; + const attempts = pendingCells(state, group.cellIds).map( + (cellId, index): FleetAttempt => ({ + id: `${id}/attempt-${index + 1}`, + cellId, + sequence: state.attempts.filter((a) => a.cellId === cellId).length + 1, + assignmentId: id, + report: null, + disposition: 'pending', + }), + ); + const assignment: FleetAssignment = { + id, + groupId: group.id, + workerId: command.workerId, + generation: ++group.generation, + attemptIds: attempts.map((a) => a.id), + status: 'active', + expiresAt: state.clock + state.policy.leaseMs, + }; + state.attempts.push(...attempts); + state.assignments.push(assignment); + reply.assignment = structuredClone(assignment); + reply.work = [workFor(state, assignment)]; + break; + } + case 'report': { + const attempt = state.attempts.find((a) => a.id === command.attemptId); + if (!attempt) throw new Error('unknown attempt'); + const assignment = requireAssignment(state, command.workerId, attempt.assignmentId); + const report = validateFleetReport(command.report); + if (attempt.report) { + if (stableJsonStringify(attempt.report) !== stableJsonStringify(report)) { + throw new Error('conflicting attempt report'); + } + } else { + attempt.report = report; + attempt.disposition = + assignment.status !== 'active' ? 'late' : selectable(report) ? 'selected' : 'retryable'; + if ( + attempt.disposition !== 'late' && + report.environmentFailure && + !state.blocked.includes(report.environmentFailure) + ) + state.blocked.push(report.environmentFailure); + } + reply.disposition = attempt.disposition; + break; + } + case 'finish': { + const assignment = requireAssignment(state, command.workerId, command.assignmentId); + if (assignment.status !== 'active') break; + if (assignment.attemptIds.some((id) => !state.attempts.find((a) => a.id === id)?.report)) { + throw new Error('cannot finish before all cell reports are committed'); + } + settle(state, assignment, 'finished'); + break; + } + case 'repair': + state.blocked = state.blocked.filter((reason) => reason !== command.reason); + break; + case 'tick': + break; + } + return { state, reply }; +} + +export function summarizeFleet(state: FleetState) { + const selected = state.attempts.filter((a) => a.disposition === 'selected'); + const cells = state.groups.flatMap((g) => g.cellIds); + const running = active(state); + const exhausted = cells.filter( + (id) => + !selected.some((a) => a.cellId === id) && + !running.some((a) => + a.attemptIds.some((attemptId) => + state.attempts.some((attempt) => attempt.id === attemptId && attempt.cellId === id), + ), + ) && + state.attempts.filter((a) => a.cellId === id).length >= state.policy.maxAttemptsPerCell, + ); + const reported = state.attempts.filter((a) => a.report); + return { + runId: state.policy.runId, + totalCells: cells.length, + completedCells: selected.length, + incompleteCells: cells.length - selected.length, + exhaustedCells: exhausted.length, + activeGroups: running.length, + blocked: [...state.blocked], + paused: state.paused, + settled: running.length === 0 && cells.length === selected.length + exhausted.length, + observedCostUsd: reported.reduce((sum, a) => sum + (a.report!.result.costUsd ?? 0), 0), + usageComplete: state.attempts.every((a) => a.report?.usage === 'complete'), + results: selected.map((a) => ({ + cellId: a.cellId, + attemptId: a.id, + result: selectedResult(a.report!), + artifacts: structuredClone(a.report!.artifacts), + })), + }; +} + +export function validateFleetReport(value: FleetReport): FleetReport { + const report = structuredClone(value); + report.result = decodeEvalResult(value.result); + member(report.execution, ['completed', 'subject_failed', 'not_started', 'unknown'], 'execution'); + member(report.verification, ['valid', 'invalid', 'not_run'], 'verification'); + member(report.usage, ['complete', 'partial', 'missing'], 'usage'); + member(report.cleanup, ['confirmed', 'unknown'], 'cleanup'); + if (report.environmentFailure !== null) nonempty(report.environmentFailure, 'environmentFailure'); + if (!Array.isArray(report.artifacts)) throw new Error('invalid artifacts'); + for (const artifact of report.artifacts) { + if ( + !/^[a-f0-9]{64}$/.test(artifact.sha256) || + !Number.isSafeInteger(artifact.bytes) || + artifact.bytes < 0 + ) { + throw new Error('invalid artifact reference'); + } + } + if (report.usage === 'missing' && report.result.usage !== null) + throw new Error('missing usage has tokens'); + if (report.usage === 'complete' && report.result.usage === null) + throw new Error('complete usage lacks tokens'); + if (report.usage !== 'complete' && report.result.costUsd !== null) + throw new Error('incomplete usage has settled cost'); + if ( + report.verification === 'valid' && + (report.result.score === null || !['completed', 'subject_failed'].includes(report.execution)) + ) + throw new Error('invalid verification evidence'); + return report; +} + +function selectedResult(report: FleetReport): EvalResult { + // Keep raw execution evidence in the attempt; project settled status only in selected results. + return { + ...report.result, + status: + report.execution === 'subject_failed' || report.result.status === 'subject_failed' + ? 'subject_failed' + : 'completed', + }; +} + +function selectable(report: FleetReport) { + return ( + ['completed', 'subject_failed'].includes(report.execution) && + report.verification === 'valid' && + report.cleanup === 'confirmed' && + report.environmentFailure === null + ); +} + +function active(state: FleetState) { + return state.assignments.filter((a) => a.status === 'active'); +} +function workFor(state: FleetState, assignment: FleetAssignment): FleetWork { + const cells = expandExperiment(state.spec); + return { + assignment: structuredClone(assignment), + cells: assignment.attemptIds.flatMap((id) => { + const attempt = state.attempts.find((a) => a.id === id)!; + return attempt.report + ? [] + : [{ attemptId: id, cell: cells.find((c) => c.id === attempt.cellId)! }]; + }), + }; +} +function pendingCells(state: FleetState, ids: string[]) { + return ids.filter( + (id) => + !state.attempts.some((a) => a.cellId === id && a.disposition === 'selected') && + state.attempts.filter((a) => a.cellId === id).length < state.policy.maxAttemptsPerCell, + ); +} +function requireWorker(state: FleetState, id: string) { + const worker = state.workers.find((w) => w.description.id === id); + if (!worker) throw new Error('unknown worker incarnation'); + return worker; +} +function requireAssignment(state: FleetState, workerId: string, id: string) { + const assignment = state.assignments.find((a) => a.id === id); + if (!assignment || assignment.workerId !== workerId) + throw new Error('unknown or foreign assignment'); + return assignment; +} +function settle(state: FleetState, assignment: FleetAssignment, status: 'finished' | 'expired') { + assignment.status = status; + state.groups.find((g) => g.id === assignment.groupId)!.readyAt = + state.clock + state.policy.retryBackoffMs; +} +function positive(value: number, name: string) { + if (!Number.isSafeInteger(value) || value < 1) throw new Error(`invalid ${name}`); +} +function nonempty(value: string, name: string) { + if (typeof value !== 'string' || !value.trim()) throw new Error(`invalid ${name}`); +} +function member(value: string, values: string[], name: string) { + if (!values.includes(value)) throw new Error(`invalid ${name}`); +} diff --git a/packages/eval/src/index.ts b/packages/eval/src/index.ts index a1706e4e8e..0f59e0697c 100644 --- a/packages/eval/src/index.ts +++ b/packages/eval/src/index.ts @@ -22,6 +22,9 @@ export * from './cli.js'; export * from './experiment-directory.js'; export * from './experiment.js'; export * from './external-subject.js'; +export * from './fleet.js'; +export * from './fleet-store.js'; +export * from './fleet-worker.js'; export * from './harness-executor.js'; export * from './maka-subject.js'; export * from './result.js';