Fence runner activation session authority

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

@@ -201,14 +201,37 @@ rotation is a prepare/commit transaction. Issuance stores a ten-minute pending
 replacement beside the active predecessor; the pending transport may register
 but is staged in a separate bounded session slot and is not dispatchable.
 Pending reconnect or flood traffic may replace only that pending slot; the
-active predecessor remains connected and dispatchable until commit. The runner
+active predecessor remains connected and dispatchable until activation begins.
+Activation then installs a reversible per-host session fence before changing
+credential memory or durable storage, so neither new dispatch nor an inbound
+result from either transport is authoritative during persistence. A failed
+save removes the fence only after restoring the predecessor inventory; a
+successful save atomically promotes the exact prepared transport before the
+fence is removed. If exact promotion cannot commit, compensating persistence
+must restore the predecessor inventory before its authority is unfenced, while
+an indeterminate compensating save removes and closes both transports.
+The runner
+may exchange liveness ping/pong traffic while prepared, but pending map
+membership never authorizes inbound results, operation-query responses, or
+deployment progress. Every authority-bearing inbound delivery is bound to
+pointer-identical membership in the active session map while the server lock
+also protects the pending request/subscription lookup and channel send. Typed
+requests, durable-operation queries, and their result channels additionally
+carry an immutable per-WebSocket authority generation, so promotion cannot
+transfer predecessor work to the replacement even though both transports have
+the same canonical host identity. This same rule rejects a displaced
+predecessor immediately after the atomic map swap, even before its deferred
+socket cleanup completes; immutable admission metadata is not used as live
+transport authority.
+The runner
 first durably replaces a private pending
 health marker carrying the current installer-generated activation nonce, then
 calls the authenticated activation method. Activation atomically removes the
 exact predecessor set, removes the replacement expiry/pending state, and
 promotes the exact registered pending session while the token inventory lock
-serializes both decisions. The server persists the activated inventory before
-the bounded session-map swap and closes the displaced predecessor transport
+serializes both decisions. The server fences the session maps before persisting
+the activated inventory, completes the bounded session-map swap before
+unfencing, and closes the displaced predecessor transport
 only after both locks are released. If the exact pending transport vanished or
 was replaced, the server durably restores the prior inventory and returns
 conflict; if that compensating persistence fails, memory remains aligned with

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

@@ -63,12 +63,20 @@ commit. It requires the current `agent:exec` bearer, exact organization,
 canonical agent/hostname binding, and an exact registered pending transport.
 A pending transport is visible to this commit proof but unavailable to action
 dispatch and cannot evict or interrupt the active predecessor. The commit
-requires durable token persistence, clears the pending expiry, removes only the
-server-recorded predecessor IDs, and atomically promotes the exact replacement
-session while invalidating exact stale sessions without allowing a
-caller-selected token ID. Persistence or promotion failure preserves one
-coherent durable inventory/session outcome; the route never reports success for
-an in-memory-only activation or a vanished pending connection. Repeating
+requires durable token persistence and begins by installing a reversible
+per-host session fence, which temporarily rejects new dispatch plus inbound
+results from both predecessor and replacement while credential storage is in
+flight. It then clears the pending expiry, removes only the server-recorded
+predecessor IDs, and atomically promotes the exact replacement session before
+unfencing. Pending requests and durable-operation queries carry an immutable
+per-WebSocket authority generation, so promotion cannot transfer in-flight
+predecessor work to the replacement. A failed durable save restores the prior
+inventory before rolling back the fence; failed exact promotion performs a
+compensating durable save, and an indeterminate compensation fail-closes both
+session slots. Socket cleanup occurs only after the credential and session
+locks are released. The route never reports success for an in-memory-only
+activation or a vanished pending connection and never allows a caller-selected
+token ID. Repeating
 activation for an already active exact session is idempotent so a lost HTTP
 response can be reconciled safely.
 

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

@@ -109,7 +109,16 @@ rejected before any action credential is minted. The pending credential may auth
 exact runner transport in a separate bounded pending slot but cannot become
 dispatch authority. Pending reconnects replace only that slot; they cannot
 close or displace the active predecessor, which remains the dispatch target
-until commit. After the runner
+until activation begins. Activation first fences the host's session authority
+under the execution-server lock, so neither transport can dispatch or satisfy
+inbound work while the credential inventory is being changed. Every typed
+request and durable-operation query is also bound to an immutable per-WebSocket
+authority generation; a promoted replacement cannot complete work sent to the
+predecessor even though both share the same host identity. A failed initial
+save restores the token inventory before removing the fence, successful
+persistence commits the exact pending map swap before unfencing, and an
+indeterminate compensating save removes both active and pending transports.
+After the runner
 durably records its current activation nonce as pending, its authenticated
 activation request durably activates the replacement and revokes the
 server-recorded predecessor set, then atomically promotes the exact pending
@@ -119,7 +128,8 @@ commit. If the pending transport vanished or was superseded, a compensating
 durable save restores both sides of the transition and returns conflict. If the
 compensating save itself fails, memory remains aligned with the last known
 durable activated inventory and the response is indeterminate rather than a
-false success. An initial persistence failure also restores both sides;
+false success. Session cleanup and socket I/O occur only after both locks are
+released. An initial persistence failure also restores both sides;
 an unactivated replacement expires without revoking the predecessor.
 Installer recovery stops the replacement and may restore a predecessor only
 after a bodyless self-cancellation durably removes the exact pending credential

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

@@ -49,9 +49,14 @@ delete, restore, or reclassify storage/recovery state.
 Runner credential rotation follows the shared two-phase token-inventory commit
 boundary: issuance durably prepares a bounded, non-dispatchable replacement
 without removing or disconnecting the active predecessor; activation after
-durable runner health proof requires durable token persistence and atomically
-promotes the exact registered replacement while removing only its recorded
-predecessor set. The installer may restore predecessor files only after the
+durable runner health proof fences both runner transports before credential
+persistence, atomically promotes the exact registered replacement before
+unfencing, and removes only its recorded predecessor set. In-flight typed work
+remains bound to the predecessor's immutable WebSocket generation and cannot be
+reinterpreted as replacement or storage-recovery evidence after promotion. A
+failed save restores predecessor credential/session authority only after the
+durable inventory is restored; indeterminate compensation fail-closes both
+transports. The installer may restore predecessor files only after the
 server atomically cancels and durably removes the exact still-pending bearer;
 activation-winning, persistence-failure, and indeterminate outcomes retain the
 replacement instead. Failed persistence or session promotion preserves one

internal/agentexec/server.go

@@ -65,6 +65,7 @@ type Server struct {
 	mu                                 sync.RWMutex
 	agents                             map[string]*agentConn                           // organizationID + agentID -> connection
 	pendingActionRunners               map[string]*agentConn                           // organizationID + agentID -> exact prepared runner transport
+	actionRunnerPromotionFences        map[string]*ActionRunnerSessionPromotion        // organizationID + agentID -> activation transaction fencing dispatch/results
 	pendingReqs                        map[string]chan CommandResultPayload            // scoped request key -> response channel
 	pendingHostStorageCleanups         map[string]chan HostStorageCleanupResultPayload // scoped request key -> typed storage-cleanup response
 	pendingHostUpdates                 map[string]chan HostUpdateResultPayload         // scoped request key -> typed host-update response
@@ -133,6 +134,7 @@ type agentConn struct {
 	agent            ConnectedAgent
 	admission        AgentAdmission
 	sessionKey       string
+	authorityKey     string
 	approvalGrantKey []byte
 	writeMu          sync.Mutex
 	done             chan struct{}
@@ -151,6 +153,19 @@ type pendingOperationQuery struct {
 	ch       chan operationreceipt.QueryResult
 }
 
+// ActionRunnerSessionPromotion is a reversible in-memory activation
+// transaction. Begin fences both dispatch and inbound-result authority before
+// credential persistence; Commit swaps the exact prepared transport into the
+// active map, while Rollback restores predecessor authority after a failed
+// durable write. Socket cleanup is deliberately deferred until Cleanup runs
+// after config.Mu has been released.
+type ActionRunnerSessionPromotion struct {
+	server  *Server
+	key     string
+	pending *agentConn
+	cleanup []*agentConn
+}
+
 func (ac *agentConn) signalDone() {
 	ac.doneOnce.Do(func() {
 		defer func() {
@@ -200,6 +215,7 @@ func NewServerWithAdmissionValidator(admit AgentRegistrationValidator, validateS
 	return &Server{
 		agents:                             make(map[string]*agentConn),
 		pendingActionRunners:               make(map[string]*agentConn),
+		actionRunnerPromotionFences:        make(map[string]*ActionRunnerSessionPromotion),
 		actionRunnerAdmissionTombstones:    make(map[string]time.Time),
 		pendingReqs:                        make(map[string]chan CommandResultPayload),
 		pendingHostStorageCleanups:         make(map[string]chan HostStorageCleanupResultPayload),
@@ -355,6 +371,32 @@ func connectionSessionKey(ac *agentConn) string {
 	return agentSessionKey(ac.admission.OrganizationID, ac.agent.AgentID)
 }
 
+// connectionAuthorityKey identifies one admitted WebSocket generation. Host
+// identity is intentionally insufficient for request correlation: an action-
+// runner replacement must not inherit work dispatched to its predecessor.
+func connectionAuthorityKey(ac *agentConn) string {
+	if ac == nil {
+		return ""
+	}
+	sessionKey := connectionSessionKey(ac)
+	if strings.TrimSpace(ac.authorityKey) == "" {
+		return sessionKey
+	}
+	return sessionKey + "\x00" + ac.authorityKey
+}
+
+// activeConnectionLocked reports whether ac is the exact transport currently
+// authorized to satisfy inbound work for its tenant and host. The caller must
+// hold s.mu for reading or writing across this check and the corresponding
+// channel delivery so action-runner promotion cannot create a check/send race.
+func (s *Server) activeConnectionLocked(ac *agentConn) bool {
+	if s == nil || ac == nil {
+		return false
+	}
+	key := connectionSessionKey(ac)
+	return s.actionRunnerPromotionFences[key] == nil && s.agents[key] == ac
+}
+
 // SetCommandAuthorizationVerifier installs the server-owned authorization
 // consumer used for approval-gated arbitrary commands.
 func (s *Server) SetCommandAuthorizationVerifier(verifier func(CommandAuthorizationRequest) error) {
@@ -401,8 +443,8 @@ func (s *Server) isShuttingDown() bool {
 	}
 }
 
-func pendingRequestKey(agentID, requestID string) string {
-	return agentID + "\x00" + requestID
+func pendingRequestKey(authorityKey, requestID string) string {
+	return authorityKey + "\x00" + requestID
 }
 
 func (s *Server) connectionForOrganization(organizationID, agentID string) (*agentConn, bool) {
@@ -412,8 +454,9 @@ func (s *Server) connectionForOrganization(organizationID, agentID string) (*age
 	key := agentSessionKey(organizationID, agentID)
 	s.mu.RLock()
 	ac, ok := s.agents[key]
+	fenced := s.actionRunnerPromotionFences[key] != nil
 	s.mu.RUnlock()
-	if !ok {
+	if !ok || fenced {
 		return nil, false
 	}
 	// Prepared action runners live only in pendingActionRunners. Membership in
@@ -452,36 +495,134 @@ func (s *Server) HasActionRunnerSession(admission AgentAdmission) bool {
 	return ok && current != nil && admission.ActivationPending && sameActionRunnerAdmission(current.admission, admission)
 }
 
-// PromoteActionRunnerSessionForCommit performs only the bounded map mutation
-// needed by the credential transaction. Callers may invoke it while holding
-// config.Mu; they must run the returned cleanup only after releasing that lock.
-// This preserves the sole nested order config.Mu -> Server.mu and keeps socket
-// close/logging I/O outside both locks.
-func (s *Server) PromoteActionRunnerSessionForCommit(admission AgentAdmission) (func(), bool) {
+// BeginActionRunnerSessionPromotion fences the host's active map entry before
+// the credential inventory can be durably changed. The fence makes both new
+// dispatch and inbound-result delivery fail closed while persistence is in
+// progress. Callers must resolve the returned transaction with Commit,
+// Rollback, or FailClosed.
+func (s *Server) BeginActionRunnerSessionPromotion(admission AgentAdmission) (*ActionRunnerSessionPromotion, bool) {
 	if s == nil {
 		return nil, false
 	}
 	key := agentSessionKey(admission.OrganizationID, admission.AgentID)
 	s.mu.Lock()
-	pending, ok := s.pendingActionRunners[key]
-	if !ok || pending == nil || !admission.ActivationPending || !sameActionRunnerAdmission(pending.admission, admission) {
-		s.mu.Unlock()
+	defer s.mu.Unlock()
+	if s.actionRunnerPromotionFences[key] != nil {
 		return nil, false
 	}
-	delete(s.pendingActionRunners, key)
-	displaced := s.agents[key]
-	s.agents[key] = pending
+	pending, ok := s.pendingActionRunners[key]
+	if !ok || pending == nil || !admission.ActivationPending || !sameActionRunnerAdmission(pending.admission, admission) {
+		return nil, false
+	}
+	tx := &ActionRunnerSessionPromotion{server: s, key: key, pending: pending}
+	s.actionRunnerPromotionFences[key] = tx
+	return tx, true
+}
+
+// Commit atomically promotes the exact transport captured by Begin. It
+// returns false if that prepared socket disconnected or was replaced while
+// persistence was in progress; the fence remains in place until Rollback or
+// FailClosed resolves the transaction.
+func (tx *ActionRunnerSessionPromotion) Commit() bool {
+	if tx == nil || tx.server == nil {
+		return false
+	}
+	s := tx.server
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.actionRunnerPromotionFences[tx.key] != tx || s.pendingActionRunners[tx.key] != tx.pending {
+		return false
+	}
+	delete(s.pendingActionRunners, tx.key)
+	displaced := s.agents[tx.key]
+	s.agents[tx.key] = tx.pending
+	delete(s.actionRunnerPromotionFences, tx.key)
+	if displaced != nil && displaced != tx.pending {
+		tx.cleanup = append(tx.cleanup, displaced)
+	}
+	return true
+}
+
+// Rollback removes this transaction's fence without changing either session
+// map. It is valid only after the prior credential inventory is known durable.
+func (tx *ActionRunnerSessionPromotion) Rollback() {
+	if tx == nil || tx.server == nil {
+		return
+	}
+	s := tx.server
+	s.mu.Lock()
+	if s.actionRunnerPromotionFences[tx.key] != tx {
+		s.mu.Unlock()
+		return
+	}
+	delete(s.actionRunnerPromotionFences, tx.key)
 	s.mu.Unlock()
-	var cleanup func()
-	if displaced != nil && displaced != pending {
-		cleanup = func() {
-			displaced.signalDone()
-			if displaced.conn != nil {
-				_ = displaced.conn.Close()
-			}
+}
+
+// FailClosed resolves an indeterminate durable activation by removing every
+// active or prepared transport for this host. Neither the potentially revoked
+// predecessor nor an uncommitted replacement may retain runtime authority.
+func (tx *ActionRunnerSessionPromotion) FailClosed() {
+	if tx == nil || tx.server == nil {
+		return
+	}
+	s := tx.server
+	s.mu.Lock()
+	if s.actionRunnerPromotionFences[tx.key] != tx {
+		s.mu.Unlock()
+		return
+	}
+	delete(s.actionRunnerPromotionFences, tx.key)
+	if active := s.agents[tx.key]; active != nil {
+		delete(s.agents, tx.key)
+		tx.cleanup = append(tx.cleanup, active)
+	}
+	if pending := s.pendingActionRunners[tx.key]; pending != nil {
+		delete(s.pendingActionRunners, tx.key)
+		tx.cleanup = append(tx.cleanup, pending)
+	}
+	s.mu.Unlock()
+}
+
+// Cleanup closes transports displaced by Commit or removed by FailClosed. It
+// must run after the caller releases config.Mu so socket I/O never occurs
+// while either the credential or session-map transaction lock is held.
+func (tx *ActionRunnerSessionPromotion) Cleanup() {
+	if tx == nil {
+		return
+	}
+	seen := make(map[*agentConn]struct{}, len(tx.cleanup))
+	for _, ac := range tx.cleanup {
+		if ac == nil {
+			continue
+		}
+		if _, duplicate := seen[ac]; duplicate {
+			continue
+		}
+		seen[ac] = struct{}{}
+		ac.signalDone()
+		if ac.conn != nil {
+			_ = ac.conn.Close()
 		}
 	}
-	return cleanup, true
+}
+
+// PromoteActionRunnerSessionForCommit is the bounded compatibility wrapper
+// used by direct callers and tests. Production credential activation uses the
+// full Begin/Commit/Rollback transaction so the persistence interval is fenced.
+func (s *Server) PromoteActionRunnerSessionForCommit(admission AgentAdmission) (func(), bool) {
+	tx, ok := s.BeginActionRunnerSessionPromotion(admission)
+	if !ok {
+		return nil, false
+	}
+	if !tx.Commit() {
+		tx.Rollback()
+		return nil, false
+	}
+	if len(tx.cleanup) == 0 {
+		return nil, true
+	}
+	return tx.Cleanup, true
 }
 
 // PromoteActionRunnerSession is the non-transactional compatibility wrapper.
@@ -602,8 +743,8 @@ func (s *Server) connectionForContext(ctx context.Context, agentID string) (*age
 	return s.connectionForOrganization(organizationIDFromContext(ctx), agentID)
 }
 
-func (s *Server) claimPendingHostOperation(agentID, requestID, actionID, operation string) (string, error) {
-	key := pendingRequestKey(agentID, requestID)
+func (s *Server) claimPendingHostOperation(authorityKey, requestID, actionID, operation string) (string, error) {
+	key := pendingRequestKey(authorityKey, requestID)
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if _, exists := s.pendingHostOperations[key]; exists {
@@ -616,8 +757,8 @@ func (s *Server) claimPendingHostOperation(agentID, requestID, actionID, operati
 	return key, nil
 }
 
-func (s *Server) matchesPendingHostOperation(agentID, requestID, actionID, operation string) bool {
-	key := pendingRequestKey(agentID, requestID)
+func (s *Server) matchesPendingHostOperation(authorityKey, requestID, actionID, operation string) bool {
+	key := pendingRequestKey(authorityKey, requestID)
 	s.mu.RLock()
 	expected, ok := s.pendingHostOperations[key]
 	s.mu.RUnlock()
@@ -628,12 +769,12 @@ func (s *Server) claimPendingDockerOperation(identity operationreceipt.Identity,
 	return s.claimPendingDockerOperationForSession(identity.AgentID, identity, containerID)
 }
 
-func (s *Server) claimPendingDockerOperationForSession(sessionKey string, identity operationreceipt.Identity, containerID string) (string, error) {
+func (s *Server) claimPendingDockerOperationForSession(authorityKey string, identity operationreceipt.Identity, containerID string) (string, error) {
 	identity, err := operationreceipt.NormalizeIdentity(identity)
 	if err != nil {
 		return "", err
 	}
-	key := pendingRequestKey(sessionKey, identity.AttemptID)
+	key := pendingRequestKey(authorityKey, identity.AttemptID)
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if _, exists := s.pendingHostOperations[key]; exists {
@@ -647,8 +788,8 @@ func (s *Server) matchesPendingDockerOperation(agentID string, result DockerCont
 	return s.matchesPendingDockerOperationForSession(agentID, agentID, result)
 }
 
-func (s *Server) matchesPendingDockerOperationForSession(sessionKey, agentID string, result DockerContainerLifecycleResultPayload) bool {
-	key := pendingRequestKey(sessionKey, result.RequestID)
+func (s *Server) matchesPendingDockerOperationForSession(authorityKey, agentID string, result DockerContainerLifecycleResultPayload) bool {
+	key := pendingRequestKey(authorityKey, result.RequestID)
 	s.mu.RLock()
 	expected, ok := s.pendingHostOperations[key]
 	s.mu.RUnlock()
@@ -660,8 +801,8 @@ func (s *Server) matchesPendingDockerUpdateOperation(agentID string, result Dock
 	return s.matchesPendingDockerUpdateOperationForSession(agentID, agentID, result)
 }
 
-func (s *Server) matchesPendingDockerUpdateOperationForSession(sessionKey, agentID string, result DockerContainerUpdateResultPayload) bool {
-	key := pendingRequestKey(sessionKey, result.RequestID)
+func (s *Server) matchesPendingDockerUpdateOperationForSession(authorityKey, agentID string, result DockerContainerUpdateResultPayload) bool {
+	key := pendingRequestKey(authorityKey, result.RequestID)
 	s.mu.RLock()
 	expected, ok := s.pendingHostOperations[key]
 	s.mu.RUnlock()
@@ -669,8 +810,8 @@ func (s *Server) matchesPendingDockerUpdateOperationForSession(sessionKey, agent
 	return ok && expected.identity == actual && expected.subjectID == strings.ToLower(strings.TrimSpace(result.ContainerID))
 }
 
-func (s *Server) matchesPendingProxmoxGuestOperationForSession(sessionKey, agentID string, result ProxmoxGuestLifecycleResultPayload) bool {
-	key := pendingRequestKey(sessionKey, result.RequestID)
+func (s *Server) matchesPendingProxmoxGuestOperationForSession(authorityKey, agentID string, result ProxmoxGuestLifecycleResultPayload) bool {
+	key := pendingRequestKey(authorityKey, result.RequestID)
 	s.mu.RLock()
 	expected, ok := s.pendingHostOperations[key]
 	s.mu.RUnlock()
@@ -1237,6 +1378,7 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
 		},
 		admission:        admission,
 		sessionKey:       agentSessionKey(admission.OrganizationID, admission.AgentID),
+		authorityKey:     uuid.NewString(),
 		approvalGrantKey: DeriveApprovalGrantKey(reg.Token),
 		done:             make(chan struct{}),
 	}
@@ -1466,6 +1608,24 @@ func (s *Server) readLoop(ac *agentConn) {
 			continue
 		}
 
+		// A prepared action runner may prove liveness while activation is
+		// pending, but it has no authority to satisfy work dispatched to the
+		// active predecessor. Recheck exact active-map membership in each result
+		// handler as well, under the same lock as delivery, because promotion can
+		// race this early rejection boundary.
+		if msg.Type != MsgTypeAgentPing {
+			s.mu.RLock()
+			active := s.activeConnectionLocked(ac)
+			s.mu.RUnlock()
+			if !active {
+				log.Warn().
+					Str("agent_id", ac.agent.AgentID).
+					Str("message_type", string(msg.Type)).
+					Msg("Dropping inbound message from non-active agent session")
+				continue
+			}
+		}
+
 		switch msg.Type {
 		case MsgTypeAgentPing:
 			pongMsg, err := NewMessage(MsgTypePong, "", nil)
@@ -1499,10 +1659,9 @@ func (s *Server) readLoop(ac *agentConn) {
 			}
 
 			s.mu.RLock()
-			ch, ok := s.pendingReqs[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
-			s.mu.RUnlock()
-
-			if ok {
+			ch, ok := s.pendingReqs[pendingRequestKey(connectionAuthorityKey(ac), result.RequestID)]
+			active := s.activeConnectionLocked(ac)
+			if active && ok {
 				select {
 				case ch <- result:
 					log.Debug().
@@ -1518,7 +1677,9 @@ func (s *Server) readLoop(ac *agentConn) {
 						Str("request_id", result.RequestID).
 						Msg("Result channel full, dropping")
 				}
-			} else {
+			}
+			s.mu.RUnlock()
+			if !active || !ok {
 				log.Warn().
 					Str("agent_id", ac.agent.AgentID).
 					Str("request_id", result.RequestID).
@@ -1531,20 +1692,20 @@ func (s *Server) readLoop(ac *agentConn) {
 				log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid host update result")
 				continue
 			}
-			if !s.matchesPendingHostOperation(connectionSessionKey(ac), result.RequestID, result.ActionID, HostUpdateOperationInstall) {
+			if !s.matchesPendingHostOperation(connectionAuthorityKey(ac), result.RequestID, result.ActionID, HostUpdateOperationInstall) {
 				log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Dropping uncorrelated host update result")
 				continue
 			}
 			s.mu.RLock()
-			ch, ok := s.pendingHostUpdates[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
-			s.mu.RUnlock()
-			if ok {
+			ch, ok := s.pendingHostUpdates[pendingRequestKey(connectionAuthorityKey(ac), result.RequestID)]
+			if s.activeConnectionLocked(ac) && ok {
 				select {
 				case ch <- result:
 				default:
 					log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Host update result channel full, dropping")
 				}
 			}
+			s.mu.RUnlock()
 
 		case MsgTypeActionPreflightResult:
 			result, decodeErr := DecodeActionPreflightResultPayload(msg.Payload)
@@ -1553,15 +1714,15 @@ func (s *Server) readLoop(ac *agentConn) {
 				continue
 			}
 			s.mu.RLock()
-			ch, ok := s.pendingActionPreflights[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
-			s.mu.RUnlock()
-			if ok {
+			ch, ok := s.pendingActionPreflights[pendingRequestKey(connectionAuthorityKey(ac), result.RequestID)]
+			if s.activeConnectionLocked(ac) && ok {
 				select {
 				case ch <- result:
 				default:
 					log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Action preflight result channel full, dropping")
 				}
 			}
+			s.mu.RUnlock()
 
 		case MsgTypeDockerContainerObserveResult:
 			result, decodeErr := DecodeDockerContainerObservationResultPayload(msg.Payload)
@@ -1570,15 +1731,15 @@ func (s *Server) readLoop(ac *agentConn) {
 				continue
 			}
 			s.mu.RLock()
-			ch, ok := s.pendingDockerContainerObservations[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
-			s.mu.RUnlock()
-			if ok {
+			ch, ok := s.pendingDockerContainerObservations[pendingRequestKey(connectionAuthorityKey(ac), result.RequestID)]
+			if s.activeConnectionLocked(ac) && ok {
 				select {
 				case ch <- result:
 				default:
 					log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Docker observation result channel full, dropping")
 				}
 			}
+			s.mu.RUnlock()
 
 		case MsgTypeHostStorageCleanupResult:
 			result, decodeErr := DecodeHostStorageCleanupResultPayload(msg.Payload)
@@ -1586,20 +1747,20 @@ func (s *Server) readLoop(ac *agentConn) {
 				log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid host storage cleanup result")
 				continue
 			}
-			if !s.matchesPendingHostOperation(connectionSessionKey(ac), result.RequestID, result.ActionID, HostStorageCleanupOperationPackageCache) {
+			if !s.matchesPendingHostOperation(connectionAuthorityKey(ac), result.RequestID, result.ActionID, HostStorageCleanupOperationPackageCache) {
 				log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Dropping uncorrelated host storage cleanup result")
 				continue
 			}
 			s.mu.RLock()
-			ch, ok := s.pendingHostStorageCleanups[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
-			s.mu.RUnlock()
-			if ok {
+			ch, ok := s.pendingHostStorageCleanups[pendingRequestKey(connectionAuthorityKey(ac), result.RequestID)]
+			if s.activeConnectionLocked(ac) && ok {
 				select {
 				case ch <- result:
 				default:
 					log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Host storage cleanup result channel full, dropping")
 				}
 			}
+			s.mu.RUnlock()
 
 		case MsgTypeProxmoxGuestLifecycleResult:
 			result, decodeErr := DecodeProxmoxGuestLifecycleResultPayload(msg.Payload)
@@ -1607,20 +1768,20 @@ func (s *Server) readLoop(ac *agentConn) {
 				log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid Proxmox guest lifecycle result")
 				continue
 			}
-			if !s.matchesPendingProxmoxGuestOperationForSession(connectionSessionKey(ac), ac.agent.AgentID, result) {
+			if !s.matchesPendingProxmoxGuestOperationForSession(connectionAuthorityKey(ac), ac.agent.AgentID, result) {
 				log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Dropping uncorrelated Proxmox guest lifecycle result")
 				continue
 			}
 			s.mu.RLock()
-			ch, ok := s.pendingProxmoxGuestLifecycles[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
-			s.mu.RUnlock()
-			if ok {
+			ch, ok := s.pendingProxmoxGuestLifecycles[pendingRequestKey(connectionAuthorityKey(ac), result.RequestID)]
+			if s.activeConnectionLocked(ac) && ok {
 				select {
 				case ch <- result:
 				default:
 					log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Proxmox guest lifecycle result channel full, dropping")
 				}
 			}
+			s.mu.RUnlock()
 
 		case MsgTypeDockerContainerLifecycleResult:
 			result, decodeErr := DecodeDockerContainerLifecycleResultPayload(msg.Payload)
@@ -1628,20 +1789,20 @@ func (s *Server) readLoop(ac *agentConn) {
 				log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid docker container lifecycle result")
 				continue
 			}
-			if !s.matchesPendingDockerOperationForSession(connectionSessionKey(ac), ac.agent.AgentID, result) {
+			if !s.matchesPendingDockerOperationForSession(connectionAuthorityKey(ac), ac.agent.AgentID, result) {
 				log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Dropping uncorrelated docker lifecycle result")
 				continue
 			}
 			s.mu.RLock()
-			ch, ok := s.pendingDockerContainerLifecycles[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
-			s.mu.RUnlock()
-			if ok {
+			ch, ok := s.pendingDockerContainerLifecycles[pendingRequestKey(connectionAuthorityKey(ac), result.RequestID)]
+			if s.activeConnectionLocked(ac) && ok {
 				select {
 				case ch <- result:
 				default:
 					log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Docker lifecycle result channel full, dropping")
 				}
 			}
+			s.mu.RUnlock()
 
 		case MsgTypeDockerContainerUpdateResult:
 			result, decodeErr := DecodeDockerContainerUpdateResultPayload(msg.Payload)
@@ -1649,20 +1810,20 @@ func (s *Server) readLoop(ac *agentConn) {
 				log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid docker container update result")
 				continue
 			}
-			if !s.matchesPendingDockerUpdateOperationForSession(connectionSessionKey(ac), ac.agent.AgentID, result) {
+			if !s.matchesPendingDockerUpdateOperationForSession(connectionAuthorityKey(ac), ac.agent.AgentID, result) {
 				log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Dropping uncorrelated docker update result")
 				continue
 			}
 			s.mu.RLock()
-			ch, ok := s.pendingDockerContainerUpdates[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
-			s.mu.RUnlock()
-			if ok {
+			ch, ok := s.pendingDockerContainerUpdates[pendingRequestKey(connectionAuthorityKey(ac), result.RequestID)]
+			if s.activeConnectionLocked(ac) && ok {
 				select {
 				case ch <- result:
 				default:
 					log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Docker update result channel full, dropping")
 				}
 			}
+			s.mu.RUnlock()
 
 		case MsgTypeOperationQueryResult:
 			result, decodeErr := operationreceipt.DecodeQueryResult(msg.Payload)
@@ -1670,7 +1831,7 @@ func (s *Server) readLoop(ac *agentConn) {
 				log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid operation query result")
 				continue
 			}
-			key := pendingRequestKey(connectionSessionKey(ac), strings.TrimSpace(msg.ID))
+			key := pendingRequestKey(connectionAuthorityKey(ac), strings.TrimSpace(msg.ID))
 			s.mu.RLock()
 			pending, ok := s.pendingOperationQueries[key]
 			s.mu.RUnlock()
@@ -1685,10 +1846,15 @@ func (s *Server) readLoop(ac *agentConn) {
 				log.Warn().Err(err).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid correlated operation query result")
 				continue
 			}
-			select {
-			case pending.ch <- result:
-			default:
+			s.mu.RLock()
+			current, stillPending := s.pendingOperationQueries[key]
+			if s.activeConnectionLocked(ac) && stillPending && current.ch == pending.ch && current.identity == pending.identity {
+				select {
+				case pending.ch <- result:
+				default:
+				}
 			}
+			s.mu.RUnlock()
 		case MsgTypeDeployProgress:
 			var progress DeployProgressPayload
 			if err := msg.DecodePayload(&progress); err != nil {
@@ -1708,7 +1874,8 @@ func (s *Server) readLoop(ac *agentConn) {
 			sent := false
 			s.mu.RLock()
 			ch, ok := s.deploySubs[subKey]
-			if ok {
+			active := s.activeConnectionLocked(ac)
+			if active && ok {
 				select {
 				case ch <- progress:
 					sent = true
@@ -1719,7 +1886,7 @@ func (s *Server) readLoop(ac *agentConn) {
 
 			// Final messages must be delivered — retry with backoff if the
 			// initial non-blocking send failed (channel was full).
-			if ok && !sent && progress.Final {
+			if active && ok && !sent && progress.Final {
 				deadline := time.After(5 * time.Second)
 				ticker := time.NewTicker(50 * time.Millisecond)
 			retryLoop:
@@ -1733,7 +1900,7 @@ func (s *Server) readLoop(ac *agentConn) {
 						// Force-close the subscription so the consumer goroutine
 						// unblocks on channel close and can finalize the job.
 						s.mu.Lock()
-						if closeCh, exists := s.deploySubs[subKey]; exists {
+						if closeCh, exists := s.deploySubs[subKey]; s.activeConnectionLocked(ac) && exists {
 							delete(s.deploySubs, subKey)
 							close(closeCh)
 						}
@@ -1742,7 +1909,8 @@ func (s *Server) readLoop(ac *agentConn) {
 					case <-ticker.C:
 						s.mu.RLock()
 						ch, ok = s.deploySubs[subKey]
-						if !ok {
+						active := s.activeConnectionLocked(ac)
+						if !ok || !active {
 							s.mu.RUnlock()
 							break retryLoop // channel was closed/unsubscribed
 						}
@@ -1757,7 +1925,7 @@ func (s *Server) readLoop(ac *agentConn) {
 					}
 				}
 				ticker.Stop()
-			} else if ok && !sent {
+			} else if active && ok && !sent {
 				log.Warn().
 					Str("agent_id", ac.agent.AgentID).
 					Str("job_id", progress.JobID).
@@ -1950,7 +2118,7 @@ func (s *Server) ExecuteCommand(ctx context.Context, agentID string, cmd Execute
 
 	// Create response channel
 	respCh := make(chan CommandResultPayload, 1)
-	reqKey := pendingRequestKey(connectionSessionKey(ac), cmd.RequestID)
+	reqKey := pendingRequestKey(connectionAuthorityKey(ac), cmd.RequestID)
 	s.mu.Lock()
 	if _, exists := s.pendingReqs[reqKey]; exists {
 		s.mu.Unlock()
@@ -2105,9 +2273,9 @@ func dispatchHostOperation[Req hostOperationPayload, Res any](ctx context.Contex
 	}
 
 	respCh := make(chan Res, 1)
-	sessionKey := connectionSessionKey(ac)
-	reqKey := pendingRequestKey(sessionKey, requestID)
-	hostOperationKey, err := s.claimPendingHostOperation(sessionKey, requestID, actionID, operation)
+	authorityKey := connectionAuthorityKey(ac)
+	reqKey := pendingRequestKey(authorityKey, requestID)
+	hostOperationKey, err := s.claimPendingHostOperation(authorityKey, requestID, actionID, operation)
 	if err != nil {
 		return nil, err
 	}
@@ -2195,7 +2363,7 @@ func (s *Server) PreflightAction(ctx context.Context, agentID string, req Action
 		return nil, fmt.Errorf("agent does not support action preflight protocol")
 	}
 	ch := make(chan ActionPreflightResultPayload, 1)
-	key := pendingRequestKey(connectionSessionKey(ac), req.RequestID)
+	key := pendingRequestKey(connectionAuthorityKey(ac), req.RequestID)
 	s.mu.Lock()
 	if _, exists := s.pendingActionPreflights[key]; exists {
 		s.mu.Unlock()
@@ -2265,7 +2433,7 @@ func (s *Server) ObserveDockerContainer(ctx context.Context, agentID string, req
 		return nil, fmt.Errorf("agent does not support docker observation protocol")
 	}
 	ch := make(chan DockerContainerObservationResultPayload, 1)
-	key := pendingRequestKey(connectionSessionKey(ac), req.RequestID)
+	key := pendingRequestKey(connectionAuthorityKey(ac), req.RequestID)
 	s.mu.Lock()
 	if _, exists := s.pendingDockerContainerObservations[key]; exists {
 		s.mu.Unlock()
@@ -2406,9 +2574,9 @@ func dispatchTypedDockerContainerOperation[Res any](
 	}
 
 	respCh := make(chan Res, 1)
-	sessionKey := connectionSessionKey(ac)
-	reqKey := pendingRequestKey(sessionKey, requestID)
-	hostOperationKey, err := s.claimPendingDockerOperationForSession(sessionKey, identity, containerID)
+	authorityKey := connectionAuthorityKey(ac)
+	reqKey := pendingRequestKey(authorityKey, requestID)
+	hostOperationKey, err := s.claimPendingDockerOperationForSession(authorityKey, identity, containerID)
 	if err != nil {
 		return nil, err
 	}
@@ -2525,7 +2693,7 @@ func (s *Server) QueryAgentOperation(ctx context.Context, agentID string, identi
 		return operationreceipt.QueryResult{}, fmt.Errorf("agent does not support durable operation receipts")
 	}
 	queryID := identity.AttemptID + ".query." + uuid.NewString()
-	key := pendingRequestKey(connectionSessionKey(ac), queryID)
+	key := pendingRequestKey(connectionAuthorityKey(ac), queryID)
 	ch := make(chan operationreceipt.QueryResult, 1)
 	s.mu.Lock()
 	if _, exists := s.pendingOperationQueries[key]; exists {
@@ -2606,7 +2774,7 @@ func (s *Server) ReadFile(ctx context.Context, agentID string, req ReadFilePaylo
 
 	// Create response channel
 	respCh := make(chan CommandResultPayload, 1)
-	reqKey := pendingRequestKey(connectionSessionKey(ac), req.RequestID)
+	reqKey := pendingRequestKey(connectionAuthorityKey(ac), req.RequestID)
 	s.mu.Lock()
 	if _, exists := s.pendingReqs[reqKey]; exists {
 		s.mu.Unlock()

internal/agentexec/server_websocket_test.go

@@ -373,6 +373,170 @@ func TestPreparedActionRunnerReconnectCannotDisplaceActiveDispatchAndExactPromot
 	}
 }
 
+func TestActionRunnerInboundResultsRequireExactActiveSessionAcrossPromotion(t *testing.T) {
+	active := AgentAdmission{
+		OrganizationID: "org-a", TokenID: "active-token", AgentID: "machine-a", Hostname: "node.example",
+		RuntimeRole: RuntimeRoleActionRunner, ActionCapability: ActionCapabilityTypedV1,
+	}
+	pending := active
+	pending.TokenID = "pending-token"
+	pending.ActivationPending = true
+	admissions := map[string]AgentAdmission{active.TokenID: active, pending.TokenID: pending}
+	s := NewServerWithAdmissionValidator(func(token, _, _ string) (AgentAdmission, bool) {
+		admission, ok := admissions[token]
+		return admission, ok
+	}, func(AgentAdmission) bool { return true })
+	ts := newWSServer(t, s)
+	defer ts.Close()
+
+	register := func(admission AgentAdmission) *websocket.Conn {
+		t.Helper()
+		conn, _, err := dialAgentExecWebSocket(t, ts.URL)
+		if err != nil {
+			t.Fatal(err)
+		}
+		wsWriteMessage(t, conn, mustNewMessage(t, MsgTypeAgentRegister, "", AgentRegisterPayload{
+			AgentID: admission.AgentID, Hostname: admission.Hostname, Token: admission.TokenID,
+			RuntimeRole: admission.RuntimeRole, ActionCapability: admission.ActionCapability,
+			OperationReceiptVersion: operationreceipt.ProtocolVersion,
+		}))
+		if ack := wsReadRegisteredPayload(t, conn); !ack.Success {
+			conn.Close()
+			t.Fatalf("registration for %s failed: %s", admission.TokenID, ack.Message)
+		}
+		return conn
+	}
+	activeConn := register(active)
+	defer activeConn.Close()
+	pendingConn := register(pending)
+	defer pendingConn.Close()
+
+	type updateOutcome struct {
+		result *HostUpdateResultPayload
+		err    error
+	}
+	dispatch := func(requestID, actionID, inventoryHash string) <-chan updateOutcome {
+		t.Helper()
+		out := make(chan updateOutcome, 1)
+		go func() {
+			ctx, cancel := context.WithTimeout(WithOrganizationID(context.Background(), active.OrganizationID), 5*time.Second)
+			defer cancel()
+			result, err := s.ExecuteHostUpdate(ctx, active.AgentID, HostUpdatePayload{
+				RequestID: requestID, ActionID: actionID, Operation: HostUpdateOperationInstall,
+				ExpectedInventoryHash: inventoryHash, Timeout: 5,
+			})
+			out <- updateOutcome{result: result, err: err}
+		}()
+		return out
+	}
+	readRequest := func(conn *websocket.Conn) HostUpdatePayload {
+		t.Helper()
+		msg, err := wsReadRawMessageWithTimeout(conn, 2*time.Second)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if msg.Type != MsgTypeHostUpdate || msg.Payload == nil {
+			t.Fatalf("message = %#v, want typed host update", msg)
+		}
+		var req HostUpdatePayload
+		if err := json.Unmarshal(*msg.Payload, &req); err != nil {
+			t.Fatal(err)
+		}
+		return req
+	}
+	resultFor := func(req HostUpdatePayload, afterHash string) HostUpdateResultPayload {
+		now := time.Now().UTC()
+		return HostUpdateResultPayload{
+			RequestID: req.RequestID, ActionID: req.ActionID, Success: true,
+			ExecutionPhase: HostUpdatePhaseComplete,
+			Before:         HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: req.ExpectedInventoryHash, PendingCount: 1, CheckedAt: now.Add(-time.Second)},
+			After:          HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: afterHash, PendingCount: 0, CheckedAt: now},
+			HealthChecked:  true, PackageManagerHealthy: true, Verification: HostUpdateVerificationVerified,
+		}
+	}
+	requirePong := func(conn *websocket.Conn, label string) {
+		t.Helper()
+		wsWriteMessage(t, conn, mustNewMessage(t, MsgTypeAgentPing, "", nil))
+		msg, err := wsReadRawMessageWithTimeout(conn, 2*time.Second)
+		if err != nil {
+			t.Fatalf("%s ping barrier: %v", label, err)
+		}
+		if msg.Type != MsgTypePong {
+			t.Fatalf("%s ping barrier received %q, want pong", label, msg.Type)
+		}
+	}
+	assertStillPending := func(out <-chan updateOutcome, label string) {
+		t.Helper()
+		select {
+		case got := <-out:
+			t.Fatalf("%s satisfied dispatch: result=%#v err=%v", label, got.result, got.err)
+		case <-time.After(100 * time.Millisecond):
+		}
+	}
+	assertSuccess := func(out <-chan updateOutcome, label, expectedAfterHash string) {
+		t.Helper()
+		select {
+		case got := <-out:
+			if got.err != nil || got.result == nil || !got.result.Success || got.result.After.InventoryHash != expectedAfterHash {
+				t.Fatalf("%s result=%#v err=%v", label, got.result, got.err)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatalf("%s did not complete dispatch", label)
+		}
+	}
+	assertDisconnected := func(out <-chan updateOutcome, label string) {
+		t.Helper()
+		select {
+		case got := <-out:
+			if got.result != nil || got.err == nil || !strings.Contains(got.err.Error(), "disconnected") {
+				t.Fatalf("%s result=%#v err=%v, want predecessor disconnect", label, got.result, got.err)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatalf("%s did not terminate after predecessor cleanup", label)
+		}
+	}
+
+	before := dispatch("before-promotion", "action-before", "sha256:"+strings.Repeat("a", 64))
+	beforeReq := readRequest(activeConn)
+	pendingForgery := resultFor(beforeReq, "sha256:"+strings.Repeat("e", 64))
+	wsWriteMessage(t, pendingConn, mustNewMessage(t, MsgTypeHostUpdateResult, beforeReq.RequestID, pendingForgery))
+	requirePong(pendingConn, "prepared replacement after forged result")
+	assertStillPending(before, "prepared replacement")
+
+	failedSaveTx, begun := s.BeginActionRunnerSessionPromotion(pending)
+	if !begun {
+		t.Fatal("failed to begin promotion fence")
+	}
+	fencedForgery := resultFor(beforeReq, "sha256:"+strings.Repeat("f", 64))
+	wsWriteMessage(t, activeConn, mustNewMessage(t, MsgTypeHostUpdateResult, beforeReq.RequestID, fencedForgery))
+	requirePong(activeConn, "predecessor during promotion fence")
+	assertStillPending(before, "fenced predecessor")
+	failedSaveTx.Rollback()
+
+	beforeResult := resultFor(beforeReq, "sha256:"+strings.Repeat("b", 64))
+	wsWriteMessage(t, activeConn, mustNewMessage(t, MsgTypeHostUpdateResult, beforeReq.RequestID, beforeResult))
+	assertSuccess(before, "active predecessor after rollback", beforeResult.After.InventoryHash)
+
+	across := dispatch("across-promotion", "action-across", "sha256:"+strings.Repeat("7", 64))
+	acrossReq := readRequest(activeConn)
+	promotion, begun := s.BeginActionRunnerSessionPromotion(pending)
+	if !begun || !promotion.Commit() {
+		t.Fatal("prepared replacement was not promoted")
+	}
+	transferredForgery := resultFor(acrossReq, "sha256:"+strings.Repeat("8", 64))
+	wsWriteMessage(t, pendingConn, mustNewMessage(t, MsgTypeHostUpdateResult, acrossReq.RequestID, transferredForgery))
+	requirePong(pendingConn, "promoted replacement after predecessor-request forgery")
+	assertStillPending(across, "promoted replacement for predecessor request")
+	promotion.Cleanup()
+	assertDisconnected(across, "predecessor request across promotion")
+
+	after := dispatch("after-promotion", "action-after", "sha256:"+strings.Repeat("c", 64))
+	afterReq := readRequest(pendingConn)
+	afterResult := resultFor(afterReq, "sha256:"+strings.Repeat("d", 64))
+	wsWriteMessage(t, pendingConn, mustNewMessage(t, MsgTypeHostUpdateResult, afterReq.RequestID, afterResult))
+	assertSuccess(after, "promoted replacement", afterResult.After.InventoryHash)
+}
+
 func TestActionRunnerPromotionVersusDisconnectNeverRetainsDeadSession(t *testing.T) {
 	for attempt := 0; attempt < 20; attempt++ {
 		admission := AgentAdmission{
@@ -420,6 +584,41 @@ func TestActionRunnerPromotionVersusDisconnectNeverRetainsDeadSession(t *testing
 	}
 }
 
+func TestStaleActionRunnerPromotionCannotFailCloseLaterSessions(t *testing.T) {
+	admission := AgentAdmission{
+		OrganizationID: "org-a", TokenID: "pending-token", AgentID: "machine-a", Hostname: "node.example",
+		RuntimeRole: RuntimeRoleActionRunner, ActionCapability: ActionCapabilityTypedV1, ActivationPending: true,
+	}
+	s := NewServerWithAdmissionValidator(func(string, string, string) (AgentAdmission, bool) {
+		return AgentAdmission{}, false
+	}, nil)
+	key := agentSessionKey(admission.OrganizationID, admission.AgentID)
+	predecessor := &agentConn{agent: ConnectedAgent{AgentID: admission.AgentID}, sessionKey: key, authorityKey: "predecessor", done: make(chan struct{})}
+	pending := &agentConn{agent: ConnectedAgent{AgentID: admission.AgentID}, admission: admission, sessionKey: key, authorityKey: "pending", done: make(chan struct{})}
+	s.agents[key] = predecessor
+	s.pendingActionRunners[key] = pending
+
+	tx, begun := s.BeginActionRunnerSessionPromotion(admission)
+	if !begun {
+		t.Fatal("failed to begin promotion")
+	}
+	tx.Rollback()
+
+	laterActive := &agentConn{agent: ConnectedAgent{AgentID: admission.AgentID}, sessionKey: key, authorityKey: "later-active", done: make(chan struct{})}
+	laterPending := &agentConn{agent: ConnectedAgent{AgentID: admission.AgentID}, admission: admission, sessionKey: key, authorityKey: "later-pending", done: make(chan struct{})}
+	s.mu.Lock()
+	s.agents[key] = laterActive
+	s.pendingActionRunners[key] = laterPending
+	s.mu.Unlock()
+
+	tx.FailClosed()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	if s.agents[key] != laterActive || s.pendingActionRunners[key] != laterPending {
+		t.Fatal("stale promotion transaction removed later session occupants")
+	}
+}
+
 func TestLegacyCredentialCannotAssertActionRunnerRole(t *testing.T) {
 	s := NewServerWithAdmissionValidator(func(token, _, _ string) (AgentAdmission, bool) {
 		return AgentAdmission{TokenID: token, AgentID: "a1", Hostname: "host1", RuntimeRole: RuntimeRoleLegacyFullTrust}, token == "legacy"

internal/api/action_runner_credentials.go

@@ -175,17 +175,18 @@ func (r *Router) handleActivateActionRunnerCredential(w http.ResponseWriter, req
 		ActionCapability:  agentexec.ActionCapabilityTypedV1,
 		ActivationPending: true,
 	}
-	var promotionCleanup func()
+	var promotion *agentexec.ActionRunnerSessionPromotion
 	_, revoked, changed, err := agenttokens.ActivateActionRunnerAndPersistWithPromotion(
 		r.config, r.persistence, caller.ID, payload.AgentID, payload.Hostname,
-		func() bool {
-			cleanup, promoted := r.agentExecServer.PromoteActionRunnerSessionForCommit(admission)
-			if promoted {
-				promotionCleanup = cleanup
-			}
-			return promoted
+		func() (agenttokens.ActionRunnerPromotionTransaction, bool) {
+			var begun bool
+			promotion, begun = r.agentExecServer.BeginActionRunnerSessionPromotion(admission)
+			return promotion, begun
 		},
 	)
+	if promotion != nil {
+		promotion.Cleanup()
+	}
 	if err != nil {
 		status := http.StatusInternalServerError
 		if errors.Is(err, agenttokens.ErrRecord) {
@@ -197,9 +198,6 @@ func (r *Router) handleActivateActionRunnerCredential(w http.ResponseWriter, req
 		return
 	}
 	if changed {
-		if promotionCleanup != nil {
-			promotionCleanup()
-		}
 		for _, previous := range revoked {
 			r.invalidateActionRunnerRecord(previous)
 		}

internal/api/action_runner_credentials_test.go

@@ -10,6 +10,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -24,6 +25,12 @@ import (
 
 type actionRunnerFailingPersistenceFS struct{}
 
+type actionRunnerBlockingFailingPersistenceFS struct {
+	once    sync.Once
+	entered chan struct{}
+	release chan struct{}
+}
+
 func (actionRunnerFailingPersistenceFS) ReadFile(name string) ([]byte, error) {
 	return os.ReadFile(name)
 }
@@ -41,6 +48,29 @@ func (actionRunnerFailingPersistenceFS) MkdirAll(path string, perm os.FileMode)
 	return os.MkdirAll(path, perm)
 }
 
+func (fs *actionRunnerBlockingFailingPersistenceFS) ReadFile(name string) ([]byte, error) {
+	return os.ReadFile(name)
+}
+func (fs *actionRunnerBlockingFailingPersistenceFS) WriteFile(string, []byte, os.FileMode) error {
+	fs.once.Do(func() {
+		close(fs.entered)
+		<-fs.release
+	})
+	return errors.New("injected blocked action-runner persistence failure")
+}
+func (fs *actionRunnerBlockingFailingPersistenceFS) Rename(oldPath, newPath string) error {
+	return os.Rename(oldPath, newPath)
+}
+func (fs *actionRunnerBlockingFailingPersistenceFS) Remove(name string) error {
+	return os.Remove(name)
+}
+func (fs *actionRunnerBlockingFailingPersistenceFS) Stat(name string) (os.FileInfo, error) {
+	return os.Stat(name)
+}
+func (fs *actionRunnerBlockingFailingPersistenceFS) MkdirAll(path string, perm os.FileMode) error {
+	return os.MkdirAll(path, perm)
+}
+
 func newActionRunnerCredentialTestRouter(t *testing.T) (*Router, *config.Config, string) {
 	t.Helper()
 	cfg := &config.Config{DataPath: t.TempDir(), AuthUser: "admin", AuthPass: "$2a$10$dummy"}
@@ -380,6 +410,120 @@ func TestActivateActionRunnerCredentialPersistenceFailureKeepsBothCredentialsAnd
 	}
 }
 
+func testActionRunnerCredentialFencesPredecessorResultsAcrossPersistence(t *testing.T) {
+	router, _, hostID := newActionRunnerCredentialTestRouter(t)
+	router.agentExecServer = agentexec.NewServerWithAdmissionValidator(router.admitAgentExecToken, router.validateAgentExecSession)
+	first := issueActionRunnerCredentialForTest(t, router, hostID, "host-1.local")
+	commitActionRunnerCredentialForTest(t, router, first)
+	activeConn, activeServer := connectActionRunnerCredentialForTest(t, router.agentExecServer, first)
+	defer activeConn.Close()
+	defer activeServer.Close()
+	second := issueActionRunnerCredentialForTest(t, router, hostID, "host-1.local")
+	pendingConn, pendingServer := connectActionRunnerCredentialForTest(t, router.agentExecServer, second)
+	defer pendingConn.Close()
+	defer pendingServer.Close()
+
+	type updateOutcome struct {
+		result *agentexec.HostUpdateResultPayload
+		err    error
+	}
+	updateDone := make(chan updateOutcome, 1)
+	expectedHash := "sha256:" + strings.Repeat("a", 64)
+	go func() {
+		ctx, cancel := context.WithTimeout(agentexec.WithOrganizationID(context.Background(), "default"), 5*time.Second)
+		defer cancel()
+		result, err := router.agentExecServer.ExecuteHostUpdate(ctx, hostID, agentexec.HostUpdatePayload{
+			RequestID: "persistence-fence", ActionID: "action-fence", Operation: agentexec.HostUpdateOperationInstall,
+			ExpectedInventoryHash: expectedHash, Timeout: 5,
+		})
+		updateDone <- updateOutcome{result: result, err: err}
+	}()
+	_ = activeConn.SetReadDeadline(time.Now().Add(2 * time.Second))
+	var dispatch agentexec.Message
+	if err := activeConn.ReadJSON(&dispatch); err != nil {
+		t.Fatal(err)
+	}
+	_ = activeConn.SetReadDeadline(time.Time{})
+	if dispatch.Type != agentexec.MsgTypeHostUpdate {
+		t.Fatalf("dispatch type = %q, want host update", dispatch.Type)
+	}
+	var updateRequest agentexec.HostUpdatePayload
+	if err := dispatch.DecodePayload(&updateRequest); err != nil {
+		t.Fatal(err)
+	}
+	resultFor := func(afterHash string) agentexec.HostUpdateResultPayload {
+		now := time.Now().UTC()
+		return agentexec.HostUpdateResultPayload{
+			RequestID: updateRequest.RequestID, ActionID: updateRequest.ActionID, Success: true,
+			ExecutionPhase: agentexec.HostUpdatePhaseComplete,
+			Before:         agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: updateRequest.ExpectedInventoryHash, PendingCount: 1, CheckedAt: now.Add(-time.Second)},
+			After:          agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: afterHash, PendingCount: 0, CheckedAt: now},
+			HealthChecked:  true, PackageManagerHealthy: true, Verification: agentexec.HostUpdateVerificationVerified,
+		}
+	}
+	send := func(conn *websocket.Conn, messageType agentexec.MessageType, id string, payload any) {
+		t.Helper()
+		message, err := agentexec.NewMessage(messageType, id, payload)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := conn.WriteJSON(message); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	router.persistence.SetFileSystem(&actionRunnerBlockingFailingPersistenceFS{entered: entered, release: release})
+	activationDone := make(chan *httptest.ResponseRecorder, 1)
+	go func() {
+		activationDone <- requestActionRunnerActivationForTest(t, router, second)
+	}()
+	select {
+	case <-entered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("activation did not enter blocked persistence")
+	}
+
+	blockedHash := "sha256:" + strings.Repeat("b", 64)
+	send(activeConn, agentexec.MsgTypeHostUpdateResult, updateRequest.RequestID, resultFor(blockedHash))
+	send(activeConn, agentexec.MsgTypeAgentPing, "", nil)
+	_ = activeConn.SetReadDeadline(time.Now().Add(2 * time.Second))
+	var pong agentexec.Message
+	if err := activeConn.ReadJSON(&pong); err != nil {
+		t.Fatal(err)
+	}
+	_ = activeConn.SetReadDeadline(time.Time{})
+	if pong.Type != agentexec.MsgTypePong {
+		t.Fatalf("persistence-fence barrier = %q, want pong", pong.Type)
+	}
+	select {
+	case outcome := <-updateDone:
+		t.Fatalf("fenced predecessor satisfied request: result=%#v err=%v", outcome.result, outcome.err)
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	close(release)
+	select {
+	case rec := <-activationDone:
+		if rec.Code != http.StatusInternalServerError {
+			t.Fatalf("failed activation status = %d, body=%s", rec.Code, rec.Body.String())
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("activation did not return after persistence failure")
+	}
+	acceptedHash := "sha256:" + strings.Repeat("c", 64)
+	send(activeConn, agentexec.MsgTypeHostUpdateResult, updateRequest.RequestID, resultFor(acceptedHash))
+	select {
+	case outcome := <-updateDone:
+		if outcome.err != nil || outcome.result == nil || outcome.result.After.InventoryHash != acceptedHash {
+			t.Fatalf("restored predecessor result=%#v err=%v", outcome.result, outcome.err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("restored predecessor did not satisfy request")
+	}
+}
+
 func TestActivateActionRunnerCredentialRequiresExactRegisteredSession(t *testing.T) {
 	router, cfg, hostID := newActionRunnerCredentialTestRouter(t)
 	router.agentExecServer = agentexec.NewServerWithAdmissionValidator(router.admitAgentExecToken, router.validateAgentExecSession)

internal/api/agenttokens/install.go

@@ -312,24 +312,30 @@ func ActivateActionRunnerAndPersist(cfg *config.Config, persistence *config.Conf
 	return activateActionRunnerAndPersist(cfg, persistence, tokenID, agentID, hostname, false, nil)
 }
 
-// ActivateActionRunnerAndPersistWithPromotion durably activates an action
-// runner credential only when the exact prepared transport can be promoted in
-// the same serialized transaction. The promotion callback must perform only a
-// bounded in-memory mutation: it runs while config.Mu is held and may acquire
-// the agent-exec server mutex, establishing the sole nested lock order
-// config.Mu -> agentexec.Server.mu. It must not close sockets, log, or wait.
-func ActivateActionRunnerAndPersistWithPromotion(cfg *config.Config, persistence *config.ConfigPersistence, tokenID, agentID, hostname string, promote func() bool) (*config.APITokenRecord, []config.APITokenRecord, bool, error) {
-	return activateActionRunnerAndPersist(cfg, persistence, tokenID, agentID, hostname, true, promote)
+// ActionRunnerPromotionTransaction fences session authority across credential
+// persistence. Implementations must keep all methods bounded to in-memory map
+// mutation and preserve the lock order config.Mu -> session mutex.
+type ActionRunnerPromotionTransaction interface {
+	Commit() bool
+	Rollback()
+	FailClosed()
 }
 
-func activateActionRunnerAndPersist(cfg *config.Config, persistence *config.ConfigPersistence, tokenID, agentID, hostname string, requireDurablePromotion bool, promote func() bool) (*config.APITokenRecord, []config.APITokenRecord, bool, error) {
+// ActivateActionRunnerAndPersistWithPromotion durably activates an action
+// runner credential only when the exact prepared transport can be fenced
+// before persistence and promoted in the same serialized transaction.
+func ActivateActionRunnerAndPersistWithPromotion(cfg *config.Config, persistence *config.ConfigPersistence, tokenID, agentID, hostname string, beginPromotion func() (ActionRunnerPromotionTransaction, bool)) (*config.APITokenRecord, []config.APITokenRecord, bool, error) {
+	return activateActionRunnerAndPersist(cfg, persistence, tokenID, agentID, hostname, true, beginPromotion)
+}
+
+func activateActionRunnerAndPersist(cfg *config.Config, persistence *config.ConfigPersistence, tokenID, agentID, hostname string, requireDurablePromotion bool, beginPromotion func() (ActionRunnerPromotionTransaction, bool)) (*config.APITokenRecord, []config.APITokenRecord, bool, error) {
 	if cfg == nil {
 		return nil, nil, false, fmt.Errorf("%w: config is required", ErrRecord)
 	}
 	if persistence == nil {
 		return nil, nil, false, fmt.Errorf("%w: durable persistence is required", ErrPersist)
 	}
-	if requireDurablePromotion && promote == nil {
+	if requireDurablePromotion && beginPromotion == nil {
 		return nil, nil, false, fmt.Errorf("%w: prepared session promotion is required", ErrActionRunnerSessionUnavailable)
 	}
 	tokenID = strings.TrimSpace(tokenID)
@@ -362,6 +368,14 @@ func activateActionRunnerAndPersist(cfg *config.Config, persistence *config.Conf
 		clone := record.Clone()
 		return &clone, nil, false, nil
 	}
+	var promotion ActionRunnerPromotionTransaction
+	if beginPromotion != nil {
+		var ok bool
+		promotion, ok = beginPromotion()
+		if !ok || promotion == nil {
+			return nil, nil, false, ErrActionRunnerSessionUnavailable
+		}
+	}
 
 	previousTokens := cloneAPITokenRecords(cfg.APITokens)
 	replaceIDs := make(map[string]struct{})
@@ -388,17 +402,22 @@ func activateActionRunnerAndPersist(cfg *config.Config, persistence *config.Conf
 	if err := persistence.SaveAPITokens(cfg.APITokens); err != nil {
 		cfg.APITokens = previousTokens
 		cfg.SortAPITokens()
+		if promotion != nil {
+			promotion.Rollback()
+		}
 		return nil, nil, false, fmt.Errorf("%w: %w", ErrPersist, err)
 	}
-	if promote != nil && !promote() {
+	if promotion != nil && !promotion.Commit() {
 		if err := persistence.SaveAPITokens(previousTokens); err != nil {
 			// The activation inventory was the last state known to reach durable
 			// storage. Keep memory aligned with that state and force repair rather
 			// than exposing pending memory against an active on-disk credential.
+			promotion.FailClosed()
 			return &activated, revoked, true, fmt.Errorf("%w: rollback persistence failed: %v", ErrActionRunnerActivationIndeterminate, err)
 		}
 		cfg.APITokens = previousTokens
 		cfg.SortAPITokens()
+		promotion.Rollback()
 		return nil, nil, false, ErrActionRunnerSessionUnavailable
 	}
 	return &activated, revoked, true, nil

internal/api/agenttokens/install_test.go

@@ -26,6 +26,26 @@ type actionRunnerFailAfterWritesFS struct {
 	allowWrites int
 }
 
+type actionRunnerPromotionTestTransaction struct {
+	commitResult    bool
+	commitCalls     int
+	rollbackCalls   int
+	failClosedCalls int
+}
+
+func (tx *actionRunnerPromotionTestTransaction) Commit() bool {
+	tx.commitCalls++
+	return tx.commitResult
+}
+
+func (tx *actionRunnerPromotionTestTransaction) Rollback() {
+	tx.rollbackCalls++
+}
+
+func (tx *actionRunnerPromotionTestTransaction) FailClosed() {
+	tx.failClosedCalls++
+}
+
 func (fs *actionRunnerFailAfterWritesFS) ReadFile(name string) ([]byte, error) {
 	return os.ReadFile(name)
 }
@@ -312,6 +332,59 @@ func TestActivateActionRunnerAndPersistFailureRestoresPendingAndActiveInventory(
 	}
 }
 
+func TestActivateActionRunnerPromotionFencesBeforePersistenceAndRollsBackOnSaveFailure(t *testing.T) {
+	cfg := &config.Config{DataPath: t.TempDir()}
+	_, predecessor, err := IssueActionRunnerAndPersist(cfg, nil, ActionRunnerIssueOptions{OrgID: "org-a", AgentID: "machine-123", Hostname: "node.example"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, _, _, err := ActivateActionRunnerAndPersist(cfg, config.NewConfigPersistence(cfg.DataPath), predecessor.ID, "machine-123", "node.example"); err != nil {
+		t.Fatal(err)
+	}
+	_, pending, err := IssueActionRunnerAndPersist(cfg, nil, ActionRunnerIssueOptions{OrgID: "org-a", AgentID: "machine-123", Hostname: "node.example"})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	persistence := config.NewConfigPersistence(cfg.DataPath)
+	persistence.SetFileSystem(&actionRunnerTestFS{entered: entered, release: release, writeErr: errors.New("injected activation save failure")})
+	tx := &actionRunnerPromotionTestTransaction{commitResult: true}
+	type activationOutcome struct {
+		changed bool
+		err     error
+	}
+	done := make(chan activationOutcome, 1)
+	go func() {
+		_, _, changed, err := ActivateActionRunnerAndPersistWithPromotion(cfg, persistence, pending.ID, "machine-123", "node.example", func() (ActionRunnerPromotionTransaction, bool) {
+			return tx, true
+		})
+		done <- activationOutcome{changed: changed, err: err}
+	}()
+
+	select {
+	case <-entered:
+		if tx.commitCalls != 0 || tx.rollbackCalls != 0 || tx.failClosedCalls != 0 {
+			t.Fatalf("promotion resolved before persistence outcome: commit %d rollback %d fail-closed %d", tx.commitCalls, tx.rollbackCalls, tx.failClosedCalls)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("activation did not enter persistence after beginning promotion")
+	}
+	close(release)
+	select {
+	case outcome := <-done:
+		if !errors.Is(outcome.err, ErrPersist) || outcome.changed {
+			t.Fatalf("activation outcome = changed %v err %v", outcome.changed, outcome.err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("activation did not return after persistence failure")
+	}
+	if tx.commitCalls != 0 || tx.rollbackCalls != 1 || tx.failClosedCalls != 0 {
+		t.Fatalf("failed-save promotion lifecycle = commit %d rollback %d fail-closed %d", tx.commitCalls, tx.rollbackCalls, tx.failClosedCalls)
+	}
+}
+
 func TestActivateActionRunnerAndPersistWithPromotionRollsBackWhenExactTransportVanished(t *testing.T) {
 	cfg := &config.Config{DataPath: t.TempDir()}
 	_, predecessor, err := IssueActionRunnerAndPersist(cfg, nil, ActionRunnerIssueOptions{OrgID: "org-a", AgentID: "machine-123", Hostname: "node.example"})
@@ -326,16 +399,17 @@ func TestActivateActionRunnerAndPersistWithPromotionRollsBackWhenExactTransportV
 		t.Fatal(err)
 	}
 	persistence := config.NewConfigPersistence(cfg.DataPath)
-	promotionCalls := 0
-	activated, revoked, changed, err := ActivateActionRunnerAndPersistWithPromotion(cfg, persistence, pending.ID, "machine-123", "node.example", func() bool {
-		promotionCalls++
-		return false
+	tx := &actionRunnerPromotionTestTransaction{}
+	beginCalls := 0
+	activated, revoked, changed, err := ActivateActionRunnerAndPersistWithPromotion(cfg, persistence, pending.ID, "machine-123", "node.example", func() (ActionRunnerPromotionTransaction, bool) {
+		beginCalls++
+		return tx, true
 	})
 	if !errors.Is(err, ErrActionRunnerSessionUnavailable) || activated != nil || revoked != nil || changed {
 		t.Fatalf("vanished transport activation = activated %#v revoked %#v changed %v err %v", activated, revoked, changed, err)
 	}
-	if promotionCalls != 1 {
-		t.Fatalf("promotion calls = %d, want 1", promotionCalls)
+	if beginCalls != 1 || tx.commitCalls != 1 || tx.rollbackCalls != 1 || tx.failClosedCalls != 0 {
+		t.Fatalf("promotion lifecycle = begin %d commit %d rollback %d fail-closed %d", beginCalls, tx.commitCalls, tx.rollbackCalls, tx.failClosedCalls)
 	}
 	if len(cfg.APITokens) != 2 {
 		t.Fatalf("rolled-back inventory = %#v", cfg.APITokens)
@@ -379,13 +453,19 @@ func TestActivateActionRunnerPromotionRollbackPersistenceFailureKeepsLastDurable
 	// one backup plus the new token file. Permit both, then fail the compensating
 	// rollback writes.
 	persistence.SetFileSystem(&actionRunnerFailAfterWritesFS{allowWrites: 2})
-	activated, revoked, changed, err := ActivateActionRunnerAndPersistWithPromotion(cfg, persistence, pending.ID, "machine-123", "node.example", func() bool { return false })
+	tx := &actionRunnerPromotionTestTransaction{}
+	activated, revoked, changed, err := ActivateActionRunnerAndPersistWithPromotion(cfg, persistence, pending.ID, "machine-123", "node.example", func() (ActionRunnerPromotionTransaction, bool) {
+		return tx, true
+	})
 	if !errors.Is(err, ErrActionRunnerActivationIndeterminate) || activated == nil || activated.ID != pending.ID || len(revoked) != 1 || revoked[0].ID != predecessor.ID || !changed {
 		t.Fatalf("rollback persistence failure = activated %#v revoked %#v changed %v err %v", activated, revoked, changed, err)
 	}
 	if len(cfg.APITokens) != 1 || cfg.APITokens[0].ID != pending.ID || cfg.APITokens[0].ExpiresAt != nil || cfg.APITokens[0].Metadata[ActionRunnerActivationPendingMetadataKey] != "" {
 		t.Fatalf("memory diverged from last durable activation: %#v", cfg.APITokens)
 	}
+	if tx.commitCalls != 1 || tx.rollbackCalls != 0 || tx.failClosedCalls != 1 {
+		t.Fatalf("indeterminate promotion lifecycle = commit %d rollback %d fail-closed %d", tx.commitCalls, tx.rollbackCalls, tx.failClosedCalls)
+	}
 	persisted, err := config.NewConfigPersistence(cfg.DataPath).LoadAPITokens()
 	if err != nil {
 		t.Fatal(err)

internal/api/security_regression_test.go

@@ -768,6 +768,10 @@ func TestActionRunnerCredentialRotationRevokesPreviousSecretOnlyAtActivation(t *
 	}
 }
 
+func TestSecurityActionRunnerActivationFencesPredecessorResultsAcrossPersistence(t *testing.T) {
+	testActionRunnerCredentialFencesPredecessorResultsAcrossPersistence(t)
+}
+
 func TestAgentExecTokenRejectsAmbiguousMultiOrganizationAuthority(t *testing.T) {
 	rawToken := "multi-org-agent-token-123.12345678"
 	record := newTokenRecord(t, rawToken, []string{config.ScopeAgentExec}, map[string]string{