telemetry: expose agent version info via data-api and data-cli (#3530)

## Summary of Changes

- Add `GET /device-link/agent-versions` endpoint to data-api and
`agent-versions` subcommand to data-cli, surfacing the telemetry agent
version and commit hash from `DeviceLatencySamplesHeader` written
onchain since #3506
- Add `GetDeviceLatencySamplesHeader` SDK method using Solana RPC
`DataSlice` to fetch only the 350-byte header instead of the full ~140KB
account
- Extract `ToV1Header()` on `DeviceLatencySamplesHeaderV0` for
header-only V0→V1 conversion, reused by existing `ToV1()`
- Bound concurrency in `GetAgentVersions` with a semaphore (cap 64) to
prevent unbounded goroutine fan-out

## Diff Breakdown

| Category     | Files | Lines (+/-)  | Net  |
|--------------|-------|--------------|------|
| Core logic   |     3 | +211 / -23   | +188 |
| Scaffolding  |     6 | +129 / -0    | +129 |
| Tests        |     5 | +368 / -1    | +367 |

~340 lines of non-test code; roughly half is core logic, half is wiring.

<details>
<summary>Key files (click to expand)</summary>

-
[`controlplane/telemetry/internal/data/device/agent_versions.go`](https://github.com/malbeclabs/doublezero/pull/3530/files#diff-0ead065494d1fcf29685878803a66c06bfaa5fa494008fd8202266177cf1ada4)
— Provider method that groups circuits by origin device, fetches headers
(current + previous epoch fallback), and returns version/commit per
device
-
[`controlplane/telemetry/internal/data/cli/agent_versions.go`](https://github.com/malbeclabs/doublezero/pull/3530/files#diff-405da9067b67de87fa28b31337f60a06ab176732a464bf90a31fff58f660b899)
— CLI subcommand that calls GetAgentVersions and renders a table
-
[`smartcontract/sdk/go/telemetry/client.go`](https://github.com/malbeclabs/doublezero/pull/3530/files#diff-80b8d7ca3aa6fd5d642584653ca346aee5be57bc77e79a89db885b782566c0ff)
— New `GetDeviceLatencySamplesHeader` using DataSlice to fetch only 350
bytes, with V0/V1 deserialization
-
[`smartcontract/sdk/go/telemetry/state_v0.go`](https://github.com/malbeclabs/doublezero/pull/3530/files#diff-8f5df3efc61241621c8867509a324f0532369db21cf40d38fe2467362c927fac)
— Extract `ToV1Header()` from `ToV1()` for header-only conversion
without samples
-
[`controlplane/telemetry/internal/data/device/server.go`](https://github.com/malbeclabs/doublezero/pull/3530/files#diff-c60979e397bc54fb0d2a666df2410d9d93b24537a1a23bd17b43ec276104e664)
— Register and handle `/device-link/agent-versions` route

</details>

## Testing Verification

- New provider tests for `GetAgentVersions`: valid headers,
account-not-found skipping, zero-sample-index skipping, staleness
filtering (>24h), epoch fallback, empty circuits
- New HTTP handler tests for `/device-link/agent-versions`: success
response, invalid env (400), empty results, provider error (500)
- Existing SDK V0/V1 serialization round-trip tests pass after
`ToV1Header` refactor

Author: nikw9944

CHANGELOG.md

@@ -8,6 +8,8 @@ All notable changes to this project will be documented in this file.
 
 ### Changes
 
+- Telemetry
+  - Add `GET /device-link/agent-versions` endpoint to data-api and `agent-versions` subcommand to data-cli, exposing per-device telemetry agent version and commit from onchain `DeviceLatencySamplesHeader`
 - Smartcontract
   - Allow `SubscribeMulticastGroup` for users in `Pending` status so that `CreateSubscribeUser` can be followed by additional subscribe calls before the activator runs ([#3521](https://github.com/malbeclabs/doublezero/pull/3521))
   - Add optional `owner` field to `UpdateMulticastGroup` instruction, allowing foundation members to reassign ownership of a multicast group ([#3527](https://github.com/malbeclabs/doublezero/pull/3527))

controlplane/telemetry/internal/data/cli/agent_versions.go

@@ -0,0 +1,88 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	devicedata "github.com/malbeclabs/doublezero/controlplane/telemetry/internal/data/device"
+	"github.com/olekukonko/tablewriter"
+	"github.com/spf13/cobra"
+)
+
+type AgentVersionsCmd struct{}
+
+func NewAgentVersionsCmd() *AgentVersionsCmd {
+	return &AgentVersionsCmd{}
+}
+
+func (c *AgentVersionsCmd) Command() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "agent-versions",
+		Short: "Show telemetry agent version for each device",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			verbose, err := cmd.Root().PersistentFlags().GetBool("verbose")
+			if err != nil {
+				return fmt.Errorf("failed to get verbose flag: %w", err)
+			}
+			env, err := cmd.Root().PersistentFlags().GetString("env")
+			if err != nil {
+				return fmt.Errorf("failed to get env flag: %w", err)
+			}
+
+			log := newLogger(verbose)
+
+			ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+			defer cancel()
+
+			provider, _, err := newDeviceProvider(log, env)
+			if err != nil {
+				log.Error("Failed to get provider", "error", err)
+				os.Exit(1)
+			}
+
+			versions, err := provider.GetAgentVersions(ctx)
+			if err != nil {
+				log.Error("Failed to get agent versions", "error", err)
+				os.Exit(1)
+			}
+
+			printAgentVersions(versions, env)
+			return nil
+		},
+	}
+
+	return cmd
+}
+
+func printAgentVersions(versions []devicedata.DeviceAgentVersion, env string) {
+	fmt.Println("Environment:", env)
+	fmt.Printf("Devices reporting: %d\n", len(versions))
+
+	table := tablewriter.NewWriter(os.Stdout)
+	table.SetAutoWrapText(false)
+	table.SetHeaderAlignment(tablewriter.ALIGN_CENTER)
+	table.SetAutoFormatHeaders(false)
+	table.SetBorder(true)
+	table.SetRowLine(true)
+	table.SetHeader([]string{
+		"Device PK",
+		"Device Code",
+		"Version",
+		"Commit",
+		"Last Sample",
+	})
+
+	for _, v := range versions {
+		table.Append([]string{
+			v.DevicePK,
+			v.DeviceCode,
+			v.Version,
+			v.Commit,
+			v.Timestamp,
+		})
+	}
+	table.Render()
+}

controlplane/telemetry/internal/data/cli/root.go

@@ -40,6 +40,7 @@ func Run() ExitCode {
 	rootCmd.AddCommand(
 		NewDeviceCmd().Command(),
 		NewInternetCmd().Command(),
+		NewAgentVersionsCmd().Command(),
 	)
 
 	if err := rootCmd.Execute(); err != nil {

controlplane/telemetry/internal/data/device/agent_versions.go

@@ -0,0 +1,125 @@
+package data
+
+import (
+	"context"
+	"errors"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/gagliardetto/solana-go"
+	"github.com/malbeclabs/doublezero/smartcontract/sdk/go/telemetry"
+)
+
+const maxAgentVersionStaleness = 24 * time.Hour
+
+func (p *provider) GetAgentVersions(ctx context.Context) ([]DeviceAgentVersion, error) {
+	circuits, err := p.GetCircuits(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	currentEpoch, err := p.cfg.EpochFinder.ApproximateAtTime(ctx, time.Now().UTC())
+	if err != nil {
+		return nil, err
+	}
+
+	// Group circuits by origin device, keeping one representative circuit per device.
+	type deviceCircuit struct {
+		devicePK   solana.PublicKey
+		deviceCode string
+		circuit    Circuit
+	}
+	seen := map[solana.PublicKey]struct{}{}
+	var devices []deviceCircuit
+	for _, c := range circuits {
+		if _, ok := seen[c.OriginDevice.PK]; ok {
+			continue
+		}
+		seen[c.OriginDevice.PK] = struct{}{}
+		devices = append(devices, deviceCircuit{
+			devicePK:   c.OriginDevice.PK,
+			deviceCode: c.OriginDevice.Code,
+			circuit:    c,
+		})
+	}
+
+	var mu sync.Mutex
+	var wg sync.WaitGroup
+	var results []DeviceAgentVersion
+	now := time.Now().UTC()
+
+	sem := make(chan struct{}, defaultGetCircuitLatenciesPoolSize)
+	for _, dev := range devices {
+		wg.Add(1)
+		sem <- struct{}{}
+		go func(dev deviceCircuit) {
+			defer func() { <-sem; wg.Done() }()
+
+			// Try current epoch first, then the previous epoch.
+			var hdr *telemetry.DeviceLatencySamplesHeader
+			for _, ep := range []uint64{currentEpoch, currentEpoch - 1} {
+				h, err := p.cfg.TelemetryClient.GetDeviceLatencySamplesHeader(
+					ctx,
+					dev.circuit.OriginDevice.PK,
+					dev.circuit.TargetDevice.PK,
+					dev.circuit.Link.PK,
+					ep,
+				)
+				if err != nil {
+					if errors.Is(err, telemetry.ErrAccountNotFound) {
+						continue
+					}
+					p.log.Warn("failed to get samples header", "device", dev.deviceCode, "epoch", ep, "error", err)
+					continue
+				}
+				if h.NextSampleIndex == 0 {
+					continue
+				}
+				hdr = h
+				break
+			}
+
+			if hdr == nil || hdr.NextSampleIndex == 0 {
+				return
+			}
+
+			ts := lastSampleTime(hdr)
+			if now.Sub(ts) > maxAgentVersionStaleness {
+				return
+			}
+
+			version := strings.TrimRight(string(hdr.AgentVersion[:]), "\x00")
+			commit := strings.TrimRight(string(hdr.AgentCommit[:]), "\x00")
+
+			mu.Lock()
+			results = append(results, DeviceAgentVersion{
+				DevicePK:   dev.devicePK.String(),
+				DeviceCode: dev.deviceCode,
+				Version:    version,
+				Commit:     commit,
+				Timestamp:  ts.Format(time.RFC3339),
+			})
+			mu.Unlock()
+		}(dev)
+	}
+	wg.Wait()
+
+	sort.Slice(results, func(i, j int) bool {
+		return results[i].DeviceCode < results[j].DeviceCode
+	})
+
+	return results, nil
+}
+
+func lastSampleTime(hdr *telemetry.DeviceLatencySamplesHeader) time.Time {
+	if hdr.NextSampleIndex == 0 {
+		return time.Time{}
+	}
+	tsMicros := hdr.StartTimestampMicroseconds +
+		uint64(hdr.NextSampleIndex-1)*hdr.SamplingIntervalMicroseconds
+	secs := int64(tsMicros / 1_000_000)
+	nanos := int64(tsMicros%1_000_000) * 1000
+	return time.Unix(secs, nanos)
+}