feat(iac): add resource ownership contract

cmd/wfctl/iac_typed_adapter.go

@@ -55,6 +55,7 @@ const (
 	iacServiceDriftDetector     = "workflow.plugin.external.iac.IaCProviderDriftDetector"
 	iacServiceCredentialRevoker = "workflow.plugin.external.iac.IaCProviderCredentialRevoker"
 	iacServiceRegionLister      = "workflow.plugin.external.iac.IaCProviderRegionLister"
+	iacServiceOwnership         = "workflow.plugin.external.iac.IaCProviderOwnership"
 	iacServiceMigrationRepairer = "workflow.plugin.external.iac.IaCProviderMigrationRepairer"
 	iacServiceValidator         = "workflow.plugin.external.iac.IaCProviderValidator"
 	iacServiceDriftConfigDetect = "workflow.plugin.external.iac.IaCProviderDriftConfigDetector"
@@ -83,6 +84,7 @@ type typedIaCAdapter struct {
 	drift        pb.IaCProviderDriftDetectorClient
 	revoker      pb.IaCProviderCredentialRevokerClient
 	regionLister pb.IaCProviderRegionListerClient
+	ownership    pb.IaCProviderOwnershipClient
 	repairer     pb.IaCProviderMigrationRepairerClient
 	validator    pb.IaCProviderValidatorClient
 	driftCfg     pb.IaCProviderDriftConfigDetectorClient
@@ -121,6 +123,9 @@ func newTypedIaCAdapter(conn *grpc.ClientConn, registered map[string]bool) *type
 	if registered[iacServiceRegionLister] {
 		a.regionLister = pb.NewIaCProviderRegionListerClient(conn)
 	}
+	if registered[iacServiceOwnership] {
+		a.ownership = pb.NewIaCProviderOwnershipClient(conn)
+	}
 	if registered[iacServiceMigrationRepairer] {
 		a.repairer = pb.NewIaCProviderMigrationRepairerClient(conn)
 	}
@@ -235,6 +240,13 @@ func (a *typedIaCAdapter) RegionLister() pb.IaCProviderRegionListerClient {
 	return a.regionLister
 }
 
+// Ownership returns the typed pb.IaCProviderOwnershipClient or nil when the
+// plugin did not register IaCProviderOwnership. Used by wfctl infra ownership
+// gates and resource-owner enumeration.
+func (a *typedIaCAdapter) Ownership() pb.IaCProviderOwnershipClient {
+	return a.ownership
+}
+
 // ListProviderRegions queries the optional region lister and returns sorted
 // provider region identifiers. The display name is intentionally ignored here
 // because infra-admin's current response shape carries region IDs only.
@@ -568,6 +580,60 @@ func (a *typedIaCAdapter) EnumerateByTag(ctx context.Context, tag string) ([]int
 	return refsFromPB(resp.GetRefs()), nil
 }
 
+// GetOwner satisfies interfaces.OwnershipProvider.
+func (a *typedIaCAdapter) GetOwner(ctx context.Context, ref interfaces.ResourceRef) (*interfaces.ResourceOwner, error) {
+	if a.ownership == nil {
+		return nil, unimplementedOptional(iacServiceOwnership)
+	}
+	resp, err := a.ownership.GetOwner(ctx, &pb.GetOwnerRequest{Ref: refToPB(ref)})
+	if err != nil {
+		return nil, translateRPCErr(err)
+	}
+	return &interfaces.ResourceOwner{
+		Ref:    ref,
+		Owner:  resp.GetOwner(),
+		Source: resp.GetSource(),
+	}, nil
+}
+
+// SetOwner satisfies interfaces.OwnershipProvider.
+func (a *typedIaCAdapter) SetOwner(ctx context.Context, ref interfaces.ResourceRef, owner string) error {
+	if a.ownership == nil {
+		return unimplementedOptional(iacServiceOwnership)
+	}
+	_, err := a.ownership.SetOwner(ctx, &pb.SetOwnerRequest{
+		Ref:   refToPB(ref),
+		Owner: owner,
+	})
+	return translateRPCErr(err)
+}
+
+// ListOwners satisfies interfaces.OwnershipProvider.
+func (a *typedIaCAdapter) ListOwners(ctx context.Context, filter interfaces.OwnerFilter) ([]interfaces.ResourceOwner, error) {
+	if a.ownership == nil {
+		return nil, unimplementedOptional(iacServiceOwnership)
+	}
+	resp, err := a.ownership.ListOwners(ctx, &pb.ListOwnersRequest{
+		Owner:        filter.Owner,
+		ResourceType: filter.ResourceType,
+	})
+	if err != nil {
+		return nil, translateRPCErr(err)
+	}
+	out := make([]interfaces.ResourceOwner, 0, len(resp.GetResources()))
+	for _, r := range resp.GetResources() {
+		if r == nil {
+			continue
+		}
+		out = append(out, interfaces.ResourceOwner{
+			Ref:    refFromPB(r.GetRef()),
+			Owner:  r.GetOwner(),
+			Source: r.GetSource(),
+		})
+	}
+	return out, nil
+}
+
 // DetectDriftWithSpecs satisfies interfaces.DriftConfigDetector. Routed
 // through the typed IaCProviderDriftConfigDetector service when the
 // plugin advertises it.
@@ -1449,6 +1515,7 @@ var (
 	_ interfaces.IaCProvider               = (*typedIaCAdapter)(nil)
 	_ interfaces.Enumerator                = (*typedIaCAdapter)(nil)
 	_ interfaces.EnumeratorAll             = (*typedIaCAdapter)(nil)
+	_ interfaces.OwnershipProvider         = (*typedIaCAdapter)(nil)
 	_ interfaces.DriftConfigDetector       = (*typedIaCAdapter)(nil)
 	_ interfaces.ProviderValidator         = (*typedIaCAdapter)(nil)
 	_ interfaces.ProviderCredentialRevoker = (*typedIaCAdapter)(nil)

cmd/wfctl/iac_typed_adapter_test.go

@@ -50,6 +50,9 @@ func TestTypedAdapter_SatisfiesIaCProvider(t *testing.T) {
 	if _, ok := any(a).(interfaces.EnumeratorAll); !ok {
 		t.Fatalf("typedIaCAdapter must satisfy interfaces.EnumeratorAll")
 	}
+	if _, ok := any(a).(interfaces.OwnershipProvider); !ok {
+		t.Fatalf("typedIaCAdapter must satisfy interfaces.OwnershipProvider")
+	}
 	if _, ok := any(a).(interfaces.DriftConfigDetector); !ok {
 		t.Fatalf("typedIaCAdapter must satisfy interfaces.DriftConfigDetector")
 	}
@@ -83,6 +86,17 @@ func TestTypedAdapter_OptionalReturnsUnimplementedSentinel(t *testing.T) {
 			_, err := a.EnumerateByTag(context.Background(), "production")
 			return err
 		}},
+		{"GetOwner", func() error {
+			_, err := a.GetOwner(context.Background(), interfaces.ResourceRef{Name: "app", Type: "infra.container_service"})
+			return err
+		}},
+		{"SetOwner", func() error {
+			return a.SetOwner(context.Background(), interfaces.ResourceRef{Name: "app", Type: "infra.container_service"}, "team-a")
+		}},
+		{"ListOwners", func() error {
+			_, err := a.ListOwners(context.Background(), interfaces.OwnerFilter{Owner: "team-a"})
+			return err
+		}},
 		{"DetectDrift", func() error {
 			_, err := a.DetectDrift(context.Background(), nil)
 			return err
@@ -420,6 +434,56 @@ func TestTypedAdapter_RegionLister_NilWhenNotRegistered(t *testing.T) {
 	}
 }
 
+func TestTypedAdapter_Ownership_PopulatedWhenRegistered(t *testing.T) {
+	conn := dialLazyConn(t)
+	adapter := newTypedIaCAdapter(conn, map[string]bool{
+		iacServiceOwnership: true,
+	})
+	if adapter.Ownership() == nil {
+		t.Error("Ownership() returned nil when IaCProviderOwnership is in registered set")
+	}
+}
+
+func TestTypedAdapter_Ownership_NilWhenNotRegistered(t *testing.T) {
+	conn := dialLazyConn(t)
+	adapter := newTypedIaCAdapter(conn, map[string]bool{
+		iacServiceEnumerator: true,
+	})
+	if adapter.Ownership() != nil {
+		t.Error("Ownership() returned non-nil when IaCProviderOwnership not registered")
+	}
+}
+
+func TestTypedAdapter_OwnershipRoundTrip(t *testing.T) {
+	stub := &ownershipStub{}
+	adapter := fixtureTypedAdapter{Ownership: stub}.build(t)
+	ref := interfaces.ResourceRef{Name: "app", Type: "infra.container_service", ProviderID: "app-1"}
+
+	got, err := adapter.GetOwner(context.Background(), ref)
+	if err != nil {
+		t.Fatalf("GetOwner: %v", err)
+	}
+	if got.Owner != "team-a" || got.Source != "tag:managed-by" || got.Ref != ref {
+		t.Fatalf("GetOwner = %+v, want owner/source/ref", got)
+	}
+	if err := adapter.SetOwner(context.Background(), ref, "team-b"); err != nil {
+		t.Fatalf("SetOwner: %v", err)
+	}
+	if stub.setOwner != "team-b" || stub.setRef != ref {
+		t.Fatalf("SetOwner recorded ref=%+v owner=%q, want %+v/team-b", stub.setRef, stub.setOwner, ref)
+	}
+	owners, err := adapter.ListOwners(context.Background(), interfaces.OwnerFilter{Owner: "team-a", ResourceType: "infra.container_service"})
+	if err != nil {
+		t.Fatalf("ListOwners: %v", err)
+	}
+	if len(owners) != 1 || owners[0].Owner != "team-a" || owners[0].Ref != ref {
+		t.Fatalf("ListOwners = %+v, want one owned resource", owners)
+	}
+	if stub.listOwner != "team-a" || stub.listType != "infra.container_service" {
+		t.Fatalf("ListOwners request owner/type = %q/%q", stub.listOwner, stub.listType)
+	}
+}
+
 func TestTypedAdapter_ListProviderRegions(t *testing.T) {
 	adapter := fixtureTypedAdapter{
 		RegionLister: regionListerStub{},
@@ -440,6 +504,37 @@ func TestTypedAdapter_ListProviderRegions(t *testing.T) {
 	}
 }
 
+type ownershipStub struct {
+	pb.UnimplementedIaCProviderOwnershipServer
+	setRef    interfaces.ResourceRef
+	setOwner  string
+	listOwner string
+	listType  string
+}
+
+func (s *ownershipStub) GetOwner(_ context.Context, req *pb.GetOwnerRequest) (*pb.GetOwnerResponse, error) {
+	if got := refFromPB(req.GetRef()); got.Name != "app" {
+		return nil, status.Errorf(codes.InvalidArgument, "name = %q, want app", got.Name)
+	}
+	return &pb.GetOwnerResponse{Owner: "team-a", Source: "tag:managed-by"}, nil
+}
+
+func (s *ownershipStub) SetOwner(_ context.Context, req *pb.SetOwnerRequest) (*pb.SetOwnerResponse, error) {
+	s.setRef = refFromPB(req.GetRef())
+	s.setOwner = req.GetOwner()
+	return &pb.SetOwnerResponse{}, nil
+}
+
+func (s *ownershipStub) ListOwners(_ context.Context, req *pb.ListOwnersRequest) (*pb.ListOwnersResponse, error) {
+	s.listOwner = req.GetOwner()
+	s.listType = req.GetResourceType()
+	return &pb.ListOwnersResponse{Resources: []*pb.OwnedResource{{
+		Ref:    refToPB(interfaces.ResourceRef{Name: "app", Type: "infra.container_service", ProviderID: "app-1"}),
+		Owner:  "team-a",
+		Source: "tag:managed-by",
+	}}}, nil
+}
+
 type regionListerStub struct {
 	pb.UnimplementedIaCProviderRegionListerServer
 }

cmd/wfctl/iac_typed_fixture_test.go

@@ -94,6 +94,7 @@ type fixtureTypedAdapter struct {
 	DriftDetector     pb.IaCProviderDriftDetectorServer
 	CredentialRevoker pb.IaCProviderCredentialRevokerServer
 	RegionLister      pb.IaCProviderRegionListerServer
+	Ownership         pb.IaCProviderOwnershipServer
 	MigrationRepairer pb.IaCProviderMigrationRepairerServer
 	Validator         pb.IaCProviderValidatorServer
 	DriftConfigDetect pb.IaCProviderDriftConfigDetectorServer
@@ -136,6 +137,10 @@ func (f fixtureTypedAdapter) build(t *testing.T) *typedIaCAdapter {
 		pb.RegisterIaCProviderRegionListerServer(server, f.RegionLister)
 		registered[iacServiceRegionLister] = true
 	}
+	if f.Ownership != nil {
+		pb.RegisterIaCProviderOwnershipServer(server, f.Ownership)
+		registered[iacServiceOwnership] = true
+	}
 	if f.MigrationRepairer != nil {
 		pb.RegisterIaCProviderMigrationRepairerServer(server, f.MigrationRepairer)
 		registered[iacServiceMigrationRepairer] = true

cmd/wfctl/infra.go

@@ -95,6 +95,8 @@ func runInfra(args []string) error {
 		return runInfraSecurityCheck(args[1:])
 	case "cleanup":
 		return runInfraCleanup(args[1:])
+	case "owners":
+		return runInfraOwners(args[1:])
 	case "audit-secrets":
 		if rc := runInfraAuditSecrets(args[1:], os.Stdout); rc != 0 {
 			return fmt.Errorf("audit-secrets exited with code %d", rc)
@@ -143,6 +145,7 @@ Actions:
                    (list-resources, get-resource, list-types,
                     list-providers, generate-config, audit-tail)
   cleanup        Tag-based force-cleanup across providers (--tag NAME [--fix])
+  owners         List cloud resources carrying a provider ownership marker
   audit-secrets  Report provider_credential anti-patterns in secrets.generate
   audit-keys     List cloud-side resources of --type via the provider's EnumeratorAll
   prune          Destructively delete cloud resources by --created-before / --exclude-access-key (two-key opt-in)
@@ -159,6 +162,8 @@ Options:
   --output <file>      Write plan to JSON file (plan only)
   --show-sensitive/-S  Show sensitive values in plaintext (plan/apply only)
   --tag <name>         Tag to match resources (cleanup only; required)
+  --owner <name>       Owner identity for ownership checks/listing
+  --force-owner        Override mismatched ownership on apply (requires --owner)
   --dry-run            Preview only (cleanup; default true)
   --write              Update config file in place (derive only)
   --non-interactive    Fail instead of prompting for ambiguous choices (derive only)
@@ -1309,6 +1314,10 @@ func runInfraApply(args []string) error {
 	fs.BoolVar(&skipBootstrapFlag, "skip-bootstrap", false, "Skip auto-bootstrap before apply; use only when required secrets/state already exist")
 	var waitFlag bool
 	fs.BoolVar(&waitFlag, "wait", false, "Wait for deployable infra resources to become healthy before exiting")
+	var ownerFlag string
+	fs.StringVar(&ownerFlag, "owner", "", "Owner identity for generic cloud-resource ownership checks (defaults to WORKFLOW_RESOURCE_OWNER)")
+	var forceOwnerFlag bool
+	fs.BoolVar(&forceOwnerFlag, "force-owner", false, "Override mismatched generic cloud-resource ownership for this apply (requires --owner or WORKFLOW_RESOURCE_OWNER)")
 	var allowReplaceFlag string
 	fs.StringVar(&allowReplaceFlag, "allow-replace", "",
 		"Comma-separated list of resource names whose protected: true status is overridden for this apply (replace/delete actions only)")
@@ -1338,6 +1347,11 @@ func runInfraApply(args []string) error {
 		return fmt.Errorf("--include cannot be combined with --plan (use --include at plan time, then apply with --plan; the plan already carries the scope)")
 	}
 
+	owner := ownerFromFlagOrEnv(ownerFlag)
+	if forceOwnerFlag && owner == "" {
+		return fmt.Errorf("--force-owner requires --owner or WORKFLOW_RESOURCE_OWNER")
+	}
+
 	// W-6/T6.1: publish the parsed --allow-replace set for the apply
 	// path's gate (validateAllowReplaceProtected, called from both
 	// applyWithProviderAndStore and applyPrecomputedPlanWithStore).
@@ -1351,6 +1365,15 @@ func runInfraApply(args []string) error {
 	currentInfraApplyWait = waitFlag
 	defer func() { currentInfraApplyWait = prevInfraApplyWait }()
 
+	prevApplyOwner := currentApplyOwner
+	prevApplyForceOwner := currentApplyForceOwner
+	currentApplyOwner = owner
+	currentApplyForceOwner = forceOwnerFlag
+	defer func() {
+		currentApplyOwner = prevApplyOwner
+		currentApplyForceOwner = prevApplyForceOwner
+	}()
+
 	// Publish the --include flag value for the apply path's filter helpers
 	// (including dry-run). Reset to "" at the top of every invocation so the
 	// filter fails open (all-resources) on subsequent invocations that do not

cmd/wfctl/infra_apply.go

@@ -460,6 +460,7 @@ func applyWithProviderAndStore(ctx context.Context, provider interfaces.IaCProvi
 	// postcondition + IaCProviderFinalizer fan-out).
 	hooks := statePersistenceHooks(store, secretsProvider, provider, providerType, plan.ID, hydratedOut)
 	wireDNSGateIntoHooks(&hooks, provider)
+	wireOwnershipGateIntoHooks(&hooks, provider)
 	result, err := applyV2ApplyPlanWithHooksFn(ctx, provider, &plan, hooks)
 	// printDriftReportIfAny surfaces input-drift to the operator on
 	// success OR partial failure — silently no-ops on empty reports.
@@ -1576,6 +1577,7 @@ func applyPrecomputedPlanWithStore(ctx context.Context, plan interfaces.IaCPlan,
 	// v2 is the only supported dispatch per ADR 0024 + workflow#699.
 	hooks := statePersistenceHooks(store, secretsProvider, provider, providerType, plan.ID, hydratedOut)
 	wireDNSGateIntoHooks(&hooks, provider)
+	wireOwnershipGateIntoHooks(&hooks, provider)
 	result, err := applyV2ApplyPlanWithHooksFn(ctx, provider, &plan, hooks)
 	if result != nil {
 		printDriftReportIfAny(w, result)

cmd/wfctl/infra_apply_dns_gate.go

@@ -102,7 +102,7 @@ func extractDNSRecords(records any) []map[string]any {
 // AFTER constructing the hooks via statePersistenceHooks so the OnBefore
 // closure shares the same provider reference.
 func wireDNSGateIntoHooks(hooks *wfctlhelpers.ApplyPlanHooks, provider interfaces.IaCProvider) {
-	hooks.OnBeforeAction = dnsGateHook(provider)
+	appendOnBeforeActionHook(hooks, dnsGateHook(provider))
 }
 
 // Compile-time guard that the policy + gate packages stay in dependency