feat(prism): enable unattested Lium dispatch for PRISM jobs (T14)

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
+```

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

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

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 == []