Add unified connections API (list + probe) with Disabled flag

Introduces GET /api/connections and POST /api/connections/probe as the backend half of the one-ledger / one-editor connection redesign. - GET /api/connections aggregates PVE/PBS/PMG/VMware/TrueNAS/agent rows into a unified Connection shape with derived state (active, paused, unauthorized, unreachable, stale, pending) computed from in-memory scheduler health plus agent Host.LastSeen. No new persisted state. - POST /api/connections/probe fingerprints a host across the five supported products in parallel (2s dial + 1s read, 3s total, max 5 concurrent). Admin-gated (RequireAdmin + ScopeSettingsWrite) to block unauthenticated SSRF against internal hosts. - Disabled bool on PVEInstance/PBSInstance/PMGInstance (zero-value = enabled, preserves existing nodes.json); pollers skip disabled instances at client init, reconnect, and per-node iteration. - NodeConfigRequest/Response gain Enabled; write path translates *bool -> Disabled so omitted field leaves state untouched. - ConnectionsAPI frontend client (list/probe) typed off the Go shape. Contracts updated: api-contracts, monitoring, agent-lifecycle, performance-and-scalability, storage-recovery. Proofs added: contract_test.go JSON snapshot for Connection and ProbeResponse, monitoring guardrails for the Disabled-skip behavior, and a vitest mock-client test for ConnectionsAPI. Frontend editor / drawer / table rewrite lands in a separate block.
2026-09-10 02:25:56 +00:00 · 2026-04-19 11:42:53 +01:00
parent fae27b3a8f
commit 9c3d96cab2
27 changed files with 1856 additions and 5 deletions
@@ -303,6 +303,16 @@ an add-only capacity posture.
    VMware must not collect external vCenter inventory before that canonical
    capacity view is safe, so fleet/setup surfaces cannot bypass the
    monitored-system accounting boundary through direct API writes.
+    The same lifecycle-adjacent platform-connections boundary now also owns
+    the unified connections ledger (`GET /api/connections`) and address
+    probe (`POST /api/connections/probe`). Lifecycle surfaces may observe
+    agent `Host.LastSeen`-backed rows on that ledger, but must not
+    reinterpret derived `state` (active/paused/unauthorized/unreachable/
+    stale/pending) as install authority or treat the probe response as
+    enrollment state; ledger writes still flow through the per-type config
+    endpoints that own admission checks, and the `Disabled` flag on
+    PVE/PBS/PMG surfaced by that ledger must remain a pause-only signal
+    rather than an installer pre-flight gate.

 ## Forbidden Paths

@@ -79,6 +79,11 @@ Own canonical runtime payload shapes between backend and frontend.
 54. `internal/api/security_status_capabilities.go`
 55. `internal/api/demo_middleware.go`
 56. `frontend-modern/src/stores/aiRuntimeState.ts`
+57. `internal/api/connections_types.go`
+58. `internal/api/connections_aggregator.go`
+59. `internal/api/connections_handlers.go`
+60. `internal/api/connections_probe.go`
+61. `frontend-modern/src/api/connections.ts`

 ## Shared Boundaries

@@ -311,6 +316,19 @@ the canonical monitored-system blocked payload.
    presentation is active, so VMware, storage, and infrastructure series stay
    aligned with `/api/resources` and `/api/state` instead of drifting onto the
    live store-backed graph.
+39. Route the unified connections ledger and address probe through
+    `internal/api/connections_types.go`,
+    `internal/api/connections_aggregator.go`,
+    `internal/api/connections_handlers.go`,
+    `internal/api/connections_probe.go`, and
+    `frontend-modern/src/api/connections.ts` together so `GET /api/connections`
+    and `POST /api/connections/probe` stay on one canonical payload shape
+    instead of re-deriving state from per-type config stores in the frontend.
+    State must remain a derived field sourced from in-memory scheduler health
+    (`monitoring.Monitor.SchedulerHealth()`) plus agent `Host.LastSeen`; the
+    endpoint must not introduce new persisted per-connection state. The probe
+    endpoint must remain admin-gated (`RequireAdmin` + `ScopeSettingsWrite`)
+    to block unauthenticated SSRF against internal hosts.

 ## Forbidden Paths

@@ -71,6 +71,12 @@ truth for live infrastructure data.
 10. Add or change mock chart synthesis, seeded history continuity, or mock-owned
   chart fallbacks through `internal/monitoring/mock_metrics_history.go` and
   `internal/monitoring/mock_chart_history.go`
+11. Honor the per-instance `Disabled` flag on PVE/PBS/PMG at poller client
+   init, reconnect, and per-node iteration so disabled connections do not
+   drive API calls, scheduler health, or surface ingest. Zero-value
+   `Disabled=false` must remain the migration-safe default for existing
+   `nodes.json` content; the poller must never create a client or mark an
+   instance reachable when `Disabled` is true.

 ## Forbidden Paths

@@ -275,6 +275,16 @@ regression protection.
    colors, but they must not fall back to inline `style=` attributes on the
    public shell just to express virtualization spacers, alert accents, or
    workload metric bars.
+36. Keep the unified connections ledger off the polling hot path. Changes to
+    `internal/api/router.go` that register `/api/connections` and
+    `/api/connections/probe` must keep those routes off the monitoring fan-out
+    budget: `GET /api/connections` must resolve purely from
+    `monitoring.Monitor.SchedulerHealth()` plus existing per-type config
+    stores without triggering any live network probes, and
+    `POST /api/connections/probe` must remain bounded at 3s total /
+    2s dial / 1s read with at most 5 concurrent fingerprints so the probe
+    endpoint cannot be repurposed into a slow-leak scanner that starves the
+    dashboard hot path.

 ## Forbidden Paths

@@ -343,6 +343,18 @@ querying, and the operator-facing storage health presentation layer.
    canonical storage page model, recovery presenters, and shared summary
    caches. Header chrome must not become a second owner for storage filters,
    recovery posture, commercial purchase state, or transport selection.
+39. Keep the unified connections ledger owner-neutral toward storage and
+    recovery. Shared `internal/api/router.go` may mount the
+    `/api/connections` and `/api/connections/probe` routes alongside the
+    existing storage/recovery-adjacent API surfaces, and
+    `internal/api/config_handlers.go` and `internal/api/config_node_handlers.go`
+    may carry the new per-instance `Enabled`/`Disabled` round-trip, but
+    storage and recovery consumers must not reinterpret the derived
+    connection `state` (active/paused/unauthorized/unreachable/stale/pending)
+    as storage health, backup-job posture, or recovery verification state;
+    storage and recovery UI must keep sourcing those signals from their
+    existing canonical page models instead of polling the connections
+    ledger for per-datastore or per-backup truth.

 ## Forbidden Paths

@@ -0,0 +1,72 @@
+import { describe, expect, it, vi, beforeEach } from 'vitest';
+import { ConnectionsAPI, type Connection, type ProbeResponse } from '../connections';
+import { apiFetchJSON } from '@/utils/apiClient';
+
+vi.mock('@/utils/apiClient', () => ({
+  apiFetchJSON: vi.fn(),
+}));
+
+const mockedApiFetchJSON = vi.mocked(apiFetchJSON);
+
+describe('ConnectionsAPI', () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it('list() calls GET /api/connections and returns the connections array', async () => {
+    const connections: Connection[] = [
+      {
+        id: 'pve-lab',
+        type: 'pve',
+        name: 'lab',
+        address: 'https://pve.lab:8006',
+        state: 'active',
+        stateReason: '',
+        enabled: true,
+        surfaces: ['vms', 'containers'],
+        scope: { vms: true, containers: true },
+        lastSeen: '2026-04-19T10:00:00Z',
+        lastError: null,
+        source: 'manual',
+        capabilities: { supportsPause: true, supportsScope: true, supportsTest: true },
+      },
+    ];
+    mockedApiFetchJSON.mockResolvedValueOnce({ connections });
+
+    const result = await ConnectionsAPI.list();
+
+    expect(mockedApiFetchJSON).toHaveBeenCalledWith('/api/connections');
+    expect(result).toEqual(connections);
+  });
+
+  it('list() returns [] when the backend omits the connections field', async () => {
+    mockedApiFetchJSON.mockResolvedValueOnce({} as { connections?: Connection[] });
+
+    const result = await ConnectionsAPI.list();
+
+    expect(result).toEqual([]);
+  });
+
+  it('probe() POSTs the address JSON and returns the candidates envelope', async () => {
+    const response: ProbeResponse = {
+      candidates: [
+        {
+          type: 'pve',
+          host: 'https://pve.lab:8006',
+          port: 8006,
+          hints: { product: 'Proxmox VE', version: '8.2.4' },
+        },
+      ],
+      probedMs: 812,
+    };
+    mockedApiFetchJSON.mockResolvedValueOnce(response);
+
+    const result = await ConnectionsAPI.probe('pve.lab');
+
+    expect(mockedApiFetchJSON).toHaveBeenCalledWith('/api/connections/probe', {
+      method: 'POST',
+      body: JSON.stringify({ address: 'pve.lab' }),
+    });
+    expect(result).toEqual(response);
+  });
+});
@@ -0,0 +1,84 @@
+import { apiFetchJSON } from '@/utils/apiClient';
+
+export type ConnectionType =
+  | 'pve'
+  | 'pbs'
+  | 'pmg'
+  | 'vmware'
+  | 'truenas'
+  | 'agent'
+  | 'docker'
+  | 'kubernetes';
+
+export type ConnectionState =
+  | 'active'
+  | 'paused'
+  | 'unauthorized'
+  | 'unreachable'
+  | 'stale'
+  | 'pending';
+
+export type ConnectionSource = 'manual' | 'agent' | 'script';
+
+export interface ConnectionCapabilities {
+  supportsPause: boolean;
+  supportsScope: boolean;
+  supportsTest: boolean;
+}
+
+export interface ConnectionError {
+  message: string;
+  at: string;
+}
+
+export interface Connection {
+  id: string;
+  type: ConnectionType;
+  name: string;
+  address: string;
+  state: ConnectionState;
+  stateReason: string;
+  enabled: boolean;
+  surfaces: string[];
+  scope: Record<string, boolean>;
+  lastSeen: string | null;
+  lastError: ConnectionError | null;
+  source: ConnectionSource;
+  capabilities: ConnectionCapabilities;
+}
+
+export interface ConnectionsListResponse {
+  connections: Connection[];
+}
+
+export interface ProbeRequest {
+  address: string;
+}
+
+export interface ProbeCandidate {
+  type: ConnectionType;
+  host: string;
+  port: number;
+  hints?: Record<string, string>;
+}
+
+export interface ProbeResponse {
+  candidates: ProbeCandidate[];
+  probedMs: number;
+}
+
+export class ConnectionsAPI {
+  private static readonly baseUrl = '/api/connections';
+
+  static async list(): Promise<Connection[]> {
+    const response: ConnectionsListResponse = await apiFetchJSON(this.baseUrl);
+    return response.connections ?? [];
+  }
+
+  static async probe(address: string): Promise<ProbeResponse> {
+    return apiFetchJSON(`${this.baseUrl}/probe`, {
+      method: 'POST',
+      body: JSON.stringify({ address } satisfies ProbeRequest),
+    });
+  }
+}
@@ -642,6 +642,7 @@ type NodeConfigRequest struct {
 	MonitorQueues                *bool    `json:"monitorQueues,omitempty"`                // PMG only
 	MonitorQuarantine            *bool    `json:"monitorQuarantine,omitempty"`            // PMG only
 	MonitorDomainStats           *bool    `json:"monitorDomainStats,omitempty"`           // PMG only
+	Enabled                      *bool    `json:"enabled,omitempty"`                      // Lifecycle toggle; nil on update preserves current
 }

 // NodeResponse represents a node in API responses
@@ -674,7 +675,8 @@ type NodeResponse struct {
 	MonitorQueues                bool                      `json:"monitorQueues,omitempty"`
 	MonitorQuarantine            bool                      `json:"monitorQuarantine,omitempty"`
 	MonitorDomainStats           bool                      `json:"monitorDomainStats,omitempty"`
-	Status                       string                    `json:"status"` // "connected", "disconnected", "error"
+	Enabled                      bool                      `json:"enabled"` // Lifecycle; false = paused
+	Status                       string                    `json:"status"`  // "connected", "disconnected", "error"
 	IsCluster                    bool                      `json:"isCluster,omitempty"`
 	ClusterName                  string                    `json:"clusterName,omitempty"`
 	ClusterEndpoints             []ClusterEndpointResponse `json:"clusterEndpoints"`
@@ -1296,6 +1298,7 @@ func (h *ConfigHandlers) GetAllNodesForAPI(ctx context.Context) []NodeResponse {
 			MonitorPhysicalDisks:         pve.MonitorPhysicalDisks,
 			PhysicalDiskPollingMinutes:   pve.PhysicalDiskPollingMinutes,
 			TemperatureMonitoringEnabled: pve.TemperatureMonitoringEnabled,
+			Enabled:                      !pve.Disabled,
 			Status:                       h.getNodeStatus(ctx, "pve", pve.Name),
 			IsCluster:                    pve.IsCluster,
 			ClusterName:                  pve.ClusterName,
@@ -1326,6 +1329,7 @@ func (h *ConfigHandlers) GetAllNodesForAPI(ctx context.Context) []NodeResponse {
 			MonitorPruneJobs:             pbs.MonitorPruneJobs,
 			MonitorGarbageJobs:           pbs.MonitorGarbageJobs,
 			ExcludeDatastores:            pbs.ExcludeDatastores,
+			Enabled:                      !pbs.Disabled,
 			Status:                       h.getNodeStatus(ctx, "pbs", pbs.Name),
 			Source:                       pbs.Source,
 		}.NormalizeCollections()
@@ -1356,6 +1360,7 @@ func (h *ConfigHandlers) GetAllNodesForAPI(ctx context.Context) []NodeResponse {
 			MonitorQueues:                pmgInst.MonitorQueues,
 			MonitorQuarantine:            pmgInst.MonitorQuarantine,
 			MonitorDomainStats:           pmgInst.MonitorDomainStats,
+			Enabled:                      !pmgInst.Disabled,
 			Status:                       h.getNodeStatus(ctx, "pmg", pmgInst.Name),
 		}.NormalizeCollections()
 		nodes = append(nodes, node)
@@ -1208,6 +1208,9 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request
 		if req.TemperatureMonitoringEnabled != nil {
 			updated.TemperatureMonitoringEnabled = req.TemperatureMonitoringEnabled
 		}
+		if req.Enabled != nil {
+			updated.Disabled = !*req.Enabled
+		}

 		if enforceMonitoredSystemLimitForConfigReplacement(
 			w,
@@ -1312,6 +1315,9 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request
 		if req.ExcludeDatastores != nil {
 			updated.ExcludeDatastores = req.ExcludeDatastores
 		}
+		if req.Enabled != nil {
+			updated.Disabled = !*req.Enabled
+		}

 		if enforceMonitoredSystemLimitForConfigReplacement(
 			w,
@@ -1400,6 +1406,9 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request
 		if req.TemperatureMonitoringEnabled != nil {
 			updated.TemperatureMonitoringEnabled = req.TemperatureMonitoringEnabled
 		}
+		if req.Enabled != nil {
+			updated.Disabled = !*req.Enabled
+		}

 		if enforceMonitoredSystemLimitForConfigReplacement(
 			w,
@@ -0,0 +1,349 @@
+package api
+
+import (
+	"fmt"
+	"regexp"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
+)
+
+// connectionStaleThreshold is the baseline "haven't heard from this connection
+// recently" cutoff used to transition `active` → `stale`. Per-type poll
+// intervals vary (PVE 5-10s, TrueNAS 60s, agents 30-60s); 2 minutes sits
+// comfortably above 2× the slowest default so we don't flap on a single
+// dropped tick. Refined later once per-type intervals become first-class.
+const connectionStaleThreshold = 2 * time.Minute
+
+// connectionAuthErrorPattern matches the error strings pollers surface when a
+// credential is wrong or the token lacks scope. Kept centrally so
+// state-derivation stays the same across types.
+var connectionAuthErrorPattern = regexp.MustCompile(`(?i)401|403|unauthori[sz]ed|forbidden|authentication|permission denied|invalid (credentials|token|api key)`)
+
+// aggregatorInputs bundles everything the aggregator reads. Separating inputs
+// from the handler makes the aggregator unit-testable without spinning up a
+// monitor or persistence layer.
+type aggregatorInputs struct {
+	pveInstances     []config.PVEInstance
+	pbsInstances     []config.PBSInstance
+	pmgInstances     []config.PMGInstance
+	vmwareInstances  []config.VMwareVCenterInstance
+	truenasInstances []config.TrueNASInstance
+	hosts            []models.Host
+	instanceHealth   map[string]monitoring.InstanceHealth
+	now              time.Time
+}
+
+// buildConnections produces a stable, sorted list of connection rows across
+// every supported infrastructure type. The function is pure — it does not
+// perform any I/O and does not mutate its inputs.
+func buildConnections(in aggregatorInputs) []Connection {
+	now := in.now
+	if now.IsZero() {
+		now = time.Now()
+	}
+
+	out := make([]Connection, 0,
+		len(in.pveInstances)+len(in.pbsInstances)+len(in.pmgInstances)+
+			len(in.vmwareInstances)+len(in.truenasInstances)+len(in.hosts))
+
+	for _, pve := range in.pveInstances {
+		out = append(out, buildPVEConnection(pve, in.instanceHealth, now))
+	}
+	for _, pbs := range in.pbsInstances {
+		out = append(out, buildPBSConnection(pbs, in.instanceHealth, now))
+	}
+	for _, pmg := range in.pmgInstances {
+		out = append(out, buildPMGConnection(pmg, in.instanceHealth, now))
+	}
+	for _, vmw := range in.vmwareInstances {
+		out = append(out, buildVMwareConnection(vmw, in.instanceHealth, now))
+	}
+	for _, tn := range in.truenasInstances {
+		out = append(out, buildTrueNASConnection(tn, in.instanceHealth, now))
+	}
+	for _, host := range in.hosts {
+		out = append(out, buildAgentConnection(host, now))
+	}
+
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].Type != out[j].Type {
+			return out[i].Type < out[j].Type
+		}
+		return strings.ToLower(out[i].Name) < strings.ToLower(out[j].Name)
+	})
+
+	return out
+}
+
+func buildPVEConnection(inst config.PVEInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection {
+	enabled := !inst.Disabled
+	surfaces := []string{"vms", "containers", "storage", "backups"}
+	scope := map[string]bool{
+		"vms":        inst.MonitorVMs,
+		"containers": inst.MonitorContainers,
+		"storage":    inst.MonitorStorage,
+		"backups":    inst.MonitorBackups,
+	}
+	h := health["pve::"+inst.Name]
+	state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now)
+	return Connection{
+		ID:           "pve:" + inst.Name,
+		Type:         ConnectionTypePVE,
+		Name:         inst.Name,
+		Address:      inst.Host,
+		State:        state,
+		StateReason:  reason,
+		Enabled:      enabled,
+		Surfaces:     surfaces,
+		Scope:        scope,
+		LastSeen:     lastSeen,
+		LastError:    lastError,
+		Source:       sourceFromString(inst.Source),
+		Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: true, SupportsTest: true},
+	}
+}
+
+func buildPBSConnection(inst config.PBSInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection {
+	enabled := !inst.Disabled
+	surfaces := []string{"backups", "datastores", "syncJobs", "verifyJobs", "pruneJobs", "garbageJobs"}
+	scope := map[string]bool{
+		"backups":     inst.MonitorBackups,
+		"datastores":  inst.MonitorDatastores,
+		"syncJobs":    inst.MonitorSyncJobs,
+		"verifyJobs":  inst.MonitorVerifyJobs,
+		"pruneJobs":   inst.MonitorPruneJobs,
+		"garbageJobs": inst.MonitorGarbageJobs,
+	}
+	h := health["pbs::"+inst.Name]
+	state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now)
+	return Connection{
+		ID:           "pbs:" + inst.Name,
+		Type:         ConnectionTypePBS,
+		Name:         inst.Name,
+		Address:      inst.Host,
+		State:        state,
+		StateReason:  reason,
+		Enabled:      enabled,
+		Surfaces:     surfaces,
+		Scope:        scope,
+		LastSeen:     lastSeen,
+		LastError:    lastError,
+		Source:       sourceFromString(inst.Source),
+		Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: true, SupportsTest: true},
+	}
+}
+
+func buildPMGConnection(inst config.PMGInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection {
+	enabled := !inst.Disabled
+	surfaces := []string{"mailStats", "queues", "quarantine", "domainStats"}
+	scope := map[string]bool{
+		"mailStats":   inst.MonitorMailStats,
+		"queues":      inst.MonitorQueues,
+		"quarantine":  inst.MonitorQuarantine,
+		"domainStats": inst.MonitorDomainStats,
+	}
+	h := health["pmg::"+inst.Name]
+	state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now)
+	return Connection{
+		ID:           "pmg:" + inst.Name,
+		Type:         ConnectionTypePMG,
+		Name:         inst.Name,
+		Address:      inst.Host,
+		State:        state,
+		StateReason:  reason,
+		Enabled:      enabled,
+		Surfaces:     surfaces,
+		Scope:        scope,
+		LastSeen:     lastSeen,
+		LastError:    lastError,
+		Source:       ConnectionSourceManual,
+		Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: true, SupportsTest: true},
+	}
+}
+
+func buildVMwareConnection(inst config.VMwareVCenterInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection {
+	enabled := inst.Enabled
+	surfaces := []string{"vms", "hosts", "datastores"}
+	h := health["vmware::"+inst.ID]
+	state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now)
+	port := inst.Port
+	if port == 0 {
+		port = 443
+	}
+	return Connection{
+		ID:           "vmware:" + inst.ID,
+		Type:         ConnectionTypeVMware,
+		Name:         inst.Name,
+		Address:      fmt.Sprintf("https://%s:%d", inst.Host, port),
+		State:        state,
+		StateReason:  reason,
+		Enabled:      enabled,
+		Surfaces:     surfaces,
+		Scope:        map[string]bool{"vms": true, "hosts": true, "datastores": true},
+		LastSeen:     lastSeen,
+		LastError:    lastError,
+		Source:       ConnectionSourceManual,
+		Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: false, SupportsTest: true},
+	}
+}
+
+func buildTrueNASConnection(inst config.TrueNASInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection {
+	enabled := inst.Enabled
+	surfaces := []string{"datasets", "pools", "replication"}
+	h := health["truenas::"+inst.ID]
+	state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now)
+	scheme := "https"
+	if !inst.UseHTTPS {
+		scheme = "http"
+	}
+	port := inst.Port
+	if port == 0 {
+		if inst.UseHTTPS {
+			port = 443
+		} else {
+			port = 80
+		}
+	}
+	return Connection{
+		ID:           "truenas:" + inst.ID,
+		Type:         ConnectionTypeTrueNAS,
+		Name:         inst.Name,
+		Address:      fmt.Sprintf("%s://%s:%d", scheme, inst.Host, port),
+		State:        state,
+		StateReason:  reason,
+		Enabled:      enabled,
+		Surfaces:     surfaces,
+		Scope:        map[string]bool{"datasets": true, "pools": true, "replication": true},
+		LastSeen:     lastSeen,
+		LastError:    lastError,
+		Source:       ConnectionSourceManual,
+		Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: false, SupportsTest: true},
+	}
+}
+
+// buildAgentConnection derives a connection row from an agent Host record.
+// Agents have no pause toggle and no scope — reports are all-or-nothing —
+// so capability flags are off.
+func buildAgentConnection(host models.Host, now time.Time) Connection {
+	name := host.DisplayName
+	if strings.TrimSpace(name) == "" {
+		name = host.Hostname
+	}
+	if strings.TrimSpace(name) == "" {
+		name = host.ID
+	}
+	address := host.Hostname
+	if strings.TrimSpace(address) == "" {
+		address = host.ReportIP
+	}
+
+	var lastSeen *time.Time
+	if !host.LastSeen.IsZero() {
+		t := host.LastSeen
+		lastSeen = &t
+	}
+
+	state := ConnectionStatePending
+	reason := ""
+	switch {
+	case lastSeen == nil:
+		state = ConnectionStatePending
+	case now.Sub(*lastSeen) > connectionStaleThreshold:
+		state = ConnectionStateStale
+		reason = fmt.Sprintf("no heartbeat in %s", now.Sub(*lastSeen).Round(time.Second))
+	default:
+		state = ConnectionStateActive
+	}
+
+	return Connection{
+		ID:           "agent:" + host.ID,
+		Type:         ConnectionTypeAgent,
+		Name:         name,
+		Address:      address,
+		State:        state,
+		StateReason:  reason,
+		Enabled:      true,
+		Surfaces:     []string{"host"},
+		Scope:        map[string]bool{"host": true},
+		LastSeen:     lastSeen,
+		LastError:    nil,
+		Source:       ConnectionSourceAgent,
+		Capabilities: ConnectionCapabilities{SupportsPause: false, SupportsScope: false, SupportsTest: false},
+	}
+}
+
+// deriveConnectionState maps (Enabled, InstanceHealth) onto the unified state
+// vocabulary. No new state is persisted — the inputs come from the existing
+// monitoring scheduler.
+func deriveConnectionState(enabled bool, h monitoring.InstanceHealth, now time.Time) (ConnectionState, string, *time.Time, *ConnectionError) {
+	var lastSeen *time.Time
+	if h.PollStatus.LastSuccess != nil && !h.PollStatus.LastSuccess.IsZero() {
+		t := *h.PollStatus.LastSuccess
+		lastSeen = &t
+	}
+
+	var lastError *ConnectionError
+	if h.PollStatus.LastError != nil && h.PollStatus.LastError.Message != "" {
+		lastError = &ConnectionError{
+			At:       h.PollStatus.LastError.At,
+			Message:  h.PollStatus.LastError.Message,
+			Category: h.PollStatus.LastError.Category,
+		}
+	}
+
+	if !enabled {
+		return ConnectionStatePaused, "paused by user", lastSeen, lastError
+	}
+
+	if lastSeen == nil && lastError == nil {
+		return ConnectionStatePending, "awaiting first poll", nil, nil
+	}
+
+	if lastError != nil && connectionAuthErrorPattern.MatchString(lastError.Message) {
+		return ConnectionStateUnauthorized, lastError.Message, lastSeen, lastError
+	}
+
+	if strings.EqualFold(h.Breaker.State, "open") {
+		reason := "circuit breaker open"
+		if lastError != nil {
+			reason = lastError.Message
+		}
+		return ConnectionStateUnreachable, reason, lastSeen, lastError
+	}
+
+	if lastSeen != nil && now.Sub(*lastSeen) > connectionStaleThreshold {
+		return ConnectionStateStale, fmt.Sprintf("no successful poll in %s", now.Sub(*lastSeen).Round(time.Second)), lastSeen, lastError
+	}
+
+	return ConnectionStateActive, "", lastSeen, lastError
+}
+
+func sourceFromString(s string) ConnectionSource {
+	switch strings.ToLower(strings.TrimSpace(s)) {
+	case "agent":
+		return ConnectionSourceAgent
+	case "script":
+		return ConnectionSourceScript
+	default:
+		return ConnectionSourceManual
+	}
+}
+
+// instanceHealthByKey flattens SchedulerHealthResponse.Instances into a
+// lookup map keyed by schedulerKey ("pve::instance-name"). The aggregator
+// consults this to pick up LastSuccess and LastError without re-polling.
+func instanceHealthByKey(resp monitoring.SchedulerHealthResponse) map[string]monitoring.InstanceHealth {
+	out := make(map[string]monitoring.InstanceHealth, len(resp.Instances))
+	for _, inst := range resp.Instances {
+		if inst.Key == "" {
+			continue
+		}
+		out[inst.Key] = inst
+	}
+	return out
+}
@@ -0,0 +1,211 @@
+package api
+
+import (
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
+)
+
+func ptrTime(t time.Time) *time.Time { return &t }
+
+func healthEntry(lastSuccess *time.Time, errMessage, errCategory string, breakerState string) monitoring.InstanceHealth {
+	ps := monitoring.InstancePollStatus{LastSuccess: lastSuccess}
+	if errMessage != "" {
+		ps.LastError = &monitoring.ErrorDetail{
+			At:       time.Now(),
+			Message:  errMessage,
+			Category: errCategory,
+		}
+	}
+	return monitoring.InstanceHealth{
+		PollStatus: ps,
+		Breaker:    monitoring.InstanceBreaker{State: breakerState},
+	}
+}
+
+func TestDeriveConnectionState_Paused(t *testing.T) {
+	state, reason, _, _ := deriveConnectionState(false, monitoring.InstanceHealth{}, time.Now())
+	if state != ConnectionStatePaused {
+		t.Fatalf("got %q, want %q", state, ConnectionStatePaused)
+	}
+	if reason == "" {
+		t.Fatal("expected non-empty reason for paused state")
+	}
+}
+
+func TestDeriveConnectionState_Pending(t *testing.T) {
+	state, _, lastSeen, lastError := deriveConnectionState(true, monitoring.InstanceHealth{}, time.Now())
+	if state != ConnectionStatePending {
+		t.Fatalf("got %q, want %q", state, ConnectionStatePending)
+	}
+	if lastSeen != nil || lastError != nil {
+		t.Fatal("expected nil lastSeen and lastError in pending state")
+	}
+}
+
+func TestDeriveConnectionState_Unauthorized(t *testing.T) {
+	now := time.Now()
+	h := healthEntry(ptrTime(now.Add(-30*time.Second)), "401 Unauthorized: token invalid", "auth", "closed")
+	state, reason, _, err := deriveConnectionState(true, h, now)
+	if state != ConnectionStateUnauthorized {
+		t.Fatalf("got %q, want %q", state, ConnectionStateUnauthorized)
+	}
+	if reason == "" {
+		t.Fatal("expected reason to include error message")
+	}
+	if err == nil || err.Message != "401 Unauthorized: token invalid" {
+		t.Fatalf("unexpected lastError: %+v", err)
+	}
+}
+
+func TestDeriveConnectionState_Unreachable(t *testing.T) {
+	now := time.Now()
+	h := healthEntry(ptrTime(now.Add(-30*time.Second)), "connection refused", "network", "open")
+	state, _, _, _ := deriveConnectionState(true, h, now)
+	if state != ConnectionStateUnreachable {
+		t.Fatalf("got %q, want %q", state, ConnectionStateUnreachable)
+	}
+}
+
+func TestDeriveConnectionState_Stale(t *testing.T) {
+	now := time.Now()
+	stale := now.Add(-5 * time.Minute)
+	h := healthEntry(ptrTime(stale), "", "", "closed")
+	state, reason, _, _ := deriveConnectionState(true, h, now)
+	if state != ConnectionStateStale {
+		t.Fatalf("got %q, want %q", state, ConnectionStateStale)
+	}
+	if reason == "" {
+		t.Fatal("expected reason to describe staleness")
+	}
+}
+
+func TestDeriveConnectionState_Active(t *testing.T) {
+	now := time.Now()
+	h := healthEntry(ptrTime(now.Add(-10*time.Second)), "", "", "closed")
+	state, _, _, _ := deriveConnectionState(true, h, now)
+	if state != ConnectionStateActive {
+		t.Fatalf("got %q, want %q", state, ConnectionStateActive)
+	}
+}
+
+func TestBuildConnections_SortsByTypeThenName(t *testing.T) {
+	now := time.Now()
+	in := aggregatorInputs{
+		pveInstances: []config.PVEInstance{
+			{Name: "beta", Host: "https://b.lan:8006", MonitorVMs: true},
+			{Name: "alpha", Host: "https://a.lan:8006", MonitorVMs: true},
+		},
+		pbsInstances: []config.PBSInstance{
+			{Name: "backups", Host: "https://bkp.lan:8007"},
+		},
+		now: now,
+	}
+	got := buildConnections(in)
+	if len(got) != 3 {
+		t.Fatalf("expected 3 connections, got %d", len(got))
+	}
+	if got[0].Type != ConnectionTypePBS {
+		t.Fatalf("expected PBS first, got %q", got[0].Type)
+	}
+	if got[1].Name != "alpha" || got[2].Name != "beta" {
+		t.Fatalf("PVE order wrong: %q, %q", got[1].Name, got[2].Name)
+	}
+}
+
+func TestBuildConnections_PVEPausedRespectsDisabled(t *testing.T) {
+	in := aggregatorInputs{
+		pveInstances: []config.PVEInstance{{Name: "pve1", Host: "https://pve1.lan:8006", Disabled: true}},
+		now:          time.Now(),
+	}
+	got := buildConnections(in)
+	if len(got) != 1 {
+		t.Fatalf("expected 1 connection, got %d", len(got))
+	}
+	if got[0].State != ConnectionStatePaused {
+		t.Fatalf("expected paused, got %q", got[0].State)
+	}
+	if got[0].Enabled {
+		t.Fatal("expected Enabled=false for Disabled PVE instance")
+	}
+}
+
+func TestBuildConnections_AgentStateFromLastSeen(t *testing.T) {
+	now := time.Now()
+	in := aggregatorInputs{
+		hosts: []models.Host{
+			{ID: "fresh", Hostname: "h1", LastSeen: now.Add(-10 * time.Second)},
+			{ID: "stale", Hostname: "h2", LastSeen: now.Add(-5 * time.Minute)},
+			{ID: "never", Hostname: "h3"},
+		},
+		now: now,
+	}
+	got := buildConnections(in)
+	byID := map[string]Connection{}
+	for _, c := range got {
+		byID[c.ID] = c
+	}
+	if byID["agent:fresh"].State != ConnectionStateActive {
+		t.Fatalf("fresh agent: got %q, want active", byID["agent:fresh"].State)
+	}
+	if byID["agent:stale"].State != ConnectionStateStale {
+		t.Fatalf("stale agent: got %q, want stale", byID["agent:stale"].State)
+	}
+	if byID["agent:never"].State != ConnectionStatePending {
+		t.Fatalf("never-reported agent: got %q, want pending", byID["agent:never"].State)
+	}
+	for _, c := range got {
+		if c.Capabilities.SupportsPause || c.Capabilities.SupportsScope {
+			t.Fatalf("agents must not advertise pause/scope capabilities: %+v", c)
+		}
+	}
+}
+
+func TestBuildConnections_VMwareAndTrueNASEnabledFlag(t *testing.T) {
+	in := aggregatorInputs{
+		vmwareInstances:  []config.VMwareVCenterInstance{{ID: "vc1", Name: "vc", Host: "vc.lan", Enabled: false}},
+		truenasInstances: []config.TrueNASInstance{{ID: "tn1", Name: "tn", Host: "tn.lan", Enabled: true, UseHTTPS: true}},
+		now:              time.Now(),
+	}
+	got := buildConnections(in)
+	var vmw, tn Connection
+	for _, c := range got {
+		switch c.Type {
+		case ConnectionTypeVMware:
+			vmw = c
+		case ConnectionTypeTrueNAS:
+			tn = c
+		}
+	}
+	if vmw.State != ConnectionStatePaused || vmw.Enabled {
+		t.Fatalf("vmware with Enabled=false should be paused, got state=%q enabled=%v", vmw.State, vmw.Enabled)
+	}
+	if tn.State != ConnectionStatePending {
+		t.Fatalf("truenas with no health yet should be pending, got %q", tn.State)
+	}
+	if !tn.Enabled {
+		t.Fatal("truenas with Enabled=true should surface enabled=true")
+	}
+}
+
+func TestBuildConnections_UsesHealthLookup(t *testing.T) {
+	now := time.Now()
+	ls := now.Add(-15 * time.Second)
+	in := aggregatorInputs{
+		pveInstances: []config.PVEInstance{{Name: "pve1", Host: "https://pve1.lan:8006"}},
+		instanceHealth: map[string]monitoring.InstanceHealth{
+			"pve::pve1": healthEntry(&ls, "", "", "closed"),
+		},
+		now: now,
+	}
+	got := buildConnections(in)
+	if got[0].State != ConnectionStateActive {
+		t.Fatalf("expected active, got %q", got[0].State)
+	}
+	if got[0].LastSeen == nil || !got[0].LastSeen.Equal(ls) {
+		t.Fatalf("lastSeen not propagated: %+v", got[0].LastSeen)
+	}
+}
@@ -0,0 +1,111 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
+)
+
+// ConnectionsHandlers serves the unified connections ledger. It does not own
+// any persistence of its own — it composes per-type stores and the
+// monitoring scheduler's in-memory health data into a single list.
+type ConnectionsHandlers struct {
+	getConfig      func(ctx context.Context) *config.Config
+	getPersistence func(ctx context.Context) *config.ConfigPersistence
+	getMonitor     func(ctx context.Context) *monitoring.Monitor
+}
+
+// NewConnectionsHandlers wires the aggregator behind the request-scoped
+// tenant resolvers already used by ConfigHandlers, so the endpoint shares
+// the same multi-tenant behavior as the per-type routes it aggregates.
+func NewConnectionsHandlers(
+	getConfig func(ctx context.Context) *config.Config,
+	getPersistence func(ctx context.Context) *config.ConfigPersistence,
+	getMonitor func(ctx context.Context) *monitoring.Monitor,
+) *ConnectionsHandlers {
+	return &ConnectionsHandlers{
+		getConfig:      getConfig,
+		getPersistence: getPersistence,
+		getMonitor:     getMonitor,
+	}
+}
+
+// HandleList returns every configured connection as a unified Connection row.
+// No probing or network I/O happens here — state is derived purely from
+// cached poller health.
+func (h *ConnectionsHandlers) HandleList(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	ctx := r.Context()
+	cfg := h.getConfig(ctx)
+	persistence := h.getPersistence(ctx)
+	monitor := h.getMonitor(ctx)
+
+	inputs := aggregatorInputs{}
+
+	if cfg != nil {
+		inputs.pveInstances = cfg.PVEInstances
+		inputs.pbsInstances = cfg.PBSInstances
+		inputs.pmgInstances = cfg.PMGInstances
+	}
+
+	if persistence != nil {
+		if vmw, err := persistence.LoadVMwareConfig(); err == nil {
+			inputs.vmwareInstances = vmw
+		}
+		if tn, err := persistence.LoadTrueNASConfig(); err == nil {
+			inputs.truenasInstances = tn
+		}
+	}
+
+	if monitor != nil {
+		snapshot := monitor.GetState()
+		inputs.hosts = snapshot.Hosts
+		inputs.instanceHealth = instanceHealthByKey(monitor.SchedulerHealth())
+	} else {
+		inputs.hosts = []models.Host{}
+		inputs.instanceHealth = map[string]monitoring.InstanceHealth{}
+	}
+
+	writeJSON(w, http.StatusOK, ConnectionsListResponse{
+		Connections: buildConnections(inputs),
+	})
+}
+
+// HandleProbe fingerprints a user-supplied address and returns the product
+// types it detected, if any. No configuration is persisted — the caller
+// uses the response to pick a credential slot in ConnectionEditor.
+func (h *ConnectionsHandlers) HandleProbe(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	r.Body = http.MaxBytesReader(w, r.Body, 4*1024)
+	defer r.Body.Close()
+
+	var req ProbeRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid JSON body", nil)
+		return
+	}
+
+	host, port, err := parseProbeAddress(req.Address)
+	if err != nil {
+		writeErrorResponse(w, http.StatusBadRequest, "invalid_address", err.Error(), nil)
+		return
+	}
+
+	candidates, elapsed := runProbe(r.Context(), host, port, probeHTTPClient())
+	writeJSON(w, http.StatusOK, ProbeResponse{
+		Candidates: candidates,
+		ProbedMs:   elapsed.Milliseconds(),
+	})
+}
@@ -0,0 +1,328 @@
+package api
+
+import (
+	"context"
+	"crypto/sha256"
+	"crypto/tls"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+// Probe budget. Per-candidate and whole-request ceilings are kept tight to
+// make the endpoint unusable as a slow-leak scanner. Each candidate gets
+// 2s connect + 1s read, and the whole request is capped at the total budget.
+const (
+	probeDialTimeout     = 2 * time.Second
+	probeTotalBudget     = 3 * time.Second
+	probeMaxConcurrent   = 5
+	probeMaxAddressBytes = 512
+)
+
+// ProbeRequest is the POST body for /api/connections/probe.
+type ProbeRequest struct {
+	Address string `json:"address"`
+}
+
+// ProbeCandidate is one detected product at a host:port.
+type ProbeCandidate struct {
+	Type  ConnectionType    `json:"type"`
+	Host  string            `json:"host"`
+	Port  int               `json:"port"`
+	Hints map[string]string `json:"hints,omitempty"`
+}
+
+// ProbeResponse is the envelope returned to the frontend.
+type ProbeResponse struct {
+	Candidates []ProbeCandidate `json:"candidates"`
+	ProbedMs   int64            `json:"probedMs"`
+}
+
+// probeTarget is a single {type, scheme, port, path} fingerprint we attempt.
+type probeTarget struct {
+	Type       ConnectionType
+	Scheme     string
+	Port       int
+	Path       string
+	identifyFn func(resp *http.Response, body []byte) (match bool, hints map[string]string)
+}
+
+var defaultProbeTargets = []probeTarget{
+	{
+		Type:   ConnectionTypePVE,
+		Scheme: "https",
+		Port:   8006,
+		Path:   "/api2/json/version",
+		identifyFn: func(resp *http.Response, body []byte) (bool, map[string]string) {
+			server := strings.ToLower(resp.Header.Get("Server"))
+			if strings.Contains(server, "pve-api-daemon") {
+				return true, versionHintsFromProxmoxBody(body, "Proxmox VE")
+			}
+			if strings.Contains(string(body), `"repoid"`) && strings.Contains(string(body), `"version"`) &&
+				!strings.Contains(string(body), `"product":"pmg"`) {
+				return true, versionHintsFromProxmoxBody(body, "Proxmox VE")
+			}
+			return false, nil
+		},
+	},
+	{
+		Type:   ConnectionTypePBS,
+		Scheme: "https",
+		Port:   8007,
+		Path:   "/api2/json/version",
+		identifyFn: func(resp *http.Response, body []byte) (bool, map[string]string) {
+			server := strings.ToLower(resp.Header.Get("Server"))
+			if strings.Contains(server, "proxmox-backup-api") {
+				return true, versionHintsFromProxmoxBody(body, "Proxmox Backup Server")
+			}
+			return false, nil
+		},
+	},
+	{
+		Type:   ConnectionTypePMG,
+		Scheme: "https",
+		Port:   8006,
+		Path:   "/api2/json/version",
+		identifyFn: func(resp *http.Response, body []byte) (bool, map[string]string) {
+			server := strings.ToLower(resp.Header.Get("Server"))
+			if strings.Contains(server, "pmg-api-daemon") {
+				return true, versionHintsFromProxmoxBody(body, "Proxmox Mail Gateway")
+			}
+			if strings.Contains(string(body), `"product":"pmg"`) {
+				return true, versionHintsFromProxmoxBody(body, "Proxmox Mail Gateway")
+			}
+			return false, nil
+		},
+	},
+	{
+		Type:   ConnectionTypeVMware,
+		Scheme: "https",
+		Port:   443,
+		Path:   "/sdk/vimServiceVersions.xml",
+		identifyFn: func(_ *http.Response, body []byte) (bool, map[string]string) {
+			if strings.Contains(string(body), "urn:vim25") {
+				return true, map[string]string{"product": "VMware vCenter"}
+			}
+			return false, nil
+		},
+	},
+	{
+		Type:   ConnectionTypeTrueNAS,
+		Scheme: "https",
+		Port:   443,
+		Path:   "/api/v2.0/system/product_name",
+		identifyFn: func(_ *http.Response, body []byte) (bool, map[string]string) {
+			upper := strings.ToUpper(string(body))
+			if strings.Contains(upper, "TRUENAS") {
+				return true, map[string]string{"product": "TrueNAS"}
+			}
+			return false, nil
+		},
+	},
+}
+
+// versionHintsFromProxmoxBody pulls the minimum fields out of a Proxmox API
+// /version response so the frontend can show the user something useful
+// beyond "we think this is PVE." Any shape mismatch just yields the product
+// name alone — we never fail a probe on a hint-parse error.
+func versionHintsFromProxmoxBody(body []byte, productName string) map[string]string {
+	hints := map[string]string{"product": productName}
+	var wrapper struct {
+		Data struct {
+			Version string `json:"version"`
+			Release string `json:"release"`
+			Repoid  string `json:"repoid"`
+		} `json:"data"`
+	}
+	if err := json.Unmarshal(body, &wrapper); err == nil {
+		if wrapper.Data.Version != "" {
+			hints["version"] = wrapper.Data.Version
+		}
+		if wrapper.Data.Release != "" {
+			hints["release"] = wrapper.Data.Release
+		}
+	}
+	return hints
+}
+
+// parseProbeAddress normalizes user input into (host, explicitPort).
+// Accepted forms: "host", "host:port", "ip", "ip:port", "scheme://host[:port]".
+// explicitPort == 0 means "probe all defaults"; otherwise only probe targets
+// that match the given port.
+func parseProbeAddress(raw string) (host string, explicitPort int, err error) {
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return "", 0, fmt.Errorf("address is required")
+	}
+	if len(raw) > probeMaxAddressBytes {
+		return "", 0, fmt.Errorf("address is too long")
+	}
+
+	if strings.Contains(raw, "://") {
+		u, err := url.Parse(raw)
+		if err != nil || u.Host == "" {
+			return "", 0, fmt.Errorf("invalid URL: %s", raw)
+		}
+		host = u.Hostname()
+		if portStr := u.Port(); portStr != "" {
+			p, err := strconv.Atoi(portStr)
+			if err != nil || p < 1 || p > 65535 {
+				return "", 0, fmt.Errorf("invalid port in URL")
+			}
+			explicitPort = p
+		}
+		return host, explicitPort, nil
+	}
+
+	if h, p, splitErr := net.SplitHostPort(raw); splitErr == nil {
+		portNum, convErr := strconv.Atoi(p)
+		if convErr != nil || portNum < 1 || portNum > 65535 {
+			return "", 0, fmt.Errorf("invalid port")
+		}
+		return h, portNum, nil
+	}
+
+	return raw, 0, nil
+}
+
+// targetsForPort narrows defaultProbeTargets to only those matching the
+// user's explicit port. Same host can serve PVE on 8006 and PBS on 8007, so
+// zero port means "try them all."
+func targetsForPort(port int) []probeTarget {
+	if port == 0 {
+		return defaultProbeTargets
+	}
+	out := make([]probeTarget, 0, 2)
+	for _, t := range defaultProbeTargets {
+		if t.Port == port {
+			out = append(out, t)
+		}
+	}
+	return out
+}
+
+// probeHTTPClient builds one client per probe so that dial timeouts,
+// TLS-skip, and cancellation are self-contained. InsecureSkipVerify is on
+// because the whole point of probing is to talk to a server whose cert we
+// haven't trusted yet.
+func probeHTTPClient() *http.Client {
+	return &http.Client{
+		Transport: &http.Transport{
+			DialContext: (&net.Dialer{
+				Timeout: probeDialTimeout,
+			}).DialContext,
+			TLSHandshakeTimeout: probeDialTimeout,
+			TLSClientConfig:     &tls.Config{InsecureSkipVerify: true},
+			DisableKeepAlives:   true,
+		},
+		// Overall per-request ceiling is the dial + read budget.
+		Timeout: probeDialTimeout + time.Second,
+	}
+}
+
+// runProbe fans out probe requests against every candidate target. It
+// returns (sorted-deduped candidates, total elapsed). The function never
+// returns an error — individual probe failures are swallowed as "not that
+// type" rather than bubbling up and confusing the caller.
+func runProbe(ctx context.Context, host string, port int, client *http.Client) ([]ProbeCandidate, time.Duration) {
+	start := time.Now()
+	budget := probeTotalBudget
+	ctx, cancel := context.WithTimeout(ctx, budget)
+	defer cancel()
+
+	targets := targetsForPort(port)
+	if len(targets) == 0 {
+		return []ProbeCandidate{}, time.Since(start)
+	}
+
+	sem := make(chan struct{}, probeMaxConcurrent)
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	results := make([]ProbeCandidate, 0, len(targets))
+
+	for _, target := range targets {
+		target := target
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			select {
+			case sem <- struct{}{}:
+				defer func() { <-sem }()
+			case <-ctx.Done():
+				return
+			}
+
+			if cand, ok := probeOne(ctx, host, target, client); ok {
+				mu.Lock()
+				results = append(results, cand)
+				mu.Unlock()
+			}
+		}()
+	}
+	wg.Wait()
+
+	sort.Slice(results, func(i, j int) bool {
+		if results[i].Type != results[j].Type {
+			return results[i].Type < results[j].Type
+		}
+		return results[i].Port < results[j].Port
+	})
+
+	return results, time.Since(start)
+}
+
+// probeOne runs a single probe. It returns (candidate, true) only when the
+// target's identifyFn confirms the product.
+func probeOne(ctx context.Context, host string, target probeTarget, client *http.Client) (ProbeCandidate, bool) {
+	endpoint := fmt.Sprintf("%s://%s:%d%s", target.Scheme, host, target.Port, target.Path)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
+	if err != nil {
+		return ProbeCandidate{}, false
+	}
+	req.Header.Set("User-Agent", "Pulse/connections-probe")
+	req.Header.Set("Accept", "application/json,text/xml,*/*")
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return ProbeCandidate{}, false
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 500 {
+		return ProbeCandidate{}, false
+	}
+
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 64*1024))
+	if err != nil {
+		return ProbeCandidate{}, false
+	}
+
+	match, hints := target.identifyFn(resp, body)
+	if !match {
+		return ProbeCandidate{}, false
+	}
+
+	if hints == nil {
+		hints = map[string]string{}
+	}
+	if resp.TLS != nil && len(resp.TLS.PeerCertificates) > 0 {
+		sum := sha256.Sum256(resp.TLS.PeerCertificates[0].Raw)
+		hints["fingerprint"] = "SHA256:" + hex.EncodeToString(sum[:])
+	}
+
+	return ProbeCandidate{
+		Type:  target.Type,
+		Host:  fmt.Sprintf("%s://%s:%d", target.Scheme, host, target.Port),
+		Port:  target.Port,
+		Hints: hints,
+	}, true
+}
@@ -0,0 +1,333 @@
+package api
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestParseProbeAddress_Shapes(t *testing.T) {
+	cases := []struct {
+		in          string
+		wantHost    string
+		wantPort    int
+		expectError bool
+	}{
+		{"pve01.lan", "pve01.lan", 0, false},
+		{"pve01.lan:8006", "pve01.lan", 8006, false},
+		{"192.168.1.10", "192.168.1.10", 0, false},
+		{"192.168.1.10:443", "192.168.1.10", 443, false},
+		{"https://pve01.lan:8006", "pve01.lan", 8006, false},
+		{"https://pve01.lan", "pve01.lan", 0, false},
+		{"   ", "", 0, true},
+		{"pve01.lan:99999", "", 0, true},
+	}
+	for _, c := range cases {
+		t.Run(c.in, func(t *testing.T) {
+			host, port, err := parseProbeAddress(c.in)
+			if c.expectError {
+				if err == nil {
+					t.Fatalf("expected error for %q", c.in)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if host != c.wantHost {
+				t.Fatalf("host = %q, want %q", host, c.wantHost)
+			}
+			if port != c.wantPort {
+				t.Fatalf("port = %d, want %d", port, c.wantPort)
+			}
+		})
+	}
+}
+
+func TestTargetsForPort_ExplicitNarrowsToMatching(t *testing.T) {
+	targets := targetsForPort(8006)
+	seen := map[ConnectionType]bool{}
+	for _, t := range targets {
+		seen[t.Type] = true
+	}
+	if !seen[ConnectionTypePVE] {
+		t.Fatal("expected PVE target for port 8006")
+	}
+	if !seen[ConnectionTypePMG] {
+		t.Fatal("expected PMG target for port 8006")
+	}
+	if seen[ConnectionTypePBS] {
+		t.Fatal("did not expect PBS target for port 8006")
+	}
+}
+
+func TestTargetsForPort_ZeroReturnsAll(t *testing.T) {
+	if len(targetsForPort(0)) != len(defaultProbeTargets) {
+		t.Fatal("zero port should return all default targets")
+	}
+}
+
+// probeTestServer spins up an HTTPS server on a random port, overriding the
+// target port in defaultProbeTargets so probeOne can hit it. Restores the
+// original targets in cleanup. Returns the host:port string.
+func probeTestServer(t *testing.T, handler http.HandlerFunc) (host string, port int, cleanup func()) {
+	t.Helper()
+	srv := httptest.NewTLSServer(handler)
+
+	u := srv.URL
+	u = strings.TrimPrefix(u, "https://")
+	h, p, err := net.SplitHostPort(u)
+	if err != nil {
+		t.Fatalf("failed to split host:port %q: %v", u, err)
+	}
+	portInt, err := strconv.Atoi(p)
+	if err != nil {
+		t.Fatalf("failed to parse port %q: %v", p, err)
+	}
+	return h, portInt, func() {
+		srv.Close()
+	}
+}
+
+func pveHandler(body string) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Server", "pve-api-daemon/3.4")
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, body)
+	}
+}
+
+func pbsHandler(body string) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Server", "proxmox-backup-api/3.2")
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, body)
+	}
+}
+
+func pmgHandler(body string) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Server", "pmg-api-daemon/8.1")
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, body)
+	}
+}
+
+func vmwareHandler() http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/xml")
+		fmt.Fprint(w, `<?xml version="1.0" encoding="UTF-8"?>
+<namespaces>
+  <namespace>urn:vim25</namespace>
+</namespaces>`)
+	}
+}
+
+func truenasHandler() http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, `"TRUENAS-SCALE-24.04"`)
+	}
+}
+
+// testProbeClient mirrors probeHTTPClient but shorter timeouts for tests.
+func testProbeClient() *http.Client {
+	return &http.Client{
+		Transport: &http.Transport{
+			DialContext:       (&net.Dialer{Timeout: 500 * time.Millisecond}).DialContext,
+			TLSClientConfig:   &tls.Config{InsecureSkipVerify: true},
+			DisableKeepAlives: true,
+		},
+		Timeout: 1 * time.Second,
+	}
+}
+
+func withSingleTarget(t *testing.T, tgt probeTarget) func() {
+	t.Helper()
+	orig := defaultProbeTargets
+	defaultProbeTargets = []probeTarget{tgt}
+	return func() { defaultProbeTargets = orig }
+}
+
+func TestProbeOne_DetectsPVE(t *testing.T) {
+	host, port, done := probeTestServer(t, pveHandler(`{"data":{"version":"8.2.4","release":"8.2","repoid":"abc"}}`))
+	defer done()
+
+	tgt := probeTarget{
+		Type:       ConnectionTypePVE,
+		Scheme:     "https",
+		Port:       port,
+		Path:       "/api2/json/version",
+		identifyFn: defaultProbeTargets[0].identifyFn,
+	}
+	cand, ok := probeOne(context.Background(), host, tgt, testProbeClient())
+	if !ok {
+		t.Fatal("expected PVE detection")
+	}
+	if cand.Type != ConnectionTypePVE {
+		t.Fatalf("type = %q, want pve", cand.Type)
+	}
+	if cand.Hints["version"] != "8.2.4" {
+		t.Fatalf("version hint not propagated: %+v", cand.Hints)
+	}
+}
+
+func TestProbeOne_DetectsPBS(t *testing.T) {
+	host, port, done := probeTestServer(t, pbsHandler(`{"data":{"version":"3.2.1","release":"3.2"}}`))
+	defer done()
+
+	// Find the PBS identifyFn from defaults.
+	var pbsFn func(*http.Response, []byte) (bool, map[string]string)
+	for _, tgt := range defaultProbeTargets {
+		if tgt.Type == ConnectionTypePBS {
+			pbsFn = tgt.identifyFn
+		}
+	}
+	if pbsFn == nil {
+		t.Fatal("pbs target not in defaults")
+	}
+	tgt := probeTarget{Type: ConnectionTypePBS, Scheme: "https", Port: port, Path: "/api2/json/version", identifyFn: pbsFn}
+	if _, ok := probeOne(context.Background(), host, tgt, testProbeClient()); !ok {
+		t.Fatal("expected PBS detection")
+	}
+}
+
+func TestProbeOne_DetectsPMGByProductField(t *testing.T) {
+	// Simulate a server that serves PMG content but with no distinctive Server header.
+	host, port, done := probeTestServer(t, func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, `{"data":{"product":"pmg","version":"8.1.2"}}`)
+	})
+	defer done()
+
+	var pmgFn func(*http.Response, []byte) (bool, map[string]string)
+	for _, tgt := range defaultProbeTargets {
+		if tgt.Type == ConnectionTypePMG {
+			pmgFn = tgt.identifyFn
+		}
+	}
+	tgt := probeTarget{Type: ConnectionTypePMG, Scheme: "https", Port: port, Path: "/api2/json/version", identifyFn: pmgFn}
+	if _, ok := probeOne(context.Background(), host, tgt, testProbeClient()); !ok {
+		t.Fatal("expected PMG detection via product field")
+	}
+}
+
+func TestProbeOne_DetectsVMware(t *testing.T) {
+	host, port, done := probeTestServer(t, vmwareHandler())
+	defer done()
+
+	var vmFn func(*http.Response, []byte) (bool, map[string]string)
+	for _, tgt := range defaultProbeTargets {
+		if tgt.Type == ConnectionTypeVMware {
+			vmFn = tgt.identifyFn
+		}
+	}
+	tgt := probeTarget{Type: ConnectionTypeVMware, Scheme: "https", Port: port, Path: "/sdk/vimServiceVersions.xml", identifyFn: vmFn}
+	if _, ok := probeOne(context.Background(), host, tgt, testProbeClient()); !ok {
+		t.Fatal("expected VMware detection")
+	}
+}
+
+func TestProbeOne_DetectsTrueNAS(t *testing.T) {
+	host, port, done := probeTestServer(t, truenasHandler())
+	defer done()
+
+	var tnFn func(*http.Response, []byte) (bool, map[string]string)
+	for _, tgt := range defaultProbeTargets {
+		if tgt.Type == ConnectionTypeTrueNAS {
+			tnFn = tgt.identifyFn
+		}
+	}
+	tgt := probeTarget{Type: ConnectionTypeTrueNAS, Scheme: "https", Port: port, Path: "/api/v2.0/system/product_name", identifyFn: tnFn}
+	if _, ok := probeOne(context.Background(), host, tgt, testProbeClient()); !ok {
+		t.Fatal("expected TrueNAS detection")
+	}
+}
+
+func TestProbeOne_RejectsNonMatchingServer(t *testing.T) {
+	host, port, done := probeTestServer(t, func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Server", "nginx/1.24")
+		fmt.Fprint(w, `<html><body>hello</body></html>`)
+	})
+	defer done()
+
+	for _, tgt := range defaultProbeTargets {
+		t := probeTarget{Type: tgt.Type, Scheme: "https", Port: port, Path: tgt.Path, identifyFn: tgt.identifyFn}
+		if _, ok := probeOne(context.Background(), host, t, testProbeClient()); ok {
+			return
+		}
+	}
+	// None matched — expected outcome.
+}
+
+func TestProbeOne_DoesNotConfusePVEForPMG(t *testing.T) {
+	host, port, done := probeTestServer(t, pveHandler(`{"data":{"version":"8.2.4","release":"8.2","repoid":"abc"}}`))
+	defer done()
+
+	var pmgFn func(*http.Response, []byte) (bool, map[string]string)
+	for _, tgt := range defaultProbeTargets {
+		if tgt.Type == ConnectionTypePMG {
+			pmgFn = tgt.identifyFn
+		}
+	}
+	tgt := probeTarget{Type: ConnectionTypePMG, Scheme: "https", Port: port, Path: "/api2/json/version", identifyFn: pmgFn}
+	if _, ok := probeOne(context.Background(), host, tgt, testProbeClient()); ok {
+		t.Fatal("PMG identifyFn should not match a PVE server")
+	}
+}
+
+func TestRunProbe_ReturnsSortedCandidates(t *testing.T) {
+	// Spin up one PVE server; runProbe should invoke all candidates but only
+	// detect PVE. Shorten the budget for the test.
+	host, port, done := probeTestServer(t, pveHandler(`{"data":{"version":"8.2.4","release":"8.2","repoid":"abc"}}`))
+	defer done()
+
+	// Override defaultProbeTargets to only the PVE target at our ephemeral port.
+	var pveFn func(*http.Response, []byte) (bool, map[string]string)
+	for _, tgt := range defaultProbeTargets {
+		if tgt.Type == ConnectionTypePVE {
+			pveFn = tgt.identifyFn
+		}
+	}
+	restore := withSingleTarget(t, probeTarget{Type: ConnectionTypePVE, Scheme: "https", Port: port, Path: "/api2/json/version", identifyFn: pveFn})
+	defer restore()
+
+	results, elapsed := runProbe(context.Background(), host, port, testProbeClient())
+	if len(results) != 1 || results[0].Type != ConnectionTypePVE {
+		t.Fatalf("unexpected results: %+v", results)
+	}
+	if elapsed <= 0 {
+		t.Fatal("elapsed should be positive")
+	}
+}
+
+func TestRunProbe_TotalBudgetEnforced(t *testing.T) {
+	// Build a dummy target that points nowhere — dial will be slow/timeout.
+	restore := withSingleTarget(t, probeTarget{
+		Type:   ConnectionTypePVE,
+		Scheme: "https",
+		Port:   1, // unlikely open
+		Path:   "/api2/json/version",
+		identifyFn: func(*http.Response, []byte) (bool, map[string]string) {
+			return true, nil
+		},
+	})
+	defer restore()
+
+	start := time.Now()
+	results, _ := runProbe(context.Background(), "127.0.0.1", 0, testProbeClient())
+	elapsed := time.Since(start)
+	if len(results) != 0 {
+		t.Fatalf("expected no results from closed port, got %+v", results)
+	}
+	if elapsed > 2*time.Second {
+		t.Fatalf("probe exceeded expected timeout: %s", elapsed)
+	}
+}
@@ -0,0 +1,84 @@
+package api
+
+import "time"
+
+// ConnectionState is the derived lifecycle state shown in the unified
+// connections ledger. Derivation rules live in connections_aggregator.go; no
+// new state is persisted — every value comes from existing runtime signals.
+type ConnectionState string
+
+const (
+	ConnectionStateActive       ConnectionState = "active"
+	ConnectionStatePaused       ConnectionState = "paused"
+	ConnectionStateUnauthorized ConnectionState = "unauthorized"
+	ConnectionStateUnreachable  ConnectionState = "unreachable"
+	ConnectionStateStale        ConnectionState = "stale"
+	ConnectionStatePending      ConnectionState = "pending"
+)
+
+// ConnectionType is the product family that owns the connection. The value is
+// the discriminator the frontend switches on to render per-type credential
+// slots in ConnectionEditor.
+type ConnectionType string
+
+const (
+	ConnectionTypePVE        ConnectionType = "pve"
+	ConnectionTypePBS        ConnectionType = "pbs"
+	ConnectionTypePMG        ConnectionType = "pmg"
+	ConnectionTypeVMware     ConnectionType = "vmware"
+	ConnectionTypeTrueNAS    ConnectionType = "truenas"
+	ConnectionTypeAgent      ConnectionType = "agent"
+	ConnectionTypeDocker     ConnectionType = "docker"
+	ConnectionTypeKubernetes ConnectionType = "kubernetes"
+)
+
+// ConnectionSource records how a connection entered Pulse.
+type ConnectionSource string
+
+const (
+	ConnectionSourceManual ConnectionSource = "manual"
+	ConnectionSourceAgent  ConnectionSource = "agent"
+	ConnectionSourceScript ConnectionSource = "script"
+)
+
+// ConnectionCapabilities tells the frontend which controls to render for a
+// connection. Agents cannot pause or partial-scope; manual Proxmox/PBS/PMG
+// can do both. This avoids putting the decision inside the editor.
+type ConnectionCapabilities struct {
+	SupportsPause bool `json:"supportsPause"`
+	SupportsScope bool `json:"supportsScope"`
+	SupportsTest  bool `json:"supportsTest"`
+}
+
+// ConnectionError is the runtime error shape surfaced on a connection row.
+// Mirrors monitoring.ErrorDetail but lives in the api package so the type
+// stays stable if the internal monitoring shape evolves.
+type ConnectionError struct {
+	At       time.Time `json:"at"`
+	Message  string    `json:"message"`
+	Category string    `json:"category,omitempty"`
+}
+
+// Connection is the unified row the frontend consumes. It replaces the
+// per-type shapes that today require separate fetches and separate table
+// renderers.
+type Connection struct {
+	ID           string                 `json:"id"`
+	Type         ConnectionType         `json:"type"`
+	Name         string                 `json:"name"`
+	Address      string                 `json:"address"`
+	State        ConnectionState        `json:"state"`
+	StateReason  string                 `json:"stateReason,omitempty"`
+	Enabled      bool                   `json:"enabled"`
+	Surfaces     []string               `json:"surfaces"`
+	Scope        map[string]bool        `json:"scope"`
+	LastSeen     *time.Time             `json:"lastSeen,omitempty"`
+	LastError    *ConnectionError       `json:"lastError,omitempty"`
+	Source       ConnectionSource       `json:"source"`
+	Capabilities ConnectionCapabilities `json:"capabilities"`
+}
+
+// ConnectionsListResponse is the envelope for GET /api/connections.
+type ConnectionsListResponse struct {
+	Connections []Connection `json:"connections"`
+}
@@ -10433,6 +10433,59 @@ func TestContract_ShippedSecurityDocReferencesStayLocal(t *testing.T) {
 	}
 }

+// TestContract_ConnectionPayloadShapeStaysCanonical pins the JSON shape of
+// the unified connections ledger so the Go types in
+// `internal/api/connections_types.go` stay in lockstep with the
+// `frontend-modern/src/api/connections.ts` client. Adding or renaming fields
+// on this wire shape requires an explicit update to both ends in the same
+// commit.
+func TestContract_ConnectionPayloadShapeStaysCanonical(t *testing.T) {
+	conn := Connection{
+		ID:           "pve-lab",
+		Type:         ConnectionTypePVE,
+		Name:         "lab",
+		Address:      "https://pve.lab:8006",
+		State:        ConnectionStateActive,
+		StateReason:  "",
+		Enabled:      true,
+		Surfaces:     []string{"vms", "containers"},
+		Scope:        map[string]bool{"vms": true, "containers": true},
+		LastSeen:     timePtr(time.Date(2026, 4, 19, 10, 0, 0, 0, time.UTC)),
+		LastError:    nil,
+		Source:       ConnectionSourceManual,
+		Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: true, SupportsTest: true},
+	}
+	body, err := json.Marshal(ConnectionsListResponse{Connections: []Connection{conn}})
+	if err != nil {
+		t.Fatalf("marshal Connection: %v", err)
+	}
+	want := `{"connections":[{"id":"pve-lab","type":"pve","name":"lab","address":"https://pve.lab:8006","state":"active","enabled":true,"surfaces":["vms","containers"],"scope":{"containers":true,"vms":true},"lastSeen":"2026-04-19T10:00:00Z","source":"manual","capabilities":{"supportsPause":true,"supportsScope":true,"supportsTest":true}}]}`
+	assertJSONSnapshot(t, body, want)
+}
+
+// TestContract_ProbePayloadShapeStaysCanonical pins the POST
+// /api/connections/probe wire shape. Hint keys are free-form; the envelope
+// fields (type, host, port, hints) are the contract boundary.
+func TestContract_ProbePayloadShapeStaysCanonical(t *testing.T) {
+	resp := ProbeResponse{
+		Candidates: []ProbeCandidate{
+			{
+				Type:  ConnectionTypePVE,
+				Host:  "https://pve.lab:8006",
+				Port:  8006,
+				Hints: map[string]string{"product": "Proxmox VE", "version": "8.2.4"},
+			},
+		},
+		ProbedMs: 812,
+	}
+	body, err := json.Marshal(resp)
+	if err != nil {
+		t.Fatalf("marshal ProbeResponse: %v", err)
+	}
+	want := `{"candidates":[{"type":"pve","host":"https://pve.lab:8006","port":8006,"hints":{"product":"Proxmox VE","version":"8.2.4"}}],"probedMs":812}`
+	assertJSONSnapshot(t, body, want)
+}
+
 func mustStreamEvent(t *testing.T, eventType string, data interface{}) chat.StreamEvent {
 	t.Helper()

@@ -300,6 +300,8 @@ var bareRouteAllowlist = []string{
 	"/api/config/nodes/test-config",
 	"/api/config/nodes/test-connection",
 	"/api/config/system",
+	"/api/connections",
+	"/api/connections/probe",
 	"/api/truenas/connections",
 	"/api/truenas/connections/preview",
 	"/api/truenas/connections/test",
@@ -433,6 +435,8 @@ var allRouteAllowlist = []string{
 	"/api/config/nodes/test-config",
 	"/api/config/nodes/test-connection",
 	"/api/config/nodes/",
+	"/api/connections",
+	"/api/connections/probe",
 	"/api/truenas/connections",
 	"/api/truenas/connections/preview",
 	"/api/truenas/connections/test",
@@ -72,6 +72,7 @@ type Router struct {
 	configHandlers             *ConfigHandlers
 	trueNASHandlers            *TrueNASHandlers
 	vmwareHandlers             *VMwareHandlers
+	connectionsHandlers        *ConnectionsHandlers
 	notificationHandlers       *NotificationHandlers
 	notificationQueueHandlers  *NotificationQueueHandlers
 	dockerAgentHandlers        *DockerAgentHandlers
@@ -386,6 +387,11 @@ func (r *Router) setupRoutes() {
 		getMonitor:     r.configHandlers.getMonitor,
 		getPoller:      func(context.Context) *monitoring.VMwarePoller { return r.vmwarePoller },
 	}
+	r.connectionsHandlers = NewConnectionsHandlers(
+		r.configHandlers.getConfig,
+		r.configHandlers.getPersistence,
+		r.configHandlers.getMonitor,
+	)
 	recoveryManager := recoverymanager.New(r.multiTenant)
 	r.recoveryHandlers = NewRecoveryHandlers(recoveryManager)
 	if r.mtMonitor != nil {
@@ -176,6 +176,32 @@ func (r *Router) registerConfigSystemRoutes(updateHandlers *UpdateHandlers) {
 		}
 	})

+	// Unified connections ledger — aggregates PVE/PBS/PMG/VMware/TrueNAS/agent rows.
+	r.mux.HandleFunc("/api/connections", func(w http.ResponseWriter, req *http.Request) {
+		if r.connectionsHandlers == nil {
+			writeErrorResponse(w, http.StatusServiceUnavailable, "connections_unavailable", "Connections service unavailable", nil)
+			return
+		}
+		if req.Method != http.MethodGet {
+			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+			return
+		}
+		RequireAdmin(r.config, RequireScope(config.ScopeSettingsRead, r.connectionsHandlers.HandleList))(w, req)
+	})
+
+	// Connection address probe — stateless type detection before credential entry.
+	r.mux.HandleFunc("/api/connections/probe", func(w http.ResponseWriter, req *http.Request) {
+		if r.connectionsHandlers == nil {
+			writeErrorResponse(w, http.StatusServiceUnavailable, "connections_unavailable", "Connections service unavailable", nil)
+			return
+		}
+		if req.Method != http.MethodPost {
+			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+			return
+		}
+		RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.connectionsHandlers.HandleProbe))(w, req)
+	})
+
 	// TrueNAS connection management
 	r.mux.HandleFunc("/api/truenas/connections", func(w http.ResponseWriter, req *http.Request) {
 		if r.trueNASHandlers == nil {
@@ -543,6 +543,11 @@ type PVEInstance struct {
 	// Agent tracking
 	Source      string // "agent" or "script" - how this node was registered (empty = legacy/manual)
 	DisableCeph bool   // Disable Ceph status polling for this instance
+
+	// Lifecycle. Disabled (not Enabled) so the zero value means "poll" —
+	// existing nodes.json records from before this field existed load
+	// unpaused without a schema migration.
+	Disabled bool `json:"disabled,omitempty"`
 }

 // ClusterEndpoint represents a single node in a cluster
@@ -595,6 +600,9 @@ type PBSInstance struct {

 	// Datastore exclusion (for unmounted/removable datastores that cause log noise)
 	ExcludeDatastores []string
+
+	// Lifecycle. See PVEInstance.Disabled.
+	Disabled bool `json:"disabled,omitempty"`
 }

 // PMGInstance represents a Proxmox Mail Gateway connection
@@ -615,6 +623,9 @@ type PMGInstance struct {
 	MonitorDomainStats           bool
 	TemperatureMonitoringEnabled *bool // Monitor temperature via SSH (nil = use global setting, true/false = override)
 	SSHPort                      int   // SSH port for temperature monitoring (0 = use global default)
+
+	// Lifecycle. See PVEInstance.Disabled.
+	Disabled bool `json:"disabled,omitempty"`
 }

 // Global persistence instance for saving
@@ -1331,3 +1331,43 @@ func TestReloadAndRuntimeContextStayOnCanonicalMonitoringPath(t *testing.T) {
 		}
 	}
 }
+
+// TestPollersHonorDisabledInstanceFlag asserts that the PVE, PBS, and PMG
+// pollers skip instances whose `Disabled` flag is set, both at client
+// initialization/reconnect time and on every per-instance iteration of the
+// poll loop. The unified connections ledger surfaces `Disabled` as a
+// `paused` state, so the monitoring runtime must not drive API calls or
+// mark a disabled instance reachable; that behavior is what keeps the
+// ledger's pause semantics honest across restarts.
+func TestPollersHonorDisabledInstanceFlag(t *testing.T) {
+	expectations := map[string][]string{
+		"monitor_client_init.go": {
+			"if pve.Disabled {",
+			"if pbsInst.Disabled {",
+			"if pmgInst.Disabled {",
+		},
+		"monitor_client_reconnect.go": {
+			"if pve.Disabled {",
+			"if pbsInst.Disabled {",
+		},
+		"monitor_pve.go": {
+			"if instanceCfg.Disabled {",
+		},
+		"monitor_pbs_pmg.go": {
+			"if instanceCfg.Disabled {",
+		},
+	}
+
+	for file, snippets := range expectations {
+		data, err := os.ReadFile(file)
+		if err != nil {
+			t.Fatalf("failed to read %s: %v", file, err)
+		}
+		source := string(data)
+		for _, snippet := range snippets {
+			if !strings.Contains(source, snippet) {
+				t.Fatalf("%s must contain %q so disabled instances stay off the monitoring hot path", file, snippet)
+			}
+		}
+	}
+}
@@ -14,6 +14,10 @@ import (
 func (m *Monitor) initPVEClients(cfg *config.Config) {
 	log.Info().Int("count", len(cfg.PVEInstances)).Msg("initializing PVE clients")
 	for _, pve := range cfg.PVEInstances {
+		if pve.Disabled {
+			log.Info().Str("instance", pve.Name).Msg("Skipping PVE client init: instance is paused")
+			continue
+		}
 		log.Info().
 			Str("name", pve.Name).
 			Str("host", pve.Host).
@@ -78,6 +82,10 @@ func (m *Monitor) initPVEClients(cfg *config.Config) {
 func (m *Monitor) initPBSClients(cfg *config.Config) {
 	log.Info().Int("count", len(cfg.PBSInstances)).Msg("initializing PBS clients")
 	for _, pbsInst := range cfg.PBSInstances {
+		if pbsInst.Disabled {
+			log.Info().Str("instance", pbsInst.Name).Msg("Skipping PBS client init: instance is paused")
+			continue
+		}
 		log.Info().
 			Str("name", pbsInst.Name).
 			Str("host", pbsInst.Host).
@@ -112,6 +120,10 @@ func (m *Monitor) initPBSClients(cfg *config.Config) {
 func (m *Monitor) initPMGClients(cfg *config.Config) {
 	log.Info().Int("count", len(cfg.PMGInstances)).Msg("initializing PMG clients")
 	for _, pmgInst := range cfg.PMGInstances {
+		if pmgInst.Disabled {
+			log.Info().Str("instance", pmgInst.Name).Msg("Skipping PMG client init: instance is paused")
+			continue
+		}
 		log.Info().
 			Str("name", pmgInst.Name).
 			Str("host", pmgInst.Host).
@@ -68,15 +68,23 @@ func (m *Monitor) retryFailedConnections(ctx context.Context) {
 		if m.config != nil {
 			connectionTimeout = m.config.ConnectionTimeout

-			// Find PVE instances without clients
+			// Find PVE instances without clients. Skip paused instances —
+			// a missing client for a paused instance is intentional, not a failure.
 			for _, pve := range m.config.PVEInstances {
+				if pve.Disabled {
+					continue
+				}
 				if _, exists := m.pveClients[pve.Name]; !exists {
 					missingPVE = append(missingPVE, pve)
 				}
 			}

-			// Find PBS instances without clients
+			// Find PBS instances without clients. Skip paused instances —
+			// a missing client for a paused instance is intentional, not a failure.
 			for _, pbsInst := range m.config.PBSInstances {
+				if pbsInst.Disabled {
+					continue
+				}
 				if _, exists := m.pbsClients[pbsInst.Name]; !exists {
 					missingPBS = append(missingPBS, pbsInst)
 				}
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"strings"
 	"testing"
 	"time"
@@ -198,3 +199,33 @@ func TestMonitor_PollPBSInstance_DatastoreDetails(t *testing.T) {
 		t.Error("DS2 not found")
 	}
 }
+
+// TestPBSAndPMGPollSkipDisabledInstances asserts that the PBS and PMG poll
+// entry points short-circuit when their resolved instance config carries
+// `Disabled: true`. This is a source-level guardrail for the discovery
+// provider surface: the unified connections ledger surfaces `Disabled` as
+// `paused`, and the PBS/PMG pollers must not drive live API calls or
+// surface ingest while that flag is set, across restarts or reloads.
+func TestPBSAndPMGPollSkipDisabledInstances(t *testing.T) {
+	data, err := os.ReadFile("monitor_pbs_pmg.go")
+	if err != nil {
+		t.Fatalf("failed to read monitor_pbs_pmg.go: %v", err)
+	}
+	source := string(data)
+
+	// Both PBS and PMG poll flows must explicitly guard on Disabled.
+	if count := strings.Count(source, "if instanceCfg.Disabled {"); count < 2 {
+		t.Fatalf("monitor_pbs_pmg.go must contain the Disabled-skip guard in both PBS and PMG poll entry points; found %d", count)
+	}
+
+	// The guards must short-circuit the poll with an early return so no
+	// downstream API client is constructed for a paused instance.
+	for _, snippet := range []string{
+		"Skipping PBS poll: instance is paused",
+		"Skipping PMG poll: instance is paused",
+	} {
+		if !strings.Contains(source, snippet) {
+			t.Fatalf("monitor_pbs_pmg.go must emit debug-log %q when skipping a disabled instance so operators can correlate paused ledger rows with runtime behavior", snippet)
+		}
+	}
+}
@@ -128,6 +128,12 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
 		log.Error().Str("instance", instanceName).Msg("PBS instance config not found")
 		return
 	}
+	if instanceCfg.Disabled {
+		if debugEnabled {
+			log.Debug().Str("instance", instanceName).Msg("Skipping PBS poll: instance is paused")
+		}
+		return
+	}

 	// Initialize PBS instance with default values
 	pbsInst := models.PBSInstance{
@@ -502,6 +508,12 @@ func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, clie
 		pollErr = fmt.Errorf("pmg instance config not found for %s", instanceName)
 		return
 	}
+	if instanceCfg.Disabled {
+		if debugEnabled {
+			log.Debug().Str("instance", instanceName).Msg("Skipping PMG poll: instance is paused")
+		}
+		return
+	}

 	now := time.Now()
 	pmgInst := models.PMGInstance{
@@ -957,6 +957,12 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie
 		pollErr = fmt.Errorf("pve instance config not found for %s", instanceName)
 		return
 	}
+	if instanceCfg.Disabled {
+		if debugEnabled {
+			log.Debug().Str("instance", instanceName).Msg("Skipping PVE poll: instance is paused")
+		}
+		return
+	}

 	// Poll nodes
 	nodes, updatedClient, err := m.fetchPVENodes(ctx, instanceName, instanceCfg, client)
@@ -3780,8 +3780,8 @@ class SubsystemLookupTest(unittest.TestCase):
                {
                    "heading": "## Shared Boundaries",
                    "path": "internal/api/access_control_handlers.go",
-                    "line": 116,
-                    "heading_line": 83,
+                    "line": 121,
+                    "heading_line": 88,
                }
            ],
        )