P5: capacity-aware node selection (control side)

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
 

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):

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

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"]