agents_and_robots/internal/api/poller.go

package api

import (
	"context"
	"time"

	"github.com/enmanuel/agents/shell/process"
)

// StatusDiff is published to the "status" topic when an agent's running state changes.
type StatusDiff struct {
	AgentID   string `json:"agent_id"`
	OldStatus bool   `json:"old_running"`
	NewStatus bool   `json:"new_running"`
	PID       int    `json:"pid,omitempty"`
}

// pollStatus polls StatusAll every 2s and publishes StatusDiff events on changes.
func (s *Server) pollStatus(ctx context.Context) {
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	// Seed the previous state map.
	prev := make(map[string]bool)
	if statuses, err := s.statusAllAuto(); err == nil {
		for _, st := range statuses {
			prev[st.ID] = st.Running
		}
	}

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.checkAndPublishDiffs(prev)
		}
	}
}

func (s *Server) checkAndPublishDiffs(prev map[string]bool) {
	statuses, err := s.statusAllAuto()
	if err != nil {
		return
	}

	for _, st := range statuses {
		old, known := prev[st.ID]
		if !known || old != st.Running {
			s.bus.Publish("status", StatusDiff{
				AgentID:   st.ID,
				OldStatus: old,
				NewStatus: st.Running,
				PID:       st.PID,
			})
			prev[st.ID] = st.Running
		}
	}

	// Handle agents that were removed (disappeared from scan)
	current := make(map[string]bool, len(statuses))
	for _, st := range statuses {
		current[st.ID] = true
	}
	for id, wasRunning := range prev {
		if !current[id] {
			if wasRunning {
				s.bus.Publish("status", StatusDiff{
					AgentID:   id,
					OldStatus: true,
					NewStatus: false,
				})
			}
			delete(prev, id)
		}
	}
}

// agentInfoByID finds AgentInfo by ID in a StatusAll scan.
func agentInfoByID(mgr *process.Manager, id string) (*process.AgentInfo, error) {
	agents, err := mgr.Scan()
	if err != nil {
		return nil, err
	}
	for i, a := range agents {
		if a.ID == id {
			return &agents[i], nil
		}
	}
	return nil, nil
}