refactor: close bounded recovery drain and invalidation seams

Move removed-replica drain and replica-scoped invalidation onto explicit core-command paths so the widened multi-replica runtime no longer depends on coarse host-side recovery handling.

Made-with: Cursor

sw-block/.private/phase/phase-16-log.md

@@ -1025,3 +1025,127 @@ Conclusion:
    core-command-driven on the core-present path
 2. this slice still does not claim broad multi-replica recovery-loop closure or
    broad failover/publication closure
+
+---
+
+#### `16J` Start Note Rev 1
+
+Date: 2026-04-04
+Scope: bounded removed-replica recovery drain ownership on the core-present path
+
+Why this slice exists:
+
+1. `16A-16I` have moved bounded assignment, startup, execution, addressing, and
+   observation seams toward explicit core ownership
+2. but removed-replica recovery drain on the core-present path still depends on
+   direct `orchestrator.ProcessAssignment(...).Removed` handling in
+   `weed/server`
+3. that means one visible recovery-loop branch is still outside the explicit
+   core-owned command / event seam
+
+Chosen implementation rule:
+
+1. add one bounded core-owned seam for removed-replica drain
+2. rebind only the core-present host path to consume that seam
+3. leave legacy no-core compatibility handling intact
+
+---
+
+#### `16J` Delivery Note Rev 1
+
+Date: 2026-04-04
+Scope: bounded removed-replica recovery drain ownership on the core-present path
+
+What changed:
+
+1. `sw-block/engine/replication/command.go`
+   - added bounded `drain_recovery_task` as an explicit core command
+2. `sw-block/engine/replication/engine.go`
+   - assignment delivery now emits `drain_recovery_task` for previously
+     recovery-owned replica targets that are no longer in the bounded target set
+3. `weed/server/blockcmd/dispatch.go` and `weed/server/blockcmd/service_ops.go`
+   - added server-adapter handling for the new drain command
+4. `weed/server/block_recovery.go`
+   - exposed a bounded recovery drain method that cancels and drains removed
+     replica work from the new command seam while leaving legacy no-core methods
+     intact
+5. `weed/server/volume_server_block.go`
+   - removed the core-present direct dependency on
+     `HandleRemovedAssignments(result)`
+6. focused proofs:
+   - `sw-block/engine/replication/phase14_command_test.go`
+   - `weed/server/blockcmd/dispatch_test.go`
+   - `weed/server/volume_server_block_test.go`
+
+Proof / evidence:
+
+1. `go test ./...` from `sw-block/engine/replication`
+2. `go test ./weed/server/blockcmd -count=1`
+3. `go test ./weed/server -count=1 -timeout 120s -run "TestBlockService_(ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|PrimaryMultiReplica_UsesCoreStartRecoveryTaskPerReplica|RemovedReplica_UsesCoreDrainRecoveryTask|RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart)|BarrierRejected_ExecutesCoreInvalidateSession|DebugInfoForVolume|CollectBlockVolumeHeartbeat|ReadinessSnapshot|HeartbeatReplicaDegraded)"`
+4. result: `PASS`
+
+Conclusion:
+
+1. removed-replica recovery drain on the core-present path is now
+   core-command-driven rather than primarily orchestrator-result-driven
+2. this slice still does not claim broad recovery-loop closure or
+   failover/publication closure
+
+---
+
+#### `16K` Start Note Rev 1
+
+Date: 2026-04-04
+Scope: bounded replica-scoped session invalidation on the core-present path
+
+Why this slice exists:
+
+1. `16F-16J` already made recovery command addressing, observation events,
+   startup, and removed-task drain much more replica-scoped
+2. but `InvalidateSessionCommand` still invalidates all replica sessions for a
+   volume even when the triggering recovery event is already replica-scoped
+3. that over-broad seam becomes more problematic as bounded multi-replica
+   runtime ownership widens
+
+Chosen implementation rule:
+
+1. extend invalidation command addressing to optionally target one replica
+2. emit replica-scoped invalidation only from replica-scoped recovery events
+3. preserve volume-wide invalidation for truly volume-scoped failures
+
+---
+
+#### `16K` Delivery Note Rev 1
+
+Date: 2026-04-04
+Scope: bounded replica-scoped session invalidation on the core-present path
+
+What changed:
+
+1. `sw-block/engine/replication/command.go`
+   - widened `invalidate_session` command addressing to optionally target one
+     replica
+2. `sw-block/engine/replication/engine.go`
+   - replica-scoped recovery escalation now emits targeted invalidation while
+     volume-scoped barrier rejection remains volume-wide
+3. `weed/server/blockcmd/dispatch.go` and `weed/server/blockcmd/service_ops.go`
+   - server adapter now invalidates one sender when `ReplicaID` is present and
+     still invalidates all projection replicas for volume-wide paths
+4. focused proofs:
+   - `sw-block/engine/replication/phase14_command_test.go`
+   - `weed/server/blockcmd/dispatch_test.go`
+   - `weed/server/volume_server_block_test.go`
+
+Proof / evidence:
+
+1. `go test ./...` from `sw-block/engine/replication`
+2. `go test ./weed/server/blockcmd -count=1`
+3. `go test ./weed/server -count=1 -timeout 120s -run "TestBlockService_(ApplyAssignments_(PrimaryRole_UsesCoreStartRecoveryTaskForCatchUp|PrimaryMultiReplica_UsesCoreStartRecoveryTaskPerReplica|RemovedReplica_UsesCoreDrainRecoveryTask|RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart)|BarrierRejected_ExecutesCoreInvalidateSession|BarrierRejected_DoesNotReexecuteInvalidateOnSameReason|NeedsRebuildObserved_InvalidatesOnlyTargetReplica|DebugInfoForVolume|CollectBlockVolumeHeartbeat|ReadinessSnapshot|HeartbeatReplicaDegraded)"`
+4. result: `PASS`
+
+Conclusion:
+
+1. replica-scoped recovery invalidation on the core-present path no longer
+   relies on broad volume-wide invalidation
+2. this slice still does not claim broad failover/publication closure or full
+   recovery-loop closure

sw-block/.private/phase/phase-16.md

@@ -370,6 +370,91 @@ Evidence:
 
 1. focused working-tree change after `92c006eb2`
 
+### `16J`: Removed-Replica Recovery Drain Ownership
+
+Goal:
+
+1. close one more bounded recovery-loop gap by moving removed-replica recovery
+   drain ownership off the direct orchestrator-result seam and onto an explicit
+   core-owned path
+2. keep the slice limited to removed-replica drain / invalidation on the
+   core-present path, not broad recovery-loop closure
+
+Acceptance object:
+
+1. on the core-present path, replica removal no longer requires
+   `HandleRemovedAssignments(result)` as the primary recovery-drain trigger
+2. the host drains removed recovery work because of an explicit core-owned
+   command or event seam
+3. current bounded startup / execution / observation proofs remain green
+4. this slice still does not yet claim broad failover/publication closure
+
+Current chosen path:
+
+1. define one bounded core-owned seam for removed-replica recovery drain
+2. rebind the core-present host path to consume that seam instead of direct
+   orchestrator-result removal handling
+3. keep legacy no-core compatibility unchanged
+
+Status:
+
+1. delivered
+
+Delivered result:
+
+1. the core now emits a bounded `drain_recovery_task` command when assignment
+   delivery removes a previously recovery-owned replica target
+2. the core-present host path drains removed recovery work from that command
+   instead of using direct `orchestrator.ProcessAssignment(...).Removed` as the
+   primary trigger
+3. legacy no-core compatibility remains isolated in `RecoveryManager`
+
+Evidence:
+
+1. focused working-tree change after `5fd9ec0ed`
+
+### `16K`: Replica-Scoped Session Invalidation
+
+Goal:
+
+1. close one bounded multi-replica runtime gap by making per-replica failure
+   invalidation explicit instead of broad volume-wide invalidation
+2. keep the slice limited to replica-scoped invalidation for replica-scoped
+   recovery events, not broad failover/publication closure
+
+Acceptance object:
+
+1. a replica-scoped recovery failure/escalation on the core-present path can
+   invalidate only the affected replica session
+2. volume-wide invalidation paths remain volume-wide where the event itself is
+   volume-scoped
+3. current bounded startup / execution / drain proofs remain green
+4. this slice still does not yet claim broad failover/publication closure
+
+Current chosen path:
+
+1. widen `InvalidateSessionCommand` from volume-only addressing to optional
+   replica-scoped addressing
+2. emit replica-scoped invalidation from replica-scoped recovery events such as
+   `NeedsRebuildObserved`
+3. keep `BarrierRejected` and other volume-scoped invalidation paths unchanged
+
+Status:
+
+1. delivered
+
+Delivered result:
+
+1. `InvalidateSessionCommand` now supports bounded replica-scoped addressing in
+   addition to volume-wide invalidation
+2. replica-scoped recovery escalation now invalidates only the affected replica
+   session on the core-present path
+3. volume-scoped invalidation paths such as `BarrierRejected` remain unchanged
+
+Evidence:
+
+1. focused working-tree change after `5fd9ec0ed`
+
 ## Current Checkpoint Review Target
 
 The current review target is the current widened bounded runtime checkpoint
@@ -434,11 +519,17 @@ boundary:
      volume projection layer
 10. `16I` delivered:
    - multi-replica primary catch-up startup ownership is core-command-driven
+11. `16J` delivered:
+   - removed-replica recovery drain is core-command-driven on the core-present
+     path
+12. `16K` delivered:
+   - replica-scoped recovery invalidation no longer depends on a remaining
+     volume-wide invalidation seam
 
 After this checkpoint:
 
 1. keep `legacy P4` only as a compatibility guard
-2. identify the next bounded runtime gap after multi-replica startup ownership,
-   most likely around broader recovery-loop closure rather than assignment entry
+2. continue closing broader recovery-loop gaps one bounded seam at a time after
+   replica-scoped invalidation
 3. do not yet claim full recovery-loop closure
 4. do not broaden into launch claims

sw-block/design/v2-product-completion-overview.md

@@ -1,6 +1,6 @@
 # V2 Product Completion Overview
 
-Date: 2026-03-31
+Date: 2026-04-04
 Status: active
 Purpose: provide one product-level overview of current V2 engineering completion, V1 reuse strategy, and the roadmap from the accepted candidate path to a production-ready block engine
 
@@ -23,8 +23,8 @@ This document is the product-completion view.
 It complements:
 
 1. `v2-protocol-truths.md` for accepted semantics
-2. `../docs/archive/design/v2-production-roadmap.md` for the older roadmap ladder
-3. `../.private/phase/phase-08.md` for current phase contract
+2. `v2-phase-development-plan.md` for the current phase ladder
+3. `../.private/phase/phase-16.md` for the active bounded runtime-closure contract
 
 ## Current Position
 
@@ -37,13 +37,24 @@ The accepted first candidate path is:
 5. `v2bridge` translates real storage/control truth
 6. `blockvol` remains the execution backend
 
-This means the project is no longer at "algorithm only".
+This means the project is no longer at "algorithm only" or even only at
+"bounded prototype".
 It already has:
 
 1. accepted protocol truths
-2. accepted engine execution closure
-3. accepted hardening replay on a real integrated path
-4. one bounded candidate statement
+2. accepted engine execution closure on the chosen path
+3. accepted control-plane and product-surface rebinding on the chosen path
+4. accepted hardening evidence for the bounded chosen path
+5. an active `Phase 16` runtime-closure program that is moving live ownership
+   from host-local orchestration toward explicit core-driven command / event /
+   projection seams
+
+The most important current distinction is:
+
+1. the runtime backbone is now largely present
+2. product completion is no longer blocked by "missing first implementation"
+3. product completion is still blocked by broader closure and launch-envelope
+   proof
 
 ## Engineering Completion Snapshot
 
@@ -53,15 +64,17 @@ These levels are rough engineering estimates, not exact percentages.
 |------|---------------|-------|
 | Algorithm / protocol truths | Strong | Core V2 semantics are accepted and should remain stable unless contradicted by live evidence. |
 | Simulator / prototype evidence | Strong | Main failure classes and protocol boundaries are already well-exercised. |
-| Engine recovery core | Strong | Sender/session/orchestrator/driver/executor are substantially implemented. |
+| Engine recovery core | Strong | Sender/session/orchestrator/driver/executor are substantially implemented and remain the semantic center. |
 | Weed bridge integration | Strong | Reader / pinner / control / executor are real and tested on the chosen path. |
-| Integrated candidate path | Strong on chosen path | Backend, control-plane, and selected product surfaces are now accepted on one bounded chosen path. |
-| Runtime ownership inside live server loop | Strong on chosen path | Accepted chosen-path execution/control ownership exists; later work is restart/disturbance hardening, not first-closure rebinding. |
+| Integrated candidate path | Strong on chosen path | Backend, control-plane, and selected product surfaces are accepted on one bounded chosen path. |
+| Runtime ownership inside live server loop | Strong on bounded path, still widening | `Phase 16A-16I` now give one substantial core-command-driven runtime path, but broad recovery-loop closure is not yet claimed. |
 | Production-grade data transfer | Strong on chosen path | `TransferFullBase` and `TransferSnapshot` execution closure are accepted on the chosen path; later work is hardening. |
 | Truncation / replica-ahead execution | Strong on chosen path | `TruncateWAL` narrow chosen-path closure is accepted; later work is hardening/planning improvement. |
 | End-to-end control-plane closure | Strong on chosen path | `Phase 10` accepted bounded end-to-end control-path closure on the chosen path. |
 | Product surfaces (`CSI`, `NVMe`, `iSCSI`, snapshot productization) | Strong on chosen path | `Phase 11` accepted bounded product-surface rebinding on the chosen path. |
-| Production hardening / ops | Partial | `Phase 12` is now the next active stage. |
+| Production hardening / ops | Strong on bounded path, not full launch proof | `Phase 12` closed the bounded hardening bar, but not the whole first-launch envelope. |
+| Multi-replica catch-up runtime ownership | Strong on bounded startup/execution path | `16F-16I` made command/event/pending/aggregation/startup ownership replica-scoped enough for the bounded primary multi-replica path. |
+| Broad failover / publication / launch envelope | Partial | Main remaining product gap: prove wider runtime closure and freeze a supported launch envelope. |
 
 ## Reuse Strategy
 
@@ -128,133 +141,89 @@ For the chosen `RF=2 sync_all` path, the project can already claim:
 8. committed/checkpoint separation accepted for this candidate path:
    - `CommittedLSN = WALHeadLSN`
    - `CheckpointLSN` remains the durable base-image boundary
+9. bounded live runtime ownership has materially improved after `Phase 15`:
+   - assignment entry is core-owned on the bounded path
+   - `apply_role`, `start_receiver`, `configure_shipper`, and
+     `invalidate_session` are command-driven
+   - bounded catch-up / rebuild execution starts from core-emitted recovery
+     commands
+   - recovery command addressing and observation events are replica-scoped
+   - multi-replica catch-up aggregation and startup ownership are now bounded on
+     the primary path
 
 ## What Is Still Missing For Product Completion
 
-The biggest remaining product-completion gaps are now production-hardening gaps:
+The biggest remaining product-completion gaps are no longer "invent the first
+working path". They are closure and launch-envelope gaps:
 
-1. restart / recovery disturbance hardening
-   - accepted chosen-path behavior must remain correct under restart, rejoin, and repeated failover
-2. long-run / soak stability
-   - accepted behavior must remain stable across repeated cycles and longer-running operation
-3. operational diagnosability
-   - blockers, symptoms, and operator-visible diagnosis quality must be explicit
-4. performance floor and rollout gates
-   - production claims need bounded floor numbers and explicit rollout criteria
+1. broader recovery-loop closure
+   - current `Phase 16` evidence is strong on one bounded runtime path, but not
+     yet a claim that all recovery lifecycle branches are core-owned end to end
+2. broader failover / publication closure
+   - product-facing truth under failover, replay, and disturbance still needs a
+     stronger whole-chain statement, not only local runtime proofs
+3. restart / disturbance preservation
+   - accepted behavior must remain correct under restart, rejoin, repeated
+     failover, and address churn without reopening semantic ambiguity
+4. long-run / soak / performance floor
+   - accepted behavior must remain stable under longer operation with explicit
+     floor numbers and bounded rollout gates
+5. launch-envelope freeze
+   - the first supported launch matrix, exclusions, operator expectations, and
+     stop conditions must be written down explicitly
 
 ## Recommended Completion Roadmap
 
-### Stage 1: Finish Phase 08 cleanly
+### Stage 1: Finish `Phase 16` runtime closure cleanly
 
 Target:
 
-1. close candidate-path judgment with explicit bounds and package it cleanly inside `Phase 08 P4`
+1. close the next bounded `Phase 16` runtime gap with explicit bounds and
+   package one clear reviewable runtime checkpoint
 
 Main output:
 
-1. one accepted candidate package for the chosen path
-
-### Stage 2: Phase 09 Production Execution Closure
-
-Target:
-
-1. turn validation-grade execution into production-grade execution
-
-Main work:
-
-1. real `TransferFullBase`
-2. real `TransferSnapshot`
-3. real `TruncateWAL`
-4. stronger runtime ownership of recovery execution
+1. one accepted bounded `V2`-native runtime path that is explicit about what is
+   core-owned, what is adapter-owned, and what is still only compatibility
+2. one short list of residual recovery-loop gaps after `16I`
 
 Why it matters:
 
-This is the largest remaining engineering block between "candidate-safe-with-bounds" and a serious product path.
+This is now the main semantic engineering blocker between a strong bounded path
+and a launchable product statement.
 
-### Stage 3: Phase 10 Real Control-Plane Closure
+### Stage 2: Freeze the first supported launch envelope
 
 Target:
 
-1. strengthen from accepted assignment-entry closure to fuller end-to-end control-path closure
+1. convert accepted protocol, runtime, control-plane, and hardening evidence
+   into one bounded first-launch support statement
 
 Status:
 
-1. accepted and closed on the chosen path
+1. not yet frozen
 
 Main work:
 
-1. heartbeat/gRPC-level proof
-2. stronger control/result convergence
-3. better identity completeness for local and remote server roles
+1. define the supported environment / replication / durability matrix
+2. define explicit exclusions that remain outside the first launch claim
+3. bind product-facing surfaces to the named supported envelope
+4. define preflight, success, and stop conditions
 
-### Stage 4: Phase 11 Product Surface Rebinding
+### Stage 3: Internal pilot and incident-driven hardening
 
 Target:
 
-1. connect product-facing surfaces to the V2-backed block path
-
-Status:
-
-1. accepted and closed on the chosen path
-
-Candidate areas:
-
-1. snapshot product path
-2. `CSI`
-3. `NVMe`
-4. `iSCSI`
-
-Recommended first cut:
-
-1. snapshot product path first
-2. `CSI` and `NVMe` / `iSCSI` after one bounded product-visible surface is already accepted
-
-Suggested order inside `Phase 11`:
-
-1. `P1` snapshot product path
-2. `P2` `CSI`
-3. `P3` `NVMe` / `iSCSI`
-4. `P4` broader residual workflow closure if still required
-
-Rule:
-
-Do this after the backend engine/runtime path is strong enough, not before.
-
-### Stage 5: Phase 12 Production Hardening
-
-Target:
-
-1. move from candidate-safe to production-safe
-
-Status:
-
-1. accepted and closed on the bounded chosen path
+1. validate the frozen launch envelope without silently broadening scope
 
 Main work:
 
-1. soak / restart / repeated failover
-2. operational diagnosis quality
-3. performance floor and cost characterization
-4. explicit production blockers / rollout gates
-
-### Stage 6: Post-`Phase 12` Productionization Program
-
-Target:
-
-1. turn the accepted `Phase 12` chosen path into a bounded first-launch product envelope without reopening protocol discovery
-
-Status:
-
-1. next active stage after `Phase 12`
-
-Main work:
-
-1. freeze the first supported launch envelope from accepted `P1`-`P4` evidence
-2. define a limited internal pilot package with explicit preflight, success, and stop conditions
-3. run incident-driven hardening with explicit classification:
+1. run a limited internal pilot package
+2. route incidents with explicit classification:
    - config / environment issue
    - known exclusion
    - true product bug
+3. harden only within the named supported envelope
 4. perform controlled rollout review only within the named supported envelope
 
 Rules:
@@ -272,45 +241,53 @@ Rules:
 
 The most important gates from here are:
 
-1. execution gate
-   - validation-grade transfer/truncation must become production-grade
-2. runtime ownership gate
-   - V2 recovery must be a stronger live runtime path, not only a bounded tested path
-3. control-plane gate
-   - stronger end-to-end control delivery proof
-4. product-surface gate
-   - front-end surfaces should only rebind after backend correctness is strong enough
-5. production-hardening gate
-   - restart, soak, diagnosis, and repeated disturbance must be acceptable
-6. productionization gate
-   - first launch envelope, pilot discipline, incident routing, and controlled rollout review must be explicit
+1. runtime-closure gate
+   - the strongest live recovery path must be explicit, bounded, and
+     semantically owned by the core rather than spread across legacy host logic
+2. failover/publication gate
+   - outward truth under disturbance must be strong enough to make a product
+     statement, not only a local runtime statement
+3. restart/disturbance gate
+   - restart, rejoin, address change, and repeated failover must preserve the
+     accepted bounded semantics
+4. launch-envelope gate
+   - the first supported matrix and explicit exclusions must be written down
+5. pilot/rollout gate
+   - internal pilot discipline, incident routing, and rollout review must be
+     explicit before widening scope
 
 ## Near-Term Planning Guidance
 
 If the goal is to maximize product completion efficiently:
 
 1. do not reopen accepted execution, control-plane, or product-surface semantics casually
-2. finish `Phase 12` hardening cleanly
-3. then freeze the first supported launch envelope
-4. then run a limited internal pilot
-5. then widen only through explicit incident review and rollout-gate review
+2. treat the remaining `Phase 16` work as runtime-closure work, not as another
+   round of broad redesign
+3. finish the next bounded recovery-loop gaps before claiming more product scope
+4. then freeze the first supported launch envelope
+5. then run a limited internal pilot
+6. then widen only through explicit incident review and rollout-gate review
 
 In short:
 
-1. chosen-path closure first
-2. production hardening second
+1. runtime closure first
+2. launch-envelope freeze second
 3. bounded productionization third
 
 ## Short Summary
 
 The V2 line is already beyond "algorithm only".
-It has an accepted bounded chosen path through backend, control-plane, selected product surfaces, and `Phase 12` hardening.
+It has an accepted bounded chosen path through backend, control-plane, selected
+product surfaces, bounded hardening, and substantial `Phase 16` runtime closure.
 
-But the remaining work is still substantial, and it is mostly engineering work:
+The main remaining work is not "build the first real thing".
+It is:
 
-1. freeze the first supported launch envelope from accepted evidence
-2. run a limited internal pilot with explicit stop conditions
-3. harden from incidents without silently broadening scope
-4. perform controlled rollout review inside a bounded launch envelope
+1. finish broader recovery-loop and failover/publication closure strongly enough
+   for a product statement
+2. freeze the first supported launch envelope from accepted evidence
+3. run a limited internal pilot with explicit stop conditions
+4. harden from incidents without silently broadening scope
 
-That is the practical path from the current production-safe chosen path to a bounded first-launch block product.
+That is the practical path from the current bounded runtime-complete candidate
+to a bounded first-launch block product.

sw-block/engine/replication/command.go

@@ -35,6 +35,14 @@ type StartRecoveryTaskCommand struct {
 
 func (StartRecoveryTaskCommand) commandName() string { return "start_recovery_task" }
 
+type DrainRecoveryTaskCommand struct {
+	VolumeID  string
+	ReplicaID string
+	Reason    string
+}
+
+func (DrainRecoveryTaskCommand) commandName() string { return "drain_recovery_task" }
+
 type StartCatchUpCommand struct {
 	VolumeID  string
 	ReplicaID string
@@ -52,8 +60,9 @@ type StartRebuildCommand struct {
 func (StartRebuildCommand) commandName() string { return "start_rebuild" }
 
 type InvalidateSessionCommand struct {
-	VolumeID string
-	Reason   string
+	VolumeID  string
+	ReplicaID string
+	Reason    string
 }
 
 func (InvalidateSessionCommand) commandName() string { return "invalidate_session" }

sw-block/engine/replication/engine.go

@@ -1,6 +1,9 @@
 package replication
 
-import "reflect"
+import (
+	"reflect"
+	"sort"
+)
 
 // CoreEngine is the first explicit Phase 14 V2 core shell.
 // It is deterministic and side-effect free: one event in, updated state and
@@ -172,9 +175,11 @@ func (e *CoreEngine) ApplyEvent(ev Event) ApplyResult {
 		st.Recovery.Phase = RecoveryNeedsRebuild
 		st.Recovery.Reason = v.Reason
 		if st.shouldInvalidate(v.Reason) {
+			replicaID, _ := st.recoveryCommandReplicaIDFromEvent(v.ReplicaID)
 			cmds = append(cmds, InvalidateSessionCommand{
-				VolumeID: st.VolumeID,
-				Reason:   v.Reason,
+				VolumeID:  st.VolumeID,
+				ReplicaID: replicaID,
+				Reason:    v.Reason,
 			})
 			st.commands.InvalidationIssued = true
 			st.commands.InvalidationReason = v.Reason
@@ -318,6 +323,7 @@ func (e *CoreEngine) applyAssignment(st *VolumeState, ev AssignmentDelivered) []
 	epochChanged := st.Epoch != ev.Epoch
 	replicasChanged := !sameReplicaAssignments(st.DesiredReplicas, ev.Replicas)
 	recoveryTargetChanged := st.recoveryTarget != ev.RecoveryTarget
+	previousRecoveryTaskTargets := recoveryTaskTargetReplicaIDs(st.commands.RecoveryTaskTargets)
 
 	st.Epoch = ev.Epoch
 	st.Role = ev.Role
@@ -385,6 +391,13 @@ func (e *CoreEngine) applyAssignment(st *VolumeState, ev AssignmentDelivered) []
 		st.commands.ShipperConfigEpoch = st.Epoch
 		st.commands.ShipperConfigReplicas = append([]ReplicaAssignment(nil), st.DesiredReplicas...)
 	}
+	for _, replicaID := range removedRecoveryTaskReplicaIDs(previousRecoveryTaskTargets, st.recoveryTaskReplicaIDs()) {
+		cmds = append(cmds, DrainRecoveryTaskCommand{
+			VolumeID:  st.VolumeID,
+			ReplicaID: replicaID,
+			Reason:    "assignment_removed",
+		})
+	}
 	for _, replicaID := range st.recoveryTaskReplicaIDs() {
 		if !st.shouldStartRecoveryTask(replicaID) {
 			continue
@@ -515,6 +528,39 @@ func (st *VolumeState) recoveryTaskReplicaIDs() []string {
 	return replicaIDs
 }
 
+func recoveryTaskTargetReplicaIDs(targets map[string]SessionKind) []string {
+	if len(targets) == 0 {
+		return nil
+	}
+	replicaIDs := make([]string, 0, len(targets))
+	for replicaID := range targets {
+		if replicaID == "" {
+			continue
+		}
+		replicaIDs = append(replicaIDs, replicaID)
+	}
+	sort.Strings(replicaIDs)
+	return replicaIDs
+}
+
+func removedRecoveryTaskReplicaIDs(previousTargets, currentTargets []string) []string {
+	if len(previousTargets) == 0 {
+		return nil
+	}
+	current := make(map[string]struct{}, len(currentTargets))
+	for _, replicaID := range currentTargets {
+		current[replicaID] = struct{}{}
+	}
+	removed := make([]string, 0, len(previousTargets))
+	for _, replicaID := range previousTargets {
+		if _, ok := current[replicaID]; ok {
+			continue
+		}
+		removed = append(removed, replicaID)
+	}
+	return removed
+}
+
 func (st *VolumeState) recoveryCommandReplicaIDFromEvent(replicaID string) (string, bool) {
 	if replicaID != "" {
 		for _, replica := range st.DesiredReplicas {

sw-block/engine/replication/phase14_command_test.go

@@ -241,6 +241,76 @@ func TestPhase14_CommandSequence_RebuildingAssignmentStartsRecoveryTaskWithoutRe
 	})
 }
 
+func TestPhase14_CommandSequence_AssignmentChangeDrainsRemovedRecoveryReplica(t *testing.T) {
+	core := NewCoreEngine()
+
+	core.ApplyEvent(AssignmentDelivered{
+		ID:             "vol-cmd-remove-replica",
+		Epoch:          1,
+		Role:           RolePrimary,
+		RecoveryTarget: SessionCatchUp,
+		Replicas: []ReplicaAssignment{
+			{ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.30:9333", CtrlAddr: "10.0.0.30:9334", Version: 1}},
+			{ReplicaID: "replica-2", Endpoint: Endpoint{DataAddr: "10.0.0.31:9333", CtrlAddr: "10.0.0.31:9334", Version: 1}},
+		},
+	})
+
+	result := core.ApplyEvent(AssignmentDelivered{
+		ID:             "vol-cmd-remove-replica",
+		Epoch:          2,
+		Role:           RolePrimary,
+		RecoveryTarget: SessionCatchUp,
+		Replicas: []ReplicaAssignment{
+			{ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.30:9333", CtrlAddr: "10.0.0.30:9334", Version: 1}},
+		},
+	})
+	assertCommandNames(t, result.Commands, []string{
+		"apply_role",
+		"configure_shipper",
+		"drain_recovery_task",
+		"start_recovery_task",
+		"publish_projection",
+	})
+	if got := drainRecoveryTaskReplicaIDs(result.Commands); !reflect.DeepEqual(got, []string{"replica-2"}) {
+		t.Fatalf("drain recovery task replicas=%v", got)
+	}
+	if got := recoveryTaskReplicaIDs(result.Commands); !reflect.DeepEqual(got, []string{"replica-1"}) {
+		t.Fatalf("start recovery task replicas=%v", got)
+	}
+}
+
+func TestPhase14_CommandSequence_NeedsRebuildInvalidatesOnlyAffectedReplica(t *testing.T) {
+	core := NewCoreEngine()
+
+	core.ApplyEvent(AssignmentDelivered{
+		ID:             "vol-cmd-needs-rebuild-targeted",
+		Epoch:          1,
+		Role:           RolePrimary,
+		RecoveryTarget: SessionCatchUp,
+		Replicas: []ReplicaAssignment{
+			{ReplicaID: "replica-1", Endpoint: Endpoint{DataAddr: "10.0.0.32:9333", CtrlAddr: "10.0.0.32:9334", Version: 1}},
+			{ReplicaID: "replica-2", Endpoint: Endpoint{DataAddr: "10.0.0.33:9333", CtrlAddr: "10.0.0.33:9334", Version: 1}},
+		},
+	})
+
+	result := core.ApplyEvent(NeedsRebuildObserved{
+		ID:        "vol-cmd-needs-rebuild-targeted",
+		ReplicaID: "replica-2",
+		Reason:    "gap_too_large",
+	})
+	assertCommandNames(t, result.Commands, []string{
+		"invalidate_session",
+		"publish_projection",
+	})
+	invalidate, ok := result.Commands[0].(InvalidateSessionCommand)
+	if !ok {
+		t.Fatalf("cmd0=%T", result.Commands[0])
+	}
+	if invalidate.ReplicaID != "replica-2" {
+		t.Fatalf("invalidate replica_id=%q", invalidate.ReplicaID)
+	}
+}
+
 func TestPhase14_CommandSequence_AssignmentChangeAllowsFreshRecoveryStart(t *testing.T) {
 	core := NewCoreEngine()
 
@@ -311,3 +381,15 @@ func recoveryTaskReplicaIDs(cmds []Command) []string {
 	}
 	return replicaIDs
 }
+
+func drainRecoveryTaskReplicaIDs(cmds []Command) []string {
+	var replicaIDs []string
+	for _, cmd := range cmds {
+		drain, ok := cmd.(DrainRecoveryTaskCommand)
+		if !ok {
+			continue
+		}
+		replicaIDs = append(replicaIDs, drain.ReplicaID)
+	}
+	return replicaIDs
+}

weed/server/block_recovery.go

@@ -72,7 +72,7 @@ func NewRecoveryManager(bs *BlockService) *RecoveryManager {
 // goroutines. Core-present paths should use StartRecoveryTask instead.
 func (rm *RecoveryManager) HandleAssignmentResult(result engine.AssignmentResult, assignments []blockvol.BlockVolumeAssignment) {
 	for _, replicaID := range result.Removed {
-		rm.cancelAndDrain(replicaID, true)
+		rm.cancelAndDrainWithReason(replicaID, true, "recovery_removed")
 	}
 	for _, replicaID := range result.SessionsSuperseded {
 		rm.cancelAndDrain(replicaID, false)
@@ -89,7 +89,7 @@ func (rm *RecoveryManager) HandleAssignmentResult(result engine.AssignmentResult
 // execution on the bounded live path.
 func (rm *RecoveryManager) HandleRemovedAssignments(result engine.AssignmentResult) {
 	for _, replicaID := range result.Removed {
-		rm.cancelAndDrain(replicaID, true)
+		rm.cancelAndDrainWithReason(replicaID, true, "recovery_removed")
 	}
 }
 
@@ -100,9 +100,26 @@ func (rm *RecoveryManager) StartRecoveryTask(replicaID string, assignments []blo
 	rm.startTask(replicaID, assignments)
 }
 
+// DrainRecoveryTask drains removed recovery work from an explicit core-owned
+// command seam on the core-present path.
+func (rm *RecoveryManager) DrainRecoveryTask(replicaID, reason string) {
+	if reason == "" {
+		reason = "recovery_removed"
+	}
+	rm.cancelAndDrainWithReason(replicaID, true, reason)
+}
+
 // cancelAndDrain cancels a running task and WAITS for it to exit.
 // This ensures no overlap between old and new owners.
 func (rm *RecoveryManager) cancelAndDrain(replicaID string, invalidateSession bool) {
+	reason := ""
+	if invalidateSession {
+		reason = "recovery_removed"
+	}
+	rm.cancelAndDrainWithReason(replicaID, invalidateSession, reason)
+}
+
+func (rm *RecoveryManager) cancelAndDrainWithReason(replicaID string, invalidateSession bool, reason string) {
 	rm.mu.Lock()
 	task, ok := rm.tasks[replicaID]
 	if !ok {
@@ -113,7 +130,7 @@ func (rm *RecoveryManager) cancelAndDrain(replicaID string, invalidateSession bo
 	task.cancel()
 	if invalidateSession && rm.bs.v2Orchestrator != nil {
 		if s := rm.bs.v2Orchestrator.Registry.Sender(replicaID); s != nil {
-			s.InvalidateSession("recovery_removed", engine.StateDisconnected)
+			s.InvalidateSession(reason, engine.StateDisconnected)
 		}
 	}
 	delete(rm.tasks, replicaID)

weed/server/blockcmd/dispatch.go

@@ -14,7 +14,8 @@ type Ops interface {
 	StartReceiver(assignment blockvol.BlockVolumeAssignment) (bool, error)
 	ConfigureShipper(volumeID string, replicas []engine.ReplicaAssignment) (executed bool, shipperConnected bool, err error)
 	StartRecoveryTask(replicaID string, assignment blockvol.BlockVolumeAssignment) (bool, error)
-	InvalidateSession(volumeID, reason string) (bool, error)
+	DrainRecoveryTask(replicaID, reason string) (bool, error)
+	InvalidateSession(volumeID, replicaID, reason string) (bool, error)
 	StartCatchUp(replicaID string, targetLSN uint64) (bool, error)
 	StartRebuild(replicaID string, targetLSN uint64) (bool, error)
 }
@@ -95,8 +96,19 @@ func (d *Dispatcher) Run(cmds []engine.Command, assignment *blockvol.BlockVolume
 			if executed {
 				d.effects.RecordCommand(v.VolumeID, "start_recovery_task")
 			}
+		case engine.DrainRecoveryTaskCommand:
+			if v.ReplicaID == "" {
+				continue
+			}
+			executed, err := d.ops.DrainRecoveryTask(v.ReplicaID, v.Reason)
+			if err != nil {
+				return err
+			}
+			if executed {
+				d.effects.RecordCommand(v.VolumeID, "drain_recovery_task")
+			}
 		case engine.InvalidateSessionCommand:
-			executed, err := d.ops.InvalidateSession(v.VolumeID, v.Reason)
+			executed, err := d.ops.InvalidateSession(v.VolumeID, v.ReplicaID, v.Reason)
 			if err != nil {
 				return err
 			}

weed/server/blockcmd/dispatch_test.go

@@ -14,7 +14,8 @@ type fakeOps struct {
 	startReceiverFn     func(blockvol.BlockVolumeAssignment) (bool, error)
 	configureShipperFn  func(string, []engine.ReplicaAssignment) (bool, bool, error)
 	startRecoveryTaskFn func(string, blockvol.BlockVolumeAssignment) (bool, error)
-	invalidateSessionFn func(string, string) (bool, error)
+	drainRecoveryTaskFn func(string, string) (bool, error)
+	invalidateSessionFn func(string, string, string) (bool, error)
 	startCatchUpFn      func(string, uint64) (bool, error)
 	startRebuildFn      func(string, uint64) (bool, error)
 }
@@ -47,11 +48,18 @@ func (f fakeOps) StartRecoveryTask(replicaID string, assignment blockvol.BlockVo
 	return f.startRecoveryTaskFn(replicaID, assignment)
 }
 
-func (f fakeOps) InvalidateSession(volumeID, reason string) (bool, error) {
+func (f fakeOps) DrainRecoveryTask(replicaID, reason string) (bool, error) {
+	if f.drainRecoveryTaskFn == nil {
+		return false, nil
+	}
+	return f.drainRecoveryTaskFn(replicaID, reason)
+}
+
+func (f fakeOps) InvalidateSession(volumeID, replicaID, reason string) (bool, error) {
 	if f.invalidateSessionFn == nil {
 		return false, nil
 	}
-	return f.invalidateSessionFn(volumeID, reason)
+	return f.invalidateSessionFn(volumeID, replicaID, reason)
 }
 
 func (f fakeOps) StartCatchUp(replicaID string, targetLSN uint64) (bool, error) {
@@ -188,7 +196,11 @@ func TestDispatcher_StopsOnFirstError(t *testing.T) {
 type fakeRecoveryCoordinator struct {
 	startedReplica string
 	startedAssigns []blockvol.BlockVolumeAssignment
-	catchUpCalls   []struct {
+	drained        []struct {
+		replicaID string
+		reason    string
+	}
+	catchUpCalls []struct {
 		replicaID string
 		targetLSN uint64
 	}
@@ -203,6 +215,13 @@ func (f *fakeRecoveryCoordinator) StartRecoveryTask(replicaID string, assignment
 	f.startedAssigns = assignments
 }
 
+func (f *fakeRecoveryCoordinator) DrainRecoveryTask(replicaID, reason string) {
+	f.drained = append(f.drained, struct {
+		replicaID string
+		reason    string
+	}{replicaID: replicaID, reason: reason})
+}
+
 func (f *fakeRecoveryCoordinator) ExecutePendingCatchUp(replicaID string, targetLSN uint64) error {
 	f.catchUpCalls = append(f.catchUpCalls, struct {
 		replicaID string
@@ -268,6 +287,29 @@ func TestServiceOps_StartRecoveryTaskUsesRecoveryCoordinator(t *testing.T) {
 	}
 }
 
+func TestDispatcher_DrainRecoveryTaskRecordsExecution(t *testing.T) {
+	effects := &fakeEffects{}
+	var drained []string
+	d := NewDispatcher(fakeOps{
+		drainRecoveryTaskFn: func(replicaID, reason string) (bool, error) {
+			drained = append(drained, replicaID+":"+reason)
+			return true, nil
+		},
+	}, effects)
+	err := d.Run([]engine.Command{
+		engine.DrainRecoveryTaskCommand{VolumeID: "vol1", ReplicaID: "vol1/vs2", Reason: "assignment_removed"},
+	}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(drained, []string{"vol1/vs2:assignment_removed"}) {
+		t.Fatalf("drained=%v", drained)
+	}
+	if !reflect.DeepEqual(effects.recorded, []string{"vol1:drain_recovery_task"}) {
+		t.Fatalf("recorded=%v", effects.recorded)
+	}
+}
+
 func TestHostEffects_PublishProjectionPrefersLatestCoreProjection(t *testing.T) {
 	cache := &fakeProjectionCache{}
 	effects := NewHostEffects(
@@ -342,7 +384,7 @@ func TestServiceOps_InvalidateSessionUsesProjectionAndSenderResolver(t *testing.
 			}
 		},
 	)
-	executed, err := ops.InvalidateSession("vol1", "test_reason")
+	executed, err := ops.InvalidateSession("vol1", "", "test_reason")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -357,6 +399,60 @@ func TestServiceOps_InvalidateSessionUsesProjectionAndSenderResolver(t *testing.
 	}
 }
 
+func TestServiceOps_InvalidateSessionTargetsSingleReplicaWhenProvided(t *testing.T) {
+	s1 := &fakeSessionInvalidator{}
+	s2 := &fakeSessionInvalidator{}
+	ops := NewServiceOps(
+		fakeOps{},
+		nil,
+		fakeProjectionReader{
+			ok: true,
+			proj: engine.PublicationProjection{
+				VolumeID:   "vol1",
+				ReplicaIDs: []string{"vol1/vs2", "vol1/vs3"},
+			},
+		},
+		func(replicaID string) SessionInvalidator {
+			switch replicaID {
+			case "vol1/vs2":
+				return s1
+			case "vol1/vs3":
+				return s2
+			default:
+				return nil
+			}
+		},
+	)
+	executed, err := ops.InvalidateSession("vol1", "vol1/vs2", "test_reason")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !executed {
+		t.Fatal("expected executed")
+	}
+	if !reflect.DeepEqual(s1.reasons, []string{"test_reason"}) {
+		t.Fatalf("reasons1=%v", s1.reasons)
+	}
+	if len(s2.reasons) != 0 {
+		t.Fatalf("reasons2=%v", s2.reasons)
+	}
+}
+
+func TestServiceOps_DrainRecoveryTaskUsesRecoveryCoordinator(t *testing.T) {
+	rec := &fakeRecoveryCoordinator{}
+	ops := NewServiceOps(fakeOps{}, rec, nil, nil)
+	executed, err := ops.DrainRecoveryTask("vol1/vs2", "assignment_removed")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !executed {
+		t.Fatal("expected executed")
+	}
+	if len(rec.drained) != 1 || rec.drained[0].replicaID != "vol1/vs2" || rec.drained[0].reason != "assignment_removed" {
+		t.Fatalf("drained=%v", rec.drained)
+	}
+}
+
 func TestServiceOps_StartRecoveryTask_NilRecoveryIsNoop(t *testing.T) {
 	ops := NewServiceOps(fakeOps{}, nil, nil, nil)
 	executed, err := ops.StartRecoveryTask("vol1/vs2", blockvol.BlockVolumeAssignment{Path: "vol1"})

weed/server/blockcmd/service_ops.go

@@ -16,6 +16,7 @@ type BackendOps interface {
 // RecoveryCoordinator is the runtime recovery surface used by command ops.
 type RecoveryCoordinator interface {
 	StartRecoveryTask(replicaID string, assignments []blockvol.BlockVolumeAssignment)
+	DrainRecoveryTask(replicaID, reason string)
 	ExecutePendingCatchUp(replicaID string, targetLSN uint64) error
 	ExecutePendingRebuild(replicaID string, targetLSN uint64) error
 }
@@ -85,10 +86,26 @@ func (ops *ServiceOps) StartRecoveryTask(replicaID string, assignment blockvol.B
 	return true, nil
 }
 
-func (ops *ServiceOps) InvalidateSession(volumeID, reason string) (bool, error) {
+func (ops *ServiceOps) DrainRecoveryTask(replicaID, reason string) (bool, error) {
+	if ops == nil || ops.recovery == nil {
+		return false, nil
+	}
+	ops.recovery.DrainRecoveryTask(replicaID, reason)
+	return true, nil
+}
+
+func (ops *ServiceOps) InvalidateSession(volumeID, replicaID, reason string) (bool, error) {
 	if ops == nil || ops.projection == nil || ops.senderByID == nil {
 		return false, nil
 	}
+	if replicaID != "" {
+		sender := ops.senderByID(replicaID)
+		if sender == nil {
+			return false, nil
+		}
+		sender.InvalidateSession(reason, engine.StateDisconnected)
+		return true, nil
+	}
 	proj, ok := ops.projection.Projection(volumeID)
 	if !ok {
 		return false, nil

weed/server/volume_server_block.go

@@ -473,7 +473,6 @@ func (bs *BlockService) ProcessAssignments(assignments []blockvol.BlockVolumeAss
 func (bs *BlockService) ApplyAssignments(assignments []blockvol.BlockVolumeAssignment) []error {
 	errs := make([]error, len(assignments))
 	var legacyRecoveryResults []engine.AssignmentResult
-	var removedRecoveryResults []engine.AssignmentResult
 
 	// V2 bridge: convert and deliver to engine orchestrator (Phase 08 P1).
 	// P3: skip V2 processing for repeated unchanged assignments.
@@ -499,8 +498,6 @@ func (bs *BlockService) ApplyAssignments(assignments []blockvol.BlockVolumeAssig
 					if len(result.SessionsCreated) > 0 || len(result.SessionsSuperseded) > 0 || len(result.Removed) > 0 {
 						legacyRecoveryResults = append(legacyRecoveryResults, result)
 					}
-				} else if len(result.Removed) > 0 {
-					removedRecoveryResults = append(removedRecoveryResults, result)
 				}
 			}
 		}
@@ -533,9 +530,6 @@ func (bs *BlockService) ApplyAssignments(assignments []blockvol.BlockVolumeAssig
 		for _, result := range legacyRecoveryResults {
 			bs.v2Recovery.HandleAssignmentResult(result, assignments)
 		}
-		for _, result := range removedRecoveryResults {
-			bs.v2Recovery.HandleRemovedAssignments(result)
-		}
 	}
 	return errs
 }

weed/server/volume_server_block_test.go

@@ -583,6 +583,99 @@ func TestBlockService_ApplyAssignments_PrimaryMultiReplica_UsesCoreStartRecovery
 	}
 }
 
+func TestBlockService_ApplyAssignments_RemovedReplica_UsesCoreDrainRecoveryTask(t *testing.T) {
+	bs := newTestBlockServiceDirect(t)
+	bs.v2Bridge = newTestControlBridge()
+	bs.v2Orchestrator = newTestOrchestrator()
+	bs.v2Recovery = NewRecoveryManager(bs)
+	defer bs.v2Recovery.Shutdown()
+
+	path := createTestVolDirect(t, bs, "vol-core-cmd-drain-removed")
+	if err := bs.blockStore.WithVolume(path, func(vol *blockvol.BlockVol) error {
+		for i := 0; i < 5; i++ {
+			if err := vol.WriteLBA(uint64(i), make([]byte, 4096)); err != nil {
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	holdTask := make(chan struct{})
+	taskReached := make(chan struct{}, 1)
+	bs.v2Recovery.OnBeforeExecute = func(replicaID string) {
+		if replicaID == path+"/vs-2" {
+			taskReached <- struct{}{}
+			<-holdTask
+		}
+	}
+
+	errs := bs.ApplyAssignments([]blockvol.BlockVolumeAssignment{{
+		Path:            path,
+		Epoch:           1,
+		Role:            blockvol.RoleToWire(blockvol.RolePrimary),
+		LeaseTtlMs:      30000,
+		ReplicaServerID: "vs-2",
+		ReplicaDataAddr: "10.0.0.2:4260",
+		ReplicaCtrlAddr: "10.0.0.2:4261",
+	}})
+	if len(errs) != 1 || errs[0] != nil {
+		t.Fatalf("apply errs=%v", errs)
+	}
+
+	select {
+	case <-taskReached:
+	case <-time.After(5 * time.Second):
+		t.Fatal("recovery task did not reach OnBeforeExecute")
+	}
+
+	replicaID := path + "/vs-2"
+	bs.v2Recovery.mu.Lock()
+	oldTask := bs.v2Recovery.tasks[replicaID]
+	bs.v2Recovery.mu.Unlock()
+	if oldTask == nil {
+		t.Fatal("expected active recovery task before removal")
+	}
+	oldDone := oldTask.done
+
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		close(holdTask)
+	}()
+
+	errs = bs.ApplyAssignments([]blockvol.BlockVolumeAssignment{{
+		Path:       path,
+		Epoch:      2,
+		Role:       blockvol.RoleToWire(blockvol.RolePrimary),
+		LeaseTtlMs: 30000,
+	}})
+	if len(errs) != 1 || errs[0] != nil {
+		t.Fatalf("remove apply errs=%v", errs)
+	}
+
+	select {
+	case <-oldDone:
+	case <-time.After(5 * time.Second):
+		t.Fatal("removed recovery task did not drain")
+	}
+
+	if got := bs.v2Recovery.ActiveTaskCount(); got != 0 {
+		t.Fatalf("active tasks=%d, want 0", got)
+	}
+	if sender := bs.v2Orchestrator.Registry.Sender(replicaID); sender != nil {
+		t.Fatalf("removed sender should be gone, got %v", sender)
+	}
+
+	cmds := bs.ExecutedCoreCommands(path)
+	if got := countCommandName(cmds, "start_recovery_task"); got != 1 {
+		t.Fatalf("start_recovery_task count=%d cmds=%v", got, cmds)
+	}
+	if got := countCommandName(cmds, "drain_recovery_task"); got != 1 {
+		t.Fatalf("drain_recovery_task count=%d cmds=%v", got, cmds)
+	}
+}
+
 func TestBlockService_ApplyAssignments_RebuildingRole_UsesCoreRecoveryPathWithoutLegacyDirectStart(t *testing.T) {
 	bs := newTestBlockServiceDirect(t)
 	bs.v2Bridge = newTestControlBridge()
@@ -782,6 +875,48 @@ func TestBlockService_BarrierRejected_DoesNotReexecuteInvalidateOnSameReason(t *
 	}
 }
 
+func TestBlockService_NeedsRebuildObserved_InvalidatesOnlyTargetReplica(t *testing.T) {
+	bs := newTestBlockServiceDirect(t)
+	bs.v2Bridge = newTestControlBridge()
+	bs.v2Orchestrator = newTestOrchestrator()
+	path := createTestVolDirect(t, bs, "vol-core-cmd-targeted-invalidate")
+
+	errs := bs.ApplyAssignments([]blockvol.BlockVolumeAssignment{{
+		Path:       path,
+		Epoch:      1,
+		Role:       blockvol.RoleToWire(blockvol.RolePrimary),
+		LeaseTtlMs: 30000,
+		ReplicaAddrs: []blockvol.ReplicaAddr{
+			{ServerID: "vs-2", DataAddr: "10.0.0.2:4260", CtrlAddr: "10.0.0.2:4261"},
+			{ServerID: "vs-3", DataAddr: "10.0.0.3:4260", CtrlAddr: "10.0.0.3:4261"},
+		},
+	}})
+	if len(errs) != 1 || errs[0] != nil {
+		t.Fatalf("apply errs=%v", errs)
+	}
+
+	replica2 := bs.v2Orchestrator.Registry.Sender(path + "/vs-2")
+	replica3 := bs.v2Orchestrator.Registry.Sender(path + "/vs-3")
+	if replica2 == nil || replica3 == nil {
+		t.Fatalf("senders not found: vs2=%v vs3=%v", replica2 != nil, replica3 != nil)
+	}
+	if !replica2.HasActiveSession() || !replica3.HasActiveSession() {
+		t.Fatal("both senders should start with active sessions")
+	}
+
+	bs.applyCoreEvent(engine.NeedsRebuildObserved{ID: path, ReplicaID: path + "/vs-2", Reason: "gap_too_large"})
+
+	if replica2.HasActiveSession() {
+		t.Fatal("target replica session should be invalidated")
+	}
+	if !replica3.HasActiveSession() {
+		t.Fatal("non-target replica session should remain active")
+	}
+	if got := bs.ExecutedCoreCommands(path); countCommandName(got, "invalidate_session") != 1 {
+		t.Fatalf("executed commands=%v", got)
+	}
+}
+
 func TestBlockService_DebugInfoForVolume_UsesCoreProjectionPrimaryPath(t *testing.T) {
 	bs := newTestBlockServiceDirect(t)
 	path := createTestVolDirect(t, bs, "vol-debug-primary")