diff --git a/cmd/wfctl/infra.go b/cmd/wfctl/infra.go
index ea085ab8..5a3e9de0 100644
--- a/cmd/wfctl/infra.go
+++ b/cmd/wfctl/infra.go
@@ -381,6 +381,7 @@ func runInfraPlan(args []string) error {
 		// Embed a hash of the desired-state inputs so wfctl infra apply --plan
 		// can detect stale plans when the config changes after plan generation.
 		plan.DesiredHash = desiredStateHash(desired)
+		plan.Include = sortedIncludeNames(planIncludeSet)
 		// Stamp generator metadata (wfctl version + IaC plugin versions) so
 		// operators can inspect what toolchain version produced this plan.
 		meta := buildGeneratorMetadata()
@@ -1433,6 +1434,7 @@ func runInfraApply(args []string) error {
 		if err != nil {
 			return fmt.Errorf("parse infra resource specs: %w", err)
 		}
+		planIncludeSet := includeSetFromNames(plan.Include)
 		if plan.DesiredHash == "" {
 			return fmt.Errorf("plan file has no hash — regenerate with: wfctl infra plan -o plan.json")
 		}
@@ -1464,6 +1466,11 @@ func runInfraApply(args []string) error {
 			if stateErr != nil {
 				return fmt.Errorf("load state for stale-check: %w", stateErr)
 			}
+			if err := validateIncludeSet(planIncludeSet, desired, currentState); err != nil {
+				return fmt.Errorf("validate plan include scope: %w", err)
+			}
+			desired = filterSpecsByInclude(desired, planIncludeSet)
+			currentState = filterStatesByInclude(currentState, planIncludeSet)
 			planApplyCfg, cfgErr := config.LoadFromFile(cfgFile)
 			if cfgErr != nil {
 				return fmt.Errorf("load config for stale-check: %w", cfgErr)
diff --git a/cmd/wfctl/infra_apply_include.go b/cmd/wfctl/infra_apply_include.go
index bcf6f187..a1c64dc7 100644
--- a/cmd/wfctl/infra_apply_include.go
+++ b/cmd/wfctl/infra_apply_include.go
@@ -30,6 +30,35 @@ func parseIncludeFlag(raw string) map[string]struct{} {
 	return out
 }
 
+func sortedIncludeNames(include map[string]struct{}) []string {
+	if len(include) == 0 {
+		return nil
+	}
+	names := make([]string, 0, len(include))
+	for name := range include {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	return names
+}
+
+func includeSetFromNames(names []string) map[string]struct{} {
+	if len(names) == 0 {
+		return nil
+	}
+	out := make(map[string]struct{}, len(names))
+	for _, name := range names {
+		if strings.TrimSpace(name) == "" {
+			continue
+		}
+		out[name] = struct{}{}
+	}
+	if len(out) == 0 {
+		return nil
+	}
+	return out
+}
+
 // validateIncludeSet returns an error if any name in the include set is not
 // declared in either specs or states. Resources can be in either side
 // (state-only resource is eligible for delete; spec-only resource is
diff --git a/cmd/wfctl/infra_apply_include_test.go b/cmd/wfctl/infra_apply_include_test.go
index 315a9e91..78bb1845 100644
--- a/cmd/wfctl/infra_apply_include_test.go
+++ b/cmd/wfctl/infra_apply_include_test.go
@@ -53,6 +53,52 @@ func TestParseIncludeFlag_AllWhitespace_ReturnsNil(t *testing.T) {
 	}
 }
 
+// ── include scope serialisation helpers ───────────────────────────────────────
+
+func TestSortedIncludeNames_NilInclude_ReturnsNil(t *testing.T) {
+	if got := sortedIncludeNames(nil); got != nil {
+		t.Errorf("nil include should yield nil names; got %v", got)
+	}
+}
+
+func TestSortedIncludeNames_ReturnsSortedNames(t *testing.T) {
+	got := sortedIncludeNames(map[string]struct{}{
+		"res-C": {},
+		"res-A": {},
+		"res-B": {},
+	})
+	want := []string{"res-A", "res-B", "res-C"}
+	if len(got) != len(want) {
+		t.Fatalf("expected %d names; got %v", len(want), got)
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("sorted names = %v, want %v", got, want)
+		}
+	}
+}
+
+func TestIncludeSetFromNames_EmptyInput_ReturnsNil(t *testing.T) {
+	if got := includeSetFromNames(nil); got != nil {
+		t.Errorf("nil names should yield nil include set; got %v", got)
+	}
+	if got := includeSetFromNames([]string{"", "   "}); got != nil {
+		t.Errorf("blank names should yield nil include set; got %v", got)
+	}
+}
+
+func TestIncludeSetFromNames_TrimsBlankNames(t *testing.T) {
+	got := includeSetFromNames([]string{"res-A", "", "   ", "res-B"})
+	if len(got) != 2 {
+		t.Fatalf("expected 2 names; got %v", got)
+	}
+	for _, name := range []string{"res-A", "res-B"} {
+		if _, ok := got[name]; !ok {
+			t.Errorf("expected %q in set; got %v", name, got)
+		}
+	}
+}
+
 // ── validateIncludeSet ────────────────────────────────────────────────────────
 
 func TestValidateIncludeSet_NilInclude_NoError(t *testing.T) {
diff --git a/cmd/wfctl/infra_apply_plan_test.go b/cmd/wfctl/infra_apply_plan_test.go
index bf43ec1f..b612e24a 100644
--- a/cmd/wfctl/infra_apply_plan_test.go
+++ b/cmd/wfctl/infra_apply_plan_test.go
@@ -149,6 +149,63 @@ modules:
 	}
 }
 
+// TestInfraApplyConsumesScopedPlan verifies that a persisted plan produced with
+// --include is hash-checked against the same scoped desired set at apply time.
+func TestInfraApplyConsumesScopedPlan(t *testing.T) {
+	dir := t.TempDir()
+	cfgPath := filepath.Join(dir, "infra.yaml")
+	if err := os.WriteFile(cfgPath, []byte(`
+modules:
+  - name: test-provider
+    type: iac.provider
+    config:
+      provider: fake-cloud
+      token: "test-token"
+
+  - name: my-db
+    type: infra.database
+    config:
+      provider: test-provider
+      engine: postgres
+      size: s
+
+  - name: other-db
+    type: infra.database
+    config:
+      provider: test-provider
+      engine: postgres
+      size: s
+`), 0o600); err != nil {
+		t.Fatalf("write config: %v", err)
+	}
+	planPath := filepath.Join(dir, "plan.json")
+
+	fake := &applyCapture{}
+	origResolve := resolveIaCProvider
+	resolveIaCProvider = func(_ context.Context, _ string, _ map[string]any) (interfaces.IaCProvider, io.Closer, error) {
+		return fake, nil, nil
+	}
+	t.Cleanup(func() { resolveIaCProvider = origResolve })
+	fake.installAsV2Dispatch(t)
+
+	if err := runInfraPlan([]string{"--config", cfgPath, "--include=my-db", "--output", planPath}); err != nil {
+		t.Fatalf("runInfraPlan: %v", err)
+	}
+
+	if err := runInfraApply([]string{"--auto-approve", "--config", cfgPath, "--plan", planPath}); err != nil {
+		t.Fatalf("runInfraApply scoped plan: %v", err)
+	}
+	if !fake.applyCalled {
+		t.Fatal("scoped plan was not applied")
+	}
+	if fake.appliedPlan == nil || len(fake.appliedPlan.Actions) != 1 {
+		t.Fatalf("applied plan actions = %+v, want exactly one action", fake.appliedPlan)
+	}
+	if got := fake.appliedPlan.Actions[0].Resource.Name; got != "my-db" {
+		t.Fatalf("applied resource = %q, want my-db", got)
+	}
+}
+
 func TestInfraApplyPlanSkipBootstrap(t *testing.T) {
 	dir := t.TempDir()
 	cfgPath := filepath.Join(dir, "infra.yaml")
diff --git a/interfaces/iac_state.go b/interfaces/iac_state.go
index 2031c0ac..7d78eb53 100644
--- a/interfaces/iac_state.go
+++ b/interfaces/iac_state.go
@@ -84,6 +84,12 @@ type IaCPlan struct {
 	// --plan compares this against the current config to detect stale plans.
 	DesiredHash string `json:"plan_hash,omitempty"`
 
+	// Include records the sorted resource names passed through
+	// `wfctl infra plan --include`. Empty means the plan covered the full
+	// desired set. wfctl infra apply --plan uses this to recompute DesiredHash
+	// against the same scope without requiring --include at apply time.
+	Include []string `json:"include,omitempty"`
+
 	// SchemaVersion is bumped when on-disk plan format changes (W-5 sets to 2 when JIT is required).
 	SchemaVersion int `json:"schema_version,omitempty"`
 
diff --git a/interfaces/iac_state_test.go b/interfaces/iac_state_test.go
index c9074794..0eaf7ba4 100644
--- a/interfaces/iac_state_test.go
+++ b/interfaces/iac_state_test.go
@@ -36,6 +36,21 @@ func TestIaCPlan_InputSnapshotField(t *testing.T) {
 	}
 }
 
+func TestIaCPlan_IncludeField(t *testing.T) {
+	p := IaCPlan{Include: []string{"bmw-dns"}}
+	data, err := json.Marshal(p)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got IaCPlan
+	if err := json.Unmarshal(data, &got); err != nil {
+		t.Fatal(err)
+	}
+	if len(got.Include) != 1 || got.Include[0] != "bmw-dns" {
+		t.Errorf("Include roundtrip failed: %v", got.Include)
+	}
+}
+
 func TestPlanAction_ResolvedConfigHashField(t *testing.T) {
 	// platform.ConfigHash returns a lower-case hex sha256 digest with no
 	// "sha256:" prefix; use a realistic 64-hex value so the test's expected