Describe v0.8.4

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

CHANGELOG.md

@@ -18,6 +18,32 @@ 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.8.4 — 2026-09-28
+
+### Node logs on the admin page
+
+Each node on the cluster page links to its logs — the agent's own, and the
+output of every engine running there — with an option to follow them. `llmgw
+ctl logs NODE` reads the same. They come over the agent's existing stream, so
+no ssh, jump host or root is needed to see what a node did. The agent keeps its
+last 2000 lines in memory; each engine its last few hundred.
+
+## v0.8.3 — 2026-09-28
+
+Found by building a node from nothing with v0.8.2.
+
+### Fixes
+
+- **A new node's first engine install failed on apt's lock.** cloud-init and
+  unattended-upgrades run apt in a machine's first minutes, and
+  `DPkg::Lock::Timeout` does not cover the lock `apt-get update` takes. A
+  package command held up by a lock now waits for up to ten minutes, and says
+  so; a failed install is retried after a minute rather than ten.
+- **A new node reported no disk** until its first download created the weights
+  directory. It is created when the agent starts.
+- **Onboarding took one control plane address.** It takes several, comma
+  separated, so a new node fails over like the rest.
+
 ## v0.8.2 — 2026-09-28
 
 A node no longer runs whatever engine somebody once put on it. It installs its

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

@@ -101,6 +101,7 @@ type AgentMessage struct {
 	//	*AgentMessage_Register
 	//	*AgentMessage_Heartbeat
 	//	*AgentMessage_Shard
+	//	*AgentMessage_Logs
 	Msg           isAgentMessage_Msg `protobuf_oneof:"msg"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
@@ -170,6 +171,15 @@ func (x *AgentMessage) GetShard() *ShardStatus {
 	return nil
 }
 
+func (x *AgentMessage) GetLogs() *LogsReply {
+	if x != nil {
+		if x, ok := x.Msg.(*AgentMessage_Logs); ok {
+			return x.Logs
+		}
+	}
+	return nil
+}
+
 type isAgentMessage_Msg interface {
 	isAgentMessage_Msg()
 }
@@ -186,12 +196,18 @@ type AgentMessage_Shard struct {
 	Shard *ShardStatus `protobuf:"bytes,3,opt,name=shard,proto3,oneof"`
 }
 
+type AgentMessage_Logs struct {
+	Logs *LogsReply `protobuf:"bytes,4,opt,name=logs,proto3,oneof"`
+}
+
 func (*AgentMessage_Register) isAgentMessage_Msg() {}
 
 func (*AgentMessage_Heartbeat) isAgentMessage_Msg() {}
 
 func (*AgentMessage_Shard) isAgentMessage_Msg() {}
 
+func (*AgentMessage_Logs) isAgentMessage_Msg() {}
+
 // ControlMessage is anything the control plane tells a node.
 type ControlMessage struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
@@ -203,6 +219,7 @@ type ControlMessage struct {
 	//	*ControlMessage_Ack
 	//	*ControlMessage_RemoveWeights
 	//	*ControlMessage_Provision
+	//	*ControlMessage_Logs
 	Msg           isControlMessage_Msg `protobuf_oneof:"msg"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
@@ -299,6 +316,15 @@ func (x *ControlMessage) GetProvision() *Provision {
 	return nil
 }
 
+func (x *ControlMessage) GetLogs() *LogsRequest {
+	if x != nil {
+		if x, ok := x.Msg.(*ControlMessage_Logs); ok {
+			return x.Logs
+		}
+	}
+	return nil
+}
+
 type isControlMessage_Msg interface {
 	isControlMessage_Msg()
 }
@@ -327,6 +353,10 @@ type ControlMessage_Provision struct {
 	Provision *Provision `protobuf:"bytes,6,opt,name=provision,proto3,oneof"`
 }
 
+type ControlMessage_Logs struct {
+	Logs *LogsRequest `protobuf:"bytes,7,opt,name=logs,proto3,oneof"`
+}
+
 func (*ControlMessage_Start) isControlMessage_Msg() {}
 
 func (*ControlMessage_Stop) isControlMessage_Msg() {}
@@ -339,6 +369,8 @@ func (*ControlMessage_RemoveWeights) isControlMessage_Msg() {}
 
 func (*ControlMessage_Provision) isControlMessage_Msg() {}
 
+func (*ControlMessage_Logs) 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.
@@ -2168,23 +2200,166 @@ func (x *Ack) GetError() string {
 	return ""
 }
 
+// LogsRequest asks a node for the recent lines of one of its logs.
+//
+// Over the stream the agent already holds, because that is the one way into a
+// node every deployment has: the admin page otherwise means ssh through a jump
+// host to a machine whose address nobody remembers, to read a journal only
+// root may open.
+type LogsRequest struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// request_id pairs the reply with its request; several may be in flight.
+	RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
+	// source is "agent" for the agent's own log, or a shard id for that
+	// engine's output.
+	Source        string `protobuf:"bytes,2,opt,name=source,proto3" json:"source,omitempty"`
+	Lines         int32  `protobuf:"varint,3,opt,name=lines,proto3" json:"lines,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *LogsRequest) Reset() {
+	*x = LogsRequest{}
+	mi := &file_llmgw_v1_agent_proto_msgTypes[21]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *LogsRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*LogsRequest) ProtoMessage() {}
+
+func (x *LogsRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_llmgw_v1_agent_proto_msgTypes[21]
+	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 LogsRequest.ProtoReflect.Descriptor instead.
+func (*LogsRequest) Descriptor() ([]byte, []int) {
+	return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{21}
+}
+
+func (x *LogsRequest) GetRequestId() string {
+	if x != nil {
+		return x.RequestId
+	}
+	return ""
+}
+
+func (x *LogsRequest) GetSource() string {
+	if x != nil {
+		return x.Source
+	}
+	return ""
+}
+
+func (x *LogsRequest) GetLines() int32 {
+	if x != nil {
+		return x.Lines
+	}
+	return 0
+}
+
+// LogsReply answers a LogsRequest.
+type LogsReply struct {
+	state     protoimpl.MessageState `protogen:"open.v1"`
+	RequestId string                 `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
+	Lines     []string               `protobuf:"bytes,2,rep,name=lines,proto3" json:"lines,omitempty"`
+	// sources are every log this node can show, so a page can offer them
+	// without knowing what runs there.
+	Sources []string `protobuf:"bytes,3,rep,name=sources,proto3" json:"sources,omitempty"`
+	// error says why the lines are absent: an unknown source, say.
+	Error         string `protobuf:"bytes,4,opt,name=error,proto3" json:"error,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *LogsReply) Reset() {
+	*x = LogsReply{}
+	mi := &file_llmgw_v1_agent_proto_msgTypes[22]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *LogsReply) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*LogsReply) ProtoMessage() {}
+
+func (x *LogsReply) ProtoReflect() protoreflect.Message {
+	mi := &file_llmgw_v1_agent_proto_msgTypes[22]
+	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 LogsReply.ProtoReflect.Descriptor instead.
+func (*LogsReply) Descriptor() ([]byte, []int) {
+	return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{22}
+}
+
+func (x *LogsReply) GetRequestId() string {
+	if x != nil {
+		return x.RequestId
+	}
+	return ""
+}
+
+func (x *LogsReply) GetLines() []string {
+	if x != nil {
+		return x.Lines
+	}
+	return nil
+}
+
+func (x *LogsReply) GetSources() []string {
+	if x != nil {
+		return x.Sources
+	}
+	return nil
+}
+
+func (x *LogsReply) GetError() string {
+	if x != nil {
+		return x.Error
+	}
+	return ""
+}
+
 var File_llmgw_v1_agent_proto protoreflect.FileDescriptor
 
 const file_llmgw_v1_agent_proto_rawDesc = "" +
 	"\n" +
-	"\x14llmgw/v1/agent.proto\x12\bllmgw.v1\"\xab\x01\n" +
+	"\x14llmgw/v1/agent.proto\x12\bllmgw.v1\"\xd6\x01\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\x05shardB\x05\n" +
-	"\x03msg\"\xb8\x02\n" +
+	"\x05shard\x18\x03 \x01(\v2\x15.llmgw.v1.ShardStatusH\x00R\x05shard\x12)\n" +
+	"\x04logs\x18\x04 \x01(\v2\x13.llmgw.v1.LogsReplyH\x00R\x04logsB\x05\n" +
+	"\x03msg\"\xe5\x02\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" +
 	"\x05drain\x18\x03 \x01(\v2\x14.llmgw.v1.DrainShardH\x00R\x05drain\x12!\n" +
 	"\x03ack\x18\x04 \x01(\v2\r.llmgw.v1.AckH\x00R\x03ack\x12@\n" +
 	"\x0eremove_weights\x18\x05 \x01(\v2\x17.llmgw.v1.RemoveWeightsH\x00R\rremoveWeights\x123\n" +
-	"\tprovision\x18\x06 \x01(\v2\x13.llmgw.v1.ProvisionH\x00R\tprovisionB\x05\n" +
+	"\tprovision\x18\x06 \x01(\v2\x13.llmgw.v1.ProvisionH\x00R\tprovision\x12+\n" +
+	"\x04logs\x18\a \x01(\v2\x15.llmgw.v1.LogsRequestH\x00R\x04logsB\x05\n" +
 	"\x03msg\"\xf2\x05\n" +
 	"\bRegister\x12\x17\n" +
 	"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x1a\n" +
@@ -2379,7 +2554,18 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
 	"\n" +
 	"message_id\x18\x01 \x01(\tR\tmessageId\x12\x0e\n" +
 	"\x02ok\x18\x02 \x01(\bR\x02ok\x12\x14\n" +
-	"\x05error\x18\x03 \x01(\tR\x05error2O\n" +
+	"\x05error\x18\x03 \x01(\tR\x05error\"Z\n" +
+	"\vLogsRequest\x12\x1d\n" +
+	"\n" +
+	"request_id\x18\x01 \x01(\tR\trequestId\x12\x16\n" +
+	"\x06source\x18\x02 \x01(\tR\x06source\x12\x14\n" +
+	"\x05lines\x18\x03 \x01(\x05R\x05lines\"p\n" +
+	"\tLogsReply\x12\x1d\n" +
+	"\n" +
+	"request_id\x18\x01 \x01(\tR\trequestId\x12\x14\n" +
+	"\x05lines\x18\x02 \x03(\tR\x05lines\x12\x18\n" +
+	"\asources\x18\x03 \x03(\tR\asources\x12\x14\n" +
+	"\x05error\x18\x04 \x01(\tR\x05error2O\n" +
 	"\fAgentService\x12?\n" +
 	"\aConnect\x12\x16.llmgw.v1.AgentMessage\x1a\x18.llmgw.v1.ControlMessage(\x010\x01B\x93\x01\n" +
 	"\fcom.llmgw.v1B\n" +
@@ -2398,7 +2584,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, 25)
+var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 27)
 var file_llmgw_v1_agent_proto_goTypes = []any{
 	(ShardStatus_State)(0),   // 0: llmgw.v1.ShardStatus.State
 	(*AgentMessage)(nil),     // 1: llmgw.v1.AgentMessage
@@ -2422,46 +2608,50 @@ var file_llmgw_v1_agent_proto_goTypes = []any{
 	(*StopShard)(nil),        // 19: llmgw.v1.StopShard
 	(*DrainShard)(nil),       // 20: llmgw.v1.DrainShard
 	(*Ack)(nil),              // 21: llmgw.v1.Ack
-	nil,                      // 22: llmgw.v1.Register.LabelsEntry
-	nil,                      // 23: llmgw.v1.Register.EngineVersionsEntry
-	nil,                      // 24: llmgw.v1.Heartbeat.EngineVersionsEntry
-	nil,                      // 25: llmgw.v1.EngineProfile.EnvEntry
+	(*LogsRequest)(nil),      // 22: llmgw.v1.LogsRequest
+	(*LogsReply)(nil),        // 23: llmgw.v1.LogsReply
+	nil,                      // 24: llmgw.v1.Register.LabelsEntry
+	nil,                      // 25: llmgw.v1.Register.EngineVersionsEntry
+	nil,                      // 26: llmgw.v1.Heartbeat.EngineVersionsEntry
+	nil,                      // 27: llmgw.v1.EngineProfile.EnvEntry
 }
 var file_llmgw_v1_agent_proto_depIdxs = []int32{
 	3,  // 0: llmgw.v1.AgentMessage.register:type_name -> llmgw.v1.Register
 	10, // 1: llmgw.v1.AgentMessage.heartbeat:type_name -> llmgw.v1.Heartbeat
 	14, // 2: llmgw.v1.AgentMessage.shard:type_name -> llmgw.v1.ShardStatus
-	16, // 3: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard
-	19, // 4: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard
-	20, // 5: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard
-	21, // 6: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack
-	18, // 7: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights
-	11, // 8: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision
-	22, // 9: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
-	7,  // 10: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
-	8,  // 11: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
-	23, // 12: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
-	6,  // 13: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
-	9,  // 14: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
-	5,  // 15: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
-	4,  // 16: llmgw.v1.Register.host:type_name -> llmgw.v1.Host
-	0,  // 17: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
-	5,  // 18: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
-	9,  // 19: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
-	24, // 20: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
-	13, // 21: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus
-	12, // 22: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant
-	0,  // 23: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
-	15, // 24: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
-	17, // 25: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
-	25, // 26: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
-	1,  // 27: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
-	2,  // 28: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
-	28, // [28:29] is the sub-list for method output_type
-	27, // [27:28] is the sub-list for method input_type
-	27, // [27:27] is the sub-list for extension type_name
-	27, // [27:27] is the sub-list for extension extendee
-	0,  // [0:27] is the sub-list for field type_name
+	23, // 3: llmgw.v1.AgentMessage.logs:type_name -> llmgw.v1.LogsReply
+	16, // 4: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard
+	19, // 5: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard
+	20, // 6: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard
+	21, // 7: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack
+	18, // 8: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights
+	11, // 9: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision
+	22, // 10: llmgw.v1.ControlMessage.logs:type_name -> llmgw.v1.LogsRequest
+	24, // 11: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
+	7,  // 12: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
+	8,  // 13: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
+	25, // 14: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
+	6,  // 15: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
+	9,  // 16: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
+	5,  // 17: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
+	4,  // 18: llmgw.v1.Register.host:type_name -> llmgw.v1.Host
+	0,  // 19: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
+	5,  // 20: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
+	9,  // 21: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
+	26, // 22: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
+	13, // 23: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus
+	12, // 24: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant
+	0,  // 25: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
+	15, // 26: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
+	17, // 27: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
+	27, // 28: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
+	1,  // 29: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
+	2,  // 30: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
+	30, // [30:31] is the sub-list for method output_type
+	29, // [29:30] is the sub-list for method input_type
+	29, // [29:29] is the sub-list for extension type_name
+	29, // [29:29] is the sub-list for extension extendee
+	0,  // [0:29] is the sub-list for field type_name
 }
 
 func init() { file_llmgw_v1_agent_proto_init() }
@@ -2473,6 +2663,7 @@ func file_llmgw_v1_agent_proto_init() {
 		(*AgentMessage_Register)(nil),
 		(*AgentMessage_Heartbeat)(nil),
 		(*AgentMessage_Shard)(nil),
+		(*AgentMessage_Logs)(nil),
 	}
 	file_llmgw_v1_agent_proto_msgTypes[1].OneofWrappers = []any{
 		(*ControlMessage_Start)(nil),
@@ -2481,6 +2672,7 @@ func file_llmgw_v1_agent_proto_init() {
 		(*ControlMessage_Ack)(nil),
 		(*ControlMessage_RemoveWeights)(nil),
 		(*ControlMessage_Provision)(nil),
+		(*ControlMessage_Logs)(nil),
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -2488,7 +2680,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:   25,
+			NumMessages:   27,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

api/proto/llmgw/v1/agent.proto

@@ -23,6 +23,7 @@ message AgentMessage {
     Register register = 1;
     Heartbeat heartbeat = 2;
     ShardStatus shard = 3;
+    LogsReply logs = 4;
   }
 }
 
@@ -35,6 +36,7 @@ message ControlMessage {
     Ack ack = 4;
     RemoveWeights remove_weights = 5;
     Provision provision = 6;
+    LogsRequest logs = 7;
   }
 }
 
@@ -418,3 +420,29 @@ message Ack {
   bool ok = 2;
   string error = 3;
 }
+
+// LogsRequest asks a node for the recent lines of one of its logs.
+//
+// Over the stream the agent already holds, because that is the one way into a
+// node every deployment has: the admin page otherwise means ssh through a jump
+// host to a machine whose address nobody remembers, to read a journal only
+// root may open.
+message LogsRequest {
+  // request_id pairs the reply with its request; several may be in flight.
+  string request_id = 1;
+  // source is "agent" for the agent's own log, or a shard id for that
+  // engine's output.
+  string source = 2;
+  int32 lines = 3;
+}
+
+// LogsReply answers a LogsRequest.
+message LogsReply {
+  string request_id = 1;
+  repeated string lines = 2;
+  // sources are every log this node can show, so a page can offer them
+  // without knowing what runs there.
+  repeated string sources = 3;
+  // error says why the lines are absent: an unknown source, say.
+  string error = 4;
+}

deploy/bootstrap/onboard-node.sh

@@ -24,7 +24,8 @@ usage: $0 --node ID --addr HOST --server URL [options]
 
   --node ID         node id, hostname and certificate common name
   --addr HOST       how to reach the machine over ssh
-  --server URL      the control plane's agent endpoint, e.g. https://10.0.0.140:7443
+  --server URL      the control plane's agent endpoint, e.g. https://10.0.0.140:7443;
+                    several, comma separated, for a node that fails over
   --fabric NAME     InfiniBand fabric this node is attached to; repeatable as
                     --fabric NAME:hca1,hca2 . Omit for a node with no fabric,
                     which can host single-node replicas and will never be given
@@ -169,7 +170,14 @@ NODE_IS_HOSTNAME=no
 # below is written only when it was given, as an override.
 {
   echo "# $NODE, written by onboard-node.sh."
-  echo "server: \"$SERVER\""
+  # Several addresses, comma separated, are every control plane the node may
+  # fail over to; a standby refuses it and it moves on to the next.
+  if [ "${SERVER#*,}" != "$SERVER" ]; then
+    echo "servers:"
+    printf '%s\n' "$SERVER" | tr ',' '\n' | sed 's/^ *//; s/^/  - "/; s/$/"/'
+  else
+    echo "server: \"$SERVER\""
+  fi
   echo
   echo "tls:"
   echo "  ca: /etc/llmgw/ca.pem"

internal/agent/agent.go

@@ -326,6 +326,8 @@ type Agent struct {
 	// it asked for, nil until it has; provStatus is where each stands and
 	// provSent the fingerprint of what the control plane last heard. All
 	// guarded by mu. provKick starts a pass without waiting for the timer.
+	logRing *logRing
+
 	prov       *provision.Installer
 	provWants  []provision.Want
 	provStatus map[string]provision.Status
@@ -372,6 +374,11 @@ func New(cfg Config, httpClient connectHTTPClient, log *slog.Logger) (*Agent, er
 	if log == nil {
 		log = slog.Default()
 	}
+	// Everything the agent logs also goes to a ring the admin page can read,
+	// so seeing what a node did does not mean ssh and a journal only root may
+	// open.
+	ring := newLogRing(agentLogLines)
+	log = slog.New(teeHandler{log.Handler(), slog.NewTextHandler(ring, nil)})
 	if httpClient == nil {
 		httpClient = http.DefaultClient
 	}
@@ -387,6 +394,7 @@ func New(cfg Config, httpClient connectHTTPClient, log *slog.Logger) (*Agent, er
 		http:         httpClient,
 		shardReplica: map[string]string{},
 		provStatus:   map[string]provision.Status{},
+		logRing:      ring,
 		provKick:     make(chan struct{}, 1),
 	}
 	a.prov = &provision.Installer{Root: cfg.EnginesDir, Log: log, OnStatus: a.provisionStatus}
@@ -409,6 +417,11 @@ func (a *Agent) Run(ctx context.Context) error {
 		a.log.Error("hardware discovery failed", "err", err)
 	}
 	a.recoverOrphans()
+	// Made now rather than by the first download, so a new node reports its
+	// disk from its first heartbeat instead of only once it holds a model.
+	if err := os.MkdirAll(a.cfg.WeightsCache, 0o755); err != nil {
+		a.log.Warn("cannot create the weights directory", "dir", a.cfg.WeightsCache, "err", err)
+	}
 	go a.provisionLoop(ctx)
 
 	attempt := 0
@@ -955,6 +968,16 @@ func (a *Agent) handle(ctx context.Context, msg *llmgwv1.ControlMessage, out cha
 		a.removeWeights(m.RemoveWeights.GetPaths())
 	case *llmgwv1.ControlMessage_Provision:
 		a.setProvision(m.Provision)
+	case *llmgwv1.ControlMessage_Logs:
+		reply := a.logsReply(m.Logs)
+		// Off the receive loop: a full send queue must not stop commands being
+		// read.
+		go func() {
+			select {
+			case out <- &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Logs{Logs: reply}}:
+			case <-ctx.Done():
+			}
+		}()
 	case *llmgwv1.ControlMessage_Ack:
 		if !m.Ack.GetOk() {
 			a.log.Warn("control plane rejected a message",
@@ -1265,14 +1288,17 @@ func (a *Agent) setProvision(p *llmgwv1.Provision) {
 
 // provisionLoop installs and re-checks the wanted engines, one pass at a time.
 func (a *Agent) provisionLoop(ctx context.Context) {
-	t := time.NewTicker(provisionEvery)
-	defer t.Stop()
+	// After a failure the next pass comes sooner, and backs off: a lock held
+	// by a new machine's own updates clears in a minute or two, while ten
+	// minutes' wait left a node idle for no reason.
+	next := provisionEvery
+	retry := time.Minute
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-a.provKick:
-		case <-t.C:
+		case <-time.After(next):
 		}
 		a.mu.Lock()
 		wants := slices.Clone(a.provWants)
@@ -1283,6 +1309,16 @@ func (a *Agent) provisionLoop(ctx context.Context) {
 		}
 
 		statuses := a.prov.Converge(ctx, wants)
+		next = provisionEvery
+		for _, st := range statuses {
+			if st.State == provision.Failed {
+				next, retry = retry, min(retry*2, provisionEvery)
+				break
+			}
+		}
+		if next == provisionEvery {
+			retry = time.Minute
+		}
 		installed := map[string]string{}
 		keep := map[string]bool{}
 		for _, st := range statuses {
@@ -1376,3 +1412,30 @@ func enginesFingerprint(vers map[string]string) string {
 	}
 	return b.String()
 }
+
+// logsReply answers a request for one of this node's logs.
+func (a *Agent) logsReply(req *llmgwv1.LogsRequest) *llmgwv1.LogsReply {
+	n := int(req.GetLines())
+	if n <= 0 {
+		n = 200
+	}
+	n = min(n, agentLogLines)
+	shards := a.sup.IDs()
+	slices.Sort(shards)
+	out := &llmgwv1.LogsReply{
+		RequestId: req.GetRequestId(),
+		Sources:   append([]string{"agent"}, shards...),
+	}
+	switch src := req.GetSource(); src {
+	case "", "agent":
+		out.Lines = a.logRing.tail(n)
+	default:
+		sh, err := a.sup.Get(src)
+		if err != nil {
+			out.Error = "no engine " + src + " runs on this node"
+			return out
+		}
+		out.Lines = sh.Logs(n)
+	}
+	return out
+}

internal/agent/logring.go

@@ -0,0 +1,84 @@
+package agent
+
+import (
+	"context"
+	"log/slog"
+	"strings"
+	"sync"
+)
+
+// agentLogLines is how much of its own log an agent keeps for the admin page.
+// Enough to cover an engine install or a failed start; the journal keeps the
+// rest for anyone who can reach it.
+const agentLogLines = 2000
+
+// logRing keeps the most recent lines written to it.
+type logRing struct {
+	mu    sync.Mutex
+	lines []string
+	next  int
+	full  bool
+}
+
+func newLogRing(n int) *logRing { return &logRing{lines: make([]string, n)} }
+
+// Write takes whole lines, as a text handler writes them.
+func (r *logRing) Write(p []byte) (int, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for _, line := range strings.Split(strings.TrimRight(string(p), "\n"), "\n") {
+		r.lines[r.next] = line
+		r.next = (r.next + 1) % len(r.lines)
+		if r.next == 0 {
+			r.full = true
+		}
+	}
+	return len(p), nil
+}
+
+// tail returns up to n of the most recent lines, oldest first.
+func (r *logRing) tail(n int) []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	size := r.next
+	if r.full {
+		size = len(r.lines)
+	}
+	n = min(n, size)
+	out := make([]string, 0, n)
+	for i := size - n; i < size; i++ {
+		idx := i
+		if r.full {
+			idx = (r.next + i) % len(r.lines)
+		}
+		out = append(out, r.lines[idx])
+	}
+	return out
+}
+
+// teeHandler sends every record to two handlers: wherever the agent already
+// logs, and the ring the admin page reads.
+type teeHandler struct{ a, b slog.Handler }
+
+func (t teeHandler) Enabled(ctx context.Context, l slog.Level) bool {
+	return t.a.Enabled(ctx, l) || t.b.Enabled(ctx, l)
+}
+
+func (t teeHandler) Handle(ctx context.Context, r slog.Record) error {
+	var err error
+	if t.a.Enabled(ctx, r.Level) {
+		err = t.a.Handle(ctx, r.Clone())
+	}
+	if t.b.Enabled(ctx, r.Level) {
+		_ = t.b.Handle(ctx, r.Clone())
+	}
+	return err
+}
+
+func (t teeHandler) WithAttrs(as []slog.Attr) slog.Handler {
+	return teeHandler{t.a.WithAttrs(as), t.b.WithAttrs(as)}
+}
+
+func (t teeHandler) WithGroup(name string) slog.Handler {
+	return teeHandler{t.a.WithGroup(name), t.b.WithGroup(name)}
+}

internal/agent/logring_test.go

@@ -0,0 +1,49 @@
+package agent
+
+import (
+	"fmt"
+	"slices"
+	"strings"
+	"testing"
+
+	llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
+)
+
+func TestTheLogRingKeepsTheNewestLinesInOrder(t *testing.T) {
+	r := newLogRing(3)
+	if got := r.tail(10); len(got) != 0 {
+		t.Fatalf("empty ring gave %v", got)
+	}
+	for i := 1; i <= 5; i++ {
+		fmt.Fprintf(r, "line %d\n", i)
+	}
+	if got := r.tail(10); !slices.Equal(got, []string{"line 3", "line 4", "line 5"}) {
+		t.Errorf("after wrapping: %v", got)
+	}
+	if got := r.tail(2); !slices.Equal(got, []string{"line 4", "line 5"}) {
+		t.Errorf("tail(2): %v", got)
+	}
+}
+
+func TestANodeAnswersForItsOwnLog(t *testing.T) {
+	cfg := Config{Server: "127.0.0.1:7443", NodeID: "n1", WeightsCache: t.TempDir(), EnginesDir: t.TempDir()}
+	cfg.Superv.WorkDir = t.TempDir()
+	a, err := New(cfg, nil, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	a.log.Info("engine installation", "state", "installing")
+
+	r := a.logsReply(&llmgwv1.LogsRequest{RequestId: "x", Source: "agent", Lines: 50})
+	if r.GetRequestId() != "x" || !strings.Contains(strings.Join(r.GetLines(), "\n"), "engine installation") {
+		t.Errorf("the agent's own line is not in its log: %v", r.GetLines())
+	}
+	if !slices.Contains(r.GetSources(), "agent") {
+		t.Errorf("sources %v", r.GetSources())
+	}
+
+	r = a.logsReply(&llmgwv1.LogsRequest{RequestId: "y", Source: "no-such-shard"})
+	if r.GetError() == "" {
+		t.Error("an unknown engine was answered as if it existed")
+	}
+}

internal/agent/provision/install.go

@@ -283,13 +283,57 @@ func (in *Installer) packages(ctx context.Context, byFamily map[string][]string,
 	}
 	in.progress(st, "installing %s with %s", strings.Join(missing, ", "), tool)
 	for _, c := range cmds {
-		if out, err := in.run(ctx, pm.env, c[0], c[1:]...); err != nil {
-			return fmt.Errorf("%s: %v: %s", strings.Join(c, " "), err, lastLine(out))
+		if err := in.runWaitingForLock(ctx, pm.env, c, st); err != nil {
+			return err
 		}
 	}
 	return nil
 }
 
+// lockWait bounds how long a package command waits for another to finish.
+const lockWait = 10 * time.Minute
+
+// lockRetry is how often a locked package manager is tried again.
+var lockRetry = 10 * time.Second
+
+// runWaitingForLock runs a package command, waiting while another holds the
+// package manager's lock.
+//
+// A machine a minute old is running cloud-init's and unattended-upgrades' own
+// apt. apt's DPkg::Lock::Timeout covers dpkg's lock and not the one `apt-get
+// update` takes on its lists, so the first install on a new node failed with
+// "Unable to lock directory /var/lib/apt/lists/" — found by building one.
+func (in *Installer) runWaitingForLock(ctx context.Context, env, c []string, st *Status) error {
+	deadline := time.Now().Add(lockWait)
+	for {
+		out, err := in.run(ctx, env, c[0], c[1:]...)
+		if err == nil {
+			return nil
+		}
+		if !lockHeld(string(out)) || time.Now().After(deadline) {
+			return fmt.Errorf("%s: %v: %s", strings.Join(c, " "), err, lastLine(out))
+		}
+		in.progress(st, "waiting for another %s to finish (a new machine runs its own updates first)", c[0])
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(lockRetry):
+		}
+	}
+}
+
+// lockHeld reports whether a package command failed only because another
+// held the lock.
+func lockHeld(out string) bool {
+	for _, s := range []string{"Unable to lock", "Could not get lock", "is locked by another process",
+		"waiting for process with pid", "System management is locked"} {
+		if strings.Contains(out, s) {
+			return true
+		}
+	}
+	return false
+}
+
 // packageManager is how one packaging family asks and installs.
 type packageManager struct {
 	// query exits zero when the package named after it is installed.

internal/agent/provision/install_test.go

@@ -16,6 +16,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
 )
@@ -89,6 +90,8 @@ type fakeSystem struct {
 	tools     map[string]bool
 	lddOut    string
 	calls     []string
+	// locked is how many more package commands find the lock held.
+	locked int
 }
 
 var errExit = errors.New("exit 1")
@@ -110,6 +113,10 @@ func (f *fakeSystem) run(_ context.Context, _ []string, name string, args ...str
 		}
 		return nil, errExit
 	case name == "apt-get" || name == "dnf":
+		if f.locked > 0 {
+			f.locked--
+			return []byte("E: Unable to lock directory /var/lib/apt/lists/\n"), errExit
+		}
 		for i, a := range args {
 			if a == "install" {
 				for _, p := range args[i+1:] {
@@ -300,3 +307,26 @@ func TestASystemTooOldForTheBuildSaysWhatItLacks(t *testing.T) {
 		t.Errorf("the same shortfall was listed more than once: %q", got[0].Detail)
 	}
 }
+
+func TestAPackageManagerBusyWithItsOwnUpdatesIsWaitedFor(t *testing.T) {
+	// A new node's first minutes: cloud-init and unattended-upgrades hold
+	// apt's lists lock, which DPkg::Lock::Timeout does not cover.
+	lockRetry = time.Millisecond
+	t.Cleanup(func() { lockRetry = 10 * time.Second })
+
+	rel := llamaRelease(t)
+	sys := &fakeSystem{installed: map[string]bool{}, locked: 3}
+	in, seen := installer(t, rel.serve(t), sys, ubuntu)
+
+	got := in.Converge(t.Context(), []Want{{Engine: "llamacpp", Version: "b1"}})
+	if got[0].State != Ready {
+		t.Fatalf("gave up on a lock another apt held: %+v", got[0])
+	}
+	waited := false
+	for _, s := range *seen {
+		waited = waited || strings.Contains(s.Detail, "waiting for another apt-get")
+	}
+	if !waited {
+		t.Error("the wait was not reported")
+	}
+}

internal/cli/ctl.go

@@ -116,6 +116,7 @@ func newCtlCmd() *cobra.Command {
 
 	cmd.AddCommand(
 		newNodesCmd(client, &asJSON),
+		newLogsCmd(client, &asJSON),
 		newModelsCmd(client, &asJSON),
 		newReplicasCmd(client, &asJSON),
 		newExplainCmd(client, &asJSON),

internal/cli/ctl_logs.go

@@ -0,0 +1,58 @@
+package cli
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/url"
+	"strconv"
+	"strings"
+
+	"github.com/spf13/cobra"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
+)
+
+func newLogsCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
+	var (
+		source string
+		lines  int
+	)
+	c := &cobra.Command{
+		Use:   "logs NODE",
+		Short: "Show the recent lines of a node's agent log, or of an engine on it",
+		Long: "Read through the agent's own stream, so it needs no access to the node:\n" +
+			"no ssh, no jump host, no root for the journal. The agent keeps its last\n" +
+			"2000 lines and each engine's last few hundred.\n\n" +
+			"  llmgw ctl logs llmgw-node06                  the agent\n" +
+			"  llmgw ctl logs llmgw-node06 --source SHARD   an engine's output",
+		Args: cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			q := url.Values{}
+			q.Set("source", source)
+			q.Set("lines", strconv.Itoa(lines))
+			var v adminapi.LogsView
+			err := client().get(cmd.Context(),
+				"/api/v1/nodes/"+url.PathEscape(args[0])+"/logs?"+q.Encode(), &v)
+			if *asJSON {
+				return json.NewEncoder(cmd.OutOrStdout()).Encode(v)
+			}
+			if err != nil {
+				return err
+			}
+			for _, l := range v.Lines {
+				fmt.Fprintln(cmd.OutOrStdout(), l)
+			}
+			if v.Error != "" {
+				return fmt.Errorf("%s", v.Error)
+			}
+			// Where else to look, since the source list is the one thing a
+			// person cannot guess.
+			fmt.Fprintf(cmd.ErrOrStderr(), "\n-- %s on %s; other sources: %s\n",
+				v.Source, v.Node, strings.Join(v.Sources, ", "))
+			return nil
+		},
+	}
+	c.Flags().StringVar(&source, "source", "agent", `"agent", or a shard id for that engine's output`)
+	c.Flags().IntVarP(&lines, "lines", "n", 200, "how many of the most recent lines")
+	return c
+}

internal/cli/server.go

@@ -203,6 +203,7 @@ func newServerCmd() *cobra.Command {
 					api, usage, version.Version, 5*time.Second, true,
 					web.NavExtras{History: history, Window: cfg.Monitoring.Window}, log))
 				adminMux.Handle("GET /_static/", web.Static())
+				adminMux.HandleFunc("GET /nodes/{id}/logs", web.NodeLogsPage(api, true, version.Version))
 
 				// The same page on the client-facing port at /admin, behind a
 				// session and an admin flag. Handing the router the data rather
@@ -466,6 +467,21 @@ func (r *dispatcherRef) RemoveWeights(ctx context.Context, nodeID string, paths
 	return rm.RemoveWeights(ctx, nodeID, paths)
 }
 
+// Logs forwards a node log request, the same way RemoveWeights does: the
+// controller finds it by type assertion, so a forwarder without it would make
+// the capability silently absent at runtime.
+func (r *dispatcherRef) Logs(ctx context.Context, nodeID, source string, lines int) ([]string, []string, error) {
+	d, err := r.get()
+	if err != nil {
+		return nil, nil, err
+	}
+	lr, ok := d.(control.LogReader)
+	if !ok {
+		return nil, nil, errors.New("this dispatcher cannot read node logs")
+	}
+	return lr.Logs(ctx, nodeID, source, lines)
+}
+
 func (r *dispatcherRef) StartShard(ctx context.Context, cmd reconciler.ShardCommand) error {
 	d, err := r.get()
 	if err != nil {

internal/control/adminapi/api.go

@@ -56,6 +56,7 @@ func (a *API) Register(mux *http.ServeMux) {
 	mux.HandleFunc("POST /api/v1/weights/collect", a.collectWeights)
 	mux.HandleFunc("GET /api/v1/explain-placement", a.explainPlacement)
 	mux.HandleFunc("POST /api/v1/nodes/{id}/state", a.setNodeState)
+	mux.HandleFunc("GET /api/v1/nodes/{id}/logs", a.nodeLogs)
 }
 
 // --- responses ---

internal/control/adminapi/logs.go

@@ -0,0 +1,60 @@
+package adminapi
+
+import (
+	"context"
+	"net/http"
+	"strconv"
+)
+
+// LogsView is one of a node's logs, as the node sent it.
+type LogsView struct {
+	Node   string   `json:"node"`
+	Source string   `json:"source"`
+	Lines  []string `json:"lines"`
+	// Sources are every log the node can show: "agent", and one per engine
+	// running there.
+	Sources []string `json:"sources"`
+	Error   string   `json:"error,omitempty"`
+}
+
+// maxLogLines bounds a request. The agent keeps two thousand lines of its own
+// and a few hundred per engine, so more would only be refused further down.
+const maxLogLines = 2000
+
+// NodeLogs asks a node for the recent lines of one of its logs.
+//
+// Exported because the cluster page renders it server-side, the same as
+// Nodes.
+func (a *API) NodeLogs(ctx context.Context, node, source string, lines int) LogsView {
+	if source == "" {
+		source = "agent"
+	}
+	v := LogsView{Node: node, Source: source}
+	lr := a.ctl.LogReader()
+	if lr == nil {
+		v.Error = "this control plane cannot read node logs"
+		return v
+	}
+	if lines <= 0 {
+		lines = 300
+	}
+	out, sources, err := lr.Logs(ctx, node, source, min(lines, maxLogLines))
+	v.Lines, v.Sources = out, sources
+	if err != nil {
+		v.Error = err.Error()
+	}
+	if len(v.Sources) == 0 {
+		v.Sources = []string{"agent"}
+	}
+	return v
+}
+
+func (a *API) nodeLogs(w http.ResponseWriter, r *http.Request) {
+	lines, _ := strconv.Atoi(r.URL.Query().Get("lines"))
+	v := a.NodeLogs(r.Context(), r.PathValue("id"), r.URL.Query().Get("source"), lines)
+	code := http.StatusOK
+	if v.Error != "" && len(v.Lines) == 0 {
+		code = http.StatusBadGateway
+	}
+	writeJSON(w, code, v)
+}

internal/control/agentsvc/heartbeat_test.go

@@ -155,3 +155,36 @@ func TestAHeartbeatCarriesAnEngineInstalledAfterRegistration(t *testing.T) {
 		t.Errorf("the newly installed engine is not known: %v", n.EngineVersions)
 	}
 }
+
+func TestALogRequestIsAnsweredByTheNodeItWasSentTo(t *testing.T) {
+	s := newTestService()
+	c := &conn{nodeID: "n1", send: make(chan *llmgwv1.ControlMessage, 4), closed: make(chan struct{})}
+	s.conns["n1"] = c
+
+	type result struct {
+		lines, sources []string
+		err            error
+	}
+	done := make(chan result, 1)
+	go func() {
+		l, src, err := s.Logs(t.Context(), "n1", "agent", 10)
+		done <- result{l, src, err}
+	}()
+
+	req := (<-c.send).GetLogs()
+	if req.GetSource() != "agent" || req.GetLines() != 10 {
+		t.Fatalf("sent %v", req)
+	}
+	s.handle("n1", &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Logs{Logs: &llmgwv1.LogsReply{
+		RequestId: req.GetRequestId(), Lines: []string{"hello"}, Sources: []string{"agent", "shard-1"},
+	}}})
+
+	r := <-done
+	if r.err != nil || len(r.lines) != 1 || r.lines[0] != "hello" || len(r.sources) != 2 {
+		t.Errorf("got %+v", r)
+	}
+
+	if _, _, err := s.Logs(t.Context(), "nobody", "agent", 10); err == nil {
+		t.Error("a node with no stream answered")
+	}
+}

internal/control/agentsvc/service.go

@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"slices"
 	"sync"
+	"time"
 
 	"connectrpc.com/connect"
 
@@ -52,6 +53,11 @@ type Service struct {
 
 	// engines is what every node is told to install, set once at start.
 	engines map[string]string
+
+	// logWaits are the log requests awaiting a reply, by request id.
+	logMu    sync.Mutex
+	logWaits map[string]chan *llmgwv1.LogsReply
+	logSeq   uint64
 }
 
 // SetEngines sets the engines every node installs, by name and version.
@@ -276,6 +282,15 @@ func (s *Service) handle(nodeID string, msg *llmgwv1.AgentMessage) {
 				"last_output", m.Shard.GetLastOutput())
 		}
 
+	case *llmgwv1.AgentMessage_Logs:
+		s.logMu.Lock()
+		ch, ok := s.logWaits[m.Logs.GetRequestId()]
+		delete(s.logWaits, m.Logs.GetRequestId())
+		s.logMu.Unlock()
+		if ok {
+			ch <- m.Logs // buffered; the asker may have given up
+		}
+
 	case *llmgwv1.AgentMessage_Register:
 		// A re-registration on a live stream means the agent's view of its own
 		// hardware changed — a GPU fell out, say. Take it.
@@ -511,3 +526,46 @@ func (s *Service) send(nodeID string, msg *llmgwv1.ControlMessage) error {
 		return fmt.Errorf("node %s is not keeping up with commands", nodeID)
 	}
 }
+
+// Logs asks a node for the recent lines of one of its logs: "agent", or a
+// shard id for that engine's output.
+//
+// It returns the lines, and every source the node can show.
+func (s *Service) Logs(ctx context.Context, nodeID, source string, lines int) ([]string, []string, error) {
+	s.logMu.Lock()
+	if s.logWaits == nil {
+		s.logWaits = map[string]chan *llmgwv1.LogsReply{}
+	}
+	s.logSeq++
+	id := fmt.Sprintf("logs-%d", s.logSeq)
+	ch := make(chan *llmgwv1.LogsReply, 1)
+	s.logWaits[id] = ch
+	s.logMu.Unlock()
+	defer func() {
+		s.logMu.Lock()
+		delete(s.logWaits, id)
+		s.logMu.Unlock()
+	}()
+
+	if source == "" {
+		source = "agent"
+	}
+	if err := s.send(nodeID, &llmgwv1.ControlMessage{Msg: &llmgwv1.ControlMessage_Logs{
+		Logs: &llmgwv1.LogsRequest{RequestId: id, Source: source, Lines: int32(lines)}, //nolint:gosec // bounded by the caller
+	}}); err != nil {
+		return nil, nil, err
+	}
+	// Bounded, because a page is waiting: an agent older than this request
+	// never answers it at all.
+	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+	select {
+	case r := <-ch:
+		if r.GetError() != "" {
+			return nil, r.GetSources(), errors.New(r.GetError())
+		}
+		return r.GetLines(), r.GetSources(), nil
+	case <-ctx.Done():
+		return nil, nil, fmt.Errorf("node %s did not answer; an agent older than v0.8.4 cannot", nodeID)
+	}
+}

internal/control/control.go

@@ -204,6 +204,17 @@ func (c *Controller) WeightRemover() WeightRemover {
 	return r
 }
 
+// LogReader can ask a node for the recent lines of one of its logs.
+type LogReader interface {
+	Logs(ctx context.Context, nodeID, source string, lines int) (out, sources []string, err error)
+}
+
+// LogReader returns the dispatcher's log capability, or nil.
+func (c *Controller) LogReader() LogReader {
+	r, _ := c.disp.(LogReader)
+	return r
+}
+
 // Start runs the reconciler and begins publishing endpoints.
 func (c *Controller) Start(ctx context.Context) {
 	if c.lead != nil {

internal/router/handlers.go

@@ -157,6 +157,8 @@ func (r *Router) Handler() http.Handler {
 				web.NavExtras{
 					Mirror: r.mirror != nil, History: r.history, Window: r.histWindow,
 				}, r.log)))
+			mux.HandleFunc("GET /admin/nodes/{id}/logs", pages.AdminPage(
+				web.NodeLogsPage(r.clusterView, false, version.Version)))
 			pages.RegisterAdminUsers(mux, r.store)
 			// Quotas are written on the same tree the accounts page builds, so
 			// they are mounted together.

internal/web/node_logs.go

@@ -0,0 +1,68 @@
+package web
+
+import (
+	"context"
+	"net/http"
+	"strconv"
+	"time"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
+)
+
+// LogSource reads a node's logs. Asked for by type assertion on the cluster
+// source, so a source that cannot — a test's, an older one — leaves the page
+// saying so rather than every Source having to grow the method.
+type LogSource interface {
+	NodeLogs(ctx context.Context, node, source string, lines int) adminapi.LogsView
+}
+
+type nodeLogsData struct {
+	adminapi.LogsView
+	OnAdminListener bool
+	// Base is where this page lives, which differs between the admin
+	// listener and the client-facing port.
+	Base    string
+	Back    string
+	Count   int
+	Counts  []int
+	Follow  bool
+	Version string
+	Now     string
+}
+
+// NodeLogsPage shows the recent lines of one of a node's logs: the agent's
+// own, or an engine's output.
+//
+// What a node did used to be learned by ssh through a jump host and a journal
+// only root may read — which, when an engine install fails at two in the
+// morning, is several steps between the page that says it failed and the line
+// that says why.
+func NodeLogsPage(src Source, onAdminListener bool, version string) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		node := r.PathValue("id")
+		d := nodeLogsData{
+			OnAdminListener: onAdminListener,
+			Base:            "/admin/nodes/" + node + "/logs",
+			Back:            "/admin",
+			Counts:          []int{100, 300, 1000, 2000},
+			Follow:          r.URL.Query().Get("follow") != "",
+			Version:         version,
+			Now:             time.Now().UTC().Format(time.RFC3339),
+		}
+		if onAdminListener {
+			d.Base, d.Back = "/nodes/"+node+"/logs", "/"
+		}
+		d.Count, _ = strconv.Atoi(r.URL.Query().Get("lines"))
+		if d.Count <= 0 {
+			d.Count = 300
+		}
+		ls, ok := src.(LogSource)
+		if !ok {
+			d.LogsView = adminapi.LogsView{Node: node, Error: "this control plane cannot read node logs"}
+		} else {
+			d.LogsView = ls.NodeLogs(r.Context(), node, r.URL.Query().Get("source"), d.Count)
+		}
+		noStore(w)
+		render(w, "node_logs", d)
+	}
+}

internal/web/node_logs_test.go

@@ -0,0 +1,31 @@
+package web
+
+import (
+	"context"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
+)
+
+type logsFleet struct{ fakeSource }
+
+func (logsFleet) NodeLogs(_ context.Context, node, source string, _ int) adminapi.LogsView {
+	return adminapi.LogsView{Node: node, Source: source, Lines: []string{"installed an engine <b>"},
+		Sources: []string{"agent", "shard-1"}}
+}
+
+func TestANodesLogIsShownWithItsOtherSources(t *testing.T) {
+	req := httptest.NewRequest("GET", "/admin/nodes/n6/logs?source=agent", nil)
+	req.SetPathValue("id", "n6")
+	rec := httptest.NewRecorder()
+	NodeLogsPage(logsFleet{}, false, "test")(rec, req)
+
+	body := rec.Body.String()
+	for _, want := range []string{"installed an engine &lt;b&gt;", "source=shard-1", "/admin/nodes/n6/logs"} {
+		if !strings.Contains(body, want) {
+			t.Errorf("page lacks %q", want)
+		}
+	}
+}

internal/web/templates/dashboard.html

@@ -103,7 +103,8 @@ requested whenever a fallback served it.</p>
 {{if .Nodes}}
 <table><tr><th>Node</th><th>State</th><th class="num">GPUs</th><th class="num">RAM</th><th class="num">Disk</th><th>Engines</th><th>Classes</th><th>Fabrics</th><th class="num">Models</th><th class="num">Replicas</th><th>Last seen</th></tr>
 {{range .Nodes}}<tr>
-  <td><code>{{.ID}}</code>{{if .MgmtAddr}} <span class="note">{{.MgmtAddr}}</span>{{end}}</td>
+  <td><code>{{.ID}}</code>{{if .MgmtAddr}} <span class="note">{{.MgmtAddr}}</span>{{end}}
+    <a class="note" href="{{if $.OnAdminListener}}/nodes/{{.ID}}/logs{{else}}/admin/nodes/{{.ID}}/logs{{end}}">logs</a></td>
   <td class="state state-{{.State}}">{{.State}}</td>
   <td class="num">{{.GPUFree}}/{{.GPUTotal}}</td>
   <td class="num">{{if .MemoryGB}}{{.MemoryGB}}G{{else}}—{{end}}</td>

internal/web/templates/node_logs.html

@@ -0,0 +1,31 @@
+{{define "node_logs"}}<!doctype html>
+<html lang="en"><head>
+<meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
+{{if .Follow}}<meta http-equiv="refresh" content="5">{{end}}
+<title>llmgw — {{.Node}} logs</title><link rel="stylesheet" href="/_static/base.css">
+</head><body><div class="wrap">
+<header>
+  <h1>{{.Node}}</h1>
+  <p class="sub">{{if eq .Source "agent"}}the agent's log{{else}}output of engine <code>{{.Source}}</code>{{end}} · last {{.Count}} lines</p>
+  <nav><a href="{{.Back}}">cluster</a></nav>
+</header>
+
+<p>
+{{- $d := . -}}
+{{range .Sources}}<a href="{{$d.Base}}?source={{.}}&lines={{$d.Count}}{{if $d.Follow}}&follow=1{{end}}"{{if eq . $d.Source}} aria-current="page"><strong>{{.}}</strong>{{else}}>{{.}}{{end}}</a> &nbsp; {{end}}
+</p>
+<p class="note">
+{{range .Counts}}<a href="{{$d.Base}}?source={{$d.Source}}&lines={{.}}{{if $d.Follow}}&follow=1{{end}}">{{.}}</a> {{end}}lines ·
+{{if .Follow}}<a href="{{.Base}}?source={{.Source}}&lines={{.Count}}">stop following</a>{{else}}<a href="{{.Base}}?source={{.Source}}&lines={{.Count}}&follow=1">follow</a> (refresh every 5s){{end}}
+</p>
+
+{{if .Error}}<p class="bad">{{.Error}}</p>{{end}}
+{{if .Lines}}<pre id="log">{{range .Lines}}{{.}}
+{{end}}</pre>
+{{/* Newest at the bottom, like a terminal, and scrolled there: the line
+     worth reading is almost always the last one. */}}
+<script>window.scrollTo(0, document.body.scrollHeight);</script>
+{{else if not .Error}}<p class="empty">Nothing logged yet.</p>{{end}}
+
+<footer class="note">llmgw {{.Version}} · {{.Now}}</footer>
+</div></body></html>{{end}}