Add report-only Unified Agent observer destinations

2026-09-10 02:25:56 +00:00 · 2026-07-19 15:30:34 +01:00
parent ff93d99fa1
commit 4b066a0f1e
29 changed files with 1771 additions and 294 deletions
@@ -25,6 +25,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agentexec"
+	"github.com/rcourtman/pulse-go-rewrite/internal/agenttarget"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agentupdate"
 	"github.com/rcourtman/pulse-go-rewrite/internal/dockeragent"
 	"github.com/rcourtman/pulse-go-rewrite/internal/hostagent"
@@ -310,6 +311,7 @@ func run(ctx context.Context, args []string, getenv func(string) string) error {
 	logger.Info().
 		Str("version", Version).
 		Str("pulse_url", cfg.PulseURL).
+		Int("observer_destinations", len(cfg.Observers)).
 		Bool("host_enabled", cfg.EnableHost).
 		Bool("docker_enabled", cfg.EnableDocker).
 		Bool("kubernetes_enabled", cfg.EnableKubernetes).
@@ -322,6 +324,13 @@ func run(ctx context.Context, args []string, getenv func(string) string) error {
 			Str("pulse_url", cfg.PulseURL).
 			Msg("--allow-plaintext-http is set: the agent API token travels in cleartext to any non-loopback Pulse URL; only use this on a network you fully control")
 	}
+	for _, observer := range cfg.Observers {
+		if observer.AllowPlaintextHTTP {
+			logger.Warn().
+				Str("destination", observer.Name).
+				Msg("Observer destination explicitly permits plaintext HTTP; its API token may travel in cleartext")
+		}
+	}

 	// 5. Set prometheus info metric
 	agentInfo.WithLabelValues(
@@ -395,6 +404,7 @@ func run(ctx context.Context, args []string, getenv func(string) string) error {
 			AppliedConfig:      cfg.AppliedConfig,
 			UpdateStatus:       updater.Snapshot,
 			ModuleStatus:       runtimeStatus.moduleStatuses,
+			Observers:          hostObserverTargets(cfg.Observers),

 			DockerContainerUpdater: dockerUpdaterBridge,
 		}
@@ -439,6 +449,7 @@ func run(ctx context.Context, args []string, getenv func(string) string) error {
 			IncludeTasks:        true,
 			CollectDiskMetrics:  false,
 			DiskExclude:         cfg.DiskExclude,
+			Targets:             dockerReportTargets(cfg),
 		}

 		dockerAgent, err = newDockerAgent(dockerCfg)
@@ -496,6 +507,7 @@ func run(ctx context.Context, args []string, getenv func(string) string) error {
 			IncludeAllPods:        cfg.KubeIncludeAllPods,
 			IncludeAllDeployments: cfg.KubeIncludeAllDeployments,
 			MaxPods:               cfg.KubeMaxPods,
+			Targets:               kubernetesReportTargets(cfg),
 		}

 		agent, err := newKubeAgent(kubeCfg)
@@ -776,6 +788,8 @@ type Config struct {
 	AllowPlaintextHTTP bool
 	CACertPath         string
 	ServerFingerprint  string
+	ObserversFile      string
+	Observers          []agenttarget.Observer
 	DeploySSHUser      string
 	LogLevel           zerolog.Level
 	LogFile            string
@@ -826,6 +840,76 @@ type Config struct {
 	KubeMaxPods               int
 }

+func hostObserverTargets(observers []agenttarget.Observer) []hostagent.ObserverTarget {
+	targets := make([]hostagent.ObserverTarget, 0, len(observers))
+	for _, observer := range observers {
+		targets = append(targets, hostagent.ObserverTarget{
+			Name:               observer.Name,
+			ID:                 observer.ID,
+			PulseURL:           observer.URL,
+			APIToken:           observer.Token,
+			InsecureSkipVerify: observer.InsecureSkipVerify,
+			AllowPlaintextHTTP: observer.AllowPlaintextHTTP,
+			CACertPath:         observer.CACertPath,
+			ServerFingerprint:  observer.ServerFingerprint,
+			ProvisionProxmox:   observer.ProvisionProxmox,
+		})
+	}
+	return targets
+}
+
+func dockerReportTargets(cfg Config) []dockeragent.TargetConfig {
+	targets := make([]dockeragent.TargetConfig, 0, len(cfg.Observers)+1)
+	targets = append(targets, dockeragent.TargetConfig{
+		Name:               "primary",
+		URL:                cfg.PulseURL,
+		Token:              cfg.APIToken,
+		InsecureSkipVerify: cfg.InsecureSkipVerify,
+		AllowPlaintextHTTP: cfg.AllowPlaintextHTTP,
+		CACertPath:         cfg.CACertPath,
+		ServerFingerprint:  cfg.ServerFingerprint,
+		Authoritative:      true,
+	})
+	for _, observer := range cfg.Observers {
+		targets = append(targets, dockeragent.TargetConfig{
+			Name:               observer.Name,
+			URL:                observer.URL,
+			Token:              observer.Token,
+			InsecureSkipVerify: observer.InsecureSkipVerify,
+			AllowPlaintextHTTP: observer.AllowPlaintextHTTP,
+			CACertPath:         observer.CACertPath,
+			ServerFingerprint:  observer.ServerFingerprint,
+		})
+	}
+	return targets
+}
+
+func kubernetesReportTargets(cfg Config) []kubernetesagent.TargetConfig {
+	targets := make([]kubernetesagent.TargetConfig, 0, len(cfg.Observers)+1)
+	targets = append(targets, kubernetesagent.TargetConfig{
+		Name:               "primary",
+		URL:                cfg.PulseURL,
+		Token:              cfg.APIToken,
+		InsecureSkipVerify: cfg.InsecureSkipVerify,
+		AllowPlaintextHTTP: cfg.AllowPlaintextHTTP,
+		CACertPath:         cfg.CACertPath,
+		ServerFingerprint:  cfg.ServerFingerprint,
+		Authoritative:      true,
+	})
+	for _, observer := range cfg.Observers {
+		targets = append(targets, kubernetesagent.TargetConfig{
+			Name:               observer.Name,
+			URL:                observer.URL,
+			Token:              observer.Token,
+			InsecureSkipVerify: observer.InsecureSkipVerify,
+			AllowPlaintextHTTP: observer.AllowPlaintextHTTP,
+			CACertPath:         observer.CACertPath,
+			ServerFingerprint:  observer.ServerFingerprint,
+		})
+	}
+	return targets
+}
+
 func loadConfig(args []string, getenv func(string) string) (Config, error) {
 	// Environment Variables
 	envURL := strings.TrimSpace(getenv("PULSE_URL"))
@@ -867,6 +951,7 @@ func loadConfig(args []string, getenv func(string) string) (Config, error) {
 	envDiskExclude := strings.TrimSpace(getenv("PULSE_DISK_EXCLUDE"))
 	envReportIP := strings.TrimSpace(getenv("PULSE_REPORT_IP"))
 	envDisableCeph := strings.TrimSpace(getenv("PULSE_DISABLE_CEPH"))
+	envObserversFile := strings.TrimSpace(getenv("PULSE_OBSERVERS_FILE"))

 	// Defaults
 	defaultInterval := 30 * time.Second
@@ -920,6 +1005,7 @@ func loadConfig(args []string, getenv func(string) string) (Config, error) {
 	allowPlaintextHTTPFlag := fs.Bool("allow-plaintext-http", utils.ParseBool(strings.TrimSpace(getenv("PULSE_AGENT_ALLOW_PLAINTEXT_HTTP"))), "Allow plain HTTP to a Pulse server that does not look local (sends the API token in cleartext; only for networks you fully control)")
 	caCertFlag := fs.String("cacert", envCACertPath, "Path to custom CA bundle for agent HTTPS transport")
 	serverFingerprintFlag := fs.String("server-fingerprint", envServerFingerprint, "Expected Pulse server TLS certificate fingerprint (SHA256)")
+	observersFileFlag := fs.String("observers-file", envObserversFile, "Absolute path to a private JSON file defining report-only Pulse observer destinations")
 	deploySSHUserFlag := fs.String("deploy-ssh-user", envDeploySSHUser, "SSH user for peer deploy fan-out (default: root; non-root requires passwordless sudo)")
 	logLevelFlag := fs.String("log-level", defaultLogLevel(envLogLevel), "Log level")
 	logFileFlag := fs.String("log-file", envLogFile, "Write rotating JSON logs to this file")
@@ -977,6 +1063,10 @@ func loadConfig(args []string, getenv func(string) string) (Config, error) {

 	// Resolve token with priority: --token > --token-file > env > default file
 	token := resolveToken(*tokenFlag, *tokenFileFlag, envToken)
+	observers, err := agenttarget.LoadObservers(strings.TrimSpace(*observersFileFlag), pulseURL)
+	if err != nil {
+		return Config{}, fmt.Errorf("load observer destinations: %w", err)
+	}

 	// When --enroll is set and a runtime token already exists from a previous
 	// enrollment, use it instead of the bootstrap token embedded in the service
@@ -1057,6 +1147,8 @@ func loadConfig(args []string, getenv func(string) string) (Config, error) {
 		AllowPlaintextHTTP:        *allowPlaintextHTTPFlag,
 		CACertPath:                strings.TrimSpace(*caCertFlag),
 		ServerFingerprint:         strings.TrimSpace(*serverFingerprintFlag),
+		ObserversFile:             strings.TrimSpace(*observersFileFlag),
+		Observers:                 observers,
 		DeploySSHUser:             deploySSHUser,
 		LogLevel:                  logLevel,
 		LogFile:                   strings.TrimSpace(*logFileFlag),
@@ -0,0 +1,40 @@
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestLoadConfigBuildsReportOnlyObserverTargets(t *testing.T) {
+	dir := t.TempDir()
+	tokenPath := filepath.Join(dir, "dev.token")
+	if err := os.WriteFile(tokenPath, []byte("observer-token"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	configPath := filepath.Join(dir, "observers.json")
+	config := `{"version":1,"observers":[{"name":"dev","url":"http://127.0.0.1:7656","tokenFile":"` + tokenPath + `"}]}`
+	if err := os.WriteFile(configPath, []byte(config), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	cfg, err := loadConfig([]string{
+		"--url", "http://127.0.0.1:7655",
+		"--token", "primary-token",
+		"--observers-file", configPath,
+	}, func(string) string { return "" })
+	if err != nil {
+		t.Fatalf("loadConfig: %v", err)
+	}
+	if len(cfg.Observers) != 1 || cfg.Observers[0].Token != "observer-token" {
+		t.Fatalf("observers = %+v", cfg.Observers)
+	}
+	dockerTargets := dockerReportTargets(cfg)
+	if len(dockerTargets) != 2 || !dockerTargets[0].Authoritative || dockerTargets[1].Authoritative {
+		t.Fatalf("docker targets = %+v", dockerTargets)
+	}
+	kubeTargets := kubernetesReportTargets(cfg)
+	if len(kubeTargets) != 2 || !kubeTargets[0].Authoritative || kubeTargets[1].Authoritative {
+		t.Fatalf("kubernetes targets = %+v", kubeTargets)
+	}
+}
@@ -102,6 +102,7 @@ func (ws *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, cha
 			AppliedConfig:      ws.cfg.AppliedConfig,
 			UpdateStatus:       updater.Snapshot,
 			ModuleStatus:       runtimeStatus.moduleStatuses,
+			Observers:          hostObserverTargets(ws.cfg.Observers),
 		}
 		agent, err := hostagent.New(hostCfg)
 		if err != nil {
@@ -146,6 +147,7 @@ func (ws *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, cha
 			IncludeServices:    true,
 			IncludeTasks:       true,
 			CollectDiskMetrics: true,
+			Targets:            dockerReportTargets(ws.cfg),
 		}

 		agent, err := dockeragent.New(dockerCfg)
@@ -191,6 +193,7 @@ func (ws *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, cha
 			IncludeAllPods:        ws.cfg.KubeIncludeAllPods,
 			IncludeAllDeployments: ws.cfg.KubeIncludeAllDeployments,
 			MaxPods:               ws.cfg.KubeMaxPods,
+			Targets:               kubernetesReportTargets(ws.cfg),
 		}
 		agent, err := kubernetesagent.New(kubeCfg)
 		if err != nil {
@@ -31,11 +31,13 @@ Run it on the host that already has the v5 `pulse-agent` service to replace the
 binary and service configuration in place; do not uninstall the old service
 first unless you are intentionally removing that host from Pulse.

-An installed agent is configured for one Pulse URL and one token. Do not point
-one running service at both a v5 server and a v6 server. After the upgrade,
-check the relevant platform page or **Machines** view once the agent has
-reported, and confirm the host-local version with `pulse-agent --version` if
-the UI has not received a fresh report yet.
+An installed agent has one **primary** Pulse URL and token. The primary is the
+only server allowed to supply remote configuration, commands, enrollment, or
+updates. The same collection can also be sent to explicitly configured,
+report-only **observer** instances; see [Observer destinations](#observer-destinations).
+After an upgrade, check the relevant platform page or **Machines** view once
+the agent has reported, and confirm the host-local version with
+`pulse-agent --version` if the UI has not received a fresh report yet.

 ### Linux (systemd)
 ```bash
@@ -88,6 +90,7 @@ curl -fsSL http://<pulse-ip>:7655/install.sh | \
 |------|---------|-------------|---------|
 | `--url` | `PULSE_URL` | Pulse server URL | `http://localhost:7655` |
 | `--token` | `PULSE_TOKEN` | API token | *(required)* |
+| `--observers-file` | `PULSE_OBSERVERS_FILE` | Private JSON file defining report-only destinations | *(none)* |
 | `--token-file` | - | Read API token from file | *(unset)* |
 | `--interval` | `PULSE_INTERVAL` | Reporting interval | `30s` |
 | `--enable-host` | `PULSE_ENABLE_HOST` | Enable host metrics | `true` |
@@ -124,6 +127,52 @@ health/metrics endpoint over the network. Use `--health-addr ""` or

 **Token resolution order**: `--token` → `--token-file` → `PULSE_TOKEN` → `/var/lib/pulse-agent/token`.

+## Observer destinations
+
+Observer destinations receive the same already-collected host, Docker/Podman,
+and Kubernetes reports. Collection runs once per interval. Delivery, retries,
+and persisted host-report buffers are isolated per destination, so an observer
+outage does not replay or block the primary stream. Observer responses cannot
+change configuration, execute commands, enroll the agent, or select updates.
+
+Create a separate API token on each observer and store every token in its own
+absolute-path file. On Unix, both the JSON file and token files must be regular,
+non-symlink files with no group or other permissions (for example mode `0600`).
+
+```json
+{
+  "version": 1,
+  "observers": [
+    {
+      "name": "dev",
+      "url": "https://pulse-dev.example.test",
+      "tokenFile": "/etc/pulse-agent/dev-observer.token",
+      "serverFingerprint": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+      "provisionProxmox": true
+    }
+  ]
+}
+```
+
+Start or install the service with
+`--observers-file /etc/pulse-agent/observers.json`. Plaintext remote HTTP is
+rejected unless that observer explicitly sets `"allowPlaintextHTTP": true`.
+`insecureSkipVerify` is available per observer but should be replaced with a
+CA file or certificate fingerprint wherever possible.
+
+When Proxmox integration is enabled, each observer gets a distinct
+destination-scoped PVE/PBS API token and registration-state directory. Pulse
+must answer the registration check before the agent creates or rotates any
+Proxmox token; an unavailable destination therefore leaves existing
+credentials unchanged. Set `"provisionProxmox": false` when an observer should
+receive only Unified Agent telemetry and no separately registered PVE/PBS
+source.
+
+Per-destination delivery status is exported on the health listener as
+`pulse_agent_destination_configured` and
+`pulse_agent_destination_delivery_up`, labelled by module, destination, and
+role.
+
 ### Advanced Flags

 - `--version`: Print the agent version and exit.
@@ -123,14 +123,16 @@ pulse-agent --version
 systemctl status pulse-agent
 ```

-### Can one installed Pulse Unified Agent report to both a Pulse v5 instance and a Pulse v6 instance at the same time?
+### Can one installed Pulse Unified Agent report to two Pulse instances at the same time?

-Not as a supported in-place setup. A running Unified Agent installation is
-configured against one Pulse URL and one token, and it fetches remote config
-from that one Pulse server. If you need side-by-side evaluation, use a
-separate test host or VM, a cloned lab machine, or a separate isolated agent
-installation instead of trying to point one running agent service at two Pulse
-servers.
+Yes. Configure one instance as the primary with `--url` and its token, then add
+the other as a report-only observer with `--observers-file`. Only the primary
+can supply remote configuration, commands, enrollment, or updates; observer
+delivery and retries are isolated. Each instance needs its own Pulse API token,
+and Proxmox observers use separate PVE/PBS tokens. See
+[Observer destinations](UNIFIED_AGENT.md#observer-destinations) for the file
+format and security requirements. Use a v6-capable agent for this topology;
+older v5 agents do not understand observer configuration.

 ### Can I keep Pulse v5 stable while I test Pulse v6?

@@ -1,6 +1,6 @@
 # Pulse v6 Source Of Truth

-Last updated: 2026-07-14
+Last updated: 2026-07-19
 Status: ACTIVE

 This file is the stable human governance layer for the active v6 release
@@ -1,6 +1,6 @@
 {
  "version": "6.0",
-  "updated_at": "2026-07-14",
+  "updated_at": "2026-07-19",
  "scope": {
    "active_repos": [
      "pulse",
@@ -80,6 +80,21 @@
        }
      ],
      "evidence": [
+        {
+          "repo": "pulse",
+          "path": "cmd/pulse-agent/main.go",
+          "kind": "file"
+        },
+        {
+          "repo": "pulse",
+          "path": "cmd/pulse-agent/observers_config_test.go",
+          "kind": "file"
+        },
+        {
+          "repo": "pulse",
+          "path": "cmd/pulse-agent/service_windows.go",
+          "kind": "file"
+        },
        {
          "repo": "pulse",
          "path": "frontend-modern/src/components/Infrastructure/__tests__/UnifiedResourceTable.workloads-link.test.tsx",
@@ -5927,7 +5942,7 @@
      "status": "target-met",
      "completion": {
        "state": "complete",
-        "summary": "Agent lifecycle and fleet operations are at the current governed RC floor: canonical auto-register now converges on one v6 contract, install-command consumers fail closed through shared validated boundaries, profile-management surfaces preserve the canonical malformed-payload and missing-profile resync contract across settings and deploy surfaces, and the runtime-side Unified Agent reporting path now uses the canonical product terminology.",
+        "summary": "Agent lifecycle and fleet operations are at the current governed RC floor: canonical auto-register converges on one v6 contract, install-command consumers fail closed through shared validated boundaries, and one Unified Agent can fan a single collection to one authoritative primary plus isolated report-only observers. Observer tokens, TLS policy, retry buffers, delivery metrics, and Proxmox credentials are destination-scoped; Proxmox token mutation is gated on a successful destination registration check.",
        "tracking": []
      },
      "blockers": [],
@@ -6085,6 +6100,16 @@
          "path": "frontend-modern/src/utils/nodeModalPresentation.ts",
          "kind": "file"
        },
+        {
+          "repo": "pulse",
+          "path": "internal/agenttarget/config.go",
+          "kind": "file"
+        },
+        {
+          "repo": "pulse",
+          "path": "internal/agenttarget/config_test.go",
+          "kind": "file"
+        },
        {
          "repo": "pulse",
          "path": "internal/agentupdate/update.go",
@@ -6120,11 +6145,21 @@
          "path": "internal/api/unified_agent.go",
          "kind": "file"
        },
+        {
+          "repo": "pulse",
+          "path": "internal/dockeragent/multi_target_test.go",
+          "kind": "file"
+        },
        {
          "repo": "pulse",
          "path": "internal/hostagent/agent.go",
          "kind": "file"
        },
+        {
+          "repo": "pulse",
+          "path": "internal/hostagent/observer_delivery_test.go",
+          "kind": "file"
+        },
        {
          "repo": "pulse",
          "path": "internal/hostagent/proxmox_setup.go",
@@ -6135,6 +6170,16 @@
          "path": "internal/hostagent/proxmox_setup_test.go",
          "kind": "file"
        },
+        {
+          "repo": "pulse",
+          "path": "internal/kubernetesagent/multi_target_test.go",
+          "kind": "file"
+        },
+        {
+          "repo": "pulse",
+          "path": "scripts/install.ps1",
+          "kind": "file"
+        },
        {
          "repo": "pulse",
          "path": "scripts/install.sh",
@@ -4642,6 +4642,13 @@ file. Proxmox registration is also destination-scoped: the primary retains its
 legacy token name for upgrade continuity, observers use distinct token names
 and state markers, and every setup path must obtain a successful registration
 state response before any create, delete, or rotation command is executed.
+Plaintext transport consent is destination-scoped too. The primary keeps the
+existing local-network compatibility and explicit process-level override, but
+an observer must opt in independently for every non-loopback HTTP URL; primary
+consent cannot silently widen an observer. That policy must survive the shared
+config loader, host/Docker/Kubernetes target normalization, and observer
+Proxmox registration path. Unix and Windows service installation must preserve
+the absolute observer-config path across an in-place update.

 The adjacent recovery handlers under `internal/api/` do not widen this agent
 lifecycle boundary. Protection posture is a read-only `monitoring:read`
@@ -2443,3 +2443,13 @@ on community and Pro binaries. The rollback tests in
 `internal/updates/manager_rollback_test.go`, the rollback handler tests in
 `internal/api/updates_test.go`, and the route inventory pin for
 `/api/updates/rollback` are the proof surface for this path.
+
+### Observer destination installation continuity
+
+Unix and Windows installers accept `--observers-file` and preserve the absolute
+path in the installed service command. Unix installation rejects relative,
+missing, and symlink configuration paths before service mutation. The runtime
+remains the final schema, permission, token-file, URL, and TLS-policy validator.
+Updates recover the observer-file argument from the existing service command so
+an in-place binary refresh does not silently collapse a multi-destination
+installation back to primary-only reporting.
@@ -1750,6 +1750,12 @@ The local agent health listener exports
 to `primary` or `observer`; destination names come from validated configuration.
 Observer delivery failure is visible but does not make the primary authority
 unready or merge observer retry state into primary delivery health.
+Host, Docker/Podman, and Kubernetes reporters each fan out the already-collected
+snapshot without triggering a second collection. Their retry queues and latest
+delivery gauges remain per destination, and Kubernetes observer transport uses
+the same destination-scoped TLS and explicit plaintext policy as the host and
+Docker reporters. Observer acknowledgements never change the canonical
+monitoring configuration returned by the primary.

 ### PBS protection evidence collection

@@ -1101,6 +1101,7 @@
      "contract": "docs/release-control/v6/internal/subsystems/agent-lifecycle.md",
      "owned_prefixes": [
        "internal/agentexec/",
+        "internal/agenttarget/",
        "internal/agentupdate/",
        "internal/hostagent/",
        "scripts/intelligence_lab/"
@@ -1108,6 +1109,7 @@
      "owned_files": [
        ".github/workflows/unified-agent-native.yml",
        "cmd/pulse-agent/main.go",
+        "cmd/pulse-agent/service_windows.go",
        "frontend-modern/src/api/agentProfiles.ts",
        "frontend-modern/src/api/nodes.ts",
        "frontend-modern/src/components/Settings/agentProfileSettings.ts",
@@ -1215,12 +1217,14 @@
            "label": "pulse-agent CLI entrypoint proof",
            "match_prefixes": [],
            "match_files": [
-              "cmd/pulse-agent/main.go"
+              "cmd/pulse-agent/main.go",
+              "cmd/pulse-agent/service_windows.go"
            ],
            "allow_same_subsystem_tests": false,
            "test_prefixes": [],
            "exact_files": [
-              "cmd/pulse-agent/main_test.go"
+              "cmd/pulse-agent/main_test.go",
+              "cmd/pulse-agent/observers_config_test.go"
            ]
          },
          {
@@ -1263,7 +1267,9 @@
          {
            "id": "agent-runtime-transport-trust",
            "label": "agent runtime transport trust proof",
-            "match_prefixes": [],
+            "match_prefixes": [
+              "internal/agenttarget/"
+            ],
            "match_files": [
              "internal/agenttls/config.go",
              "internal/dockeragent/agent.go",
@@ -1275,9 +1281,12 @@
            "allow_same_subsystem_tests": false,
            "test_prefixes": [],
            "exact_files": [
+              "internal/agenttarget/config_test.go",
              "internal/agenttls/config_test.go",
              "internal/dockeragent/agent_internal_test.go",
+              "internal/dockeragent/multi_target_test.go",
              "internal/kubernetesagent/agent_new_test.go",
+              "internal/kubernetesagent/multi_target_test.go",
              "internal/remoteconfig/client_additional_test.go",
              "internal/remoteconfig/client_test.go",
              "internal/securityutil/httpurl_test.go"
@@ -1317,6 +1326,7 @@
              "internal/hostagent/commands_host_update_test.go",
              "internal/hostagent/commands_storage_cleanup_test.go",
              "internal/hostagent/docker_lifecycle_test.go",
+              "internal/hostagent/observer_delivery_test.go",
              "internal/hostagent/package_updates_test.go",
              "internal/hostagent/send_report_test.go",
              "internal/hostagent/storage_cleanup_test.go"
@@ -5078,6 +5088,7 @@
            "exact_files": [
              "internal/kubernetesagent/agent_inventory_test.go",
              "internal/kubernetesagent/agent_new_test.go",
+              "internal/kubernetesagent/multi_target_test.go",
              "internal/monitoring/kubernetes_agents_test.go",
              "internal/unifiedresources/adapter_coverage_test.go",
              "internal/unifiedresources/kubernetes_registry_test.go"
@@ -1434,6 +1434,10 @@ enforce TLS, CA, fingerprint, and explicit plaintext policy. Observer payload
 responses are report acknowledgements only and cannot authorize configuration,
 commands, enrollment, or updates. Per-destination Proxmox tokens prevent one
 Pulse instance from rotating credentials used by another.
+The process-wide plaintext override belongs only to the authoritative primary:
+it cannot satisfy an observer's opt-in. Every non-loopback HTTP observer must
+declare its own `allowPlaintextHTTP` consent, and that decision must propagate
+unchanged through host, Docker, Kubernetes, and observer Proxmox transports.

 Operational Trust action offers enforce current plan, approve, and execute
 authority before a mutating affordance is returned. Planning repeats those
@@ -0,0 +1,242 @@
+package agenttarget
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/url"
+	"os"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"unicode"
+
+	"github.com/rcourtman/pulse-go-rewrite/pkg/securityutil"
+)
+
+const ConfigVersion = 1
+
+const maxConfigBytes = 256 * 1024
+const maxObservers = 16
+const maxTokenBytes = 64 * 1024
+
+// Observer is a report-only Pulse destination. The primary Pulse URL remains
+// the sole authority for remote configuration, commands, and agent updates.
+type Observer struct {
+	Name               string
+	URL                string
+	Token              string
+	InsecureSkipVerify bool
+	AllowPlaintextHTTP bool
+	CACertPath         string
+	ServerFingerprint  string
+	ProvisionProxmox   bool
+	ID                 string
+}
+
+type observerFile struct {
+	Version   int                  `json:"version"`
+	Observers []observerFileTarget `json:"observers"`
+}
+
+type observerFileTarget struct {
+	Name               string `json:"name"`
+	URL                string `json:"url"`
+	TokenFile          string `json:"tokenFile"`
+	InsecureSkipVerify bool   `json:"insecureSkipVerify,omitempty"`
+	AllowPlaintextHTTP bool   `json:"allowPlaintextHTTP,omitempty"`
+	CACertPath         string `json:"caCertPath,omitempty"`
+	ServerFingerprint  string `json:"serverFingerprint,omitempty"`
+	ProvisionProxmox   *bool  `json:"provisionProxmox,omitempty"`
+}
+
+// LoadObservers validates and resolves an owner-private observer configuration.
+// Raw tokens are never accepted in the JSON document; each destination must
+// reference a separately permissioned token file.
+func LoadObservers(path string, primaryURL string) ([]Observer, error) {
+	path = strings.TrimSpace(path)
+	if path == "" {
+		return nil, nil
+	}
+	if !filepath.IsAbs(path) {
+		return nil, errors.New("observer config path must be absolute")
+	}
+	if err := requirePrivateRegularFile(path, "observer config", maxConfigBytes); err != nil {
+		return nil, err
+	}
+
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, fmt.Errorf("open observer config: %w", err)
+	}
+	defer f.Close()
+
+	dec := json.NewDecoder(io.LimitReader(f, maxConfigBytes+1))
+	dec.DisallowUnknownFields()
+	var parsed observerFile
+	if err := dec.Decode(&parsed); err != nil {
+		return nil, fmt.Errorf("decode observer config: %w", err)
+	}
+	if err := ensureJSONEOF(dec); err != nil {
+		return nil, err
+	}
+	if parsed.Version != ConfigVersion {
+		return nil, fmt.Errorf("observer config version %d is unsupported; expected %d", parsed.Version, ConfigVersion)
+	}
+	if len(parsed.Observers) > maxObservers {
+		return nil, fmt.Errorf("observer config has %d destinations; maximum is %d", len(parsed.Observers), maxObservers)
+	}
+
+	normalizedPrimary := ""
+	if strings.TrimSpace(primaryURL) != "" {
+		primary, err := NormalizePulseURL(primaryURL, true, true)
+		if err != nil {
+			return nil, fmt.Errorf("normalize primary Pulse URL: %w", err)
+		}
+		normalizedPrimary = primary
+	}
+
+	result := make([]Observer, 0, len(parsed.Observers))
+	seenNames := make(map[string]struct{}, len(parsed.Observers))
+	seenURLs := make(map[string]struct{}, len(parsed.Observers))
+	for index, item := range parsed.Observers {
+		observer, err := resolveObserver(item)
+		if err != nil {
+			return nil, fmt.Errorf("observer %d: %w", index+1, err)
+		}
+		nameKey := strings.ToLower(observer.Name)
+		if _, exists := seenNames[nameKey]; exists {
+			return nil, fmt.Errorf("observer %d: duplicate name %q", index+1, observer.Name)
+		}
+		if _, exists := seenURLs[observer.URL]; exists {
+			return nil, fmt.Errorf("observer %d: duplicate URL", index+1)
+		}
+		if normalizedPrimary != "" && observer.URL == normalizedPrimary {
+			return nil, fmt.Errorf("observer %d: URL duplicates the primary Pulse destination", index+1)
+		}
+		seenNames[nameKey] = struct{}{}
+		seenURLs[observer.URL] = struct{}{}
+		result = append(result, observer)
+	}
+	return result, nil
+}
+
+func resolveObserver(item observerFileTarget) (Observer, error) {
+	name := strings.TrimSpace(item.Name)
+	if err := validateName(name); err != nil {
+		return Observer{}, err
+	}
+	url, err := NormalizePulseURL(item.URL, false, item.AllowPlaintextHTTP)
+	if err != nil {
+		return Observer{}, fmt.Errorf("invalid URL: %w", err)
+	}
+
+	tokenPath := strings.TrimSpace(item.TokenFile)
+	if !filepath.IsAbs(tokenPath) {
+		return Observer{}, errors.New("tokenFile must be an absolute path")
+	}
+	if err := requirePrivateRegularFile(tokenPath, "observer token", maxTokenBytes); err != nil {
+		return Observer{}, err
+	}
+	tokenBytes, err := os.ReadFile(tokenPath)
+	if err != nil {
+		return Observer{}, fmt.Errorf("read observer token: %w", err)
+	}
+	token := strings.TrimSpace(string(tokenBytes))
+	if token == "" {
+		return Observer{}, errors.New("observer token file is empty")
+	}
+	if strings.IndexFunc(token, unicode.IsSpace) >= 0 {
+		return Observer{}, errors.New("observer token contains whitespace")
+	}
+
+	provisionProxmox := true
+	if item.ProvisionProxmox != nil {
+		provisionProxmox = *item.ProvisionProxmox
+	}
+	return Observer{
+		Name:               name,
+		URL:                url,
+		Token:              token,
+		InsecureSkipVerify: item.InsecureSkipVerify,
+		AllowPlaintextHTTP: item.AllowPlaintextHTTP,
+		CACertPath:         strings.TrimSpace(item.CACertPath),
+		ServerFingerprint:  strings.TrimSpace(item.ServerFingerprint),
+		ProvisionProxmox:   provisionProxmox,
+		ID:                 stableID(url),
+	}, nil
+}
+
+// NormalizePulseURL applies the transport policy for one report destination.
+// Authoritative primary destinations retain the established self-hosted
+// local-network HTTP allowance. Report-only observers require an explicit
+// per-destination opt-in for every non-loopback plaintext URL, even when the
+// process-wide primary override is enabled.
+func NormalizePulseURL(raw string, authoritative bool, allowPlaintext bool) (string, error) {
+	if !authoritative && !allowPlaintext {
+		parsed, err := url.Parse(strings.TrimSpace(raw))
+		if err == nil && parsed.Hostname() != "" && strings.EqualFold(parsed.Scheme, "http") &&
+			!securityutil.IsLoopbackHost(parsed.Hostname()) {
+			return "", fmt.Errorf("Pulse URL %q must use https unless the observer explicitly allows plaintext HTTP", raw)
+		}
+	}
+	parsed, err := securityutil.NormalizePulseHTTPBaseURLWithOptions(raw, securityutil.PulseURLValidationOptions{
+		AllowLocalNetworkHTTP:      authoritative,
+		AllowOperatorPlaintextHTTP: allowPlaintext,
+	})
+	if err != nil {
+		return "", err
+	}
+	return strings.TrimRight(parsed.String(), "/"), nil
+}
+
+func validateName(name string) error {
+	if name == "" {
+		return errors.New("name is required")
+	}
+	if len(name) > 64 {
+		return errors.New("name must be 64 characters or fewer")
+	}
+	for _, r := range name {
+		if unicode.IsLetter(r) || unicode.IsDigit(r) || r == '-' || r == '_' || r == '.' {
+			continue
+		}
+		return fmt.Errorf("name %q contains unsupported characters", name)
+	}
+	return nil
+}
+
+func requirePrivateRegularFile(path string, label string, maxBytes int64) error {
+	info, err := os.Lstat(path)
+	if err != nil {
+		return fmt.Errorf("stat %s file: %w", label, err)
+	}
+	if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() {
+		return fmt.Errorf("%s file must be a regular file, not a symlink", label)
+	}
+	if info.Size() > maxBytes {
+		return fmt.Errorf("%s file exceeds the %d-byte size limit", label, maxBytes)
+	}
+	if runtime.GOOS != "windows" && info.Mode().Perm()&0o077 != 0 {
+		return fmt.Errorf("%s file permissions must not grant group or other access", label)
+	}
+	return nil
+}
+
+func ensureJSONEOF(dec *json.Decoder) error {
+	var extra any
+	if err := dec.Decode(&extra); err == io.EOF {
+		return nil
+	} else if err != nil {
+		return fmt.Errorf("decode observer config trailer: %w", err)
+	}
+	return errors.New("observer config contains multiple JSON values")
+}
+
+func stableID(url string) string {
+	sum := sha256.Sum256([]byte(strings.TrimSpace(url)))
+	return hex.EncodeToString(sum[:8])
+}
@@ -0,0 +1,163 @@
+package agenttarget
+
+import (
+	"os"
+	"path/filepath"
+	"runtime"
+	"testing"
+
+	"github.com/rcourtman/pulse-go-rewrite/pkg/securityutil"
+)
+
+func writePrivateFile(t *testing.T, path string, content string) {
+	t.Helper()
+	if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestLoadObservers(t *testing.T) {
+	dir := t.TempDir()
+	tokenPath := filepath.Join(dir, "dev.token")
+	configPath := filepath.Join(dir, "observers.json")
+	writePrivateFile(t, tokenPath, "observer-secret\n")
+	writePrivateFile(t, configPath, `{
+  "version": 1,
+  "observers": [{
+    "name": "dev",
+    "url": "http://192.168.0.10:7655/",
+    "tokenFile": "`+tokenPath+`",
+    "allowPlaintextHTTP": true,
+    "provisionProxmox": true
+  }]
+}`)
+
+	observers, err := LoadObservers(configPath, "https://prod.example.com")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(observers) != 1 {
+		t.Fatalf("len(observers) = %d, want 1", len(observers))
+	}
+	got := observers[0]
+	if got.Name != "dev" || got.URL != "http://192.168.0.10:7655" || got.Token != "observer-secret" || !got.ProvisionProxmox || got.ID == "" {
+		t.Fatalf("observer = %+v", got)
+	}
+}
+
+func TestLoadObserversRejectsPrimaryDuplicate(t *testing.T) {
+	dir := t.TempDir()
+	tokenPath := filepath.Join(dir, "dev.token")
+	configPath := filepath.Join(dir, "observers.json")
+	writePrivateFile(t, tokenPath, "observer-secret")
+	writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"https://pulse.example.com/","tokenFile":"`+tokenPath+`"}]}`)
+
+	if _, err := LoadObservers(configPath, "https://pulse.example.com"); err == nil {
+		t.Fatal("expected duplicate primary URL rejection")
+	}
+}
+
+func TestLoadObserversRejectsInlineOrLooseSecrets(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("POSIX permission check")
+	}
+	dir := t.TempDir()
+	tokenPath := filepath.Join(dir, "dev.token")
+	configPath := filepath.Join(dir, "observers.json")
+	writePrivateFile(t, tokenPath, "observer-secret")
+	if err := os.Chmod(tokenPath, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"https://dev.example.com","tokenFile":"`+tokenPath+`"}]}`)
+
+	if _, err := LoadObservers(configPath, "https://prod.example.com"); err == nil {
+		t.Fatal("expected loose token permission rejection")
+	}
+}
+
+func TestLoadObserversRejectsInlineTokenField(t *testing.T) {
+	dir := t.TempDir()
+	configPath := filepath.Join(dir, "observers.json")
+	writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"https://dev.example.com","token":"inline-secret"}]}`)
+	if _, err := LoadObservers(configPath, "https://prod.example.com"); err == nil {
+		t.Fatal("expected inline token field rejection")
+	}
+}
+
+func TestLoadObserversRejectsSymlinkedTokenFile(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("symlink creation requires platform-specific privileges")
+	}
+	dir := t.TempDir()
+	tokenPath := filepath.Join(dir, "dev.token")
+	linkPath := filepath.Join(dir, "dev-link.token")
+	configPath := filepath.Join(dir, "observers.json")
+	writePrivateFile(t, tokenPath, "observer-secret")
+	if err := os.Symlink(tokenPath, linkPath); err != nil {
+		t.Fatal(err)
+	}
+	writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"https://dev.example.com","tokenFile":"`+linkPath+`"}]}`)
+	if _, err := LoadObservers(configPath, "https://prod.example.com"); err == nil {
+		t.Fatal("expected symlinked token file rejection")
+	}
+}
+
+func TestLoadObserversRequiresExplicitPublicPlaintextOptIn(t *testing.T) {
+	dir := t.TempDir()
+	tokenPath := filepath.Join(dir, "dev.token")
+	configPath := filepath.Join(dir, "observers.json")
+	writePrivateFile(t, tokenPath, "observer-secret")
+	writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"http://203.0.113.10:7655","tokenFile":"`+tokenPath+`"}]}`)
+	if _, err := LoadObservers(configPath, "https://prod.example.com"); err == nil {
+		t.Fatal("expected public plaintext URL rejection")
+	}
+	writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"http://203.0.113.10:7655","tokenFile":"`+tokenPath+`","allowPlaintextHTTP":true}]}`)
+	if _, err := LoadObservers(configPath, "https://prod.example.com"); err != nil {
+		t.Fatalf("explicit plaintext opt-in: %v", err)
+	}
+}
+
+func TestLoadObserversRequiresExplicitPrivatePlaintextOptIn(t *testing.T) {
+	dir := t.TempDir()
+	tokenPath := filepath.Join(dir, "dev.token")
+	configPath := filepath.Join(dir, "observers.json")
+	writePrivateFile(t, tokenPath, "observer-secret")
+	writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"http://192.168.50.10:7655","tokenFile":"`+tokenPath+`"}]}`)
+	if _, err := LoadObservers(configPath, "https://prod.example.com"); err == nil {
+		t.Fatal("expected private-network plaintext URL rejection without observer opt-in")
+	}
+	writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"http://192.168.50.10:7655","tokenFile":"`+tokenPath+`","allowPlaintextHTTP":true}]}`)
+	if _, err := LoadObservers(configPath, "https://prod.example.com"); err != nil {
+		t.Fatalf("explicit private-network plaintext opt-in: %v", err)
+	}
+}
+
+func TestObserverPlaintextPolicyDoesNotInheritPrimaryProcessConsent(t *testing.T) {
+	securityutil.SetOperatorPlaintextHTTPConsent(true)
+	defer securityutil.SetOperatorPlaintextHTTPConsent(false)
+
+	if _, err := NormalizePulseURL("http://203.0.113.10:7655", false, false); err == nil {
+		t.Fatal("observer without explicit opt-in inherited process-wide primary plaintext consent")
+	}
+	if got, err := NormalizePulseURL("http://203.0.113.10:7655", false, true); err != nil {
+		t.Fatalf("explicit observer plaintext opt-in: %v", err)
+	} else if got != "http://203.0.113.10:7655" {
+		t.Fatalf("normalized URL = %q", got)
+	}
+}
+
+func TestLoadObserversDefaultsProxmoxProvisioningOn(t *testing.T) {
+	dir := t.TempDir()
+	tokenPath := filepath.Join(dir, "dev.token")
+	configPath := filepath.Join(dir, "observers.json")
+	writePrivateFile(t, tokenPath, "observer-secret")
+	writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"https://dev.example.com","tokenFile":"`+tokenPath+`"}]}`)
+
+	observers, err := LoadObservers(configPath, "https://prod.example.com")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !observers[0].ProvisionProxmox {
+		t.Fatal("expected Proxmox provisioning to default on")
+	}
+}
@@ -0,0 +1,29 @@
+package agenttarget
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	destinationConfigured = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "pulse_agent_destination_configured",
+		Help: "Configured Pulse report destinations by module and authority role.",
+	}, []string{"module", "destination", "role"})
+	destinationDeliveryUp = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "pulse_agent_destination_delivery_up",
+		Help: "Whether the most recent report delivery to a Pulse destination succeeded.",
+	}, []string{"module", "destination", "role"})
+)
+
+func MarkConfigured(module, destination, role string) {
+	destinationConfigured.WithLabelValues(module, destination, role).Set(1)
+}
+
+func MarkDelivery(module, destination, role string, success bool) {
+	value := 0.0
+	if success {
+		value = 1
+	}
+	destinationDeliveryUp.WithLabelValues(module, destination, role).Set(value)
+}
@@ -15,8 +15,8 @@ import (

 	systemtypes "github.com/moby/moby/api/types/system"
 	"github.com/moby/moby/client"
+	"github.com/rcourtman/pulse-go-rewrite/internal/agenttarget"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agenttls"
-	"github.com/rcourtman/pulse-go-rewrite/internal/securityutil"
 	"github.com/rcourtman/pulse-go-rewrite/internal/utils"
 	agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker"
 	"github.com/rs/zerolog"
@@ -24,11 +24,14 @@ import (

 // TargetConfig describes a single Pulse backend the agent should report to.
 type TargetConfig struct {
+	Name               string
 	URL                string
 	Token              string
 	InsecureSkipVerify bool
+	AllowPlaintextHTTP bool
 	CACertPath         string
 	ServerFingerprint  string
+	Authoritative      bool
 }

 // Config describes runtime configuration for the Docker / Podman collection module.
@@ -124,6 +127,7 @@ type Agent struct {
 	prevContainerCPU   map[string]cpuSample
 	cpuMu              sync.Mutex // protects prevContainerCPU
 	reportBuffer       *utils.Queue[agentsdocker.Report]
+	reportBuffers      map[string]*utils.Queue[agentsdocker.Report]
 	registryChecker    *RegistryChecker // For checking container image updates
 	collectMu          sync.Mutex       // serializes collectOnce calls
 	backgroundMu       sync.Mutex       // protects updateCheckRunning, cleanupTaskRunning
@@ -162,11 +166,13 @@ func New(cfg Config) (*Agent, error) {
 		}

 		targets, err = normalizeTargetsFn([]TargetConfig{{
+			Name:               "primary",
 			URL:                url,
 			Token:              token,
 			InsecureSkipVerify: cfg.InsecureSkipVerify,
 			CACertPath:         cfg.CACertPath,
 			ServerFingerprint:  cfg.ServerFingerprint,
+			Authoritative:      true,
 		}})
 		if err != nil {
 			return nil, fmt.Errorf("dockeragent.New: normalize fallback target: %w", err)
@@ -242,6 +248,11 @@ func New(cfg Config) (*Agent, error) {
 	hasSecure := false
 	hasInsecure := false
 	for _, target := range cfg.Targets {
+		role := "observer"
+		if target.Authoritative {
+			role = "primary"
+		}
+		agenttarget.MarkConfigured("docker", target.Name, role)
 		if target.InsecureSkipVerify {
 			hasInsecure = true
 		} else {
@@ -288,6 +299,20 @@ func New(cfg Config) (*Agent, error) {

 	const bufferCapacity = 60

+	reportBuffers := make(map[string]*utils.Queue[agentsdocker.Report], len(cfg.Targets))
+	for _, target := range cfg.Targets {
+		reportBuffers[target.Name] = utils.New[agentsdocker.Report](bufferCapacity)
+	}
+	primaryBuffer := reportBuffers["primary"]
+	if primaryBuffer == nil {
+		for _, target := range cfg.Targets {
+			if target.Authoritative {
+				primaryBuffer = reportBuffers[target.Name]
+				break
+			}
+		}
+	}
+
 	agent := &Agent{
 		cfg:                cfg,
 		docker:             dockerClient,
@@ -306,7 +331,8 @@ func New(cfg Config) (*Agent, error) {
 		allowedStates:      make(map[string]struct{}, len(stateFilters)),
 		stateFilters:       stateFilters,
 		prevContainerCPU:   make(map[string]cpuSample),
-		reportBuffer:       utils.New[agentsdocker.Report](bufferCapacity),
+		reportBuffer:       primaryBuffer,
+		reportBuffers:      reportBuffers,
 		registryChecker:    newRegistryCheckerWithConfig(*logger, !cfg.DisableUpdateChecks),
 	}

@@ -326,8 +352,10 @@ func normalizeTargets(raw []TargetConfig) ([]TargetConfig, error) {

 	normalized := make([]TargetConfig, 0, len(raw))
 	seen := make(map[string]struct{}, len(raw))
+	seenNames := make(map[string]struct{}, len(raw))

-	for _, target := range raw {
+	authoritativeCount := 0
+	for index, target := range raw {
 		targetURL := strings.TrimSpace(target.URL)
 		token := strings.TrimSpace(target.Token)
 		if targetURL == "" && token == "" {
@@ -341,40 +369,67 @@ func normalizeTargets(raw []TargetConfig) ([]TargetConfig, error) {
 			return nil, fmt.Errorf("pulse target %s is missing API token", targetURL)
 		}

-		normalizedURL, err := normalizeTargetURL(targetURL)
+		if len(normalized) == 0 && !target.Authoritative {
+			target.Authoritative = true
+		}
+		normalizedURL, err := normalizeTargetURLWithPolicy(
+			targetURL,
+			target.Authoritative,
+			target.AllowPlaintextHTTP,
+		)
 		if err != nil {
 			return nil, fmt.Errorf("invalid pulse target URL %q: %w", targetURL, err)
 		}

 		caCertPath := strings.TrimSpace(target.CACertPath)
 		serverFingerprint := strings.TrimSpace(target.ServerFingerprint)
-		key := fmt.Sprintf("%s|%s|%t|%s|%s", normalizedURL, token, target.InsecureSkipVerify, caCertPath, serverFingerprint)
+		name := strings.TrimSpace(target.Name)
+		if name == "" {
+			if len(normalized) == 0 {
+				name = "primary"
+			} else {
+				name = fmt.Sprintf("observer-%d", index)
+			}
+		}
+		if _, exists := seenNames[name]; exists {
+			return nil, fmt.Errorf("duplicate Pulse target name %q", name)
+		}
+		key := normalizedURL
 		if _, exists := seen[key]; exists {
 			continue
 		}
+		if target.Authoritative {
+			authoritativeCount++
+		}
 		seen[key] = struct{}{}
+		seenNames[name] = struct{}{}

 		normalized = append(normalized, TargetConfig{
+			Name:               name,
 			URL:                normalizedURL,
 			Token:              token,
 			InsecureSkipVerify: target.InsecureSkipVerify,
+			AllowPlaintextHTTP: target.AllowPlaintextHTTP,
 			CACertPath:         caCertPath,
 			ServerFingerprint:  serverFingerprint,
+			Authoritative:      target.Authoritative,
 		})
 	}
-
+	if len(normalized) > 0 && authoritativeCount != 1 {
+		return nil, fmt.Errorf("exactly one authoritative Pulse target is required (got %d)", authoritativeCount)
+	}
 	return normalized, nil
 }

 func normalizeTargetURL(raw string) (string, error) {
-	parsed, err := securityutil.NormalizePulseHTTPBaseURLWithOptions(raw, securityutil.PulseURLValidationOptions{
-		AllowLocalNetworkHTTP: true,
-	})
+	return normalizeTargetURLWithPolicy(raw, true, false)
+}
+
+func normalizeTargetURLWithPolicy(raw string, authoritative bool, allowPlaintext bool) (string, error) {
+	normalized, err := agenttarget.NormalizePulseURL(raw, authoritative, allowPlaintext)
 	if err != nil {
 		return "", err
 	}
-
-	normalized := strings.TrimRight(parsed.String(), "/")
 	if normalized == "" {
 		return "", errors.New("URL is empty after normalization")
 	}
@@ -760,59 +815,116 @@ func (a *Agent) collectOnce(ctx context.Context) error {
 		return fmt.Errorf("build docker report: %w", err)
 	}

-	if err := a.sendReport(ctx, report); err != nil {
-		if errors.Is(err, ErrStopRequested) {
-			return nil
-		}
-		a.logger.Warn().
-			Err(err).
-			Int("buffered_reports", a.bufferedReports()).
-			Int("targets", len(a.targets)).
-			Msg("Failed to send docker report, buffering")
-		a.reportBuffer.Push(report)
-		return nil
-	}
-
-	a.flushBuffer(ctx)
-	return nil
+	return a.deliverReport(ctx, report)
 }

 func (a *Agent) flushBuffer(ctx context.Context) {
-	report, ok := a.reportBuffer.Peek()
-	if !ok {
+	a.ensureReportBuffers()
+	for _, target := range a.targets {
+		a.flushTargetBuffer(ctx, target)
+	}
+}
+
+func (a *Agent) deliverReport(ctx context.Context, report agentsdocker.Report) error {
+	a.ensureReportBuffers()
+	payload, err := json.Marshal(report)
+	if err != nil {
+		return fmt.Errorf("marshal report: %w", err)
+	}
+	compressed, err := utils.CompressJSON(payload)
+	if err != nil {
+		return fmt.Errorf("compress report: %w", err)
+	}
+	for _, target := range a.targets {
+		if err := a.sendReportToTarget(ctx, target, compressed, len(report.Containers)); err != nil {
+			agenttarget.MarkDelivery("docker", target.Name, targetRole(target), false)
+			if errors.Is(err, ErrStopRequested) && target.Authoritative {
+				return nil
+			}
+			a.reportBuffers[target.Name].Push(report)
+			a.logger.Warn().Err(err).Str("destination", target.Name).
+				Bool("authoritative", target.Authoritative).
+				Int("buffered_reports", a.reportBuffers[target.Name].Len()).
+				Msg("Failed to send docker report, buffering only for this destination")
+			continue
+		}
+		agenttarget.MarkDelivery("docker", target.Name, targetRole(target), true)
+		a.flushTargetBuffer(ctx, target)
+	}
+	return nil
+}
+
+func targetRole(target TargetConfig) string {
+	if target.Authoritative {
+		return "primary"
+	}
+	return "observer"
+}
+
+func (a *Agent) flushTargetBuffer(ctx context.Context, target TargetConfig) {
+	a.ensureReportBuffers()
+	queue := a.reportBuffers[target.Name]
+	if queue == nil {
 		return
 	}
-
-	a.logger.Info().Int("count", a.reportBuffer.Len()).Msg("Flushing buffered docker reports")
-
 	for {
-		if err := a.sendReport(ctx, report); err != nil {
-			if errors.Is(err, ErrStopRequested) {
-				return
-			}
-			a.logger.Warn().
-				Err(err).
-				Int("remaining_reports", a.bufferedReports()).
-				Msg("Failed to flush buffered docker report, stopping flush")
-			return
-		}
-		a.reportBuffer.Pop()
-
-		report, ok = a.reportBuffer.Peek()
+		report, ok := queue.Peek()
 		if !ok {
 			return
 		}
+		payload, err := json.Marshal(report)
+		if err != nil {
+			queue.Pop()
+			continue
+		}
+		compressed, err := utils.CompressJSON(payload)
+		if err != nil {
+			return
+		}
+		if err := a.sendReportToTarget(ctx, target, compressed, len(report.Containers)); err != nil {
+			return
+		}
+		queue.Pop()
 	}
 }

 func (a *Agent) bufferedReports() int {
-	if a.reportBuffer == nil {
-		return 0
+	a.ensureReportBuffers()
+	total := 0
+	for _, queue := range a.reportBuffers {
+		if queue != nil {
+			total += queue.Len()
+		}
+	}
+	return total
+}
+
+func (a *Agent) ensureReportBuffers() {
+	if a.reportBuffers != nil {
+		return
+	}
+	a.reportBuffers = make(map[string]*utils.Queue[agentsdocker.Report], len(a.targets))
+	for index := range a.targets {
+		if strings.TrimSpace(a.targets[index].Name) == "" {
+			if index == 0 {
+				a.targets[index].Name = "primary"
+			} else {
+				a.targets[index].Name = fmt.Sprintf("observer-%d", index)
+			}
+		}
+		if index == 0 {
+			a.targets[index].Authoritative = true
+		}
+		queue := utils.New[agentsdocker.Report](60)
+		if index == 0 && a.reportBuffer != nil {
+			queue = a.reportBuffer
+		}
+		a.reportBuffers[a.targets[index].Name] = queue
 	}
-	return a.reportBuffer.Len()
 }

 func (a *Agent) sendReport(ctx context.Context, report agentsdocker.Report) error {
+	a.ensureReportBuffers()
 	payload, err := json.Marshal(report)
 	if err != nil {
 		return fmt.Errorf("marshal report: %w", err)
@@ -917,6 +1029,9 @@ func (a *Agent) sendReportToTarget(ctx context.Context, target TargetConfig, pay
 	if len(body) == 0 {
 		return nil
 	}
+	if !target.Authoritative && target.Name != "" {
+		return nil
+	}

 	var reportResp agentsdocker.ReportResponse
 	if err := json.Unmarshal(body, &reportResp); err != nil {
@@ -1171,10 +1286,15 @@ func (a *Agent) sendCommandAckWithPayload(ctx context.Context, target TargetConf
 }

 func (a *Agent) primaryTarget() TargetConfig {
-	if len(a.targets) == 0 {
-		return TargetConfig{}
+	for _, target := range a.targets {
+		if target.Authoritative {
+			return target
+		}
 	}
-	return a.targets[0]
+	if len(a.targets) > 0 {
+		return a.targets[0]
+	}
+	return TargetConfig{}
 }

 func (a *Agent) httpClientFor(target TargetConfig) *http.Client {
@@ -0,0 +1,71 @@
+package dockeragent
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/utils"
+	agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker"
+	"github.com/rs/zerolog"
+)
+
+func TestDeliverReportBuffersOnlyFailedDockerDestination(t *testing.T) {
+	primary := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }))
+	defer primary.Close()
+	observer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusBadGateway) }))
+	defer observer.Close()
+
+	targets := []TargetConfig{
+		{Name: "primary", URL: primary.URL, Token: "p", Authoritative: true},
+		{Name: "dev", URL: observer.URL, Token: "o"},
+	}
+	a := &Agent{
+		logger: zerolog.Nop(), targets: targets,
+		httpClients:        map[bool]*http.Client{false: http.DefaultClient},
+		trustedHTTPClients: map[string]*http.Client{},
+		reportBuffer:       utils.New[agentsdocker.Report](10),
+		reportBuffers: map[string]*utils.Queue[agentsdocker.Report]{
+			"primary": utils.New[agentsdocker.Report](10),
+			"dev":     utils.New[agentsdocker.Report](10),
+		},
+	}
+	a.reportBuffer = a.reportBuffers["primary"]
+	if err := a.deliverReport(context.Background(), agentsdocker.Report{}); err != nil {
+		t.Fatalf("deliver report: %v", err)
+	}
+	if a.reportBuffers["primary"].Len() != 0 || a.reportBuffers["dev"].Len() != 1 {
+		t.Fatalf("buffer depths primary=%d observer=%d", a.reportBuffers["primary"].Len(), a.reportBuffers["dev"].Len())
+	}
+}
+
+func TestDockerObserverCannotIssueCommands(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"commands":[{"id":"cmd1","type":"stop"}]}`))
+	}))
+	defer server.Close()
+	a := &Agent{logger: zerolog.Nop(), httpClients: map[bool]*http.Client{false: server.Client()}, trustedHTTPClients: map[string]*http.Client{}}
+	if err := a.sendReportToTarget(context.Background(), TargetConfig{Name: "dev", URL: server.URL, Token: "o"}, []byte(`{}`), 0); err != nil {
+		t.Fatalf("observer response must be acknowledgement-only: %v", err)
+	}
+}
+
+func TestDockerObserverPlaintextPolicyIsDestinationScoped(t *testing.T) {
+	targets := []TargetConfig{
+		{Name: "primary", URL: "https://primary.example.test", Token: "p", Authoritative: true},
+		{Name: "observer", URL: "http://203.0.113.10:7655", Token: "o"},
+	}
+	if _, err := normalizeTargets(targets); err == nil {
+		t.Fatal("expected observer plaintext URL rejection without destination opt-in")
+	}
+	targets[1].AllowPlaintextHTTP = true
+	normalized, err := normalizeTargets(targets)
+	if err != nil {
+		t.Fatalf("explicit observer plaintext opt-in: %v", err)
+	}
+	if !normalized[1].AllowPlaintextHTTP {
+		t.Fatal("observer plaintext policy was not preserved after normalization")
+	}
+}
@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"net/netip"
@@ -19,6 +20,7 @@ import (
 	"sync"
 	"time"

+	"github.com/rcourtman/pulse-go-rewrite/internal/agenttarget"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agenttls"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agentupdate"
 	"github.com/rcourtman/pulse-go-rewrite/internal/platformsupport"
@@ -44,6 +46,7 @@ type Config struct {
 	InsecureSkipVerify bool
 	CACertPath         string
 	ServerFingerprint  string
+	Observers          []ObserverTarget
 	RunOnce            bool
 	LogLevel           zerolog.Level
 	Logger             *zerolog.Logger
@@ -92,6 +95,28 @@ type Config struct {
 	storageCleanup       *storageCleanupManager
 }

+// ObserverTarget is a report-only Pulse destination. It can receive the same
+// collected host payload and an independently provisioned Proxmox source, but
+// it never supplies remote config, commands, enrollment, or update authority.
+type ObserverTarget struct {
+	Name               string
+	ID                 string
+	PulseURL           string
+	APIToken           string
+	InsecureSkipVerify bool
+	AllowPlaintextHTTP bool
+	CACertPath         string
+	ServerFingerprint  string
+	ProvisionProxmox   bool
+}
+
+type observerReporter struct {
+	target             ObserverTarget
+	httpClient         *http.Client
+	reportBuffer       *utils.Queue[agentshost.Report]
+	lastAuthFailureLog time.Time
+}
+
 // Agent is responsible for collecting host metrics and shipping them to Pulse.
 type Agent struct {
 	cfg        Config
@@ -117,6 +142,7 @@ type Agent struct {
 	configMu               sync.RWMutex
 	remoteConfigChanged    chan struct{}
 	reportBuffer           *utils.Queue[agentshost.Report]
+	observerReporters      []*observerReporter
 	commandClient          *CommandClient
 	commandClientMu        sync.Mutex
 	commandClientRunCancel context.CancelFunc
@@ -270,23 +296,18 @@ func New(cfg Config) (*Agent, error) {
 	if arch == "" {
 		arch = runtime.GOARCH
 	}
-	tlsConfig, err := agenttls.NewClientTLSConfig(cfg.CACertPath, cfg.InsecureSkipVerify, cfg.ServerFingerprint)
+	client, err := newAgentHTTPClient(cfg.CACertPath, cfg.InsecureSkipVerify, cfg.ServerFingerprint)
 	if err != nil {
 		return nil, fmt.Errorf("invalid TLS configuration: %w", err)
 	}

-	client := &http.Client{
-		Timeout: 15 * time.Second,
-		Transport: &http.Transport{
-			Proxy:           http.ProxyFromEnvironment,
-			TLSClientConfig: tlsConfig,
-		},
-		// Disallow redirects for agent API calls. If a reverse proxy redirects
-		// HTTP to HTTPS, Go's default behavior converts POST to GET (per HTTP spec),
-		// causing 405 errors. Return an error with guidance instead.
-		CheckRedirect: func(req *http.Request, via []*http.Request) error {
-			return fmt.Errorf("server returned redirect to %s - if using a reverse proxy, ensure you use the correct protocol (https:// instead of http://) in your --url flag", req.URL)
-		},
+	observerReporters, err := newObserverReporters(cfg.Observers)
+	if err != nil {
+		return nil, err
+	}
+	agenttarget.MarkConfigured("host", "primary", "primary")
+	for _, observer := range observerReporters {
+		agenttarget.MarkConfigured("host", observer.target.Name, "observer")
 	}

 	trimmedTags := make([]string, 0, len(cfg.Tags))
@@ -349,6 +370,7 @@ func New(cfg Config) (*Agent, error) {
 		trimmedPulseURL:     pulseURL,
 		remoteConfigChanged: make(chan struct{}, 1),
 		reportBuffer:        utils.New[agentshost.Report](bufferCapacity),
+		observerReporters:   observerReporters,
 		collector:           collector,
 		newCommandClient:    newCommandClientFn,
 		runCommandClient:    runCommandClientFn,
@@ -367,6 +389,58 @@ func New(cfg Config) (*Agent, error) {
 	return agent, nil
 }

+func newAgentHTTPClient(caCertPath string, insecureSkipVerify bool, serverFingerprint string) (*http.Client, error) {
+	tlsConfig, err := agenttls.NewClientTLSConfig(caCertPath, insecureSkipVerify, serverFingerprint)
+	if err != nil {
+		return nil, err
+	}
+	return &http.Client{
+		Timeout: 15 * time.Second,
+		Transport: &http.Transport{
+			Proxy:           http.ProxyFromEnvironment,
+			TLSClientConfig: tlsConfig,
+		},
+		// Redirects can rewrite report POSTs to GETs and must never cross an
+		// authority boundary implicitly.
+		CheckRedirect: func(req *http.Request, _ []*http.Request) error {
+			return fmt.Errorf("server returned redirect to %s - use the final Pulse URL explicitly", req.URL)
+		},
+	}, nil
+}
+
+func newObserverReporters(targets []ObserverTarget) ([]*observerReporter, error) {
+	const bufferCapacity = 60
+	result := make([]*observerReporter, 0, len(targets))
+	seen := make(map[string]struct{}, len(targets))
+	for _, target := range targets {
+		target.Name = strings.TrimSpace(target.Name)
+		target.ID = strings.TrimSpace(target.ID)
+		target.APIToken = strings.TrimSpace(target.APIToken)
+		if target.Name == "" || target.ID == "" || target.APIToken == "" {
+			return nil, errors.New("observer destination requires name, id, and API token")
+		}
+		url, err := agenttarget.NormalizePulseURL(target.PulseURL, false, target.AllowPlaintextHTTP)
+		if err != nil {
+			return nil, fmt.Errorf("invalid observer %q Pulse URL: %w", target.Name, err)
+		}
+		target.PulseURL = url
+		if _, exists := seen[target.ID]; exists {
+			return nil, fmt.Errorf("duplicate observer destination id for %q", target.Name)
+		}
+		seen[target.ID] = struct{}{}
+		client, err := newAgentHTTPClient(target.CACertPath, target.InsecureSkipVerify, target.ServerFingerprint)
+		if err != nil {
+			return nil, fmt.Errorf("configure observer %q TLS: %w", target.Name, err)
+		}
+		result = append(result, &observerReporter{
+			target:       target,
+			httpClient:   client,
+			reportBuffer: utils.New[agentshost.Report](bufferCapacity),
+		})
+	}
+	return result, nil
+}
+
 func normalizeProxmoxType(raw string) (string, error) {
 	value := strings.TrimSpace(strings.ToLower(raw))
 	switch value {
@@ -442,10 +516,12 @@ func (a *Agent) Run(ctx context.Context) error {

 	// Load any reports buffered from a previous shutdown
 	a.loadPersistedBuffer()
+	a.loadPersistedObserverBuffers()

 	ticker := time.NewTicker(a.currentInterval())
 	defer ticker.Stop()
 	defer a.persistBuffer()
+	defer a.persistObserverBuffers()

 	if err := a.process(ctx); err != nil && !errors.Is(err, context.Canceled) {
 		a.logger.Error().
@@ -619,7 +695,16 @@ func (a *Agent) process(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("build report: %w", err)
 	}
+	primaryErr := a.deliverPrimaryReport(ctx, report)
+	for _, observer := range a.observerReporters {
+		a.deliverObserverReport(ctx, observer, report)
+	}
+	return primaryErr
+}
+
+func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Report) error {
 	if err := a.sendReport(ctx, report); err != nil {
+		agenttarget.MarkDelivery("host", "primary", "primary", false)
 		var statusErr *reportHTTPStatusError
 		if errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusForbidden {
 			a.logger.Error().
@@ -652,6 +737,7 @@ func (a *Agent) process(ctx context.Context) error {
 		event.Msg("Failed to send report, buffering")
 		return nil
 	}
+	agenttarget.MarkDelivery("host", "primary", "primary", true)

 	// A successful report means the token is accepted again; reset the auth
 	// failure throttle so a later rejection is reported promptly.
@@ -667,6 +753,56 @@ func (a *Agent) process(ctx context.Context) error {
 	return nil
 }

+func (a *Agent) deliverObserverReport(ctx context.Context, observer *observerReporter, report agentshost.Report) {
+	if err := a.sendReportToDestination(ctx, report, observer.target.PulseURL, observer.target.APIToken, observer.httpClient, false); err != nil {
+		agenttarget.MarkDelivery("host", observer.target.Name, "observer", false)
+		var statusErr *reportHTTPStatusError
+		if errors.As(err, &statusErr) && (statusErr.StatusCode == http.StatusUnauthorized || statusErr.StatusCode == http.StatusForbidden) {
+			if time.Since(observer.lastAuthFailureLog) >= authFailureLogInterval {
+				observer.lastAuthFailureLog = time.Now()
+				a.logger.Error().Err(err).
+					Str("destination", observer.target.Name).
+					Str("role", "observer").
+					Str("pulse_url", observer.target.PulseURL).
+					Msg("Pulse observer rejected its API token; reports for this destination are dropped until the token is replaced")
+			}
+			return
+		}
+		observer.reportBuffer.Push(report)
+		a.logger.Warn().Err(err).
+			Str("destination", observer.target.Name).
+			Str("role", "observer").
+			Int("buffered_reports", observer.reportBuffer.Len()).
+			Msg("Failed to send observer report, buffering only for this destination")
+		return
+	}
+
+	agenttarget.MarkDelivery("host", observer.target.Name, "observer", true)
+	observer.lastAuthFailureLog = time.Time{}
+	a.flushObserverBuffer(ctx, observer)
+}
+
+func (a *Agent) flushObserverBuffer(ctx context.Context, observer *observerReporter) {
+	for !observer.reportBuffer.IsEmpty() {
+		report, ok := observer.reportBuffer.Peek()
+		if !ok {
+			return
+		}
+		if err := a.sendReportToDestination(ctx, report, observer.target.PulseURL, observer.target.APIToken, observer.httpClient, false); err != nil {
+			var statusErr *reportHTTPStatusError
+			if errors.As(err, &statusErr) && (statusErr.StatusCode == http.StatusUnauthorized || statusErr.StatusCode == http.StatusForbidden) {
+				for {
+					if _, ok := observer.reportBuffer.Pop(); !ok {
+						break
+					}
+				}
+			}
+			return
+		}
+		observer.reportBuffer.Pop()
+	}
+}
+
 // logAuthFailure emits an actionable, throttled error when the server rejects
 // the agent's API token with 401. It is only called from the single-threaded
 // report loop, so lastAuthFailureLog needs no additional synchronisation.
@@ -739,11 +875,25 @@ func (a *Agent) flushBuffer(ctx context.Context) {

 const bufferFileName = "report-buffer.json"

+func observerBufferFileName(id string) string {
+	return "report-buffer-observer-" + id + ".json"
+}
+
 // persistBuffer writes buffered reports to disk on shutdown so they can be
 // retransmitted on the next startup. Uses atomic write (tmp + rename) to
 // prevent corruption if the process is killed mid-write.
 func (a *Agent) persistBuffer() {
-	items := a.reportBuffer.Items()
+	a.persistReportQueue(a.reportBuffer, bufferFileName, "primary")
+}
+
+func (a *Agent) persistObserverBuffers() {
+	for _, observer := range a.observerReporters {
+		a.persistReportQueue(observer.reportBuffer, observerBufferFileName(observer.target.ID), observer.target.Name)
+	}
+}
+
+func (a *Agent) persistReportQueue(queue *utils.Queue[agentshost.Report], fileName, destination string) {
+	items := queue.Items()
 	if len(items) == 0 {
 		return
 	}
@@ -764,7 +914,7 @@ func (a *Agent) persistBuffer() {
 		return
 	}

-	path := filepath.Join(a.stateDir, bufferFileName)
+	path := filepath.Join(a.stateDir, fileName)
 	tmpPath := path + ".tmp"
 	if err := os.WriteFile(tmpPath, data, 0600); err != nil {
 		a.logger.Warn().Err(err).Str("path", tmpPath).Msg("Failed to write report buffer temp file")
@@ -776,18 +926,28 @@ func (a *Agent) persistBuffer() {
 		return
 	}

-	a.logger.Info().Int("count", len(items)).Str("path", path).Msg("Persisted report buffer to disk")
+	a.logger.Info().Int("count", len(items)).Str("path", path).Str("destination", destination).Msg("Persisted report buffer to disk")
 }

 // loadPersistedBuffer loads buffered reports from a previous shutdown and
 // attempts to flush them. The file is deleted after loading regardless of
 // whether the flush succeeds (items are pushed back into the in-memory buffer).
 func (a *Agent) loadPersistedBuffer() {
+	a.loadPersistedReportQueue(a.reportBuffer, bufferFileName, "primary")
+}
+
+func (a *Agent) loadPersistedObserverBuffers() {
+	for _, observer := range a.observerReporters {
+		a.loadPersistedReportQueue(observer.reportBuffer, observerBufferFileName(observer.target.ID), observer.target.Name)
+	}
+}
+
+func (a *Agent) loadPersistedReportQueue(queue *utils.Queue[agentshost.Report], fileName, destination string) {
 	if a.stateDir == "" {
 		return
 	}

-	path := filepath.Join(a.stateDir, bufferFileName)
+	path := filepath.Join(a.stateDir, fileName)
 	data, err := os.ReadFile(path)
 	if err != nil {
 		if !os.IsNotExist(err) {
@@ -810,10 +970,10 @@ func (a *Agent) loadPersistedBuffer() {
 	}

 	for _, item := range items {
-		a.reportBuffer.Push(item)
+		queue.Push(item)
 	}

-	a.logger.Info().Int("count", len(items)).Msg("Loaded persisted report buffer from disk")
+	a.logger.Info().Int("count", len(items)).Str("destination", destination).Msg("Loaded persisted report buffer from disk")
 }

 func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) {
@@ -960,6 +1120,10 @@ func (a *Agent) currentStorageCleanupStatus(ctx context.Context) *agentshost.Sto
 }

 func (a *Agent) sendReport(ctx context.Context, report agentshost.Report) error {
+	return a.sendReportToDestination(ctx, report, a.trimmedPulseURL, a.cfg.APIToken, a.httpClient, true)
+}
+
+func (a *Agent) sendReportToDestination(ctx context.Context, report agentshost.Report, pulseURL, token string, client *http.Client, authoritative bool) error {
 	payload, err := json.Marshal(report)
 	if err != nil {
 		return fmt.Errorf("marshal report: %w", err)
@@ -971,7 +1135,7 @@ func (a *Agent) sendReport(ctx context.Context, report agentshost.Report) error
 	}

 	endpoint := agentReportEndpoint
-	url := fmt.Sprintf("%s%s", a.trimmedPulseURL, endpoint)
+	url := fmt.Sprintf("%s%s", strings.TrimRight(pulseURL, "/"), endpoint)
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(compressed))
 	if err != nil {
 		return fmt.Errorf("create request: %w", err)
@@ -979,13 +1143,13 @@ func (a *Agent) sendReport(ctx context.Context, report agentshost.Report) error

 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Content-Encoding", "gzip")
-	if token := strings.TrimSpace(a.cfg.APIToken); token != "" {
+	if token := strings.TrimSpace(token); token != "" {
 		req.Header.Set("Authorization", "Bearer "+token)
 		req.Header.Set("X-API-Token", token)
 	}
 	req.Header.Set("User-Agent", "pulse-agent/"+Version)

-	resp, err := a.httpClient.Do(req)
+	resp, err := client.Do(req)
 	if err != nil {
 		return fmt.Errorf("send request: %w", err)
 	}
@@ -1002,6 +1166,10 @@ func (a *Agent) sendReport(ctx context.Context, report agentshost.Report) error
 			StatusCode: resp.StatusCode,
 		}
 	}
+	if !authoritative {
+		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64*1024))
+		return nil
+	}

 	// Parse response to check for server-side config overrides
 	var reportResp struct {
@@ -1795,50 +1963,56 @@ func (a *Agent) collectSMARTData(ctx context.Context, diskExclude []string) []ag
 // Supports hosts with multiple Proxmox products (e.g., PVE + PBS on same host).
 func (a *Agent) runProxmoxSetup(ctx context.Context) {
 	a.logger.Info().Msg("Proxmox mode enabled, checking setup...")
-
-	setup := NewProxmoxSetup(
-		a.logger,
-		a.httpClient,
-		a.collector,
-		a.trimmedPulseURL,
-		a.cfg.APIToken,
-		a.cfg.ProxmoxType,
-		a.hostname,
-		a.currentReportIP(),
-		a.stateDir,
-		a.cfg.InsecureSkipVerify,
-	)
-
-	// Use RunAll to detect and register all Proxmox products on this host
-	results, err := setup.RunAll(ctx)
-	if err != nil {
-		a.logger.Error().Err(err).Msg("Proxmox setup failed")
-		return
-	}
-
-	if len(results) == 0 {
-		// All types already registered
-		a.logger.Info().Msg("All detected Proxmox products already registered")
-		return
-	}
-
-	// Log results for each registered type
-	for _, result := range results {
-		if result.Registered {
-			a.logger.Info().
-				Str("type", result.ProxmoxType).
-				Str("host", result.NodeHost).
-				Str("token_id", result.TokenID).
-				Msg("Proxmox node registered successfully")
-		} else {
-			a.logger.Warn().
-				Str("type", result.ProxmoxType).
-				Str("host", result.NodeHost).
-				Msg("Proxmox token created but registration failed (node may need manual configuration)")
+	for _, destination := range a.proxmoxDestinations() {
+		results, err := destination.setup.RunAll(ctx)
+		if err != nil {
+			a.logger.Error().Err(err).Str("destination", destination.name).Msg("Proxmox setup failed")
+			continue
+		}
+		for _, result := range results {
+			if result.Registered {
+				a.logger.Info().Str("destination", destination.name).
+					Str("type", result.ProxmoxType).
+					Str("host", result.NodeHost).
+					Str("token_id", result.TokenID).
+					Msg("Proxmox node registered successfully")
+			} else {
+				a.logger.Warn().Str("destination", destination.name).
+					Str("type", result.ProxmoxType).
+					Str("host", result.NodeHost).
+					Msg("Proxmox token created but registration failed")
+			}
 		}
 	}
 }

+type proxmoxDestination struct {
+	name  string
+	setup *ProxmoxSetup
+}
+
+func (a *Agent) proxmoxDestinations() []proxmoxDestination {
+	destinations := []proxmoxDestination{{
+		name: "primary",
+		setup: NewProxmoxSetup(a.logger, a.httpClient, a.collector, a.trimmedPulseURL, a.cfg.APIToken,
+			a.cfg.ProxmoxType, a.hostname, a.currentReportIP(), a.stateDir, a.cfg.InsecureSkipVerify),
+	}}
+	for _, observer := range a.observerReporters {
+		if !observer.target.ProvisionProxmox {
+			continue
+		}
+		stateDir := filepath.Join(a.stateDir, "observers", observer.target.ID)
+		tokenName := "pulse-" + proxmoxTokenScope(a.hostname, a.trimmedPulseURL) + "-" + observer.target.ID
+		setup := NewProxmoxSetup(a.logger, observer.httpClient, a.collector, observer.target.PulseURL,
+			observer.target.APIToken, a.cfg.ProxmoxType, a.hostname, a.currentReportIP(), stateDir,
+			observer.target.InsecureSkipVerify).
+			WithObserverPlaintextPolicy(observer.target.AllowPlaintextHTTP).
+			WithMonitorTokenName(tokenName)
+		destinations = append(destinations, proxmoxDestination{name: observer.target.Name, setup: setup})
+	}
+	return destinations
+}
+
 const (
 	proxmoxHealthCheckInitialDelay = 2 * time.Minute
 	proxmoxHealthCheckInterval     = 5 * time.Minute
@@ -1865,34 +2039,22 @@ func (a *Agent) runProxmoxHealthCheckLoop(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			setup := NewProxmoxSetup(
-				a.logger,
-				a.httpClient,
-				a.collector,
-				a.trimmedPulseURL,
-				a.cfg.APIToken,
-				a.cfg.ProxmoxType,
-				a.hostname,
-				a.currentReportIP(),
-				a.stateDir,
-				a.cfg.InsecureSkipVerify,
-			)
-			results, err := setup.RunHealthCheck(ctx)
-			if err != nil {
-				a.logger.Warn().Err(err).Msg("Proxmox health check failed")
-				continue
-			}
-			for _, result := range results {
-				if result.Registered {
-					a.logger.Info().
+			for _, destination := range a.proxmoxDestinations() {
+				results, err := destination.setup.RunHealthCheck(ctx)
+				if err != nil {
+					a.logger.Warn().Err(err).Str("destination", destination.name).Msg("Proxmox health check failed")
+					continue
+				}
+				for _, result := range results {
+					event := a.logger.Info()
+					if !result.Registered {
+						event = a.logger.Warn()
+					}
+					event.Str("destination", destination.name).
 						Str("type", result.ProxmoxType).
 						Str("host", result.NodeHost).
-						Msg("Proxmox node re-registered via health check")
-				} else {
-					a.logger.Warn().
-						Str("type", result.ProxmoxType).
-						Str("host", result.NodeHost).
-						Msg("Proxmox health check: token rotated but registration failed")
+						Bool("registered", result.Registered).
+						Msg("Proxmox destination health repair completed")
 				}
 			}
 		}
@@ -0,0 +1,94 @@
+package hostagent
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/utils"
+	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
+	"github.com/rs/zerolog"
+)
+
+func TestObserverDeliveryIsIsolatedAndReportOnly(t *testing.T) {
+	primaryRequests := atomic.Int32{}
+	primary := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		primaryRequests.Add(1)
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"success":true,"config":{"commandsEnabled":false}}`))
+	}))
+	defer primary.Close()
+
+	observerRequests := atomic.Int32{}
+	observerHealthy := atomic.Bool{}
+	observer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		observerRequests.Add(1)
+		if !observerHealthy.Load() {
+			w.WriteHeader(http.StatusBadGateway)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"success":true,"config":{"commandsEnabled":true}}`))
+	}))
+	defer observer.Close()
+
+	logger := zerolog.Nop()
+	a := &Agent{
+		cfg:                 Config{APIToken: "primary-token"},
+		logger:              logger,
+		httpClient:          primary.Client(),
+		trimmedPulseURL:     primary.URL,
+		reportBuffer:        utils.New[agentshost.Report](10),
+		remoteConfigChanged: make(chan struct{}, 1),
+		observerReporters: []*observerReporter{{
+			target:     ObserverTarget{Name: "dev", ID: "dev12345", PulseURL: observer.URL, APIToken: "observer-token"},
+			httpClient: observer.Client(), reportBuffer: utils.New[agentshost.Report](10),
+		}},
+	}
+	report := agentshost.Report{Host: agentshost.HostInfo{Hostname: "node"}}
+
+	if err := a.deliverPrimaryReport(context.Background(), report); err != nil {
+		t.Fatalf("primary delivery: %v", err)
+	}
+	a.deliverObserverReport(context.Background(), a.observerReporters[0], report)
+	if got := a.reportBuffer.Len(); got != 0 {
+		t.Fatalf("primary buffer depth = %d, want 0", got)
+	}
+	if got := a.observerReporters[0].reportBuffer.Len(); got != 1 {
+		t.Fatalf("observer buffer depth = %d, want 1", got)
+	}
+
+	observerHealthy.Store(true)
+	a.deliverObserverReport(context.Background(), a.observerReporters[0], report)
+	if got := a.observerReporters[0].reportBuffer.Len(); got != 0 {
+		t.Fatalf("observer buffer depth after recovery = %d, want 0", got)
+	}
+	if a.cfg.EnableCommands {
+		t.Fatal("observer response changed authoritative command configuration")
+	}
+	if primaryRequests.Load() != 1 || observerRequests.Load() != 3 {
+		t.Fatalf("requests primary=%d observer=%d, want 1 and 3", primaryRequests.Load(), observerRequests.Load())
+	}
+}
+
+func TestObserverReporterPlaintextPolicyIsDestinationScoped(t *testing.T) {
+	target := ObserverTarget{
+		Name:     "observer",
+		ID:       "observer-id",
+		PulseURL: "http://203.0.113.10:7655",
+		APIToken: "observer-token",
+	}
+	if _, err := newObserverReporters([]ObserverTarget{target}); err == nil {
+		t.Fatal("expected observer plaintext URL rejection without destination opt-in")
+	}
+	target.AllowPlaintextHTTP = true
+	reporters, err := newObserverReporters([]ObserverTarget{target})
+	if err != nil {
+		t.Fatalf("explicit observer plaintext opt-in: %v", err)
+	}
+	if len(reporters) != 1 || !reporters[0].target.AllowPlaintextHTTP {
+		t.Fatalf("observer plaintext policy was not preserved: %+v", reporters)
+	}
+}
@@ -16,6 +16,7 @@ import (
 	"strings"
 	"time"

+	"github.com/rcourtman/pulse-go-rewrite/internal/agenttarget"
 	"github.com/rs/zerolog"
 )

@@ -32,6 +33,9 @@ type ProxmoxSetup struct {
 	collector          SystemCollector
 	stateDir           string          // directory for registration state files
 	retryBackoffs      []time.Duration // overridable for testing; nil uses defaults
+	monitorToken       string          // optional destination-scoped token name
+	authoritative      bool            // primary destinations retain local-network HTTP compatibility
+	allowPlaintextHTTP bool            // explicit observer plaintext transport policy
 }

 // ProxmoxSetupResult contains the result of a successful Proxmox setup.
@@ -200,6 +204,9 @@ func proxmoxTokenScope(candidates ...string) string {
 }

 func (p *ProxmoxSetup) monitorTokenName() string {
+	if tokenName := strings.TrimSpace(p.monitorToken); tokenName != "" {
+		return tokenName
+	}
 	return "pulse-" + proxmoxTokenScope(p.hostname, p.pulseURL)
 }

@@ -347,15 +354,33 @@ func NewProxmoxSetup(logger zerolog.Logger, httpClient *http.Client, collector S
 		reportIP:           reportIP,
 		stateDir:           stateDir,
 		insecureSkipVerify: insecure,
+		authoritative:      true,
 	}
 }

+// WithObserverPlaintextPolicy marks this setup as an observer transport. Unlike
+// the primary, every non-loopback plaintext observer requires its own explicit
+// opt-in and cannot inherit process-wide primary transport consent.
+func (p *ProxmoxSetup) WithObserverPlaintextPolicy(allowPlaintext bool) *ProxmoxSetup {
+	p.authoritative = false
+	p.allowPlaintextHTTP = allowPlaintext
+	return p
+}
+
+// WithMonitorTokenName isolates Proxmox credentials for an additional Pulse
+// destination. The primary destination deliberately keeps the legacy token
+// name for upgrade compatibility.
+func (p *ProxmoxSetup) WithMonitorTokenName(tokenName string) *ProxmoxSetup {
+	p.monitorToken = strings.TrimSpace(tokenName)
+	return p
+}
+
 // Run executes the Proxmox setup process:
 // 1. Detects Proxmox type (if not specified)
 // 2. Creates the monitoring user and API token
 // 3. Registers the node with Pulse via auto-register
 func (p *ProxmoxSetup) Run(ctx context.Context) (*ProxmoxSetupResult, error) {
-	pulseURL, err := normalizePulseURL(p.pulseURL)
+	pulseURL, err := agenttarget.NormalizePulseURL(p.pulseURL, p.authoritative, p.allowPlaintextHTTP)
 	if err != nil {
 		return nil, fmt.Errorf("invalid pulse URL: %w", err)
 	}
@@ -377,53 +402,11 @@ func (p *ProxmoxSetup) Run(ctx context.Context) (*ProxmoxSetupResult, error) {
 		p.logger.Info().Str("type", string(ptype)).Msg("Auto-detected Proxmox type")
 	}

-	hostURL := p.getHostURL(ctx, ptype)
-
-	// Check if already registered (idempotency)
-	if p.isAlreadyRegistered() {
-		registered, err := p.checkRegistrationWithPulse(ctx, ptype, hostURL)
-		if err != nil {
-			p.logger.Warn().Err(err).Msg("Failed to verify Proxmox registration state with Pulse; keeping local marker behavior")
-			return nil, nil
-		}
-		if registered {
-			p.logger.Info().Msg("Proxmox node already registered, skipping setup")
-			return nil, nil
-		}
-		p.logger.Info().Str("type", string(ptype)).Str("host", hostURL).Msg("Local Proxmox registration marker exists but Pulse has no matching node; re-registering")
-	}
-
-	// Create monitoring user and token
-	tokenID, tokenValue, err := p.setupToken(ctx, ptype)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create Proxmox API token: %w", err)
-	}
-
-	p.logger.Info().Str("token_id", tokenID).Msg("Created Proxmox API token")
-
-	// Register with Pulse
-	registered := false
-	registerResp, err := p.registerWithPulse(ctx, ptype, hostURL, tokenID, tokenValue)
-	if err != nil {
-		p.logger.Warn().Err(err).Msg("Failed to register with Pulse (node may already exist)")
-	} else {
-		registered = true
+	result, err := p.runForType(ctx, ptype)
+	if result != nil && result.Registered {
 		p.markAsRegistered()
-		p.markTypeAsRegistered(ptype)
-		if strings.TrimSpace(registerResp.Host) != "" {
-			hostURL = registerResp.Host
-		}
-		p.logger.Info().Str("host", hostURL).Str("node", registerResp.NodeName).Msg("Successfully registered Proxmox node with Pulse")
 	}
-
-	return &ProxmoxSetupResult{
-		ProxmoxType: string(ptype),
-		TokenID:     tokenID,
-		TokenValue:  tokenValue,
-		NodeHost:    hostURL,
-		NodeName:    registerResp.NodeName,
-		Registered:  registered,
-	}, nil
+	return result, err
 }

 // RunAll detects and registers ALL Proxmox products on this system.
@@ -431,7 +414,7 @@ func (p *ProxmoxSetup) Run(ctx context.Context) (*ProxmoxSetupResult, error) {
 // supported configuration). Each type gets its own registration and state tracking.
 // Returns results for all types that were processed (skipping already-registered ones).
 func (p *ProxmoxSetup) RunAll(ctx context.Context) ([]*ProxmoxSetupResult, error) {
-	pulseURL, err := normalizePulseURL(p.pulseURL)
+	pulseURL, err := agenttarget.NormalizePulseURL(p.pulseURL, p.authoritative, p.allowPlaintextHTTP)
 	if err != nil {
 		return nil, fmt.Errorf("invalid pulse URL: %w", err)
 	}
@@ -490,7 +473,7 @@ func (p *ProxmoxSetup) RunAll(ctx context.Context) ([]*ProxmoxSetupResult, error
 // health-check rotation, which would cause uncontrolled token churn if Pulse
 // is temporarily unreachable.
 func (p *ProxmoxSetup) RunHealthCheck(ctx context.Context) ([]*ProxmoxSetupResult, error) {
-	pulseURL, err := normalizePulseURL(p.pulseURL)
+	pulseURL, err := agenttarget.NormalizePulseURL(p.pulseURL, p.authoritative, p.allowPlaintextHTTP)
 	if err != nil {
 		return nil, fmt.Errorf("invalid pulse URL: %w", err)
 	}
@@ -552,20 +535,26 @@ func (p *ProxmoxSetup) runForType(ctx context.Context, ptype proxmoxProductType)
 	ptype = proxmoxProductType(normalizedTypeStr)
 	hostURL := p.getHostURL(ctx, ptype)

-	// Check if this type is already registered
+	// Always establish destination reachability and registration state before
+	// mutating a Proxmox token. This prevents a missing local marker or a Pulse
+	// startup outage from rotating credentials that a healthy Pulse instance is
+	// still using.
+	registered, err := p.checkRegistrationWithPulse(ctx, ptype, hostURL)
+	if err != nil {
+		p.logger.Warn().
+			Err(err).
+			Str("type", string(ptype)).
+			Msg("Failed to verify Proxmox registration state with Pulse; leaving credentials unchanged")
+		return nil, nil
+	}
+	if registered {
+		if !p.isTypeRegistered(ptype) {
+			p.markTypeAsRegistered(ptype)
+		}
+		p.logger.Info().Str("type", string(ptype)).Msg("Proxmox type already registered, skipping")
+		return nil, nil
+	}
 	if p.isTypeRegistered(ptype) {
-		registered, err := p.checkRegistrationWithPulse(ctx, ptype, hostURL)
-		if err != nil {
-			p.logger.Warn().
-				Err(err).
-				Str("type", string(ptype)).
-				Msg("Failed to verify Proxmox registration state with Pulse; keeping local marker behavior")
-			return nil, nil
-		}
-		if registered {
-			p.logger.Info().Str("type", string(ptype)).Msg("Proxmox type already registered, skipping")
-			return nil, nil
-		}
 		p.logger.Info().
 			Str("type", string(ptype)).
 			Str("host", hostURL).
@@ -588,7 +577,7 @@ func (p *ProxmoxSetup) runForType(ctx context.Context, ptype proxmoxProductType)
 	p.logger.Info().Str("type", string(ptype)).Str("token_id", tokenID).Msg("Created Proxmox API token")

 	// Register with Pulse
-	registered := false
+	registered = false
 	registerResp, err := p.registerWithPulse(ctx, ptype, hostURL, tokenID, tokenValue)
 	if err != nil {
 		p.logger.Warn().Err(err).Str("type", string(ptype)).Msg("Failed to register with Pulse (node may already exist)")
@@ -560,6 +560,7 @@ func TestGetIPThatReachesPulse_IPv6Target(t *testing.T) {
 func TestProxmoxSetup_RunForType(t *testing.T) {
 	mc := &mockCollector{}
 	var expectedTokenID string
+	registrationExists := true
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		switch r.URL.Path {
 		case "/api/setup-script-url":
@@ -573,7 +574,7 @@ func TestProxmoxSetup_RunForType(t *testing.T) {
 			}
 			if check, _ := payload["checkRegistration"].(bool); check {
 				w.Header().Set("Content-Type", "application/json")
-				_, _ = w.Write([]byte(`{"registered":true}`))
+				_, _ = w.Write([]byte(fmt.Sprintf(`{"registered":%t}`, registrationExists)))
 				return
 			}
 			w.Header().Set("Content-Type", "application/json")
@@ -601,6 +602,7 @@ func TestProxmoxSetup_RunForType(t *testing.T) {
 	})

 	t.Run("performs registration successfully", func(t *testing.T) {
+		registrationExists = false
 		mc.statFn = func(name string) (os.FileInfo, error) { return nil, os.ErrNotExist }
 		mc.commandCombinedOutputFn = func(ctx context.Context, name string, arg ...string) (string, error) {
 			// pveum user token add pulse-monitor@pve ... --privsep 1
@@ -653,6 +655,44 @@ func TestProxmoxSetup_MonitorTokenNameIsNodeScoped(t *testing.T) {
 	}
 }

+func TestProxmoxSetup_DoesNotMutateTokenBeforeDestinationCheck(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer server.Close()
+
+	mutations := 0
+	mc := &mockCollector{
+		statFn: func(string) (os.FileInfo, error) { return nil, os.ErrNotExist },
+		dialTimeoutFn: func(string, string, time.Duration) (net.Conn, error) {
+			return &mockConn{localAddr: &net.UDPAddr{IP: net.ParseIP("10.0.0.1")}}, nil
+		},
+		commandCombinedOutputFn: func(_ context.Context, name string, _ ...string) (string, error) {
+			if name == "pveum" || name == "proxmox-backup-manager" {
+				mutations++
+			}
+			return "", nil
+		},
+	}
+	p := NewProxmoxSetup(zerolog.Nop(), server.Client(), mc, server.URL, "token", "pve", "node", "", t.TempDir(), false)
+	p.retryBackoffs = []time.Duration{}
+	result, err := p.runForType(context.Background(), proxmoxProductPVE)
+	if err != nil {
+		t.Fatalf("runForType: %v", err)
+	}
+	if result != nil {
+		t.Fatalf("result = %+v, want nil while destination is unavailable", result)
+	}
+	if mutations != 0 {
+		t.Fatalf("Proxmox command mutations = %d, want 0", mutations)
+	}
+
+	p.WithMonitorTokenName("pulse-node-observer-deadbeef")
+	if got := p.monitorTokenName(); got != "pulse-node-observer-deadbeef" {
+		t.Fatalf("observer token name = %q", got)
+	}
+}
+
 func TestProxmoxSetup_SetupPVETokenUsesPrivilegeSeparatedTokenACLs(t *testing.T) {
 	mc := &mockCollector{}
 	calls := make([]commandCall, 0)
@@ -967,8 +1007,8 @@ func TestProxmoxSetup_RunAll(t *testing.T) {
 		}

 		results, _ := p.RunAll(context.Background())
-		if len(results) != 1 || results[0].ProxmoxType != "pbs" {
-			t.Errorf("expected pbs result")
+		if len(results) != 0 {
+			t.Errorf("expected no token mutation while Pulse registration state is unreachable")
 		}
 	})

@@ -1437,8 +1477,8 @@ func TestProxmoxSetup_Run_TopLevel(t *testing.T) {
 		if gotPayload.AuthToken != "setup-token-optional" {
 			t.Fatalf("authToken = %q, want %q", gotPayload.AuthToken, "setup-token-optional")
 		}
-		if requestCount != 2 {
-			t.Fatalf("requestCount = %d, want 2", requestCount)
+		if requestCount != 4 {
+			t.Fatalf("requestCount = %d, want 4 (reachability check plus registration)", requestCount)
 		}
 	})

@@ -1456,6 +1496,27 @@ func TestProxmoxSetup_Run_TopLevel(t *testing.T) {
 	})
 }

+func TestObserverProxmoxSetupRequiresDestinationPlaintextOptIn(t *testing.T) {
+	setup := NewProxmoxSetup(
+		zerolog.Nop(),
+		http.DefaultClient,
+		nil,
+		"http://203.0.113.10:7655",
+		"observer-token",
+		"pve",
+		"node",
+		"",
+		t.TempDir(),
+		false,
+	).WithObserverPlaintextPolicy(false)
+
+	if _, err := setup.Run(context.Background()); err == nil {
+		t.Fatal("expected observer Proxmox setup to reject plaintext without destination opt-in")
+	} else if !strings.Contains(err.Error(), "explicitly allows plaintext HTTP") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
 // Helpers

 type mockConn struct {
@@ -19,6 +19,7 @@ import (
 	"time"

 	"github.com/IGLOU-EU/go-wildcard/v2"
+	"github.com/rcourtman/pulse-go-rewrite/internal/agenttarget"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agenttls"
 	"github.com/rcourtman/pulse-go-rewrite/internal/securityutil"
 	"github.com/rcourtman/pulse-go-rewrite/internal/utils"
@@ -57,6 +58,7 @@ type Config struct {
 	InsecureSkipVerify bool
 	CACertPath         string
 	ServerFingerprint  string
+	Targets            []TargetConfig
 	LogLevel           zerolog.Level
 	Logger             *zerolog.Logger

@@ -72,6 +74,25 @@ type Config struct {
 	MaxPods               int  // Max pods included in the report
 }

+// TargetConfig describes one Pulse report destination. Exactly one target is
+// authoritative; additional targets are report-only observers.
+type TargetConfig struct {
+	Name               string
+	URL                string
+	Token              string
+	InsecureSkipVerify bool
+	AllowPlaintextHTTP bool
+	CACertPath         string
+	ServerFingerprint  string
+	Authoritative      bool
+}
+
+type reportTarget struct {
+	config TargetConfig
+	client *http.Client
+	buffer *utils.Queue[agentsk8s.Report]
+}
+
 // Agent collects and reports Kubernetes cluster state to Pulse.
 // It periodically gathers pod, deployment, and node metrics, then sends
 // them to the configured Pulse URL. The agent handles authentication,
@@ -100,6 +121,7 @@ type Agent struct {
 	excludeNamespaces []string

 	reportBuffer *utils.Queue[agentsk8s.Report]
+	targets      []*reportTarget
 }

 const (
@@ -147,19 +169,32 @@ func New(cfg Config) (*Agent, error) {
 		logger = &scoped
 	}

-	if strings.TrimSpace(cfg.APIToken) == "" {
-		return nil, fmt.Errorf("api token is required")
-	}
-
-	pulseURL := strings.TrimSpace(cfg.PulseURL)
-	if pulseURL == "" {
-		pulseURL = "http://localhost:7655"
-	}
-	pulseURL, err := normalizePulseURL(pulseURL)
+	legacyTarget := len(cfg.Targets) == 0
+	targetConfigs, err := normalizeKubernetesTargets(cfg)
 	if err != nil {
-		return nil, fmt.Errorf("invalid pulse URL: %w", err)
+		if legacyTarget {
+			if strings.TrimSpace(cfg.APIToken) == "" {
+				return nil, fmt.Errorf("api token is required")
+			}
+			return nil, fmt.Errorf("invalid pulse URL: %w", err)
+		}
+		return nil, err
 	}
-	cfg.PulseURL = pulseURL
+	cfg.Targets = targetConfigs
+	primary := targetConfigs[0]
+	for _, target := range targetConfigs {
+		role := "observer"
+		if target.Authoritative {
+			role = "primary"
+		}
+		agenttarget.MarkConfigured("kubernetes", target.Name, role)
+		if target.Authoritative {
+			primary = target
+		}
+	}
+	cfg.PulseURL = primary.URL
+	cfg.APIToken = primary.Token
+	pulseURL := primary.URL

 	restCfg, contextName, err := buildRESTConfig(cfg.KubeconfigPath, cfg.KubeContext)
 	if err != nil {
@@ -183,22 +218,23 @@ func New(cfg Config) (*Agent, error) {
 		agentVersion = Version
 	}

-	tlsConfig, err := agenttls.NewClientTLSConfig(cfg.CACertPath, cfg.InsecureSkipVerify, cfg.ServerFingerprint)
-	if err != nil {
-		return nil, fmt.Errorf("configure Pulse TLS client: %w", err)
-	}
-	httpClient := &http.Client{
-		Timeout: 15 * time.Second,
-		Transport: &http.Transport{
-			Proxy:           http.ProxyFromEnvironment,
-			TLSClientConfig: tlsConfig,
-		},
-		// Disallow redirects for agent API calls. If a reverse proxy redirects
-		// HTTP to HTTPS, Go's default behavior converts POST to GET (per HTTP spec),
-		// causing 405 errors. Return an error with guidance instead.
-		CheckRedirect: func(req *http.Request, via []*http.Request) error {
-			return fmt.Errorf("server returned redirect to %s - if using a reverse proxy, ensure you use the correct protocol (https:// instead of http://) in your --url flag", req.URL)
-		},
+	reportTargets := make([]*reportTarget, 0, len(targetConfigs))
+	var httpClient *http.Client
+	var primaryBuffer *utils.Queue[agentsk8s.Report]
+	for _, target := range targetConfigs {
+		client, err := newKubernetesHTTPClient(target)
+		if err != nil {
+			if legacyTarget {
+				return nil, fmt.Errorf("configure Pulse TLS client: %w", err)
+			}
+			return nil, fmt.Errorf("configure Pulse target %q: %w", target.Name, err)
+		}
+		buffer := utils.New[agentsk8s.Report](60)
+		reportTargets = append(reportTargets, &reportTarget{config: target, client: client, buffer: buffer})
+		if target.Authoritative {
+			httpClient = client
+			primaryBuffer = buffer
+		}
 	}

 	clusterServer := strings.TrimSpace(restCfg.Host)
@@ -228,7 +264,8 @@ func New(cfg Config) (*Agent, error) {
 		clusterContext:    clusterContext,
 		includeNamespaces: cfg.IncludeNamespaces,
 		excludeNamespaces: cfg.ExcludeNamespaces,
-		reportBuffer:      utils.New[agentsk8s.Report](60),
+		reportBuffer:      primaryBuffer,
+		targets:           reportTargets,
 	}

 	if err := agent.discoverClusterMetadata(context.Background()); err != nil {
@@ -256,6 +293,85 @@ func normalizePulseURL(rawURL string) (string, error) {
 	return parsed.String(), nil
 }

+func normalizeKubernetesTargets(cfg Config) ([]TargetConfig, error) {
+	raw := append([]TargetConfig(nil), cfg.Targets...)
+	if len(raw) == 0 {
+		url := strings.TrimSpace(cfg.PulseURL)
+		if url == "" {
+			url = "http://localhost:7655"
+		}
+		raw = []TargetConfig{{
+			Name: "primary", URL: url, Token: cfg.APIToken, InsecureSkipVerify: cfg.InsecureSkipVerify,
+			CACertPath: cfg.CACertPath, ServerFingerprint: cfg.ServerFingerprint, Authoritative: true,
+		}}
+	}
+
+	result := make([]TargetConfig, 0, len(raw))
+	seenURLs := make(map[string]struct{}, len(raw))
+	seenNames := make(map[string]struct{}, len(raw))
+	authoritative := 0
+	for index, target := range raw {
+		name := strings.TrimSpace(target.Name)
+		if name == "" {
+			if index == 0 {
+				name = "primary"
+			} else {
+				name = fmt.Sprintf("observer-%d", index)
+			}
+		}
+		if _, exists := seenNames[name]; exists {
+			return nil, fmt.Errorf("duplicate Pulse target name %q", name)
+		}
+		if index == 0 && !target.Authoritative {
+			target.Authoritative = true
+		}
+		url, err := agenttarget.NormalizePulseURL(
+			strings.TrimSpace(target.URL),
+			target.Authoritative,
+			target.AllowPlaintextHTTP,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("invalid Pulse target %q URL: %w", name, err)
+		}
+		if _, exists := seenURLs[url]; exists {
+			return nil, fmt.Errorf("duplicate Pulse target URL %q", url)
+		}
+		token := strings.TrimSpace(target.Token)
+		if token == "" {
+			return nil, fmt.Errorf("Pulse target %q API token is required", name)
+		}
+		if target.Authoritative {
+			authoritative++
+		}
+		target.Name = name
+		target.URL = url
+		target.Token = token
+		target.CACertPath = strings.TrimSpace(target.CACertPath)
+		target.ServerFingerprint = strings.TrimSpace(target.ServerFingerprint)
+		seenNames[name] = struct{}{}
+		seenURLs[url] = struct{}{}
+		result = append(result, target)
+	}
+	if authoritative != 1 {
+		return nil, fmt.Errorf("exactly one authoritative Pulse target is required (got %d)", authoritative)
+	}
+	return result, nil
+}
+
+func newKubernetesHTTPClient(target TargetConfig) (*http.Client, error) {
+	tlsConfig, err := agenttls.NewClientTLSConfig(target.CACertPath, target.InsecureSkipVerify, target.ServerFingerprint)
+	if err != nil {
+		return nil, err
+	}
+	return &http.Client{
+		Timeout:   15 * time.Second,
+		Transport: &http.Transport{Proxy: http.ProxyFromEnvironment, TLSClientConfig: tlsConfig},
+		CheckRedirect: func(req *http.Request, _ []*http.Request) error {
+			return fmt.Errorf("server returned redirect to %s - use the final Pulse URL explicitly", req.URL)
+		},
+	}, nil
+}
+
 func buildRESTConfig(kubeconfigPath, kubeContext string) (*rest.Config, string, error) {
 	kubeconfigPath = strings.TrimSpace(kubeconfigPath)
 	kubeContext = strings.TrimSpace(kubeContext)
@@ -335,8 +451,16 @@ func (a *Agent) discoverClusterMetadata(ctx context.Context) error {
 }

 func (a *Agent) closeIdleConnections() {
-	if a.httpClient != nil {
-		a.httpClient.CloseIdleConnections()
+	if len(a.targets) == 0 {
+		if a.httpClient != nil {
+			a.httpClient.CloseIdleConnections()
+		}
+		return
+	}
+	for _, target := range a.targets {
+		if target.client != nil {
+			target.client.CloseIdleConnections()
+		}
 	}
 }

@@ -359,10 +483,20 @@ func (a *Agent) Run(ctx context.Context) error {
 }

 func (a *Agent) bufferedReportCount() int {
-	if a == nil || a.reportBuffer == nil {
+	if a == nil {
 		return 0
 	}
-	return a.reportBuffer.Len()
+	if len(a.targets) == 0 {
+		if a.reportBuffer == nil {
+			return 0
+		}
+		return a.reportBuffer.Len()
+	}
+	total := 0
+	for _, target := range a.targets {
+		total += target.buffer.Len()
+	}
+	return total
 }

 func (a *Agent) runOnce(ctx context.Context) {
@@ -379,59 +513,75 @@ func (a *Agent) runOnce(ctx context.Context) {
 		return
 	}

-	if err := a.sendReport(ctx, report); err != nil {
-		a.logger.Warn().
-			Err(err).
-			Str("phase", "send_report").
-			Str("cluster_id", a.clusterID).
-			Str("agent_id", a.agentID).
-			Int("report_nodes", len(report.Nodes)).
-			Int("report_pods", len(report.Pods)).
-			Int("report_deployments", len(report.Deployments)).
-			Int("buffer_depth_before", a.bufferedReportCount()).
-			Msg("Failed to send Kubernetes report, buffering")
-		a.reportBuffer.Push(report)
-		a.logger.Debug().
-			Str("phase", "buffer_report").
-			Str("cluster_id", a.clusterID).
-			Int("buffer_depth_after", a.bufferedReportCount()).
-			Msg("Buffered Kubernetes report for retry")
+	for _, target := range a.targets {
+		if err := a.sendReportToTarget(ctx, report, target); err != nil {
+			agenttarget.MarkDelivery("kubernetes", target.config.Name, kubernetesTargetRole(target.config), false)
+			target.buffer.Push(report)
+			a.logger.Warn().Err(err).
+				Str("destination", target.config.Name).
+				Bool("authoritative", target.config.Authoritative).
+				Int("buffer_depth", target.buffer.Len()).
+				Msg("Failed to send Kubernetes report, buffering only for this destination")
+			continue
+		}
+		agenttarget.MarkDelivery("kubernetes", target.config.Name, kubernetesTargetRole(target.config), true)
+		a.flushTargetReports(ctx, target)
+	}
+	if len(a.targets) == 0 {
+		if err := a.sendReport(ctx, report); err != nil {
+			a.reportBuffer.Push(report)
+		}
 	}
 }

+func kubernetesTargetRole(target TargetConfig) string {
+	if target.Authoritative {
+		return "primary"
+	}
+	return "observer"
+}
+
 func (a *Agent) flushReports(ctx context.Context) {
-	flushed := 0
-	for {
-		report, ok := a.reportBuffer.Peek()
-		if !ok {
-			if flushed > 0 {
-				a.logger.Debug().
-					Str("phase", "flush_buffered_reports").
-					Str("cluster_id", a.clusterID).
-					Str("agent_id", a.agentID).
-					Int("flushed_reports", flushed).
-					Int("buffer_depth_remaining", a.bufferedReportCount()).
-					Msg("Flushed buffered Kubernetes reports")
+	if len(a.targets) == 0 {
+		for {
+			report, ok := a.reportBuffer.Peek()
+			if !ok || a.sendReport(ctx, report) != nil {
+				return
 			}
+			a.reportBuffer.Pop()
+		}
+	}
+	for _, target := range a.targets {
+		a.flushTargetReports(ctx, target)
+	}
+}
+
+func (a *Agent) flushTargetReports(ctx context.Context, target *reportTarget) {
+	for {
+		report, ok := target.buffer.Peek()
+		if !ok {
 			return
 		}
-		if err := a.sendReport(ctx, report); err != nil {
+		if err := a.sendReportToTarget(ctx, report, target); err != nil {
+			agenttarget.MarkDelivery("kubernetes", target.config.Name, kubernetesTargetRole(target.config), false)
 			a.logger.Warn().
 				Err(err).
+				Str("destination", target.config.Name).
 				Str("phase", "flush_buffered_report").
 				Str("cluster_id", a.clusterID).
 				Str("agent_id", a.agentID).
 				Int("report_nodes", len(report.Nodes)).
 				Int("report_pods", len(report.Pods)).
 				Int("report_deployments", len(report.Deployments)).
-				Int("buffer_depth", a.bufferedReportCount()).
+				Int("buffer_depth", target.buffer.Len()).
 				Msg("Failed to flush buffered Kubernetes report")
 			return
 		}
-		if _, ok := a.reportBuffer.Pop(); !ok {
+		if _, ok := target.buffer.Pop(); !ok {
 			a.logger.Debug().Msg("Failed to remove buffered report after successful send")
 			return
 		}
+		agenttarget.MarkDelivery("kubernetes", target.config.Name, kubernetesTargetRole(target.config), true)
 	}
 }

@@ -3476,6 +3626,19 @@ func isProblemDeployment(dep appsv1.Deployment) bool {
 }

 func (a *Agent) sendReport(ctx context.Context, report agentsk8s.Report) (retErr error) {
+	for _, target := range a.targets {
+		if target.config.Authoritative {
+			return a.sendReportToTarget(ctx, report, target)
+		}
+	}
+	// Preserve direct-construction test and legacy behavior.
+	return a.sendReportToTarget(ctx, report, &reportTarget{
+		config: TargetConfig{URL: a.pulseURL, Token: a.cfg.APIToken, Authoritative: true},
+		client: a.httpClient,
+	})
+}
+
+func (a *Agent) sendReportToTarget(ctx context.Context, report agentsk8s.Report, target *reportTarget) (retErr error) {
 	payload, err := json.Marshal(report)
 	if err != nil {
 		return fmt.Errorf("marshal report: %w", err)
@@ -3486,7 +3649,7 @@ func (a *Agent) sendReport(ctx context.Context, report agentsk8s.Report) (retErr
 		return fmt.Errorf("compress report: %w", err)
 	}

-	reportURL := fmt.Sprintf("%s/api/agents/kubernetes/report", a.pulseURL)
+	reportURL := fmt.Sprintf("%s/api/agents/kubernetes/report", target.config.URL)
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, reportURL, bytes.NewReader(compressed))
 	if err != nil {
 		return fmt.Errorf("create request for %s: %w", reportURL, err)
@@ -3494,11 +3657,15 @@ func (a *Agent) sendReport(ctx context.Context, report agentsk8s.Report) (retErr

 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Content-Encoding", "gzip")
-	req.Header.Set("Authorization", "Bearer "+a.cfg.APIToken)
-	req.Header.Set("X-API-Token", a.cfg.APIToken)
+	req.Header.Set("Authorization", "Bearer "+target.config.Token)
+	req.Header.Set("X-API-Token", target.config.Token)
 	req.Header.Set("User-Agent", reportUserAgent+a.agentVersion)

-	resp, err := a.httpClient.Do(req)
+	client := target.client
+	if client == nil {
+		client = a.httpClient
+	}
+	resp, err := client.Do(req)
 	if err != nil {
 		return fmt.Errorf("send request to %s: %w", reportURL, err)
 	}
@@ -0,0 +1,52 @@
+package kubernetesagent
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/utils"
+	agentsk8s "github.com/rcourtman/pulse-go-rewrite/pkg/agents/kubernetes"
+	"github.com/rs/zerolog"
+)
+
+func TestKubernetesDestinationBuffersAreIndependent(t *testing.T) {
+	primary := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }))
+	defer primary.Close()
+	observer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusServiceUnavailable) }))
+	defer observer.Close()
+
+	primaryTarget := &reportTarget{config: TargetConfig{Name: "primary", URL: primary.URL, Token: "p", Authoritative: true}, client: primary.Client(), buffer: utils.New[agentsk8s.Report](10)}
+	observerTarget := &reportTarget{config: TargetConfig{Name: "dev", URL: observer.URL, Token: "o"}, client: observer.Client(), buffer: utils.New[agentsk8s.Report](10)}
+	a := &Agent{logger: zerolog.Nop(), agentVersion: "test", targets: []*reportTarget{primaryTarget, observerTarget}, reportBuffer: primaryTarget.buffer}
+	report := agentsk8s.Report{}
+	for _, target := range a.targets {
+		if err := a.sendReportToTarget(context.Background(), report, target); err != nil {
+			target.buffer.Push(report)
+		}
+	}
+	if primaryTarget.buffer.Len() != 0 || observerTarget.buffer.Len() != 1 {
+		t.Fatalf("buffer depths primary=%d observer=%d", primaryTarget.buffer.Len(), observerTarget.buffer.Len())
+	}
+}
+
+func TestKubernetesObserverPlaintextPolicyIsDestinationScoped(t *testing.T) {
+	cfg := Config{
+		Targets: []TargetConfig{
+			{Name: "primary", URL: "https://primary.example.test", Token: "p", Authoritative: true},
+			{Name: "observer", URL: "http://203.0.113.10:7655", Token: "o"},
+		},
+	}
+	if _, err := normalizeKubernetesTargets(cfg); err == nil {
+		t.Fatal("expected observer plaintext URL rejection without destination opt-in")
+	}
+	cfg.Targets[1].AllowPlaintextHTTP = true
+	normalized, err := normalizeKubernetesTargets(cfg)
+	if err != nil {
+		t.Fatalf("explicit observer plaintext opt-in: %v", err)
+	}
+	if !normalized[1].AllowPlaintextHTTP {
+		t.Fatal("observer plaintext policy was not preserved after normalization")
+	}
+}
@@ -20,6 +20,7 @@ param (
    [bool]$Uninstall = $false,
    [string]$CACertPath = $env:PULSE_CACERT,
    [string]$ServerFingerprint = $env:PULSE_SERVER_FINGERPRINT,
+    [string]$ObserversFile = $env:PULSE_OBSERVERS_FILE,
    [string]$AgentId = $env:PULSE_AGENT_ID,
    [string]$Hostname = $env:PULSE_HOSTNAME,
    [string]$TokenFile = $env:PULSE_TOKEN_FILE,
@@ -436,6 +437,9 @@ function Save-ConnectionState {
    if (-not [string]::IsNullOrWhiteSpace($ServerFingerprint)) {
        $lines += "PULSE_SERVER_FINGERPRINT='$ServerFingerprint'"
    }
+    if (-not [string]::IsNullOrWhiteSpace($ObserversFile)) {
+        $lines += "PULSE_OBSERVERS_FILE='$ObserversFile'"
+    }

    New-Item -ItemType Directory -Path $StateDir -Force | Out-Null
    Set-Content -Path $ConnectionStatePath -Value ($lines -join "`n") -Encoding UTF8
@@ -614,6 +618,18 @@ if ([string]::IsNullOrWhiteSpace($Token) -and -not [string]::IsNullOrWhiteSpace(
    }
 }

+if (-not [string]::IsNullOrWhiteSpace($ObserversFile)) {
+    try {
+        $ObserversFile = $ExecutionContext.SessionState.Path.GetUnresolvedProviderPathFromPSPath($ObserversFile)
+        if (-not [System.IO.Path]::IsPathRooted($ObserversFile) -or -not (Test-Path $ObserversFile -PathType Leaf)) {
+            throw "path must be absolute and identify an existing file"
+        }
+    } catch {
+        Show-Error "Invalid observer config file.`nProvided: $ObserversFile`nError: $_"
+        Exit 1
+    }
+}
+
 # --- Input Validation ---
 Write-Host "Validating parameters..." -ForegroundColor Cyan

@@ -936,6 +952,7 @@ if ($EnableCommands) { $ServiceArgs += "--enable-commands" }
 if ($Insecure) { $ServiceArgs += "--insecure" }
 if (-not [string]::IsNullOrWhiteSpace($CACertPath)) { $ServiceArgs += @("--cacert", "`"$CACertPath`"") }
 if (-not [string]::IsNullOrWhiteSpace($ServerFingerprint)) { $ServiceArgs += @("--server-fingerprint", "`"$ServerFingerprint`"") }
+if (-not [string]::IsNullOrWhiteSpace($ObserversFile)) { $ServiceArgs += @("--observers-file", "`"$ObserversFile`"") }
 if (-not [string]::IsNullOrWhiteSpace($AgentId)) { $ServiceArgs += @("--agent-id", "`"$AgentId`"") }
 if (-not [string]::IsNullOrWhiteSpace($Hostname)) { $ServiceArgs += @("--hostname", "`"$Hostname`"") }
 $ServiceArgs += @("--log-file", "`"$LogFile`"")
@@ -23,6 +23,7 @@
 #   --disk-exclude <pattern>  Exclude mount points matching pattern (repeatable)
 #   --insecure          Skip TLS certificate verification
 #   --server-fingerprint <sha256> Pin the Pulse server leaf certificate
+#   --observers-file <path> Report to additional observer Pulse instances
 #   --enable-commands   Enable Pulse command execution on agent (disabled by default; required for Patrol actions and Proxmox LXC Docker inventory)
 #   --health-addr <addr> Health/metrics listener address (default: 127.0.0.1:9191, use "" to disable)
 #   --update            Update an existing agent using saved connection state
@@ -79,6 +80,7 @@ UPDATE_ONLY="false"
 UNINSTALL="false"
 INSECURE="false"
 SERVER_FINGERPRINT="${PULSE_SERVER_FINGERPRINT:-}"
+OBSERVERS_FILE="${PULSE_OBSERVERS_FILE:-}"
 AGENT_ID=""
 HOSTNAME_OVERRIDE=""
 ENABLE_COMMANDS="false"
@@ -304,6 +306,7 @@ Options:
  --insecure              Skip TLS verification (auto-enabled for http:// URLs)
  --cacert <path>         Custom CA certificate for TLS (used by curl and agent)
  --server-fingerprint <sha256> Pin the Pulse server leaf certificate for agent connections
+  --observers-file <path> Absolute path to private JSON config for report-only observer Pulse destinations
  --enable-commands       Enable Pulse command execution (disabled by default; required for Patrol actions and Proxmox LXC Docker inventory)
  --health-addr <addr>    Health/metrics listener address (default: 127.0.0.1:9191; use "" to disable)
  --enroll                Exchange bootstrap token for runtime token (deploy wizard)
@@ -1451,6 +1454,7 @@ build_exec_arg_items() {
    if [[ -n "$PROXMOX_TYPE" ]]; then EXEC_ARG_ITEMS+=(--proxmox-type "$PROXMOX_TYPE"); fi
    if [[ "$INSECURE" == "true" ]]; then EXEC_ARG_ITEMS+=(--insecure); fi
    if [[ -n "$SERVER_FINGERPRINT" ]]; then EXEC_ARG_ITEMS+=(--server-fingerprint "$SERVER_FINGERPRINT"); fi
+    if [[ -n "$OBSERVERS_FILE" ]]; then EXEC_ARG_ITEMS+=(--observers-file "$OBSERVERS_FILE"); fi
    if [[ "$ENABLE_COMMANDS" == "true" ]]; then EXEC_ARG_ITEMS+=(--enable-commands); fi
    if [[ "$HEALTH_ADDR_SET" == "true" ]]; then EXEC_ARG_ITEMS+=(--health-addr "$HEALTH_ADDR"); fi
    if [[ "$ENROLL" == "true" ]]; then EXEC_ARG_ITEMS+=(--enroll); fi
@@ -1718,6 +1722,10 @@ apply_recovered_agent_arg_value() {
            if [[ -z "$SERVER_FINGERPRINT" ]]; then SERVER_FINGERPRINT="$value"; fi
            RECOVERED_AGENT_ARG_STATE="true"
            ;;
+        observers-file)
+            if [[ -z "$OBSERVERS_FILE" ]]; then OBSERVERS_FILE="$value"; fi
+            RECOVERED_AGENT_ARG_STATE="true"
+            ;;
        health-addr)
            if [[ "$HEALTH_ADDR_SET" != "true" ]]; then
                HEALTH_ADDR="$value"
@@ -1772,10 +1780,10 @@ recover_connection_state_from_arg_stream() {
        fi

        case "$arg" in
-            --url|--pulse-url|--token|--token-file|--interval|--agent-id|--hostname|--cacert|--server-fingerprint|--health-addr|--state-dir|--kubeconfig|--proxmox-type|--disk-exclude|-url|-pulse-url|-token|-token-file|-interval|-agent-id|-hostname|-cacert|-server-fingerprint|-health-addr|-state-dir|-kubeconfig|-proxmox-type|-disk-exclude)
+            --url|--pulse-url|--token|--token-file|--interval|--agent-id|--hostname|--cacert|--server-fingerprint|--observers-file|--health-addr|--state-dir|--kubeconfig|--proxmox-type|--disk-exclude|-url|-pulse-url|-token|-token-file|-interval|-agent-id|-hostname|-cacert|-server-fingerprint|-observers-file|-health-addr|-state-dir|-kubeconfig|-proxmox-type|-disk-exclude)
                pending_key=$(normalize_recovered_agent_arg_key "$arg")
                ;;
-            --url=*|--pulse-url=*|--token=*|--token-file=*|--interval=*|--agent-id=*|--hostname=*|--cacert=*|--server-fingerprint=*|--health-addr=*|--state-dir=*|--kubeconfig=*|--proxmox-type=*|--disk-exclude=*|-url=*|-pulse-url=*|-token=*|-token-file=*|-interval=*|-agent-id=*|-hostname=*|-cacert=*|-server-fingerprint=*|-health-addr=*|-state-dir=*|-kubeconfig=*|-proxmox-type=*|-disk-exclude=*)
+            --url=*|--pulse-url=*|--token=*|--token-file=*|--interval=*|--agent-id=*|--hostname=*|--cacert=*|--server-fingerprint=*|--observers-file=*|--health-addr=*|--state-dir=*|--kubeconfig=*|--proxmox-type=*|--disk-exclude=*|-url=*|-pulse-url=*|-token=*|-token-file=*|-interval=*|-agent-id=*|-hostname=*|-cacert=*|-server-fingerprint=*|-observers-file=*|-health-addr=*|-state-dir=*|-kubeconfig=*|-proxmox-type=*|-disk-exclude=*)
                key="${arg%%=*}"
                value="${arg#*=}"
                apply_recovered_agent_arg_value "$key" "$value"
@@ -2288,6 +2296,7 @@ while [[ $# -gt 0 ]]; do
        --insecure) INSECURE="true"; shift ;;
        --cacert) CURL_CA_BUNDLE="$2"; shift 2 ;;
        --server-fingerprint) SERVER_FINGERPRINT="$2"; shift 2 ;;
+        --observers-file) OBSERVERS_FILE="$2"; shift 2 ;;
        --enable-commands) ENABLE_COMMANDS="true"; shift ;;
        --health-addr) HEALTH_ADDR="$2"; HEALTH_ADDR_SET="true"; shift 2 ;;
        --enroll) ENROLL="true"; shift ;;
@@ -2308,6 +2317,15 @@ while [[ $# -gt 0 ]]; do
    esac
 done

+if [[ -n "$OBSERVERS_FILE" ]]; then
+    if [[ "$OBSERVERS_FILE" != /* ]]; then
+        fail "Observer config path must be absolute: ${OBSERVERS_FILE}" "$EXIT_MISSING_ARGS"
+    fi
+    if [[ ! -f "$OBSERVERS_FILE" || -L "$OBSERVERS_FILE" ]]; then
+        fail "Observer config must be a regular non-symlink file: ${OBSERVERS_FILE}" "$EXIT_MISSING_ARGS"
+    fi
+fi
+
 # Read token from file if --token-file was provided
 if [[ -n "$TOKEN_FILE_PATH" ]]; then
    if [[ ! -f "$TOKEN_FILE_PATH" ]]; then
@@ -136,6 +136,27 @@ func TestInstallPS1PersistsAndVerifiesServerFingerprint(t *testing.T) {
 	}
 }

+func TestInstallPS1PreservesObserverDestinationConfig(t *testing.T) {
+	content, err := os.ReadFile(repoFile("scripts", "install.ps1"))
+	if err != nil {
+		t.Fatalf("read install.ps1: %v", err)
+	}
+
+	script := string(content)
+	required := []string{
+		`[string]$ObserversFile = $env:PULSE_OBSERVERS_FILE,`,
+		`$lines += "PULSE_OBSERVERS_FILE='$ObserversFile'"`,
+		`[System.IO.Path]::IsPathRooted($ObserversFile)`,
+		`Test-Path $ObserversFile -PathType Leaf`,
+		`$ServiceArgs += @("--observers-file", "` + "`" + `"$ObserversFile` + "`" + `"")`,
+	}
+	for _, needle := range required {
+		if !strings.Contains(script, needle) {
+			t.Fatalf("install.ps1 missing observer-config lifecycle contract: %s", needle)
+		}
+	}
+}
+
 func TestInstallPS1AllowsMissingTokenForOptionalAuth(t *testing.T) {
 	content, err := os.ReadFile(repoFile("scripts", "install.ps1"))
 	if err != nil {
@@ -711,7 +711,7 @@ func TestInstallSHSupportsSavedStateUpdateMode(t *testing.T) {
 		`recover_connection_state_from_arg_stream`,
 		`recover_token_from_default_agent_token_file() {`,
 		`normalize_recovered_agent_arg_key() {`,
-		`-url|-pulse-url|-token|-token-file|-interval|-agent-id|-hostname|-cacert|-server-fingerprint|-health-addr|-state-dir|-kubeconfig|-proxmox-type|-disk-exclude)`,
+		`-url|-pulse-url|-token|-token-file|-interval|-agent-id|-hostname|-cacert|-server-fingerprint|-observers-file|-health-addr|-state-dir|-kubeconfig|-proxmox-type|-disk-exclude)`,
 		`--enable-host|-enable-host|--enable-host=true|-enable-host=true)`,
 		`recover_connection_state_from_env_stream`,
 		`recovered_connection_state_ready() {`,
@@ -556,6 +556,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
                        "internal/hostagent/commands_host_update_test.go",
                        "internal/hostagent/commands_storage_cleanup_test.go",
                        "internal/hostagent/docker_lifecycle_test.go",
+                        "internal/hostagent/observer_delivery_test.go",
                        "internal/hostagent/package_updates_test.go",
                        "internal/hostagent/send_report_test.go",
                        "internal/hostagent/storage_cleanup_test.go",
@@ -4437,6 +4437,7 @@ class SubsystemLookupTest(unittest.TestCase):
                "internal/hostagent/commands_host_update_test.go",
                "internal/hostagent/commands_storage_cleanup_test.go",
                "internal/hostagent/docker_lifecycle_test.go",
+                "internal/hostagent/observer_delivery_test.go",
                "internal/hostagent/package_updates_test.go",
                "internal/hostagent/send_report_test.go",
                "internal/hostagent/storage_cleanup_test.go",