Jf/onboarding clarity enhancement clean

decentralized-api/apiconfig/constants.go

@@ -0,0 +1,27 @@
+package apiconfig
+
+// MLNodeOnboardingState represents the onboarding state of an ML node
+type MLNodeOnboardingState string
+
+// Constants for MLNodeOnboardingState
+const (
+	MLNodeState_WAITING_FOR_POC MLNodeOnboardingState = "WAITING_FOR_POC"
+	MLNodeState_TESTING         MLNodeOnboardingState = "TESTING"
+	MLNodeState_TEST_FAILED     MLNodeOnboardingState = "TEST_FAILED"
+)
+
+// Timing constants used across broker/admin components
+const (
+	DefaultBlockTimeSeconds           = 6.0
+	AutoTestMinSecondsBeforePoC int64 = 3600
+	OnlineAlertLeadSeconds      int64 = 600
+)
+
+// ParticipantState represents the state of a participant
+type ParticipantState string
+
+// Constants for ParticipantState
+const (
+	ParticipantState_INACTIVE_WAITING     ParticipantState = "INACTIVE_WAITING"
+	ParticipantState_ACTIVE_PARTICIPATING ParticipantState = "ACTIVE_PARTICIPATING"
+)

decentralized-api/broker/broker.go

@@ -107,22 +107,23 @@ func (b *BrokerChainBridgeImpl) GetParams() (*types.QueryParamsResponse, error)
 }
 
 type Broker struct {
-	highPriorityCommands chan Command
-	lowPriorityCommands  chan Command
-	nodes                map[string]*NodeWithState
-	mu                   sync.RWMutex
-	curMaxNodesNum       atomic.Uint64
-	chainBridge          BrokerChainBridge
-	nodeWorkGroup        *NodeWorkGroup
-	phaseTracker         *chainphase.ChainPhaseTracker
-	participantInfo      participant.CurrenParticipantInfo
-	callbackUrl          string
-	mlNodeClientFactory  mlnodeclient.ClientFactory
-	reconcileTrigger     chan struct{}
-	lastEpochIndex       uint64
-	lastEpochPhase       types.EpochPhase
-	statusQueryTrigger   chan statusQuerySignal
-	configManager        *apiconfig.ConfigManager
+	highPriorityCommands  chan Command
+	lowPriorityCommands   chan Command
+	nodes                 map[string]*NodeWithState
+	mu                    sync.RWMutex
+	curMaxNodesNum        atomic.Uint64
+	chainBridge           BrokerChainBridge
+	nodeWorkGroup         *NodeWorkGroup
+	phaseTracker          *chainphase.ChainPhaseTracker
+	participantInfo       participant.CurrenParticipantInfo
+	callbackUrl           string
+	mlNodeClientFactory   mlnodeclient.ClientFactory
+	reconcileTrigger      chan struct{}
+	lastEpochIndex        uint64
+	lastEpochPhase        types.EpochPhase
+	lastParticipantWeight int64
+	statusQueryTrigger    chan statusQuerySignal
+	configManager         *apiconfig.ConfigManager
 }
 
 // GetParticipantAddress returns the current participant's address if available.
@@ -254,6 +255,14 @@ type NodeState struct {
 	// Epoch-specific data, populated from the chain
 	EpochModels  map[string]types.Model      `json:"epoch_models"`
 	EpochMLNodes map[string]types.MLNodeInfo `json:"epoch_ml_nodes"`
+
+	Timing *TimingInfo `json:"timing,omitempty"`
+
+	UserMessage           string `json:"user_message,omitempty"`
+	Guidance              string `json:"guidance,omitempty"`
+	ParticipantState      string `json:"participant_state,omitempty"`
+	MLNodeOnboardingState string `json:"mlnode_state,omitempty"`
+	ParticipantWeight     int64  `json:"participant_weight,omitempty"`
 }
 
 func (s NodeState) MarshalJSON() ([]byte, error) {
@@ -349,6 +358,13 @@ type NodeResponse struct {
 	State NodeState `json:"state"`
 }
 
+type TimingInfo struct {
+	CurrentPhase        string `json:"current_phase"`
+	BlocksUntilNextPoC  int64  `json:"blocks_until_next_poc"`
+	SecondsUntilNextPoC int64  `json:"seconds_until_next_poc"`
+	ShouldBeOnline      bool   `json:"should_be_online"`
+}
+
 func NewBroker(chainBridge BrokerChainBridge, phaseTracker *chainphase.ChainPhaseTracker, participantInfo participant.CurrenParticipantInfo, callbackUrl string, clientFactory mlnodeclient.ClientFactory, configManager *apiconfig.ConfigManager) *Broker {
 	broker := &Broker{
 		highPriorityCommands: make(chan Command, 100),
@@ -373,6 +389,18 @@ func NewBroker(chainBridge BrokerChainBridge, phaseTracker *chainphase.ChainPhas
 	// go nodeReconciliationWorker(broker)
 	go nodeStatusQueryWorker(broker)
 	go broker.reconcilerLoop()
+
+	// Startup: try to populate epoch data once chain is synced to expose participant status early
+	go func() {
+		for i := 0; i < 10; i++ {
+			es := broker.phaseTracker.GetCurrentEpochState()
+			if es != nil && es.IsSynced {
+				_ = broker.UpdateNodeWithEpochData(es)
+				return
+			}
+			time.Sleep(1 * time.Second)
+		}
+	}()
 	return broker
 }
 
@@ -472,6 +500,10 @@ func (b *Broker) executeCommand(command Command) {
 		command.Execute(b)
 	case UpdateNodeResultCommand:
 		command.Execute(b)
+	case SetNodeFailureReasonCommand:
+		command.Execute(b)
+	case SetNodeMLNodeOnboardingStateCommand:
+		command.Execute(b)
 	default:
 		logging.Error("Unregistered command type", types.Nodes, "type", reflect.TypeOf(command).String())
 	}
@@ -492,6 +524,8 @@ func (b *Broker) QueueMessage(command Command) error {
 	switch command.(type) {
 	case StartPocCommand, InitValidateCommand, InferenceUpAllCommand, UpdateNodeResultCommand, SetNodesActualStatusCommand, SetNodeAdminStateCommand, RegisterNode, RemoveNode, StartTrainingCommand, LockNodesForTrainingCommand, SyncNodesCommand:
 		b.highPriorityCommands <- command
+	case SetNodeFailureReasonCommand:
+		b.highPriorityCommands <- command
 	default:
 		b.lowPriorityCommands <- command
 	}
@@ -1522,6 +1556,9 @@ func (b *Broker) UpdateNodeWithEpochData(epochState *chainphase.EpochState) erro
 
 	parentEpochData := parentGroupResp.GetEpochGroupData()
 
+	// Calculate current participant weight by scanning validation weights across subgroups
+	currentWeight := int64(0)
+
 	b.clearNodeEpochData()
 
 	// 2. Track which nodes are found in epoch data
@@ -1549,6 +1586,12 @@ func (b *Broker) UpdateNodeWithEpochData(epochState *chainphase.EpochState) erro
 		for _, weightInfo := range subgroup.ValidationWeights {
 			// Check if the participant is the one this broker is managing
 			if weightInfo.MemberAddress == b.participantInfo.GetAddress() {
+				// Track participant weight (use ConfirmationWeight if present, else Weight)
+				w := weightInfo.ConfirmationWeight
+				if w == 0 {
+					w = weightInfo.Weight
+				}
+				currentWeight += w
 				// 5. Iterate through the ML nodes for this participant in the epoch data
 				b.UpdateNodeEpochData(weightInfo.MlNodes, modelId, *subgroup.ModelSnapshot)
 				// Mark these nodes as found in epoch
@@ -1559,6 +1602,19 @@ func (b *Broker) UpdateNodeWithEpochData(epochState *chainphase.EpochState) erro
 		}
 	}
 
+	// If participant weight changed, log and update cached weight
+	if currentWeight != b.lastParticipantWeight {
+		logging.Info("Participant weight changed", types.Participants, "old", b.lastParticipantWeight, "new", currentWeight, "epoch", epochState.LatestEpoch.EpochIndex)
+		b.lastParticipantWeight = currentWeight
+	}
+
+	// Store participant weight on each node state for visibility in admin APIs
+	b.mu.Lock()
+	for _, node := range b.nodes {
+		node.State.ParticipantWeight = currentWeight
+	}
+	b.mu.Unlock()
+
 	// 6. Populate governance models for nodes not in epoch data (disabled nodes)
 	b.mu.RLock()
 	nodeIds := make([]string, 0, len(b.nodes))
@@ -1792,3 +1848,33 @@ func (b *Broker) MergeModelArgs(epochArgs []string, localArgs []string) []string
 
 	return mergedArgs
 }
+
+func (b *Broker) IsParticipantActiveOnChain() (bool, error) {
+	resp, err := b.chainBridge.GetCurrentEpochGroupData()
+	if err != nil {
+		return false, err
+	}
+	if resp == nil {
+		return false, nil
+	}
+	epochIndex := resp.EpochGroupData.EpochIndex
+	subModels := resp.EpochGroupData.SubGroupModels
+	addr := b.participantInfo.GetAddress()
+	for _, mid := range subModels {
+		subgroupResp, err := b.chainBridge.GetEpochGroupDataByModelId(epochIndex, mid)
+		if err != nil {
+			continue
+		}
+		if subgroupResp == nil {
+			continue
+		}
+		for _, w := range subgroupResp.EpochGroupData.ValidationWeights {
+			if w.MemberAddress == addr {
+				if len(w.MlNodes) > 0 {
+					return true, nil
+				}
+			}
+		}
+	}
+	return false, nil
+}

decentralized-api/broker/commands.go

@@ -50,6 +50,27 @@ func (c GetNodesCommand) Execute(b *Broker) {
 	b.mu.RLock()
 	defer b.mu.RUnlock()
 
+	// Precompute timing information from current epoch state
+	var (
+		blocksUntilNextPoC  int64
+		secondsUntilNextPoC int64
+		currentPhase        types.EpochPhase
+		hasEpochInfo        bool
+	)
+	epochState := b.phaseTracker.GetCurrentEpochState()
+	if epochState != nil && epochState.IsSynced {
+		hasEpochInfo = true
+		currentPhase = epochState.CurrentPhase
+		currentHeight := epochState.CurrentBlock.Height
+		nextPoC := epochState.LatestEpoch.NextPoCStart()
+		blocksUntilNextPoC = nextPoC - currentHeight
+		if blocksUntilNextPoC < 0 {
+			blocksUntilNextPoC = 0
+		}
+		// Use default block time constant
+		secondsUntilNextPoC = int64(float64(blocksUntilNextPoC) * apiconfig.DefaultBlockTimeSeconds)
+	}
+
 	nodeResponses := make([]NodeResponse, 0, len(b.nodes))
 	for _, nodeWithState := range b.nodes {
 		// --- Deep copy Node ---
@@ -99,6 +120,15 @@ func (c GetNodesCommand) Execute(b *Broker) {
 			Node:  nodeCopy,
 			State: stateCopy,
 		})
+		if hasEpochInfo {
+			shouldOnline := currentPhase == types.PoCGeneratePhase || currentPhase == types.PoCGenerateWindDownPhase || currentPhase == types.PoCValidatePhase || currentPhase == types.PoCValidateWindDownPhase || secondsUntilNextPoC <= apiconfig.OnlineAlertLeadSeconds
+			nodeResponses[len(nodeResponses)-1].State.Timing = &TimingInfo{
+				CurrentPhase:        string(currentPhase),
+				BlocksUntilNextPoC:  blocksUntilNextPoC,
+				SecondsUntilNextPoC: secondsUntilNextPoC,
+				ShouldBeOnline:      shouldOnline,
+			}
+		}
 	}
 	logging.Debug("Got nodes", types.Nodes, "size", len(nodeResponses))
 	c.Response <- nodeResponses
@@ -242,6 +272,9 @@ func (c UpdateNodeResultCommand) Execute(b *Broker) {
 		return
 	}
 
+	prevStatus := node.State.CurrentStatus
+	prevFailure := node.State.FailureReason
+
 	// Update state
 	logging.Info("Finalizing state transition for node", types.Nodes,
 		"node_id", c.NodeId,
@@ -260,6 +293,9 @@ func (c UpdateNodeResultCommand) Execute(b *Broker) {
 	} else {
 		// Clear failure reason on success
 		node.State.FailureReason = ""
+		if prevFailure != "" {
+			logging.Info("Node status recovered", types.Nodes, "node_id", c.NodeId, "blockHeight", blockHeight)
+		}
 	}
 
 	// Reset POC fields when moving away from POC status
@@ -268,5 +304,41 @@ func (c UpdateNodeResultCommand) Execute(b *Broker) {
 		node.State.PocCurrentStatus = PocStatusIdle
 	}
 
+	if prevStatus == types.HardwareNodeStatus_POC && c.Result.FinalStatus == types.HardwareNodeStatus_INFERENCE {
+		logging.Info("Onboarding transition POC->INFERENCE", types.Nodes, "node_id", c.NodeId, "blockHeight", blockHeight)
+	}
+
+	c.Response <- true
+}
+
+// SetNodeFailureReasonCommand sets the FailureReason field on a node state directly
+// without requiring an in-flight reconciliation.
+type SetNodeFailureReasonCommand struct {
+	NodeId   string
+	Reason   string
+	Response chan bool
+}
+
+func NewSetNodeFailureReasonCommand(nodeId string, reason string) SetNodeFailureReasonCommand {
+	return SetNodeFailureReasonCommand{
+		NodeId:   nodeId,
+		Reason:   reason,
+		Response: make(chan bool, 2),
+	}
+}
+
+func (c SetNodeFailureReasonCommand) GetResponseChannelCapacity() int { return cap(c.Response) }
+
+func (c SetNodeFailureReasonCommand) Execute(b *Broker) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	node, exists := b.nodes[c.NodeId]
+	if !exists {
+		logging.Warn("SetNodeFailureReason: node not found", types.Nodes, "node_id", c.NodeId)
+		c.Response <- false
+		return
+	}
+	node.State.FailureReason = c.Reason
 	c.Response <- true
 }

decentralized-api/broker/lock_helpers.go

@@ -221,7 +221,9 @@ func DoWithLockedNodeHTTPRetry(
 			}
 			outcome = InferenceError{Message: msg}
 		}
-		_ = b.QueueMessage(ReleaseNode{NodeId: node.Id, Outcome: outcome, Response: make(chan bool, 2)})
+		if err := b.QueueMessage(ReleaseNode{NodeId: node.Id, Outcome: outcome, Response: make(chan bool, 2)}); err != nil {
+			logging.Warn("Failed to queue ReleaseNode message", types.Inferences, "node_id", node.Id, "error", err)
+		}
 
 		if retry {
 			if triggerRecheck {

decentralized-api/broker/node_admin_commands.go

@@ -169,6 +169,8 @@ func (c RegisterNode) Execute(b *Broker) {
 
 	logging.Info("RegisterNode. Registered node", types.Nodes, "node", c.Node)
 	c.Response <- NodeCommandResponse{Node: &c.Node, Error: nil}
+
+	// Auto-test now handled by admin orchestrator
 }
 
 // UpdateNode updates an existing node's configuration while preserving runtime state
@@ -266,6 +268,8 @@ func (c UpdateNode) Execute(b *Broker) {
 
 	logging.Info("UpdateNode. Updated node configuration", types.Nodes, "node_id", c.Node.Id)
 	c.Response <- NodeCommandResponse{Node: &c.Node, Error: nil}
+
+	// Auto-test now handled by admin orchestrator
 }
 
 type RemoveNode struct {
@@ -370,3 +374,43 @@ func (c UpdateNodeHardwareCommand) Execute(b *Broker) {
 	logging.Info("Updated node hardware", types.Nodes, "node_id", c.NodeId, "hardware_count", len(c.Hardware))
 	c.Response <- nil
 }
+
+// SetNodeMLNodeOnboardingStateCommand updates the MLNodeOnboardingState of a node
+type SetNodeMLNodeOnboardingStateCommand struct {
+	NodeId   string
+	NewState string
+	Response chan bool
+}
+
+func NewSetNodeMLNodeOnboardingStateCommand(nodeId string, newState string) SetNodeMLNodeOnboardingStateCommand {
+	return SetNodeMLNodeOnboardingStateCommand{
+		NodeId:   nodeId,
+		NewState: newState,
+		Response: make(chan bool, 2),
+	}
+}
+
+func (c SetNodeMLNodeOnboardingStateCommand) GetResponseChannelCapacity() int {
+	return cap(c.Response)
+}
+
+func (c SetNodeMLNodeOnboardingStateCommand) Execute(b *Broker) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	node, exists := b.nodes[c.NodeId]
+	if !exists {
+		logging.Error("Cannot set MLNodeOnboardingState: node not found", types.Nodes, "node_id", c.NodeId)
+		c.Response <- false
+		return
+	}
+
+	logging.Info("Setting MLNodeOnboardingState for node", types.Nodes,
+		"node_id", c.NodeId,
+		"old_state", node.State.MLNodeOnboardingState,
+		"new_state", c.NewState)
+
+	node.State.MLNodeOnboardingState = c.NewState
+
+	c.Response <- true
+}