diff --git a/extensions/common/metadata_object.cc b/extensions/common/metadata_object.cc
index 08c02b76fdd..e3067fa6b31 100644
--- a/extensions/common/metadata_object.cc
+++ b/extensions/common/metadata_object.cc
@@ -24,7 +24,10 @@ namespace Istio {
 namespace Common {
 
 namespace {
-static absl::flat_hash_map<absl::string_view, BaggageToken> ALL_BAGGAGE_TOKENS = {
+
+// This maps field names into baggage tokens. We use it to decode field names
+// when WorkloadMetadataObject content is accessed through the Envoy API.
+static absl::flat_hash_map<absl::string_view, BaggageToken> ALL_METADATA_FIELDS = {
     {NamespaceNameToken, BaggageToken::NamespaceName},
     {ClusterNameToken, BaggageToken::ClusterName},
     {ServiceNameToken, BaggageToken::ServiceName},
@@ -36,6 +39,22 @@ static absl::flat_hash_map<absl::string_view, BaggageToken> ALL_BAGGAGE_TOKENS =
     {InstanceNameToken, BaggageToken::InstanceName},
 };
 
+// This maps baggage keys into baggage tokens. We use it to decode baggage keys
+// coming over the wire when building WorkloadMetadataObject.
+static absl::flat_hash_map<absl::string_view, BaggageToken> ALL_BAGGAGE_TOKENS = {
+    {NamespaceNameBaggageToken, BaggageToken::NamespaceName},
+    {ClusterNameBaggageToken, BaggageToken::ClusterName},
+    {ServiceNameBaggageToken, BaggageToken::ServiceName},
+    {ServiceVersionBaggageToken, BaggageToken::ServiceVersion},
+    {AppNameBaggageToken, BaggageToken::AppName},
+    {AppVersionBaggageToken, BaggageToken::AppVersion},
+    {DeploymentNameBaggageToken, BaggageToken::WorkloadName},
+    {PodNameBaggageToken, BaggageToken::WorkloadName},
+    {CronjobNameBaggageToken, BaggageToken::WorkloadName},
+    {JobNameBaggageToken, BaggageToken::WorkloadName},
+    {InstanceNameBaggageToken, BaggageToken::InstanceName},
+};
+
 static absl::flat_hash_map<absl::string_view, WorkloadType> ALL_WORKLOAD_TOKENS = {
     {PodSuffix, WorkloadType::Pod},
     {DeploymentSuffix, WorkloadType::Deployment},
@@ -69,13 +88,13 @@ std::string WorkloadMetadataObject::baggage() const {
   }
   // Map the workload metadata fields to baggage tokens
   const std::vector<std::pair<absl::string_view, absl::string_view>> field_to_baggage = {
-      {Istio::Common::NamespaceNameToken, "k8s.namespace.name"},
-      {Istio::Common::ClusterNameToken, "k8s.cluster.name"},
-      {Istio::Common::ServiceNameToken, "service.name"},
-      {Istio::Common::ServiceVersionToken, "service.version"},
-      {Istio::Common::AppNameToken, "app.name"},
-      {Istio::Common::AppVersionToken, "app.version"},
-      {Istio::Common::InstanceNameToken, "k8s.instance.name"},
+      {Istio::Common::NamespaceNameToken, Istio::Common::NamespaceNameBaggageToken},
+      {Istio::Common::ClusterNameToken, Istio::Common::ClusterNameBaggageToken},
+      {Istio::Common::ServiceNameToken, Istio::Common::ServiceNameBaggageToken},
+      {Istio::Common::ServiceVersionToken, Istio::Common::ServiceVersionBaggageToken},
+      {Istio::Common::AppNameToken, Istio::Common::AppNameBaggageToken},
+      {Istio::Common::AppVersionToken, Istio::Common::AppVersionBaggageToken},
+      {Istio::Common::InstanceNameToken, Istio::Common::InstanceNameBaggageToken},
   };
 
   for (const auto& [field_name, baggage_key] : field_to_baggage) {
@@ -320,8 +339,8 @@ std::string serializeToStringDeterministic(const google::protobuf::Struct& metad
 
 WorkloadMetadataObject::FieldType
 WorkloadMetadataObject::getField(absl::string_view field_name) const {
-  const auto it = ALL_BAGGAGE_TOKENS.find(field_name);
-  if (it != ALL_BAGGAGE_TOKENS.end()) {
+  const auto it = ALL_METADATA_FIELDS.find(field_name);
+  if (it != ALL_METADATA_FIELDS.end()) {
     switch (it->second) {
     case BaggageToken::NamespaceName:
       return namespace_name_;
@@ -389,15 +408,19 @@ convertBaggageToWorkloadMetadata(absl::string_view data, absl::string_view ident
       case BaggageToken::AppVersion:
         app_version = parts.second;
         break;
-      case BaggageToken::WorkloadName:
+      case BaggageToken::WorkloadName: {
         workload = parts.second;
+        std::vector<absl::string_view> splitWorkloadKey = absl::StrSplit(parts.first, ".");
+        if (splitWorkloadKey.size() >= 2 && splitWorkloadKey[0] == "k8s") {
+          workload_type = fromSuffix(splitWorkloadKey[1]);
+        }
         break;
-      case BaggageToken::WorkloadType:
-        workload_type = fromSuffix(parts.second);
-        break;
+      }
       case BaggageToken::InstanceName:
         instance = parts.second;
         break;
+      default:
+        break;
       }
     }
   }
diff --git a/extensions/common/metadata_object.h b/extensions/common/metadata_object.h
index 18b3301bd1d..7253fdd539a 100644
--- a/extensions/common/metadata_object.h
+++ b/extensions/common/metadata_object.h
@@ -67,6 +67,7 @@ enum class BaggageToken {
   InstanceName,
 };
 
+// Field names accessible from WorkloadMetadataObject.
 constexpr absl::string_view NamespaceNameToken = "namespace";
 constexpr absl::string_view ClusterNameToken = "cluster";
 constexpr absl::string_view ServiceNameToken = "service";
@@ -79,6 +80,20 @@ constexpr absl::string_view InstanceNameToken = "name";
 constexpr absl::string_view LabelsToken = "labels";
 constexpr absl::string_view IdentityToken = "identity";
 
+// Field names used to translate baggage content into
+// WorkloadMetadataObject information.
+constexpr absl::string_view NamespaceNameBaggageToken = "k8s.namespace.name";
+constexpr absl::string_view ClusterNameBaggageToken = "k8s.cluster.name";
+constexpr absl::string_view ServiceNameBaggageToken = "service.name";
+constexpr absl::string_view ServiceVersionBaggageToken = "service.version";
+constexpr absl::string_view AppNameBaggageToken = "app.name";
+constexpr absl::string_view AppVersionBaggageToken = "app.version";
+constexpr absl::string_view DeploymentNameBaggageToken = "k8s.deployment.name";
+constexpr absl::string_view PodNameBaggageToken = "k8s.pod.name";
+constexpr absl::string_view CronjobNameBaggageToken = "k8s.cronjob.name";
+constexpr absl::string_view JobNameBaggageToken = "k8s.job.name";
+constexpr absl::string_view InstanceNameBaggageToken = "k8s.instance.name";
+
 constexpr absl::string_view InstanceMetadataField = "NAME";
 constexpr absl::string_view NamespaceMetadataField = "NAMESPACE";
 constexpr absl::string_view ClusterMetadataField = "CLUSTER_ID";
diff --git a/extensions/common/metadata_object_test.cc b/extensions/common/metadata_object_test.cc
index 90003b95570..188defa777f 100644
--- a/extensions/common/metadata_object_test.cc
+++ b/extensions/common/metadata_object_test.cc
@@ -82,8 +82,9 @@ TEST(WorkloadMetadataObjectTest, ConversionWithLabels) {
 TEST(WorkloadMetadataObjectTest, Conversion) {
   {
     const auto r = convertBaggageToWorkloadMetadata(
-        "type=deployment,workload=foo,cluster=my-cluster,"
-        "namespace=default,service=foo-service,revision=v1alpha3,app=foo-app,version=latest");
+        "k8s.deployment.name=foo,k8s.cluster.name=my-cluster,"
+        "k8s.namespace.name=default,service.name=foo-service,service.version=v1alpha3,app.name=foo-"
+        "app,app.version=latest");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("service")), "foo-service");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("revision")), "v1alpha3");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("type")), DeploymentSuffix);
@@ -97,14 +98,14 @@ TEST(WorkloadMetadataObjectTest, Conversion) {
   }
 
   {
-    const auto r =
-        convertBaggageToWorkloadMetadata("type=pod,name=foo-pod-435,cluster=my-cluster,namespace="
-                                         "test,service=foo-service,revision=v1beta2");
+    const auto r = convertBaggageToWorkloadMetadata(
+        "k8s.pod.name=foo-pod-435,k8s.cluster.name=my-cluster,k8s.namespace.name="
+        "test,k8s.instance.name=foo-instance-435,service.name=foo-service,service.version=v1beta2");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("service")), "foo-service");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("revision")), "v1beta2");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("type")), PodSuffix);
-    EXPECT_EQ(absl::get<absl::string_view>(r->getField("workload")), "");
-    EXPECT_EQ(absl::get<absl::string_view>(r->getField("name")), "foo-pod-435");
+    EXPECT_EQ(absl::get<absl::string_view>(r->getField("workload")), "foo-pod-435");
+    EXPECT_EQ(absl::get<absl::string_view>(r->getField("name")), "foo-instance-435");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("namespace")), "test");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("cluster")), "my-cluster");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("app")), "");
@@ -113,23 +114,23 @@ TEST(WorkloadMetadataObjectTest, Conversion) {
   }
 
   {
-    const auto r =
-        convertBaggageToWorkloadMetadata("type=job,name=foo-job-435,cluster=my-cluster,namespace="
-                                         "test,service=foo-service,revision=v1beta4");
+    const auto r = convertBaggageToWorkloadMetadata(
+        "k8s.job.name=foo-job-435,k8s.cluster.name=my-cluster,k8s.namespace.name="
+        "test,k8s.instance.name=foo-instance-435,service.name=foo-service,service.version=v1beta4");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("service")), "foo-service");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("revision")), "v1beta4");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("type")), JobSuffix);
-    EXPECT_EQ(absl::get<absl::string_view>(r->getField("workload")), "");
-    EXPECT_EQ(absl::get<absl::string_view>(r->getField("name")), "foo-job-435");
+    EXPECT_EQ(absl::get<absl::string_view>(r->getField("workload")), "foo-job-435");
+    EXPECT_EQ(absl::get<absl::string_view>(r->getField("name")), "foo-instance-435");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("namespace")), "test");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("cluster")), "my-cluster");
     checkStructConversion(*r);
   }
 
   {
-    const auto r =
-        convertBaggageToWorkloadMetadata("type=cronjob,workload=foo-cronjob,cluster=my-cluster,"
-                                         "namespace=test,service=foo-service,revision=v1beta4");
+    const auto r = convertBaggageToWorkloadMetadata(
+        "k8s.cronjob.name=foo-cronjob,k8s.cluster.name=my-cluster,"
+        "k8s.namespace.name=test,service.name=foo-service,service.version=v1beta4");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("service")), "foo-service");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("revision")), "v1beta4");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("type")), CronJobSuffix);
@@ -141,8 +142,9 @@ TEST(WorkloadMetadataObjectTest, Conversion) {
   }
 
   {
-    const auto r = convertBaggageToWorkloadMetadata(
-        "type=deployment,workload=foo,namespace=default,service=foo-service,revision=v1alpha3");
+    const auto r =
+        convertBaggageToWorkloadMetadata("k8s.deployment.name=foo,k8s.namespace.name=default,"
+                                         "service.name=foo-service,service.version=v1alpha3");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("service")), "foo-service");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("revision")), "v1alpha3");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("type")), DeploymentSuffix);
@@ -153,7 +155,7 @@ TEST(WorkloadMetadataObjectTest, Conversion) {
   }
 
   {
-    const auto r = convertBaggageToWorkloadMetadata("namespace=default");
+    const auto r = convertBaggageToWorkloadMetadata("k8s.namespace.name=default");
     EXPECT_EQ(absl::get<absl::string_view>(r->getField("namespace")), "default");
     checkStructConversion(*r);
   }
diff --git a/source/extensions/filters/http/peer_metadata/filter.cc b/source/extensions/filters/http/peer_metadata/filter.cc
index d226d508b42..80981b9bbee 100644
--- a/source/extensions/filters/http/peer_metadata/filter.cc
+++ b/source/extensions/filters/http/peer_metadata/filter.cc
@@ -238,6 +238,23 @@ void BaggagePropagationMethod::inject(const StreamInfo::StreamInfo&, Http::Heade
   headers.setReference(Headers::get().Baggage, value_);
 }
 
+BaggageDiscoveryMethod::BaggageDiscoveryMethod() {}
+
+absl::optional<PeerInfo> BaggageDiscoveryMethod::derivePeerInfo(const StreamInfo::StreamInfo&,
+                                                                Http::HeaderMap& headers,
+                                                                Context&) const {
+  const auto baggage_header = headers.get(Headers::get().Baggage);
+  if (baggage_header.empty()) {
+    return {};
+  }
+  const auto baggage_value = baggage_header[0]->value().getStringView();
+  const auto workload = Istio::Common::convertBaggageToWorkloadMetadata(baggage_value);
+  if (workload) {
+    return *workload;
+  }
+  return {};
+}
+
 FilterConfig::FilterConfig(const io::istio::http::peer_metadata::Config& config,
                            Server::Configuration::FactoryContext& factory_context)
     : shared_with_upstream_(config.shared_with_upstream()),
@@ -273,6 +290,13 @@ std::vector<DiscoveryMethodPtr> FilterConfig::buildDiscoveryMethods(
       methods.push_back(std::make_unique<MXMethod>(downstream, additional_labels,
                                                    factory_context.serverFactoryContext()));
       break;
+    case io::istio::http::peer_metadata::Config::DiscoveryMethod::MethodSpecifierCase::kBaggage:
+      if (downstream) {
+        methods.push_back(std::make_unique<BaggageDiscoveryMethod>());
+      } else {
+        ENVOY_LOG(warn, "BaggageDiscovery peer metadata discovery option is only available for "
+                        "downstream peer discovery");
+      }
     case io::istio::http::peer_metadata::Config::DiscoveryMethod::MethodSpecifierCase::
         kUpstreamFilterState:
       if (!downstream) {
diff --git a/source/extensions/filters/http/peer_metadata/filter.h b/source/extensions/filters/http/peer_metadata/filter.h
index 4434d167bf9..33c470826f4 100644
--- a/source/extensions/filters/http/peer_metadata/filter.h
+++ b/source/extensions/filters/http/peer_metadata/filter.h
@@ -112,6 +112,13 @@ class BaggagePropagationMethod : public PropagationMethod {
   const std::string value_;
 };
 
+class BaggageDiscoveryMethod : public DiscoveryMethod, public Logger::Loggable<Logger::Id::filter> {
+public:
+  BaggageDiscoveryMethod();
+  absl::optional<PeerInfo> derivePeerInfo(const StreamInfo::StreamInfo&, Http::HeaderMap&,
+                                          Context&) const override;
+};
+
 class FilterConfig : public Logger::Loggable<Logger::Id::filter> {
 public:
   FilterConfig(const io::istio::http::peer_metadata::Config&,
@@ -155,7 +162,7 @@ class FilterConfig : public Logger::Loggable<Logger::Id::filter> {
 
 using FilterConfigSharedPtr = std::shared_ptr<FilterConfig>;
 
-class Filter : public Http::PassThroughFilter {
+class Filter : public Http::PassThroughFilter, public Logger::Loggable<Logger::Id::filter> {
 public:
   Filter(const FilterConfigSharedPtr& config) : config_(config) {}
   Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap&, bool) override;
diff --git a/source/extensions/filters/http/peer_metadata/filter_test.cc b/source/extensions/filters/http/peer_metadata/filter_test.cc
index 57ead416e0e..7824c9a3d9b 100644
--- a/source/extensions/filters/http/peer_metadata/filter_test.cc
+++ b/source/extensions/filters/http/peer_metadata/filter_test.cc
@@ -718,6 +718,342 @@ TEST_F(PeerMetadataTest, BaggagePropagationWithMixedConfig) {
   EXPECT_TRUE(response_headers_.has(Headers::get().Baggage));
 }
 
+// Baggage Discovery Tests
+
+TEST_F(PeerMetadataTest, DownstreamBaggageDiscoveryEmpty) {
+  initialize(R"EOF(
+    downstream_discovery:
+      - baggage: {}
+  )EOF");
+  EXPECT_EQ(0, request_headers_.size());
+  EXPECT_EQ(0, response_headers_.size());
+  checkNoPeer(true);
+  checkNoPeer(false);
+}
+
+TEST_F(PeerMetadataTest, UpstreamBaggageDiscoveryEmpty) {
+  // The baggage discovery filter should only be used for downstream
+  // peer metadata detection.
+  initialize(R"EOF(
+    upstream_discovery:
+      - baggage: {}
+  )EOF");
+  EXPECT_EQ(0, request_headers_.size());
+  EXPECT_EQ(0, response_headers_.size());
+  checkNoPeer(true);
+  checkNoPeer(false);
+}
+
+TEST_F(PeerMetadataTest, DownstreamBaggageDiscovery) {
+  request_headers_.setReference(
+      Headers::get().Baggage,
+      "k8s.namespace.name=test-namespace,k8s.cluster.name=test-cluster,"
+      "service.name=test-service,service.version=v1,k8s.deployment.name=test-workload,"
+      "k8s.workload.type=deployment,k8s.instance.name=test-instance-123,"
+      "app.name=test-app,app.version=v2.0");
+  initialize(R"EOF(
+    downstream_discovery:
+      - baggage: {}
+  )EOF");
+  EXPECT_EQ(1, request_headers_.size());
+  EXPECT_EQ(0, response_headers_.size());
+  checkPeerNamespace(true, "test-namespace");
+  checkNoPeer(false);
+  checkShared(false);
+}
+
+TEST_F(PeerMetadataTest, UpstreamBaggageDiscovery) {
+  response_headers_.setReference(
+      Headers::get().Baggage,
+      "k8s.namespace.name=upstream-namespace,k8s.cluster.name=upstream-cluster,"
+      "service.name=upstream-service,service.version=v2,k8s.workload.name=upstream-workload,"
+      "k8s.workload.type=pod,k8s.instance.name=upstream-instance-456,"
+      "app.name=upstream-app,app.version=v3.0");
+  initialize(R"EOF(
+    upstream_discovery:
+      - baggage: {}
+  )EOF");
+  EXPECT_EQ(0, request_headers_.size());
+  EXPECT_EQ(1, response_headers_.size());
+  checkNoPeer(true);
+  // Baggage discovery should ignore upstream.
+  checkNoPeer(false);
+}
+
+TEST_F(PeerMetadataTest, BothDirectionsBaggageDiscovery) {
+  request_headers_.setReference(Headers::get().Baggage,
+                                "k8s.namespace.name=downstream-ns,service.name=downstream-svc");
+  response_headers_.setReference(Headers::get().Baggage,
+                                 "k8s.namespace.name=upstream-ns,service.name=upstream-svc");
+  initialize(R"EOF(
+    downstream_discovery:
+      - baggage: {}
+    upstream_discovery:
+      - baggage: {}
+  )EOF");
+  EXPECT_EQ(1, request_headers_.size());
+  EXPECT_EQ(1, response_headers_.size());
+  checkPeerNamespace(true, "downstream-ns");
+  // Baggage discovery should ignore upstream
+  checkNoPeer(false);
+}
+
+TEST_F(PeerMetadataTest, DownstreamBaggageFallbackFirst) {
+  // Baggage is present, so XDS should not be called
+  request_headers_.setReference(
+      Headers::get().Baggage, "k8s.namespace.name=baggage-namespace,service.name=baggage-service");
+  EXPECT_CALL(*metadata_provider_, GetMetadata(_)).Times(0);
+  initialize(R"EOF(
+    downstream_discovery:
+      - baggage: {}
+      - workload_discovery: {}
+  )EOF");
+  EXPECT_EQ(1, request_headers_.size());
+  EXPECT_EQ(0, response_headers_.size());
+  checkPeerNamespace(true, "baggage-namespace");
+  checkNoPeer(false);
+}
+
+TEST_F(PeerMetadataTest, DownstreamBaggageFallbackSecond) {
+  // No baggage header, so XDS should be called as fallback
+  const WorkloadMetadataObject pod("pod-foo-1234", "my-cluster", "xds-namespace", "foo",
+                                   "foo-service", "v1alpha3", "", "",
+                                   Istio::Common::WorkloadType::Pod, "");
+  EXPECT_CALL(*metadata_provider_, GetMetadata(_))
+      .WillRepeatedly(Invoke([&](const Network::Address::InstanceConstSharedPtr& address)
+                                 -> std::optional<WorkloadMetadataObject> {
+        if (absl::StartsWith(address->asStringView(), "127.0.0.1")) {
+          return {pod};
+        }
+        return {};
+      }));
+  initialize(R"EOF(
+    downstream_discovery:
+      - baggage: {}
+      - workload_discovery: {}
+  )EOF");
+  EXPECT_EQ(0, request_headers_.size());
+  EXPECT_EQ(0, response_headers_.size());
+  checkPeerNamespace(true, "xds-namespace");
+  checkNoPeer(false);
+}
+
+TEST_F(PeerMetadataTest, UpstreamBaggageFallbackFirst) {
+  // Baggage is present, but ignored as it's coming from upstream.
+  response_headers_.setReference(
+      Headers::get().Baggage,
+      "k8s.namespace.name=baggage-upstream,service.name=baggage-upstream-service");
+  // WDS information is also present, and this is the one tha tshould be used.
+  const WorkloadMetadataObject pod("pod-foo-1234", "my-cluster", "xds-upstream", "foo",
+                                   "foo-service", "v1alpha3", "", "",
+                                   Istio::Common::WorkloadType::Pod, "");
+  EXPECT_CALL(*metadata_provider_, GetMetadata(_))
+      .WillRepeatedly(Invoke([&](const Network::Address::InstanceConstSharedPtr& address)
+                                 -> std::optional<WorkloadMetadataObject> {
+        if (absl::StartsWith(address->asStringView(), "10.0.0.1")) {
+          return {pod};
+        }
+        return {};
+      }));
+  initialize(R"EOF(
+    upstream_discovery:
+      - baggage: {}
+      - workload_discovery: {}
+  )EOF");
+  EXPECT_EQ(0, request_headers_.size());
+  EXPECT_EQ(1, response_headers_.size());
+  checkNoPeer(true);
+  checkPeerNamespace(false, "xds-upstream");
+}
+
+TEST_F(PeerMetadataTest, UpstreamBaggageFallbackSecond) {
+  // No baggage header, baggage is ignored as it's coming from upstream,
+  // but workload discovery should pick up the details.
+  const WorkloadMetadataObject pod("pod-foo-1234", "my-cluster", "xds-upstream", "foo",
+                                   "foo-service", "v1alpha3", "", "",
+                                   Istio::Common::WorkloadType::Pod, "");
+  EXPECT_CALL(*metadata_provider_, GetMetadata(_))
+      .WillRepeatedly(Invoke([&](const Network::Address::InstanceConstSharedPtr& address)
+                                 -> std::optional<WorkloadMetadataObject> {
+        if (absl::StartsWith(address->asStringView(), "10.0.0.1")) {
+          return {pod};
+        }
+        return {};
+      }));
+  initialize(R"EOF(
+    upstream_discovery:
+      - baggage: {}
+      - workload_discovery: {}
+  )EOF");
+  EXPECT_EQ(0, request_headers_.size());
+  EXPECT_EQ(0, response_headers_.size());
+  checkNoPeer(true);
+  checkPeerNamespace(false, "xds-upstream");
+}
+
+TEST_F(PeerMetadataTest, DownstreamBaggageWithMXFallback) {
+  // Baggage is present, so MX should not be used
+  request_headers_.setReference(Headers::get().Baggage,
+                                "k8s.namespace.name=baggage-ns,service.name=baggage-svc");
+  request_headers_.setReference(Headers::get().ExchangeMetadataHeaderId, "test-pod");
+  request_headers_.setReference(Headers::get().ExchangeMetadataHeader, SampleIstioHeader);
+  initialize(R"EOF(
+    downstream_discovery:
+      - baggage: {}
+      - istio_headers: {}
+  )EOF");
+  EXPECT_EQ(1, request_headers_.size());
+  EXPECT_EQ(0, response_headers_.size());
+  checkPeerNamespace(true, "baggage-ns");
+  checkNoPeer(false);
+}
+
+TEST_F(PeerMetadataTest, DownstreamMXWithBaggageFallback) {
+  // MX is first, so it should be used even if baggage is present
+  request_headers_.setReference(Headers::get().Baggage,
+                                "k8s.namespace.name=baggage-ns,service.name=baggage-svc");
+  request_headers_.setReference(Headers::get().ExchangeMetadataHeaderId, "test-pod");
+  request_headers_.setReference(Headers::get().ExchangeMetadataHeader, SampleIstioHeader);
+  initialize(R"EOF(
+    downstream_discovery:
+      - istio_headers: {}
+      - baggage: {}
+  )EOF");
+  EXPECT_EQ(1, request_headers_.size());
+  EXPECT_EQ(0, response_headers_.size());
+  // MX header has namespace "default" from SampleIstioHeader
+  checkPeerNamespace(true, "default");
+  checkNoPeer(false);
+}
+
+TEST_F(PeerMetadataTest, BaggageDiscoveryWithPropagation) {
+  request_headers_.setReference(Headers::get().Baggage,
+                                "k8s.namespace.name=discovered-ns,service.name=discovered-svc");
+  initialize(R"EOF(
+    downstream_discovery:
+      - baggage: {}
+    downstream_propagation:
+      - baggage: {}
+    upstream_propagation:
+      - baggage: {}
+  )EOF");
+  EXPECT_EQ(1, request_headers_.size());  // upstream baggage propagation
+  EXPECT_EQ(1, response_headers_.size()); // downstream baggage propagation
+  EXPECT_TRUE(request_headers_.has(Headers::get().Baggage));
+  EXPECT_TRUE(response_headers_.has(Headers::get().Baggage));
+  checkPeerNamespace(true, "discovered-ns");
+  checkNoPeer(false);
+}
+
+// Test class specifically for BaggageDiscoveryMethod unit tests
+class BaggageDiscoveryMethodTest : public testing::Test {
+protected:
+  BaggageDiscoveryMethodTest() = default;
+
+  NiceMock<Server::Configuration::MockFactoryContext> context_;
+  NiceMock<StreamInfo::MockStreamInfo> stream_info_;
+};
+
+TEST_F(BaggageDiscoveryMethodTest, DerivePeerInfoFromBaggage) {
+  BaggageDiscoveryMethod method;
+
+  Http::TestRequestHeaderMapImpl headers;
+  headers.setReference(
+      Headers::get().Baggage,
+      "k8s.namespace.name=unit-test-namespace,k8s.cluster.name=unit-test-cluster,"
+      "service.name=unit-test-service,service.version=v1.0,"
+      "k8s.deployment.name=unit-test-workload,k8s.workload.type=deployment,"
+      "k8s.instance.name=unit-test-instance,app.name=unit-test-app,app.version=v2.0");
+  Context ctx;
+
+  const auto result = method.derivePeerInfo(stream_info_, headers, ctx);
+
+  ASSERT_TRUE(result.has_value());
+  EXPECT_EQ("unit-test-namespace", result->namespace_name_);
+  EXPECT_EQ("unit-test-cluster", result->cluster_name_);
+  EXPECT_EQ("unit-test-service", result->canonical_name_);
+  EXPECT_EQ("v1.0", result->canonical_revision_);
+  EXPECT_EQ("unit-test-workload", result->workload_name_);
+  EXPECT_EQ("unit-test-instance", result->instance_name_);
+  EXPECT_EQ("unit-test-app", result->app_name_);
+  EXPECT_EQ("v2.0", result->app_version_);
+  EXPECT_EQ(Istio::Common::WorkloadType::Deployment, result->workload_type_);
+}
+
+TEST_F(BaggageDiscoveryMethodTest, DerivePeerInfoEmptyBaggage) {
+  BaggageDiscoveryMethod method;
+
+  Http::TestRequestHeaderMapImpl headers;
+  Context ctx;
+
+  const auto result = method.derivePeerInfo(stream_info_, headers, ctx);
+
+  EXPECT_FALSE(result.has_value());
+}
+
+TEST_F(BaggageDiscoveryMethodTest, DerivePeerInfoPartialBaggage) {
+  BaggageDiscoveryMethod method;
+
+  Http::TestResponseHeaderMapImpl headers;
+  headers.setReference(Headers::get().Baggage,
+                       "k8s.namespace.name=partial-ns,service.name=partial-svc");
+  Context ctx;
+
+  const auto result = method.derivePeerInfo(stream_info_, headers, ctx);
+
+  ASSERT_TRUE(result.has_value());
+  EXPECT_EQ("partial-ns", result->namespace_name_);
+  EXPECT_EQ("partial-svc", result->canonical_name_);
+  // Other fields should be empty or default
+  EXPECT_TRUE(result->cluster_name_.empty());
+  EXPECT_TRUE(result->workload_name_.empty());
+}
+
+TEST_F(BaggageDiscoveryMethodTest, DerivePeerInfoAllWorkloadTypes) {
+  BaggageDiscoveryMethod method;
+  Context ctx;
+
+  // Test Pod workload type
+  {
+    Http::TestRequestHeaderMapImpl headers;
+    headers.setReference(Headers::get().Baggage,
+                         "k8s.namespace.name=test-ns,k8s.pod.name=pod-name");
+    const auto result = method.derivePeerInfo(stream_info_, headers, ctx);
+    ASSERT_TRUE(result.has_value());
+    EXPECT_EQ(Istio::Common::WorkloadType::Pod, result->workload_type_);
+  }
+
+  // Test Deployment workload type
+  {
+    Http::TestRequestHeaderMapImpl headers;
+    headers.setReference(Headers::get().Baggage,
+                         "k8s.namespace.name=test-ns,k8s.deployment.name=deployment-name");
+    const auto result = method.derivePeerInfo(stream_info_, headers, ctx);
+    ASSERT_TRUE(result.has_value());
+    EXPECT_EQ(Istio::Common::WorkloadType::Deployment, result->workload_type_);
+  }
+
+  // Test Job workload type
+  {
+    Http::TestRequestHeaderMapImpl headers;
+    headers.setReference(Headers::get().Baggage,
+                         "k8s.namespace.name=test-ns,k8s.job.name=job-name");
+    const auto result = method.derivePeerInfo(stream_info_, headers, ctx);
+    ASSERT_TRUE(result.has_value());
+    EXPECT_EQ(Istio::Common::WorkloadType::Job, result->workload_type_);
+  }
+
+  // Test CronJob workload type
+  {
+    Http::TestRequestHeaderMapImpl headers;
+    headers.setReference(Headers::get().Baggage,
+                         "k8s.namespace.name=test-ns,k8s.cronjob.name=cronjob-name");
+    const auto result = method.derivePeerInfo(stream_info_, headers, ctx);
+    ASSERT_TRUE(result.has_value());
+    EXPECT_EQ(Istio::Common::WorkloadType::CronJob, result->workload_type_);
+  }
+}
+
 } // namespace
 } // namespace PeerMetadata
 } // namespace HttpFilters