diff --git a/cmd/plugin/plugin.json b/cmd/plugin/plugin.json
index dfc32d4..58ec347 100644
--- a/cmd/plugin/plugin.json
+++ b/cmd/plugin/plugin.json
@@ -6,7 +6,7 @@
   "license": "MIT",
   "type": "external",
   "tier": "community",
-  "minEngineVersion": "0.68.2",
+  "minEngineVersion": "0.69.1",
   "required_secrets": [
     {
       "name": "DIGITALOCEAN_TOKEN",
@@ -26,6 +26,7 @@
     "workflow.plugin.external.iac.IaCProviderLogCapture",
     "workflow.plugin.external.iac.IaCProviderRequirementMapper",
     "workflow.plugin.external.iac.IaCProviderRegionLister",
+    "workflow.plugin.external.iac.IaCProviderOwnership",
     "workflow.plugin.external.iac.IaCProviderFinalizer",
     "workflow.plugin.external.iac.ResourceDriver",
     "workflow.plugin.external.iac.IaCStateBackend"
diff --git a/go.mod b/go.mod
index f99b5a2..c40321c 100644
--- a/go.mod
+++ b/go.mod
@@ -3,7 +3,7 @@ module github.com/GoCodeAlone/workflow-plugin-digitalocean
 go 1.26.0
 
 require (
-	github.com/GoCodeAlone/workflow v0.68.2
+	github.com/GoCodeAlone/workflow v0.69.1
 	github.com/aws/aws-sdk-go-v2 v1.41.6
 	github.com/aws/aws-sdk-go-v2/config v1.32.16
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.15
diff --git a/go.sum b/go.sum
index 62435c5..1b44e64 100644
--- a/go.sum
+++ b/go.sum
@@ -22,8 +22,8 @@ github.com/GoCodeAlone/modular/modules/jsonschema v1.17.0 h1:zoWioqUvuNNDfnjHA1s
 github.com/GoCodeAlone/modular/modules/jsonschema v1.17.0/go.mod h1:GDU/jsD6AddmXKedj0wZwieUIaQsTBSGMzuj+XHXMrw=
 github.com/GoCodeAlone/modular/modules/reverseproxy/v2 v2.10.0 h1:+2M/ecyCxDiXfJM4ibcERuu/BBeIbLTQNcVgRsllR64=
 github.com/GoCodeAlone/modular/modules/reverseproxy/v2 v2.10.0/go.mod h1:tlVH1mA5yuU8CB7R7+HXIRaBixZoNid6h+5tew5u3FU=
-github.com/GoCodeAlone/workflow v0.68.2 h1:U0ksQOkIwDReuw+nz4kRoCeYwahoBaItqLzwYIRm758=
-github.com/GoCodeAlone/workflow v0.68.2/go.mod h1:4UwFYm1cM8a/AvGNb1CZAuob0b0gq7552sxcNMdDALA=
+github.com/GoCodeAlone/workflow v0.69.1 h1:X9smiC8ZkRAtD7d5d0IVfWEZzNfHeQPSwLtm0jcEiWI=
+github.com/GoCodeAlone/workflow v0.69.1/go.mod h1:4UwFYm1cM8a/AvGNb1CZAuob0b0gq7552sxcNMdDALA=
 github.com/GoCodeAlone/yaegi v0.17.2 h1:WK6Y6e0t1a6U7r+S2dN3CGWW1PizYD3zO0zneToZPxM=
 github.com/GoCodeAlone/yaegi v0.17.2/go.mod h1:z5Pr6Wse6QJcQvpgxTxzMAevFarH0N37TG88Y9dprx0=
 github.com/IBM/sarama v1.47.0 h1:GcQFEd12+KzfPYeLgN69Fh7vLCtYRhVIx0rO4TZO318=
diff --git a/internal/drivers/cache.go b/internal/drivers/cache.go
index 93b3377..5d94d2e 100644
--- a/internal/drivers/cache.go
+++ b/internal/drivers/cache.go
@@ -185,10 +185,15 @@ func (d *CacheDriver) Scale(ctx context.Context, ref interfaces.ResourceRef, rep
 }
 
 func cacheOutput(db *godo.Database) *interfaces.ResourceOutput {
+	tags := make([]any, 0, len(db.Tags))
+	for _, tag := range db.Tags {
+		tags = append(tags, tag)
+	}
 	outputs := map[string]any{
 		"engine":  db.EngineSlug,
 		"region":  db.RegionSlug,
 		"size":    db.SizeSlug,
+		"tags":    tags,
 		"version": db.VersionSlug,
 	}
 	if db.Connection != nil {
diff --git a/internal/drivers/database.go b/internal/drivers/database.go
index 2d45a18..0c50c34 100644
--- a/internal/drivers/database.go
+++ b/internal/drivers/database.go
@@ -613,11 +613,16 @@ func trustedSourceFirewallRulesFromConfig(cfg map[string]any) ([]*godo.DatabaseF
 }
 
 func dbOutput(db *godo.Database) *interfaces.ResourceOutput {
+	tags := make([]any, 0, len(db.Tags))
+	for _, tag := range db.Tags {
+		tags = append(tags, tag)
+	}
 	outputs := map[string]any{
 		"engine":    db.EngineSlug,
 		"num_nodes": float64(db.NumNodes),
 		"region":    db.RegionSlug,
 		"size":      db.SizeSlug,
+		"tags":      tags,
 		"version":   db.VersionSlug,
 	}
 	if db.Connection != nil {
diff --git a/internal/iacserver.go b/internal/iacserver.go
index 7b0bca8..017fe49 100644
--- a/internal/iacserver.go
+++ b/internal/iacserver.go
@@ -57,6 +57,7 @@ type doIaCServer struct {
 	pb.UnimplementedIaCProviderLogCaptureServer
 	pb.UnimplementedIaCProviderRequirementMapperServer
 	pb.UnimplementedIaCProviderRegionListerServer
+	pb.UnimplementedIaCProviderOwnershipServer
 	// pb.UnimplementedIaCProviderFinalizerServer satisfies the
 	// mustEmbedUnimplementedIaCProviderFinalizerServer() forward-compat
 	// requirement on pb.IaCProviderFinalizerServer (workflow#695 Phase 2.5).
@@ -121,6 +122,7 @@ var (
 	_ pb.IaCProviderLogCaptureServer          = (*doIaCServer)(nil)
 	_ pb.IaCProviderRequirementMapperServer   = (*doIaCServer)(nil)
 	_ pb.IaCProviderRegionListerServer        = (*doIaCServer)(nil)
+	_ pb.IaCProviderOwnershipServer           = (*doIaCServer)(nil)
 	// IaCProviderFinalizer is the workflow#695 Phase 2.5 optional service
 	// — DO plugin implements FinalizeApply server-side to host the
 	// deferred-flush iteration previously held inline in the v1
diff --git a/internal/iacserver_mapper_test.go b/internal/iacserver_mapper_test.go
index 25046d3..018f402 100644
--- a/internal/iacserver_mapper_test.go
+++ b/internal/iacserver_mapper_test.go
@@ -177,8 +177,8 @@ func TestPluginManifestAdvertisesRequirementMapper(t *testing.T) {
 	if err := json.Unmarshal(data, &manifest); err != nil {
 		t.Fatalf("parse plugin.json: %v", err)
 	}
-	if manifest.MinEngineVersion != "0.68.2" {
-		t.Fatalf("minEngineVersion = %q, want 0.68.2", manifest.MinEngineVersion)
+	if manifest.MinEngineVersion != "0.69.1" {
+		t.Fatalf("minEngineVersion = %q, want 0.69.1", manifest.MinEngineVersion)
 	}
 	const mapperService = "workflow.plugin.external.iac.IaCProviderRequirementMapper"
 	for _, svc := range manifest.IaCServices {
diff --git a/internal/ownership.go b/internal/ownership.go
new file mode 100644
index 0000000..9a567f7
--- /dev/null
+++ b/internal/ownership.go
@@ -0,0 +1,332 @@
+package internal
+
+import (
+	"context"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/GoCodeAlone/workflow-plugin-digitalocean/internal/drivers"
+	"github.com/GoCodeAlone/workflow/interfaces"
+	"github.com/digitalocean/godo"
+)
+
+const (
+	ownershipTagPrefix = "workflow-owner:"
+	ownershipTagSource = "tag:workflow-owner"
+	ownershipTagMaxLen = 255
+)
+
+var _ interfaces.OwnershipProvider = (*DOProvider)(nil)
+
+func (p *DOProvider) GetOwner(ctx context.Context, ref interfaces.ResourceRef) (*interfaces.ResourceOwner, error) {
+	if p.client == nil {
+		return nil, fmt.Errorf("digitalocean: GetOwner called on provider that is not initialized — call Initialize first")
+	}
+	tags, err := p.resourceTags(ctx, ref)
+	if err != nil {
+		return nil, err
+	}
+	for _, tag := range tags {
+		owner, ok := ownerFromTag(tag)
+		if ok {
+			return &interfaces.ResourceOwner{Ref: ref, Owner: owner, Source: ownershipTagSource}, nil
+		}
+	}
+	return &interfaces.ResourceOwner{Ref: ref, Source: ownershipTagSource}, nil
+}
+
+func (p *DOProvider) SetOwner(ctx context.Context, ref interfaces.ResourceRef, owner string) error {
+	if p.client == nil {
+		return fmt.Errorf("digitalocean: SetOwner called on provider that is not initialized — call Initialize first")
+	}
+	resource, err := ownershipTagResource(ref)
+	if err != nil {
+		return err
+	}
+	tag, err := ownerTagName(owner)
+	if err != nil {
+		return err
+	}
+
+	tags, err := p.resourceTags(ctx, ref)
+	if err != nil {
+		return err
+	}
+	for _, existing := range tags {
+		if _, ok := ownerFromTag(existing); !ok || existing == tag {
+			continue
+		}
+		if _, err := p.client.Tags.UntagResources(ctx, existing, &godo.UntagResourcesRequest{Resources: []godo.Resource{resource}}); err != nil {
+			return fmt.Errorf("digitalocean: remove owner tag %q from %s/%s: %w", existing, ref.Type, ref.Name, drivers.WrapGodoError(err))
+		}
+	}
+	if _, _, err := p.client.Tags.Create(ctx, &godo.TagCreateRequest{Name: tag}); err != nil && !isTagAlreadyExists(err) {
+		return fmt.Errorf("digitalocean: create owner tag %q: %w", tag, drivers.WrapGodoError(err))
+	}
+	if _, err := p.client.Tags.TagResources(ctx, tag, &godo.TagResourcesRequest{Resources: []godo.Resource{resource}}); err != nil {
+		return fmt.Errorf("digitalocean: tag %s/%s with owner %q: %w", ref.Type, ref.Name, owner, drivers.WrapGodoError(err))
+	}
+	return nil
+}
+
+func (p *DOProvider) ListOwners(ctx context.Context, filter interfaces.OwnerFilter) ([]interfaces.ResourceOwner, error) {
+	if p.client == nil {
+		return nil, fmt.Errorf("digitalocean: ListOwners called on provider that is not initialized — call Initialize first")
+	}
+	if filter.Owner != "" {
+		return p.listOwnersForTag(ctx, filter)
+	}
+
+	var out []interfaces.ResourceOwner
+	page := &godo.ListOptions{Page: 1, PerPage: 200}
+	for {
+		tags, resp, err := p.client.Tags.List(ctx, page)
+		if err != nil {
+			return nil, fmt.Errorf("digitalocean: list owner tags: %w", drivers.WrapGodoError(err))
+		}
+		for _, tag := range tags {
+			owner, ok := ownerFromTag(tag.Name)
+			if !ok {
+				continue
+			}
+			owners, err := p.listOwnersForTag(ctx, interfaces.OwnerFilter{Owner: owner, ResourceType: filter.ResourceType})
+			if err != nil {
+				return nil, err
+			}
+			out = append(out, owners...)
+		}
+		if resp == nil || resp.Links == nil || resp.Links.IsLastPage() {
+			break
+		}
+		nextPage, err := resp.Links.CurrentPage()
+		if err != nil {
+			return nil, fmt.Errorf("digitalocean: paginate owner tags: %w", err)
+		}
+		page.Page = nextPage + 1
+	}
+	return out, nil
+}
+
+func (p *DOProvider) listOwnersForTag(ctx context.Context, filter interfaces.OwnerFilter) ([]interfaces.ResourceOwner, error) {
+	tag, err := ownerTagName(filter.Owner)
+	if err != nil {
+		return nil, err
+	}
+	refs, err := p.ownershipRefsByTag(ctx, tag, filter.ResourceType)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]interfaces.ResourceOwner, 0, len(refs))
+	for _, ref := range refs {
+		if filter.ResourceType != "" && ref.Type != filter.ResourceType {
+			continue
+		}
+		out = append(out, interfaces.ResourceOwner{Ref: ref, Owner: filter.Owner, Source: ownershipTagSource})
+	}
+	return out, nil
+}
+
+func (p *DOProvider) ownershipRefsByTag(ctx context.Context, tag, resourceType string) ([]interfaces.ResourceRef, error) {
+	if resourceType == "" {
+		return p.EnumerateByTag(ctx, tag)
+	}
+	if _, _, err := p.client.Tags.Get(ctx, tag); err != nil {
+		var doErr *godo.ErrorResponse
+		if errors.As(err, &doErr) && doErr.Response != nil && doErr.Response.StatusCode == 404 {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("digitalocean: get owner tag %q: %w", tag, drivers.WrapGodoError(err))
+	}
+
+	switch resourceType {
+	case "infra.droplet":
+		return p.ownershipDropletRefsByTag(ctx, tag)
+	case "infra.volume":
+		return p.ownershipVolumeRefsByTag(ctx, tag)
+	case "infra.database", "infra.cache":
+		return p.ownershipDatabaseRefsByTag(ctx, tag, resourceType)
+	default:
+		return nil, nil
+	}
+}
+
+func (p *DOProvider) ownershipDropletRefsByTag(ctx context.Context, tag string) ([]interfaces.ResourceRef, error) {
+	var refs []interfaces.ResourceRef
+	page := &godo.ListOptions{Page: 1, PerPage: 200}
+	for {
+		droplets, resp, err := p.client.Droplets.ListByTag(ctx, tag, page)
+		if err != nil {
+			return nil, fmt.Errorf("digitalocean: list droplets by owner tag %q: %w", tag, drivers.WrapGodoError(err))
+		}
+		for _, d := range droplets {
+			refs = append(refs, interfaces.ResourceRef{Name: d.Name, Type: "infra.droplet", ProviderID: fmt.Sprint(d.ID)})
+		}
+		if resp == nil || resp.Links == nil || resp.Links.IsLastPage() {
+			break
+		}
+		nextPage, err := resp.Links.CurrentPage()
+		if err != nil {
+			return nil, fmt.Errorf("digitalocean: paginate droplets by owner tag: %w", err)
+		}
+		page.Page = nextPage + 1
+	}
+	return refs, nil
+}
+
+func (p *DOProvider) ownershipVolumeRefsByTag(ctx context.Context, tag string) ([]interfaces.ResourceRef, error) {
+	var refs []interfaces.ResourceRef
+	page := &godo.ListVolumeParams{ListOptions: &godo.ListOptions{Page: 1, PerPage: 200}}
+	for {
+		volumes, resp, err := p.client.Storage.ListVolumes(ctx, page)
+		if err != nil {
+			return nil, fmt.Errorf("digitalocean: list volumes by owner tag %q: %w", tag, drivers.WrapGodoError(err))
+		}
+		for _, v := range volumes {
+			if stringSliceContains(v.Tags, tag) {
+				refs = append(refs, interfaces.ResourceRef{Name: v.Name, Type: "infra.volume", ProviderID: v.ID})
+			}
+		}
+		if resp == nil || resp.Links == nil || resp.Links.IsLastPage() {
+			break
+		}
+		nextPage, err := resp.Links.CurrentPage()
+		if err != nil {
+			return nil, fmt.Errorf("digitalocean: paginate volumes by owner tag: %w", err)
+		}
+		page.ListOptions.Page = nextPage + 1
+	}
+	return refs, nil
+}
+
+func (p *DOProvider) ownershipDatabaseRefsByTag(ctx context.Context, tag, resourceType string) ([]interfaces.ResourceRef, error) {
+	var refs []interfaces.ResourceRef
+	page := &godo.ListOptions{Page: 1, PerPage: 200}
+	for {
+		databases, resp, err := p.client.Databases.List(ctx, page)
+		if err != nil {
+			return nil, fmt.Errorf("digitalocean: list databases by owner tag %q: %w", tag, drivers.WrapGodoError(err))
+		}
+		for _, db := range databases {
+			if !stringSliceContains(db.Tags, tag) {
+				continue
+			}
+			dbType := "infra.database"
+			if db.EngineSlug == "redis" {
+				dbType = "infra.cache"
+			}
+			if dbType == resourceType {
+				refs = append(refs, interfaces.ResourceRef{Name: db.Name, Type: dbType, ProviderID: db.ID})
+			}
+		}
+		if resp == nil || resp.Links == nil || resp.Links.IsLastPage() {
+			break
+		}
+		nextPage, err := resp.Links.CurrentPage()
+		if err != nil {
+			return nil, fmt.Errorf("digitalocean: paginate databases by owner tag: %w", err)
+		}
+		page.Page = nextPage + 1
+	}
+	return refs, nil
+}
+
+func (p *DOProvider) resourceTags(ctx context.Context, ref interfaces.ResourceRef) ([]string, error) {
+	if _, err := ownershipTagResource(ref); err != nil {
+		return nil, err
+	}
+	driver, err := p.ResourceDriver(ref.Type)
+	if err != nil {
+		return nil, err
+	}
+	out, err := driver.Read(ctx, ref)
+	if err != nil {
+		return nil, err
+	}
+	if out == nil {
+		return nil, nil
+	}
+	return stringSliceFromAny(out.Outputs["tags"]), nil
+}
+
+func ownershipTagResource(ref interfaces.ResourceRef) (godo.Resource, error) {
+	if ref.ProviderID == "" {
+		return godo.Resource{}, fmt.Errorf("digitalocean: ownership requires provider id for %s/%s", ref.Type, ref.Name)
+	}
+	switch ref.Type {
+	case "infra.droplet":
+		return godo.Resource{ID: ref.ProviderID, Type: godo.DropletResourceType}, nil
+	case "infra.volume":
+		return godo.Resource{ID: ref.ProviderID, Type: godo.VolumeResourceType}, nil
+	case "infra.database", "infra.cache":
+		return godo.Resource{ID: ref.ProviderID, Type: godo.DatabaseResourceType}, nil
+	default:
+		return godo.Resource{}, fmt.Errorf("digitalocean: ownership unsupported for %s: %w", ref.Type, interfaces.ErrProviderMethodUnimplemented)
+	}
+}
+
+func ownerTagName(owner string) (string, error) {
+	if owner == "" {
+		return "", fmt.Errorf("digitalocean: owner must be non-empty")
+	}
+	suffix := owner
+	if !isPlainOwnerSuffix(owner) {
+		suffix = "b64:" + base64.RawURLEncoding.EncodeToString([]byte(owner))
+	}
+	tag := ownershipTagPrefix + suffix
+	if len(tag) > ownershipTagMaxLen {
+		return "", fmt.Errorf("digitalocean: owner tag exceeds %d characters", ownershipTagMaxLen)
+	}
+	return tag, nil
+}
+
+func ownerFromTag(tag string) (string, bool) {
+	suffix, ok := strings.CutPrefix(tag, ownershipTagPrefix)
+	if !ok || suffix == "" {
+		return "", false
+	}
+	if encoded, ok := strings.CutPrefix(suffix, "b64:"); ok {
+		raw, err := base64.RawURLEncoding.DecodeString(encoded)
+		if err != nil {
+			return "", false
+		}
+		return string(raw), true
+	}
+	return suffix, true
+}
+
+func isPlainOwnerSuffix(owner string) bool {
+	if strings.HasPrefix(owner, "b64:") {
+		return false
+	}
+	for _, r := range owner {
+		if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == ':' || r == '-' || r == '_' {
+			continue
+		}
+		return false
+	}
+	return true
+}
+
+func isTagAlreadyExists(err error) bool {
+	var doErr *godo.ErrorResponse
+	return errors.As(err, &doErr) && doErr.Response != nil && doErr.Response.StatusCode == 422
+}
+
+func stringSliceFromAny(v any) []string {
+	switch tags := v.(type) {
+	case []string:
+		return append([]string(nil), tags...)
+	case []any:
+		out := make([]string, 0, len(tags))
+		for _, tag := range tags {
+			if s, ok := tag.(string); ok {
+				out = append(out, s)
+			}
+		}
+		return out
+	default:
+		return nil
+	}
+}
diff --git a/internal/ownership_server.go b/internal/ownership_server.go
new file mode 100644
index 0000000..206ec60
--- /dev/null
+++ b/internal/ownership_server.go
@@ -0,0 +1,42 @@
+package internal
+
+import (
+	"context"
+
+	"github.com/GoCodeAlone/workflow/interfaces"
+	pb "github.com/GoCodeAlone/workflow/plugin/external/proto"
+)
+
+func (s *doIaCServer) GetOwner(ctx context.Context, req *pb.GetOwnerRequest) (*pb.GetOwnerResponse, error) {
+	owner, err := s.provider.GetOwner(ctx, refFromPB(req.GetRef()))
+	if err != nil {
+		return nil, err
+	}
+	return &pb.GetOwnerResponse{Owner: owner.Owner, Source: owner.Source}, nil
+}
+
+func (s *doIaCServer) SetOwner(ctx context.Context, req *pb.SetOwnerRequest) (*pb.SetOwnerResponse, error) {
+	if err := s.provider.SetOwner(ctx, refFromPB(req.GetRef()), req.GetOwner()); err != nil {
+		return nil, err
+	}
+	return &pb.SetOwnerResponse{}, nil
+}
+
+func (s *doIaCServer) ListOwners(ctx context.Context, req *pb.ListOwnersRequest) (*pb.ListOwnersResponse, error) {
+	owners, err := s.provider.ListOwners(ctx, interfaces.OwnerFilter{
+		Owner:        req.GetOwner(),
+		ResourceType: req.GetResourceType(),
+	})
+	if err != nil {
+		return nil, err
+	}
+	out := make([]*pb.OwnedResource, 0, len(owners))
+	for _, owner := range owners {
+		out = append(out, &pb.OwnedResource{
+			Ref:    refToPB(owner.Ref),
+			Owner:  owner.Owner,
+			Source: owner.Source,
+		})
+	}
+	return &pb.ListOwnersResponse{Resources: out}, nil
+}
diff --git a/internal/ownership_test.go b/internal/ownership_test.go
new file mode 100644
index 0000000..960b07b
--- /dev/null
+++ b/internal/ownership_test.go
@@ -0,0 +1,373 @@
+package internal
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/GoCodeAlone/workflow-plugin-digitalocean/internal/drivers"
+	"github.com/GoCodeAlone/workflow/interfaces"
+	pb "github.com/GoCodeAlone/workflow/plugin/external/proto"
+	sdk "github.com/GoCodeAlone/workflow/plugin/external/sdk"
+	"github.com/digitalocean/godo"
+	"google.golang.org/grpc"
+)
+
+func TestDOProvider_ImplementsOwnershipProvider(t *testing.T) {
+	var _ interfaces.OwnershipProvider = (*DOProvider)(nil)
+}
+
+func TestDOIaCServer_RegistersOwnershipProvider(t *testing.T) {
+	server := grpc.NewServer()
+	if err := sdk.RegisterAllIaCProviderServices(server, NewIaCServer()); err != nil {
+		t.Fatalf("RegisterAllIaCProviderServices: %v", err)
+	}
+	if _, ok := server.GetServiceInfo()[pb.IaCProviderOwnership_ServiceDesc.ServiceName]; !ok {
+		t.Fatalf("registered services missing %s", pb.IaCProviderOwnership_ServiceDesc.ServiceName)
+	}
+}
+
+func TestPluginManifestAdvertisesOwnershipProvider(t *testing.T) {
+	data, err := os.ReadFile(filepath.Join(testRepoRoot(t), "plugin.json"))
+	if err != nil {
+		t.Fatalf("read plugin.json: %v", err)
+	}
+	var manifest struct {
+		IaCServices []string `json:"iacServices"`
+	}
+	if err := json.Unmarshal(data, &manifest); err != nil {
+		t.Fatalf("parse plugin.json: %v", err)
+	}
+	if !containsString(manifest.IaCServices, pb.IaCProviderOwnership_ServiceDesc.ServiceName) {
+		t.Fatalf("iacServices missing %s: %v", pb.IaCProviderOwnership_ServiceDesc.ServiceName, manifest.IaCServices)
+	}
+}
+
+func TestDOProvider_GetOwnerFromDropletTags(t *testing.T) {
+	api := &ownershipFakeAPI{
+		droplets: map[string]ownershipResource{
+			"1001": {id: "1001", name: "app", tags: []string{"other", "workflow-owner:team-a"}},
+		},
+	}
+	p, _ := newProviderForOwnershipTest(t, api)
+
+	owner, err := p.GetOwner(context.Background(), interfaces.ResourceRef{
+		Name:       "app",
+		Type:       "infra.droplet",
+		ProviderID: "1001",
+	})
+	if err != nil {
+		t.Fatalf("GetOwner: %v", err)
+	}
+	if owner.Owner != "team-a" {
+		t.Fatalf("owner = %q, want team-a", owner.Owner)
+	}
+	if owner.Source != "tag:workflow-owner" {
+		t.Fatalf("source = %q, want tag:workflow-owner", owner.Source)
+	}
+}
+
+func TestDOProvider_GetOwnerFromSupportedResourceTags(t *testing.T) {
+	encodedOwner := "team a@example.com"
+	encodedTag, err := ownerTagName(encodedOwner)
+	if err != nil {
+		t.Fatalf("ownerTagName: %v", err)
+	}
+	api := &ownershipFakeAPI{
+		droplets: map[string]ownershipResource{
+			"1001": {id: "1001", name: "app", tags: []string{"workflow-owner:team-a"}},
+		},
+		volumes: map[string]ownershipResource{
+			"vol-1": {id: "vol-1", name: "data", tags: []string{encodedTag}},
+		},
+		databases: map[string]ownershipResource{
+			"db-1":    {id: "db-1", name: "pg", engine: "pg", tags: []string{"workflow-owner:team-db"}},
+			"cache-1": {id: "cache-1", name: "redis", engine: "redis", tags: []string{"workflow-owner:team-cache"}},
+		},
+	}
+	p, _ := newProviderForOwnershipTest(t, api)
+
+	tests := []struct {
+		name string
+		ref  interfaces.ResourceRef
+		want string
+	}{
+		{
+			name: "droplet",
+			ref:  interfaces.ResourceRef{Name: "app", Type: "infra.droplet", ProviderID: "1001"},
+			want: "team-a",
+		},
+		{
+			name: "volume encoded owner",
+			ref:  interfaces.ResourceRef{Name: "data", Type: "infra.volume", ProviderID: "vol-1"},
+			want: encodedOwner,
+		},
+		{
+			name: "database",
+			ref:  interfaces.ResourceRef{Name: "pg", Type: "infra.database", ProviderID: "db-1"},
+			want: "team-db",
+		},
+		{
+			name: "cache",
+			ref:  interfaces.ResourceRef{Name: "redis", Type: "infra.cache", ProviderID: "cache-1"},
+			want: "team-cache",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			owner, err := p.GetOwner(context.Background(), tt.ref)
+			if err != nil {
+				t.Fatalf("GetOwner: %v", err)
+			}
+			if owner.Owner != tt.want {
+				t.Fatalf("owner = %q, want %q", owner.Owner, tt.want)
+			}
+			if owner.Source != "tag:workflow-owner" {
+				t.Fatalf("source = %q, want tag:workflow-owner", owner.Source)
+			}
+		})
+	}
+}
+
+func TestDOProvider_OwnershipRequiresInitializedClient(t *testing.T) {
+	p := NewDOProvider()
+	ref := interfaces.ResourceRef{Name: "app", Type: "infra.droplet", ProviderID: "1001"}
+
+	_, err := p.GetOwner(context.Background(), ref)
+	if err == nil || !strings.Contains(err.Error(), "call Initialize first") {
+		t.Fatalf("GetOwner error = %v, want Initialize hint", err)
+	}
+	err = p.SetOwner(context.Background(), ref, "team-a")
+	if err == nil || !strings.Contains(err.Error(), "call Initialize first") {
+		t.Fatalf("SetOwner error = %v, want Initialize hint", err)
+	}
+	_, err = p.ListOwners(context.Background(), interfaces.OwnerFilter{Owner: "team-a"})
+	if err == nil || !strings.Contains(err.Error(), "call Initialize first") {
+		t.Fatalf("ListOwners error = %v, want Initialize hint", err)
+	}
+}
+
+func TestDOProvider_SetOwnerCreatesTagAndReplacesOldOwnerTag(t *testing.T) {
+	api := &ownershipFakeAPI{
+		droplets: map[string]ownershipResource{
+			"1001": {id: "1001", name: "app", tags: []string{"workflow-owner:team-old"}},
+		},
+	}
+	p, _ := newProviderForOwnershipTest(t, api)
+
+	err := p.SetOwner(context.Background(), interfaces.ResourceRef{
+		Name:       "app",
+		Type:       "infra.droplet",
+		ProviderID: "1001",
+	}, "team-a")
+	if err != nil {
+		t.Fatalf("SetOwner: %v", err)
+	}
+
+	if !api.createdTags["workflow-owner:team-a"] {
+		t.Fatalf("expected workflow-owner:team-a tag to be created")
+	}
+	if !api.tagged["workflow-owner:team-a|droplet|1001"] {
+		t.Fatalf("expected droplet 1001 to be tagged with workflow-owner:team-a; got %+v", api.tagged)
+	}
+	if !api.untagged["workflow-owner:team-old|droplet|1001"] {
+		t.Fatalf("expected old owner tag to be removed; got %+v", api.untagged)
+	}
+}
+
+func TestDOProvider_ListOwnersByOwnerUsesTaggedResourceEnumeration(t *testing.T) {
+	mock := &enumeratorFakeAPI{
+		tagExists: map[string]bool{"workflow-owner:team-a": true},
+		dropletsByTag: map[string][]godo.Droplet{
+			"workflow-owner:team-a": {
+				{ID: 1001, Name: "app", Tags: []string{"workflow-owner:team-a"}},
+			},
+		},
+		volumes: []godo.Volume{
+			{ID: "vol-1", Name: "data", Tags: []string{"workflow-owner:team-a"}},
+		},
+		databases: []godo.Database{
+			{ID: "db-1", Name: "pg", EngineSlug: "pg", Tags: []string{"workflow-owner:team-a"}},
+		},
+	}
+	p, _ := newProviderForEnumeratorTest(t, mock)
+
+	owners, err := p.ListOwners(context.Background(), interfaces.OwnerFilter{Owner: "team-a"})
+	if err != nil {
+		t.Fatalf("ListOwners: %v", err)
+	}
+	if len(owners) != 3 {
+		t.Fatalf("owners len = %d, want 3: %+v", len(owners), owners)
+	}
+	for _, owner := range owners {
+		if owner.Owner != "team-a" {
+			t.Fatalf("owner = %q, want team-a in %+v", owner.Owner, owner)
+		}
+		if owner.Source != "tag:workflow-owner" {
+			t.Fatalf("source = %q, want tag:workflow-owner", owner.Source)
+		}
+	}
+}
+
+func TestDOProvider_ListOwnersByResourceTypeOnlyEnumeratesThatType(t *testing.T) {
+	mock := &enumeratorFakeAPI{
+		tagExists: map[string]bool{"workflow-owner:team-a": true},
+		dropletsByTag: map[string][]godo.Droplet{
+			"workflow-owner:team-a": {
+				{ID: 1001, Name: "app", Tags: []string{"workflow-owner:team-a"}},
+			},
+		},
+		volumes: []godo.Volume{
+			{ID: "vol-1", Name: "data", Tags: []string{"workflow-owner:team-a"}},
+		},
+		databases: []godo.Database{
+			{ID: "db-1", Name: "pg", EngineSlug: "pg", Tags: []string{"workflow-owner:team-a"}},
+			{ID: "cache-1", Name: "redis", EngineSlug: "redis", Tags: []string{"workflow-owner:team-a"}},
+		},
+	}
+	p, _ := newProviderForEnumeratorTest(t, mock)
+
+	owners, err := p.ListOwners(context.Background(), interfaces.OwnerFilter{
+		Owner:        "team-a",
+		ResourceType: "infra.volume",
+	})
+	if err != nil {
+		t.Fatalf("ListOwners: %v", err)
+	}
+	if len(owners) != 1 {
+		t.Fatalf("owners len = %d, want 1: %+v", len(owners), owners)
+	}
+	if owners[0].Ref.Type != "infra.volume" || owners[0].Ref.ProviderID != "vol-1" {
+		t.Fatalf("owner ref = %+v, want volume vol-1", owners[0].Ref)
+	}
+	if mock.volumeLists != 1 {
+		t.Fatalf("volume list calls = %d, want 1", mock.volumeLists)
+	}
+	if mock.dropletLists != 0 || mock.databaseLists != 0 {
+		t.Fatalf("unexpected unrelated list calls: droplets=%d databases=%d", mock.dropletLists, mock.databaseLists)
+	}
+}
+
+type ownershipResource struct {
+	id     string
+	name   string
+	engine string
+	tags   []string
+}
+
+type ownershipFakeAPI struct {
+	droplets    map[string]ownershipResource
+	volumes     map[string]ownershipResource
+	databases   map[string]ownershipResource
+	createdTags map[string]bool
+	tagged      map[string]bool
+	untagged    map[string]bool
+}
+
+func (f *ownershipFakeAPI) handler(t *testing.T) http.HandlerFunc {
+	t.Helper()
+	if f.createdTags == nil {
+		f.createdTags = map[string]bool{}
+	}
+	if f.tagged == nil {
+		f.tagged = map[string]bool{}
+	}
+	if f.untagged == nil {
+		f.untagged = map[string]bool{}
+	}
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		switch {
+		case strings.HasPrefix(r.URL.Path, "/v2/droplets/") && r.Method == http.MethodGet:
+			id := strings.TrimPrefix(r.URL.Path, "/v2/droplets/")
+			writeOwnershipResource(t, w, "droplet", f.droplets[id])
+			return
+		case strings.HasPrefix(r.URL.Path, "/v2/volumes/") && r.Method == http.MethodGet:
+			id := strings.TrimPrefix(r.URL.Path, "/v2/volumes/")
+			writeOwnershipResource(t, w, "volume", f.volumes[id])
+			return
+		case strings.HasPrefix(r.URL.Path, "/v2/databases/") && r.Method == http.MethodGet:
+			id := strings.TrimPrefix(r.URL.Path, "/v2/databases/")
+			writeOwnershipResource(t, w, "database", f.databases[id])
+			return
+		case r.URL.Path == "/v2/tags" && r.Method == http.MethodPost:
+			var req godo.TagCreateRequest
+			if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+				t.Fatalf("decode create tag: %v", err)
+			}
+			f.createdTags[req.Name] = true
+			_, _ = fmt.Fprintf(w, `{"tag":{"name":%q}}`, req.Name)
+			return
+		case strings.HasPrefix(r.URL.Path, "/v2/tags/") && strings.HasSuffix(r.URL.Path, "/resources"):
+			tag := strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/v2/tags/"), "/resources")
+			var req godo.TagResourcesRequest
+			if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+				t.Fatalf("decode tag resources: %v", err)
+			}
+			for _, resource := range req.Resources {
+				key := fmt.Sprintf("%s|%s|%s", tag, resource.Type, resource.ID)
+				switch r.Method {
+				case http.MethodPost:
+					f.tagged[key] = true
+				case http.MethodDelete:
+					f.untagged[key] = true
+				default:
+					t.Fatalf("unexpected tag resources method %s", r.Method)
+				}
+			}
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+		w.WriteHeader(http.StatusNotImplemented)
+		_, _ = fmt.Fprintf(w, `{"id":"not_implemented","message":"%s %s not handled"}`, r.Method, r.URL.Path)
+	}
+}
+
+func writeOwnershipResource(t *testing.T, w http.ResponseWriter, kind string, resource ownershipResource) {
+	t.Helper()
+	if resource.id == "" {
+		w.WriteHeader(http.StatusNotFound)
+		_, _ = w.Write([]byte(`{"id":"not_found","message":"missing"}`))
+		return
+	}
+	tags := jsonStringArray(resource.tags)
+	switch kind {
+	case "droplet":
+		_, _ = fmt.Fprintf(w, `{"droplet":{"id":%s,"name":%q,"status":"active","size":{"slug":"s-1vcpu-1gb"},"region":{"slug":"nyc3"},"tags":%s}}`, resource.id, resource.name, tags)
+	case "volume":
+		_, _ = fmt.Fprintf(w, `{"volume":{"id":%q,"name":%q,"tags":%s}}`, resource.id, resource.name, tags)
+	case "database":
+		engine := resource.engine
+		if engine == "" {
+			engine = "pg"
+		}
+		_, _ = fmt.Fprintf(w, `{"database":{"id":%q,"name":%q,"engine":%q,"tags":%s}}`, resource.id, resource.name, engine, tags)
+	}
+}
+
+func newProviderForOwnershipTest(t *testing.T, api *ownershipFakeAPI) (*DOProvider, *httptest.Server) {
+	t.Helper()
+	srv := httptest.NewServer(api.handler(t))
+	t.Cleanup(srv.Close)
+
+	client := godo.NewClient(srv.Client())
+	base, err := url.Parse(srv.URL + "/")
+	if err != nil {
+		t.Fatalf("parse httptest URL: %v", err)
+	}
+	client.BaseURL = base
+	return &DOProvider{client: client, region: "nyc3", drivers: map[string]interfaces.ResourceDriver{
+		"infra.droplet":  drivers.NewDropletDriver(client, "nyc3"),
+		"infra.volume":   drivers.NewVolumeDriver(client, "nyc3"),
+		"infra.database": drivers.NewDatabaseDriver(client, "nyc3"),
+		"infra.cache":    drivers.NewCacheDriver(client, "nyc3"),
+	}}, srv
+}
diff --git a/internal/provider_enumerator_test.go b/internal/provider_enumerator_test.go
index f182ea0..4c6c0d5 100644
--- a/internal/provider_enumerator_test.go
+++ b/internal/provider_enumerator_test.go
@@ -49,6 +49,9 @@ type enumeratorFakeAPI struct {
 	dropletsByTag map[string][]godo.Droplet
 	volumes       []godo.Volume
 	databases     []godo.Database
+	dropletLists  int
+	volumeLists   int
+	databaseLists int
 	// tagExists maps tag-name → whether GET /v2/tags/{name} returns 200.
 	// When the tag does not exist, the API returns 404; EnumerateByTag must
 	// treat 404 as "no resources" (empty result) not an error.
@@ -77,16 +80,19 @@ func (f *enumeratorFakeAPI) handler(t *testing.T) http.HandlerFunc {
 			return
 
 		case r.URL.Path == "/v2/droplets" && r.Method == http.MethodGet:
+			f.dropletLists++
 			tag := r.URL.Query().Get("tag_name")
 			droplets := f.dropletsByTag[tag]
 			writeDroplets(t, w, droplets)
 			return
 
 		case r.URL.Path == "/v2/volumes" && r.Method == http.MethodGet:
+			f.volumeLists++
 			writeVolumes(t, w, f.volumes)
 			return
 
 		case r.URL.Path == "/v2/databases" && r.Method == http.MethodGet:
+			f.databaseLists++
 			writeDatabases(t, w, f.databases)
 			return
 		}
diff --git a/plugin.json b/plugin.json
index dfc32d4..58ec347 100644
--- a/plugin.json
+++ b/plugin.json
@@ -6,7 +6,7 @@
   "license": "MIT",
   "type": "external",
   "tier": "community",
-  "minEngineVersion": "0.68.2",
+  "minEngineVersion": "0.69.1",
   "required_secrets": [
     {
       "name": "DIGITALOCEAN_TOKEN",
@@ -26,6 +26,7 @@
     "workflow.plugin.external.iac.IaCProviderLogCapture",
     "workflow.plugin.external.iac.IaCProviderRequirementMapper",
     "workflow.plugin.external.iac.IaCProviderRegionLister",
+    "workflow.plugin.external.iac.IaCProviderOwnership",
     "workflow.plugin.external.iac.IaCProviderFinalizer",
     "workflow.plugin.external.iac.ResourceDriver",
     "workflow.plugin.external.iac.IaCStateBackend"