diff --git a/.github/workflows/prepare-longhorn-v2-node.yaml b/.github/workflows/prepare-longhorn-v2-node.yaml
index bae2544077..492e3f549d 100644
--- a/.github/workflows/prepare-longhorn-v2-node.yaml
+++ b/.github/workflows/prepare-longhorn-v2-node.yaml
@@ -32,6 +32,7 @@ name: Prepare Longhorn V2 Migration
             inspect-host-memory,
             reboot-host-for-hugepages,
             observe-database,
+            register-temporary-v2-storage,
             repair-v1-reservation,
             reserve-v1-scratch,
             retain-v1-source,
@@ -151,6 +152,9 @@ jobs:
             benchmark-v1|benchmark-v2)
               [[ "${LONGHORN_V2_CONFIRM}" == benchmark-longhorn-storage ]]
               ;;
+            register-temporary-v2-storage)
+              [[ "${LONGHORN_V2_CONFIRM}" == register-longhorn-v2-temporary-storage ]]
+              ;;
             *) exit 1 ;;
           esac
       - name: Checkout reviewed source
diff --git a/internal/longhornmigration/activation.go b/internal/longhornmigration/activation.go
new file mode 100644
index 0000000000..9d7c18f519
--- /dev/null
+++ b/internal/longhornmigration/activation.go
@@ -0,0 +1,137 @@
+package longhornmigration
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"regexp"
+	"strings"
+
+	"github.com/devantler-tech/platform/internal/strictjson"
+)
+
+const (
+	// ActivationJournal is the fixed three-worker raw registration receipt.
+	ActivationJournal = "longhorn-v2-temporary-activation"
+	// ActivationOwner binds registered storage to that receipt's Kubernetes UID.
+	ActivationOwner = "storage.platform.devantler.tech/v2-activation"
+)
+
+// ActivationBinding retains the native identities verified before registration.
+type ActivationBinding struct {
+	Selection        string `json:"selection"`
+	Name             string `json:"name"`
+	NodeUID          string `json:"nodeUID"`
+	ProviderID       string `json:"providerID"`
+	MachineUUID      string `json:"machineUUID"`
+	BootID           string `json:"bootID"`
+	OSImage          string `json:"osImage"`
+	LonghornUID      string `json:"longhornUID"`
+	EphemeralUUID    string `json:"ephemeralUUID"`
+	RawPartitionUUID string `json:"rawPartitionUUID"`
+	PreparationUID   string `json:"preparationUID,omitempty"`
+}
+
+// SameStorageIdentity compares the durable worker, partition and preparation
+// identities. Boot and OS observations describe the original read bracket; a
+// later caller must freshly prove its own bracket rather than reuse those values.
+func (b ActivationBinding) SameStorageIdentity(other ActivationBinding) bool {
+	b.BootID, b.OSImage = "", ""
+	other.BootID, other.OSImage = "", ""
+	return b == other
+}
+
+// ActivationRecord is shared by the registering writer and publication readers.
+type ActivationRecord struct {
+	Version   int                          `json:"version"`
+	Source    string                       `json:"source"`
+	OwnerRun  string                       `json:"ownerRun"`
+	Targets   map[string]ActivationBinding `json:"targets"`
+	Completed bool                         `json:"completed"`
+}
+
+var errActivation = errors.New("temporary raw storage identity or registration ownership is unproven")
+
+// ActivationSelections returns the fixed roster without exposing mutable state.
+func ActivationSelections() [3]string {
+	return [3]string{"temporary-v1-1", "temporary-v1-2", "temporary-v2-1"}
+}
+
+// DecodeActivationRecord rejects oversized, duplicate, aliased and unknown JSON.
+func DecodeActivationRecord(data []byte) (ActivationRecord, error) {
+	if len(data) == 0 || len(data) > 32<<10 {
+		return ActivationRecord{}, errActivation
+	}
+	canonical := map[string][]string{"": {"version", "source", "ownerRun", "targets", "completed"}}
+	fields := []string{"selection", "name", "nodeUID", "providerID", "machineUUID", "bootID", "osImage", "longhornUID", "ephemeralUUID", "rawPartitionUUID", "preparationUID"}
+	for _, selection := range ActivationSelections() {
+		canonical["targets."+selection] = fields
+	}
+	if strictjson.Validate(data, canonical) != nil {
+		return ActivationRecord{}, errActivation
+	}
+	var record ActivationRecord
+	decoder := json.NewDecoder(bytes.NewReader(data))
+	decoder.DisallowUnknownFields()
+	if decoder.Decode(&record) != nil || record.Validate() != nil {
+		return ActivationRecord{}, errActivation
+	}
+	return record, nil
+}
+
+// Validate requires the three distinct workers, partitions and preparation UIDs.
+// It validates receipt structure; native callers still establish live acceptance.
+func (r ActivationRecord) Validate() error {
+	if r.Version != 1 || !regexp.MustCompile(`^[0-9a-f]{40}$`).MatchString(r.Source) || !regexp.MustCompile(`^[1-9][0-9]*$`).MatchString(r.OwnerRun) || len(r.Targets) != 3 {
+		return errActivation
+	}
+	uuid := regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`)
+	seen := [7]map[string]bool{}
+	preparations := map[string]bool{}
+	for i := range seen {
+		seen[i] = map[string]bool{}
+	}
+	for _, selection := range ActivationSelections() {
+		b, ok := r.Targets[selection]
+		if !ok || b.Selection != selection || !regexp.MustCompile(`^autoscale-longhorn-(v[12])-[0-9a-f]{16}$`).MatchString(b.Name) ||
+			!strings.HasPrefix(b.Name, "autoscale-longhorn-"+strings.Split(selection, "-")[1]+"-") ||
+			b.NodeUID == "" || !regexp.MustCompile(`^hcloud://[1-9][0-9]*$`).MatchString(b.ProviderID) || b.MachineUUID == "" ||
+			!uuid.MatchString(b.BootID) || b.OSImage == "" || b.LonghornUID == "" || b.EphemeralUUID == "" ||
+			!uuid.MatchString(strings.ToLower(b.RawPartitionUUID)) || strings.EqualFold(b.RawPartitionUUID, b.EphemeralUUID) {
+			return errActivation
+		}
+		if b.PreparationUID == "" || preparations[b.PreparationUID] {
+			return errActivation
+		}
+		preparations[b.PreparationUID] = true
+		for i, identity := range []string{b.Name, b.NodeUID, b.ProviderID, strings.ToLower(b.MachineUUID), b.LonghornUID, strings.ToLower(b.EphemeralUUID), strings.ToLower(b.RawPartitionUUID)} {
+			if seen[i][identity] {
+				return errActivation
+			}
+			seen[i][identity] = true
+		}
+	}
+	return nil
+}
+
+// registeredRawTransition checks stable ownership for the read-only pool
+// exception. A later normal reboot or OS upgrade does not revoke registration;
+// destructive operations continue to require their separate native boot proof.
+func registeredRawTransition(n, lh, cm Object) bool {
+	if text(cm, "metadata", "name") != ActivationJournal || text(cm, "metadata", "namespace") != "longhorn-system" ||
+		text(cm, "metadata", "uid") == "" || value(cm, "metadata", "deletionTimestamp") != nil ||
+		text(lh, "metadata", "annotations", ActivationOwner) != text(cm, "metadata", "uid") {
+		return false
+	}
+	r, err := DecodeActivationRecord([]byte(text(cm, "data", "record")))
+	if err != nil || !r.Completed {
+		return false
+	}
+	for _, b := range r.Targets {
+		if b.Name == text(n, "metadata", "name") {
+			return b.NodeUID == text(n, "metadata", "uid") && b.ProviderID == text(n, "spec", "providerID") &&
+				strings.EqualFold(b.MachineUUID, text(n, "status", "nodeInfo", "systemUUID")) && b.LonghornUID == text(lh, "metadata", "uid")
+		}
+	}
+	return false
+}
diff --git a/internal/longhornmigration/pools.go b/internal/longhornmigration/pools.go
index 2ce6d9d204..9e98931ab3 100644
--- a/internal/longhornmigration/pools.go
+++ b/internal/longhornmigration/pools.go
@@ -119,7 +119,7 @@ func VerifyNode(n Object, pool Pool) error {
 
 // Eligible grants no writes. Readers still enforce ordinary compute-node
 // exclusion on false, and treat every returned error as a hold.
-func Eligible(data []byte, n, lh Object) (bool, error) {
+func Eligible(data []byte, n, lh, activation Object) (bool, error) {
 	enabled := text(lh, "metadata", "labels", EnabledLabel)
 	if enabled == "" {
 		return false, nil
@@ -149,14 +149,20 @@ func Eligible(data []byte, n, lh Object) (bool, error) {
 		return false, errors.New("migration disk scope is incomplete")
 	}
 	var disk Object
-	if role == "v1" {
+	if role == "v1" && disks["longhorn-v2"] == nil {
 		disk, _ = disks["migration-v1"].(map[string]any)
 		if text(disk, "path") != "/var/lib/longhorn" || text(disk, "diskType") != "filesystem" {
 			return false, errors.New("migration filesystem differs from its reviewed scope")
 		}
 	} else {
+		if RequiresActivationRecord(n, lh) && !registeredRawTransition(n, lh, activation) {
+			return false, errors.New("migration raw transition lacks its completed identity-bound registration")
+		}
 		disk, _ = disks["longhorn-v2"].(map[string]any)
-		if text(disk, "path") != "/dev/mapper/luks2-r-longhorn" || text(disk, "diskType") != "block" || text(disk, "diskDriver") != "aio" {
+		path := text(disk, "path")
+		standard := path == "/dev/mapper/luks2-r-longhorn"
+		staged := role == "v2" && path == "/dev/mapper/luks2-"+RetainedStageRawID
+		if (!standard && !staged) || text(disk, "diskType") != "block" || text(disk, "diskDriver") != "aio" {
 			return false, errors.New("migration raw disk differs from its reviewed scope")
 		}
 	}
@@ -165,3 +171,12 @@ func Eligible(data []byte, n, lh Object) (bool, error) {
 	}
 	return true, nil
 }
+
+// RequiresActivationRecord identifies storage transitions that cannot use the
+// original pool's scheduling exception without a completed registration.
+func RequiresActivationRecord(node, storage Object) bool {
+	role := text(node, "metadata", "labels", PoolLabel)
+	disk := value(storage, "spec", "disks", "longhorn-v2")
+	return disk != nil && (role == "v1" || role == "v2" &&
+		text(storage, "spec", "disks", "longhorn-v2", "path") == "/dev/mapper/luks2-"+RetainedStageRawID)
+}
diff --git a/internal/longhornmigration/pools_test.go b/internal/longhornmigration/pools_test.go
index a0b3b24e66..79249434e8 100644
--- a/internal/longhornmigration/pools_test.go
+++ b/internal/longhornmigration/pools_test.go
@@ -2,6 +2,7 @@ package longhornmigration
 
 import (
 	"encoding/json"
+	"fmt"
 	"strings"
 	"testing"
 )
@@ -10,6 +11,42 @@ func sourceFixture() []byte {
 	return []byte(`{"spec":{"cluster":{"controlPlanes":3,"workers":3,"autoscaler":{"node":{"maxNodesTotal":9,"pools":[{"name":"autoscale-longhorn-v1","serverType":"cpx42","location":"fsn1","min":2,"max":2,"labels":{"storage.platform.devantler.tech/migration-pool":"v1"}},{"name":"autoscale-longhorn-v2","serverType":"cpx42","location":"fsn1","min":1,"max":1,"labels":{"storage.platform.devantler.tech/migration-pool":"v2"}}]}}},"provider":{"hetzner":{"serverLimit":10}}}}`)
 }
 
+func TestStagedRawDiskRequiresItsCompletedRegistration(t *testing.T) {
+	n, lh := eligibilityFixture("v2")
+	value(lh, "spec", "disks", "longhorn-v2").(Object)["path"] = "/dev/mapper/luks2-" + RetainedStageRawID
+	lh["metadata"].(Object)["annotations"] = Object{ActivationOwner: "owned-registration"}
+	record := ActivationRecord{Version: 1, Source: strings.Repeat("a", 40), OwnerRun: "12", Completed: true, Targets: map[string]ActivationBinding{}}
+	for i, selection := range ActivationSelections() {
+		role := strings.Split(selection, "-")[1]
+		record.Targets[selection] = ActivationBinding{
+			Selection: selection, Name: fmt.Sprintf("autoscale-longhorn-%s-%016x", role, i),
+			NodeUID: fmt.Sprintf("node-%d", i), ProviderID: fmt.Sprintf("hcloud://%d", i+1),
+			MachineUUID: fmt.Sprintf("machine-%d", i), BootID: "12345678-1234-4234-8234-000000000099", OSImage: "Talos fixture",
+			LonghornUID: fmt.Sprintf("longhorn-%d", i), EphemeralUUID: fmt.Sprintf("ephemeral-%d", i),
+			RawPartitionUUID: fmt.Sprintf("12345678-1234-4234-8234-%012x", i), PreparationUID: fmt.Sprintf("prep-%d", i),
+		}
+	}
+	stage := record.Targets["temporary-v2-1"]
+	stage.Name, stage.NodeUID, stage.ProviderID = text(n, "metadata", "name"), text(n, "metadata", "uid"), text(n, "spec", "providerID")
+	stage.MachineUUID, stage.LonghornUID = text(n, "status", "nodeInfo", "systemUUID"), text(lh, "metadata", "uid")
+	record.Targets[stage.Selection] = stage
+	cm := Object{"metadata": Object{"name": ActivationJournal, "namespace": "longhorn-system", "uid": "owned-registration"}}
+	for _, complete := range []bool{true, false} {
+		record.Completed = complete
+		encoded, err := json.Marshal(record)
+		if err != nil {
+			t.Fatal(err)
+		}
+		cm["data"] = Object{"record": string(encoded)}
+		ok, err := Eligible(sourceFixture(), n, lh, cm)
+		if ok != complete || (err == nil) != complete {
+			t.Fatalf("staged raw eligibility completed=%t got %t: %v", complete, ok, err)
+		}
+	}
+}
+
+// eligibilityFixture binds one fixed pool member to owned scale-down protection
+// and its role-specific storage, independently of the activation writer.
 func eligibilityFixture(role string) (Object, Object) {
 	name := "autoscale-longhorn-" + role + "-0123456789abcdef"
 	n := Object{"metadata": Object{"name": name, "uid": "kube-uid", "resourceVersion": "1", "labels": Object{PoolLabel: role, "ksail.io/autoscaled": "true", "node.kubernetes.io/instance-type": "cpx42"}, "annotations": Object{ScaleDownDisabled: "true", OwnerAnnotation: OwnerValue}}, "spec": Object{"providerID": "hcloud://42"}, "status": Object{"nodeInfo": Object{"systemUUID": "machine"}, "conditions": []any{Object{"type": "Ready", "status": "True"}}}}
@@ -25,10 +62,12 @@ func eligibilityFixture(role string) (Object, Object) {
 	return n, lh
 }
 
+// Only fixed reviewed pool members with intact ownership and storage can escape
+// ordinary compute-node exclusion; missing or altered authorization holds.
 func TestOnlyReviewedPinnedMigrationPoolsAreStorageExceptions(t *testing.T) {
 	for _, role := range []string{"v1", "v2"} {
 		n, lh := eligibilityFixture(role)
-		if ok, err := Eligible(sourceFixture(), n, lh); !ok || err != nil {
+		if ok, err := Eligible(sourceFixture(), n, lh, nil); !ok || err != nil {
 			t.Fatalf("verified %s pool rejected: %v", role, err)
 		}
 	}
@@ -55,13 +94,14 @@ func TestOnlyReviewedPinnedMigrationPoolsAreStorageExceptions(t *testing.T) {
 		t.Run(name, func(t *testing.T) {
 			n, lh := eligibilityFixture("v1")
 			mutate(n, lh)
-			if ok, err := Eligible(sourceFixture(), n, lh); ok || err == nil {
+			if ok, err := Eligible(sourceFixture(), n, lh, nil); ok || err == nil {
 				t.Fatal("unverified storage exception accepted")
 			}
 		})
 	}
 }
 
+// Fixed pool capacity cannot be reduced while its storage exception is active.
 func TestFixedPoolMinimumCannotBeRemovedWhileStorageIsAuthorized(t *testing.T) {
 	n, lh := eligibilityFixture("v1")
 	var cfg Object
@@ -74,11 +114,11 @@ func TestFixedPoolMinimumCannotBeRemovedWhileStorageIsAuthorized(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if ok, err := Eligible(b, n, lh); ok || err == nil {
+	if ok, err := Eligible(b, n, lh, nil); ok || err == nil {
 		t.Fatal("a zero-minimum pool retained its storage exception")
 	}
 	delete(lh["metadata"].(Object)["labels"].(Object), EnabledLabel)
-	if ok, err := Eligible(nil, n, lh); ok || err != nil {
+	if ok, err := Eligible(nil, n, lh, nil); ok || err != nil {
 		t.Fatal("ordinary compute classification depends on temporary migration source")
 	}
 }
diff --git a/k8s/providers/hetzner/infrastructure/controllers/longhorn/helm-release.yaml b/k8s/providers/hetzner/infrastructure/controllers/longhorn/helm-release.yaml
index df6f88c4a4..790ce0ae8a 100644
--- a/k8s/providers/hetzner/infrastructure/controllers/longhorn/helm-release.yaml
+++ b/k8s/providers/hetzner/infrastructure/controllers/longhorn/helm-release.yaml
@@ -32,6 +32,19 @@ spec:
   values:
     defaultSettings:
       defaultDataPath: /var/lib/longhorn
+      # Coexist during migration; retire V1 only after all active and retained
+      # application data has been recovered and verified on V2.
+      v1DataEngine: true
+      v2DataEngine: true
+      # Each migration worker has eight CPUs. Reserve two polling reactors;
+      # preserve the V1 reservation while the original replicas remain live.
+      guaranteedInstanceManagerCPU: '{"v1":"6","v2":"25"}'
+      dataEngineCPUMask: '{"v2":"0x3"}'
+      dataEngineNumberOfCPUCores: '{"v2":"0"}'
+      dataEngineCPUIsolationEnabled: '{"v2":"true"}'
+      dataEngineInterruptModeEnabled: '{"v2":"false"}'
+      dataEngineHugepageEnabled: '{"v2":"true"}'
+      dataEngineMemorySize: '{"v2":"2048"}'
       # Match the number of workers so every volume is replicated across
       # all available nodes.
       defaultReplicaCount: ${longhorn_replica_count:=3}
diff --git a/k8s/providers/hetzner/infrastructure/controllers/longhorn/kustomization.yaml b/k8s/providers/hetzner/infrastructure/controllers/longhorn/kustomization.yaml
index ff1d681b2b..793b4dce31 100644
--- a/k8s/providers/hetzner/infrastructure/controllers/longhorn/kustomization.yaml
+++ b/k8s/providers/hetzner/infrastructure/controllers/longhorn/kustomization.yaml
@@ -8,6 +8,7 @@ resources:
   # Default WaitForFirstConsumer StorageClass (see storage-class-longhorn-wffc.yaml);
   # the chart's `longhorn` class is demoted to non-default in helm-release.yaml.
   - storage-class-longhorn-wffc.yaml
+  - storage-class-longhorn-v2.yaml
   - cilium-network-policy.yaml
   - http-route.yaml
   # Default off: no benchmark namespace currently carries the opt-in label.
diff --git a/k8s/providers/hetzner/infrastructure/controllers/longhorn/limit-range.yaml b/k8s/providers/hetzner/infrastructure/controllers/longhorn/limit-range.yaml
index e7291aa7c2..c39507bc34 100644
--- a/k8s/providers/hetzner/infrastructure/controllers/longhorn/limit-range.yaml
+++ b/k8s/providers/hetzner/infrastructure/controllers/longhorn/limit-range.yaml
@@ -15,15 +15,15 @@
 #
 # This restores ONLY the CPU limit and deliberately keeps memory unset, so the
 # OOM-cascade safeguard is preserved:
-#   * default.cpu "2" is generous and compressible — historically longhorn ran
-#     fine under the old 500m CPU cap (the incident was memory, never CPU), so a
-#     4x-higher ceiling cannot throttle a rebuild into failure; a CPU limit only
-#     slows, it never OOMKills, so it cannot trigger the IM-restart cascade.
+#   * default.cpu "4" leaves management and network work headroom above V2's
+#     two polling reactors. Longhorn sets the CPU request separately; this
+#     changes its ceiling. Recovery acceptance checks the admitted request and
+#     limit before registering disks or creating a disposable probe.
 #   * NO default.memory — this is the entire point of the longhorn exclusion and
 #     must stay unset (instance-managers must be free to burst memory during a
 #     rebuild).
 #   * defaultRequest is a low floor so a container that omits requests is not
-#     handed the 2-core limit as its request.
+#     handed the 4-core limit as its request.
 apiVersion: v1
 kind: LimitRange
 metadata:
@@ -33,7 +33,7 @@ spec:
   limits:
     - type: Container
       default:
-        cpu: "2"
+        cpu: "4"
       defaultRequest:
         cpu: 15m
         memory: 128Mi
diff --git a/k8s/providers/hetzner/infrastructure/controllers/longhorn/storage-class-longhorn-v2.yaml b/k8s/providers/hetzner/infrastructure/controllers/longhorn/storage-class-longhorn-v2.yaml
new file mode 100644
index 0000000000..82be0826c0
--- /dev/null
+++ b/k8s/providers/hetzner/infrastructure/controllers/longhorn/storage-class-longhorn-v2.yaml
@@ -0,0 +1,19 @@
+---
+# Opt-in migration destination. The V1 default and existing claims remain until
+# each application's recovered bytes and service behavior have been verified.
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: longhorn-v2
+provisioner: driver.longhorn.io
+allowVolumeExpansion: true
+reclaimPolicy: Retain
+volumeBindingMode: WaitForFirstConsumer
+parameters:
+  numberOfReplicas: "3"
+  staleReplicaTimeout: "30"
+  fsType: ext4
+  dataLocality: disabled
+  unmapMarkSnapChainRemoved: disabled
+  dataEngine: v2
+  backupTargetName: default
diff --git a/scripts/maintain-longhorn-clients/admission.go b/scripts/maintain-longhorn-clients/admission.go
index b1dc550088..4867b19807 100644
--- a/scripts/maintain-longhorn-clients/admission.go
+++ b/scripts/maintain-longhorn-clients/admission.go
@@ -71,7 +71,7 @@ func (e engine) replicaExclusion(ctx context.Context) error {
 		if !strings.HasPrefix(name, "autoscale-") {
 			continue
 		}
-		eligible, proofErr := e.migrationStorageEligible(byName[name], n)
+		eligible, proofErr := e.migrationStorageEligible(ctx, byName[name], n)
 		if proofErr != nil {
 			return proofErr
 		}
diff --git a/scripts/maintain-longhorn-clients/main.go b/scripts/maintain-longhorn-clients/main.go
index fbc480b15b..d77077f358 100644
--- a/scripts/maintain-longhorn-clients/main.go
+++ b/scripts/maintain-longhorn-clients/main.go
@@ -391,6 +391,8 @@ func (e engine) detached(ctx context.Context, j journal) (bool, error) {
 	return true, nil
 }
 
+// clientsReady verifies effective Longhorn and CSI workloads and worker
+// registration, including only journal-qualified temporary storage exceptions.
 func (e engine) clientsReady(ctx context.Context) (bool, error) {
 	for _, setting := range []struct{ name, value string }{{"system-managed-components-node-selector", ""}, {"create-default-disk-labeled-nodes", "true"}} {
 		o, err := e.c.read(ctx, "settings.longhorn.io", "longhorn-system", setting.name)
@@ -474,7 +476,7 @@ func (e engine) clientsReady(ctx context.Context) (bool, error) {
 			if l == nil {
 				return false, nil
 			}
-			eligible, proofErr := e.migrationStorageEligible(n, l)
+			eligible, proofErr := e.migrationStorageEligible(ctx, n, l)
 			if proofErr != nil {
 				return false, proofErr
 			}
diff --git a/scripts/maintain-longhorn-clients/migration_capacity.go b/scripts/maintain-longhorn-clients/migration_capacity.go
index c6ca9f62b3..d9307d9d64 100644
--- a/scripts/maintain-longhorn-clients/migration_capacity.go
+++ b/scripts/maintain-longhorn-clients/migration_capacity.go
@@ -1,12 +1,15 @@
 package main
 
 import (
+	"context"
 	"os"
 
 	"github.com/devantler-tech/platform/internal/longhornmigration"
 )
 
-func (e engine) migrationStorageEligible(node, storage object) (bool, error) {
+// migrationStorageEligible reads the shared activation receipt only for an
+// enabled migration pool, retaining ordinary-node exclusion and read failures.
+func (e engine) migrationStorageEligible(ctx context.Context, node, storage object) (bool, error) {
 	if str(storage, "metadata", "labels", longhornmigration.EnabledLabel) == "" {
 		return false, nil
 	}
@@ -18,5 +21,12 @@ func (e engine) migrationStorageEligible(node, storage object) (bool, error) {
 	if err != nil {
 		return false, err
 	}
-	return longhornmigration.Eligible(data, node, storage)
+	var activation object
+	if longhornmigration.RequiresActivationRecord(node, storage) {
+		activation, err = e.c.read(ctx, "configmap", "longhorn-system", longhornmigration.ActivationJournal)
+		if err != nil {
+			return false, err
+		}
+	}
+	return longhornmigration.Eligible(data, node, storage, activation)
 }
diff --git a/scripts/maintain-longhorn-clients/migration_capacity_test.go b/scripts/maintain-longhorn-clients/migration_capacity_test.go
index 54b0cea6c7..04583601eb 100644
--- a/scripts/maintain-longhorn-clients/migration_capacity_test.go
+++ b/scripts/maintain-longhorn-clients/migration_capacity_test.go
@@ -2,6 +2,9 @@ package main
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/devantler-tech/platform/internal/longhornmigration"
@@ -53,6 +56,114 @@ func TestClientReadinessRecognizesOnlyProtectedMigrationStorage(t *testing.T) {
 	}
 }
 
+// rawMigrationFixture represents a completed registration, independently of
+// the activation builder. A pool's original allocation role remains unchanged.
+func rawMigrationFixture(t *testing.T) (*fakeAPI, engine, object, object, object) {
+	t.Helper()
+	a, e, n, lh := protectedMigrationFixture()
+	lh["metadata"].(object)["annotations"] = object{"storage.platform.devantler.tech/v2-activation": "activation-uid"}
+	lh["spec"].(object)["disks"] = object{"longhorn-v2": object{"diskType": "block", "path": "/dev/mapper/luks2-r-longhorn", "diskDriver": "aio", "allowScheduling": true, "evictionRequested": false}}
+	targets := object{}
+	for i, selection := range []string{"temporary-v1-1", "temporary-v1-2", "temporary-v2-1"} {
+		role := "v1"
+		if i == 2 {
+			role = "v2"
+		}
+		binding := object{"selection": selection, "name": fmt.Sprintf("autoscale-longhorn-%s-%016x", role, i+1), "nodeUID": fmt.Sprintf("node-%d", i), "providerID": fmt.Sprintf("hcloud://%d", i+1), "machineUUID": fmt.Sprintf("machine-%d", i), "bootID": "11111111-1111-1111-1111-111111111111", "osImage": "Talos (fixture)", "longhornUID": fmt.Sprintf("longhorn-%d", i), "ephemeralUUID": fmt.Sprintf("ephemeral-%d", i), "rawPartitionUUID": fmt.Sprintf("22222222-2222-2222-2222-%012d", i+1)}
+		binding["preparationUID"] = fmt.Sprintf("preparation-%d", i)
+		if i == 0 {
+			binding["name"], binding["nodeUID"], binding["providerID"] = str(n, "metadata", "name"), str(n, "metadata", "uid"), "hcloud://42"
+			binding["machineUUID"], binding["longhornUID"] = "machine", str(lh, "metadata", "uid")
+		}
+		targets[selection] = binding
+	}
+	record, err := json.Marshal(object{"version": 1, "source": strings.Repeat("a", 40), "ownerRun": "1234", "completed": true, "targets": targets})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cm := resource("ConfigMap", "longhorn-system", "longhorn-v2-temporary-activation")
+	cm["metadata"].(object)["uid"] = "activation-uid"
+	cm["data"] = object{"record": string(record)}
+	a.objects[fakeKey("configmap", "longhorn-system", "longhorn-v2-temporary-activation")] = []object{cm}
+	return a, e, n, lh, cm
+}
+
+// A completed raw registration must satisfy both publication and replica
+// exclusion while the pool retains its original allocation label.
+func TestCompletedRawTransitionPreservesPublicationAndReplicaExclusion(t *testing.T) {
+	_, e, _, _, _ := rawMigrationFixture(t)
+	if ok, err := e.clientsReady(context.Background()); !ok || err != nil {
+		t.Fatalf("completed raw registration blocks production publication: %v", err)
+	}
+	if err := e.replicaExclusion(context.Background()); err != nil {
+		t.Fatalf("completed raw registration loses its storage exception: %v", err)
+	}
+}
+
+func TestStagedRawTransitionReadsItsRegistration(t *testing.T) {
+	a, e, n, lh, cm := rawMigrationFixture(t)
+	record, err := longhornmigration.DecodeActivationRecord([]byte(str(cm, "data", "record")))
+	if err != nil {
+		t.Fatal(err)
+	}
+	stage := record.Targets["temporary-v2-1"]
+	meta := n["metadata"].(object)
+	meta["name"], meta["uid"] = stage.Name, stage.NodeUID
+	meta["labels"].(object)[longhornmigration.PoolLabel] = "v2"
+	n["spec"].(object)["providerID"] = stage.ProviderID
+	n["status"].(object)["nodeInfo"].(object)["systemUUID"] = stage.MachineUUID
+	lhMeta := lh["metadata"].(object)
+	lhMeta["name"], lhMeta["uid"] = stage.Name, stage.LonghornUID
+	val(lh, "spec", "disks", "longhorn-v2").(object)["path"] = "/dev/mapper/luks2-" + longhornmigration.RetainedStageRawID
+	if ok, err := e.migrationStorageEligible(t.Context(), n, lh); !ok || err != nil {
+		t.Fatalf("completed staged registration did not reach the publication reader: %v", err)
+	}
+	delete(a.objects, fakeKey("configmap", "longhorn-system", longhornmigration.ActivationJournal))
+	if ok, err := e.migrationStorageEligible(t.Context(), n, lh); ok || err == nil {
+		t.Fatal("staged raw storage bypassed its registration")
+	}
+}
+
+// An incomplete, replaced or foreign activation must fail both independent
+// readers rather than turn a matching raw path into storage authorization.
+func TestRawTransitionCannotBorrowAnotherActivationOrWorkerIdentity(t *testing.T) {
+	for name, mutate := range map[string]func(*fakeAPI, object, object, object){
+		"missing journal": func(a *fakeAPI, _, _, _ object) {
+			delete(a.objects, fakeKey("configmap", "longhorn-system", "longhorn-v2-temporary-activation"))
+		},
+		"replaced journal": func(_ *fakeAPI, _, _ object, cm object) { cm["metadata"].(object)["uid"] = "replacement" },
+		"incomplete registration": func(_ *fakeAPI, _, _ object, cm object) {
+			cm["data"].(object)["record"] = strings.Replace(str(cm, "data", "record"), `"completed":true`, `"completed":false`, 1)
+		},
+		"replaced Kubernetes node": func(_ *fakeAPI, n, _, _ object) { n["metadata"].(object)["uid"] = "replacement" },
+		"replaced provider":        func(_ *fakeAPI, n, _, _ object) { n["spec"].(object)["providerID"] = "hcloud://99" },
+		"replaced hardware": func(_ *fakeAPI, n, _, _ object) {
+			n["status"].(object)["nodeInfo"].(object)["systemUUID"] = "replacement"
+		},
+		"replaced Longhorn node": func(_ *fakeAPI, _, lh, _ object) { lh["metadata"].(object)["uid"] = "replacement" },
+		"foreign activation owner": func(_ *fakeAPI, _, lh, _ object) {
+			lh["metadata"].(object)["annotations"].(object)["storage.platform.devantler.tech/v2-activation"] = "foreign"
+		},
+		"deleting journal": func(_ *fakeAPI, _, _ object, cm object) {
+			cm["metadata"].(object)["deletionTimestamp"] = "2026-11-03T00:00:00Z"
+		},
+		"ambiguous completion": func(_ *fakeAPI, _, _ object, cm object) {
+			cm["data"].(object)["record"] = strings.Replace(str(cm, "data", "record"), `"completed":true`, `"completed":false,"completed":true`, 1)
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			a, e, n, lh, cm := rawMigrationFixture(t)
+			mutate(a, n, lh, cm)
+			if ok, err := e.clientsReady(context.Background()); ok || err == nil {
+				t.Fatal("unbound raw registration accepted for publication")
+			}
+			if err := e.replicaExclusion(context.Background()); err == nil {
+				t.Fatal("unbound raw registration accepted for replica storage")
+			}
+		})
+	}
+}
+
 func TestReplicaExclusionAdmitsOnlyProtectedMigrationReplicas(t *testing.T) {
 	_, e, n, _ := protectedMigrationFixture()
 	if err := e.replicaExclusion(context.Background()); err != nil {
diff --git a/scripts/migrate-longhorn-v2/README.md b/scripts/migrate-longhorn-v2/README.md
index 9cc1a04b89..a2b4941d3b 100644
--- a/scripts/migrate-longhorn-v2/README.md
+++ b/scripts/migrate-longhorn-v2/README.md
@@ -98,19 +98,78 @@ may adopt it only after the previous run has ended. This operation enables no V2
 storage and removes no iSCSI support.
 
 The `Prepare Longhorn V2 Migration` workflow runs guarded steps towards
-platform#5582. It does not migrate a claim, enable V2, disable V1 or remove iSCSI.
+platform#5582. Application cutover, recovery probes, V1 retirement and iSCSI
+removal have separate acceptance gates.
 
 This tool lands in stages (platform#5655). It includes `refresh-clients`, `prepare-worker`,
 `prepare-unused-temporary-worker`, read-only replica inventories and the benchmarks.
 `refresh-clients` holds until the fixed migration workers are Ready and the host
-huge-page settings are loaded. Worker preparation is below and is dormant; enabling temporary
-capacity follows in its own change.
+huge-page settings are loaded. Host recovery and a cleaned V1 baseline precede
+temporary-disk preparation. The V2 activation change must follow those completed
+operations, rather than their source PRs alone.
+
+## Temporary V2 activation
+
+The activation manifests retain V1 and the existing default StorageClass, enable
+V2, and add the opt-in `longhorn-v2` Retain class. Eight-CPU workers reserve two
+cores for polling SPDK reactors with mask `0x3`, CPU isolation, and 2048 MiB of
+huge pages. The CPU-only LimitRange rises to four cores so management and network
+work have headroom above those reactors; it adds no memory limit. The reviewed
+Longhorn 1.13.0 instance-manager image and the admitted client resources and
+arguments must match before registration. Enabling the setting proves neither
+raw-disk readiness nor application recovery.
+
+After all three temporary workers have completed their independently confirmed
+raw-layout preparation, select `register-temporary-v2-storage` and confirm
+`register-longhorn-v2-temporary-storage`. This operation always selects all three
+fixed temporary pool members. It requires intact original V1 replicas, healthy
+database quorums, the cleaned baseline, reviewed storage capacity, no competing
+maintenance, and ready V2 clients on all six workers before its first write.
+Every selected worker must present its own 144 GiB encrypted raw partition. The
+complete volume inventory must fit on each disk with the working reserves and
+free-space margin available.
+
+The journal `longhorn-v2-temporary-activation` binds each temporary selection to
+the node, provider, hardware, boot, encrypted EPHEMERAL and raw GPT partition
+identities. Complete native reads bracket the partition observations. Each rebuilt
+worker also binds its completed preparation journal UID and the successful
+first-attempt main workflow that produced it. A prepared-looking disk without that
+record, a replaced journal or a failed preparation workflow cannot authorize
+registration. The third worker needs a completed preservation receipt and a new
+encrypted staging partition with a GPT identity distinct from its original raw
+partition. Each conditional Longhorn Node update registers the selected worker's
+reviewed mapper and adds the journal UID as its ownership annotation. Matching
+unowned disks are refused.
+An uncertain response stops the run. A new first-attempt main dispatch may resume
+an owned partial registration only after the previous workflow is terminal, and
+it revalidates all three native bindings and every remaining prerequisite.
+Completion requires three independent ready raw disks and all qualified clients;
+it grants no application cutover or original-disk reset authority.
+
+The two rebuilt workers retain their original `migration-pool=v1` allocation
+label. The shared publication and replica-exclusion checks recognize their raw
+disks and the third worker's distinct staging mapper only with the completed
+activation journal, its matching ownership UID, and the unchanged Kubernetes,
+provider, hardware and Longhorn node identities.
+An ordinary later reboot or OS upgrade does not revoke this registration.
+
+The journal retains the original boot and OS observations for audit. Later
+registration recovery and acceptance compare its durable worker, partition and
+preparation identities, while each native read still requires an unchanged live
+boot bracket. A normal reboot does not adopt replacement hardware or storage.
+
+Filesystem and CSI recovery probes follow separately under platform#5809.
+They must verify restored bytes, persistence after reattachment and owned scratch
+cleanup before any application cutover. This activation operation has no data
+movement or probe entrypoint. All original application volumes and historical
+recovery states remain protected.
 
 For the unused temporary-worker route (platform#5787), finish the client refresh and the
 cleaned V1 baseline first. Select `prepare-unused-temporary-worker`, `temporary-v1-1` or
 `temporary-v1-2`, and the matching confirmation `prepare-unused-longhorn-v2-temporary-v1-1`
 or `prepare-unused-longhorn-v2-temporary-v1-2`. Each selection is resolved from the complete,
-reviewed two-member V1 pool and one-member V2 pool. The fresh V2 pool has no reset operation.
+reviewed two-member V1 pool and one-member V2 pool. This preparation operation
+admits only the two V1-born workers.
 
 This operation wipes **only the selected temporary worker's EPHEMERAL partition** and reboots
 it. Its separate journal binds the original node UID, provider and hardware identities, boot,
diff --git a/scripts/migrate-longhorn-v2/client.go b/scripts/migrate-longhorn-v2/client.go
index be9d973099..8bc47aa79a 100644
--- a/scripts/migrate-longhorn-v2/client.go
+++ b/scripts/migrate-longhorn-v2/client.go
@@ -52,7 +52,11 @@ func verifyEncryptedLocalLayout(raw, ephemeral object) error {
 }
 
 func verifyEncryptedRawDisk(raw, ephemeral object, rawGiB uint64) error {
-	if text(raw, "metadata", "id") != "r-longhorn" || text(ephemeral, "metadata", "id") != "EPHEMERAL" || text(ephemeral, "spec", "uuid") == "" {
+	return verifyNamedEncryptedRawDisk(raw, ephemeral, "r-longhorn", rawGiB)
+}
+
+func verifyNamedEncryptedRawDisk(raw, ephemeral object, rawID string, rawGiB uint64) error {
+	if text(raw, "metadata", "id") != rawID || text(ephemeral, "metadata", "id") != "EPHEMERAL" || text(ephemeral, "spec", "uuid") == "" {
 		return errors.New("encrypted local partition identities are missing")
 	}
 	for _, volume := range []object{raw, ephemeral} {
diff --git a/scripts/migrate-longhorn-v2/main.go b/scripts/migrate-longhorn-v2/main.go
index 18a3115799..b2e32317cd 100644
--- a/scripts/migrate-longhorn-v2/main.go
+++ b/scripts/migrate-longhorn-v2/main.go
@@ -11,6 +11,10 @@ import (
 // main admits explicit migration operations through their distinct confirmations
 // and protected workflow identities before constructing any production client.
 func main() {
+	if len(os.Args) > 1 {
+		fmt.Fprintln(os.Stderr, "Migration operations accept no command-line arguments.")
+		os.Exit(1)
+	}
 	operation := os.Getenv("LONGHORN_V2_OPERATION")
 	target := os.Getenv("LONGHORN_V2_NODE")
 	var authorization error
@@ -23,6 +27,8 @@ func main() {
 		authorization = authorizeRetainedTemporaryPreparation(os.Getenv, target)
 	case "benchmark-v1", "benchmark-v2":
 		authorization = authorizeBenchmark(os.Getenv, operation[len("benchmark-"):])
+	case "register-temporary-v2-storage":
+		authorization = authorizeTemporaryActivation(os.Getenv)
 	case "inventory-replicas":
 		authorization = authorizeInventory(os.Getenv, target)
 	case "inspect-host-memory":
@@ -128,6 +134,15 @@ func main() {
 		fmt.Fprintln(os.Stderr, err)
 		os.Exit(1)
 	}
+	if operation == "register-temporary-v2-storage" {
+		activation := temporaryV2Activation{c: client{run: execute, context: "admin@prod", inventoryPools: pools}, source: os.Getenv("GITHUB_SHA"), owner: os.Getenv("GITHUB_RUN_ID"), pools: pools}
+		if err = activation.run(ctx); err != nil {
+			fmt.Fprintln(os.Stderr, "Temporary V2 registration held:", err)
+			fmt.Fprintln(os.Stderr, "Preserve its identity journal and any partial registration for explicit recovery.")
+			os.Exit(1)
+		}
+		return
+	}
 	if operation == "prepare-unused-temporary-worker" || operation == "prepare-retained-temporary-worker" {
 		c := client{run: execute, context: "admin@prod", inventoryPools: pools}
 		nodes, readErr := c.list(ctx, "nodes", "")
diff --git a/scripts/migrate-longhorn-v2/node.go b/scripts/migrate-longhorn-v2/node.go
index bdebb3ff13..0ce32f903f 100644
--- a/scripts/migrate-longhorn-v2/node.go
+++ b/scripts/migrate-longhorn-v2/node.go
@@ -35,7 +35,11 @@ func verifyNode(n object, j journal, original bool) (string, error) {
 }
 
 func blockDisks() object {
-	return object{"longhorn-v2": object{"path": "/dev/mapper/luks2-r-longhorn", "diskType": "block", "diskDriver": "aio", "allowScheduling": true, "evictionRequested": false, "storageReserved": float64(4 * 1024 * 1024 * 1024), "tags": []any{}}}
+	return blockDisksFor("r-longhorn")
+}
+
+func blockDisksFor(rawID string) object {
+	return object{"longhorn-v2": object{"path": "/dev/mapper/luks2-" + rawID, "diskType": "block", "diskDriver": "aio", "allowScheduling": true, "evictionRequested": false, "storageReserved": float64(4 * 1024 * 1024 * 1024), "tags": []any{}}}
 }
 
 func verifyDiskDefaults(n object) error {
@@ -53,12 +57,16 @@ func verifyDiskDefaults(n object) error {
 }
 
 func reviewedBlockDisks(raw any) bool {
+	return reviewedBlockDisksFor(raw, "r-longhorn")
+}
+
+func reviewedBlockDisksFor(raw any, rawID string) bool {
 	disks, ok := raw.(map[string]any)
 	if !ok || len(disks) != 1 {
 		return false
 	}
 	d, ok := disks["longhorn-v2"].(map[string]any)
-	expected := blockDisks()["longhorn-v2"].(object)
+	expected := blockDisksFor(rawID)["longhorn-v2"].(object)
 	if !ok || len(d) != len(expected) {
 		return false
 	}
diff --git a/scripts/migrate-longhorn-v2/storage_root_inventory_test.go b/scripts/migrate-longhorn-v2/storage_root_inventory_test.go
index eaba358391..b8b9846fe3 100644
--- a/scripts/migrate-longhorn-v2/storage_root_inventory_test.go
+++ b/scripts/migrate-longhorn-v2/storage_root_inventory_test.go
@@ -17,6 +17,8 @@ type storageRootReaderFixture struct {
 	change, partial, reboot bool
 	blockCalls              int
 	blockError              bool
+	retainedCalls           int
+	retainedObserved        bool
 }
 
 func (r *storageRootReaderFixture) ListBlockLayout(context.Context) (longhorninventory.BlockLayoutReport, error) {
@@ -27,6 +29,17 @@ func (r *storageRootReaderFixture) ListBlockLayout(context.Context) (longhorninv
 	return longhorninventory.BlockLayoutReport{VolumeStatuses: 2, DiscoveredVolumes: 1, EphemeralPhase: "ready", RawPhase: "waiting", Digest: strings.Repeat("a", 64)}, nil
 }
 
+func (r *storageRootReaderFixture) ListRetainedBlockLayout(_ context.Context, closed bool) (longhorninventory.BlockLayoutReport, longhorninventory.RetainedLayoutIdentity, error) {
+	r.retainedCalls++
+	if closed {
+		return longhorninventory.BlockLayoutReport{}, longhorninventory.RetainedLayoutIdentity{}, errors.New("public observation must require the original mapper")
+	}
+	return longhorninventory.BlockLayoutReport{
+		VolumeStatuses: 2, DiscoveredVolumes: 2, EphemeralPhase: "ready", RawPhase: "ready",
+		RawExtentObserved: r.retainedObserved, Digest: strings.Repeat("a", 64),
+	}, longhorninventory.RetainedLayoutIdentity{PartitionUUID: "12345678-1234-4234-8234-000000000099", ExtentDigest: strings.Repeat("b", 64)}, nil
+}
+
 func (r *storageRootReaderFixture) ListStorageRoot(context.Context) (longhorninventory.Listing, error) {
 	r.calls++
 	entries := []longhorninventory.Entry{{Name: ".", Mode: uint32(os.ModeDir), IsDir: true}, {Name: "engine-binaries", Mode: uint32(os.ModeDir), IsDir: true}, {Name: "private-parent-canary", Mode: uint32(os.ModeDir), IsDir: true}}
@@ -44,9 +57,14 @@ func (r *storageRootReaderFixture) BootID(context.Context) (string, error) {
 }
 
 func TestRootObserverBracketsNativeParentMetadataAndNeverGrantsResetClearance(t *testing.T) {
-	for _, scenario := range []string{"complete", "partial", "root changed", "peer changed", "boot changed", "close failed", "layout complete", "layout read failed"} {
+	for _, scenario := range []string{"complete", "partial", "root changed", "peer changed", "boot changed", "close failed", "layout complete", "layout read failed", "layout retained", "layout retained missing proof"} {
 		t.Run(scenario, func(t *testing.T) {
 			nodes := temporaryInventoryRoster()
+			target, selected := "temporary-v1-1", nodes[0]
+			retained := strings.HasPrefix(scenario, "layout retained")
+			if retained {
+				target, selected = "temporary-v2-1", nodes[2]
+			}
 			pools, err := configuredMigrationPools(poolConfigFixture())
 			if err != nil {
 				t.Fatal(err)
@@ -57,7 +75,7 @@ func TestRootObserverBracketsNativeParentMetadataAndNeverGrantsResetClearance(t
 					t.Fatal("parent observer escaped fixed read scope")
 				}
 				if program == "talosctl" {
-					return json.Marshal(object{"metadata": object{"id": "systeminformation", "namespace": "hardware"}, "spec": object{"uuid": text(nodes[0], "status", "nodeInfo", "systemUUID")}})
+					return json.Marshal(object{"metadata": object{"id": "systeminformation", "namespace": "hardware"}, "spec": object{"uuid": text(selected, "status", "nodeInfo", "systemUUID")}})
 				}
 				api, kind := "v1", "NodeList"
 				items := []any{}
@@ -73,7 +91,7 @@ func TestRootObserverBracketsNativeParentMetadataAndNeverGrantsResetClearance(t
 				case "--raw=" + inventoryStorageNodes:
 					api = "longhorn.io/v1beta2"
 					_, lh := temporaryInventoryFixture()
-					lh["metadata"].(object)["name"] = text(nodes[0], "metadata", "name")
+					lh["metadata"].(object)["name"] = text(selected, "metadata", "name")
 					items = []any{lh}
 				case "--raw=" + inventoryReplicas:
 					api, kind = "longhorn.io/v1beta2", "ReplicaList"
@@ -82,14 +100,18 @@ func TestRootObserverBracketsNativeParentMetadataAndNeverGrantsResetClearance(t
 				}
 				return json.Marshal(object{"apiVersion": api, "kind": kind, "metadata": object{"resourceVersion": "1"}, "items": items})
 			}}
-			reader := &storageRootReaderFixture{temporaryNativeReader: temporaryNativeReader{boot: text(nodes[0], "status", "nodeInfo", "bootID")}, change: scenario == "root changed", partial: scenario == "partial", reboot: scenario == "boot changed", blockError: scenario == "layout read failed"}
+			reader := &storageRootReaderFixture{
+				temporaryNativeReader: temporaryNativeReader{boot: text(selected, "status", "nodeInfo", "bootID")},
+				change:                scenario == "root changed", partial: scenario == "partial", reboot: scenario == "boot changed",
+				blockError: scenario == "layout read failed", retainedObserved: scenario == "layout retained",
+			}
 			closed := false
 			observe := observeStorageRoot
 			if strings.HasPrefix(scenario, "layout ") {
 				observe = observeStorageLayout
 			}
-			receipt, err := observe(context.Background(), c, "temporary-v1-1", func(_ context.Context, address string) (nativeReplicaReader, func() error, error) {
-				projection, _, projectionErr := c.inventoryProjection(context.Background(), "temporary-v1-1")
+			receipt, err := observe(context.Background(), c, target, func(_ context.Context, address string) (nativeReplicaReader, func() error, error) {
+				projection, _, projectionErr := c.inventoryProjection(context.Background(), target)
 				if projectionErr != nil || address != projection.Address {
 					t.Fatal("parent observer selected another endpoint")
 				}
@@ -101,9 +123,12 @@ func TestRootObserverBracketsNativeParentMetadataAndNeverGrantsResetClearance(t
 					return nil
 				}, nil
 			})
-			if !closed || (err == nil) != (scenario == "complete" || scenario == "layout complete") {
+			if !closed || (err == nil) != (scenario == "complete" || scenario == "layout complete" || scenario == "layout retained") {
 				t.Fatal("incomplete or changed parent metadata was accepted:", err)
 			}
+			if retained != (reader.retainedCalls == 1) || (retained && reader.blockCalls != 0) {
+				t.Fatal("retained extent observation escaped its fixed worker")
+			}
 			encoded, _ := json.Marshal(receipt)
 			if strings.Contains(string(encoded), "private-parent-canary") || strings.Contains(string(encoded), "private-close-canary") {
 				t.Fatal("private names reached the receipt")
@@ -118,6 +143,10 @@ func TestRootObserverBracketsNativeParentMetadataAndNeverGrantsResetClearance(t
 				if reader.blockCalls != 1 || receipt.Schema != 2 || receipt.BlockLayout == nil || receipt.BlockLayout.RawPhase != "waiting" {
 					t.Fatal("layout mode omitted native block facts")
 				}
+			} else if retained {
+				if receipt.Schema != 2 || receipt.BlockLayout == nil || !receipt.BlockLayout.RawExtentObserved {
+					t.Fatal("retained layout omitted its native extent proof")
+				}
 			} else if reader.blockCalls != 0 || receipt.Schema != 1 || receipt.BlockLayout != nil {
 				t.Fatal("legacy root mode expanded its read scope")
 			}
diff --git a/scripts/migrate-longhorn-v2/temporary_activation.go b/scripts/migrate-longhorn-v2/temporary_activation.go
new file mode 100644
index 0000000000..a17a9b7974
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/temporary_activation.go
@@ -0,0 +1,139 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/devantler-tech/platform/internal/longhornmigration"
+)
+
+type temporaryV2Binding = longhornmigration.ActivationBinding
+
+var errTemporaryActivation = errors.New("temporary raw storage identity or registration ownership is unproven")
+
+const temporaryActivationOwner = longhornmigration.ActivationOwner
+
+// authorizeTemporaryActivation requires a distinct confirmation for registering
+// the fixed three-disk pool through the protected first-attempt main workflow.
+func authorizeTemporaryActivation(getenv func(string) string) error {
+	if getenv("LONGHORN_V2_OPERATION") != "register-temporary-v2-storage" {
+		return errTemporaryActivation
+	}
+	return authorizeWorkflow(getenv, "register-longhorn-v2-temporary-storage")
+}
+
+type temporaryV2Activation struct {
+	c             client
+	source, owner string
+	pools         map[string]migrationPool
+	reader        inventoryReaderFactory
+}
+
+func temporaryRawID(selection string) string {
+	if selection == "temporary-v2-1" {
+		return longhornmigration.RetainedStageRawID
+	}
+	return "r-longhorn"
+}
+
+// readBinding brackets the exact native partition reads with hardware and boot
+// identity, then repeats the Kubernetes projection. An incomplete read or failed
+// close cannot produce a binding for a subsequent registration mutation.
+func (a temporaryV2Activation) readBinding(ctx context.Context, selection string, node, storage object, pool migrationPool) (binding temporaryV2Binding, err error) {
+	ctx, cancel := context.WithTimeout(ctx, 45*time.Second)
+	defer cancel()
+	if !commandContextActive(ctx) {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	address, err := verifyNode(node, journal{Target: text(node, "metadata", "name"), NodeUID: text(node, "metadata", "uid"), ProviderID: text(node, "spec", "providerID"), MachineUUID: text(node, "status", "nodeInfo", "systemUUID")}, true)
+	if err != nil || a.c.verifyHost(ctx, address, node) != nil {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	factory := a.reader
+	if factory == nil {
+		factory = newInventoryReader
+	}
+	reader, closeReader, err := factory(ctx, address)
+	if err != nil || reader == nil || closeReader == nil {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	defer func() {
+		if closeReader() != nil || !commandContextActive(ctx) {
+			binding, err = temporaryV2Binding{}, errTemporaryActivation
+		}
+	}()
+	boot, err := reader.BootID(ctx)
+	if err != nil {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	observe := func(worker object) (temporaryV2Binding, error) {
+		raw, rawErr := a.c.talosVolume(ctx, address, temporaryRawID(selection))
+		if rawErr != nil {
+			return temporaryV2Binding{}, errTemporaryActivation
+		}
+		ephemeral, ephemeralErr := a.c.talosVolume(ctx, address, "EPHEMERAL")
+		if ephemeralErr != nil {
+			return temporaryV2Binding{}, errTemporaryActivation
+		}
+		return projectTemporaryV2Binding(selection, worker, storage, raw, ephemeral, pool, boot)
+	}
+	before, err := observe(node)
+	if err != nil {
+		return temporaryV2Binding{}, err
+	}
+	fresh, err := a.c.get(ctx, "node", "", before.Name, false)
+	if err != nil {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	after, err := observe(fresh)
+	if err != nil || before != after || a.c.verifyHost(ctx, address, fresh) != nil {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	finalBoot, err := reader.BootID(ctx)
+	if err != nil || finalBoot != boot {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	return after, nil
+}
+
+// temporaryRegistrationState accepts an empty fence or this journal's exact
+// partial activation. A matching disk path without the journal UID is not owned.
+func temporaryRegistrationState(storage object, journalUID, selection string) (bool, error) {
+	owner := text(storage, "metadata", "annotations", temporaryActivationOwner)
+	if owner == "" && unusedStorageFenced(storage) {
+		return false, nil
+	}
+	if journalUID == "" || owner != journalUID || text(storage, "metadata", "labels", migrationEnabledLabel) != "true" ||
+		value(storage, "spec", "allowScheduling") != true || value(storage, "spec", "evictionRequested") != false ||
+		!reviewedBlockDisksFor(value(storage, "spec", "disks"), temporaryRawID(selection)) {
+		return false, errTemporaryActivation
+	}
+	return true, nil
+}
+
+// projectTemporaryV2Binding binds a fixed temporary selection to its pinned
+// hardware and native encrypted partitions. Both legacy pool roles receive raw
+// V2 storage; this profile never selects an original data-bearing worker.
+func projectTemporaryV2Binding(selection string, node, storage, raw, ephemeral object, pool migrationPool, nativeBoot string) (temporaryV2Binding, error) {
+	b := temporaryV2Binding{Selection: selection, Name: text(node, "metadata", "name"), NodeUID: text(node, "metadata", "uid"),
+		ProviderID: text(node, "spec", "providerID"), MachineUUID: text(node, "status", "nodeInfo", "systemUUID"),
+		BootID: text(node, "status", "nodeInfo", "bootID"), OSImage: text(node, "status", "nodeInfo", "osImage"),
+		LonghornUID: text(storage, "metadata", "uid"), EphemeralUUID: text(ephemeral, "spec", "uuid"), RawPartitionUUID: text(raw, "spec", "partitionUUID")}
+	if !inventoryTemporarySelection.MatchString(selection) || !strings.HasPrefix(selection, "temporary-"+pool.role+"-") ||
+		!inventoryTemporaryNode.MatchString(b.Name) || verifyMigrationNode(node, pool) != nil ||
+		verifyNamedEncryptedRawDisk(raw, ephemeral, temporaryRawID(selection), stagingRawDiskGiB) != nil ||
+		b.NodeUID == "" || !regexp.MustCompile(`^hcloud://[1-9][0-9]*$`).MatchString(b.ProviderID) || b.MachineUUID == "" || b.OSImage == "" ||
+		!inventoryBoot.MatchString(b.BootID) || nativeBoot != b.BootID || !inventoryBoot.MatchString(strings.ToLower(b.RawPartitionUUID)) || b.RawPartitionUUID == b.EphemeralUUID ||
+		text(node, "status", "nodeInfo", "architecture") != "amd64" || !hugepagesReady(node) || text(node, "status", "allocatable", "cpu") != "8" ||
+		value(node, "spec", "unschedulable") == true || text(node, "metadata", "annotations", fenceAnnotation) != "" ||
+		text(node, "metadata", "annotations", capacityOwnerAnnotation) != capacityOwnerValue || text(node, "metadata", "annotations", scaleDownDisabled) != "true" ||
+		text(node, "metadata", "labels", "node.longhorn.io/create-default-disk") != "" ||
+		text(storage, "metadata", "name") != b.Name || text(storage, "metadata", "namespace") != "longhorn-system" || b.LonghornUID == "" ||
+		text(storage, "metadata", "resourceVersion") == "" || value(storage, "metadata", "deletionTimestamp") != nil {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	return b, nil
+}
diff --git a/scripts/migrate-longhorn-v2/temporary_activation_authorization_test.go b/scripts/migrate-longhorn-v2/temporary_activation_authorization_test.go
new file mode 100644
index 0000000000..82e727d2d4
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/temporary_activation_authorization_test.go
@@ -0,0 +1,60 @@
+package main
+
+import (
+	"os"
+	"os/exec"
+	"testing"
+
+	"gopkg.in/yaml.v3"
+)
+
+// Registration confirmation grants neither host disruption nor another
+// operation, and cannot be reused outside the protected first main attempt.
+func TestTemporaryActivationRequiresItsOwnFirstAttemptMainConfirmation(t *testing.T) {
+	env := dispatchFixture()
+	env["LONGHORN_V2_OPERATION"] = "register-temporary-v2-storage"
+	env["LONGHORN_V2_CONFIRM"] = "register-longhorn-v2-temporary-storage"
+	getenv := func(key string) string { return env[key] }
+	if err := authorizeTemporaryActivation(getenv); err != nil {
+		t.Fatal(err)
+	}
+	for key, value := range map[string]string{"LONGHORN_V2_CONFIRM": "prepare-longhorn-v2-prod-worker-1", "GITHUB_RUN_ATTEMPT": "2", "GITHUB_REF": "refs/heads/topic", "GITHUB_EVENT_NAME": "pull_request", "GITHUB_WORKFLOW_REF": "other/workflow@main"} {
+		t.Run(key, func(t *testing.T) {
+			old := env[key]
+			env[key] = value
+			defer func() { env[key] = old }()
+			if authorizeTemporaryActivation(getenv) == nil {
+				t.Fatal("unreviewed or differently confirmed activation was accepted")
+			}
+		})
+	}
+	if authorizeHostReboot(getenv, "prod-worker-1") == nil {
+		t.Fatal("disk registration authorized host disruption")
+	}
+	env["LONGHORN_V2_OPERATION"] = "canary-v2"
+	if authorizeTemporaryActivation(getenv) == nil {
+		t.Fatal("recovery data movement was admitted as disk registration")
+	}
+}
+
+// Execute the actual workflow gate so a different operation's confirmation
+// cannot reach production credentials through the registration branch.
+func TestTemporaryActivationWorkflowRequiresTheDistinctConfirmation(t *testing.T) {
+	data, err := os.ReadFile("../../.github/workflows/prepare-longhorn-v2-node.yaml")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var workflow struct {
+		Jobs map[string]struct{ Steps []struct{ Run string } }
+	}
+	if yaml.Unmarshal(data, &workflow) != nil {
+		t.Fatal("invalid workflow")
+	}
+	for _, confirmation := range []string{"register-longhorn-v2-temporary-storage", "verify-longhorn-v2-restore", "prepare-unused-longhorn-v2-temporary-v1-1"} {
+		cmd := exec.CommandContext(t.Context(), "bash", "-c", workflow.Jobs["prepare"].Steps[0].Run)
+		cmd.Env = append(os.Environ(), "DISPATCH_REF=refs/heads/main", "GITHUB_RUN_ATTEMPT=1", "LONGHORN_V2_OPERATION=register-temporary-v2-storage", "LONGHORN_V2_CONFIRM="+confirmation)
+		if err = cmd.Run(); (err == nil) != (confirmation == "register-longhorn-v2-temporary-storage") {
+			t.Fatalf("registration confirmation %s: %v", confirmation, err)
+		}
+	}
+}
diff --git a/scripts/migrate-longhorn-v2/temporary_activation_flow_test.go b/scripts/migrate-longhorn-v2/temporary_activation_flow_test.go
new file mode 100644
index 0000000000..a41f12df6a
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/temporary_activation_flow_test.go
@@ -0,0 +1,419 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"reflect"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/devantler-tech/platform/internal/longhornmigration"
+)
+
+type activationAPI struct {
+	t                   *testing.T
+	s                   inventory
+	managers, databases []object
+	baseline, record    object
+	writes              int
+	lostResponse        bool
+	priorLive           bool
+	preparations        map[string]object
+	preparationFailed   bool
+}
+
+// newActivationAPI models complete fixed-roster reads and successful native
+// preparation, with writes restricted to the activation journal and raw nodes.
+func newActivationAPI(t *testing.T) (*activationAPI, temporaryV2Activation) {
+	t.Helper()
+	s, managers := v2ReadinessFixture()
+	for i, node := range s.nodes {
+		node["metadata"].(object)["uid"] = fmt.Sprintf("node-%d", i)
+		if strings.HasPrefix(text(node, "metadata", "name"), "prod-worker-") {
+			node["metadata"].(object)["labels"].(object)["node.kubernetes.io/instance-type"] = "cx43"
+		}
+		node["spec"] = object{"providerID": fmt.Sprintf("hcloud://%d", i+100), "unschedulable": false}
+		node["status"].(object)["nodeInfo"] = object{"systemUUID": fmt.Sprintf("hardware-%d", i), "bootID": fmt.Sprintf("12345678-1234-4234-8234-%012x", i+1), "osImage": "Talos fixture", "architecture": "amd64"}
+		node["status"].(object)["addresses"] = []any{object{"type": "InternalIP", "address": fmt.Sprintf("10.0.99.%d", i+10)}}
+	}
+	for _, lh := range s.storageNodes {
+		lh["metadata"].(object)["namespace"] = "longhorn-system"
+		lh["spec"].(object)["allowScheduling"] = true
+		lh["spec"].(object)["evictionRequested"] = false
+		if strings.HasPrefix(text(lh, "metadata", "name"), "autoscale-longhorn-") {
+			lh["spec"] = object{"allowScheduling": false, "evictionRequested": true, "disks": object{}}
+			lh["status"] = object{"diskStatus": object{}}
+		}
+	}
+	f := &activationAPI{t: t, s: s, managers: managers, baseline: unusedBaselineFixture(t, s)}
+	for key, profile := range hostDatabases {
+		namespace, name, _ := strings.Cut(key, "/")
+		db := resource(name)
+		db["metadata"].(object)["namespace"] = namespace
+		db["spec"] = object{"instances": profile.instances}
+		db["status"] = object{"readyInstances": profile.instances, "systemID": "123456789", "phase": "Cluster in healthy state"}
+		f.databases = append(f.databases, db)
+	}
+	for _, volume := range s.volumes {
+		valueObject(volume, "spec")["size"] = "1073741824"
+		valueObject(volume, "status")["actualSize"] = float64(1 << 20)
+	}
+	pools, err := configuredMigrationPools(poolConfigFixture())
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.preparations = map[string]object{}
+	for ordinal, selection := range []string{"temporary-v1-1", "temporary-v1-2", "temporary-v2-1"} {
+		name, _, err := resolveTemporaryInventoryTarget(selection, s.nodes, pools)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for nodeIndex, node := range s.nodes {
+			if text(node, "metadata", "name") != name {
+				continue
+			}
+			var lh object
+			for _, storage := range s.storageNodes {
+				if text(storage, "metadata", "name") == name {
+					lh = storage
+				}
+			}
+			binding := temporaryV2Binding{Selection: selection, Name: name, NodeUID: text(node, "metadata", "uid"), ProviderID: text(node, "spec", "providerID"),
+				MachineUUID: text(node, "status", "nodeInfo", "systemUUID"), LonghornUID: text(lh, "metadata", "uid"),
+				RawPartitionUUID: fmt.Sprintf("12345678-1234-4234-8234-%012x", nodeIndex+100)}
+			cm := temporaryPreparedFixture(t, binding)
+			cm["data"].(object)["record"] = strings.Replace(text(cm, "data", "record"), `"ownerRun":"7777"`, fmt.Sprintf(`"ownerRun":"%d"`, 7777+ordinal), 1)
+			f.preparations[selection] = cm
+		}
+	}
+	a := temporaryV2Activation{c: client{context: "fixture", run: f.run}, source: strings.Repeat("a", 40), owner: "456", pools: pools}
+	a.reader = func(_ context.Context, address string) (nativeReplicaReader, func() error, error) {
+		for _, node := range f.s.nodes {
+			addresses, _ := value(node, "status", "addresses").([]any)
+			if len(addresses) == 1 && text(addresses[0].(object), "address") == address {
+				boot := text(node, "status", "nodeInfo", "bootID")
+				return &activationBootReader{before: boot, after: boot}, func() error { return nil }, nil
+			}
+		}
+		return nil, nil, errTemporaryActivation
+	}
+	return f, a
+}
+
+// run enforces the production command boundary and identity-preconditioned
+// writes, including an uncertain response after an actual disk registration.
+func (f *activationAPI) run(_ context.Context, input []byte, program string, args ...string) ([]byte, error) {
+	f.t.Helper()
+	if program == "git" && reflect.DeepEqual(args, []string{"ls-remote", "--exit-code", "https://github.com/devantler-tech/platform.git", "refs/heads/main"}) {
+		return []byte(strings.Repeat("a", 40) + "\trefs/heads/main\n"), nil
+	}
+	if program == "gh" && len(args) == 2 && args[0] == "api" {
+		source := strings.Repeat("1", 40)
+		status := "completed"
+		conclusion := "success"
+		if strings.Contains(args[1], "/7777/") || strings.Contains(args[1], "/7778/") || strings.Contains(args[1], "/7779/") {
+			if f.preparationFailed {
+				conclusion = "failure"
+			}
+		} else if !strings.Contains(args[1], "/5678/") {
+			source = strings.Repeat("a", 40)
+			if f.priorLive {
+				status = "in_progress"
+			}
+		}
+		return json.Marshal(object{"status": status, "conclusion": conclusion, "head_sha": source, "head_branch": "main", "event": "workflow_dispatch", "run_attempt": 1, "path": ".github/workflows/prepare-longhorn-v2-node.yaml"})
+	}
+	if program == "talosctl" && len(args) == 5 && args[1] == "get" && args[4] == "--output=json" {
+		for i, node := range f.s.nodes {
+			if args[0] != "--nodes="+fmt.Sprintf("10.0.99.%d", i+10) {
+				continue
+			}
+			if args[2] == "systeminformation" && args[3] == "systeminformation" {
+				return json.Marshal(object{"metadata": object{"id": "systeminformation", "namespace": "hardware"}, "spec": object{"uuid": text(node, "status", "nodeInfo", "systemUUID")}})
+			}
+			raw, eph := rawFixture()
+			valueObject(raw, "spec")["size"] = float64(144 << 30)
+			raw["spec"].(object)["partitionUUID"] = fmt.Sprintf("12345678-1234-4234-8234-%012x", i+100)
+			eph["spec"].(object)["uuid"] = fmt.Sprintf("ephemeral-%d", i)
+			role := text(node, "metadata", "labels", migrationPoolLabel)
+			rawID := temporaryRawID("temporary-" + role + "-1")
+			if args[2] == "volumestatus" && args[3] == rawID {
+				raw["metadata"].(object)["id"] = rawID
+				return json.Marshal(raw)
+			}
+			if args[2] == "volumestatus" && args[3] == "EPHEMERAL" {
+				return json.Marshal(eph)
+			}
+		}
+	}
+	if program != "kubectl" || len(args) < 4 || args[0] != "--context=fixture" || args[1] != "--request-timeout=20s" {
+		f.t.Fatalf("activation escaped its native scope: %s %v", program, args)
+	}
+	args = args[2:]
+	if args[0] == "get" {
+		kind := args[1]
+		if args[2] == "--output=json" {
+			collections := map[string][]object{"nodes": f.s.nodes, "nodes.longhorn.io": f.s.storageNodes, "volumes.longhorn.io": f.s.volumes, "replicas.longhorn.io": f.s.replicas, "engines.longhorn.io": f.s.engines, "orphans.longhorn.io": f.s.orphans, "pods": f.s.pods, "instancemanagers.longhorn.io": f.managers, "clusters.postgresql.cnpg.io": f.databases}
+			items, ok := collections[kind]
+			if !ok {
+				f.t.Fatalf("unexpected collection %s", kind)
+			}
+			if items == nil {
+				items = []object{}
+			}
+			return json.Marshal(object{"items": items})
+		}
+		name := args[2]
+		if kind == "configmap" {
+			for selection, prepared := range f.preparations {
+				if name == "longhorn-v2-prepare-"+selection {
+					if prepared == nil {
+						return nil, nil
+					}
+					return json.Marshal(prepared)
+				}
+			}
+			if name == temporaryActivationJournal {
+				if f.record == nil {
+					return nil, nil
+				}
+				return json.Marshal(f.record)
+			}
+			if name == "longhorn-v2-benchmark-v1" {
+				return json.Marshal(f.baseline)
+			}
+		}
+		if kind == "settings.longhorn.io" {
+			settings := map[string]string{"v1-data-engine": "true", "v2-data-engine": "true", "storage-over-provisioning-percentage": "100", "storage-minimal-available-percentage": "25", "guaranteed-instance-manager-cpu": `{"v1":"6","v2":"25"}`, "data-engine-cpu-mask": `{"v2":"0x3"}`, "data-engine-number-of-cpu-cores": `{"v2":"0"}`, "data-engine-cpu-isolation-enabled": `{"v2":"true"}`, "data-engine-interrupt-mode-enabled": `{"v2":"false"}`, "data-engine-hugepage-enabled": `{"v2":"true"}`, "data-engine-memory-size": `{"v2":"2048"}`}
+			setting, ok := settings[name]
+			if !ok {
+				f.t.Fatalf("unexpected setting %s", name)
+			}
+			o := resource(name)
+			o["value"] = setting
+			return json.Marshal(o)
+		}
+		for _, o := range append(append([]object{}, f.s.nodes...), f.s.storageNodes...) {
+			if text(o, "metadata", "name") == name && (kind == "node" && text(o, "metadata", "namespace") == "" || kind == "nodes.longhorn.io" && text(o, "metadata", "namespace") == "longhorn-system") {
+				return json.Marshal(o)
+			}
+		}
+	}
+	if args[0] == "create" {
+		if f.record != nil {
+			return nil, errConflict
+		}
+		var created object
+		if json.Unmarshal(input, &created) != nil || text(created, "kind") != "ConfigMap" || text(created, "metadata", "name") != temporaryActivationJournal || text(created, "metadata", "namespace") != "longhorn-system" {
+			f.t.Fatal("activation created a resource outside its fixed journal")
+		}
+		created["metadata"].(object)["uid"] = "activation-journal-uid"
+		created["metadata"].(object)["resourceVersion"] = "1"
+		f.record = created
+		return json.Marshal(created)
+	}
+	if args[0] == "patch" {
+		var target object
+		if args[1] == "configmap" && args[2] == temporaryActivationJournal {
+			target = f.record
+		}
+		if args[1] == "nodes.longhorn.io" {
+			for _, lh := range f.s.storageNodes {
+				if text(lh, "metadata", "name") == args[2] && strings.HasPrefix(args[2], "autoscale-longhorn-") {
+					target = lh
+				}
+			}
+		}
+		var changes []object
+		if target == nil || json.Unmarshal(input, &changes) != nil || len(changes) < 3 || !reflect.DeepEqual(changes[:2], identityTests(target)) {
+			f.t.Fatal("activation write lost its exact native identity or exceeded temporary scope")
+		}
+		for _, change := range changes[2:] {
+			parts := strings.Split(strings.TrimPrefix(text(change, "path"), "/"), "/")
+			parent := target
+			for _, part := range parts[:len(parts)-1] {
+				parent = parent[part].(object)
+			}
+			parent[strings.ReplaceAll(parts[len(parts)-1], "~1", "/")] = change["value"]
+		}
+		version, _ := strconv.Atoi(text(target, "metadata", "resourceVersion"))
+		target["metadata"].(object)["resourceVersion"] = strconv.Itoa(version + 1)
+		if args[1] == "nodes.longhorn.io" {
+			f.writes++
+			target["status"] = object{"conditions": readyConditions(), "diskStatus": object{"longhorn-v2": object{"diskUUID": "disk-" + args[2], "conditions": readyConditions()}}}
+			if f.lostResponse {
+				f.lostResponse = false
+				return nil, errors.New("response lost after registration")
+			}
+		}
+		return nil, nil
+	}
+	f.t.Fatalf("unexpected activation command: %s %v", program, args)
+	return nil, nil
+}
+
+// Registered disks must remain eligible to the independent publication reader;
+// replaying the completed operation must not repeat a disk write.
+func TestTemporaryActivationRegistersOnlyThreeJournalOwnedRawDisks(t *testing.T) {
+	f, a := newActivationAPI(t)
+	if err := a.run(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	_, record, err := a.read(t.Context())
+	if err != nil || !record.Completed || f.writes != 3 {
+		t.Fatal("activation did not prove all three owned raw disks:", err)
+	}
+	if err = a.run(t.Context()); err != nil || f.writes != 3 {
+		t.Fatal("completed activation repeated a registration:", err)
+	}
+	for _, node := range f.s.nodes {
+		if text(node, "metadata", "labels", longhornmigration.PoolLabel) == "" {
+			continue
+		}
+		var storage object
+		for _, lh := range f.s.storageNodes {
+			if text(lh, "metadata", "name") == text(node, "metadata", "name") {
+				storage = lh
+			}
+		}
+		if eligible, err := longhornmigration.Eligible(poolConfigFixture(), node, storage, f.record); !eligible || err != nil {
+			t.Fatal("actual registration output blocks the shared publication gate:", err)
+		}
+	}
+}
+
+// No journal or disk may be written without successful native preparation.
+func TestTemporaryActivationHoldsBeforeWritesWithoutSuccessfulPreparation(t *testing.T) {
+	for _, mode := range []string{"missing preparation", "failed preparation", "unfinished preparation"} {
+		t.Run(mode, func(t *testing.T) {
+			f, a := newActivationAPI(t)
+			switch mode {
+			case "missing preparation":
+				f.preparations["temporary-v1-1"] = nil
+			case "failed preparation":
+				f.preparationFailed = true
+			case "unfinished preparation":
+				cm := f.preparations["temporary-v1-1"]
+				cm["data"].(object)["record"] = strings.Replace(text(cm, "data", "record"), `"phase":"prepared"`, `"phase":"reset-requested"`, 1)
+			}
+			if err := a.run(t.Context()); err == nil || f.writes != 0 || f.record != nil {
+				t.Fatal("activation wrote before successful native preparation was bound:", err)
+			}
+		})
+	}
+}
+
+func TestTemporaryActivationHoldsWhenTheDatasetOutgrowsStaging(t *testing.T) {
+	f, a := newActivationAPI(t)
+	valueObject(f.s.volumes[0], "spec")["size"] = fmt.Sprint(135 << 30)
+	if err := a.run(t.Context()); err == nil || f.writes != 0 || f.record != nil {
+		t.Fatal("activation accepted data beyond the per-worker storage budget", err)
+	}
+}
+
+// Only a later owner may resume an uncertain owned patch; an unrelated storage
+// owner's annotation cannot be adopted even when the desired disk matches.
+func TestTemporaryActivationResumesItsUncertainWriteWithoutAdoptingForeignStorage(t *testing.T) {
+	f, a := newActivationAPI(t)
+	f.lostResponse = true
+	if a.run(t.Context()) == nil || f.writes != 1 {
+		t.Fatal("uncertain registration was accepted or retried")
+	}
+	a.owner = "789"
+	if err := a.run(t.Context()); err != nil || f.writes != 3 {
+		t.Fatal("owned partial activation could not safely resume:", err)
+	}
+	f, a = newActivationAPI(t)
+	f.s.storageNodes[3]["metadata"].(object)["annotations"] = object{temporaryActivationOwner: "foreign"}
+	if a.run(t.Context()) == nil || f.writes != 0 {
+		t.Fatal("unowned storage was adopted")
+	}
+}
+
+// Completed and interrupted registration survive a normal host reboot without
+// replaying an owned disk write or rewriting the original observation evidence.
+func TestTemporaryActivationPreservesStorageIdentityAcrossReboot(t *testing.T) {
+	for _, interrupted := range []bool{false, true} {
+		t.Run(fmt.Sprintf("interrupted=%t", interrupted), func(t *testing.T) {
+			f, a := newActivationAPI(t)
+			f.lostResponse = interrupted
+			if err := a.run(t.Context()); (err != nil) != interrupted {
+				t.Fatal("initial registration did not reach the selected boundary:", err)
+			}
+			_, original, err := a.read(t.Context())
+			if err != nil {
+				t.Fatal(err)
+			}
+			for i, node := range f.s.nodes {
+				if !strings.HasPrefix(text(node, "metadata", "name"), "autoscale-longhorn-") {
+					continue
+				}
+				info := value(node, "status", "nodeInfo").(object)
+				info["bootID"] = fmt.Sprintf("12345678-1234-4234-8234-%012x", i+500)
+				info["osImage"] = "Talos upgraded fixture"
+			}
+			a.owner = "789"
+			if err = a.run(t.Context()); err != nil || f.writes != 3 {
+				t.Fatal("normal reboot invalidated intact prepared storage:", err)
+			}
+			if err = a.accepted(t.Context()); err != nil {
+				t.Fatal("completed storage was not accepted after reboot:", err)
+			}
+			_, recovered, err := a.read(t.Context())
+			if err != nil || !reflect.DeepEqual(original.Targets, recovered.Targets) {
+				t.Fatal("recovery rewrote the original registration observations:", err)
+			}
+		})
+	}
+}
+
+// Matching but unreviewed manager images must fail before disk registration.
+func TestTemporaryActivationRejectsUnqualifiedClientsBeforeRegisteringDisks(t *testing.T) {
+	f, a := newActivationAPI(t)
+	f.managers[0]["spec"].(object)["image"] = "example.invalid/unreviewed:latest"
+	value(f.s.pods[0], "spec", "containers").([]any)[0].(object)["image"] = "example.invalid/unreviewed:latest"
+	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+	defer cancel()
+	if a.run(ctx) == nil || f.writes != 0 {
+		t.Fatal("unqualified V2 clients reached raw-disk registration")
+	}
+}
+
+// Read-only acceptance must revalidate the completed workflow and current
+// storage identities rather than trust a cached completion bit.
+func TestCompletedTemporaryActivationRequiresCurrentOwnedStorage(t *testing.T) {
+	f, a := newActivationAPI(t)
+	if a.accepted(t.Context()) == nil || f.writes != 0 {
+		t.Fatal("an unregistered pool authorized the canary")
+	}
+	for _, mode := range []string{"valid", "journal replaced", "worker changed", "workflow live", "disk fenced", "preparation replaced"} {
+		t.Run(mode, func(t *testing.T) {
+			f, a := newActivationAPI(t)
+			if err := a.run(t.Context()); err != nil {
+				t.Fatal(err)
+			}
+			switch mode {
+			case "journal replaced":
+				f.record["metadata"].(object)["uid"] = "replacement"
+			case "worker changed":
+				f.s.nodes[3]["spec"].(object)["providerID"] = "hcloud://999"
+			case "workflow live":
+				f.priorLive = true
+			case "disk fenced":
+				f.s.storageNodes[3]["spec"].(object)["allowScheduling"] = false
+			case "preparation replaced":
+				f.preparations["temporary-v1-1"]["metadata"].(object)["uid"] = "replacement-preparation"
+			}
+			before, _ := json.Marshal(f.record)
+			err := a.accepted(t.Context())
+			after, _ := json.Marshal(f.record)
+			if (err == nil) != (mode == "valid") || f.writes != 3 || string(before) != string(after) {
+				t.Fatal("canary acceptance ignored changed evidence or mutated the plan:", err)
+			}
+		})
+	}
+}
diff --git a/scripts/migrate-longhorn-v2/temporary_activation_native_test.go b/scripts/migrate-longhorn-v2/temporary_activation_native_test.go
new file mode 100644
index 0000000000..cd9c3a8a21
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/temporary_activation_native_test.go
@@ -0,0 +1,89 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"reflect"
+	"testing"
+)
+
+type activationBootReader struct {
+	nativeReplicaReader
+	before, after string
+	reads         int
+}
+
+// BootID supplies independently selected observations on the two bracket sides.
+func (r *activationBootReader) BootID(context.Context) (string, error) {
+	r.reads++
+	if r.reads == 1 {
+		return r.before, nil
+	}
+	return r.after, nil
+}
+
+// A binding requires matching native boot, hardware and API identities plus a
+// successful reader close; durable identity comparison cannot waive this bracket.
+func TestTemporaryRawBindingBracketsNativeReadsAndClosesItsReader(t *testing.T) {
+	for _, mode := range []string{"valid", "boot changed", "hardware changed", "node changed", "partition unreadable", "partition changed", "close failed"} {
+		t.Run(mode, func(t *testing.T) {
+			n, storage, raw, eph, pool := temporaryRawBindingFixture()
+			n["status"].(object)["addresses"] = []any{object{"type": "InternalIP", "address": "10.0.99.10"}}
+			boot := text(n, "status", "nodeInfo", "bootID")
+			reader := &activationBootReader{before: boot, after: boot}
+			if mode == "boot changed" {
+				reader.after = "12345678-1234-4234-8234-000000000099"
+			}
+			closed, volumes := 0, 0
+			activation := temporaryV2Activation{reader: func(context.Context, string) (nativeReplicaReader, func() error, error) {
+				return reader, func() error {
+					closed++
+					if mode == "close failed" {
+						return errors.New("reader close failed")
+					}
+					return nil
+				}, nil
+			}}
+			activation.c = client{context: "fixture", run: func(_ context.Context, _ []byte, program string, args ...string) ([]byte, error) {
+				if program == "talosctl" && len(args) == 5 && args[0] == "--nodes=10.0.99.10" && args[1] == "get" && args[4] == "--output=json" {
+					switch {
+					case args[2] == "systeminformation" && args[3] == "systeminformation":
+						uuid := "hardware"
+						if mode == "hardware changed" {
+							uuid = "other"
+						}
+						return json.Marshal(object{"metadata": object{"namespace": "hardware", "id": "systeminformation"}, "spec": object{"uuid": uuid}})
+					case args[2] == "volumestatus" && args[3] == "r-longhorn":
+						volumes++
+						if mode == "partition unreadable" {
+							return nil, errors.New("native read failed")
+						}
+						if mode == "partition changed" && volumes > 2 {
+							raw["spec"].(object)["partitionUUID"] = "12345678-1234-4234-8234-000000000099"
+						}
+						return json.Marshal(raw)
+					case args[2] == "volumestatus" && args[3] == "EPHEMERAL":
+						volumes++
+						return json.Marshal(eph)
+					}
+				}
+				if program == "kubectl" && reflect.DeepEqual(args, []string{"--context=fixture", "--request-timeout=20s", "get", "node", text(n, "metadata", "name"), "--output=json"}) {
+					if mode == "node changed" {
+						n["metadata"].(object)["uid"] = "replacement"
+					}
+					return json.Marshal(n)
+				}
+				t.Fatalf("native binding escaped its read-only fixed scope: %s %v", program, args)
+				return nil, nil
+			}}
+			binding, err := activation.readBinding(t.Context(), "temporary-v1-1", n, storage, pool)
+			if (err == nil) != (mode == "valid") {
+				t.Fatal("native binding validity did not match its complete identity bracket:", err)
+			}
+			if mode == "valid" && (closed != 1 || reader.reads != 2 || volumes != 4 || binding.BootID != boot) {
+				t.Fatal("successful binding omitted a native read or close")
+			}
+		})
+	}
+}
diff --git a/scripts/migrate-longhorn-v2/temporary_activation_prepared.go b/scripts/migrate-longhorn-v2/temporary_activation_prepared.go
new file mode 100644
index 0000000000..140d8cb4f5
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/temporary_activation_prepared.go
@@ -0,0 +1,56 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+)
+
+// preparedTemporaryBinding validates a completed unused-storage reset journal
+// against the current worker and its changed boot and EPHEMERAL partition.
+func preparedTemporaryBinding(cm object, binding temporaryV2Binding) (journal, error) {
+	name := "longhorn-v2-prepare-" + binding.Selection
+	if text(cm, "metadata", "name") != name || text(cm, "metadata", "namespace") != "longhorn-system" ||
+		text(cm, "metadata", "uid") == "" || text(cm, "metadata", "resourceVersion") == "" || value(cm, "metadata", "deletionTimestamp") != nil {
+		return journal{}, errTemporaryActivation
+	}
+	record, err := decodePreparationJournal([]byte(text(cm, "data", "record")))
+	if err != nil {
+		return journal{}, errTemporaryActivation
+	}
+	p := preparation{target: binding.Name, name: name, temporarySelection: binding.Selection}
+	if p.validateRecord(record) != nil || (record.Schema != 2 && record.Schema != 3) || !record.StagingLayout || record.Phase != "prepared" || record.NodeUID != binding.NodeUID ||
+		record.ProviderID != binding.ProviderID || record.MachineUUID != binding.MachineUUID || record.LonghornUID != binding.LonghornUID ||
+		record.OriginalBootID == binding.BootID || strings.EqualFold(record.EphemeralUUID, binding.EphemeralUUID) {
+		return journal{}, errTemporaryActivation
+	}
+	if record.Schema == 3 && !strings.EqualFold(record.PreparedRawUUID, binding.RawPartitionUUID) {
+		return journal{}, errTemporaryActivation
+	}
+	return record, nil
+}
+
+// prepared requires successful completion of the matching protected preparation
+// before enabling a rebuilt worker's raw disk.
+func (a temporaryV2Activation) prepared(ctx context.Context, binding temporaryV2Binding) (temporaryV2Binding, error) {
+	if !inventoryTemporarySelection.MatchString(binding.Selection) {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	cm, err := a.c.get(ctx, "configmap", "longhorn-system", "longhorn-v2-prepare-"+binding.Selection, false)
+	if err != nil {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	record, err := preparedTemporaryBinding(cm, binding)
+	if err != nil {
+		return temporaryV2Binding{}, err
+	}
+	data, err := a.c.run(ctx, nil, "gh", "api", "repos/devantler-tech/platform/actions/runs/"+record.OwnerRun+"/attempts/1")
+	var run object
+	if err != nil || json.Unmarshal(data, &run) != nil || text(run, "status") != "completed" || text(run, "conclusion") != "success" ||
+		text(run, "head_sha") != record.Source || text(run, "head_branch") != "main" || text(run, "event") != "workflow_dispatch" ||
+		value(run, "run_attempt") != float64(1) || !reviewedWorkflowPath(text(run, "path")) {
+		return temporaryV2Binding{}, errTemporaryActivation
+	}
+	binding.PreparationUID = text(cm, "metadata", "uid")
+	return binding, nil
+}
diff --git a/scripts/migrate-longhorn-v2/temporary_activation_prepared_test.go b/scripts/migrate-longhorn-v2/temporary_activation_prepared_test.go
new file mode 100644
index 0000000000..b537891cfb
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/temporary_activation_prepared_test.go
@@ -0,0 +1,117 @@
+package main
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+// temporaryPreparedFixture records an earlier encrypted filesystem and boot,
+// both distinct from the post-reset binding used for activation.
+func temporaryPreparedFixture(t *testing.T, binding temporaryV2Binding) object {
+	t.Helper()
+	j := journal{Schema: 2, Target: binding.Name, Source: strings.Repeat("1", 40), OwnerRun: "7777", Phase: "prepared",
+		NodeUID: binding.NodeUID, ProviderID: binding.ProviderID, MachineUUID: binding.MachineUUID, LonghornUID: binding.LonghornUID,
+		EphemeralUUID: "original-ephemeral", OriginalDisks: object{}, TemporarySelection: binding.Selection,
+		NativeEmptyDigest: strings.Repeat("c", 64), OriginalBootID: "12345678-1234-4234-8234-000000000099", RawVolumeAbsent: true, StagingLayout: true}
+	if binding.Selection == "temporary-v2-1" {
+		j.Schema, j.RawVolumeAbsent = 3, false
+		j.RetainedExtentDigest = strings.Repeat("f", 64)
+		j.RetainedRawUUID = "12345678-1234-4234-8234-000000000089"
+		j.PreparedRawUUID = binding.RawPartitionUUID
+	}
+	data, err := json.Marshal(j)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cm := resource("longhorn-v2-prepare-" + binding.Selection)
+	cm["metadata"].(object)["namespace"] = "longhorn-system"
+	cm["data"] = object{"record": string(data)}
+	return cm
+}
+
+// Foreign, unfinished, replaced or ambiguous preparation cannot authorize a
+// raw disk, and completion must prove the original filesystem was replaced.
+func TestTemporaryActivationRequiresTheCompletedPreparationIdentity(t *testing.T) {
+	n, lh, raw, eph, pool := temporaryRawBindingFixture()
+	binding, err := projectTemporaryV2Binding("temporary-v1-1", n, lh, raw, eph, pool, text(n, "status", "nodeInfo", "bootID"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	cm := temporaryPreparedFixture(t, binding)
+	if _, err = preparedTemporaryBinding(cm, binding); err != nil {
+		t.Fatal("completed matching preparation was refused:", err)
+	}
+	for name, change := range map[string]func(object){
+		"foreign scope":    func(cm object) { cm["metadata"].(object)["namespace"] = "other" },
+		"missing identity": func(cm object) { delete(cm["metadata"].(object), "uid") },
+		"pending deletion": func(cm object) { cm["metadata"].(object)["deletionTimestamp"] = "2026-11-03T09:00:00Z" },
+		"unfinished preparation": func(cm object) {
+			cm["data"].(object)["record"] = strings.Replace(text(cm, "data", "record"), `"phase":"prepared"`, `"phase":"reset-requested"`, 1)
+		},
+		"old smaller preparation": func(cm object) {
+			cm["data"].(object)["record"] = strings.Replace(text(cm, "data", "record"), `,"stagingLayout":true`, "", 1)
+		},
+		"replaced worker": func(cm object) {
+			cm["data"].(object)["record"] = strings.Replace(text(cm, "data", "record"), binding.NodeUID, "other-node", 1)
+		},
+		"unchanged boot": func(cm object) {
+			cm["data"].(object)["record"] = strings.Replace(text(cm, "data", "record"), "12345678-1234-4234-8234-000000000099", binding.BootID, 1)
+		},
+		"unchanged ephemeral partition": func(cm object) {
+			cm["data"].(object)["record"] = strings.Replace(text(cm, "data", "record"), "original-ephemeral", binding.EphemeralUUID, 1)
+		},
+		"ambiguous phase": func(cm object) {
+			cm["data"].(object)["record"] = strings.Replace(text(cm, "data", "record"), `"phase":"prepared"`, `"phase":"reset-requested","phase":"prepared"`, 1)
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			cm := temporaryPreparedFixture(t, binding)
+			change(cm)
+			if _, err := preparedTemporaryBinding(cm, binding); err == nil {
+				t.Fatal("incomplete or foreign preparation authorized raw storage")
+			}
+		})
+	}
+}
+
+func TestTemporaryActivationRequiresThePreservingPreparationIdentity(t *testing.T) {
+	n, lh, raw, eph, pool := temporaryRawBindingFixture()
+	binding, err := projectTemporaryV2Binding("temporary-v1-1", n, lh, raw, eph, pool, text(n, "status", "nodeInfo", "bootID"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	binding.Selection = "temporary-v2-1"
+	binding.Name = strings.Replace(binding.Name, "-v1-", "-v2-", 1)
+	cm := temporaryPreparedFixture(t, binding)
+	var record journal
+	if err = json.Unmarshal([]byte(text(cm, "data", "record")), &record); err != nil {
+		t.Fatal(err)
+	}
+	for _, tc := range []struct {
+		name   string
+		change func(*journal)
+	}{
+		{"complete", nil},
+		{"missing extent proof", func(r *journal) { r.RetainedExtentDigest = "" }},
+		{"unexpected raw absence", func(r *journal) { r.RawVolumeAbsent = true }},
+		{"wrong destination", func(r *journal) { r.PreparedRawUUID = "12345678-1234-4234-8234-000000000099" }},
+		{"retained partition reused", func(r *journal) { r.RetainedRawUUID = r.PreparedRawUUID }},
+		{"unfinished reset", func(r *journal) { r.Phase = "reset-requested" }},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			candidate := record
+			if tc.change != nil {
+				tc.change(&candidate)
+			}
+			data, marshalErr := json.Marshal(candidate)
+			if marshalErr != nil {
+				t.Fatal(marshalErr)
+			}
+			cm["data"] = object{"record": string(data)}
+			if _, err := preparedTemporaryBinding(cm, binding); (err == nil) != (tc.change == nil) {
+				t.Fatalf("unexpected preserving-preparation verdict: %v", err)
+			}
+		})
+	}
+}
diff --git a/scripts/migrate-longhorn-v2/temporary_activation_record.go b/scripts/migrate-longhorn-v2/temporary_activation_record.go
new file mode 100644
index 0000000000..d492617d0a
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/temporary_activation_record.go
@@ -0,0 +1,19 @@
+package main
+
+import "github.com/devantler-tech/platform/internal/longhornmigration"
+
+const temporaryActivationJournal = longhornmigration.ActivationJournal
+
+var temporaryV2Selections = longhornmigration.ActivationSelections()
+
+type temporaryActivationRecord = longhornmigration.ActivationRecord
+
+// decodeTemporaryActivationRecord shares the strict receipt parser with the
+// read-only publication gate, while preserving this operation's error boundary.
+func decodeTemporaryActivationRecord(data []byte) (temporaryActivationRecord, error) {
+	record, err := longhornmigration.DecodeActivationRecord(data)
+	if err != nil {
+		return temporaryActivationRecord{}, errTemporaryActivation
+	}
+	return record, nil
+}
diff --git a/scripts/migrate-longhorn-v2/temporary_activation_record_test.go b/scripts/migrate-longhorn-v2/temporary_activation_record_test.go
new file mode 100644
index 0000000000..a928e2856f
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/temporary_activation_record_test.go
@@ -0,0 +1,103 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+)
+
+// temporaryActivationRecordFixture gives each selected worker independent
+// provider, hardware, partition and preparation identities.
+func temporaryActivationRecordFixture(t *testing.T) temporaryActivationRecord {
+	t.Helper()
+	n, lh, raw, eph, pool := temporaryRawBindingFixture()
+	binding, err := projectTemporaryV2Binding("temporary-v1-1", n, lh, raw, eph, pool, text(n, "status", "nodeInfo", "bootID"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	record := temporaryActivationRecord{Version: 1, Source: strings.Repeat("a", 40), OwnerRun: "123", Targets: map[string]temporaryV2Binding{}}
+	for i, selection := range []string{"temporary-v1-1", "temporary-v1-2", "temporary-v2-1"} {
+		b := binding
+		b.Selection = selection
+		b.Name = fmt.Sprintf("autoscale-longhorn-%s-%016x", strings.Split(selection, "-")[1], i)
+		b.NodeUID, b.ProviderID, b.MachineUUID = fmt.Sprintf("node-%d", i), fmt.Sprintf("hcloud://%d", 100+i), fmt.Sprintf("hardware-%d", i)
+		b.LonghornUID, b.EphemeralUUID = fmt.Sprintf("longhorn-%d", i), fmt.Sprintf("ephemeral-%d", i)
+		b.RawPartitionUUID = fmt.Sprintf("12345678-1234-4234-8234-%012x", i+10)
+		b.PreparationUID = fmt.Sprintf("preparation-%d", i)
+		record.Targets[selection] = b
+	}
+	return record
+}
+
+// The decoder must reject missing or reused identities and ambiguous JSON
+// rather than let a partial or differently interpreted plan authorize storage.
+func TestTemporaryActivationRecordRequiresThreeDistinctCompleteBindings(t *testing.T) {
+	record := temporaryActivationRecordFixture(t)
+	data, _ := json.Marshal(record)
+	if _, err := decodeTemporaryActivationRecord(data); err != nil {
+		t.Fatal("complete fixed temporary activation record rejected:", err)
+	}
+	for name, change := range map[string]func(*temporaryActivationRecord){
+		"unreviewed source":  func(r *temporaryActivationRecord) { r.Source = "unknown" },
+		"missing owner":      func(r *temporaryActivationRecord) { r.OwnerRun = "" },
+		"missing third disk": func(r *temporaryActivationRecord) { delete(r.Targets, "temporary-v2-1") },
+		"missing preparation identity": func(r *temporaryActivationRecord) {
+			b := r.Targets["temporary-v1-1"]
+			b.PreparationUID = ""
+			r.Targets["temporary-v1-1"] = b
+		},
+		"missing preserving preparation": func(r *temporaryActivationRecord) {
+			b := r.Targets["temporary-v2-1"]
+			b.PreparationUID = ""
+			r.Targets["temporary-v2-1"] = b
+		},
+		"duplicated preparation identity": func(r *temporaryActivationRecord) {
+			b := r.Targets["temporary-v1-2"]
+			b.PreparationUID = r.Targets["temporary-v1-1"].PreparationUID
+			r.Targets["temporary-v1-2"] = b
+		},
+		"wrong target pool": func(r *temporaryActivationRecord) {
+			b := r.Targets["temporary-v2-1"]
+			b.Name = r.Targets["temporary-v1-1"].Name
+			r.Targets["temporary-v2-1"] = b
+		},
+		"duplicate provider": func(r *temporaryActivationRecord) {
+			b := r.Targets["temporary-v1-2"]
+			b.ProviderID = r.Targets["temporary-v1-1"].ProviderID
+			r.Targets["temporary-v1-2"] = b
+		},
+		"duplicate partition": func(r *temporaryActivationRecord) {
+			b := r.Targets["temporary-v1-2"]
+			b.RawPartitionUUID = r.Targets["temporary-v1-1"].RawPartitionUUID
+			r.Targets["temporary-v1-2"] = b
+		},
+		"missing native identity": func(r *temporaryActivationRecord) {
+			b := r.Targets["temporary-v1-2"]
+			b.RawPartitionUUID = ""
+			r.Targets["temporary-v1-2"] = b
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			r := temporaryActivationRecordFixture(t)
+			change(&r)
+			data, _ := json.Marshal(r)
+			if _, err := decodeTemporaryActivationRecord(data); err == nil {
+				t.Fatal("unproven activation record was accepted")
+			}
+		})
+	}
+	for name, altered := range map[string]string{
+		"duplicate field":   strings.Replace(string(data), `"completed":false`, `"completed":false,"completed":true`, 1),
+		"root case alias":   strings.Replace(string(data), `"source":`, `"Source":`, 1),
+		"nested case alias": strings.Replace(string(data), `"nodeUID":`, `"NodeUID":`, 1),
+		"unknown field":     strings.Replace(string(data), `"version":1`, `"version":1,"extra":true`, 1),
+		"trailing document": string(data) + ` {}`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			if _, err := decodeTemporaryActivationRecord([]byte(altered)); err == nil {
+				t.Fatal("ambiguous activation record was accepted")
+			}
+		})
+	}
+}
diff --git a/scripts/migrate-longhorn-v2/temporary_activation_run.go b/scripts/migrate-longhorn-v2/temporary_activation_run.go
new file mode 100644
index 0000000000..17a55351cb
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/temporary_activation_run.go
@@ -0,0 +1,288 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+)
+
+// read accepts only the fixed activation journal with a complete native API
+// identity and a strictly decoded three-member plan.
+func (a temporaryV2Activation) read(ctx context.Context) (object, temporaryActivationRecord, error) {
+	cm, err := a.c.get(ctx, "configmap", "longhorn-system", temporaryActivationJournal, true)
+	if err != nil || cm == nil {
+		return cm, temporaryActivationRecord{}, err
+	}
+	if text(cm, "metadata", "name") != temporaryActivationJournal || text(cm, "metadata", "namespace") != "longhorn-system" || text(cm, "metadata", "uid") == "" || text(cm, "metadata", "resourceVersion") == "" || value(cm, "metadata", "deletionTimestamp") != nil {
+		return nil, temporaryActivationRecord{}, errTemporaryActivation
+	}
+	record, err := decodeTemporaryActivationRecord([]byte(text(cm, "data", "record")))
+	return cm, record, err
+}
+
+// persist conditionally writes the plan on current main, then verifies the
+// same journal identity and complete record through a fresh native read.
+func (a temporaryV2Activation) persist(ctx context.Context, cm object, record temporaryActivationRecord) (object, error) {
+	if record.Validate() != nil || record.Source != a.source || record.OwnerRun != a.owner {
+		return nil, errTemporaryActivation
+	}
+	data, err := json.Marshal(record)
+	if err != nil || sourceCurrent(ctx, a.c.run, a.source) != nil {
+		return nil, errTemporaryActivation
+	}
+	if cm == nil {
+		cm, err = a.c.create(ctx, a.source, object{"apiVersion": "v1", "kind": "ConfigMap", "metadata": object{"name": temporaryActivationJournal, "namespace": "longhorn-system"}, "data": object{"record": string(data)}})
+	} else {
+		err = a.c.patch(ctx, "configmap", "longhorn-system", cm, object{"op": "replace", "path": "/data/record", "value": string(data)})
+	}
+	if err != nil {
+		return nil, err
+	}
+	readback, actual, err := a.read(ctx)
+	if err != nil || text(readback, "metadata", "uid") != text(cm, "metadata", "uid") || !reflect.DeepEqual(record, actual) {
+		return nil, errTemporaryActivation
+	}
+	return readback, nil
+}
+
+// owned rebinds journal identity, owner, plan and current main immediately
+// before every registration or completion write.
+func (a temporaryV2Activation) owned(ctx context.Context, uid string, expected temporaryActivationRecord) (object, error) {
+	cm, actual, err := a.read(ctx)
+	if err != nil || uid == "" || text(cm, "metadata", "uid") != uid || actual.Source != a.source || actual.OwnerRun != a.owner || !reflect.DeepEqual(expected, actual) {
+		return nil, errTemporaryActivation
+	}
+	return cm, sourceCurrent(ctx, a.c.run, a.source)
+}
+
+// environment preserves the original V1 data and database quorums throughout
+// raw registration. All six clients need their reserved capacity and every
+// temporary worker must retain its owned autoscaler pin.
+func (a temporaryV2Activation) environment(ctx context.Context) (inventory, error) {
+	if err := verifyBudgetSettings(ctx, a.c); err != nil {
+		return inventory{}, err
+	}
+	if err := verifyV2Settings(ctx, a.c); err != nil {
+		return inventory{}, err
+	}
+	s, err := a.c.snapshot(ctx)
+	if err != nil || verifyMigrationCapacity(s.nodes, a.pools) != nil || verifyRedundancy(s) != nil {
+		return inventory{}, errTemporaryActivation
+	}
+	if err = verifyStagingBudget(s); err != nil {
+		return inventory{}, err
+	}
+	managers, err := a.c.list(ctx, "instancemanagers.longhorn.io", "longhorn-system")
+	if err != nil || verifyV2Clients(s, managers, a.pools, v2ManagerImage) != nil {
+		return inventory{}, errTemporaryActivation
+	}
+	for _, node := range s.nodes {
+		if strings.HasPrefix(text(node, "metadata", "name"), "prod-control-plane-") {
+			continue
+		}
+		if value(node, "spec", "unschedulable") == true || text(node, "metadata", "annotations", fenceAnnotation) != "" || text(node, "status", "allocatable", "cpu") != "8" {
+			return inventory{}, errTemporaryActivation
+		}
+	}
+	for _, volume := range s.volumes {
+		if text(volume, "spec", "dataEngine") != "v1" {
+			return inventory{}, errTemporaryActivation
+		}
+	}
+	if _, err = (hostMemoryReboot{c: a.c}).databases(ctx, nil); err != nil {
+		return inventory{}, err
+	}
+	if err = requireBenchmarkBaseline(ctx, a.c, s.nodes); err != nil {
+		return inventory{}, err
+	}
+	return s, nil
+}
+
+// targets collects all native bindings before any disk is enabled. A resumed
+// plan must still name the same physical workers and partitions, and no existing
+// replica may be assigned to a temporary member during this pre-cutover step.
+func (a temporaryV2Activation) targets(ctx context.Context, s inventory, journalUID string, expected map[string]temporaryV2Binding) (map[string]temporaryV2Binding, error) {
+	nodes, err := indexed(s.nodes)
+	if err != nil {
+		return nil, err
+	}
+	storage, err := indexed(s.storageNodes)
+	if err != nil {
+		return nil, err
+	}
+	result := map[string]temporaryV2Binding{}
+	for _, selection := range temporaryV2Selections {
+		name, _, err := resolveTemporaryInventoryTarget(selection, s.nodes, a.pools)
+		if err != nil {
+			return nil, fmt.Errorf("temporary raw registration held [stage=pool-selection]: %w", err)
+		}
+		for _, replica := range s.replicas {
+			if text(replica, "spec", "nodeID") == name {
+				return nil, errTemporaryActivation
+			}
+		}
+		if _, err = temporaryRegistrationState(storage[name], journalUID, selection); err != nil {
+			return nil, err
+		}
+		role := text(nodes[name], "metadata", "labels", migrationPoolLabel)
+		binding, err := a.readBinding(ctx, selection, nodes[name], storage[name], a.pools[role])
+		if err != nil {
+			return nil, errTemporaryActivation
+		}
+		binding, err = a.prepared(ctx, binding)
+		if err != nil || expected != nil && !expected[selection].SameStorageIdentity(binding) {
+			return nil, errTemporaryActivation
+		}
+		result[selection] = binding
+	}
+	return result, nil
+}
+
+// registrationChanges adds ownership and the canonical raw disk in one
+// conditional Longhorn-node patch, preserving unrelated metadata.
+func registrationChanges(storage object, journalUID, selection string) []object {
+	changes := []object{}
+	if value(storage, "metadata", "labels") == nil {
+		changes = append(changes, object{"op": "add", "path": "/metadata/labels", "value": object{}})
+	}
+	if value(storage, "metadata", "annotations") == nil {
+		changes = append(changes, object{"op": "add", "path": "/metadata/annotations", "value": object{}})
+	}
+	changes = append(changes,
+		object{"op": "add", "path": "/metadata/labels/" + strings.ReplaceAll(migrationEnabledLabel, "/", "~1"), "value": "true"},
+		object{"op": "add", "path": "/metadata/annotations/" + strings.ReplaceAll(temporaryActivationOwner, "/", "~1"), "value": journalUID},
+		object{"op": "replace", "path": "/spec/disks", "value": blockDisksFor(temporaryRawID(selection))},
+		object{"op": "replace", "path": "/spec/allowScheduling", "value": true},
+		object{"op": "replace", "path": "/spec/evictionRequested", "value": false})
+	return changes
+}
+
+// run registers the three temporary raw disks under one durable identity plan.
+// An uncertain patch stops the invocation; only a later confirmed owner can
+// adopt its exact partial result after the previous workflow has terminated.
+func (a temporaryV2Activation) run(ctx context.Context) error {
+	if err := sourceCurrent(ctx, a.c.run, a.source); err != nil {
+		return err
+	}
+	s, err := a.environment(ctx)
+	if err != nil {
+		return err
+	}
+	cm, record, err := a.read(ctx)
+	if err != nil {
+		return err
+	}
+	if cm == nil {
+		targets, err := a.targets(ctx, s, "", nil)
+		if err != nil {
+			return err
+		}
+		record = temporaryActivationRecord{Version: 1, Source: a.source, OwnerRun: a.owner, Targets: targets}
+		cm, err = a.persist(ctx, nil, record)
+		if err != nil {
+			return err
+		}
+	} else {
+		if _, err = a.targets(ctx, s, text(cm, "metadata", "uid"), record.Targets); err != nil {
+			return err
+		}
+		if record.OwnerRun != a.owner || record.Source != a.source {
+			if err = previousRunCompleted(ctx, a.c, record.OwnerRun, record.Source); err != nil {
+				return err
+			}
+			record.OwnerRun, record.Source = a.owner, a.source
+			cm, err = a.persist(ctx, cm, record)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	uid := text(cm, "metadata", "uid")
+	for _, selection := range temporaryV2Selections {
+		s, err = a.environment(ctx)
+		if err != nil {
+			return err
+		}
+		if _, err = a.targets(ctx, s, uid, record.Targets); err != nil {
+			return err
+		}
+		storage, err := a.c.get(ctx, "nodes.longhorn.io", "longhorn-system", record.Targets[selection].Name, false)
+		if err != nil || text(storage, "metadata", "uid") != record.Targets[selection].LonghornUID {
+			return errTemporaryActivation
+		}
+		registered, err := temporaryRegistrationState(storage, uid, selection)
+		if err != nil {
+			return err
+		}
+		if registered {
+			continue
+		}
+		if _, err = a.owned(ctx, uid, record); err != nil {
+			return err
+		}
+		if err = a.c.patch(ctx, "nodes.longhorn.io", "longhorn-system", storage, registrationChanges(storage, uid, selection)...); err != nil {
+			return err
+		}
+	}
+	if err = waitReady(ctx, func(ctx context.Context) error {
+		if _, err := a.owned(ctx, uid, record); err != nil {
+			return errors.Join(errReadinessTerminal, err)
+		}
+		s, err := a.environment(ctx)
+		if err != nil {
+			return err
+		}
+		if _, err = a.targets(ctx, s, uid, record.Targets); err != nil {
+			return errors.Join(errReadinessTerminal, err)
+		}
+		_, err = v2Environment(ctx, a.c, a.pools)
+		return err
+	}); err != nil {
+		return err
+	}
+	if !record.Completed {
+		cm, err = a.owned(ctx, uid, record)
+		if err != nil {
+			return err
+		}
+		record.Completed = true
+		if _, err = a.persist(ctx, cm, record); err != nil {
+			return err
+		}
+	}
+	fmt.Println("Temporary V2 storage is registered on three identity-bound raw disks; clients and original V1 copies are healthy.")
+	return nil
+}
+
+// accepted is a read-only registration acceptance check. The completed journal must
+// come from a terminated protected workflow and still match all three native
+// partitions and journal-owned registrations; its cached completion bit alone
+// never authorizes IO.
+func (a temporaryV2Activation) accepted(ctx context.Context) error {
+	cm, record, err := a.read(ctx)
+	if err != nil || cm == nil || !record.Completed || previousRunCompleted(ctx, a.c, record.OwnerRun, record.Source) != nil {
+		return errTemporaryActivation
+	}
+	s, err := a.environment(ctx)
+	if err != nil {
+		return err
+	}
+	uid := text(cm, "metadata", "uid")
+	if _, err = a.targets(ctx, s, uid, record.Targets); err != nil {
+		return err
+	}
+	storage, err := indexed(s.storageNodes)
+	if err != nil {
+		return err
+	}
+	for _, binding := range record.Targets {
+		registered, err := temporaryRegistrationState(storage[binding.Name], uid, binding.Selection)
+		if err != nil || !registered {
+			return errTemporaryActivation
+		}
+	}
+	return nil
+}
diff --git a/scripts/migrate-longhorn-v2/temporary_activation_test.go b/scripts/migrate-longhorn-v2/temporary_activation_test.go
new file mode 100644
index 0000000000..d7fef9de52
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/temporary_activation_test.go
@@ -0,0 +1,104 @@
+package main
+
+import (
+	"testing"
+)
+
+// temporaryRawBindingFixture supplies one pinned worker and separate encrypted
+// filesystem and raw partitions without using the registration builder.
+func temporaryRawBindingFixture() (object, object, object, object, migrationPool) {
+	n := capacityFixture()[3]
+	n["spec"] = object{"providerID": "hcloud://123", "unschedulable": false}
+	n["status"].(object)["nodeInfo"] = object{"systemUUID": "hardware", "bootID": "12345678-1234-4234-8234-000000000001", "osImage": "Talos fixture", "architecture": "amd64"}
+	n["status"].(object)["allocatable"].(object)["cpu"] = "8"
+	lh := resource(text(n, "metadata", "name"))
+	lh["metadata"].(object)["namespace"] = "longhorn-system"
+	raw, eph := rawFixture()
+	valueObject(raw, "spec")["size"] = float64(144 << 30)
+	raw["spec"].(object)["partitionUUID"] = "12345678-1234-4234-8234-000000000002"
+	return n, lh, raw, eph, migrationPool{name: "autoscale-longhorn-v1", role: "v1", serverType: "cpx42", count: 2}
+}
+
+// Wrong ownership, boot, encryption, capacity or pool selection must prevent a
+// native partition observation from becoming registration evidence.
+func TestTemporaryRawRegistrationBindsTheNativePartitionsAndPoolMember(t *testing.T) {
+	n, lh, raw, eph, pool := temporaryRawBindingFixture()
+	binding, err := projectTemporaryV2Binding("temporary-v1-1", n, lh, raw, eph, pool, text(n, "status", "nodeInfo", "bootID"))
+	if err != nil || binding.Name != text(n, "metadata", "name") || binding.RawPartitionUUID != text(raw, "spec", "partitionUUID") || binding.EphemeralUUID != text(eph, "spec", "uuid") {
+		t.Fatal("qualified temporary raw storage was not bound to its native identities:", err)
+	}
+	for name, change := range map[string]func(object, object, object, object){
+		"foreign storage identity":           func(_ object, lh object, _, _ object) { lh["metadata"].(object)["name"] = "other" },
+		"missing raw identity":               func(_, _ object, raw, _ object) { delete(raw["spec"].(object), "partitionUUID") },
+		"raw and filesystem identity reused": func(_, _ object, raw, eph object) { raw["spec"].(object)["partitionUUID"] = text(eph, "spec", "uuid") },
+		"unencrypted raw disk":               func(_, _ object, raw, _ object) { raw["spec"].(object)["encryptionProvider"] = "" },
+		"original smaller raw capacity":      func(_, _ object, raw, _ object) { raw["spec"].(object)["size"] = float64(96 << 30) },
+		"cordoned member":                    func(n, _, _, _ object) { n["spec"].(object)["unschedulable"] = true },
+		"unowned capacity": func(n, _, _, _ object) {
+			delete(n["metadata"].(object)["annotations"].(object), capacityOwnerAnnotation)
+		},
+		"automatic disk creation enabled": func(n, _, _, _ object) {
+			n["metadata"].(object)["labels"].(object)["node.longhorn.io/create-default-disk"] = "config"
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			n, lh, raw, eph, pool := temporaryRawBindingFixture()
+			change(n, lh, raw, eph)
+			if _, err := projectTemporaryV2Binding("temporary-v1-1", n, lh, raw, eph, pool, text(n, "status", "nodeInfo", "bootID")); err == nil {
+				t.Fatal("unproven temporary registration binding was accepted")
+			}
+		})
+	}
+	if _, err := projectTemporaryV2Binding("temporary-v1-1", n, lh, raw, eph, pool, "12345678-1234-4234-8234-000000000099"); err == nil {
+		t.Fatal("a native boot mismatch was accepted")
+	}
+	if _, err := projectTemporaryV2Binding("temporary-v2-1", n, lh, raw, eph, pool, text(n, "status", "nodeInfo", "bootID")); err == nil {
+		t.Fatal("a selection from another pool was accepted")
+	}
+}
+
+// Resumption recognizes only an empty fence or this journal's exact raw disk;
+// matching paths without ownership never authorize adoption.
+func TestTemporaryRawRegistrationAcceptsOnlyItsOwnPartialActivation(t *testing.T) {
+	storage := func(enabled bool) object {
+		_, lh, _, _, _ := temporaryRawBindingFixture()
+		lh["metadata"].(object)["annotations"] = object{}
+		lh["metadata"].(object)["labels"] = object{}
+		lh["spec"] = object{"allowScheduling": false, "evictionRequested": true, "disks": object{}}
+		lh["status"] = object{"diskStatus": object{}}
+		if enabled {
+			lh["metadata"].(object)["annotations"].(object)[temporaryActivationOwner] = "journal-uid"
+			lh["metadata"].(object)["labels"].(object)[migrationEnabledLabel] = "true"
+			lh["spec"] = object{"allowScheduling": true, "evictionRequested": false, "disks": blockDisks()}
+		}
+		return lh
+	}
+	if registered, err := temporaryRegistrationState(storage(false), "journal-uid", "temporary-v1-1"); err != nil || registered {
+		t.Fatal("empty fenced storage could not be registered:", err)
+	}
+	if registered, err := temporaryRegistrationState(storage(true), "journal-uid", "temporary-v1-1"); err != nil || !registered {
+		t.Fatal("the journal's exact existing activation could not resume:", err)
+	}
+	if _, err := temporaryRegistrationState(storage(true), "journal-uid", "temporary-v2-1"); err == nil {
+		t.Fatal("the retained mapper qualified as staged storage")
+	}
+	for name, change := range map[string]func(object){
+		"foreign journal":   func(lh object) { lh["metadata"].(object)["annotations"].(object)[temporaryActivationOwner] = "foreign" },
+		"missing ownership": func(lh object) { delete(lh["metadata"].(object)["annotations"].(object), temporaryActivationOwner) },
+		"filesystem substitution": func(lh object) {
+			lh["spec"].(object)["disks"] = object{"filesystem": object{"path": "/var/lib/longhorn"}}
+		},
+		"unknown eviction state": func(lh object) { delete(lh["spec"].(object), "evictionRequested") },
+	} {
+		t.Run(name, func(t *testing.T) {
+			lh := storage(true)
+			change(lh)
+			if _, err := temporaryRegistrationState(lh, "journal-uid", "temporary-v1-1"); err == nil {
+				t.Fatal("a foreign or changed disk registration was adopted")
+			}
+		})
+	}
+	if _, err := temporaryRegistrationState(storage(true), "", "temporary-v1-1"); err == nil {
+		t.Fatal("an unjournaled existing registration was adopted")
+	}
+}
diff --git a/scripts/migrate-longhorn-v2/v2_environment.go b/scripts/migrate-longhorn-v2/v2_environment.go
new file mode 100644
index 0000000000..ba4c3ae47c
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/v2_environment.go
@@ -0,0 +1,58 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"reflect"
+	"strings"
+)
+
+// v2ManagerImage is qualified against the reviewed Longhorn 1.13.0 chart;
+// matching unreviewed live declarations cannot establish image provenance.
+const v2ManagerImage = "longhornio/longhorn-instance-manager:v1.13.0"
+
+// verifyV2Settings binds coexistence and effective SPDK configuration to the
+// reviewed profile before either registration or subsequent storage acceptance.
+func verifyV2Settings(ctx context.Context, c client) error {
+	for name, expected := range map[string]string{
+		"v1-data-engine": "true", "v2-data-engine": "true", "guaranteed-instance-manager-cpu": `{"v1":"6","v2":"25"}`,
+		"data-engine-cpu-mask": `{"v2":"0x3"}`, "data-engine-number-of-cpu-cores": `{"v2":"0"}`, "data-engine-cpu-isolation-enabled": `{"v2":"true"}`,
+		"data-engine-interrupt-mode-enabled": `{"v2":"false"}`, "data-engine-hugepage-enabled": `{"v2":"true"}`, "data-engine-memory-size": `{"v2":"2048"}`,
+	} {
+		o, err := c.get(ctx, "settings.longhorn.io", "longhorn-system", name, false)
+		if err != nil {
+			return err
+		}
+		actual := text(o, "value")
+		if strings.HasPrefix(expected, "{") {
+			var want, got map[string]string
+			if json.Unmarshal([]byte(expected), &want) != nil || json.Unmarshal([]byte(actual), &got) != nil || !reflect.DeepEqual(want, got) {
+				return errors.New("live V2 settings have not adopted the reviewed configuration")
+			}
+		} else if actual != expected {
+			return errors.New("both engines must remain enabled during storage qualification")
+		}
+	}
+	return nil
+}
+
+// v2Environment proves effective worker clients and three independent ready
+// raw disks. Merely enabling the V2 setting cannot establish storage readiness.
+func v2Environment(ctx context.Context, c client, pools map[string]migrationPool) (inventory, error) {
+	if err := verifyV2Settings(ctx, c); err != nil {
+		return inventory{}, err
+	}
+	s, err := c.snapshot(ctx)
+	if err != nil {
+		return inventory{}, err
+	}
+	managers, err := c.list(ctx, "instancemanagers.longhorn.io", "longhorn-system")
+	if err != nil {
+		return inventory{}, err
+	}
+	if err = verifyV2Runtime(s, managers, pools, v2ManagerImage); err != nil {
+		return inventory{}, err
+	}
+	return s, nil
+}
diff --git a/scripts/migrate-longhorn-v2/v2_readiness.go b/scripts/migrate-longhorn-v2/v2_readiness.go
new file mode 100644
index 0000000000..ca5ed7f3e1
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/v2_readiness.go
@@ -0,0 +1,200 @@
+package main
+
+import (
+	"errors"
+	"strings"
+)
+
+// verifyV2Runtime requires ready SPDK clients on every worker and three
+// independent raw disks on the temporary workers. The original workers retain
+// the V1 replicas throughout the disposable recovery proof.
+func verifyV2Runtime(s inventory, managers []object, pools map[string]migrationPool, expectedManagerImage string) error {
+	if expectedManagerImage == "" || strings.TrimSpace(expectedManagerImage) != expectedManagerImage {
+		return errors.New("the reviewed instance-manager image is unresolved")
+	}
+	if err := verifyMigrationCapacity(s.nodes, pools); err != nil {
+		return err
+	}
+	if err := verifyRedundancy(s); err != nil {
+		return err
+	}
+	nodes, err := indexed(s.nodes)
+	if err != nil {
+		return err
+	}
+	storage, err := indexed(s.storageNodes)
+	if err != nil {
+		return err
+	}
+	if _, err = indexed(managers); err != nil {
+		return err
+	}
+	workers, raw := map[string]bool{}, map[string]string{}
+	for name, n := range nodes {
+		if strings.HasPrefix(name, "prod-control-plane-") {
+			continue
+		}
+		if value(n, "spec", "unschedulable") == true || text(n, "status", "allocatable", "cpu") != "8" {
+			return errors.New("V2 migration workers must be schedulable with eight allocatable CPUs")
+		}
+		workers[name] = true
+		role := text(n, "metadata", "labels", migrationPoolLabel)
+		if _, temporary := pools[role]; temporary {
+			raw[name] = temporaryRawID("temporary-" + role + "-1")
+		}
+	}
+	if len(workers) != 6 || len(raw) != 3 {
+		return errors.New("V2 canary storage and client roster is incomplete")
+	}
+	diskIDs := map[string]bool{}
+	for name, rawID := range raw {
+		lh := storage[name]
+		status, _ := value(lh, "status").(object)
+		diskStatus, _ := value(lh, "status", "diskStatus").(object)
+		disk, _ := diskStatus["longhorn-v2"].(object)
+		id := text(disk, "diskUUID")
+		if !reviewedBlockDisksFor(value(lh, "spec", "disks"), rawID) || value(lh, "spec", "allowScheduling") != true || value(lh, "spec", "evictionRequested") != false ||
+			!ready(status) || len(diskStatus) != 1 || !ready(disk) || id == "" || diskIDs[id] {
+			return errors.New("three distinct ready canonical raw disks are not proven")
+		}
+		if override := value(lh, "spec", "instanceManagerCPURequest"); override != nil && override != float64(0) {
+			return errors.New("a node CPU override supersedes the reviewed V2 reservation")
+		}
+		diskIDs[id] = true
+	}
+	return verifyV2ClientPods(s, managers, workers, expectedManagerImage)
+}
+
+// verifyV2Clients qualifies the effective clients before any raw disk is
+// registered, independently of the later three-disk readiness requirement.
+func verifyV2Clients(s inventory, managers []object, pools map[string]migrationPool, expectedManagerImage string) error {
+	if err := verifyMigrationCapacity(s.nodes, pools); err != nil {
+		return err
+	}
+	nodes, err := indexed(s.nodes)
+	if err != nil {
+		return err
+	}
+	workers := map[string]bool{}
+	for name, node := range nodes {
+		if strings.HasPrefix(name, "prod-control-plane-") {
+			continue
+		}
+		if value(node, "spec", "unschedulable") == true || text(node, "status", "allocatable", "cpu") != "8" {
+			return errors.New("V2 migration workers must be schedulable with eight allocatable CPUs")
+		}
+		workers[name] = true
+	}
+	if len(workers) != 6 {
+		return errors.New("V2 client roster is incomplete")
+	}
+	return verifyV2ClientPods(s, managers, workers, expectedManagerImage)
+}
+
+// verifyV2ClientPods binds each client to its controller and reviewed image,
+// reservations, reactor arguments and isolation on the already checked roster.
+func verifyV2ClientPods(s inventory, managers []object, workers map[string]bool, expectedManagerImage string) error {
+	if expectedManagerImage == "" || strings.TrimSpace(expectedManagerImage) != expectedManagerImage {
+		return errors.New("the reviewed instance-manager image is unresolved")
+	}
+	if _, err := indexed(managers); err != nil {
+		return err
+	}
+	pods := map[string]object{}
+	for _, p := range s.pods {
+		if text(p, "metadata", "namespace") != "longhorn-system" {
+			continue
+		}
+		name := text(p, "metadata", "name")
+		if pods[name] != nil {
+			return errors.New("instance-manager pod inventory is duplicated")
+		}
+		pods[name] = p
+	}
+	seen := map[string]bool{}
+	for _, im := range managers {
+		if text(im, "spec", "dataEngine") == "v1" {
+			continue
+		}
+		if text(im, "spec", "dataEngine") != "v2" {
+			return errors.New("an instance manager has an unrecognized data engine")
+		}
+		node := text(im, "spec", "nodeID")
+		if !workers[node] || seen[node] || text(im, "spec", "type") != "aio" || value(im, "metadata", "deletionTimestamp") != nil ||
+			text(im, "status", "currentState") != "running" || text(im, "metadata", "namespace") != "longhorn-system" {
+			return errors.New("V2 instance managers do not match the ready worker roster")
+		}
+		p := pods[text(im, "metadata", "name")]
+		status, _ := value(p, "status").(object)
+		if text(p, "metadata", "uid") == "" || value(p, "metadata", "deletionTimestamp") != nil || text(p, "spec", "nodeName") != node || text(p, "status", "phase") != "Running" || !ready(status) {
+			return errors.New("V2 instance-manager pod is missing, misplaced or unready")
+		}
+		owned := false
+		refs, _ := value(p, "metadata", "ownerReferences").([]any)
+		for _, ref := range refs {
+			r, _ := ref.(object)
+			if text(r, "apiVersion") == "longhorn.io/v1beta2" && text(r, "kind") == "InstanceManager" && text(r, "name") == text(im, "metadata", "name") && text(r, "uid") == text(im, "metadata", "uid") && value(r, "controller") == true {
+				owned = true
+			}
+		}
+		containers, _ := value(p, "spec", "containers").([]any)
+		if !owned || len(containers) != 1 {
+			return errors.New("V2 instance-manager workload ownership is unproven")
+		}
+		container, _ := containers[0].(object)
+		if command := value(container, "command"); command != nil {
+			items, ok := command.([]any)
+			if !ok || len(items) != 0 {
+				return errors.New("V2 workload overrides the reviewed image entrypoint")
+			}
+		}
+		cpu := text(container, "resources", "requests", "cpu")
+		cpuLimit := text(container, "resources", "limits", "cpu")
+		hugepages := text(container, "resources", "limits", "hugepages-2Mi")
+		if text(container, "name") != "instance-manager" || text(container, "image") != expectedManagerImage || text(container, "image") != text(im, "spec", "image") || (cpu != "2" && cpu != "2000m") || (cpuLimit != "4" && cpuLimit != "4000m") || (hugepages != "2Gi" && hugepages != "2048Mi") {
+			return errors.New("V2 workload lacks its effective two-core and two-GiB huge-page reservation")
+		}
+		args, _ := value(container, "args").([]any)
+		if len(args) == 0 || args[0] != "start-spdk-tgt" {
+			return errors.New("V2 workload does not launch the reviewed SPDK target wrapper")
+		}
+		flags := map[string]int{}
+		positions := map[string]int{}
+		for i, arg := range args {
+			a, ok := arg.(string)
+			if !ok {
+				return errors.New("V2 workload arguments are malformed")
+			}
+			for _, flag := range []string{"--spdk-cpumask", "--spdk-core-number", "--spdk-memory-size", "--enable-spdk", "--spdk-enabled", "--enable-irq-affinity", "--enable-workqueue-affinity", "--enable-rps", "--spdk-no-hugepage", "--spdk-interrupt-mode"} {
+				if strings.HasPrefix(a, flag+"=") {
+					return errors.New("V2 workload overrides a reviewed flag using an assignment")
+				}
+			}
+			flags[a]++
+			positions[a] = i
+		}
+		paired := func(flag, expected string) bool {
+			count := 0
+			for i, arg := range args {
+				if arg == flag {
+					if i+1 >= len(args) || args[i+1] != expected {
+						return false
+					}
+					count++
+				}
+			}
+			return count == 1
+		}
+		if !paired("--spdk-cpumask", "0x3") || !paired("--spdk-core-number", "0") || !paired("--spdk-memory-size", "2048") ||
+			flags["start-spdk-tgt"] != 1 || flags["daemon"] != 1 || flags["--enable-spdk"] != 1 || flags["--spdk-enabled"] != 1 ||
+			positions["--enable-spdk"] >= positions["daemon"] || positions["--spdk-enabled"] <= positions["daemon"] ||
+			flags["--enable-irq-affinity"] != 1 || flags["--enable-workqueue-affinity"] != 1 || flags["--enable-rps"] != 1 || flags["--spdk-no-hugepage"] != 0 || flags["--spdk-interrupt-mode"] != 0 {
+			return errors.New("V2 workload has not adopted the reviewed polling reactors, huge pages and CPU isolation")
+		}
+		seen[node] = true
+	}
+	if len(seen) != len(workers) {
+		return errors.New("V2 volume clients are not ready on every worker")
+	}
+	return nil
+}
diff --git a/scripts/migrate-longhorn-v2/v2_readiness_test.go b/scripts/migrate-longhorn-v2/v2_readiness_test.go
new file mode 100644
index 0000000000..8299e617e7
--- /dev/null
+++ b/scripts/migrate-longhorn-v2/v2_readiness_test.go
@@ -0,0 +1,166 @@
+package main
+
+import (
+	"fmt"
+	"testing"
+)
+
+// v2ReadinessFixture keeps original V1 replicas on permanent workers while
+// independent temporary disks supply the prospective V2 storage.
+func v2ReadinessFixture() (inventory, []object) {
+	return v2ReadinessFixtureForRoute(false)
+}
+
+// v2ReadinessFixtureForRoute can represent the superseded original-disk route
+// solely to prove that it cannot satisfy temporary-only activation admission.
+func v2ReadinessFixtureForRoute(legacy bool) (inventory, []object) {
+	s := evacuatedFixture()
+	s.nodes = capacityFixture()
+	old := []string{"storage-1", "storage-2", "storage-3"}
+	v1Nodes := []string{"prod-worker-1", "prod-worker-2", "prod-worker-3"}
+	if legacy {
+		v1Nodes = []string{"prod-worker-3", text(s.nodes[3], "metadata", "name"), text(s.nodes[4], "metadata", "name")}
+	}
+	for i, name := range v1Nodes {
+		s.storageNodes[i]["metadata"].(object)["name"] = name
+		for _, r := range s.replicas {
+			if text(r, "spec", "nodeID") == old[i] {
+				r["spec"].(object)["nodeID"] = name
+			}
+		}
+	}
+	for _, n := range s.nodes[:6] {
+		n["status"].(object)["allocatable"].(object)["cpu"] = "8"
+		role := text(n, "metadata", "labels", migrationPoolLabel)
+		if legacy && (text(n, "metadata", "name") == "prod-worker-3" || role == "v1") || !legacy && role == "" {
+			continue
+		}
+		lh := resource(text(n, "metadata", "name"))
+		lh["spec"] = object{"disks": blockDisksFor(temporaryRawID("temporary-" + role + "-1")), "allowScheduling": true, "evictionRequested": false, "instanceManagerCPURequest": float64(0)}
+		lh["status"] = object{"conditions": readyConditions(), "diskStatus": object{"longhorn-v2": object{"diskUUID": "raw-" + text(n, "metadata", "name"), "conditions": readyConditions()}}}
+		s.storageNodes = append(s.storageNodes, lh)
+	}
+	managers := []object{}
+	for i, n := range s.nodes[:6] {
+		im := resource(fmt.Sprintf("im-%d", i))
+		im["metadata"].(object)["namespace"] = "longhorn-system"
+		im["spec"] = object{"nodeID": text(n, "metadata", "name"), "dataEngine": "v2", "type": "aio", "image": "longhornio/longhorn-instance-manager:v1.13.0"}
+		im["status"] = object{"currentState": "running"}
+		managers = append(managers, im)
+		p := resource(text(im, "metadata", "name"))
+		p["metadata"].(object)["namespace"] = "longhorn-system"
+		p["metadata"].(object)["ownerReferences"] = []any{object{"apiVersion": "longhorn.io/v1beta2", "kind": "InstanceManager", "name": text(im, "metadata", "name"), "uid": text(im, "metadata", "uid"), "controller": true}}
+		p["spec"] = object{"nodeName": text(n, "metadata", "name"), "containers": []any{object{"name": "instance-manager", "image": text(im, "spec", "image"),
+			"args":      []any{"start-spdk-tgt", "--spdk-cpumask", "0x3", "--spdk-core-number", "0", "--spdk-memory-size", "2048", "--enable-spdk", "daemon", "--spdk-enabled", "--enable-irq-affinity", "--enable-workqueue-affinity", "--enable-rps"},
+			"resources": object{"requests": object{"cpu": "2"}, "limits": object{"cpu": "4", "hugepages-2Mi": "2Gi"}}}}}
+		p["status"] = object{"phase": "Running", "conditions": readyConditions()}
+		s.pods = append(s.pods, p)
+	}
+	return s, managers
+}
+
+// Raw storage on original workers cannot borrow the temporary-route authority.
+func TestV2RuntimeRejectsTheEarlierRebuiltPermanentWorkerRoute(t *testing.T) {
+	pools, err := configuredMigrationPools(poolConfigFixture())
+	if err != nil {
+		t.Fatal(err)
+	}
+	s, managers := v2ReadinessFixtureForRoute(true)
+	if verifyV2Runtime(s, managers, pools, "longhornio/longhorn-instance-manager:v1.13.0") == nil {
+		t.Fatal("the temporary-only route accepted raw storage on original workers")
+	}
+}
+
+// Ready disk identities, original redundancy and effective SPDK ownership,
+// executable arguments and reservations are all required for storage readiness.
+func TestV2RuntimeRequiresRealRawDisksAndReservedReactors(t *testing.T) {
+	pools, err := configuredMigrationPools(poolConfigFixture())
+	if err != nil {
+		t.Fatal(err)
+	}
+	s, managers := v2ReadinessFixture()
+	if err = verifyV2Runtime(s, managers, pools, "longhornio/longhorn-instance-manager:v1.13.0"); err != nil {
+		t.Fatal(err)
+	}
+	for name, change := range map[string]func(*inventory, *[]object){
+		"unrebuilt disk":     func(s *inventory, _ *[]object) { s.storageNodes[3]["spec"].(object)["disks"] = object{} },
+		"raw disk not ready": func(s *inventory, _ *[]object) { s.storageNodes[3]["status"].(object)["diskStatus"] = object{} },
+		"raw disk identity reused": func(s *inventory, _ *[]object) {
+			s.storageNodes[4]["status"].(object)["diskStatus"].(object)["longhorn-v2"].(object)["diskUUID"] = text(s.storageNodes[3], "status", "diskStatus", "longhorn-v2", "diskUUID")
+		},
+		"retained copy lost":         func(s *inventory, _ *[]object) { s.replicas[3]["spec"].(object)["healthyAt"] = "" },
+		"client manager missing":     func(_ *inventory, m *[]object) { *m = (*m)[1:] },
+		"client manager not running": func(_ *inventory, m *[]object) { (*m)[2]["status"].(object)["currentState"] = "starting" },
+		"manager pod unready":        func(s *inventory, _ *[]object) { s.pods[0]["status"].(object)["conditions"] = []any{} },
+		"manager type replaced":      func(_ *inventory, m *[]object) { (*m)[0]["spec"].(object)["type"] = "engine" },
+		"manager deletion pending": func(_ *inventory, m *[]object) {
+			(*m)[0]["metadata"].(object)["deletionTimestamp"] = "2026-11-03T09:00:00Z"
+		},
+		"manager identity missing": func(_ *inventory, m *[]object) { delete((*m)[0]["metadata"].(object), "uid") },
+		"foreign manager pod":      func(s *inventory, _ *[]object) { s.pods[0]["metadata"].(object)["ownerReferences"] = []any{} },
+		"entrypoint overridden": func(s *inventory, _ *[]object) {
+			value(s.pods[0], "spec", "containers").([]any)[0].(object)["command"] = []any{"/bin/true"}
+		},
+		"SPDK bootstrap absent": func(s *inventory, _ *[]object) {
+			value(s.pods[0], "spec", "containers").([]any)[0].(object)["args"].([]any)[0] = "daemon"
+		},
+		"SPDK launch disabled": func(s *inventory, _ *[]object) {
+			value(s.pods[0], "spec", "containers").([]any)[0].(object)["args"].([]any)[7] = "--debug"
+		},
+		"V2 daemon disabled": func(s *inventory, _ *[]object) {
+			value(s.pods[0], "spec", "containers").([]any)[0].(object)["args"].([]any)[9] = "--debug"
+		},
+		"insufficient cpu": func(s *inventory, _ *[]object) {
+			value(s.pods[0], "spec", "containers").([]any)[0].(object)["resources"].(object)["requests"] = object{"cpu": "480m"}
+		},
+		"reactors consume whole cpu limit": func(s *inventory, _ *[]object) {
+			value(s.pods[0], "spec", "containers").([]any)[0].(object)["resources"].(object)["limits"].(object)["cpu"] = "2"
+		},
+		"hugepages not reserved": func(s *inventory, _ *[]object) {
+			value(s.pods[0], "spec", "containers").([]any)[0].(object)["resources"].(object)["limits"] = object{"hugepages-2Mi": "0"}
+		},
+		"one busy reactor": func(s *inventory, _ *[]object) {
+			value(s.pods[0], "spec", "containers").([]any)[0].(object)["args"].([]any)[2] = "0x1"
+		},
+		"cpu isolation missing": func(s *inventory, _ *[]object) {
+			c := value(s.pods[0], "spec", "containers").([]any)[0].(object)
+			c["args"] = c["args"].([]any)[:7]
+		},
+		"node pressured": func(s *inventory, _ *[]object) { s.nodes[0]["status"].(object)["conditions"] = readyConditions() },
+		"raw cpu override": func(s *inventory, _ *[]object) {
+			s.storageNodes[3]["spec"].(object)["instanceManagerCPURequest"] = float64(480)
+		},
+		"raw eviction status missing": func(s *inventory, _ *[]object) {
+			delete(s.storageNodes[3]["spec"].(object), "evictionRequested")
+		},
+		"matching but unreviewed image": func(s *inventory, m *[]object) {
+			(*m)[0]["spec"].(object)["image"] = "example.invalid/foreign:latest"
+			value(s.pods[0], "spec", "containers").([]any)[0].(object)["image"] = "example.invalid/foreign:latest"
+		},
+		"interrupt mode assignment": func(s *inventory, _ *[]object) {
+			c := value(s.pods[0], "spec", "containers").([]any)[0].(object)
+			c["args"] = append(c["args"].([]any), "--spdk-interrupt-mode=true")
+		},
+		"hugepage bypass assignment": func(s *inventory, _ *[]object) {
+			c := value(s.pods[0], "spec", "containers").([]any)[0].(object)
+			c["args"] = append(c["args"].([]any), "--spdk-no-hugepage=true")
+		},
+		"isolation disabled by later assignment": func(s *inventory, _ *[]object) {
+			c := value(s.pods[0], "spec", "containers").([]any)[0].(object)
+			c["args"] = append(c["args"].([]any), "--enable-irq-affinity=false")
+		},
+		"unknown extra manager engine": func(_ *inventory, m *[]object) {
+			im := resource("unknown-engine")
+			im["spec"] = object{"dataEngine": "v3"}
+			*m = append(*m, im)
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			s, m := v2ReadinessFixture()
+			change(&s, &m)
+			if verifyV2Runtime(s, m, pools, "longhornio/longhorn-instance-manager:v1.13.0") == nil {
+				t.Fatal("V2 canary admitted without its real runtime prerequisites")
+			}
+		})
+	}
+}
diff --git a/scripts/validate-eks-ci-role-policy/approved-surface.txt b/scripts/validate-eks-ci-role-policy/approved-surface.txt
index 5f28df44b1..519618f280 100644
--- a/scripts/validate-eks-ci-role-policy/approved-surface.txt
+++ b/scripts/validate-eks-ci-role-policy/approved-surface.txt
@@ -96,7 +96,7 @@ helm.toolkit.fluxcd.io/v2|HelmRelease|kubescape|kubescape cd313c675760792a4acd04
 
 helm.toolkit.fluxcd.io/v2|HelmRelease|kyverno|kyverno cba1d622347666781a22316d3c466cb9df54ea4951809e8efa9917b048032867
 
-helm.toolkit.fluxcd.io/v2|HelmRelease|longhorn-system|longhorn 3687bc9ddaa8177bc383eb7d25fc8f1bff732a935e2874656fdd8b438aea6e03
+helm.toolkit.fluxcd.io/v2|HelmRelease|longhorn-system|longhorn 10154822248452b0a2b5419cd8d898204fc4f75554b2805407a6cb9bef15ca75
 
 helm.toolkit.fluxcd.io/v2|HelmRelease|oauth2-proxy|oauth2-proxy d16150bb9a7d264f002fd3cb8d82d459a368d6428207e23fd0b8479e4194f60d