diff --git a/docs/lium-dispatch.md b/docs/lium-dispatch.md new file mode 100644 index 000000000..79587a67d --- /dev/null +++ b/docs/lium-dispatch.md @@ -0,0 +1,76 @@ +# PRISM → Lium dispatch (T14) + +Master-owned path that admits Prism GPU training/eval work onto **our** Lium +machines. This is **compute only** — not TEE, not constation elevation. + +## Trigger (ops) + +```bash +# Master (preferred for "our machines") +export BASE_LIUM_TRAINING__ENABLED=true +export BASE_LIUM_TRAINING__API_KEY_FILE=/run/secrets/lium_api_key # never log +export BASE_LIUM_TRAINING__SSH_PUBLIC_KEY_FILE=/run/secrets/lium_ssh.pub +# Optional spend / concurrency guards (defaults in LiumTrainingSettings): +# BASE_LIUM_TRAINING__MAX_PRICE_PER_HOUR=1.50 +# BASE_LIUM_TRAINING__CONCURRENCY_CAP=3 +# BASE_LIUM_TRAINING__DAILY_SPEND_CEILING_USD=50 + +# Optional worker label (default stays base_gpu) +export PRISM_EXECUTION_BACKEND=lium +``` + +Fail-closed: + +- `lium_training.enabled=false` (default) → `try_build_lium_capacity_scheduler` returns `None`; no rentals. +- `enabled=true` without `api_key` / `api_key_file` → hard `build_*` raises `ValueError`; soft `try_build_*` returns `None` so master still boots. + +## Code path + +1. `cli_app/main.py` — `try_build_lium_capacity_scheduler(settings)` when master starts. +2. `MasterOrchestrationDriver.bridge_pending_work` — Prism GPU units → `lium_scheduler.enqueue(submission_id, job_id)`. +3. `run_once` — `lium_scheduler.tick()` admits FIFO when 1-GPU Blackwell inventory is free. +4. `LiumCapacityScheduler` — provisions via training-locked `LiumClient.for_prism_training` (`lium_capacity.py` + `lium_training_wiring.py`). +5. Worker construct — `PRISM_EXECUTION_BACKEND=lium` / `execution_backend=lium` is always allowed at `PrismWorker` construction (**no** constation bundle). Bundle remains optional API-compat only. + +Empty inventory → lease stays `queued` with `reason=capacity_wait` (never a terminal fail for capacity). + +## Config keys (`LiumTrainingSettings`) + +| Field | Env | Default | Notes | +|-------|-----|---------|-------| +| `enabled` | `BASE_LIUM_TRAINING__ENABLED` | `false` | Master plane off until ops flips | +| `api_key` | `BASE_LIUM_TRAINING__API_KEY` | `null` | Prefer file; never log | +| `api_key_file` | `BASE_LIUM_TRAINING__API_KEY_FILE` | `null` | File contents read lazily | +| `ssh_public_key_file` | `BASE_LIUM_TRAINING__SSH_PUBLIC_KEY_FILE` | `null` | Injected into pods | +| `max_price_per_hour` | `BASE_LIUM_TRAINING__MAX_PRICE_PER_HOUR` | `1.50` | Per-GPU hourly ceiling | +| `max_lifetime_hours` | `BASE_LIUM_TRAINING__MAX_LIFETIME_HOURS` | `4.0` | Pod lifetime guard | +| `concurrency_cap` | `BASE_LIUM_TRAINING__CONCURRENCY_CAP` | `3` | Max concurrent training pods | +| `daily_spend_ceiling_usd` | `BASE_LIUM_TRAINING__DAILY_SPEND_CEILING_USD` | `50.0` | Blocks **new** admissions only | +| `pod_name_prefix` | `BASE_LIUM_TRAINING__POD_NAME_PREFIX` | `prism-train-` | Recover matches this prefix | + +Prism worker label: `execution_backend` / `PRISM_EXECUTION_BACKEND` (`base_gpu` default; `lium` allowed bare). + +## What this is not + +- **Not TEE.** No attestation claim, no constation_ok elevation from selecting `lium`. +- **Not live billable by default.** Tests mock the client; live rentals require explicit ops env + real key. +- Constation modules stay in-tree for the separate ingestion elevation path (todos 19–22). + +## T15 handoff (live 1M e2e) + +1. Pack `examples/tiny-1m` (or equivalent tiny recipe) as the Prism work unit. +2. Set `BASE_LIVE_PROVIDER_TESTS=1` **and** a real key via `BASE_LIUM_TRAINING__API_KEY_FILE` only in a controlled env. +3. Run master with `BASE_LIUM_TRAINING__ENABLED=true`; confirm bridge enqueue + tick provisions a pod; tear down pods after. +4. Do **not** claim TEE / constation from a successful rental. +5. Capture spend + pod IDs (not API key material) in evidence. + +## Tests + +```bash +UV_CACHE_DIR=/var/tmp/uv-cache uv run pytest \ + packages/challenges/prism/tests/test_execution_backend_constation_gate.py \ + tests/unit/test_lium_dispatch_unattested.py \ + tests/unit/test_orchestration_lium.py \ + tests/unit/test_lium_training_wiring.py \ + -q +``` diff --git a/packages/challenges/prism/config.example.yaml b/packages/challenges/prism/config.example.yaml index 6268c854c..6734ac58f 100644 --- a/packages/challenges/prism/config.example.yaml +++ b/packages/challenges/prism/config.example.yaml @@ -32,6 +32,8 @@ fineweb_sample_count: 128 # base_eval_train_data_dir: /data/fineweb-edu/train # base_eval_val_data_dir: /data/fineweb-edu/val # Phase-2 sample-100BT root is the sibling .../fineweb-edu/sample-100BT unless overridden. +# Compute backend for PrismWorker. Default base_gpu. lium is allowed bare (T14 master-owned / unattested compute; not TEE). +# Override: PRISM_EXECUTION_BACKEND=lium execution_backend: base_gpu prism_role: challenge public_submissions_enabled: true diff --git a/packages/challenges/prism/tests/test_execution_backend_constation_gate.py b/packages/challenges/prism/tests/test_execution_backend_constation_gate.py index 4b2cc1072..7d6cba7e3 100644 --- a/packages/challenges/prism/tests/test_execution_backend_constation_gate.py +++ b/packages/challenges/prism/tests/test_execution_backend_constation_gate.py @@ -1,16 +1,25 @@ -"""Lium execution backend is gated on a full constation bundle (todo 19). +"""Execution-backend gate: bare ``lium`` is allowed (T14 unattested / master-owned). -Renamed from the misleading test_lium_client.py (which tested no client). +Container backends stay always-on. ``constation_bundle`` remains an API-compat +kwarg but does **not** gate Lium selection. ``constation_ok`` score elevation is +a separate ingestion path; constation modules are not deleted. """ from __future__ import annotations +from types import SimpleNamespace +from unittest.mock import AsyncMock + import pytest +from prism_challenge.config import PrismSettings from prism_challenge.constation import ConstationBundle +from prism_challenge.evaluator.interface import PrismContext from prism_challenge.queue import ( + GATED_EXECUTION_BACKENDS, LIUM_EXECUTION_BACKEND, SUPPORTED_EXECUTION_BACKENDS, + PrismWorker, is_execution_backend_supported, require_execution_backend, ) @@ -42,21 +51,39 @@ def test_base_gpu_always_supported_without_bundle() -> None: require_execution_backend("base_gpu") # no raise -def test_lium_without_bundle_rejected() -> None: - assert is_execution_backend_supported(LIUM_EXECUTION_BACKEND) is False - assert is_execution_backend_supported(LIUM_EXECUTION_BACKEND, constation_bundle=None) is False - with pytest.raises(ValueError, match="constation bundle required for lium"): - require_execution_backend(LIUM_EXECUTION_BACKEND) - with pytest.raises(ValueError, match="constation bundle required for lium"): - require_execution_backend("lium", constation_bundle=None) +def test_lium_without_bundle_accepted_unattested() -> None: + """T14: bare ``lium`` is a supported compute backend (no constation required).""" + assert LIUM_EXECUTION_BACKEND in SUPPORTED_EXECUTION_BACKENDS + assert LIUM_EXECUTION_BACKEND not in GATED_EXECUTION_BACKENDS + assert is_execution_backend_supported(LIUM_EXECUTION_BACKEND) is True + assert ( + is_execution_backend_supported(LIUM_EXECUTION_BACKEND, constation_bundle=None) + is True + ) + require_execution_backend(LIUM_EXECUTION_BACKEND) # no raise + require_execution_backend("lium", constation_bundle=None) # no raise -def test_lium_with_full_bundle_accepted() -> None: +def test_lium_with_full_bundle_still_accepted() -> None: + """Bundle remains optional API-compat; still accepted when supplied.""" bundle = _bundle() assert is_execution_backend_supported("lium", constation_bundle=bundle) is True require_execution_backend("lium", constation_bundle=bundle) # no raise +def test_prism_worker_constructs_with_bare_lium() -> None: + """PrismWorker(execution_backend=lium) constructs without constation_bundle.""" + repo = SimpleNamespace(claim_next=AsyncMock(return_value=None)) + worker = PrismWorker( + repository=repo, # type: ignore[arg-type] + ctx=PrismContext(), + execution_backend="lium", + settings=PrismSettings(docker_enabled=False, plagiarism_enabled=False), + ) + assert worker.execution_backend == "lium" + assert worker._constation_bundle is None # noqa: SLF001 + + def test_remote_provider_and_local_cpu_still_rejected() -> None: assert "remote_provider" not in SUPPORTED_EXECUTION_BACKENDS assert "local_cpu" not in SUPPORTED_EXECUTION_BACKENDS diff --git a/tests/unit/test_lium_dispatch_unattested.py b/tests/unit/test_lium_dispatch_unattested.py new file mode 100644 index 000000000..9f8094df9 --- /dev/null +++ b/tests/unit/test_lium_dispatch_unattested.py @@ -0,0 +1,196 @@ +"""T14 — unattested master-owned Lium dispatch path (mocked; no billable rent). + +Proves the code path that submits Prism GPU work onto Lium capacity without +constation elevation or live provider calls. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + +import pytest +from sqlalchemy import select + +from base.compute.lium_capacity import ( + InMemoryLeaseStore, + LeaseState, + LiumCapacityScheduler, +) +from base.compute.lium_training_wiring import ( + build_lium_capacity_scheduler, + try_build_lium_capacity_scheduler, +) +from base.compute.provider import Instance, InstanceSpec, Offer +from base.config.settings import LiumTrainingSettings, Settings +from base.db import Base, create_engine, create_session_factory +from base.db.models import WorkAssignment +from base.master.assignment import AssignmentService +from base.master.orchestration import ChallengePendingWork, MasterOrchestrationDriver +from base.master.validator_coordination import ValidatorCoordinationService + +NOW = datetime(2026, 6, 27, 12, 0, 0, tzinfo=UTC) + +_BLACKWELL = "NVIDIA RTX PRO 6000 Blackwell Server Edition" + + +@dataclass +class FakeLiumClient: + """In-memory Lium surface — never hits the network.""" + + offers: list[Offer] = field(default_factory=list) + provisioned: list[InstanceSpec] = field(default_factory=list) + terminated: list[str] = field(default_factory=list) + pods: list[dict[str, Any]] = field(default_factory=list) + _seq: int = 0 + + async def list_offers( + self, *, max_price_per_hour: float | None = None + ) -> list[Offer]: + out: list[Offer] = [] + for offer in self.offers: + if ( + max_price_per_hour is not None + and offer.price_per_hour > max_price_per_hour + ): + continue + out.append(offer) + return out + + async def list_pods(self) -> list[dict[str, Any]]: + return list(self.pods) + + async def provision( + self, spec: InstanceSpec, *, offer: Offer | None = None + ) -> Instance: + self.provisioned.append(spec) + self._seq += 1 + pod_id = f"pod-fake-{self._seq}" + self.pods.append( + { + "id": pod_id, + "pod_name": spec.name, + "name": spec.name, + "status": "RUNNING", + "provider": "lium", + } + ) + if offer is not None: + self.offers = [o for o in self.offers if o.id != offer.id] + elif self.offers: + self.offers = self.offers[1:] + return Instance(id=pod_id, status="RUNNING", provider="lium") + + async def terminate(self, instance_id: str) -> None: + self.terminated.append(instance_id) + self.pods = [p for p in self.pods if str(p.get("id")) != str(instance_id)] + + +def _blackwell_offer(*, offer_id: str = "offer-bw-1", price: float = 1.0) -> Offer: + return Offer( + id=offer_id, + gpu_type=_BLACKWELL, + gpu_count=1, + price_per_hour=price, + provider="lium", + raw={"machine_name": "rtx-pro-6000-blackwell-server"}, + ) + + +def test_try_build_scheduler_none_when_disabled() -> None: + """Given lium_training.enabled=False, When try_build, Then None (default off).""" + settings = Settings(lium_training=LiumTrainingSettings(enabled=False)) + assert try_build_lium_capacity_scheduler(settings) is None + + +def test_try_build_scheduler_fail_closed_missing_key_returns_none() -> None: + """Given enabled without key, When try_build, Then None (master still boots).""" + settings = Settings(lium_training=LiumTrainingSettings(enabled=True)) + assert try_build_lium_capacity_scheduler(settings) is None + + +def test_build_scheduler_fail_closed_raises_without_key() -> None: + """Given enabled without key, When hard build, Then ValueError names key fields.""" + settings = Settings(lium_training=LiumTrainingSettings(enabled=True)) + with pytest.raises(ValueError, match="api_key|api_key_file|lium_training"): + build_lium_capacity_scheduler(settings) + + +async def test_dispatch_enqueue_and_tick_provisions_mocked_pod() -> None: + """Given Prism pending work + fake Lium inventory, When bridge+tick, Then pod provisioned. + + End-to-end master-owned dispatch without constation and without live API. + """ + client = FakeLiumClient(offers=[_blackwell_offer()]) + store = InMemoryLeaseStore() + scheduler = LiumCapacityScheduler( + lambda: client, + store=store, + ssh_public_keys=("ssh-ed25519 AAAA test-dispatch",), + ) + + engine = create_engine("sqlite+aiosqlite:///:memory:") + try: + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + factory = create_session_factory(engine) + service = AssignmentService(factory, now_fn=lambda: NOW) + validators = ValidatorCoordinationService(factory, now_fn=lambda: NOW) + + @dataclass + class _Src: + async def fetch_pending_work(self) -> list[ChallengePendingWork]: + return [ + ChallengePendingWork( + challenge_slug="prism", + submission_id="sub-t14-1", + submission_ref="miner-hk", + checkpoint_ref="hf://ckpt/1", + job_id="job-t14-1", + ) + ] + + driver = MasterOrchestrationDriver( + assignment_service=service, + validator_service=validators, + work_source=_Src(), + lium_scheduler=scheduler, + ) + + # bridge enqueues; run_once ticks Lium and provisions + result = await driver.run_once() + assert result.bridged["prism"] == ["sub-t14-1"] + + assert len(client.provisioned) == 1 + assert client.provisioned[0].name == "prism-train-sub-t14-1" + active = store.get("sub-t14-1") + assert active is not None + assert active.state is LeaseState.ACTIVE + assert active.pod_id is not None + + async with factory() as session: + rows = list((await session.execute(select(WorkAssignment))).scalars().all()) + assert len(rows) == 1 + assert rows[0].work_unit_id == "sub-t14-1" + assert rows[0].required_capability == "gpu" + finally: + await engine.dispose() + + +async def test_dispatch_queues_when_inventory_empty_no_terminal_fail() -> None: + """Given empty Lium inventory, When tick, Then lease stays queued (capacity_wait).""" + client = FakeLiumClient(offers=[]) + store = InMemoryLeaseStore() + scheduler = LiumCapacityScheduler( + lambda: client, + store=store, + ssh_public_keys=("ssh-ed25519 AAAA test-dispatch",), + ) + scheduler.enqueue(submission_id="sub-wait", job_id="job-wait") + await scheduler.tick() + lease = store.get("sub-wait") + assert lease is not None + assert lease.state is LeaseState.QUEUED + assert lease.reason == "capacity_wait" + assert client.provisioned == []