diff --git a/meridian_control/README.md b/meridian_control/README.md index 2fb28e4..38d3926 100644 --- a/meridian_control/README.md +++ b/meridian_control/README.md @@ -41,6 +41,7 @@ meridian-control mint-token --auto-approve | `POST /control/v1/nodes/{id}/certificate` | Rotate the node certificate (mTLS-gated) | | `POST /admin/tokens` · `/admin/claims/{id}/approve` · `/admin/nodes/{id}/desired` · `/admin/nodes/{id}/stop-authorize` · `/admin/nodes/{id}/revoke` · `/admin/restore` | Operator actions | | `GET /admin/projection` | Serving projection (gateway consumes read-only) | +| `POST /admin/place` | Capacity-aware placement: pick a node/device with enough allocatable VRAM | ## Restore safety diff --git a/meridian_control/routes.py b/meridian_control/routes.py index 4f8732d..d79cceb 100644 --- a/meridian_control/routes.py +++ b/meridian_control/routes.py @@ -130,6 +130,12 @@ def projection(request: Request): return {"endpoints": _svc(request).serving_projection()} +@router.post("/admin/place") +def place(request: Request, body: dict = Body(...)): + placement = _svc(request).select_placement(int(body.get("required_vram_bytes", 0))) + return {"placement": placement} + + def register_error_handler(app) -> None: @app.exception_handler(ControlServiceError) async def _handle(request: Request, exc: ControlServiceError): diff --git a/meridian_control/service.py b/meridian_control/service.py index 43dca4c..021b113 100644 --- a/meridian_control/service.py +++ b/meridian_control/service.py @@ -364,6 +364,29 @@ def post_observation(self, node_id: str, observation: dict) -> dict: s.commit() return {"received": True} + # --- placement (Phase 5, control side) ----------------------------- + def select_placement(self, required_vram_bytes: int) -> Optional[dict]: + """Pick a routable node + device with enough reported allocatable VRAM + (DESIGN.md 24 P3). Consumes the capacity nodes report in observations; + prefers the device with the most headroom (spread). None if nothing fits.""" + best: Optional[tuple[int, str, str]] = None # (allocatable, node_id, device_id) + with self._sf() as s: + for node in s.scalars(select(Node)): + lease_valid = node.lease_expires_at is not None and self._now() < _aware(node.lease_expires_at) + if node.revoked or not lease_valid: + continue + obs = s.get(ObservationRow, node.node_id) + if obs is None: + continue + allocatable = obs.observation.get("capacity", {}).get("allocatable_vram_bytes", {}) + for device_id, free in allocatable.items(): + free_i = int(free) + if free_i >= required_vram_bytes and (best is None or free_i > best[0]): + best = (free_i, node.node_id, device_id) + if best is None: + return None + return {"node_id": best[1], "device_id": best[2], "allocatable_vram_bytes": best[0]} + # --- read models (operator / gateway projection) ------------------- def serving_projection(self) -> list[dict]: """Read-only projection the gateway consumes (DESIGN.md 17). Routable diff --git a/tests/control/test_service.py b/tests/control/test_service.py index 48776ff..a979f37 100644 --- a/tests/control/test_service.py +++ b/tests/control/test_service.py @@ -166,6 +166,37 @@ def test_rotate_certificate_reissues_for_same_key(tmp_path, clock, node_key): svc._ca.verify_cert(rotated["certificate"]) +def _ready_node_with_capacity(svc, node_key, allocatable): + node_id = _enroll_auto(svc, node_key)["node_id"] + svc.establish_session(node_id, {"agent_session_id": "s-" + node_id}) + svc.heartbeat(node_id, {"agent_session_id": "s-" + node_id, "fencing_epoch": 1, "sequence": 1}) + svc.post_observation(node_id, {"sequence": 1, "engines": [], + "capacity": {"allocatable_vram_bytes": allocatable}}) + return node_id + + +def test_select_placement_prefers_most_headroom(tmp_path, clock, node_key): + g = 1024**3 + svc = make_service(tmp_path, clock) + _ready_node_with_capacity(svc, node_key, {"GPU-0": 8 * g}) + big = _ready_node_with_capacity(svc, node_key, {"GPU-0": 40 * g, "GPU-1": 20 * g}) + + pick = svc.select_placement(10 * g) + assert pick is not None + assert pick["node_id"] == big and pick["device_id"] == "GPU-0" # most headroom that fits + assert pick["allocatable_vram_bytes"] == 40 * g + + assert svc.select_placement(100 * g) is None # nothing fits + + +def test_select_placement_skips_expired_lease(tmp_path, clock, node_key): + g = 1024**3 + svc = make_service(tmp_path, clock) + _ready_node_with_capacity(svc, node_key, {"GPU-0": 40 * g}) + clock.advance(31) # lease (ttl 30) expired -> node not eligible + assert svc.select_placement(10 * g) is None + + def test_rotate_certificate_rejects_revoked_node(tmp_path, clock, node_key): svc = make_service(tmp_path, clock) node_id = _enroll_auto(svc, node_key)["node_id"]