Describe v0.10.1

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -20,6 +20,58 @@ is only released once its **exit criterion** is met, not when its code lands.
 Until v1.0.0 the minor version is the breaking one: configuration formats and Go
 APIs change between minors without a deprecation period.
 
+## v0.10.1 — 2026-09-29
+
+### `llmgw ctl fabric-test`: two nodes, rail by rail
+
+`llmgw doctor` checks a node on its own, and every check can pass while two
+nodes still cannot reach each other: a cable into the wrong switch, a
+partition key, a RoCE network with no route. `llmgw ctl fabric-test A B` runs
+`ib_write_bw` from each of A's adapters on a fabric to the same adapter on B,
+one rail at a time, through the agents — no SSH between nodes, and perftest is
+installed the first time — and grades each against its link's rate: ok at 80%,
+slow at 50%, fail below or when nothing got through.
+
+Agents now say which requests they answer beyond the usual (`features`), so
+one too old for the test is refused at once instead of waited for.
+
+Written against perftest's real output and tested with it, but not yet run
+over real InfiniBand or RoCE: this cluster has neither.
+
+### The mirror on its own machine, in two commands
+
+`llmgw mirror init --control-plane <address>` sets a machine up as the model
+mirror: the user, `/srv/models`, and a service serving files to every node on
+:8081 and the management API on :8082. `server init --mirror
+http://<mirror>:8082` uses it instead of running one beside the control plane.
+It took a unit written by hand and an edit to `server.yaml` before.
+
+The management API can fetch and delete models; `mirror serve --api-allow`
+makes it answer only the addresses given and the machine itself, and `mirror
+init` gives it the control plane's. With `--no-node`, the starter model is
+fetched and catalogued anyway, unpinned, and starts on the first node added.
+
+Tried on three empty VMs: mirror, control plane and a node, and a request
+answered by the starter model on the node.
+
+### `server init` asks, and `node add` takes several nodes
+
+- **`sudo llmgw server init`** in a terminal, without `--admin`, asks for the
+  administrator's email, the other names nodes will reach the machine by
+  (showing the addresses it already found), SQLite or PostgreSQL, and whether
+  to run the mirror, a node and the starter model here. Then its steps fill
+  in as a list, downloads updating in place, and the administrator's password
+  stays in a box until the end — and is printed again after, with the whole
+  run, so it is in the terminal's history. The last screen says how to add a
+  node and to back up the authority's key. `--plain`, or no terminal, prints
+  lines as before.
+- **`llmgw node add a b c`** adds each in turn and ends with which were added.
+  A fabric named for one node's adapters is recognised on the next by what the
+  adapters report, and not asked again.
+
+Both tried on empty VMs: init from the form to a starter model answering, and
+two nodes added to it in one command in 40 seconds.
+
 ## v0.10.0 — 2026-09-29
 
 Independent InfiniBand fabrics and NVLink domains, told apart, drawn and

api/gen/llmgw/v1/agent.pb.go

@@ -103,6 +103,7 @@ type AgentMessage struct {
 	//	*AgentMessage_Shard
 	//	*AgentMessage_Logs
 	//	*AgentMessage_FabricScan
+	//	*AgentMessage_FabricTest
 	Msg           isAgentMessage_Msg `protobuf_oneof:"msg"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
@@ -190,6 +191,15 @@ func (x *AgentMessage) GetFabricScan() *FabricScanReply {
 	return nil
 }
 
+func (x *AgentMessage) GetFabricTest() *FabricTestReply {
+	if x != nil {
+		if x, ok := x.Msg.(*AgentMessage_FabricTest); ok {
+			return x.FabricTest
+		}
+	}
+	return nil
+}
+
 type isAgentMessage_Msg interface {
 	isAgentMessage_Msg()
 }
@@ -214,6 +224,10 @@ type AgentMessage_FabricScan struct {
 	FabricScan *FabricScanReply `protobuf:"bytes,5,opt,name=fabric_scan,json=fabricScan,proto3,oneof"`
 }
 
+type AgentMessage_FabricTest struct {
+	FabricTest *FabricTestReply `protobuf:"bytes,6,opt,name=fabric_test,json=fabricTest,proto3,oneof"`
+}
+
 func (*AgentMessage_Register) isAgentMessage_Msg() {}
 
 func (*AgentMessage_Heartbeat) isAgentMessage_Msg() {}
@@ -224,6 +238,8 @@ func (*AgentMessage_Logs) isAgentMessage_Msg() {}
 
 func (*AgentMessage_FabricScan) isAgentMessage_Msg() {}
 
+func (*AgentMessage_FabricTest) isAgentMessage_Msg() {}
+
 // ControlMessage is anything the control plane tells a node.
 type ControlMessage struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
@@ -237,6 +253,7 @@ type ControlMessage struct {
 	//	*ControlMessage_Provision
 	//	*ControlMessage_Logs
 	//	*ControlMessage_FabricScan
+	//	*ControlMessage_FabricTest
 	Msg           isControlMessage_Msg `protobuf_oneof:"msg"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
@@ -351,6 +368,15 @@ func (x *ControlMessage) GetFabricScan() *FabricScanRequest {
 	return nil
 }
 
+func (x *ControlMessage) GetFabricTest() *FabricTestRequest {
+	if x != nil {
+		if x, ok := x.Msg.(*ControlMessage_FabricTest); ok {
+			return x.FabricTest
+		}
+	}
+	return nil
+}
+
 type isControlMessage_Msg interface {
 	isControlMessage_Msg()
 }
@@ -387,6 +413,10 @@ type ControlMessage_FabricScan struct {
 	FabricScan *FabricScanRequest `protobuf:"bytes,8,opt,name=fabric_scan,json=fabricScan,proto3,oneof"`
 }
 
+type ControlMessage_FabricTest struct {
+	FabricTest *FabricTestRequest `protobuf:"bytes,9,opt,name=fabric_test,json=fabricTest,proto3,oneof"`
+}
+
 func (*ControlMessage_Start) isControlMessage_Msg() {}
 
 func (*ControlMessage_Stop) isControlMessage_Msg() {}
@@ -403,6 +433,8 @@ func (*ControlMessage_Logs) isControlMessage_Msg() {}
 
 func (*ControlMessage_FabricScan) isControlMessage_Msg() {}
 
+func (*ControlMessage_FabricTest) isControlMessage_Msg() {}
+
 // Register is the first message on every stream. It carries the node's full
 // hardware inventory, so the control plane needs no separate discovery call and
 // a reconnecting agent automatically refreshes what the scheduler knows.
@@ -478,8 +510,13 @@ type Register struct {
 	// vLLM, it answered "unsupported", and every node of a cluster half way
 	// through an upgrade showed as failing.
 	InstallableEngines []string `protobuf:"bytes,16,rep,name=installable_engines,json=installableEngines,proto3" json:"installable_engines,omitempty"`
-	unknownFields      protoimpl.UnknownFields
-	sizeCache          protoimpl.SizeCache
+	// features are the requests this agent answers beyond the ones every agent
+	// does: "fabric-test". An agent that does not know a request ignores it, so
+	// without this the control plane would wait minutes for an answer that is
+	// never coming.
+	Features      []string `protobuf:"bytes,17,rep,name=features,proto3" json:"features,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
 }
 
 func (x *Register) Reset() {
@@ -624,6 +661,13 @@ func (x *Register) GetInstallableEngines() []string {
 	return nil
 }
 
+func (x *Register) GetFeatures() []string {
+	if x != nil {
+		return x.Features
+	}
+	return nil
+}
+
 // Host describes the operating system and processor of a node.
 type Host struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
@@ -3223,19 +3267,203 @@ func (x *FabricScanReply) GetTookMs() int64 {
 	return 0
 }
 
+// FabricTestRequest asks a node for one side of an RDMA bandwidth test
+// between two nodes: ib_write_bw, from perftest, over one adapter. The
+// server side is asked first and answers when the test is over; the client
+// dials it. What it proves is what a replica spanning nodes needs and nothing
+// on one node can show: that these two adapters reach each other through the
+// fabric, and how fast.
+type FabricTestRequest struct {
+	state     protoimpl.MessageState `protogen:"open.v1"`
+	RequestId string                 `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
+	// role is "server" or "client".
+	Role string `protobuf:"bytes,2,opt,name=role,proto3" json:"role,omitempty"`
+	Hca  string `protobuf:"bytes,3,opt,name=hca,proto3" json:"hca,omitempty"`
+	Port int32  `protobuf:"varint,4,opt,name=port,proto3" json:"port,omitempty"`
+	// peer is the server's management address, for the client. perftest
+	// exchanges its connection details over TCP before any RDMA.
+	Peer string `protobuf:"bytes,5,opt,name=peer,proto3" json:"peer,omitempty"`
+	// tcp_port is that exchange's port, the same on both sides.
+	TcpPort       int32 `protobuf:"varint,6,opt,name=tcp_port,json=tcpPort,proto3" json:"tcp_port,omitempty"`
+	Seconds       int32 `protobuf:"varint,7,opt,name=seconds,proto3" json:"seconds,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *FabricTestRequest) Reset() {
+	*x = FabricTestRequest{}
+	mi := &file_llmgw_v1_agent_proto_msgTypes[31]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *FabricTestRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*FabricTestRequest) ProtoMessage() {}
+
+func (x *FabricTestRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_llmgw_v1_agent_proto_msgTypes[31]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use FabricTestRequest.ProtoReflect.Descriptor instead.
+func (*FabricTestRequest) Descriptor() ([]byte, []int) {
+	return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{31}
+}
+
+func (x *FabricTestRequest) GetRequestId() string {
+	if x != nil {
+		return x.RequestId
+	}
+	return ""
+}
+
+func (x *FabricTestRequest) GetRole() string {
+	if x != nil {
+		return x.Role
+	}
+	return ""
+}
+
+func (x *FabricTestRequest) GetHca() string {
+	if x != nil {
+		return x.Hca
+	}
+	return ""
+}
+
+func (x *FabricTestRequest) GetPort() int32 {
+	if x != nil {
+		return x.Port
+	}
+	return 0
+}
+
+func (x *FabricTestRequest) GetPeer() string {
+	if x != nil {
+		return x.Peer
+	}
+	return ""
+}
+
+func (x *FabricTestRequest) GetTcpPort() int32 {
+	if x != nil {
+		return x.TcpPort
+	}
+	return 0
+}
+
+func (x *FabricTestRequest) GetSeconds() int32 {
+	if x != nil {
+		return x.Seconds
+	}
+	return 0
+}
+
+// FabricTestReply is what one side measured.
+type FabricTestReply struct {
+	state     protoimpl.MessageState `protogen:"open.v1"`
+	RequestId string                 `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
+	// gbps is the average bandwidth, in gigabits a second.
+	Gbps float64 `protobuf:"fixed64,2,opt,name=gbps,proto3" json:"gbps,omitempty"`
+	// output is the end of the tool's output, for when the number is not
+	// enough.
+	Output        string `protobuf:"bytes,3,opt,name=output,proto3" json:"output,omitempty"`
+	Error         string `protobuf:"bytes,4,opt,name=error,proto3" json:"error,omitempty"`
+	TookMs        int64  `protobuf:"varint,5,opt,name=took_ms,json=tookMs,proto3" json:"took_ms,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *FabricTestReply) Reset() {
+	*x = FabricTestReply{}
+	mi := &file_llmgw_v1_agent_proto_msgTypes[32]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *FabricTestReply) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*FabricTestReply) ProtoMessage() {}
+
+func (x *FabricTestReply) ProtoReflect() protoreflect.Message {
+	mi := &file_llmgw_v1_agent_proto_msgTypes[32]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use FabricTestReply.ProtoReflect.Descriptor instead.
+func (*FabricTestReply) Descriptor() ([]byte, []int) {
+	return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{32}
+}
+
+func (x *FabricTestReply) GetRequestId() string {
+	if x != nil {
+		return x.RequestId
+	}
+	return ""
+}
+
+func (x *FabricTestReply) GetGbps() float64 {
+	if x != nil {
+		return x.Gbps
+	}
+	return 0
+}
+
+func (x *FabricTestReply) GetOutput() string {
+	if x != nil {
+		return x.Output
+	}
+	return ""
+}
+
+func (x *FabricTestReply) GetError() string {
+	if x != nil {
+		return x.Error
+	}
+	return ""
+}
+
+func (x *FabricTestReply) GetTookMs() int64 {
+	if x != nil {
+		return x.TookMs
+	}
+	return 0
+}
+
 var File_llmgw_v1_agent_proto protoreflect.FileDescriptor
 
 const file_llmgw_v1_agent_proto_rawDesc = "" +
 	"\n" +
-	"\x14llmgw/v1/agent.proto\x12\bllmgw.v1\"\x94\x02\n" +
+	"\x14llmgw/v1/agent.proto\x12\bllmgw.v1\"\xd2\x02\n" +
 	"\fAgentMessage\x120\n" +
 	"\bregister\x18\x01 \x01(\v2\x12.llmgw.v1.RegisterH\x00R\bregister\x123\n" +
 	"\theartbeat\x18\x02 \x01(\v2\x13.llmgw.v1.HeartbeatH\x00R\theartbeat\x12-\n" +
 	"\x05shard\x18\x03 \x01(\v2\x15.llmgw.v1.ShardStatusH\x00R\x05shard\x12)\n" +
 	"\x04logs\x18\x04 \x01(\v2\x13.llmgw.v1.LogsReplyH\x00R\x04logs\x12<\n" +
 	"\vfabric_scan\x18\x05 \x01(\v2\x19.llmgw.v1.FabricScanReplyH\x00R\n" +
-	"fabricScanB\x05\n" +
-	"\x03msg\"\xa5\x03\n" +
+	"fabricScan\x12<\n" +
+	"\vfabric_test\x18\x06 \x01(\v2\x19.llmgw.v1.FabricTestReplyH\x00R\n" +
+	"fabricTestB\x05\n" +
+	"\x03msg\"\xe5\x03\n" +
 	"\x0eControlMessage\x12,\n" +
 	"\x05start\x18\x01 \x01(\v2\x14.llmgw.v1.StartShardH\x00R\x05start\x12)\n" +
 	"\x04stop\x18\x02 \x01(\v2\x13.llmgw.v1.StopShardH\x00R\x04stop\x12,\n" +
@@ -3245,8 +3473,10 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
 	"\tprovision\x18\x06 \x01(\v2\x13.llmgw.v1.ProvisionH\x00R\tprovision\x12+\n" +
 	"\x04logs\x18\a \x01(\v2\x15.llmgw.v1.LogsRequestH\x00R\x04logs\x12>\n" +
 	"\vfabric_scan\x18\b \x01(\v2\x1b.llmgw.v1.FabricScanRequestH\x00R\n" +
-	"fabricScanB\x05\n" +
-	"\x03msg\"\xc7\x06\n" +
+	"fabricScan\x12>\n" +
+	"\vfabric_test\x18\t \x01(\v2\x1b.llmgw.v1.FabricTestRequestH\x00R\n" +
+	"fabricTestB\x05\n" +
+	"\x03msg\"\xe3\x06\n" +
 	"\bRegister\x12\x17\n" +
 	"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x1a\n" +
 	"\bhostname\x18\x02 \x01(\tR\bhostname\x12\x1b\n" +
@@ -3264,7 +3494,8 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
 	"\x04host\x18\r \x01(\v2\x0e.llmgw.v1.HostR\x04host\x12%\n" +
 	"\x0eengines_probed\x18\x0e \x01(\bR\renginesProbed\x12\"\n" +
 	"\x04imex\x18\x0f \x01(\v2\x0e.llmgw.v1.IMEXR\x04imex\x12/\n" +
-	"\x13installable_engines\x18\x10 \x03(\tR\x12installableEngines\x1a9\n" +
+	"\x13installable_engines\x18\x10 \x03(\tR\x12installableEngines\x12\x1a\n" +
+	"\bfeatures\x18\x11 \x03(\tR\bfeatures\x1a9\n" +
 	"\vLabelsEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
 	"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\x1aA\n" +
@@ -3536,7 +3767,23 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
 	"request_id\x18\x01 \x01(\tR\trequestId\x12\x16\n" +
 	"\x06output\x18\x02 \x01(\tR\x06output\x12\x14\n" +
 	"\x05error\x18\x03 \x01(\tR\x05error\x12\x17\n" +
-	"\atook_ms\x18\x04 \x01(\x03R\x06tookMs2O\n" +
+	"\atook_ms\x18\x04 \x01(\x03R\x06tookMs\"\xb5\x01\n" +
+	"\x11FabricTestRequest\x12\x1d\n" +
+	"\n" +
+	"request_id\x18\x01 \x01(\tR\trequestId\x12\x12\n" +
+	"\x04role\x18\x02 \x01(\tR\x04role\x12\x10\n" +
+	"\x03hca\x18\x03 \x01(\tR\x03hca\x12\x12\n" +
+	"\x04port\x18\x04 \x01(\x05R\x04port\x12\x12\n" +
+	"\x04peer\x18\x05 \x01(\tR\x04peer\x12\x19\n" +
+	"\btcp_port\x18\x06 \x01(\x05R\atcpPort\x12\x18\n" +
+	"\aseconds\x18\a \x01(\x05R\aseconds\"\x8b\x01\n" +
+	"\x0fFabricTestReply\x12\x1d\n" +
+	"\n" +
+	"request_id\x18\x01 \x01(\tR\trequestId\x12\x12\n" +
+	"\x04gbps\x18\x02 \x01(\x01R\x04gbps\x12\x16\n" +
+	"\x06output\x18\x03 \x01(\tR\x06output\x12\x14\n" +
+	"\x05error\x18\x04 \x01(\tR\x05error\x12\x17\n" +
+	"\atook_ms\x18\x05 \x01(\x03R\x06tookMs2O\n" +
 	"\fAgentService\x12?\n" +
 	"\aConnect\x12\x16.llmgw.v1.AgentMessage\x1a\x18.llmgw.v1.ControlMessage(\x010\x01B\x93\x01\n" +
 	"\fcom.llmgw.v1B\n" +
@@ -3555,7 +3802,7 @@ func file_llmgw_v1_agent_proto_rawDescGZIP() []byte {
 }
 
 var file_llmgw_v1_agent_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
-var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 37)
+var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 39)
 var file_llmgw_v1_agent_proto_goTypes = []any{
 	(ShardStatus_State)(0),    // 0: llmgw.v1.ShardStatus.State
 	(*AgentMessage)(nil),      // 1: llmgw.v1.AgentMessage
@@ -3589,12 +3836,14 @@ var file_llmgw_v1_agent_proto_goTypes = []any{
 	(*LogsReply)(nil),         // 29: llmgw.v1.LogsReply
 	(*FabricScanRequest)(nil), // 30: llmgw.v1.FabricScanRequest
 	(*FabricScanReply)(nil),   // 31: llmgw.v1.FabricScanReply
-	nil,                       // 32: llmgw.v1.Register.LabelsEntry
-	nil,                       // 33: llmgw.v1.Register.EngineVersionsEntry
-	nil,                       // 34: llmgw.v1.GPU.NicPathsEntry
-	nil,                       // 35: llmgw.v1.IBPort.CountersEntry
-	nil,                       // 36: llmgw.v1.Heartbeat.EngineVersionsEntry
-	nil,                       // 37: llmgw.v1.EngineProfile.EnvEntry
+	(*FabricTestRequest)(nil), // 32: llmgw.v1.FabricTestRequest
+	(*FabricTestReply)(nil),   // 33: llmgw.v1.FabricTestReply
+	nil,                       // 34: llmgw.v1.Register.LabelsEntry
+	nil,                       // 35: llmgw.v1.Register.EngineVersionsEntry
+	nil,                       // 36: llmgw.v1.GPU.NicPathsEntry
+	nil,                       // 37: llmgw.v1.IBPort.CountersEntry
+	nil,                       // 38: llmgw.v1.Heartbeat.EngineVersionsEntry
+	nil,                       // 39: llmgw.v1.EngineProfile.EnvEntry
 }
 var file_llmgw_v1_agent_proto_depIdxs = []int32{
 	3,  // 0: llmgw.v1.AgentMessage.register:type_name -> llmgw.v1.Register
@@ -3602,49 +3851,51 @@ var file_llmgw_v1_agent_proto_depIdxs = []int32{
 	20, // 2: llmgw.v1.AgentMessage.shard:type_name -> llmgw.v1.ShardStatus
 	29, // 3: llmgw.v1.AgentMessage.logs:type_name -> llmgw.v1.LogsReply
 	31, // 4: llmgw.v1.AgentMessage.fabric_scan:type_name -> llmgw.v1.FabricScanReply
-	22, // 5: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard
-	25, // 6: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard
-	26, // 7: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard
-	27, // 8: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack
-	24, // 9: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights
-	17, // 10: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision
-	28, // 11: llmgw.v1.ControlMessage.logs:type_name -> llmgw.v1.LogsRequest
-	30, // 12: llmgw.v1.ControlMessage.fabric_scan:type_name -> llmgw.v1.FabricScanRequest
-	32, // 13: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
-	7,  // 14: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
-	8,  // 15: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
-	33, // 16: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
-	6,  // 17: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
-	9,  // 18: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
-	5,  // 19: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
-	4,  // 20: llmgw.v1.Register.host:type_name -> llmgw.v1.Host
-	14, // 21: llmgw.v1.Register.imex:type_name -> llmgw.v1.IMEX
-	0,  // 22: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
-	34, // 23: llmgw.v1.GPU.nic_paths:type_name -> llmgw.v1.GPU.NicPathsEntry
-	11, // 24: llmgw.v1.NodeLoad.shards:type_name -> llmgw.v1.ShardLoad
-	35, // 25: llmgw.v1.IBPort.counters:type_name -> llmgw.v1.IBPort.CountersEntry
-	13, // 26: llmgw.v1.IBPort.subnet_managers:type_name -> llmgw.v1.SubnetManager
-	5,  // 27: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
-	9,  // 28: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
-	36, // 29: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
-	19, // 30: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus
-	10, // 31: llmgw.v1.Heartbeat.load:type_name -> llmgw.v1.NodeLoad
-	12, // 32: llmgw.v1.Heartbeat.ib_ports:type_name -> llmgw.v1.IBPort
-	8,  // 33: llmgw.v1.Heartbeat.fabrics:type_name -> llmgw.v1.FabricAttachment
-	15, // 34: llmgw.v1.Heartbeat.gpu_fabric:type_name -> llmgw.v1.GPUFabricState
-	14, // 35: llmgw.v1.Heartbeat.imex:type_name -> llmgw.v1.IMEX
-	18, // 36: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant
-	0,  // 37: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
-	21, // 38: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
-	23, // 39: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
-	37, // 40: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
-	1,  // 41: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
-	2,  // 42: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
-	42, // [42:43] is the sub-list for method output_type
-	41, // [41:42] is the sub-list for method input_type
-	41, // [41:41] is the sub-list for extension type_name
-	41, // [41:41] is the sub-list for extension extendee
-	0,  // [0:41] is the sub-list for field type_name
+	33, // 5: llmgw.v1.AgentMessage.fabric_test:type_name -> llmgw.v1.FabricTestReply
+	22, // 6: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard
+	25, // 7: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard
+	26, // 8: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard
+	27, // 9: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack
+	24, // 10: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights
+	17, // 11: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision
+	28, // 12: llmgw.v1.ControlMessage.logs:type_name -> llmgw.v1.LogsRequest
+	30, // 13: llmgw.v1.ControlMessage.fabric_scan:type_name -> llmgw.v1.FabricScanRequest
+	32, // 14: llmgw.v1.ControlMessage.fabric_test:type_name -> llmgw.v1.FabricTestRequest
+	34, // 15: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
+	7,  // 16: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
+	8,  // 17: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
+	35, // 18: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
+	6,  // 19: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
+	9,  // 20: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
+	5,  // 21: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
+	4,  // 22: llmgw.v1.Register.host:type_name -> llmgw.v1.Host
+	14, // 23: llmgw.v1.Register.imex:type_name -> llmgw.v1.IMEX
+	0,  // 24: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
+	36, // 25: llmgw.v1.GPU.nic_paths:type_name -> llmgw.v1.GPU.NicPathsEntry
+	11, // 26: llmgw.v1.NodeLoad.shards:type_name -> llmgw.v1.ShardLoad
+	37, // 27: llmgw.v1.IBPort.counters:type_name -> llmgw.v1.IBPort.CountersEntry
+	13, // 28: llmgw.v1.IBPort.subnet_managers:type_name -> llmgw.v1.SubnetManager
+	5,  // 29: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
+	9,  // 30: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
+	38, // 31: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
+	19, // 32: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus
+	10, // 33: llmgw.v1.Heartbeat.load:type_name -> llmgw.v1.NodeLoad
+	12, // 34: llmgw.v1.Heartbeat.ib_ports:type_name -> llmgw.v1.IBPort
+	8,  // 35: llmgw.v1.Heartbeat.fabrics:type_name -> llmgw.v1.FabricAttachment
+	15, // 36: llmgw.v1.Heartbeat.gpu_fabric:type_name -> llmgw.v1.GPUFabricState
+	14, // 37: llmgw.v1.Heartbeat.imex:type_name -> llmgw.v1.IMEX
+	18, // 38: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant
+	0,  // 39: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
+	21, // 40: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
+	23, // 41: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
+	39, // 42: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
+	1,  // 43: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
+	2,  // 44: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
+	44, // [44:45] is the sub-list for method output_type
+	43, // [43:44] is the sub-list for method input_type
+	43, // [43:43] is the sub-list for extension type_name
+	43, // [43:43] is the sub-list for extension extendee
+	0,  // [0:43] is the sub-list for field type_name
 }
 
 func init() { file_llmgw_v1_agent_proto_init() }
@@ -3658,6 +3909,7 @@ func file_llmgw_v1_agent_proto_init() {
 		(*AgentMessage_Shard)(nil),
 		(*AgentMessage_Logs)(nil),
 		(*AgentMessage_FabricScan)(nil),
+		(*AgentMessage_FabricTest)(nil),
 	}
 	file_llmgw_v1_agent_proto_msgTypes[1].OneofWrappers = []any{
 		(*ControlMessage_Start)(nil),
@@ -3668,6 +3920,7 @@ func file_llmgw_v1_agent_proto_init() {
 		(*ControlMessage_Provision)(nil),
 		(*ControlMessage_Logs)(nil),
 		(*ControlMessage_FabricScan)(nil),
+		(*ControlMessage_FabricTest)(nil),
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -3675,7 +3928,7 @@ func file_llmgw_v1_agent_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_llmgw_v1_agent_proto_rawDesc), len(file_llmgw_v1_agent_proto_rawDesc)),
 			NumEnums:      1,
-			NumMessages:   37,
+			NumMessages:   39,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

api/proto/llmgw/v1/agent.proto

@@ -25,6 +25,7 @@ message AgentMessage {
     ShardStatus shard = 3;
     LogsReply logs = 4;
     FabricScanReply fabric_scan = 5;
+    FabricTestReply fabric_test = 6;
   }
 }
 
@@ -39,6 +40,7 @@ message ControlMessage {
     Provision provision = 6;
     LogsRequest logs = 7;
     FabricScanRequest fabric_scan = 8;
+    FabricTestRequest fabric_test = 9;
   }
 }
 
@@ -125,6 +127,12 @@ message Register {
   // vLLM, it answered "unsupported", and every node of a cluster half way
   // through an upgrade showed as failing.
   repeated string installable_engines = 16;
+
+  // features are the requests this agent answers beyond the ones every agent
+  // does: "fabric-test". An agent that does not know a request ignores it, so
+  // without this the control plane would wait minutes for an answer that is
+  // never coming.
+  repeated string features = 17;
 }
 
 // Host describes the operating system and processor of a node.
@@ -605,3 +613,35 @@ message FabricScanReply {
   string error = 3;
   int64 took_ms = 4;
 }
+
+// FabricTestRequest asks a node for one side of an RDMA bandwidth test
+// between two nodes: ib_write_bw, from perftest, over one adapter. The
+// server side is asked first and answers when the test is over; the client
+// dials it. What it proves is what a replica spanning nodes needs and nothing
+// on one node can show: that these two adapters reach each other through the
+// fabric, and how fast.
+message FabricTestRequest {
+  string request_id = 1;
+  // role is "server" or "client".
+  string role = 2;
+  string hca = 3;
+  int32 port = 4;
+  // peer is the server's management address, for the client. perftest
+  // exchanges its connection details over TCP before any RDMA.
+  string peer = 5;
+  // tcp_port is that exchange's port, the same on both sides.
+  int32 tcp_port = 6;
+  int32 seconds = 7;
+}
+
+// FabricTestReply is what one side measured.
+message FabricTestReply {
+  string request_id = 1;
+  // gbps is the average bandwidth, in gigabits a second.
+  double gbps = 2;
+  // output is the end of the tool's output, for when the number is not
+  // enough.
+  string output = 3;
+  string error = 4;
+  int64 took_ms = 5;
+}

docs/fabrics.md

@@ -121,6 +121,39 @@ rate and errors, and the NCCL environment a shard on each fabric would be given.
 `--json` writes every fact it read as well as what it concluded — send that
 file when the report is not enough.
 
+## Checking two nodes: `llmgw ctl fabric-test`
+
+Every check on one node can pass while two nodes still cannot reach each other
+— a cable into the wrong switch, a partition key, a RoCE network with no route
+— and a replica spanning them then hangs on its first all-reduce. Before
+placing one, on the control plane:
+
+```
+llmgw ctl fabric-test gpu01 gpu02
+```
+
+```
+gpu01 → gpu02 on ib-compute, 41s
+
+RAIL              LINK      MEASURED    VERDICT  DETAIL
+mlx5_0 → mlx5_0   400 Gb/s  388.2 Gb/s  ok       97% of 400 Gb/s
+mlx5_1 → mlx5_1   400 Gb/s  241.0 Gb/s  slow     60% of 400 Gb/s: check the cable, and the card's PCIe width and NUMA node
+mlx5_2 → mlx5_2   400 Gb/s  0.0 Gb/s    fail     gpu01: ib_write_bw: Couldn't connect to 10.1.0.22:18517
+```
+
+It runs `ib_write_bw` (perftest, installed on a node the first time) from each
+of the first node's adapters on the fabric to the same adapter on the second —
+mlx5_0 to mlx5_0, the way NCCL pairs them — one rail at a time, through the
+agents, so no SSH between nodes is needed. Each rail is graded against its
+link's rate: **ok** at 80% or more, **slow** at 50%, **fail** below that or when
+nothing got through. `--fabric` picks one when the nodes share several,
+`--seconds` the length of each rail's run. The traffic is real: run it when the
+fabric is quiet.
+
+It measures RDMA between adapters, not GPUDirect or NCCL itself: a pass says
+the fabric carries traffic between these two nodes at its rate, which is what
+a first all-reduce needs and what nothing on one node can show.
+
 ## Watching them
 
 - **Cluster page, Fabrics.** Each declared fabric and what it physically is;
@@ -172,6 +205,6 @@ workers headless. `launch_mode: ray` is there for a vLLM too old for that.
 | It | Look at |
 |---|---|
 | is not placed | `llmgw ctl explain-placement <model>`: a fabric split by identity, a node whose IMEX is down, a node refused because its adapters are cabled to two fabrics |
-| starts and hangs on its first collective | the cluster page's fabric problems; `doctor` on each node — a port down, the wrong interface, RoCE without a v2 GID |
+| starts and hangs on its first collective | `llmgw ctl fabric-test` between two of its nodes; the cluster page's fabric problems; `doctor` on each node — a port down, the wrong interface, RoCE without a v2 GID |
 | runs, slowly | `nvidia_peermem`; the map for a degraded cable; the node page's NVLink matrix for a group that crossed PCIe |
 | worked yesterday | the subnet manager table for a failover; `link_downed` on the node's ports |

docs/quick-install.md

@@ -35,10 +35,12 @@ install it.
 ## 2. Set up the control plane
 
 ```
-sudo llmgw server init --admin you@example.com
+sudo llmgw server init
 ```
 
-This one command sets up everything on this machine, and takes a few minutes:
+It asks for your email and a few choices, then sets up everything on this
+machine, in a few minutes. In a script, give them as flags instead:
+`sudo llmgw server init --admin you@example.com`.
 
 - the service, its certificates and its configuration;
 - your administrator account. **The password is printed once. Copy it now.**
@@ -54,18 +56,33 @@ sudo llmgw server init --admin you@example.com --host llm.example.com
 
 It is safe to run again. It keeps what exists and finishes what is missing.
 
+**The mirror on its own machine**, when it wants a larger disk than the
+control plane has: first on that machine, with llmgw installed as in step 1,
+
+```
+sudo llmgw mirror init --control-plane <control plane address>
+```
+
+then on the control plane `sudo llmgw server init --mirror http://<mirror
+address>:8082`. The mirror serves model files to every node, and takes
+orders only from the control plane. Untick "This machine as a node" in the
+form (or add `--no-node`) to keep models off the control plane; the starter
+model then starts on the first node you add.
+
 **Check:** open `http://<control-plane>:8080/chat`, sign in with the printed
 password, and ask something. The small model answers. Change the password on
 the account page.
 
 ## 3. Add the GPU nodes
 
-On the control plane, for each node:
+On the control plane:
 
 ```
 sudo llmgw node add ubuntu@10.0.0.21
 ```
 
+Several at once: `sudo llmgw node add ubuntu@10.0.0.21 ubuntu@10.0.0.22 ubuntu@10.0.0.23`.
+
 Replace `ubuntu` with the user you log in to the node as, and `10.0.0.21` with
 its address. That user needs `sudo` without a password on the node.
 
@@ -89,6 +106,13 @@ the same fabric the same name. In a script, give it instead:
 sudo llmgw node add ubuntu@10.0.0.21 --fabric ib-a:mlx5_0,mlx5_1,mlx5_2,mlx5_3
 ```
 
+Once two nodes on a fabric are in, check they reach each other at the link's
+rate before placing anything across them:
+
+```
+llmgw ctl fabric-test gpu01 gpu02
+```
+
 **If it cannot log in.** `sudo` runs SSH as root, with root's keys. To use your
 own keys instead, log in to the control plane with `ssh -A` and keep your
 agent through sudo:

internal/agent/agent.go

@@ -844,6 +844,7 @@ func (a *Agent) session(ctx context.Context, server string) error {
 	reg.EngineVersions = a.engines(parent)
 	reg.EnginesProbed = true
 	reg.InstallableEngines = provision.Installable()
+	reg.Features = []string{FeatureFabricTest}
 	reg.RunningShards = a.runningShards()
 	a.mu.Lock()
 	h := a.prov.Host
@@ -1145,6 +1146,16 @@ func (a *Agent) handle(ctx context.Context, msg *llmgwv1.ControlMessage, out cha
 			case <-ctx.Done():
 			}
 		}(m.FabricScan)
+	case *llmgwv1.ControlMessage_FabricTest:
+		// Seconds of traffic, and a package install the first time: off the
+		// receive loop too.
+		go func(req *llmgwv1.FabricTestRequest) {
+			reply := a.fabricTest(ctx, req)
+			select {
+			case out <- &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_FabricTest{FabricTest: reply}}:
+			case <-ctx.Done():
+			}
+		}(m.FabricTest)
 	case *llmgwv1.ControlMessage_Ack:
 		if !m.Ack.GetOk() {
 			a.log.Warn("control plane rejected a message",
@@ -1758,6 +1769,70 @@ func (a *Agent) ensureDiags(ctx context.Context) {
 	}
 }
 
+// FeatureFabricTest is the agent answering FabricTestRequest.
+const FeatureFabricTest = "fabric-test"
+
+// runWriteBW is discovery.RunWriteBW, and haveWriteBW whether it is
+// installed; both replaced in tests.
+var (
+	runWriteBW  = discovery.RunWriteBW
+	haveWriteBW = func() bool { _, err := exec.LookPath("ib_write_bw"); return err == nil }
+)
+
+// fabricTest runs this node's side of a bandwidth test with another.
+func (a *Agent) fabricTest(ctx context.Context, req *llmgwv1.FabricTestRequest) *llmgwv1.FabricTestReply {
+	start := time.Now()
+	reply := &llmgwv1.FabricTestReply{RequestId: req.GetRequestId()}
+	defer func() { reply.TookMs = time.Since(start).Milliseconds() }()
+
+	w := discovery.WriteBW{HCA: req.GetHca(), Port: int(req.GetPort()), GIDIndex: -1,
+		TCPPort: int(req.GetTcpPort()), Seconds: int(req.GetSeconds())}
+	if req.GetRole() == "client" {
+		w.Peer = req.GetPeer()
+	}
+	a.mu.Lock()
+	found := false
+	for _, h := range a.inv.HCAs {
+		for _, p := range h.Ports {
+			if h.Name == w.HCA && p.Num == w.Port {
+				found = true
+				// RoCE needs its v2 GID named, as NCCL does; InfiniBand uses the LID.
+				if p.LinkLayer == discovery.LinkEthernet && p.RoCEv2GID >= 0 {
+					w.GIDIndex = p.RoCEv2GID
+				}
+			}
+		}
+	}
+	a.mu.Unlock()
+	if !found {
+		reply.Error = fmt.Sprintf("this node has no port %s/%d", w.HCA, w.Port)
+		return reply
+	}
+	if !haveWriteBW() && a.prov != nil {
+		a.log.Info("installing perftest, for a bandwidth test with another node")
+		pkgs := map[string][]string{"deb": {"perftest"}, "rpm": {"perftest"}}
+		if err := a.prov.InstallPackages(ctx, pkgs); err != nil {
+			reply.Error = "installing perftest: " + err.Error()
+			return reply
+		}
+	}
+	gbps, out, err := runWriteBW(ctx, w)
+	reply.Gbps = gbps
+	reply.Output = tailLines(out, 12)
+	if err != nil {
+		reply.Error = err.Error()
+	}
+	return reply
+}
+
+func tailLines(s string, n int) string {
+	lines := strings.Split(strings.TrimRight(s, "\n"), "\n")
+	if len(lines) > n {
+		lines = lines[len(lines)-n:]
+	}
+	return strings.Join(lines, "\n")
+}
+
 // discoverGPUFabric and discoverIMEX are replaced in tests.
 var (
 	discoverGPUFabric = discovery.DiscoverGPUFabric

internal/agent/agent_test.go

@@ -23,6 +23,7 @@ import (
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/h2c"
 
+	llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
@@ -881,3 +882,34 @@ func TestAFabricScanIsAskedForAndAnswered(t *testing.T) {
 		t.Errorf("asked %q, got %+v", asked, r)
 	}
 }
+
+func TestABandwidthTestIsRunOnTheAdapterAskedFor(t *testing.T) {
+	oldRun, oldHave := runWriteBW, haveWriteBW
+	t.Cleanup(func() { runWriteBW, haveWriteBW = oldRun, oldHave })
+	haveWriteBW = func() bool { return true }
+	var got discovery.WriteBW
+	runWriteBW = func(_ context.Context, w discovery.WriteBW) (float64, string, error) {
+		got = w
+		return 388.2, " #bytes  #iterations  BW peak[Gb/sec]  BW average[Gb/sec]\n", nil
+	}
+	h := start(t, func(c *Config) { c.Discovery.IBRoot = ibTreeWithActivePorts(t, "mlx5_0") })
+
+	r, err := h.svc.FabricTest(t.Context(), "n1", &llmgwv1.FabricTestRequest{
+		Role: "client", Hca: "mlx5_0", Port: 1, Peer: "10.0.0.2", TcpPort: 18515, Seconds: 3})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if r.GetGbps() != 388.2 || r.GetError() != "" {
+		t.Errorf("reply %+v", r)
+	}
+	if got.HCA != "mlx5_0" || got.Peer != "10.0.0.2" || got.TCPPort != 18515 || got.GIDIndex != -1 {
+		t.Errorf("ran %+v", got)
+	}
+
+	// An adapter the node does not have is said so, without running anything.
+	got = discovery.WriteBW{}
+	r, _ = h.svc.FabricTest(t.Context(), "n1", &llmgwv1.FabricTestRequest{Role: "server", Hca: "mlx5_9", Port: 1, Seconds: 1})
+	if !strings.Contains(r.GetError(), "no port mlx5_9/1") || got.HCA != "" {
+		t.Errorf("reply %+v, ran %+v", r, got)
+	}
+}

internal/agent/discovery/perftest.go

@@ -0,0 +1,133 @@
+package discovery
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os/exec"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// WriteBW is one side of an ib_write_bw run, from perftest: an RDMA write
+// bandwidth test between two adapters, which is what a collective spanning
+// nodes rides on.
+type WriteBW struct {
+	HCA  string
+	Port int
+	// GIDIndex is the RoCE v2 GID to use; negative on InfiniBand, where the
+	// port's LID is used.
+	GIDIndex int
+	// Peer is the server's address, for the client; empty makes the server.
+	Peer    string
+	TCPPort int
+	Seconds int
+}
+
+// Args are the command line: a duration rather than a count, so the test
+// takes as long on a slow link as a fast one, and results in Gb/s.
+func (w WriteBW) Args() []string {
+	args := []string{"-d", w.HCA, "-i", strconv.Itoa(max(w.Port, 1)),
+		"-p", strconv.Itoa(w.TCPPort), "-D", strconv.Itoa(max(w.Seconds, 1)),
+		"-F", "--report_gbits"}
+	if w.GIDIndex >= 0 {
+		args = append(args, "-x", strconv.Itoa(w.GIDIndex))
+	}
+	if w.Peer != "" {
+		args = append(args, w.Peer)
+	}
+	return args
+}
+
+// RunWriteBW runs one side and returns the average bandwidth the output
+// reports, in Gb/s, and the output. The client retries its connection for a
+// while: the server is started first, but a few seconds may pass before it
+// listens.
+func RunWriteBW(ctx context.Context, w WriteBW) (float64, string, error) {
+	bin, err := exec.LookPath("ib_write_bw")
+	if err != nil {
+		return 0, "", errors.New("ib_write_bw is not installed (the perftest package)")
+	}
+	limit := time.Duration(w.Seconds+60) * time.Second
+	var out []byte
+	for attempt := 0; ; attempt++ {
+		cctx, cancel := context.WithTimeout(ctx, limit)
+		out, err = exec.CommandContext(cctx, bin, w.Args()...).CombinedOutput() //nolint:gosec // fixed tool
+		cancel()
+		if err == nil || w.Peer == "" || !refused(string(out)) || attempt >= 10 {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			return 0, string(out), ctx.Err()
+		case <-time.After(time.Second):
+		}
+	}
+	text := string(out)
+	if err != nil {
+		return 0, text, fmt.Errorf("ib_write_bw: %v: %s", err, lastNonEmpty(text))
+	}
+	gbps, perr := ParseWriteBW(text)
+	if perr != nil && w.Peer != "" {
+		// Only the client is sure to print the result; a server that does not
+		// is not a failure.
+		return 0, text, perr
+	}
+	return gbps, text, nil
+}
+
+func refused(out string) bool {
+	for _, s := range []string{"Couldn't connect", "Unable to init the socket connection", "Connection refused"} {
+		if strings.Contains(out, s) {
+			return true
+		}
+	}
+	return false
+}
+
+var columnSplit = regexp.MustCompile(`\s{2,}`)
+
+// ParseWriteBW finds the average bandwidth in perftest's result table:
+//
+//	#bytes     #iterations    BW peak[Gb/sec]    BW average[Gb/sec]   MsgRate[Mpps]
+//	65536      887036           0.00               96.93               0.184877
+//
+// by the heading's position rather than a fixed column, since versions add
+// columns and the peak reads 0.00 when the test runs for a duration.
+func ParseWriteBW(out string) (float64, error) {
+	lines := strings.Split(out, "\n")
+	for i, l := range lines {
+		if !strings.Contains(l, "BW average") {
+			continue
+		}
+		heads := columnSplit.Split(strings.TrimSpace(l), -1)
+		col := -1
+		for j, h := range heads {
+			if strings.HasPrefix(h, "BW average") {
+				col = j
+			}
+		}
+		for _, d := range lines[i+1:] {
+			f := strings.Fields(d)
+			if len(f) == 0 || strings.HasPrefix(f[0], "-") {
+				continue
+			}
+			if len(f) != len(heads) || col < 0 {
+				return 0, fmt.Errorf("perftest's result table has a shape this does not know: %q / %q", l, d)
+			}
+			v, err := strconv.ParseFloat(f[col], 64)
+			if err != nil {
+				return 0, fmt.Errorf("reading the average bandwidth %q: %w", f[col], err)
+			}
+			return v, nil
+		}
+	}
+	return 0, errors.New("perftest printed no result")
+}
+
+func lastNonEmpty(s string) string {
+	lines := strings.Split(strings.TrimSpace(s), "\n")
+	return strings.TrimSpace(lines[len(lines)-1])
+}

internal/agent/discovery/perftest_test.go

@@ -0,0 +1,49 @@
+package discovery
+
+import (
+	"slices"
+	"testing"
+)
+
+// The client's output of perftest 24.x, with a duration.
+const writeBWClient = `---------------------------------------------------------------------------------------
+                    RDMA_Write BW Test
+ Dual-port       : OFF		Device         : mlx5_0
+ Number of qps   : 1		Transport type : IB
+ Connection type : RC		Using SRQ      : OFF
+ TX depth        : 128
+ CQ Moderation   : 100
+ Mtu             : 4096[B]
+ Link type       : IB
+ Max inline data : 0[B]
+ rdma_cm QPs	 : OFF
+ Data ex. method : Ethernet
+---------------------------------------------------------------------------------------
+ local address: LID 0x05 QPN 0x0112 PSN 0x7b1f02 RKey 0x1fffbd VAddr 0x007f1d2a5f0000
+ remote address: LID 0x09 QPN 0x0113 PSN 0x2c1e44 RKey 0x1ffeaf VAddr 0x007fa4e1a60000
+---------------------------------------------------------------------------------------
+ #bytes     #iterations    BW peak[Gb/sec]    BW average[Gb/sec]   MsgRate[Mpps]
+ 65536      1851214          0.00               388.21		   0.740422
+---------------------------------------------------------------------------------------
+`
+
+func TestTheAverageBandwidthIsReadFromPerftest(t *testing.T) {
+	got, err := ParseWriteBW(writeBWClient)
+	if err != nil || got != 388.21 {
+		t.Errorf("got %v, %v", got, err)
+	}
+	if _, err := ParseWriteBW(" Couldn't connect to 10.0.0.2:18515\n"); err == nil {
+		t.Error("output with no result table gave a number")
+	}
+}
+
+func TestWriteBWArguments(t *testing.T) {
+	ib := WriteBW{HCA: "mlx5_0", Port: 1, GIDIndex: -1, TCPPort: 18515, Seconds: 3}.Args()
+	if slices.Contains(ib, "-x") {
+		t.Errorf("InfiniBand was given a GID index: %v", ib)
+	}
+	roce := WriteBW{HCA: "mlx5_0", Port: 1, GIDIndex: 3, Peer: "10.0.0.2", TCPPort: 18515, Seconds: 3}.Args()
+	if !slices.Contains(roce, "-x") || roce[len(roce)-1] != "10.0.0.2" {
+		t.Errorf("RoCE client: %v", roce)
+	}
+}

internal/cli/ctl.go

@@ -120,6 +120,7 @@ func newCtlCmd() *cobra.Command {
 		newModelsCmd(client, &asJSON),
 		newReplicasCmd(client, &asJSON),
 		newExplainCmd(client, &asJSON),
+		newFabricTestCmd(client, &asJSON),
 		newNodeStateCmd(client, "cordon", "cordoned",
 			"Stop placing new work on a node; what runs there keeps running"),
 		newNodeStateCmd(client, "uncordon", "ready",

internal/cli/ctl_fabrictest.go

@@ -0,0 +1,75 @@
+package cli
+
+import (
+	"encoding/json"
+	"fmt"
+	"text/tabwriter"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
+)
+
+func newFabricTestCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
+	var (
+		fabric  string
+		seconds int
+	)
+	c := &cobra.Command{
+		Use:   "fabric-test NODE-A NODE-B",
+		Short: "Measure RDMA bandwidth between two nodes, rail by rail",
+		Long: `Runs ib_write_bw from each of NODE-A's adapters on a fabric to the same
+adapter on NODE-B — mlx5_0 to mlx5_0, and so on, the way NCCL pairs them —
+one rail at a time, and grades each against its link's rate: ok at 80% or
+more, slow at 50%, fail below or when nothing got through.
+
+Every check on one node can pass while two nodes still cannot reach each
+other: a cable into the wrong switch, a partition key, a RoCE network with no
+route. A replica spanning them then hangs on its first all-reduce. Run this
+before placing one.
+
+It goes through the agents, so it needs no ssh between nodes. perftest is
+installed on a node the first time it is asked. The traffic is real and
+takes a few seconds a rail; run it when the fabric is quiet.`,
+		Example: `  llmgw ctl fabric-test gpu01 gpu02
+  llmgw ctl fabric-test gpu01 gpu02 --fabric ib-compute --seconds 10`,
+		Args: cobra.ExactArgs(2),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			body, _ := json.Marshal(map[string]any{"a": args[0], "b": args[1], "fabric": fabric, "seconds": seconds})
+			cl := client()
+			// Rails one after another, each for its seconds and a settle; the
+			// client's usual timeout is for questions, not measurements.
+			cl.http.Timeout = 30 * time.Minute
+			var v adminapi.FabricTestView
+			if err := cl.post(cmd.Context(), "/api/v1/fabric-test", body, &v); err != nil {
+				return err
+			}
+			if *asJSON {
+				return json.NewEncoder(cmd.OutOrStdout()).Encode(v)
+			}
+			out := cmd.OutOrStdout()
+			fmt.Fprintf(out, "%s → %s on %s, %s\n\n", v.A, v.B, v.Fabric, v.Took)
+			w := tabwriter.NewWriter(out, 0, 0, 2, ' ', 0)
+			fmt.Fprintln(w, "RAIL\tLINK\tMEASURED\tVERDICT\tDETAIL")
+			for _, r := range v.Rails {
+				link := "-"
+				if r.RateGbps > 0 {
+					link = fmt.Sprintf("%d Gb/s", r.RateGbps)
+				}
+				fmt.Fprintf(w, "%s → %s\t%s\t%.1f Gb/s\t%s\t%s\n", r.A, r.B, link, r.Gbps, r.Verdict, r.Detail)
+			}
+			_ = w.Flush()
+			for _, n := range v.Notes {
+				fmt.Fprintf(out, "\nnote: %s\n", n)
+			}
+			if !v.OK {
+				return fmt.Errorf("not every rail passed")
+			}
+			return nil
+		},
+	}
+	c.Flags().StringVar(&fabric, "fabric", "", "the fabric to test (default: the one the two nodes share)")
+	c.Flags().IntVar(&seconds, "seconds", 3, "how long each rail runs, up to 30")
+	return c
+}

internal/cli/dispatcherref_test.go

@@ -5,6 +5,7 @@ import (
 
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest"
 )
 
 func TestTheForwardingDispatcherCarriesEveryCapability(t *testing.T) {
@@ -23,10 +24,17 @@ func TestTheForwardingDispatcherCarriesEveryCapability(t *testing.T) {
 		t.Error("dispatcherRef drops WeightRemover, so the control plane cannot reclaim disk")
 	}
 
+	if _, ok := ref.(fabrictest.Asker); !ok {
+		t.Error("dispatcherRef drops the bandwidth test, so ctl fabric-test cannot reach the agents")
+	}
+
 	// And the thing it forwards to must have it, or the forwarding has nothing
 	// to reach.
 	var svc any = &agentsvc.Service{}
 	if _, ok := svc.(control.WeightRemover); !ok {
 		t.Error("the agent service does not implement WeightRemover")
 	}
+	if _, ok := svc.(fabrictest.Asker); !ok {
+		t.Error("the agent service does not implement the bandwidth test")
+	}
 }

internal/cli/mirror.go

@@ -4,7 +4,9 @@ import (
 	"encoding/json"
 	"fmt"
 	"log/slog"
+	"net"
 	"net/http"
+	"net/netip"
 	"os"
 	"strings"
 	"time"
@@ -44,6 +46,7 @@ func newMirrorCmd() *cobra.Command {
 		newMirrorUnusedCmd(&dir),
 		newMirrorRemoveCmd(&dir),
 		newMirrorServeCmd(&dir),
+		newMirrorInitCmd(&dir),
 	)
 	return c
 }
@@ -229,6 +232,7 @@ func newMirrorRemoveCmd(dir *string) *cobra.Command {
 func newMirrorServeCmd(dir *string) *cobra.Command {
 	var listen, apiListen, base string
 	var tlsc tlsconf.Config
+	var allow []string
 
 	c := &cobra.Command{
 		Use:   "serve",
@@ -268,9 +272,13 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
 				Handler:           readOnly(withManifest(*dir, http.FileServer(http.Dir(*dir)))),
 				ReadHeaderTimeout: 10 * time.Second,
 			}
+			allowed, err := parseAllow(allow)
+			if err != nil {
+				return err
+			}
 			admin := &http.Server{
 				Addr:              apiListen,
-				Handler:           api.Handler(),
+				Handler:           onlyFrom(allowed, api.Handler()),
 				ReadHeaderTimeout: 10 * time.Second,
 			}
 
@@ -279,9 +287,12 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
 			fmt.Fprintf(out, "serving %s on %s\n", *dir, listen)
 
 			go func() { errs <- fmt.Errorf("serving the API: %w", serveAPI(admin, tlsc)) }()
-			if tlsc.Enabled() {
+			switch {
+			case tlsc.Enabled():
 				fmt.Fprintf(out, "management API on %s, mutual TLS against %s\n", apiListen, tlsc.CA)
-			} else {
+			case len(allowed) > 0:
+				fmt.Fprintf(out, "management API on %s, for %s and this machine only\n", apiListen, strings.Join(allow, ", "))
+			default:
 				// Said plainly rather than left to be discovered. An API that
 				// can delete models and start downloads, open to anyone who can
 				// reach the port, is a choice the operator should have made
@@ -300,6 +311,8 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
 	c.Flags().StringVar(&apiListen, "api-listen", ":8082", "address to serve the management API on")
 	c.Flags().StringVar(&base, "base", "http://localhost:8081",
 		"the address nodes use to reach this mirror, as it should appear in catalog sources")
+	c.Flags().StringArrayVar(&allow, "api-allow", nil,
+		"an address or network the management API answers, besides this machine; repeatable (default: anyone)")
 	c.Flags().StringVar(&tlsc.CA, "tls-ca", "", "CA bundle that management clients must present a certificate from")
 	c.Flags().StringVar(&tlsc.Cert, "tls-cert", "", "this mirror's certificate")
 	c.Flags().StringVar(&tlsc.Key, "tls-key", "", "this mirror's private key")
@@ -398,3 +411,54 @@ func versionPath(p string) (model, version string, ok bool) {
 	}
 	return parts[0], parts[1], true
 }
+
+// parseAllow reads addresses and networks: 10.0.0.140, 10.0.0.0/24.
+func parseAllow(list []string) ([]netip.Prefix, error) {
+	var out []netip.Prefix
+	for _, s := range list {
+		for _, one := range strings.Split(s, ",") {
+			one = strings.TrimSpace(one)
+			if one == "" {
+				continue
+			}
+			if p, err := netip.ParsePrefix(one); err == nil {
+				out = append(out, p.Masked())
+				continue
+			}
+			a, err := netip.ParseAddr(one)
+			if err != nil {
+				return nil, fmt.Errorf("--api-allow %q is not an address or a network", one)
+			}
+			out = append(out, netip.PrefixFrom(a.Unmap(), a.Unmap().BitLen()))
+		}
+	}
+	return out, nil
+}
+
+// onlyFrom answers only this machine and the addresses allowed, when any
+// are: the management API can delete models and start downloads, and a
+// mirror on its own machine is reached over the network by one control plane
+// and no one else. Nothing allowed is the old behaviour, anyone.
+func onlyFrom(allowed []netip.Prefix, next http.Handler) http.Handler {
+	if len(allowed) == 0 {
+		return next
+	}
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		host, _, err := net.SplitHostPort(r.RemoteAddr)
+		a, perr := netip.ParseAddr(host)
+		if err != nil || perr != nil {
+			http.Error(w, "forbidden", http.StatusForbidden)
+			return
+		}
+		a = a.Unmap()
+		ok := a.IsLoopback()
+		for _, p := range allowed {
+			ok = ok || p.Contains(a)
+		}
+		if !ok {
+			http.Error(w, "this mirror is managed by its control plane only", http.StatusForbidden)
+			return
+		}
+		next.ServeHTTP(w, r)
+	})
+}

internal/cli/mirrorinit.go

@@ -0,0 +1,125 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"os"
+	"runtime"
+	"strings"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
+)
+
+// mirror init sets a machine up as the cluster's model mirror, on its own:
+// the bytes for every node on :8081, the management API on :8082 for the
+// control plane alone. server init does the same beside the control plane;
+// this is for when the mirror wants its own disk.
+
+func newMirrorInitCmd(dir *string) *cobra.Command {
+	var (
+		controlPlanes []string
+		base          string
+	)
+	cmd := &cobra.Command{
+		Use:   "init",
+		Short: "Set this machine up as the cluster's model mirror and start it",
+		Long: `Creates the llmgw user and the mirror's directory, and a systemd unit that
+serves the model files to every node on :8081 and the management API on
+:8082, and starts it.
+
+The management API can fetch and delete models, so it answers this machine
+and the control planes given with --control-plane, and nobody else. Then on
+the control plane:
+
+  sudo llmgw server init --mirror http://<this machine>:8082
+
+Safe to run again: it changes only what differs.`,
+		Example: `  sudo llmgw mirror init --control-plane 10.0.0.10`,
+		Args:    cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			if runtime.GOOS != "linux" {
+				return fmt.Errorf("mirror init sets up a Linux machine; this is %s", runtime.GOOS)
+			}
+			if os.Geteuid() != 0 {
+				return errors.New("mirror init creates a user, writes /etc and starts a service: run it as root")
+			}
+			if len(controlPlanes) == 0 {
+				return errors.New("--control-plane: which control plane manages this mirror (its address)")
+			}
+			if _, err := parseAllow(controlPlanes); err != nil {
+				return err
+			}
+			return mirrorInit(cmd.Context(), cmd.OutOrStdout(), *dir, base, controlPlanes)
+		},
+	}
+	cmd.Flags().StringArrayVar(&controlPlanes, "control-plane", nil,
+		"address of the control plane that manages this mirror; repeatable")
+	cmd.Flags().StringVar(&base, "base", "",
+		"the address nodes fetch from, http://host:8081 (default: this machine's first address)")
+	return cmd
+}
+
+func mirrorInit(ctx context.Context, out io.Writer, dir, base string, controlPlanes []string) error {
+	in := &serverInit{user: "llmgw", stateDir: "/var/lib/llmgw", out: out}
+	if err := in.ensureUser(ctx); err != nil {
+		return err
+	}
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return err
+	}
+	if err := in.chown(dir, true); err != nil {
+		return err
+	}
+	addr := ""
+	if ips := localAddrs(); len(ips) > 0 {
+		addr = ips[0].String()
+	}
+	if base == "" {
+		if addr == "" {
+			return errors.New("this machine has no address to give nodes; give one with --base")
+		}
+		base = "http://" + net.JoinHostPort(addr, "8081")
+	}
+	unit := mirrorUnitFor(dir, ":8082", base, controlPlanes,
+		"'llmgw mirror init'. Bytes on :8081 for every node; the\n# management API on :8082, for the control plane only.")
+	changed, err := in.writeUnit(mirrorUnitPath, unit)
+	if err != nil {
+		return err
+	}
+	for _, args := range [][]string{{"daemon-reload"}, {"enable", "--now", "llmgw-mirror"}} {
+		if err := in.systemctl(ctx, args...); err != nil {
+			return err
+		}
+	}
+	if changed {
+		if err := in.systemctl(ctx, "restart", "llmgw-mirror"); err != nil {
+			return err
+		}
+		in.done("model mirror in %s: files on :8081 at %s, management on :8082 for %s", dir, base, strings.Join(controlPlanes, ", "))
+	} else {
+		in.kept("model mirror in %s", dir)
+	}
+	m, err := mirror.NewClient("http://127.0.0.1:8082", tlsconf.Config{}, 5*time.Second)
+	if err != nil {
+		return err
+	}
+	if err := waitFor(ctx, 30*time.Second, "the mirror to answer", func() (bool, error) {
+		return m.Health(ctx) == nil, nil
+	}); err != nil {
+		return err
+	}
+	in.kept("answering")
+	api := "http://<this machine>:8082"
+	if addr != "" {
+		api = "http://" + net.JoinHostPort(addr, "8082")
+	}
+	fmt.Fprintf(out, "\nThe mirror is up. On the control plane:\n\n  sudo llmgw server init --mirror %s\n\n", api)
+	return nil
+}

internal/cli/node.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"slices"
 	"strings"
 	"time"
 
@@ -64,12 +65,15 @@ type nodeAdd struct {
 	noWait    bool
 	plain     bool
 	serverCfg string
+	// named are the fabric names given in this run, by what the adapters
+	// report the fabric to be.
+	named map[string]string
 }
 
 func newNodeAddCmd() *cobra.Command {
 	var na nodeAdd
 	cmd := &cobra.Command{
-		Use:   "add [user@]address",
+		Use:   "add [user@]address...",
 		Short: "Make a machine a node: check it, give it an identity, start its agent",
 		Long: `Run on the machine holding the certificate authority — the control plane,
 after 'llmgw server init' — as root. It logs in to the machine with ssh,
@@ -87,14 +91,18 @@ Nothing else needs giving, except --fabric for InfiniBand or RoCE: which
 fabric an adapter is on has a name only you can choose, and it must be the
 same on every node attached to it. Doctor suggests one when it sees adapters.
 
+Several addresses add several nodes, one after another; a fabric named for
+one is recognised on the next and not asked again.
+
 The user needs sudo on the machine without a password. Run it again at any
 time: the node gets a fresh certificate and configuration, and its running
 engines are kept.`,
 		Example: `  sudo llmgw node add 10.0.0.21
+  sudo llmgw node add ubuntu@10.0.0.21 ubuntu@10.0.0.22 ubuntu@10.0.0.23
   sudo llmgw node add ubuntu@gpu01 --fabric ib-a:mlx5_0,mlx5_1,mlx5_2,mlx5_3`,
-		Args: cobra.ExactArgs(1),
+		Args: cobra.MinimumNArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			return na.run(cmd.Context(), args[0], cmd.OutOrStdout())
+			return na.run(cmd.Context(), args, cmd.OutOrStdout())
 		},
 	}
 	f := cmd.Flags()
@@ -120,18 +128,9 @@ engines are kept.`,
 	return cmd
 }
 
-func (na *nodeAdd) prepare(addr string) error {
+// prepare works out what every node added in this run shares.
+func (na *nodeAdd) prepare() error {
 	o := &na.opts
-	o.Addr = addr
-	if user, host, ok := strings.Cut(addr, "@"); ok {
-		o.User, o.Addr = user, host
-	}
-	if o.User == "" {
-		// Under sudo, the person who ran it — their account is usually the
-		// one that exists on the nodes too. root's own keys still do the
-		// logging in, since that is who is running ssh.
-		o.User = os.Getenv("SUDO_USER")
-	}
 	for _, s := range na.fabrics {
 		fab, err := onboard.ParseFabric(s)
 		if err != nil {
@@ -158,32 +157,124 @@ func (na *nodeAdd) prepare(addr string) error {
 		o.Binary = exe
 	}
 	onboard.AgentUnit = agentUnit
+	na.named = map[string]string{}
 	return nil
 }
 
-func (na *nodeAdd) run(ctx context.Context, addr string, out io.Writer) error {
-	if err := na.prepare(addr); err != nil {
+// target is the options for one address, [user@]host.
+func (na *nodeAdd) target(addr string) onboard.Options {
+	o := na.opts
+	o.Fabrics = slices.Clone(na.opts.Fabrics)
+	o.Addr = addr
+	if user, host, ok := strings.Cut(addr, "@"); ok {
+		o.User, o.Addr = user, host
+	}
+	if o.User == "" {
+		// Under sudo, the person who ran it — their account is usually the
+		// one that exists on the nodes too. root's own keys still do the
+		// logging in, since that is who is running ssh.
+		o.User = os.Getenv("SUDO_USER")
+	}
+	return o
+}
+
+// rememberFabrics wraps the question of which fabric a node's adapters are
+// on, so a name given for one node is used for the next whose adapters
+// report the same fabric, without asking again. ask nil is a script, which
+// is asked nothing and gets only what an earlier answer said.
+func (na *nodeAdd) rememberFabrics(ask func([]onboard.Fabric) []onboard.Fabric) func([]onboard.Fabric) []onboard.Fabric {
+	return func(s []onboard.Fabric) []onboard.Fabric {
+		var known []onboard.Fabric
+		all := true
+		for i, f := range s {
+			if name, ok := na.named[f.Key]; ok && f.Key != "" && f.Key != "unknown" {
+				f.Name = name
+				s[i].Name = name
+				known = append(known, f)
+			} else {
+				all = false
+			}
+		}
+		if all || ask == nil {
+			return known
+		}
+		got := ask(s)
+		for _, f := range got {
+			if f.Key != "" && f.Key != "unknown" {
+				na.named[f.Key] = f.Name
+			}
+		}
+		return got
+	}
+}
+
+func (na *nodeAdd) run(ctx context.Context, addrs []string, out io.Writer) error {
+	if len(addrs) > 1 && na.opts.Node != "" {
+		return errors.New("--node names one node; give one address with it")
+	}
+	if err := na.prepare(); err != nil {
 		return err
 	}
-	if !na.plain && isTerminal(out) {
-		return na.runTUI(ctx)
+	tui := !na.plain && isTerminal(out)
+	type outcome struct {
+		addr, node string
+		err        error
 	}
+	var results []outcome
+	for i, addr := range addrs {
+		o := na.target(addr)
+		title := "llmgw · adding " + o.Addr
+		if len(addrs) > 1 {
+			title += fmt.Sprintf(" (%d of %d)", i+1, len(addrs))
+		}
+		var node string
+		var err error
+		if tui {
+			node, err = na.runTUI(ctx, o, title)
+		} else {
+			node, err = na.runPlain(ctx, o, title, out)
+		}
+		results = append(results, outcome{addr, node, err})
+		if ctx.Err() != nil {
+			break
+		}
+	}
+	if len(addrs) == 1 {
+		return results[0].err
+	}
+	fmt.Fprintln(out)
+	failed := 0
+	for _, r := range results {
+		if r.err != nil {
+			failed++
+			fmt.Fprintf(out, "  FAIL  %-20s %s\n", r.addr, r.err)
+		} else {
+			fmt.Fprintf(out, "  ok    %-20s %s\n", r.addr, r.node)
+		}
+	}
+	if failed > 0 {
+		return fmt.Errorf("%d of %d not added; running the same command again finishes them and leaves the rest as they are", failed, len(addrs))
+	}
+	return nil
+}
+
+func (na *nodeAdd) runPlain(ctx context.Context, o onboard.Options, title string, out io.Writer) (string, error) {
 	p := &plainProgress{out: out}
-	fmt.Fprintf(out, "adding %s\n", na.opts.Addr)
-	r := onboard.New(na.opts, p.event)
-	res, err := r.Run(ctx)
+	fmt.Fprintf(out, "%s\n", title)
+	o.AskFabrics = na.rememberFabrics(nil)
+	res, err := onboard.New(o, p.event).Run(ctx)
 	if err != nil {
-		return err
+		return "", err
 	}
 	if na.noWait {
-		fmt.Fprintf(out, "\n%s is added. 'llmgw ctl nodes' shows it once it joins.\n", res.Node)
-		return nil
+		fmt.Fprintf(out, "\n%s is added. 'llmgw ctl nodes' shows it once it joins.\n\n", res.Node)
+		return res.Node, nil
 	}
 	if err := na.watch(ctx, res.Node, p.event); err != nil {
-		return err
+		return res.Node, err
 	}
-	fmt.Fprintf(out, "\n%s is in the cluster.\n", res.Node)
-	return nil
+	fmt.Fprintf(out, "\n%s is in the cluster.\n\n", res.Node)
+	return res.Node, nil
 }
 
 // adminAPI is the control plane's admin API, from its configuration when

internal/cli/server.go

@@ -21,11 +21,13 @@ import (
 	"golang.org/x/net/http2/h2c"
 	"golang.org/x/sync/errgroup"
 
+	llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/config"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/fleetmetrics"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
@@ -493,6 +495,18 @@ func (r *dispatcherRef) Logs(ctx context.Context, nodeID, source string, lines i
 	return lr.Logs(ctx, nodeID, source, lines)
 }
 
+func (r *dispatcherRef) FabricTest(ctx context.Context, nodeID string, req *llmgwv1.FabricTestRequest) (*llmgwv1.FabricTestReply, error) {
+	d, err := r.get()
+	if err != nil {
+		return nil, err
+	}
+	ft, ok := d.(fabrictest.Asker)
+	if !ok {
+		return nil, errors.New("this dispatcher cannot run a bandwidth test")
+	}
+	return ft.FabricTest(ctx, nodeID, req)
+}
+
 func (r *dispatcherRef) StartShard(ctx context.Context, cmd reconciler.ShardCommand) error {
 	d, err := r.get()
 	if err != nil {

internal/cli/serverinit.go

@@ -73,6 +73,8 @@ type serverInit struct {
 	// make it serve something: a mirror, this machine as a node, and a small
 	// model on it.
 	noMirror, noNode, noModel bool
+	// mirrorURL is a mirror on another machine to use instead of running one.
+	mirrorURL string
 	// pgSetup prepares PostgreSQL and returns the store address; a seam for
 	// tests. Nil does it for real.
 	pgSetup func(context.Context) (string, error)
@@ -92,6 +94,7 @@ func newServerInitCmd() *cobra.Command {
 		unitPath: "/etc/systemd/system/llmgw-server.service",
 		user:     "llmgw",
 	}
+	var plain bool
 	cmd := &cobra.Command{
 		Use:   "init",
 		Short: "Set this machine up as a control plane and start it",
@@ -122,15 +125,21 @@ func newServerInitCmd() *cobra.Command {
 				return errors.New("init creates a user, writes /etc and starts a service: run it as root")
 			}
 			in.out = cmd.OutOrStdout()
+			if !plain && isTerminal(in.out) {
+				return in.runInitTUI(cmd.Context(), in.out)
+			}
 			return in.run(cmd.Context())
 		},
 	}
 	f := cmd.Flags()
+	f.BoolVar(&plain, "plain", false, "plain output, even in a terminal")
 	f.StringArrayVar(&in.hosts, "host", nil,
 		"an address agents dial besides the ones found on this machine; repeatable")
 	f.StringVar(&in.admin, "admin", "", "email of the first administrator; a password is generated and printed once")
 	f.StringVar(&in.pkiDir, "pki", in.pkiDir, "directory holding the certificate authority")
 	f.BoolVar(&in.noMirror, "no-mirror", false, "do not run a model mirror on this machine")
+	f.StringVar(&in.mirrorURL, "mirror", "",
+		"use the mirror on another machine, set up there by 'llmgw mirror init': its management API, http://host:8082")
 	f.BoolVar(&in.noNode, "no-node", false, "do not make this machine a node (and serve no model on it)")
 	f.BoolVar(&in.noModel, "no-model", false, "do not fetch and serve the starter model ("+starterName+")")
 	f.BoolVar(&in.postgres, "postgres", false,
@@ -163,7 +172,12 @@ func (in *serverInit) run(ctx context.Context) error {
 		hasSystemd = false
 	}
 	// Before the service starts, because the mirror goes into server.yaml.
-	if hasSystemd && !in.noMirror {
+	switch {
+	case in.mirrorURL != "":
+		if err := in.useMirror(ctx); err != nil {
+			return fmt.Errorf("mirror: %w", err)
+		}
+	case hasSystemd && !in.noMirror:
 		if err := in.ensureMirror(ctx); err != nil {
 			return fmt.Errorf("mirror: %w", err)
 		}
@@ -175,14 +189,14 @@ func (in *serverInit) run(ctx context.Context) error {
 		if _, err := in.ensureNode(ctx); err != nil {
 			return fmt.Errorf("node: %w", err)
 		}
-		switch {
-		case in.noModel:
-		case in.noMirror:
-			fmt.Fprintln(in.out, "  skip  no starter model without the mirror")
-		default:
-			if err := in.ensureStarter(ctx); err != nil {
-				return fmt.Errorf("starter model: %w", err)
-			}
+	}
+	switch {
+	case !hasSystemd || in.noModel:
+	case in.noMirror && in.mirrorURL == "":
+		fmt.Fprintln(in.out, "  skip  no starter model without the mirror")
+	default:
+		if err := in.ensureStarter(ctx); err != nil {
+			return fmt.Errorf("starter model: %w", err)
 		}
 	}
 	in.nextSteps()

internal/cli/serverinit_cluster.go

@@ -111,15 +111,25 @@ func (in *serverInit) writeUnit(path, body string) (bool, error) {
 
 func (in *serverInit) mirrorUnit() string {
 	base := "http://" + net.JoinHostPort(in.reachableAddr(), "8081")
+	return mirrorUnitFor(mirrorDir, mirrorAPI, base, nil,
+		"'llmgw server init'. Bytes on :8081 for every node; the\n# management API on loopback only, for the control plane beside it.")
+}
+
+// mirrorUnitFor is a mirror's systemd unit. allow is who, besides the
+// machine itself, may use the management API.
+func mirrorUnitFor(dir, apiListen, base string, allow []string, about string) string {
+	exec := "/usr/local/bin/llmgw mirror serve --dir " + dir + " --listen :8081 --api-listen " + apiListen + " --base " + base
+	for _, a := range allow {
+		exec += " --api-allow " + a
+	}
 	return `[Unit]
 Description=llmgw model mirror
 After=network-online.target
 Wants=network-online.target
 
 [Service]
-# Written by 'llmgw server init'. Bytes on :8081 for every node; the
-# management API on loopback only, for the control plane beside it.
-ExecStart=/usr/local/bin/llmgw mirror serve --dir ` + mirrorDir + ` --listen :8081 --api-listen ` + mirrorAPI + ` --base ` + base + `
+# Written by ` + about + `
+ExecStart=` + exec + `
 User=llmgw
 Group=llmgw
 Restart=on-failure
@@ -128,13 +138,53 @@ NoNewPrivileges=true
 ProtectSystem=strict
 ProtectHome=true
 PrivateTmp=true
-ReadWritePaths=` + mirrorDir + `
+ReadWritePaths=` + dir + `
 
 [Install]
 WantedBy=multi-user.target
 `
 }
 
+// mirrorAPIURL is the mirror's management API: one on its own machine, or
+// the one beside this control plane.
+func (in *serverInit) mirrorAPIURL() string {
+	if in.mirrorURL != "" {
+		return in.mirrorURL
+	}
+	return "http://" + mirrorAPI
+}
+
+// useMirror points the control plane at a mirror on another machine, set up
+// there by 'llmgw mirror init'.
+func (in *serverInit) useMirror(ctx context.Context) error {
+	m, err := mirror.NewClient(in.mirrorURL, tlsconf.Config{}, 5*time.Second)
+	if err != nil {
+		return err
+	}
+	if err := m.Health(ctx); err != nil {
+		return fmt.Errorf("the mirror at %s does not answer: %v. On that machine: sudo llmgw mirror init --control-plane <this machine's address>", in.mirrorURL, err)
+	}
+	cfg, err := config.Load(in.configPath())
+	if err != nil {
+		return err
+	}
+	switch cfg.Mirror.API {
+	case in.mirrorURL:
+		in.kept("mirror at %s", in.mirrorURL)
+		return nil
+	case "":
+	default:
+		in.kept("mirror at %s, as server.yaml says; change it there to use %s", cfg.Mirror.API, in.mirrorURL)
+		return nil
+	}
+	if err := appendMirrorBlockFor(in.configPath(), in.mirrorURL,
+		"# The model mirror, on its own machine, added by 'llmgw server init'."); err != nil {
+		return err
+	}
+	in.done("server.yaml points at the mirror at %s", in.mirrorURL)
+	return nil
+}
+
 // ensureMirror runs a mirror on this machine and points the control plane at
 // it. Its management API is on loopback, which is why it needs no certificate:
 // nothing but this machine can reach it.
@@ -174,7 +224,9 @@ func (in *serverInit) ensureMirror(ctx context.Context) error {
 	}
 
 	if cfg.Mirror.API == "" {
-		if err := appendMirrorBlock(in.configPath()); err != nil {
+		if err := appendMirrorBlockFor(in.configPath(), "http://"+mirrorAPI,
+			"# The model mirror on this machine, added by 'llmgw server init'. Its\n"+
+				"# management API is on loopback, so it needs no certificate."); err != nil {
 			return err
 		}
 		in.done("server.yaml points at the mirror")
@@ -189,9 +241,9 @@ func (in *serverInit) ensureMirror(ctx context.Context) error {
 	})
 }
 
-// appendMirrorBlock adds the mirror to server.yaml without touching the rest,
-// keeping the file as it was beside it.
-func appendMirrorBlock(path string) error {
+// appendMirrorBlockFor adds the mirror to server.yaml without touching the
+// rest, keeping the file as it was beside it.
+func appendMirrorBlockFor(path, api, comment string) error {
 	b, err := os.ReadFile(path)
 	if err != nil {
 		return err
@@ -200,9 +252,7 @@ func appendMirrorBlock(path string) error {
 	if err := os.WriteFile(backup, b, 0o640); err != nil {
 		return err
 	}
-	block := "\n# The model mirror on this machine, added by 'llmgw server init'. Its\n" +
-		"# management API is on loopback, so it needs no certificate.\n" +
-		"mirror:\n  api: http://" + mirrorAPI + "\n"
+	block := "\n" + comment + "\nmirror:\n  api: " + api + "\n"
 	info, err := os.Stat(path)
 	if err != nil {
 		return err
@@ -356,8 +406,11 @@ func hasNVIDIADevice() bool {
 
 // starterModel is the catalog entry: a GPU profile first when there is a card,
 // and a CPU profile always, both pinned to this machine.
-func starterModel(source string, gpuClasses []string) catalog.Model {
-	pin := map[string]string{localLabelKey: localLabelValue}
+func starterModel(source string, gpuClasses []string, pinned bool) catalog.Model {
+	var pin map[string]string
+	if pinned {
+		pin = map[string]string{localLabelKey: localLabelValue}
+	}
 	var profiles []catalog.Profile
 	for _, class := range gpuClasses {
 		profiles = append(profiles, catalog.Profile{
@@ -401,7 +454,7 @@ func (in *serverInit) ensureStarter(ctx context.Context) error {
 	if !ok {
 		return fmt.Errorf("cannot turn %s into a download", starterSource)
 	}
-	m, err := mirror.NewClient("http://"+mirrorAPI, tlsconf.Config{}, 15*time.Second)
+	m, err := mirror.NewClient(in.mirrorAPIURL(), tlsconf.Config{}, 15*time.Second)
 	if err != nil {
 		return err
 	}
@@ -444,14 +497,24 @@ func (in *serverInit) ensureStarter(ctx context.Context) error {
 	}
 	in.done("%s in the mirror", starterName)
 
-	classes, _ := gpuClasses(ctx)
-	body, err := json.Marshal(starterModel(st.URL, classes))
+	// Served here when this machine is a node, pinned so it does not wander
+	// to the first GPU node that joins; otherwise on whichever node joins.
+	pinned := !in.noNode
+	var classes []string
+	if pinned {
+		classes, _ = gpuClasses(ctx)
+	}
+	body, err := json.Marshal(starterModel(st.URL, classes, pinned))
 	if err != nil {
 		return err
 	}
 	if err := client.post(ctx, "/api/v1/models", body, &map[string]string{}); err != nil {
 		return fmt.Errorf("adding %s to the catalog: %w", starterName, err)
 	}
+	if !pinned {
+		in.done("%s in the catalog; it starts on the first node that joins", starterName)
+		return nil
+	}
 	where := "on CPU"
 	if len(classes) > 0 {
 		where = "on " + strings.Join(classes, ", ") + ", or CPU if the card is busy"

internal/cli/serverinit_cluster_test.go

@@ -1,6 +1,8 @@
 package cli
 
 import (
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"strings"
@@ -21,7 +23,7 @@ func TestTheInitAgentUnitIsTheShippedUnit(t *testing.T) {
 }
 
 func TestTheStarterUsesTheGPUAndFallsBackToCPUOnThisMachine(t *testing.T) {
-	m := starterModel("http://10.0.0.5:8081/qwen2.5-0.5b-instruct/starter/q.gguf", []string{"4090-24g"})
+	m := starterModel("http://10.0.0.5:8081/qwen2.5-0.5b-instruct/starter/q.gguf", []string{"4090-24g"}, true)
 	c, err := catalog.New([]catalog.Model{m})
 	if err != nil {
 		t.Fatalf("the starter is not a valid catalog entry: %v", err)
@@ -46,7 +48,7 @@ func TestTheStarterUsesTheGPUAndFallsBackToCPUOnThisMachine(t *testing.T) {
 		t.Error("the starter should be running, not waiting to be asked for")
 	}
 
-	if m := starterModel("http://x/q.gguf", nil); len(m.Profiles) != 1 || m.Profiles[0].Name != "cpu" {
+	if m := starterModel("http://x/q.gguf", nil, true); len(m.Profiles) != 1 || m.Profiles[0].Name != "cpu" {
 		t.Errorf("without a GPU only the CPU profile: %+v", m.Profiles)
 	}
 }
@@ -56,7 +58,7 @@ func TestTheMirrorBlockIsAddedAndReads(t *testing.T) {
 	if err := in.run(t.Context()); err != nil {
 		t.Fatalf("%v\n%s", err, out)
 	}
-	if err := appendMirrorBlock(in.configPath()); err != nil {
+	if err := appendMirrorBlockFor(in.configPath(), "http://"+mirrorAPI, "# the mirror"); err != nil {
 		t.Fatal(err)
 	}
 	cfg, err := config.Load(in.configPath())
@@ -87,3 +89,35 @@ func TestTheMirrorUnitAdvertisesAnAddressNodesCanReach(t *testing.T) {
 		t.Errorf("the management API is not on loopback: %s", u)
 	}
 }
+
+func TestWithNoNodeHereTheStarterGoesToTheFirstNodeThatJoins(t *testing.T) {
+	m := starterModel("http://x/q.gguf", nil, false)
+	for _, p := range m.Profiles {
+		if len(p.NodeSelector) > 0 {
+			t.Errorf("profile %s is pinned to a machine that is not a node: %v", p.Name, p.NodeSelector)
+		}
+	}
+}
+
+func TestTheMirrorAPIOnlyAnswersItsControlPlane(t *testing.T) {
+	allowed, err := parseAllow([]string{"10.0.0.140", "10.1.0.0/24"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	h := onlyFrom(allowed, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {}))
+	for addr, want := range map[string]int{
+		"10.0.0.140:5000": 200, "10.1.0.7:5000": 200, "127.0.0.1:5000": 200, "[::1]:5000": 200,
+		"10.0.0.141:5000": 403, "192.168.1.9:5000": 403,
+	} {
+		r := httptest.NewRequest("GET", "/api/v1/health", nil)
+		r.RemoteAddr = addr
+		w := httptest.NewRecorder()
+		h.ServeHTTP(w, r)
+		if w.Code != want {
+			t.Errorf("%s: %d, want %d", addr, w.Code, want)
+		}
+	}
+	if _, err := parseAllow([]string{"not-an-address"}); err == nil {
+		t.Error("nonsense was accepted")
+	}
+}

internal/cli/tui_init.go

@@ -0,0 +1,403 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/mail"
+	"strings"
+	"sync"
+
+	"github.com/charmbracelet/bubbles/spinner"
+	"github.com/charmbracelet/bubbles/textinput"
+	tea "github.com/charmbracelet/bubbletea"
+	"github.com/charmbracelet/lipgloss"
+)
+
+// The full-screen way through 'llmgw server init': a short form for what
+// init would otherwise take as flags, then init's own lines as a list that
+// fills in, with the administrator's password kept in view — and printed
+// again once the screen closes, so it is in the terminal's history.
+
+// initLineMsg is one line init wrote.
+type initLineMsg string
+
+// initDoneMsg is init finishing.
+type initDoneMsg struct{ err error }
+
+// lineWriter turns init's output into one message per line.
+type lineWriter struct {
+	mu   sync.Mutex
+	buf  []byte
+	send func(string)
+}
+
+func (w *lineWriter) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.buf = append(w.buf, p...)
+	for {
+		i := strings.IndexByte(string(w.buf), '\n')
+		if i < 0 {
+			return len(p), nil
+		}
+		w.send(string(w.buf[:i]))
+		w.buf = w.buf[i+1:]
+	}
+}
+
+// Form fields, in order.
+const (
+	fAdmin = iota
+	fHosts
+	fPostgres
+	fMirror
+	fNode
+	fModel
+	fStart
+	fCount
+)
+
+type initModel struct {
+	in    *serverInit
+	found []string // this machine's addresses
+
+	// The form, until it is submitted.
+	form      bool
+	focus     int
+	admin     textinput.Model
+	hosts     textinput.Model
+	toggles   [fCount]bool
+	formError string
+
+	// Then the run.
+	running  bool
+	lines    []string // init's lines, as it wrote them
+	password string
+	next     []string // the closing "Next:" text
+	spin     spinner.Model
+	width    int
+	done     bool
+	err      error
+	quit     bool
+	start    func()
+}
+
+func newInitModel(in *serverInit, found []string, askForm bool) initModel {
+	admin := textinput.New()
+	admin.Placeholder = "you@example.com"
+	admin.SetValue(in.admin)
+	admin.CharLimit = 200
+	admin.Width = 40
+	hosts := textinput.New()
+	hosts.Placeholder = "llm.example.com, 203.0.113.10"
+	hosts.SetValue(strings.Join(in.hosts, ", "))
+	hosts.CharLimit = 400
+	hosts.Width = 40
+	sp := spinner.New()
+	sp.Spinner = spinner.MiniDot
+	sp.Style = lipgloss.NewStyle().Foreground(tuiAccent)
+	m := initModel{in: in, found: found, form: askForm, admin: admin, hosts: hosts, spin: sp, width: 100}
+	m.toggles[fPostgres] = in.postgres
+	m.toggles[fMirror] = !in.noMirror
+	m.toggles[fNode] = !in.noNode
+	m.toggles[fModel] = !in.noModel
+	if askForm {
+		m.admin.Focus()
+	}
+	return m
+}
+
+func (m initModel) Init() tea.Cmd {
+	if m.form {
+		return textinput.Blink
+	}
+	return tea.Batch(m.spin.Tick, func() tea.Msg { return startInitMsg{} })
+}
+
+type startInitMsg struct{}
+
+func (m initModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
+	switch msg := msg.(type) {
+	case tea.WindowSizeMsg:
+		m.width = msg.Width
+	case spinner.TickMsg:
+		var cmd tea.Cmd
+		m.spin, cmd = m.spin.Update(msg)
+		return m, cmd
+	case startInitMsg:
+		m.running = true
+		if m.start != nil {
+			m.start()
+		}
+	case initLineMsg:
+		m.addLine(string(msg))
+	case initDoneMsg:
+		m.done, m.err = true, msg.err
+		return m, tea.Quit
+	case tea.KeyMsg:
+		if msg.Type == tea.KeyCtrlC {
+			m.quit = true
+			return m, tea.Quit
+		}
+		if m.form {
+			return m.formKey(msg)
+		}
+	}
+	return m, nil
+}
+
+// addLine files one of init's lines: the password aside, the closing text
+// aside, a progress line replacing the one before it.
+func (m *initModel) addLine(l string) {
+	t := strings.TrimSpace(l)
+	switch {
+	case t == "":
+		return
+	case strings.HasPrefix(t, "password: "):
+		m.password = strings.TrimPrefix(t, "password: ")
+		return
+	case strings.HasPrefix(t, "shown once"):
+		return
+	case !strings.HasPrefix(l, "  ") || len(m.next) > 0:
+		// Unindented lines are the closing text, and everything after its
+		// first line belongs to it.
+		m.next = append(m.next, l)
+		return
+	}
+	if strings.HasPrefix(t, "..") && len(m.lines) > 0 && strings.HasPrefix(strings.TrimSpace(m.lines[len(m.lines)-1]), "..") {
+		m.lines[len(m.lines)-1] = l
+		return
+	}
+	m.lines = append(m.lines, l)
+}
+
+func (m initModel) formKey(k tea.KeyMsg) (tea.Model, tea.Cmd) {
+	switch k.Type {
+	case tea.KeyTab, tea.KeyDown:
+		m.moveFocus(1)
+		return m, nil
+	case tea.KeyShiftTab, tea.KeyUp:
+		m.moveFocus(-1)
+		return m, nil
+	case tea.KeyEnter:
+		if m.focus != fStart {
+			m.moveFocus(1)
+			return m, nil
+		}
+		admin := strings.TrimSpace(m.admin.Value())
+		if admin != "" {
+			if _, err := mail.ParseAddress(admin); err != nil {
+				m.formError = "the administrator's email is not an address"
+				m.focus = fAdmin
+				m.admin.Focus()
+				return m, nil
+			}
+		}
+		m.in.admin = admin
+		m.in.hosts = nil
+		for _, h := range strings.Split(m.hosts.Value(), ",") {
+			if h = strings.TrimSpace(h); h != "" {
+				m.in.hosts = append(m.in.hosts, h)
+			}
+		}
+		m.in.postgres = m.toggles[fPostgres]
+		m.in.noMirror = !m.toggles[fMirror]
+		m.in.noNode = !m.toggles[fNode]
+		m.in.noModel = !m.toggles[fModel]
+		m.form = false
+		return m, tea.Batch(m.spin.Tick, func() tea.Msg { return startInitMsg{} })
+	}
+	switch m.focus {
+	case fAdmin:
+		var cmd tea.Cmd
+		m.admin, cmd = m.admin.Update(k)
+		return m, cmd
+	case fHosts:
+		var cmd tea.Cmd
+		m.hosts, cmd = m.hosts.Update(k)
+		return m, cmd
+	case fPostgres, fMirror, fNode, fModel:
+		if k.String() == " " || k.String() == "x" {
+			m.toggles[m.focus] = !m.toggles[m.focus]
+		}
+	}
+	return m, nil
+}
+
+func (m *initModel) moveFocus(d int) {
+	m.focus = (m.focus + d + fCount) % fCount
+	// With --mirror there is no mirror to run here, so nothing to tick.
+	if m.focus == fMirror && m.in.mirrorURL != "" {
+		m.focus = (m.focus + d + fCount) % fCount
+	}
+	m.admin.Blur()
+	m.hosts.Blur()
+	switch m.focus {
+	case fAdmin:
+		m.admin.Focus()
+	case fHosts:
+		m.hosts.Focus()
+	}
+}
+
+func (m initModel) View() string {
+	var b strings.Builder
+	b.WriteString(sTitle.Render("llmgw · setting up this machine as a control plane") + "\n\n")
+	if m.form {
+		return b.String() + m.formView()
+	}
+	for _, l := range m.visibleLines() {
+		b.WriteString(m.lineView(l) + "\n")
+	}
+	if m.password != "" {
+		box := sBox.BorderForeground(tuiWarn).Render(
+			fmt.Sprintf("Administrator %s\npassword  %s\n%s", m.in.admin,
+				lipgloss.NewStyle().Bold(true).Render(m.password),
+				sDim.Render("Shown once. Copy it now; change it on the account page after signing in.")))
+		b.WriteString("\n" + box + "\n")
+	}
+	switch {
+	case m.done && m.err != nil:
+		b.WriteString("\n" + sBad.Render("✗ ") + wrap(m.err.Error(), m.width-4, "  ") + "\n" +
+			sDim.Render("  It is safe to run init again once that is fixed: it finishes what is missing.") + "\n")
+	case m.done:
+		b.WriteString("\n" + sOK.Render("✓ ") + "The control plane is up. Add a node: " +
+			lipgloss.NewStyle().Bold(true).Render("sudo llmgw node add <user>@<address>") + "\n")
+		b.WriteString("  " + sDim.Render("Back up "+m.in.pkiDir+": its ca-key.pem signs every node's identity, and nothing recovers it.") + "\n")
+	case m.quit:
+		b.WriteString("\n" + sDim.Render("stopped; run init again to finish") + "\n")
+	case m.running:
+		b.WriteString("\n" + sDim.Render("ctrl+c to stop") + "\n")
+	}
+	return b.String()
+}
+
+// visibleLines are the last lines, enough to follow without the screen
+// outgrowing the terminal: the whole log is printed when it closes.
+func (m initModel) visibleLines() []string {
+	const keep = 16
+	if len(m.lines) > keep {
+		return m.lines[len(m.lines)-keep:]
+	}
+	return m.lines
+}
+
+func (m initModel) lineView(l string) string {
+	t := strings.TrimSpace(l)
+	word, rest, _ := strings.Cut(t, " ")
+	rest = truncate(strings.TrimSpace(rest), max(20, m.width-6))
+	switch word {
+	case "ok":
+		return "  " + sOK.Render("✓") + " " + rest
+	case "kept":
+		return "  " + sDim.Render("· "+rest)
+	case "skip":
+		return "  " + sDim.Render("- "+rest)
+	case "!!":
+		return "  " + sWarn.Render("! ") + rest
+	case "..":
+		if m.done {
+			return "  " + sDim.Render("· "+rest)
+		}
+		return "  " + m.spin.View() + " " + rest
+	}
+	return "    " + t
+}
+
+func (m initModel) formView() string {
+	var b strings.Builder
+	label := func(i int, s string) string {
+		if m.focus == i {
+			return sTitle.Render("› " + s)
+		}
+		return "  " + s
+	}
+	check := func(i int, s, hint string) string {
+		box := "[ ]"
+		if m.toggles[i] {
+			box = "[" + sOK.Render("x") + "]"
+		}
+		return fmt.Sprintf("%s %s %s\n", label(i, box), s, sDim.Render(hint))
+	}
+	fmt.Fprintf(&b, "%s\n    %s\n", label(fAdmin, "Administrator's email"), m.admin.View())
+	b.WriteString("    " + sDim.Render("A password is generated and shown once. Leave empty to create accounts later.") + "\n\n")
+	fmt.Fprintf(&b, "%s\n    %s\n", label(fHosts, "Other names nodes will reach this machine by"), m.hosts.View())
+	found := "none found"
+	if len(m.found) > 0 {
+		found = strings.Join(m.found, ", ")
+	}
+	b.WriteString("    " + sDim.Render("Already in the certificate: "+found+". Add a DNS name or NAT address here.") + "\n\n")
+	b.WriteString(check(fPostgres, "PostgreSQL", "rather than SQLite; installed here, for several control planes or a larger store"))
+	if m.in.mirrorURL != "" {
+		b.WriteString("  Model mirror " + sDim.Render("at "+m.in.mirrorURL+", from --mirror") + "\n")
+	} else {
+		b.WriteString(check(fMirror, "Model mirror", "downloads model files once for every node"))
+	}
+	b.WriteString(check(fNode, "This machine as a node", "serves models here too, on its GPUs or CPU"))
+	b.WriteString(check(fModel, "Starter model", "a small Qwen, so there is something to ask at once"))
+	b.WriteString("\n" + label(fStart, lipgloss.NewStyle().Bold(true).Render("Set up")) + "\n")
+	if m.formError != "" {
+		b.WriteString("\n" + sBad.Render(m.formError) + "\n")
+	}
+	b.WriteString("\n" + sDim.Render("tab or ↑↓ to move · space to tick · enter on Set up to start · ctrl+c to leave") + "\n")
+	return b.String()
+}
+
+// runInitTUI runs init with the screen above, asking the form first when
+// the flags left the administrator unset.
+func (in *serverInit) runInitTUI(ctx context.Context, out io.Writer) error {
+	var found []string
+	for _, ip := range localAddrs() {
+		found = append(found, ip.String())
+	}
+	m := newInitModel(in, found, in.admin == "")
+	var p *tea.Program
+
+	var (
+		logMu sync.Mutex
+		log   []string
+	)
+	w := &lineWriter{send: func(l string) {
+		logMu.Lock()
+		log = append(log, l)
+		logMu.Unlock()
+		p.Send(initLineMsg(l))
+	}}
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	started := make(chan struct{})
+	m.start = func() {
+		close(started)
+		go func() {
+			in.out = w
+			err := in.run(ctx)
+			p.Send(initDoneMsg{err})
+		}()
+	}
+	p = tea.NewProgram(m, tea.WithOutput(out))
+	final, err := p.Run()
+	if err != nil {
+		return err
+	}
+	fm := final.(initModel)
+	select {
+	case <-started:
+	default:
+		return fmt.Errorf("left before anything was done")
+	}
+	if fm.quit && !fm.done {
+		cancel()
+		return fmt.Errorf("stopped; run init again to finish what was started")
+	}
+
+	// The screen showed the last lines; the whole run goes to the terminal's
+	// history, the password with it.
+	logMu.Lock()
+	defer logMu.Unlock()
+	fmt.Fprintln(out)
+	for _, l := range log {
+		fmt.Fprintln(out, l)
+	}
+	return fm.err
+}

internal/cli/tui_init_test.go

@@ -0,0 +1,103 @@
+package cli
+
+import (
+	"strings"
+	"testing"
+
+	tea "github.com/charmbracelet/bubbletea"
+)
+
+func typeIn(m tea.Model, s string) tea.Model {
+	return send(m, tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune(s)})
+}
+
+func TestTheInitFormBecomesInitsSettings(t *testing.T) {
+	in := &serverInit{pkiDir: "/etc/llmgw/pki"}
+	var m tea.Model = newInitModel(in, []string{"10.0.0.140"}, true)
+	if v := m.View(); !strings.Contains(v, "Already in the certificate: 10.0.0.140") {
+		t.Fatalf("the addresses found are not shown:\n%s", v)
+	}
+	m = typeIn(m, "ops@example.com")
+	m = send(m, tea.KeyMsg{Type: tea.KeyTab})
+	m = typeIn(m, "llm.example.com, 203.0.113.10")
+	m = send(m, tea.KeyMsg{Type: tea.KeyTab}, tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune(" ")}) // PostgreSQL on
+	m = send(m, tea.KeyMsg{Type: tea.KeyTab}, tea.KeyMsg{Type: tea.KeyTab}, tea.KeyMsg{Type: tea.KeyTab},
+		tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune(" ")}) // starter model off
+	m = send(m, tea.KeyMsg{Type: tea.KeyTab}, tea.KeyMsg{Type: tea.KeyEnter})
+
+	if in.admin != "ops@example.com" || strings.Join(in.hosts, "|") != "llm.example.com|203.0.113.10" {
+		t.Errorf("admin %q hosts %q", in.admin, in.hosts)
+	}
+	if !in.postgres || in.noMirror || in.noNode || !in.noModel {
+		t.Errorf("toggles: postgres %v mirror %v node %v model %v", in.postgres, !in.noMirror, !in.noNode, !in.noModel)
+	}
+	if m.(initModel).form {
+		t.Error("the form stayed up after Set up")
+	}
+}
+
+func TestAnEmailThatIsNotOneIsSaidSo(t *testing.T) {
+	in := &serverInit{}
+	var m tea.Model = newInitModel(in, nil, true)
+	m = typeIn(m, "not an address")
+	for range fStart {
+		m = send(m, tea.KeyMsg{Type: tea.KeyTab})
+	}
+	m = send(m, tea.KeyMsg{Type: tea.KeyEnter})
+	if !m.(initModel).form || !strings.Contains(m.View(), "not an address") {
+		t.Errorf("a bad email was accepted:\n%s", m.View())
+	}
+}
+
+func TestInitsLinesAreFiledAndThePasswordKeptInView(t *testing.T) {
+	in := &serverInit{admin: "ops@example.com", pkiDir: "/etc/llmgw/pki"}
+	var m tea.Model = newInitModel(in, nil, false)
+	for _, l := range []string{
+		"  ok    system user llmgw",
+		"  kept  certificate authority in /etc/llmgw/pki",
+		"  ok    administrator ops@example.com (u1)",
+		"",
+		"        password: Tr0ub4dor-3",
+		"        shown once — sign in and change it from /profile",
+		"",
+		"  ..    downloading qwen2.5-0.5b-instruct into the mirror",
+		"  ..    40%",
+		"  ..    80%",
+		"  !!    an NVIDIA card is on the PCI bus but nvidia-smi does not answer",
+		"The control plane is up. Next:",
+		"  llmgw ctl status",
+	} {
+		m = send(m, initLineMsg(l))
+	}
+	m = send(m, initDoneMsg{})
+	im := m.(initModel)
+	if im.password != "Tr0ub4dor-3" {
+		t.Errorf("password %q", im.password)
+	}
+	v := m.View()
+	for _, want := range []string{"✓ system user llmgw", "Tr0ub4dor-3", "Shown once", "80%", "! an NVIDIA card", "Back up /etc/llmgw/pki", "node add"} {
+		if !strings.Contains(v, want) {
+			t.Errorf("screen lacks %q:\n%s", want, v)
+		}
+	}
+	if strings.Contains(v, "40%") {
+		t.Error("progress lines piled up instead of replacing each other")
+	}
+	if strings.Contains(v, "llmgw ctl status") {
+		t.Error("the closing text was listed as a step")
+	}
+}
+
+func TestWithAMirrorElsewhereTheFormDoesNotOfferOne(t *testing.T) {
+	in := &serverInit{mirrorURL: "http://10.0.0.150:8082"}
+	var m tea.Model = newInitModel(in, nil, true)
+	if v := m.View(); !strings.Contains(v, "at http://10.0.0.150:8082, from --mirror") {
+		t.Errorf("the mirror in use is not shown:\n%s", v)
+	}
+	// Tab from the addresses goes past the mirror line to PostgreSQL, then
+	// the node.
+	m = send(m, tea.KeyMsg{Type: tea.KeyTab}, tea.KeyMsg{Type: tea.KeyTab}, tea.KeyMsg{Type: tea.KeyTab})
+	if f := m.(initModel).focus; f != fNode {
+		t.Errorf("focus %d, want the node toggle %d", f, fNode)
+	}
+}

internal/cli/tui_node.go

@@ -176,7 +176,7 @@ func (m nodeModel) key(k tea.KeyMsg) (tea.Model, tea.Cmd) {
 			var out []onboard.Fabric
 			for i, in := range m.inputs {
 				if name := strings.TrimSpace(in.Value()); name != "" {
-					out = append(out, onboard.Fabric{Name: name, HCAs: m.fabrics.suggested[i].HCAs})
+					out = append(out, onboard.Fabric{Name: name, HCAs: m.fabrics.suggested[i].HCAs, Key: m.fabrics.suggested[i].Key})
 				}
 			}
 			m.fabrics.reply <- out
@@ -297,15 +297,15 @@ func wrap(s string, width int, indent string) string {
 	return strings.Join(lines, "\n"+indent)
 }
 
-// runTUI adds the node with the screen above.
-func (na *nodeAdd) runTUI(ctx context.Context) error {
+// runTUI adds one node with the screen above.
+func (na *nodeAdd) runTUI(ctx context.Context, o onboard.Options, title string) (string, error) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	m := newNodeModel("llmgw · adding "+na.opts.Addr, cancel)
+	m := newNodeModel(title, cancel)
 	p := tea.NewProgram(m)
 
 	emit := func(e onboard.Event) { p.Send(evMsg(e)) }
-	na.opts.AskFabrics = func(s []onboard.Fabric) []onboard.Fabric {
+	o.AskFabrics = na.rememberFabrics(func(s []onboard.Fabric) []onboard.Fabric {
 		reply := make(chan []onboard.Fabric)
 		p.Send(askFabricsMsg{s, reply})
 		select {
@@ -314,8 +314,8 @@ func (na *nodeAdd) runTUI(ctx context.Context) error {
 		case <-ctx.Done():
 			return nil
 		}
-	}
-	na.opts.AskContinue = func(fails int) bool {
+	})
+	o.AskContinue = func(fails int) bool {
 		reply := make(chan bool)
 		p.Send(askContinueMsg{fails, reply})
 		select {
@@ -327,7 +327,7 @@ func (na *nodeAdd) runTUI(ctx context.Context) error {
 	}
 
 	go func() {
-		res, err := onboard.New(na.opts, emit).Run(ctx)
+		res, err := onboard.New(o, emit).Run(ctx)
 		node := ""
 		if res != nil {
 			node = res.Node
@@ -344,11 +344,12 @@ func (na *nodeAdd) runTUI(ctx context.Context) error {
 
 	final, err := p.Run()
 	if err != nil {
-		return err
+		return "", err
 	}
 	fm := final.(nodeModel)
 	if fm.quitAsked && !fm.done {
-		return fmt.Errorf("stopped; the node may be half set up, and running this again finishes it")
+		cancel()
+		return fm.node, fmt.Errorf("stopped; the node may be half set up, and running this again finishes it")
 	}
-	return fm.err
+	return fm.node, fm.err
 }

internal/cli/tui_node_test.go

@@ -94,3 +94,24 @@ func TestANodeCountsAsJoinedOnlyWhenHeardSinceItsAgentStarted(t *testing.T) {
 		}
 	}
 }
+
+func TestAFabricNamedForOneNodeIsNotAskedAgainForTheNext(t *testing.T) {
+	na := &nodeAdd{named: map[string]string{}}
+	asked := 0
+	ask := func(s []onboard.Fabric) []onboard.Fabric {
+		asked++
+		return []onboard.Fabric{{Name: "ib-compute", HCAs: s[0].HCAs, Key: s[0].Key}}
+	}
+	first := na.rememberFabrics(ask)([]onboard.Fabric{{Name: "fabric", HCAs: []string{"mlx5_0"}, Key: "sm:0xa"}})
+	second := na.rememberFabrics(ask)([]onboard.Fabric{{Name: "fabric", HCAs: []string{"mlx5_0", "mlx5_1"}, Key: "sm:0xa"}})
+	if asked != 1 {
+		t.Errorf("asked %d times", asked)
+	}
+	if len(second) != 1 || second[0].Name != "ib-compute" || len(second[0].HCAs) != 2 || first[0].Name != "ib-compute" {
+		t.Errorf("first %+v second %+v", first, second)
+	}
+	// A fabric not seen before is asked about; in a script nothing is asked.
+	if got := na.rememberFabrics(nil)([]onboard.Fabric{{Name: "fabric", Key: "sm:0xb"}}); len(got) != 0 {
+		t.Errorf("a script was given a fabric nobody named: %+v", got)
+	}
+}

internal/control/adminapi/api.go

@@ -75,6 +75,7 @@ func (a *API) Register(mux *http.ServeMux) {
 	mux.HandleFunc("GET /api/v1/explain-placement", a.explainPlacement)
 	mux.HandleFunc("POST /api/v1/nodes/{id}/state", a.setNodeState)
 	mux.HandleFunc("DELETE /api/v1/nodes/{id}", a.removeNode)
+	mux.HandleFunc("POST /api/v1/fabric-test", a.fabricTest)
 	mux.HandleFunc("GET /api/v1/nodes/{id}/logs", a.nodeLogs)
 }
 
@@ -133,7 +134,10 @@ type NodeView struct {
 
 	LastSeenAgo  string `json:"last_seen_ago"`
 	AgentVersion string `json:"agent_version,omitempty"`
-	Replicas     int    `json:"replicas"`
+	// Features are the requests the agent answers beyond the usual:
+	// "fabric-test". Absent is an agent too old to say.
+	Features []string `json:"features,omitempty"`
+	Replicas int      `json:"replicas"`
 
 	// Load is how busy the machine is, absent when its agent has not
 	// measured it lately — which is unknown, not idle.
@@ -334,7 +338,7 @@ func (a *API) Nodes() []NodeView {
 			State: n.State.String(), Labels: n.Labels,
 			GPUTotal: len(n.GPUs), GPUFree: len(inv.FreeGPUs(n.ID)),
 			MemoryGB: n.MemoryTotalMB / 1024,
-			Fabrics:  n.FabricNames(), AgentVersion: n.AgentVersion,
+			Fabrics:  n.FabricNames(), AgentVersion: n.AgentVersion, Features: n.Features,
 			Replicas: replicasOn[n.ID],
 		}
 		if free, total := n.LocalDisk(); total > 0 {

internal/control/adminapi/fabrictest.go

@@ -0,0 +1,71 @@
+package adminapi
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"time"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest"
+)
+
+// FabricTestView is a bandwidth test between two nodes.
+type FabricTestView struct {
+	A      string     `json:"a"`
+	B      string     `json:"b"`
+	Fabric string     `json:"fabric"`
+	OK     bool       `json:"ok"`
+	Rails  []RailView `json:"rails"`
+	Notes  []string   `json:"notes,omitempty"`
+	Took   string     `json:"took"`
+}
+
+// RailView is one rail's result.
+type RailView struct {
+	A        string  `json:"a"`
+	B        string  `json:"b"`
+	RateGbps int     `json:"rate_gbps"`
+	Gbps     float64 `json:"gbps"`
+	Verdict  string  `json:"verdict"`
+	Detail   string  `json:"detail,omitempty"`
+}
+
+// fabricTest measures RDMA bandwidth between two nodes, rail by rail. It
+// sends traffic over the fabric for seconds a rail, so it is a POST and only
+// ever run when asked.
+func (a *API) fabricTest(w http.ResponseWriter, r *http.Request) {
+	var body struct {
+		A, B, Fabric string
+		Seconds      int
+	}
+	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+	ask := a.ctl.FabricTester()
+	if ask == nil {
+		writeError(w, http.StatusNotImplemented, "this control plane cannot reach its nodes' agents")
+		return
+	}
+	inv := a.ctl.Inventory()
+	na, okA := inv.Get(body.A)
+	nb, okB := inv.Get(body.B)
+	if !okA || !okB {
+		writeError(w, http.StatusNotFound, fmt.Sprintf("no node %q", map[bool]string{true: body.B, false: body.A}[okA]))
+		return
+	}
+	fabric, rails, notes, err := fabrictest.Plan(na, nb, body.Fabric)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+	start := time.Now()
+	rep := fabrictest.Run(r.Context(), ask, na, nb, fabric, rails,
+		fabrictest.Options{Seconds: min(max(body.Seconds, 1), 30), Settle: 2 * time.Second})
+	v := FabricTestView{A: na.ID, B: nb.ID, Fabric: fabric, OK: rep.OK(), Notes: append(notes, rep.Notes...),
+		Took: time.Since(start).Round(time.Second).String()}
+	for _, x := range rep.Results {
+		v.Rails = append(v.Rails, RailView{x.A, x.B, x.RateGbps, x.Gbps, x.Verdict, x.Detail})
+	}
+	writeJSON(w, http.StatusOK, v)
+}

internal/control/agentsvc/convert.go

@@ -23,6 +23,7 @@ func NodeFromRegister(r *llmgwv1.Register) inventory.Node {
 		Labels:         r.GetLabels(),
 		State:          inventory.Ready,
 		AgentVersion:   r.GetAgentVersion(),
+		Features:       r.GetFeatures(),
 		EngineVersions: r.GetEngineVersions(),
 		EnginesProbed:  r.GetEnginesProbed(),
 		MemoryTotalMB:  int(r.GetMemoryTotalMb()),

internal/control/agentsvc/service.go

@@ -19,6 +19,7 @@ import (
 	"gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1/llmgwv1connect"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
 )
@@ -61,6 +62,7 @@ type Service struct {
 	logSeq   uint64
 	// scanWaits are the fabric scans awaiting a reply, by request id.
 	scanWaits map[string]chan *llmgwv1.FabricScanReply
+	testWaits map[string]chan *llmgwv1.FabricTestReply
 }
 
 // SetEngines sets the engines every node installs, by name and version.
@@ -289,6 +291,15 @@ func (s *Service) handle(nodeID string, msg *llmgwv1.AgentMessage) {
 				"last_output", m.Shard.GetLastOutput())
 		}
 
+	case *llmgwv1.AgentMessage_FabricTest:
+		s.logMu.Lock()
+		ch, ok := s.testWaits[m.FabricTest.GetRequestId()]
+		delete(s.testWaits, m.FabricTest.GetRequestId())
+		s.logMu.Unlock()
+		if ok {
+			ch <- m.FabricTest
+		}
+
 	case *llmgwv1.AgentMessage_FabricScan:
 		s.logMu.Lock()
 		ch, ok := s.scanWaits[m.FabricScan.GetRequestId()]
@@ -717,6 +728,38 @@ func (s *Service) ScanFabric(ctx context.Context, nodeID, hca string, port int)
 	}
 }
 
+// FabricTest asks a node for one side of a bandwidth test, and waits for
+// what it measured. The request's id is filled in here.
+func (s *Service) FabricTest(ctx context.Context, nodeID string, req *llmgwv1.FabricTestRequest) (*llmgwv1.FabricTestReply, error) {
+	s.logMu.Lock()
+	if s.testWaits == nil {
+		s.testWaits = map[string]chan *llmgwv1.FabricTestReply{}
+	}
+	s.logSeq++
+	id := fmt.Sprintf("test-%d", s.logSeq)
+	ch := make(chan *llmgwv1.FabricTestReply, 1)
+	s.testWaits[id] = ch
+	s.logMu.Unlock()
+	defer func() {
+		s.logMu.Lock()
+		delete(s.testWaits, id)
+		s.logMu.Unlock()
+	}()
+	req.RequestId = id
+	if err := s.send(nodeID, &llmgwv1.ControlMessage{Msg: &llmgwv1.ControlMessage_FabricTest{FabricTest: req}}); err != nil {
+		return nil, err
+	}
+	// The test's own length, and a package install the first time.
+	ctx, cancel := context.WithTimeout(ctx, time.Duration(req.GetSeconds())*time.Second+5*time.Minute)
+	defer cancel()
+	select {
+	case r := <-ch:
+		return r, nil
+	case <-ctx.Done():
+		return nil, fmt.Errorf("node %s did not answer the bandwidth test", nodeID)
+	}
+}
+
 // ScanFabrics maps every physical InfiniBand fabric every so often, asking
 // one node on each: the first, by name, whose port on it is up. A fabric
 // looks the same from all its ports, so one is enough, and the rest of its
@@ -795,3 +838,7 @@ func installable(reg *llmgwv1.Register) []string {
 	}
 	return []string{"llamacpp"}
 }
+
+// The admin API finds the bandwidth test by asserting this; a signature that
+// drifted would make it quietly nil.
+var _ fabrictest.Asker = (*Service)(nil)

internal/control/control.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint"
@@ -215,6 +216,12 @@ func (c *Controller) LogReader() LogReader {
 	return r
 }
 
+// FabricTester returns the dispatcher's bandwidth test capability, or nil.
+func (c *Controller) FabricTester() fabrictest.Asker {
+	r, _ := c.disp.(fabrictest.Asker)
+	return r
+}
+
 // Start runs the reconciler and begins publishing endpoints.
 func (c *Controller) Start(ctx context.Context) {
 	if c.lead != nil {

internal/control/fabrictest/fabrictest.go

@@ -0,0 +1,235 @@
+// Package fabrictest measures the fabric between two nodes, rail by rail.
+//
+// Every check on one node — ports up, a subnet manager, the right modules —
+// can pass while the two nodes still cannot reach each other: a cable into the
+// wrong switch, a partition key, a RoCE network with no route. A replica
+// spanning them then starts, and hangs on its first all-reduce with nothing to
+// say why. This runs what that all-reduce would ride on — an RDMA write from
+// each adapter to its counterpart — and says how it went against the link's
+// rate, before a replica does.
+package fabrictest
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"strings"
+	"time"
+
+	llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
+)
+
+// Asker sends one side of a test to a node and waits for its answer.
+type Asker interface {
+	FabricTest(ctx context.Context, node string, req *llmgwv1.FabricTestRequest) (*llmgwv1.FabricTestReply, error)
+}
+
+// Rail is one adapter on each node, paired by their position in the fabric's
+// declaration: mlx5_0 with mlx5_0, the way NCCL pairs them.
+type Rail struct {
+	A, B         string // adapters
+	PortA, PortB int
+	// RateGbps is the slower of the two links.
+	RateGbps int
+}
+
+// Result is one rail's test.
+type Result struct {
+	Rail
+	Gbps float64 `json:"gbps"`
+	// Verdict is "ok" at 80% of the link's rate or more, "slow" at 50%,
+	// "fail" below that or when it did not run.
+	Verdict string `json:"verdict"`
+	Detail  string `json:"detail,omitempty"`
+}
+
+// Report is a whole test.
+type Report struct {
+	Fabric  string   `json:"fabric"`
+	A, B    string   `json:"-"`
+	Results []Result `json:"results"`
+	// Notes say what the test could not cover.
+	Notes []string `json:"notes,omitempty"`
+}
+
+// OK reports whether every rail passed.
+func (r Report) OK() bool {
+	for _, x := range r.Results {
+		if x.Verdict != "ok" {
+			return false
+		}
+	}
+	return len(r.Results) > 0
+}
+
+// Plan pairs the adapters two nodes have on a fabric. With no fabric named,
+// the one fabric they share is used.
+func Plan(a, b inventory.Node, fabric string) (string, []Rail, []string, error) {
+	if a.ID == b.ID {
+		return "", nil, nil, errors.New("give two different nodes")
+	}
+	if fabric == "" {
+		var common []string
+		for _, f := range a.FabricNames() {
+			if b.HasFabric(f) {
+				common = append(common, f)
+			}
+		}
+		switch len(common) {
+		case 0:
+			return "", nil, nil, fmt.Errorf("%s (%s) and %s (%s) declare no fabric in common",
+				a.ID, orNone(a.FabricNames()), b.ID, orNone(b.FabricNames()))
+		case 1:
+			fabric = common[0]
+		default:
+			return "", nil, nil, fmt.Errorf("%s and %s share %s: name one with --fabric", a.ID, b.ID, strings.Join(common, ", "))
+		}
+	}
+	for _, n := range []inventory.Node{a, b} {
+		if !slices.Contains(n.Features, "fabric-test") {
+			return "", nil, nil, fmt.Errorf("%s's agent (%s) is too old to run a bandwidth test: upgrade it", n.ID, orNone([]string{n.AgentVersion}))
+		}
+	}
+	ha, hb := a.HCAsFor(fabric), b.HCAsFor(fabric)
+	if len(ha) == 0 || len(hb) == 0 {
+		return "", nil, nil, fmt.Errorf("%s is not declared on both %s and %s", fabric, a.ID, b.ID)
+	}
+	var notes []string
+	if len(ha) != len(hb) {
+		notes = append(notes, fmt.Sprintf("%s has %d adapters on %s and %s has %d: only the first %d are paired",
+			a.ID, len(ha), fabric, b.ID, len(hb), min(len(ha), len(hb))))
+	}
+	var rails []Rail
+	for i := range min(len(ha), len(hb)) {
+		pa, ra := port(a, ha[i])
+		pb, rb := port(b, hb[i])
+		rails = append(rails, Rail{A: ha[i], B: hb[i], PortA: pa, PortB: pb, RateGbps: minPositive(ra, rb)})
+	}
+	return fabric, rails, notes, nil
+}
+
+// port is an adapter's first active port and its rate, or port 1 when the
+// node reported none.
+func port(n inventory.Node, hca string) (int, int) {
+	for _, p := range n.IBPorts {
+		if p.HCA == hca && strings.EqualFold(p.State, "ACTIVE") {
+			return p.Port, p.RateGbps
+		}
+	}
+	return 1, 0
+}
+
+// Options for a run.
+type Options struct {
+	Seconds int
+	// BasePort is perftest's TCP exchange port for the first rail; each rail
+	// takes the next, so a rail whose server is still closing never meets the
+	// next one's.
+	BasePort int
+	// Settle is how long the server is given to listen before the client
+	// dials; the client retries besides.
+	Settle time.Duration
+}
+
+// Run tests every rail, one after another: rails at once would share the
+// PCIe and memory bandwidth they are measuring.
+func Run(ctx context.Context, ask Asker, a, b inventory.Node, fabric string, rails []Rail, o Options) Report {
+	if o.Seconds <= 0 {
+		o.Seconds = 3
+	}
+	if o.BasePort == 0 {
+		o.BasePort = 18515
+	}
+	rep := Report{Fabric: fabric, A: a.ID, B: b.ID}
+	for i, r := range rails {
+		rep.Results = append(rep.Results, runRail(ctx, ask, a, b, r, o, o.BasePort+i))
+	}
+	return rep
+}
+
+func runRail(ctx context.Context, ask Asker, a, b inventory.Node, r Rail, o Options, tcp int) Result {
+	res := Result{Rail: r, Verdict: "fail"}
+	if b.MgmtAddr == "" {
+		res.Detail = b.ID + " reported no management address for the client to dial"
+		return res
+	}
+	type answer struct {
+		r   *llmgwv1.FabricTestReply
+		err error
+	}
+	server := make(chan answer, 1)
+	go func() {
+		rep, err := ask.FabricTest(ctx, b.ID, &llmgwv1.FabricTestRequest{Role: "server", Hca: r.B,
+			Port: int32(r.PortB), TcpPort: int32(tcp), Seconds: int32(o.Seconds)}) //nolint:gosec // small numbers
+		server <- answer{rep, err}
+	}()
+	select {
+	case <-ctx.Done():
+		res.Detail = ctx.Err().Error()
+		return res
+	case <-time.After(o.Settle):
+	}
+	client, err := ask.FabricTest(ctx, a.ID, &llmgwv1.FabricTestRequest{Role: "client", Hca: r.A,
+		Port: int32(r.PortA), Peer: b.MgmtAddr, TcpPort: int32(tcp), Seconds: int32(o.Seconds)}) //nolint:gosec // small numbers
+	var srv answer
+	select {
+	case srv = <-server:
+	case <-time.After(time.Duration(o.Seconds)*time.Second + time.Minute):
+		srv.err = errors.New("the server side did not finish")
+	}
+	switch {
+	case err != nil:
+		res.Detail = a.ID + ": " + err.Error()
+	case client.GetError() != "":
+		res.Detail = a.ID + ": " + client.GetError()
+		// The server's reason is often the better one: it saw no one arrive.
+		if srv.r != nil && srv.r.GetError() != "" {
+			res.Detail += "; " + b.ID + ": " + srv.r.GetError()
+		}
+	default:
+		res.Gbps = client.GetGbps()
+		res.Verdict, res.Detail = grade(res.Gbps, r.RateGbps)
+	}
+	return res
+}
+
+// grade compares a measurement with the link's rate. RDMA writes of 64 KiB
+// reach 90–97% of the signalling rate on a healthy link; well under that is
+// a degraded cable, a card in a narrow PCIe slot, or traffic crossing a
+// socket.
+func grade(gbps float64, rate int) (string, string) {
+	if gbps <= 0 {
+		return "fail", "no traffic got through"
+	}
+	if rate <= 0 {
+		return "ok", "the link's rate is not known to compare with"
+	}
+	pct := gbps / float64(rate) * 100
+	switch {
+	case pct >= 80:
+		return "ok", fmt.Sprintf("%.0f%% of %d Gb/s", pct, rate)
+	case pct >= 50:
+		return "slow", fmt.Sprintf("%.0f%% of %d Gb/s: check the cable, and the card's PCIe width and NUMA node", pct, rate)
+	default:
+		return "fail", fmt.Sprintf("%.0f%% of %d Gb/s", pct, rate)
+	}
+}
+
+func minPositive(a, b int) int {
+	switch {
+	case a <= 0:
+		return b
+	case b <= 0:
+		return a
+	}
+	return min(a, b)
+}
+
+func orNone(s []string) string {
+	if len(s) == 0 {
+		return "none"
+	}
+	return strings.Join(slices.Clone(s), ", ")
+}

internal/control/fabrictest/fabrictest_test.go

@@ -0,0 +1,100 @@
+package fabrictest
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
+)
+
+func node(id string, hcas ...string) inventory.Node {
+	n := inventory.Node{ID: id, MgmtAddr: "10.0.0." + id[len(id)-1:], Features: []string{"fabric-test"},
+		Fabrics: []inventory.FabricAttachment{{Fabric: "ib-a", HCAs: hcas}}}
+	for _, h := range hcas {
+		n.IBPorts = append(n.IBPorts, inventory.IBPort{HCA: h, Port: 1, State: "ACTIVE", RateGbps: 400})
+	}
+	return n
+}
+
+// fakeFabric answers as two agents would: the server waits for its client.
+type fakeFabric struct {
+	mu    sync.Mutex
+	gbps  map[string]float64 // by client adapter
+	fails map[string]string
+	asked []string
+}
+
+func (f *fakeFabric) FabricTest(_ context.Context, n string, req *llmgwv1.FabricTestRequest) (*llmgwv1.FabricTestReply, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.asked = append(f.asked, n+" "+req.GetRole()+" "+req.GetHca())
+	if req.GetRole() == "server" {
+		return &llmgwv1.FabricTestReply{}, nil
+	}
+	if e := f.fails[req.GetHca()]; e != "" {
+		return &llmgwv1.FabricTestReply{Error: e}, nil
+	}
+	return &llmgwv1.FabricTestReply{Gbps: f.gbps[req.GetHca()]}, nil
+}
+
+func TestRailsArePairedByPositionOnTheSharedFabric(t *testing.T) {
+	a, b := node("n1", "mlx5_0", "mlx5_1"), node("n2", "mlx5_0", "mlx5_1", "mlx5_2")
+	fabric, rails, notes, err := Plan(a, b, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if fabric != "ib-a" || len(rails) != 2 || rails[1].A != "mlx5_1" || rails[1].B != "mlx5_1" || rails[0].RateGbps != 400 {
+		t.Errorf("got %s %+v", fabric, rails)
+	}
+	if len(notes) != 1 || !strings.Contains(notes[0], "only the first 2") {
+		t.Errorf("an unequal count went unmentioned: %v", notes)
+	}
+}
+
+func TestNodesWithNoFabricInCommonAreSaidSo(t *testing.T) {
+	b := node("n2", "mlx5_0")
+	b.Fabrics[0].Fabric = "ib-b"
+	if _, _, _, err := Plan(node("n1", "mlx5_0"), b, ""); err == nil || !strings.Contains(err.Error(), "no fabric in common") {
+		t.Errorf("got %v", err)
+	}
+}
+
+func TestEachRailIsGradedAgainstItsLink(t *testing.T) {
+	a, b := node("n1", "mlx5_0", "mlx5_1", "mlx5_2"), node("n2", "mlx5_0", "mlx5_1", "mlx5_2")
+	_, rails, _, _ := Plan(a, b, "ib-a")
+	f := &fakeFabric{
+		gbps:  map[string]float64{"mlx5_0": 388, "mlx5_1": 240},
+		fails: map[string]string{"mlx5_2": "ib_write_bw: Couldn't connect to 10.0.0.2:18517"},
+	}
+	rep := Run(context.Background(), f, a, b, "ib-a", rails, Options{Seconds: 1, Settle: time.Millisecond})
+	want := []string{"ok", "slow", "fail"}
+	for i, r := range rep.Results {
+		if r.Verdict != want[i] {
+			t.Errorf("rail %d: %s (%s), want %s", i, r.Verdict, r.Detail, want[i])
+		}
+	}
+	if rep.OK() {
+		t.Error("a report with a failed rail is OK")
+	}
+	if !strings.Contains(rep.Results[2].Detail, "Couldn't connect") {
+		t.Errorf("the reason was lost: %q", rep.Results[2].Detail)
+	}
+	// Servers on the second node, clients on the first, one rail at a time.
+	if f.asked[0] != "n2 server mlx5_0" || f.asked[1] != "n1 client mlx5_0" {
+		t.Errorf("asked %v", f.asked)
+	}
+}
+
+func TestAnAgentTooOldToAnswerIsRefusedAtOnce(t *testing.T) {
+	// It would ignore the request, and the control plane would wait minutes.
+	old := node("n2", "mlx5_0")
+	old.Features, old.AgentVersion = nil, "v0.10.0"
+	_, _, _, err := Plan(node("n1", "mlx5_0"), old, "")
+	if err == nil || !strings.Contains(err.Error(), "n2's agent (v0.10.0) is too old") {
+		t.Errorf("got %v", err)
+	}
+}

internal/control/inventory/inventory.go

@@ -100,7 +100,10 @@ type Node struct {
 	LastSeen time.Time `json:"last_seen"`
 	// AgentVersion and EngineVersions are reported at registration. The engine
 	// version decides which metric names and CLI flags apply (plan §14).
-	AgentVersion   string            `json:"agent_version,omitempty"`
+	AgentVersion string `json:"agent_version,omitempty"`
+	// Features are the requests the agent answers beyond the ones every
+	// agent does.
+	Features       []string          `json:"features,omitempty"`
 	EngineVersions map[string]string `json:"engine_versions,omitempty"`
 	// EnginesProbed says EngineVersions is an answer, so an empty map means
 	// "has none" rather than "an agent too old to say".

internal/onboard/onboard.go

@@ -35,6 +35,10 @@ import (
 type Fabric struct {
 	Name string
 	HCAs []string
+	// Key is what the adapters report the fabric to be — its subnet manager,
+	// or a RoCE network — on a suggestion, so a name given for one node can
+	// be given again to the next on the same fabric.
+	Key string
 }
 
 // ParseFabric reads "name:hca1,hca2", or "name" alone.
@@ -462,7 +466,7 @@ func suggestFabrics(hcas []discovery.HCA) []Fabric {
 		if len(keys) > 1 {
 			name = fmt.Sprintf("fabric%d", i+1)
 		}
-		out = append(out, Fabric{Name: name, HCAs: byKey[k]})
+		out = append(out, Fabric{Name: name, HCAs: byKey[k], Key: k})
 	}
 	return out
 }