Register agent command channels under the identity Pulse resolves

A token binds the first agent ID its command channel presents and never
moves. The agent registered that channel before its first report, under
its configured --agent-id, or an ID Pulse had forked for a cloned
machine ID or kept from an earlier enrollment. Pulse meanwhile resolved
its reports to another host ID and acknowledged it, and the agent
persisted that ID. The config gate, which judges the resolved ID, then
turned the operator's commands off, and the next restart under the
persisted ID was refused for good.

Report acknowledgements now name commandAgentId: the resolved host ID,
only when the token's binding already holds it or a registration
presenting it would take the first bind, backfill, legacy migration,
hostname rebind or deploy repair it would get anyway. The credential
must also be one command admission takes, and the host must be held
under the reporting token. Naming grants nothing that registration
would not. An established binding to another ID is never moved from
report contents, because resolution can map a report onto another
host's continuity record by machine ID, so a forged report and a stale
one look the same. The config gate takes the same decision.

The agent keeps its channel staged until a report is acknowledged and
registers under the named identity, or under its own ID when none is
named, which is what an older Pulse or an already-bound token sees.
An unreadable ack keeps it waiting, and a 403 releases it under its own
ID because Pulse can never answer that token. A running client is never
rebuilt, since that would reopen its receipt store under in-flight
operations. A superseded client can no longer take the run slot.

Tokens already bound to a diverged ID stay as they are; re-running the
install command issues a fresh token that binds the resolved identity.

docs/release-control/v6/internal/subsystems/agent-lifecycle.md

@@ -8278,6 +8278,52 @@ report path relies on: an applied report carries its token's ID, a report the
 ordering watermark rejects returns the stored host and hostname, and mock
 mode's discard acknowledgement carries no token.
 
+### The command channel registers under the identity Pulse acknowledges
+
+A token binds the first agent ID its command channel presents and never moves
+afterwards, while Pulse resolves the agent's reports to its own host ID, which
+the acknowledgement returns as `agentId` and the agent persists and presents
+after a restart. When the two differed (a configured `--agent-id`, an ID Pulse
+forked for a cloned machine ID, or one it kept from an earlier enrollment), the
+first registration bound the presented ID, the config gate, which judges the
+resolved ID, turned commands off, and the next restart was refused for good.
+The agent therefore keeps its command channel staged until Pulse acknowledges a
+report, then registers under the identity the acknowledgement names as
+`commandAgentId`. Only a real acknowledgement negotiates: one that reports
+success and names the host, for a report presenting the agent ID and hostname
+this process registers with, so a report buffered by an earlier run under
+another identity neither names nor releases the channel. Pulse names the
+identity under the rule in the security-privacy contract, only when the
+token's binding holds the resolved ID or would take it on that registration,
+which also lets an agent restarted under a different `--agent-id` register
+under the identity its token recorded. An acknowledgement
+without it (an older Pulse, or a token already bound to the ID the agent
+presents) leaves the channel on the identity it already uses, the agent's own
+ID by default. An unreadable acknowledgement is no answer and the channel keeps
+waiting; a 403, which means the token lacks report scope and Pulse can never
+answer, releases it under the agent's own ID as before. A client Pulse has
+admitted is never rebuilt under a newly named identity, because a second client
+would reopen the operation receipt store and mark in-flight operations
+interrupted; a client whose registration Pulse has never accepted has received
+no operation, and the agent retires and rebuilds it under the named identity,
+with a compare-and-swap on the client so a retired client can never be admitted
+afterwards. Observer acknowledgements neither name nor release the channel. A
+token bound to an ID its binding will not give up keeps it, since no report
+moves a binding: re-running the install command issues a fresh token that
+binds the resolved identity.
+`TestCommandChannelRegistersUnderTheIdentityPulseAcknowledges` and the other
+`TestCommandChannel*`, `TestRunningCommandChannelIsNeverReplacedByALaterIdentity`,
+`TestRefusedCommandChannelMovesToTheIdentityALaterAckNames`,
+`TestReplayedReportAckNeverNegotiatesTheCurrentChannel`,
+`TestObserverAckNeverReleasesOrNamesTheCommandChannel`,
+`TestSupersededCommandClientCannotOccupyTheRunSlot`, and
+`TestCommandClientRegistrationAdmitsOnlyAnUnretiredClient` tests in
+`internal/hostagent/command_client_test.go` pin the agent side, and
+`TestHostAgentCommandChannelBindsTheIdentityPulseAcknowledges` in
+`internal/api/host_agent_removal_lifecycle_integration_test.go` replays a
+configured-ID install through the router: report, registration, restart, and
+the gate.
+
 ### Alert push routing does not create an agent lifecycle channel
 
 The shared API router may classify an external-probe outage for specialized

docs/release-control/v6/internal/subsystems/api-contracts.md

@@ -10607,6 +10607,19 @@ router echoes it end-to-end on enrollment acks
 (`TestHostAgentRemovalLifecycleThroughAuthenticatedRouterAndRestart`). Agents
 use it to trigger an immediate self-update check after a server upgrade. The
 field is additive: agents that predate it ignore it.
+### Unified-agent report ack names the command identity
+
+The `POST /api/agents/agent/report` acknowledgement may also carry
+`commandAgentId`: the identity the agent's command channel must register under.
+When present it equals `agentId`, and it is present only when the reporting
+token's command binding holds that ID or would take it on registration, under
+the rule in the security-privacy contract. It is omitted otherwise, including
+for tokens without `agent:exec`, tokens command admission refuses outright, and
+mock mode's unresolved acknowledgements, and agents read its absence as "keep
+the command identity you have". The field is additive: agents that predate it
+ignore it and keep registering under their own ID.
+`TestContract_UnifiedAgentReportAckNamesCommandIdentityOnlyWhenTheBindingAdmitsIt`
+in `internal/api/contract_test.go` pins the field.
 ### Security status serves an infrastructureRead settings capability
 
 `GET /api/security/status` adds `settingsCapabilities.infrastructureRead` to the

docs/release-control/v6/internal/subsystems/security-privacy.md

@@ -1509,6 +1509,33 @@ admission holds, and a failed save restores the previous metadata.
 `internal/api/agentbinding/policy_test.go`, and the `TestReportedIdentity*`
 tests and `TestReportHandlerOffersOnlyQualifyingLiveReportsToTheRecorder` in
 `internal/api/security_regression_test.go`, pin that boundary.
+Command-channel identity converges by naming, never by moving a binding.
+`commandIdentityForToken` in `internal/api/agent_ingest.go` returns the
+server-resolved host ID, which `/api/agents/agent/report` acknowledges as
+`commandAgentId` for a host the monitor resolved and holds under the reporting
+token, judged for the hostname the report presents. It names that ID only for
+a credential command admission would take on its legacy path (`agent:exec`, at
+most one organization, no runner, collector, or unsupported runtime role), and
+only when the shared `evaluateAgentExecBinding` decision admits it: the token
+already holds it, or a registration presenting it would take the first bind,
+backfill, legacy migration, hostname rebind, or one-time deploy repair it would
+get anyway. The name grants nothing that registration would not, and it
+discloses no ID the acknowledgement's `agentId` does not already carry. A token
+whose binding would refuse the resolved ID gets no name and keeps its binding.
+Identity resolution can map a report onto another host's continuity record by
+machine ID without a token check, so a stale-ID report and a forged one are
+indistinguishable, and an established binding therefore moves only through the
+moves registration could already make (legacy migration, hostname rebind, and
+the one-time deploy placeholder repair, each still decided at admission), never
+from report contents alone. Mock mode's
+unresolved acknowledgements name nothing. The agent config gate consumes the
+same predicate, so it never enables commands for an identity the channel would
+refuse. `TestReportAckNeverMovesAnEstablishedCommandBinding`,
+`TestReportAckNamesTheResolvedIdentityForAHostnameBoundDeployToken`,
+`TestReportAckLandsTheDeployPlaceholderRepairOnTheResolvedIdentity`,
+`TestReportAckNamesTheRecordedIdentityToALaterReconfiguredAgent`, and
+`TestReportAckNamesNoCommandIdentityInMockMode` in
+`internal/api/security_regression_test.go` pin that boundary.
 Pulse-minted install tokens now carry a server-authored command-policy intent
 in addition to first-use binding metadata. That intent grants no independent
 authority: first-report convergence requires the same shared binding decision

docs/release-control/v6/internal/subsystems/storage-recovery.md

@@ -6497,6 +6497,18 @@ half-bound token. This adds no backup, restore, retention, snapshot, or
 recovery authority. `TestReportedIdentityRestoresTheTokenWhenTheSaveFails` in
 `internal/api/security_regression_test.go` pins the restore and the retry.
 
+### Naming a command identity persists nothing
+
+Naming `commandAgentId` on a report acknowledgement reads the request's token
+snapshot and writes nothing: the command binding is still persisted only by
+command admission and by the report-side identity recording, each in its one
+existing save. The agent keeps the named identity in memory for the process and
+asks again after every restart. This adds no backup, restore, retention,
+snapshot, or recovery authority.
+`TestHostAgentCommandChannelBindsTheIdentityPulseAcknowledges` in
+`internal/api/host_agent_removal_lifecycle_integration_test.go` pins the
+binding the command registration persists.
+
 ### Destination severity policy preserves notification recovery identity
 
 Email, webhook, Apprise, and Relay minimum-severity floors persist through

internal/api/agent_ingest.go

@@ -159,6 +159,17 @@ func (h *UnifiedAgentHandlers) HandleReport(w http.ResponseWriter, r *http.Reque
 	if h.serverVersion != "" {
 		resp["serverVersion"] = h.serverVersion
 	}
+	// Name the identity this agent's command channel must register under, for
+	// a host the monitor resolved and holds under this token, judged for the
+	// hostname this report presents, as its registration will. Mock mode's
+	// discard acknowledgement carries no token ID, so it names nothing.
+	if tokenRecord != nil && host.TokenID == tokenRecord.ID {
+		commandHost := host
+		commandHost.Hostname = reportedHostname
+		if commandAgentID := commandIdentityForToken(tokenRecord, commandHost); commandAgentID != "" {
+			resp["commandAgentId"] = commandAgentID
+		}
+	}
 
 	// Only include config if there are actual overrides
 	if serverConfig.CommandsEnabled != nil {
@@ -688,7 +699,7 @@ func commandConfigAllowedForToken(record *config.APITokenRecord, host models.Hos
 	// commands are enabled when its channel registration would be rejected
 	// strands the host on "Remote control blocked" (the agent reports
 	// CommandsEnabled=true forever while no channel can be admitted).
-	if !evaluateAgentExecBinding(record, host.ID, host.Hostname).admit {
+	if commandIdentityForToken(record, host) == "" {
 		// This gate runs before the agent ever attempts command-channel
 		// registration, so without a log here a refused binding leaves no
 		// trace anywhere: the agent never learns commands were requested and
@@ -707,6 +718,42 @@ func commandConfigAllowedForToken(record *config.APITokenRecord, host models.Hos
 	return true
 }
 
+// commandIdentityForToken names the identity a host agent's command channel
+// must register under for its token to admit it: the host ID Pulse resolved
+// for the agent's report, which the ack returns as agentId and the agent
+// persists and presents after a restart. A token binds the first ID its
+// channel presents, so an agent that registered a configured --agent-id, or
+// an ID Pulse forked or kept from an earlier enrollment, was refused on every
+// restart while the config gate judged the resolved ID.
+//
+// The ID is named only for a credential that command admission would take on
+// its legacy path (agent:exec, at most one organization, no runner,
+// collector, or unsupported runtime role), and only when the shared binding
+// decision admits it: the token already holds it, or a registration
+// presenting it would take the first bind, backfill, legacy migration,
+// hostname rebind, or one-time deploy repair it would get anyway. Naming it
+// grants nothing beyond that registration. An established binding to any
+// other ID is never moved by report contents, which can be steered onto
+// another host's continuity record; the result is empty and the agent keeps
+// registering under the ID it presents. The config gate consumes the same
+// decision, so commands are never reported enabled for an identity the
+// channel would refuse.
+func commandIdentityForToken(record *config.APITokenRecord, host models.Host) string {
+	hostID := strings.TrimSpace(host.ID)
+	if record == nil || hostID == "" || !record.HasScope(config.ScopeAgentExec) || len(record.GetBoundOrgs()) > 1 {
+		return ""
+	}
+	switch strings.TrimSpace(record.Metadata[agenttokens.RuntimeRoleMetadataKey]) {
+	case "", agenttokens.CredentialKindLegacyFullTrust:
+	default:
+		return ""
+	}
+	if !evaluateAgentExecBinding(record, hostID, host.Hostname).admit {
+		return ""
+	}
+	return hostID
+}
+
 func (h *UnifiedAgentHandlers) ensureAgentTokenMatch(w http.ResponseWriter, r *http.Request, agentID string) bool {
 	record := getAPITokenRecordFromRequest(r)
 	if record == nil {

internal/api/contract_test.go

@@ -24560,6 +24560,91 @@ func TestUnifiedAgentHandlers_HandleReportAckOmitsEmptyServerVersion(t *testing.
 	}
 }
 
+// The report acknowledgement names the command identity only as the identity
+// the token's binding admits: equal to agentId when present, and absent for a
+// token bound elsewhere, without agent:exec, or one command admission refuses
+// outright (a collector role, more than one organization), which agents read
+// as "keep the identity you have".
+func TestContract_UnifiedAgentReportAckNamesCommandIdentityOnlyWhenTheBindingAdmitsIt(t *testing.T) {
+	cases := []struct {
+		name     string
+		scopes   []string
+		metadata map[string]string
+		orgs     []string
+		wantName bool
+	}{
+		{
+			name:     "unbound install token",
+			scopes:   []string{config.ScopeAgentReport, config.ScopeAgentExec},
+			metadata: map[string]string{"issued_via": agentInstallIssuedViaConfig, "install_type": "host"},
+			wantName: true,
+		},
+		{
+			name:   "bound to another identity",
+			scopes: []string{config.ScopeAgentReport, config.ScopeAgentExec},
+			metadata: map[string]string{
+				"issued_via": agentInstallIssuedViaConfig, "install_type": "host",
+				"bound_agent_id": "configured-id", "bound_hostname": "contract-host",
+				agentExecBindingVersionKey: agentExecBindingVersion,
+			},
+		},
+		{
+			name:     "no agent:exec",
+			scopes:   []string{config.ScopeAgentReport},
+			metadata: map[string]string{"issued_via": agentInstallIssuedViaConfig, "install_type": "host"},
+		},
+		{
+			name:   "monitoring collector",
+			scopes: []string{config.ScopeAgentReport, config.ScopeAgentExec},
+			metadata: map[string]string{
+				"issued_via": agentInstallIssuedViaConfig, "install_type": "host",
+				agenttokens.RuntimeRoleMetadataKey: agenttokens.CredentialKindMonitoringCollector,
+			},
+		},
+		{
+			name:     "bound to two organizations",
+			scopes:   []string{config.ScopeAgentReport, config.ScopeAgentExec},
+			metadata: map[string]string{"issued_via": agentInstallIssuedViaConfig, "install_type": "host"},
+			orgs:     []string{"default", "tenant-b"},
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			record := newTokenRecord(t, "contract-command-identity-token.12345678", tc.scopes, tc.metadata)
+			record.OrgIDs = tc.orgs
+			cfg := newTestConfigWithTokens(t, record)
+			handler, _ := newUnifiedAgentHandlers(t, cfg)
+
+			body, err := json.Marshal(agentshost.Report{
+				Agent:     agentshost.AgentInfo{ID: "configured-id", Version: "6.5.0"},
+				Host:      agentshost.HostInfo{ID: "machine-contract", MachineID: "machine-contract", Hostname: "contract-host", Platform: "linux"},
+				Timestamp: time.Now().UTC(),
+			})
+			if err != nil {
+				t.Fatalf("marshal report: %v", err)
+			}
+			req := httptest.NewRequest(http.MethodPost, "/api/agents/agent/report", bytes.NewReader(body))
+			attachAPITokenRecord(req, &cfg.APITokens[0])
+			rec := httptest.NewRecorder()
+			handler.HandleReport(rec, req)
+			if rec.Code != http.StatusOK {
+				t.Fatalf("status = %d: %s", rec.Code, rec.Body.String())
+			}
+			var ack map[string]any
+			if err := json.Unmarshal(rec.Body.Bytes(), &ack); err != nil {
+				t.Fatalf("decode ack: %v", err)
+			}
+			named, present := ack["commandAgentId"]
+			if present != tc.wantName {
+				t.Fatalf("commandAgentId present = %v (%v), want %v", present, named, tc.wantName)
+			}
+			if present && named != ack["agentId"] {
+				t.Fatalf("commandAgentId = %v, want it equal to agentId %v", named, ack["agentId"])
+			}
+		})
+	}
+}
+
 // App bootstrap requires presentation defaults for every authenticated role.
 // Pin both sides of the privilege boundary: the narrow runtime projection is
 // readable by a viewer, while the complete system settings payload remains

internal/api/host_agent_removal_lifecycle_integration_test.go

@@ -594,6 +594,93 @@ func TestHostAgentFreshInstallTokenReplacesStaleDisabledCommandPolicyOnce(t *tes
 	}
 }
 
+// An agent started with a configured --agent-id presents that ID while Pulse
+// resolves its reports to the machine's host ID, acknowledges that ID, and the
+// agent persists it. Its command channel used to bind the install token to the
+// configured ID on first registration, so the config gate (which judges the
+// resolved ID) turned the operator's commands-enabled install off on the next
+// report, and a restart under the persisted ID was refused for good. The
+// acknowledgement now names the resolved ID as the command identity, the
+// agent's channel binds it, and every later report, restart and gate decision
+// agrees on it.
+func TestHostAgentCommandChannelBindsTheIdentityPulseAcknowledges(t *testing.T) {
+	dataPath := t.TempDir()
+	const adminRaw = "command-identity-admin-token-123.12345678"
+	admin := newTokenRecord(t, adminRaw, []string{config.ScopeWildcard}, nil)
+	if err := config.NewConfigPersistence(dataPath).SaveAPITokens([]config.APITokenRecord{admin}); err != nil {
+		t.Fatalf("SaveAPITokens: %v", err)
+	}
+	runtime := newHostRemovalLifecycleHTTPRuntime(t, dataPath, []config.APITokenRecord{admin})
+	t.Cleanup(runtime.stop)
+
+	installRec := serveHostRemovalLifecycleRequest(t, runtime, http.MethodPost, "/api/agent-install-command", adminRaw,
+		[]byte(`{"type":"host","enableCommands":true,"name":"configured-agent-id-install"}`))
+	install := decodeHostAgentInstallResponse(t, installRec)
+	if install.Token == "" || install.Record == nil {
+		t.Fatalf("commands-enabled install omitted token record: %+v", install)
+	}
+
+	const (
+		presentedID = "configured-agent-id"
+		machineID   = "machine-configured-agent-id"
+		hostname    = "configured-host"
+	)
+	report := agentshost.Report{
+		Agent:     agentshost.AgentInfo{ID: presentedID, Version: "6.5.0", Type: "unified", CommandsEnabled: true},
+		Host:      agentshost.HostInfo{ID: machineID, MachineID: machineID, Hostname: hostname, Platform: "linux"},
+		Timestamp: time.Now().UTC(),
+	}
+	ack := func() map[string]any {
+		t.Helper()
+		body, err := json.Marshal(report)
+		if err != nil {
+			t.Fatalf("marshal report: %v", err)
+		}
+		rec := serveHostRemovalLifecycleRequest(t, runtime, http.MethodPost, "/api/agents/agent/report", install.Token, body)
+		if rec.Code != http.StatusOK {
+			t.Fatalf("report status = %d: %s", rec.Code, rec.Body.String())
+		}
+		var decoded map[string]any
+		if err := json.Unmarshal(rec.Body.Bytes(), &decoded); err != nil {
+			t.Fatalf("decode ack: %v", err)
+		}
+		report.Timestamp = report.Timestamp.Add(time.Minute)
+		return decoded
+	}
+
+	first := ack()
+	resolved, _ := first["agentId"].(string)
+	if resolved != machineID {
+		t.Fatalf("ack agentId = %q, want the resolved host ID %q", resolved, machineID)
+	}
+	if got, _ := first["commandAgentId"].(string); got != resolved {
+		t.Fatalf("ack commandAgentId = %v, want the resolved %q", first["commandAgentId"], resolved)
+	}
+
+	// The agent registers its command channel under the named identity, then
+	// restarts under the identity it persisted from the acknowledgement.
+	if _, ok := runtime.router.admitAgentExecToken(install.Token, resolved, hostname); !ok {
+		t.Fatal("first command registration under the named identity was refused")
+	}
+	if _, ok := runtime.router.admitAgentExecToken(install.Token, resolved, hostname); !ok {
+		t.Fatal("restart registration under the acknowledged identity was refused")
+	}
+
+	// The agent still presents its configured ID in reports. The gate keeps
+	// the operator's commands-enabled install on, and Pulse keeps naming the
+	// bound identity.
+	later := ack()
+	if got, _ := later["commandAgentId"].(string); got != resolved {
+		t.Fatalf("later ack commandAgentId = %v, want %q", later["commandAgentId"], resolved)
+	}
+	if cfg, _ := later["config"].(map[string]any); cfg == nil || cfg["commandsEnabled"] != true {
+		t.Fatalf("later ack config = %v, want commands kept enabled", later["config"])
+	}
+	if _, ok := runtime.router.admitAgentExecToken(install.Token, presentedID, hostname); ok {
+		t.Fatal("the install token admitted the presented identity as a second command identity")
+	}
+}
+
 func tokenRecordByID(records []config.APITokenRecord, tokenID string) *config.APITokenRecord {
 	for index := range records {
 		if records[index].ID == tokenID {

internal/api/security_regression_test.go

@@ -6046,3 +6046,164 @@ func TestReportHandlerOffersOnlyQualifyingLiveReportsToTheRecorder(t *testing.T)
 		t.Fatalf("recorder calls in mock mode = %+v, want none beyond the live report", calls[1:])
 	}
 }
+
+// postReportIdentityAck posts an authenticated report through the router and
+// returns the whole acknowledgement.
+func postReportIdentityAck(t *testing.T, router *Router, token string, report agentshost.Report) map[string]any {
+	t.Helper()
+	code, payload := issue1753PostReport(t, router, token, report)
+	if code != http.StatusOK {
+		t.Fatalf("report rejected with %d (%v)", code, payload)
+	}
+	return payload
+}
+
+// A token already bound to one identity is never moved by report contents.
+// Identity resolution can map a report onto another host's continuity record
+// by machine ID without a token check, so a report resolving elsewhere names
+// no command identity, leaves the binding as it was, and the agent keeps
+// registering under the ID it presents.
+func TestReportAckNeverMovesAnEstablishedCommandBinding(t *testing.T) {
+	const raw = "command-identity-established-token.12345678"
+	record := newTokenRecord(t, raw, agenttokens.HostScopes(true), map[string]string{
+		"issued_via":                       agentbinding.IssuedViaConfig,
+		"install_type":                     "host",
+		"bound_agent_id":                   "configured-id",
+		"bound_hostname":                   "host-established",
+		agentExecBindingVersionKey:         agentExecBindingVersion,
+		agenttokens.RuntimeRoleMetadataKey: agenttokens.CredentialKindLegacyFullTrust,
+	})
+	record.OrgID = "default"
+	cfg := newTestConfigWithTokens(t, record)
+	router := newReportIdentityRouter(t, cfg)
+
+	ack := postReportIdentityAck(t, router, raw, reportIdentityReport("configured-id", reportIdentityMachineID, "host-established"))
+	if ack["agentId"] != reportIdentityMachineID {
+		t.Fatalf("precondition: acknowledged agentId = %v, want the machine ID", ack["agentId"])
+	}
+	if named, present := ack["commandAgentId"]; present {
+		t.Fatalf("ack named command identity %v for a token bound to another identity", named)
+	}
+	if got := reportIdentityToken(t, cfg, record.ID).Metadata["bound_agent_id"]; got != "configured-id" {
+		t.Fatalf("bound_agent_id = %q after the report, want it unchanged", got)
+	}
+	if _, ok := router.admitAgentExecToken(raw, reportIdentityMachineID, "host-established"); ok {
+		t.Fatal("the established binding moved to the resolved identity")
+	}
+	if _, ok := router.admitAgentExecToken(raw, "configured-id", "host-established"); !ok {
+		t.Fatal("the agent lost its established command identity")
+	}
+}
+
+// Deploy enrollment binds the runtime token to the node's hostname only. When
+// the agent presents another ID than Pulse resolves (a configured --agent-id),
+// the report path records nothing, and the acknowledgement names the resolved
+// ID so the agent's command channel backfills that one instead.
+func TestReportAckNamesTheResolvedIdentityForAHostnameBoundDeployToken(t *testing.T) {
+	const raw = "command-identity-deploy-token.12345678"
+	record := newTokenRecord(t, raw, agenttokens.HostScopes(true), map[string]string{
+		"bound_hostname":               "pve-node5",
+		"deploy_job_id":                "dep_5",
+		agentExecBindingVersionKey:     agentExecBindingVersion,
+		agentbinding.DeployIdentityKey: agentbinding.DeployIdentityAgent,
+	})
+	record.OrgID = "default"
+	cfg := newTestConfigWithTokens(t, record)
+	router := newReportIdentityRouter(t, cfg)
+
+	ack := postReportIdentityAck(t, router, raw, reportIdentityReport("configured-id", reportIdentityMachineID, "pve-node5"))
+	if ack["commandAgentId"] != reportIdentityMachineID || ack["agentId"] != reportIdentityMachineID {
+		t.Fatalf("ack agentId=%v commandAgentId=%v, want both the resolved %q", ack["agentId"], ack["commandAgentId"], reportIdentityMachineID)
+	}
+	if _, ok := router.admitAgentExecToken(raw, reportIdentityMachineID, "pve-node5"); !ok {
+		t.Fatal("the command channel could not backfill the named identity")
+	}
+	if _, ok := router.admitAgentExecToken(raw, "configured-id", "pve-node5"); ok {
+		t.Fatal("the backfilled token admitted the presented identity too")
+	}
+	bound := reportIdentityToken(t, cfg, record.ID)
+	if !commandConfigAllowedForToken(&bound, models.Host{ID: reportIdentityMachineID, Hostname: "pve-node5"}) {
+		t.Fatal("the config gate refused the identity the command channel admits")
+	}
+}
+
+// A historical deploy token still bound to the invented agent-<hostname> may
+// move once, on exactly its hostname, to the identity that registers. Naming
+// the resolved ID lands that one move on the identity the agent's reports and
+// restarts use, not on whatever ID the agent happened to present.
+func TestReportAckLandsTheDeployPlaceholderRepairOnTheResolvedIdentity(t *testing.T) {
+	const raw = "command-identity-placeholder-token.12345678"
+	record := newTokenRecord(t, raw, agenttokens.HostScopes(true), map[string]string{
+		"bound_agent_id":           "agent-pve-node6",
+		"bound_hostname":           "pve-node6",
+		"deploy_job_id":            "dep_6",
+		agentExecBindingVersionKey: agentExecBindingVersion,
+	})
+	record.OrgID = "default"
+	cfg := newTestConfigWithTokens(t, record)
+	router := newReportIdentityRouter(t, cfg)
+
+	ack := postReportIdentityAck(t, router, raw, reportIdentityReport("configured-id", reportIdentityMachineID, "pve-node6"))
+	if ack["commandAgentId"] != reportIdentityMachineID {
+		t.Fatalf("ack commandAgentId = %v, want the resolved %q", ack["commandAgentId"], reportIdentityMachineID)
+	}
+	if _, ok := router.admitAgentExecToken(raw, reportIdentityMachineID, "pve-node6"); !ok {
+		t.Fatal("the placeholder token did not take its one-time repair to the named identity")
+	}
+	if got := reportIdentityToken(t, cfg, record.ID).Metadata; got["bound_agent_id"] != reportIdentityMachineID ||
+		got[agentbinding.DeployIdentityKey] != agentbinding.DeployIdentityRepaired {
+		t.Fatalf("binding after repair = %v, want %q marked repaired", got, reportIdentityMachineID)
+	}
+	if _, ok := router.admitAgentExecToken(raw, "configured-id", "pve-node6"); ok {
+		t.Fatal("the repaired token moved a second time")
+	}
+}
+
+// The report path records the first identity an agent reports under, and it
+// is final. An agent later restarted under a different --agent-id is refused
+// under that ID, so the acknowledgement names the recorded identity and the
+// agent's command channel registers under it instead.
+func TestReportAckNamesTheRecordedIdentityToALaterReconfiguredAgent(t *testing.T) {
+	const raw = "command-identity-reconfigured-token.12345678"
+	record := autoRegisteredInstallToken(t, raw, "pve10", true)
+	cfg := newTestConfigWithTokens(t, record)
+	router := newReportIdentityRouter(t, cfg)
+
+	postReportIdentityReport(t, router, raw, reportIdentityReport(reportIdentityMachineID, reportIdentityMachineID, "pve10"))
+	ack := postReportIdentityAck(t, router, raw, reportIdentityReport("configured-id", reportIdentityMachineID, "pve10"))
+	if ack["commandAgentId"] != reportIdentityMachineID {
+		t.Fatalf("ack commandAgentId = %v, want the recorded %q", ack["commandAgentId"], reportIdentityMachineID)
+	}
+	if _, ok := router.admitAgentExecToken(raw, reportIdentityMachineID, "pve10"); !ok {
+		t.Fatal("the reconfigured agent's registration under the named identity was refused")
+	}
+}
+
+// Mock mode discards reports and acknowledges the presented ID unresolved, on
+// a host held under no token. Naming that ID would steer a fresh token's first
+// command registration onto an identity Pulse never resolved, so a mock ack
+// names no command identity.
+func TestReportAckNamesNoCommandIdentityInMockMode(t *testing.T) {
+	const raw = "command-identity-mock-token.12345678"
+	record := newTokenRecord(t, raw, agenttokens.HostScopes(true), map[string]string{
+		"issued_via":   agentbinding.IssuedViaConfig,
+		"install_type": "host",
+	})
+	record.OrgID = "default"
+	cfg := newTestConfigWithTokens(t, record)
+	router := newReportIdentityRouter(t, cfg)
+
+	previous := mock.IsMockEnabled()
+	if err := mock.SetEnabled(true); err != nil {
+		t.Fatalf("enable mock mode: %v", err)
+	}
+	t.Cleanup(func() { _ = mock.SetEnabled(previous) })
+
+	ack := postReportIdentityAck(t, router, raw, reportIdentityReport("configured-id", reportIdentityMachineID, "host-mock"))
+	if ack["agentId"] != "configured-id" {
+		t.Fatalf("precondition: mock ack agentId = %v, want the presented ID echoed", ack["agentId"])
+	}
+	if named, present := ack["commandAgentId"]; present {
+		t.Fatalf("mock ack named command identity %v", named)
+	}
+}

internal/hostagent/agent.go

@@ -191,6 +191,16 @@ type Agent struct {
 	commandClientMu        sync.Mutex
 	commandClientRunCancel context.CancelFunc
 	commandClientParentCtx context.Context
+	// commandClientRunCtx holds Run's context until the first acknowledged
+	// report releases it as commandClientParentCtx. The command channel must
+	// not register before Pulse has said which identity its token binding
+	// will hold: a first registration binds whatever ID it presents, and an
+	// agent that later restarts under the acknowledged ID is refused for good.
+	commandClientRunCtx context.Context
+	// commandIdentity is the command-channel identity Pulse acknowledged
+	// (commandAgentId on a report ack). Empty means Pulse named none, and the
+	// channel registers under agentID as it always has.
+	commandIdentity string
 	// commandAuthorityProfile is the immutable local install-time ceiling for
 	// this process. Remote configuration may disable and later re-enable a
 	// command-capable or legacy install, but it cannot promote monitoring-only.
@@ -626,12 +636,15 @@ func (a *Agent) Run(ctx context.Context) error {
 		return a.runOnce(ctx)
 	}
 
+	// The command channel waits for the first acknowledged report, which
+	// releases this context to it (adoptCommandIdentity).
 	a.commandClientMu.Lock()
-	a.commandClientParentCtx = ctx
+	a.commandClientRunCtx = ctx
 	commandClient := a.commandClient
 	a.commandClientMu.Unlock()
 	defer func() {
 		a.commandClientMu.Lock()
+		a.commandClientRunCtx = nil
 		a.commandClientParentCtx = nil
 		a.commandClientMu.Unlock()
 		a.stopCommandClient(true)
@@ -647,7 +660,7 @@ func (a *Agent) Run(ctx context.Context) error {
 		// original was built with the bootstrap token during New().
 		if a.cfg.EnableCommands {
 			a.commandClientMu.Lock()
-			a.commandClient = a.newCommandClient(a.cfg, a.agentID, a.hostname, a.platform, a.agentVersion)
+			a.commandClient = a.newCommandClient(a.cfg, a.commandRegistrationIdentityLocked(), a.hostname, a.platform, a.agentVersion)
 			commandClient = a.commandClient
 			a.commandClientMu.Unlock()
 		}
@@ -659,7 +672,8 @@ func (a *Agent) Run(ctx context.Context) error {
 		go a.runProxmoxHealthCheckLoop(ctx)
 	}
 
-	// Start command client in background for AI command execution
+	// Stage the command client for AI command execution. It starts once the
+	// first report is acknowledged.
 	if commandClient != nil {
 		a.startCommandClient(commandClient)
 	}
@@ -813,6 +827,13 @@ func (a *Agent) startCommandClient(client *CommandClient) bool {
 		a.commandClientMu.Unlock()
 		return true
 	}
+	// A caller may hold a client that a concurrent disable already closed and
+	// cleared. Running it would occupy the run slot with a dead client, and
+	// the next enabled client would never start.
+	if a.commandClient != client {
+		a.commandClientMu.Unlock()
+		return false
+	}
 
 	parentCtx := a.commandClientParentCtx
 	if parentCtx == nil {
@@ -831,6 +852,84 @@ func (a *Agent) startCommandClient(client *CommandClient) bool {
 	return true
 }
 
+// commandRegistrationIdentityLocked returns the agent ID the command channel
+// registers under: the identity Pulse acknowledged for it, or this agent's own
+// ID when Pulse named none. Callers hold commandClientMu.
+func (a *Agent) commandRegistrationIdentityLocked() string {
+	if a.commandIdentity != "" {
+		return a.commandIdentity
+	}
+	return a.agentID
+}
+
+// adoptCommandIdentity runs after every acknowledged report. Pulse resolves
+// this agent to its own host ID, which the agent persists and presents after a
+// restart, and names that ID as commandAgentId whenever the token's binding
+// holds it or would take it on first registration. The command channel must
+// register under it, because a token binds the first ID its channel presents
+// and never moves afterwards: registering under a configured --agent-id, or an
+// ID the server forked or kept from an earlier enrollment, strands the channel
+// on the next restart.
+//
+// An ack without commandAgentId keeps the identity already in use (an older
+// Pulse, or a token already bound to the ID this agent presents). The first
+// ack also releases Run's context, so the channel never registers before Pulse
+// has answered. A client under another ID is replaced while Pulse has never
+// admitted it (staged, or still retrying a refused registration). An admitted
+// client is never replaced: its token accepted it, and building a second
+// client would reopen the operation receipt store, which marks in-flight
+// operations interrupted. A new identity then applies to the next client this
+// process builds.
+func (a *Agent) adoptCommandIdentity(commandAgentID string) {
+	commandAgentID = strings.TrimSpace(commandAgentID)
+
+	a.configMu.RLock()
+	commandCfg := a.cfg
+	a.configMu.RUnlock()
+
+	a.commandClientMu.Lock()
+	if commandAgentID != "" && commandAgentID != a.commandIdentity {
+		a.commandIdentity = commandAgentID
+		if commandAgentID != a.agentID {
+			a.logger.Info().
+				Str("agent_id", a.agentID).
+				Str("command_agent_id", commandAgentID).
+				Msg("Registering the command channel under the identity Pulse acknowledged for this agent")
+		}
+	}
+	if a.commandClientParentCtx == nil && a.commandClientRunCtx != nil {
+		a.commandClientParentCtx = a.commandClientRunCtx
+	}
+	identity := a.commandRegistrationIdentityLocked()
+	client := a.commandClient
+	if client != nil && client.agentID != identity &&
+		(a.commandClientRunCancel == nil || client.retireIfNeverAdmitted()) {
+		if a.commandClientRunCancel != nil {
+			a.commandClientRunCancel()
+			a.commandClientRunCancel = nil
+		}
+		if err := client.Close(); err != nil {
+			a.logger.Debug().Err(err).Msg("Error closing command client staged under a superseded identity")
+		}
+		client = a.newCommandClient(commandCfg, identity, a.hostname, a.platform, a.agentVersion)
+		a.commandClient = client
+	}
+	a.commandClientMu.Unlock()
+
+	if client != nil {
+		a.startCommandClient(client)
+	}
+}
+
+// releaseCommandChannelWithoutReports lets the command channel register under
+// the identity it already has when Pulse refuses this token's reports outright
+// (403: the token lacks report scope). Pulse can then never name a command
+// identity, and a token that may still run commands keeps the channel it had
+// before registration waited for an acknowledgement.
+func (a *Agent) releaseCommandChannelWithoutReports() {
+	a.adoptCommandIdentity("")
+}
+
 func (a *Agent) stopCommandClient(clearClient bool) {
 	a.commandClientMu.Lock()
 	client := a.commandClient
@@ -889,6 +988,7 @@ func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Repo
 				Str("endpoint", statusErr.Endpoint).
 				Int("status_code", statusErr.StatusCode).
 				Msg("Failed to send Unified Agent report (403 Forbidden). API token may lack 'Unified Agent reporting' scope. Set PULSE_ENABLE_HOST=false if host monitoring is not needed.")
+			a.releaseCommandChannelWithoutReports()
 			return nil
 		}
 		if errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusUnauthorized {
@@ -1057,6 +1157,9 @@ func (a *Agent) flushBuffer(ctx context.Context) {
 				// accepted with it. Drop them so the buffer cannot grow
 				// unbounded across restarts, and surface the actionable error.
 				a.logAuthFailure(statusErr)
+				if statusErr.StatusCode == http.StatusForbidden {
+					a.releaseCommandChannelWithoutReports()
+				}
 				for {
 					if _, ok := a.reportBuffer.Pop(); !ok {
 						break
@@ -1440,15 +1543,18 @@ func (a *Agent) sendReportToDestination(ctx context.Context, report agentshost.R
 
 	// Parse response to check for server-side config overrides
 	var reportResp struct {
-		Success       bool   `json:"success"`
-		AgentID       string `json:"agentId"`
-		ServerVersion string `json:"serverVersion"`
-		Config        *struct {
+		Success        bool   `json:"success"`
+		AgentID        string `json:"agentId"`
+		CommandAgentID string `json:"commandAgentId"`
+		ServerVersion  string `json:"serverVersion"`
+		Config         *struct {
 			CommandsEnabled *bool `json:"commandsEnabled"`
 		} `json:"config,omitempty"`
 	}
 	if err := json.NewDecoder(resp.Body).Decode(&reportResp); err != nil {
-		// Non-fatal: just log and continue
+		// Non-fatal: just log and continue. An unreadable ack is no answer, so
+		// the command channel keeps waiting: registering now could bind a
+		// fresh token to an ID the next readable ack would not name.
 		a.logger.Debug().Err(err).Msg("Failed to parse report response, ignoring config")
 		return nil
 	}
@@ -1473,6 +1579,17 @@ func (a *Agent) sendReportToDestination(ctx context.Context, report agentshost.R
 		a.applyRemoteConfig(*reportResp.Config.CommandsEnabled)
 	}
 
+	// Only a real acknowledgement answers for this process's identity: one
+	// that accepted the report and named its host, for a report that presents
+	// the agent ID and hostname this process registers with. A replayed report
+	// buffered by an earlier run may carry another of either, and Pulse judged
+	// its command identity for that one.
+	if reportResp.Success && canonicalAgentID != "" &&
+		strings.TrimSpace(report.Agent.ID) == a.agentID &&
+		strings.TrimSpace(report.Host.Hostname) == a.hostname {
+		a.adoptCommandIdentity(reportResp.CommandAgentID)
+	}
+
 	return nil
 }
 
@@ -1636,7 +1753,7 @@ func (a *Agent) applyRemoteConfig(commandsEnabled bool) (bool, bool) {
 	currentlyEnabled := a.commandClient != nil
 	switch {
 	case effectiveState && !currentlyEnabled:
-		clientToStart = a.newCommandClient(commandCfg, a.agentID, a.hostname, a.platform, a.agentVersion)
+		clientToStart = a.newCommandClient(commandCfg, a.commandRegistrationIdentityLocked(), a.hostname, a.platform, a.agentVersion)
 		a.commandClient = clientToStart
 	case !effectiveState && currentlyEnabled:
 		shouldStop = true

internal/hostagent/command_client_test.go

@@ -2,14 +2,21 @@ package hostagent
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
+	"net/http"
+	"net/http/httptest"
 	"reflect"
+	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/gorilla/websocket"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agentexec"
+	agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
 	"github.com/rs/zerolog"
 )
 
@@ -531,3 +538,527 @@ func TestCommandClientGuestOutcomeContract(t *testing.T) {
 	t.Run("execution-and-admission", testGuestExecProjectsGuestOutcome)
 	t.Run("terminal-dictionary", TestGuestExecTerminalDictionary)
 }
+
+// commandIdentityHarness runs an agent with commands enabled against a fake
+// Pulse whose report acknowledgement the test scripts, and records the
+// identity of every command client that starts.
+type commandIdentityHarness struct {
+	agent   *Agent
+	server  *httptest.Server
+	started chan string
+
+	mu      sync.Mutex
+	reports int
+	built   int
+	// refuse makes every started client behave as if Pulse refused its
+	// registration; otherwise a started client counts as admitted.
+	refuse bool
+}
+
+func newCommandIdentityHarness(t *testing.T, presentedID string, interval time.Duration, respond func(report int, w http.ResponseWriter)) *commandIdentityHarness {
+	t.Helper()
+	h := &commandIdentityHarness{started: make(chan string, 8)}
+	h.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		h.mu.Lock()
+		h.reports++
+		report := h.reports
+		h.mu.Unlock()
+		respond(report, w)
+	}))
+	t.Cleanup(h.server.Close)
+
+	logger := zerolog.Nop()
+	agent, err := New(Config{
+		APIToken:       "token",
+		PulseURL:       "https://pulse",
+		AgentID:        presentedID,
+		EnableCommands: true,
+		Interval:       interval,
+		StateDir:       t.TempDir(),
+		Collector:      &mockCollector{},
+		Logger:         &logger,
+		newCommandClientFn: func(cfg Config, agentID, hostname, platform, version string) *CommandClient {
+			h.mu.Lock()
+			h.built++
+			h.mu.Unlock()
+			return NewCommandClient(cfg, agentID, hostname, platform, version)
+		},
+		runCommandClientFn: func(client *CommandClient, ctx context.Context) error {
+			h.mu.Lock()
+			refuse := h.refuse
+			h.mu.Unlock()
+			if !refuse {
+				client.markAdmitted()
+			}
+			h.started <- client.agentID
+			<-ctx.Done()
+			return ctx.Err()
+		},
+	})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	agent.httpClient = h.server.Client()
+	agent.trimmedPulseURL = h.server.URL
+	h.agent = agent
+	return h
+}
+
+func (h *commandIdentityHarness) run(t *testing.T) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		_ = h.agent.Run(ctx)
+	}()
+	t.Cleanup(func() {
+		cancel()
+		<-done
+	})
+}
+
+func (h *commandIdentityHarness) waitForReports(t *testing.T, want int) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		h.mu.Lock()
+		reports := h.reports
+		h.mu.Unlock()
+		if reports >= want {
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("only %d reports were sent, want %d", reports, want)
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+}
+
+func (h *commandIdentityHarness) clientIdentity() string {
+	h.agent.commandClientMu.Lock()
+	defer h.agent.commandClientMu.Unlock()
+	if h.agent.commandClient == nil {
+		return ""
+	}
+	return h.agent.commandClient.agentID
+}
+
+func (h *commandIdentityHarness) expectStart(t *testing.T, want string) {
+	t.Helper()
+	select {
+	case got := <-h.started:
+		if got != want {
+			t.Fatalf("command channel registered as %q, want %q", got, want)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatalf("command channel never started; want it registered as %q", want)
+	}
+}
+
+func (h *commandIdentityHarness) expectNoStart(t *testing.T, within time.Duration) {
+	t.Helper()
+	select {
+	case got := <-h.started:
+		t.Fatalf("command channel registered as %q before Pulse acknowledged a report", got)
+	case <-time.After(within):
+	}
+}
+
+func (h *commandIdentityHarness) expectNoRestart(t *testing.T) {
+	t.Helper()
+	select {
+	case got := <-h.started:
+		t.Fatalf("command channel registered again, as %q", got)
+	default:
+	}
+}
+
+func writeAck(w http.ResponseWriter, body string) {
+	w.Header().Set("Content-Type", "application/json")
+	_, _ = w.Write([]byte(body))
+}
+
+// A token binds the first identity its command channel presents and never
+// moves. An agent started with a configured ID (or an ID Pulse forked, or
+// kept from an earlier enrollment) used to register that ID before its first
+// report, while Pulse resolved its reports to another host ID that the agent
+// persisted and presented after the next restart: the channel was refused for
+// good. The channel now waits for the acknowledgement and registers under the
+// identity Pulse names for it.
+func TestCommandChannelRegistersUnderTheIdentityPulseAcknowledges(t *testing.T) {
+	release := make(chan struct{})
+	h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(report int, w http.ResponseWriter) {
+		if report == 1 {
+			<-release
+		}
+		writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`)
+	})
+	h.run(t)
+
+	h.expectNoStart(t, 300*time.Millisecond)
+	close(release)
+	h.expectStart(t, "server-resolved")
+}
+
+// An older Pulse names no command identity, and a current Pulse names none for
+// a token already bound to the ID this agent presents (for example a
+// configured --agent-id). The channel then registers under the agent's own ID
+// exactly as before, once a report is acknowledged.
+func TestCommandChannelKeepsItsOwnIdentityWhenPulseNamesNone(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) {
+		writeAck(w, `{"success":true,"agentId":"server-resolved"}`)
+	})
+	h.run(t)
+
+	h.expectStart(t, "configured-id")
+}
+
+// While Pulse is unreachable or refusing reports the agent has no answer, so
+// the channel stays staged and registers once a later report is acknowledged.
+func TestCommandChannelWaitsForAnAcknowledgedReport(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) {
+		if report < 3 {
+			http.Error(w, "unavailable", http.StatusServiceUnavailable)
+			return
+		}
+		writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`)
+	})
+	h.run(t)
+
+	h.expectStart(t, "server-resolved")
+	h.mu.Lock()
+	reports := h.reports
+	h.mu.Unlock()
+	if reports < 3 {
+		t.Fatalf("command channel started after %d report attempts, want it to wait for the accepted third", reports)
+	}
+}
+
+// Once Pulse has named the channel's identity, a later acknowledgement that
+// names none (Pulse now resolves the host elsewhere, and the binding refuses
+// that ID) must not drop the channel back to a different ID: the token is
+// bound to the acknowledged one.
+func TestCommandChannelKeepsTheAcknowledgedIdentityWhenALaterAckNamesNone(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) {
+		if report == 1 {
+			writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`)
+			return
+		}
+		writeAck(w, `{"success":true,"agentId":"healed-id"}`)
+	})
+	h.run(t)
+
+	h.expectStart(t, "server-resolved")
+	h.waitForReports(t, 4)
+	h.expectNoRestart(t)
+	if got := h.clientIdentity(); got != "server-resolved" {
+		t.Fatalf("command client identity = %q, want the acknowledged %q", got, "server-resolved")
+	}
+
+	// A client rebuilt later in this process (commands disabled, then enabled
+	// again) still registers under the acknowledged identity.
+	h.agent.applyRemoteConfig(false)
+	h.agent.applyRemoteConfig(true)
+	h.expectStart(t, "server-resolved")
+}
+
+// Commands enabled by Pulse after start-up (rather than by --enable-commands)
+// stage a client before any report is acknowledged. That client must register
+// under the acknowledged identity, not the presented one.
+func TestRemotelyEnabledCommandChannelRegistersUnderTheAcknowledgedIdentity(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) {
+		writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved","config":{"commandsEnabled":true}}`)
+	})
+	h.agent.stopCommandClient(true)
+	h.agent.configMu.Lock()
+	h.agent.cfg.EnableCommands = false
+	h.agent.configMu.Unlock()
+	h.run(t)
+
+	h.expectStart(t, "server-resolved")
+}
+
+// An unreadable acknowledgement is no answer. Registering on it could bind a
+// fresh token to the presented ID, which the next readable ack would not name,
+// so the channel waits for an ack it can read.
+func TestCommandChannelIgnoresAnUnreadableAck(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) {
+		if report == 1 {
+			w.Header().Set("Content-Type", "text/html")
+			_, _ = w.Write([]byte("<html>proxy</html>"))
+			return
+		}
+		writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`)
+	})
+	h.run(t)
+
+	h.expectStart(t, "server-resolved")
+}
+
+// A token Pulse refuses reports for (403: no report scope) can never be named
+// a command identity, and it may still run commands. Its channel registers
+// under the agent's own ID as it did before registration waited for an ack.
+func TestCommandChannelRegistersUnderItsOwnIdentityWhenReportsAreForbidden(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) {
+		http.Error(w, "missing scope", http.StatusForbidden)
+	})
+	h.run(t)
+
+	h.expectStart(t, "configured-id")
+}
+
+// A running channel was admitted by its token. A later ack naming another
+// identity must not tear it down and build a second client: that reopens the
+// operation receipt store and marks in-flight operations interrupted.
+func TestRunningCommandChannelIsNeverReplacedByALaterIdentity(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) {
+		if report == 1 {
+			writeAck(w, `{"success":true,"agentId":"server-resolved"}`)
+			return
+		}
+		writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`)
+	})
+	h.run(t)
+
+	h.expectStart(t, "configured-id")
+	h.waitForReports(t, 3)
+	h.expectNoRestart(t)
+	if got := h.clientIdentity(); got != "configured-id" {
+		t.Fatalf("running command client identity = %q, want it left on %q", got, "configured-id")
+	}
+	h.mu.Lock()
+	built := h.built
+	h.mu.Unlock()
+	if built != 1 {
+		t.Fatalf("command clients built = %d, want only the one New staged", built)
+	}
+}
+
+// Observer destinations may run any Pulse version and never speak for this
+// agent's identity: their acknowledgements neither name the command identity
+// nor release the channel.
+func TestObserverAckNeverReleasesOrNamesTheCommandChannel(t *testing.T) {
+	server := newServerVersionAckServer(`{"success":true,"agentId":"observer-id","commandAgentId":"observer-id"}`)
+	defer server.Close()
+
+	h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) {
+		writeAck(w, `{"success":true,"agentId":"server-resolved"}`)
+	})
+	runCtx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	h.agent.commandClientMu.Lock()
+	h.agent.commandClientRunCtx = runCtx
+	h.agent.commandClientMu.Unlock()
+
+	report := agentshost.Report{
+		Agent: agentshost.AgentInfo{ID: "configured-id"},
+		Host:  agentshost.HostInfo{Hostname: "test-host"},
+	}
+	if err := h.agent.sendReportToDestination(context.Background(), report, server.URL, "observer-token", server.Client(), false); err != nil {
+		t.Fatalf("sendReportToDestination: %v", err)
+	}
+	h.agent.commandClientMu.Lock()
+	released := h.agent.commandClientParentCtx != nil
+	identity := h.agent.commandIdentity
+	h.agent.commandClientMu.Unlock()
+	if released || identity != "" {
+		t.Fatalf("observer ack released=%v identity=%q, want neither", released, identity)
+	}
+	h.expectNoStart(t, 100*time.Millisecond)
+}
+
+// A remote disable can close and clear a client another path is about to
+// start. Starting that superseded client must fail without occupying the run
+// slot, or the next enabled client never runs.
+func TestSupersededCommandClientCannotOccupyTheRunSlot(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) {
+		writeAck(w, `{"success":true,"agentId":"server-resolved"}`)
+	})
+	runCtx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	h.agent.commandClientMu.Lock()
+	h.agent.commandClientParentCtx = runCtx
+	superseded := h.agent.commandClient
+	h.agent.commandClientMu.Unlock()
+
+	h.agent.applyRemoteConfig(false)
+	if h.agent.startCommandClient(superseded) {
+		t.Fatal("a cleared command client was started")
+	}
+	h.agent.commandClientMu.Lock()
+	occupied := h.agent.commandClientRunCancel != nil
+	h.agent.commandClientMu.Unlock()
+	if occupied {
+		t.Fatal("a cleared command client occupied the run slot")
+	}
+	h.expectNoStart(t, 100*time.Millisecond)
+
+	h.agent.applyRemoteConfig(true)
+	h.expectStart(t, "configured-id")
+}
+
+// A client whose registration Pulse keeps refusing (for example one released
+// by a mock-mode acknowledgement that named nothing, for a token bound to the
+// resolved identity) has accepted no operation. When a later acknowledgement
+// names the identity the token admits, it is retired and rebuilt under that
+// identity instead of retrying the refused one for the life of the process.
+func TestRefusedCommandChannelMovesToTheIdentityALaterAckNames(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) {
+		if report == 1 {
+			writeAck(w, `{"success":true,"agentId":"configured-id"}`)
+			return
+		}
+		writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`)
+	})
+	h.refuse = true
+	h.run(t)
+
+	h.expectStart(t, "configured-id")
+	h.expectStart(t, "server-resolved")
+	if got := h.clientIdentity(); got != "server-resolved" {
+		t.Fatalf("command client identity = %q, want %q", got, "server-resolved")
+	}
+}
+
+// A decodable body that is not a Pulse acknowledgement (an empty object, or a
+// refusal) is no answer either: the channel keeps waiting for a real ack.
+func TestCommandChannelIgnoresAResponseThatIsNotAnAck(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) {
+		switch report {
+		case 1:
+			writeAck(w, `{}`)
+		case 2:
+			writeAck(w, `{"success":false}`)
+		default:
+			writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`)
+		}
+	})
+	h.run(t)
+
+	h.expectStart(t, "server-resolved")
+}
+
+// A report buffered by an earlier run may present another agent ID or
+// hostname, and Pulse judges the command identity for the report it
+// acknowledges. Its acknowledgement must neither name nor release this
+// process's channel.
+func TestReplayedReportAckNeverNegotiatesTheCurrentChannel(t *testing.T) {
+	h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) {
+		writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`)
+	})
+	runCtx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	h.agent.commandClientMu.Lock()
+	h.agent.commandClientRunCtx = runCtx
+	h.agent.commandClientMu.Unlock()
+
+	for _, replayed := range []agentshost.Report{
+		{Agent: agentshost.AgentInfo{ID: "earlier-id"}, Host: agentshost.HostInfo{Hostname: h.agent.hostname}},
+		{Agent: agentshost.AgentInfo{ID: "configured-id"}, Host: agentshost.HostInfo{Hostname: "earlier-hostname"}},
+	} {
+		if err := h.agent.sendReport(context.Background(), replayed); err != nil {
+			t.Fatalf("sendReport: %v", err)
+		}
+	}
+	h.agent.commandClientMu.Lock()
+	released := h.agent.commandClientParentCtx != nil
+	identity := h.agent.commandIdentity
+	h.agent.commandClientMu.Unlock()
+	if released || identity != "" {
+		t.Fatalf("replayed acks released=%v identity=%q, want neither", released, identity)
+	}
+	h.expectNoStart(t, 100*time.Millisecond)
+
+	current := agentshost.Report{Agent: agentshost.AgentInfo{ID: "configured-id"}, Host: agentshost.HostInfo{Hostname: h.agent.hostname}}
+	if err := h.agent.sendReport(context.Background(), current); err != nil {
+		t.Fatalf("sendReport: %v", err)
+	}
+	h.expectStart(t, "server-resolved")
+}
+
+// Pulse admits a client's registration before sending it anything. A client
+// it admitted can no longer be retired, and one the agent retired can never be
+// admitted afterwards.
+func TestCommandClientAdmissionAndRetirementAreExclusive(t *testing.T) {
+	admitted := &CommandClient{}
+	if !admitted.markAdmitted() || !admitted.markAdmitted() {
+		t.Fatal("an admitted client did not stay admitted across reconnects")
+	}
+	if admitted.retireIfNeverAdmitted() {
+		t.Fatal("an admitted client was retired")
+	}
+	retired := &CommandClient{}
+	if !retired.retireIfNeverAdmitted() {
+		t.Fatal("a never-admitted client could not be retired")
+	}
+	if retired.markAdmitted() {
+		t.Fatal("a retired client was admitted")
+	}
+}
+
+// The admission state is set by the real registration handshake: a client
+// Pulse accepts becomes admitted, and a client the agent retired before that
+// stops right after registering instead of handling anything.
+func TestCommandClientRegistrationAdmitsOnlyAnUnretiredClient(t *testing.T) {
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		var registration wsMessage
+		if err := conn.ReadJSON(&registration); err != nil {
+			return
+		}
+		payload, _ := json.Marshal(registeredPayload{Success: true, Message: "Registered"})
+		if err := conn.WriteJSON(wsMessage{Type: msgTypeRegistered, Timestamp: time.Now(), Payload: payload}); err != nil {
+			return
+		}
+		_, _, _ = conn.ReadMessage()
+	}))
+	defer server.Close()
+
+	newClient := func() *CommandClient {
+		return &CommandClient{
+			pulseURL: strings.TrimRight(server.URL, "/"),
+			apiToken: "token",
+			agentID:  "agent-1",
+			hostname: "host-1",
+			platform: "linux",
+			version:  "1.2.3",
+			logger:   zerolog.Nop(),
+			done:     make(chan struct{}),
+		}
+	}
+
+	admitted := newClient()
+	ctx, cancel := context.WithCancel(context.Background())
+	errCh := make(chan error, 1)
+	go func() { errCh <- admitted.connectAndHandle(ctx) }()
+	deadline := time.Now().Add(5 * time.Second)
+	for admitted.admission.Load() != commandClientAdmitted {
+		if time.Now().After(deadline) {
+			t.Fatal("an accepted registration did not admit the client")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	<-errCh
+
+	retired := newClient()
+	if !retired.retireIfNeverAdmitted() {
+		t.Fatal("precondition: a fresh client could not be retired")
+	}
+	done := make(chan error, 1)
+	go func() { done <- retired.connectAndHandle(context.Background()) }()
+	select {
+	case err := <-done:
+		if !errors.Is(err, errCommandClientSuperseded) {
+			t.Fatalf("retired client connectAndHandle = %v, want errCommandClientSuperseded", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("a retired client kept running after its registration was accepted")
+	}
+}

internal/hostagent/commands.go

@@ -16,6 +16,7 @@ import (
 	"runtime"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -146,6 +147,12 @@ type CommandClient struct {
 	healthCapabilities    []string
 	actionActivationNonce string
 	actionHealthWriter    func(bool) error
+	// admission is commandClientPending until Pulse first accepts this
+	// client's registration (commandClientAdmitted), or until the agent
+	// retires it unadmitted under a superseded identity
+	// (commandClientSuperseded). Only an unadmitted client may be retired:
+	// operations arrive only after registration, so it has accepted none.
+	admission atomic.Int32
 }
 
 const maxCancellableRequestsPerConnection = 256
@@ -452,6 +459,9 @@ func (c *CommandClient) connectAndHandle(ctx context.Context) error {
 	if err := c.waitForRegistration(conn); err != nil {
 		return fmt.Errorf("registration failed: %w", err)
 	}
+	if !c.markAdmitted() {
+		return errCommandClientSuperseded
+	}
 	if c.actionRunnerOnly {
 		if err := c.persistActionRunnerHealth(false); err != nil {
 			return fmt.Errorf("write action-runner health: %w", err)
@@ -600,6 +610,31 @@ func (c *CommandClient) waitForRegistration(conn *websocket.Conn) error {
 	return nil
 }
 
+const (
+	commandClientPending int32 = iota
+	commandClientAdmitted
+	commandClientSuperseded
+)
+
+var errCommandClientSuperseded = errors.New("command client was retired under a superseded identity before Pulse admitted it")
+
+// markAdmitted records that Pulse accepted this client's registration. It
+// reports false for a client the agent already retired, which must then
+// handle nothing.
+func (c *CommandClient) markAdmitted() bool {
+	if c.admission.CompareAndSwap(commandClientPending, commandClientAdmitted) {
+		return true
+	}
+	return c.admission.Load() == commandClientAdmitted
+}
+
+// retireIfNeverAdmitted retires a client Pulse has never admitted, so a later
+// registration success cannot let it handle anything. It reports false once
+// the client has been admitted.
+func (c *CommandClient) retireIfNeverAdmitted() bool {
+	return c.admission.CompareAndSwap(commandClientPending, commandClientSuperseded)
+}
+
 func (c *CommandClient) pingLoop(ctx context.Context, conn *websocket.Conn, done chan struct{}) {
 	ticker := time.NewTicker(30 * time.Second)
 	defer ticker.Stop()