diff --git a/Makefile b/Makefile
index a998cd2bf..0b5ba92e9 100644
--- a/Makefile
+++ b/Makefile
@@ -163,7 +163,8 @@ podmonitors.monitoring.coreos.com,$\
 apiservices.apiregistration.k8s.io,$\
 horizontalpodautoscalers.autoscaling,$\
 oidcpolicies.extensions.kuadrant.io,$\
-planpolicies.extensions.kuadrant.io
+planpolicies.extensions.kuadrant.io,$\
+pipelinepolicies.extensions.kuadrant.io
 
 clean: ## Clean all objects on cluster created by running this testsuite. Set the env variable USER to delete after someone else
 	@echo "Deleting objects for user: $(USER)"
diff --git a/config/settings.local.yaml.tpl b/config/settings.local.yaml.tpl
index 3760d9a06..2b32269a0 100644
--- a/config/settings.local.yaml.tpl
+++ b/config/settings.local.yaml.tpl
@@ -16,6 +16,8 @@
 #      password: "testPassword"
 #  spicedb:
 #    image: "SPICEDB_IMAGE"
+#  pipeline_policy_extension_service:
+#    image: "PIPELINE_POLICY_EXTENSION_SERVICE_IMAGE"
 #  auth0:
 #    client_id: "CLIENT_ID"
 #    client_secret: "CLIENT_SECRET"
diff --git a/config/settings.yaml b/config/settings.yaml
index 22e67d4c8..824c57908 100644
--- a/config/settings.yaml
+++ b/config/settings.yaml
@@ -18,6 +18,8 @@ default:
     image: "quay.io/rhn_support_azgabur/mockserver:latest"
   spicedb:
     image: "quay.io/authzed/spicedb:latest"
+  pipeline_policy_extension_service:
+    image: "quay.io/kuadrant/threat-assessment-service:latest"
   prometheus:
     project: "openshift-monitoring"
     service: "thanos-querier"
diff --git a/testsuite/kuadrant/extensions/pipeline_policy.py b/testsuite/kuadrant/extensions/pipeline_policy.py
new file mode 100644
index 000000000..f1f6e879e
--- /dev/null
+++ b/testsuite/kuadrant/extensions/pipeline_policy.py
@@ -0,0 +1,133 @@
+"""Module containing classes related to PipelinePolicy"""
+
+from __future__ import annotations
+
+from functools import cached_property
+from typing import Dict, List, Optional
+
+from testsuite.gateway import Referencable
+from testsuite.kubernetes import modify
+from testsuite.kubernetes.client import KubernetesClient
+from testsuite.kuadrant.policy import Policy
+
+
+class ActionSection:
+    """Section for request/response actions in a PipelinePolicy, mirrors ActionSpec from the Go API"""
+
+    def __init__(self, obj: "PipelinePolicy", section_name: str) -> None:
+        self.obj = obj
+        self.section_name = section_name
+
+    def modify_and_apply(self, modifier_func, retries=2, cmd_args=None):
+        """Delegates modify_and_apply to the parent PipelinePolicy"""
+
+        def _new_modifier(obj):
+            modifier_func(ActionSection(obj, self.section_name))
+
+        return self.obj.modify_and_apply(_new_modifier, retries, cmd_args)
+
+    @property
+    def committed(self):
+        """Delegates committed check to the parent PipelinePolicy"""
+        return self.obj.committed
+
+    @property
+    def section(self):
+        """Returns the action list for this section"""
+        return self.obj.model.spec.setdefault(self.section_name, [])
+
+    @modify
+    def add_grpc_method(self, method: str, var: Optional[str] = None, predicate: Optional[str] = None):
+        """Add a grpc_method action that calls an upstream"""
+        action: Dict = {"type": "grpc_method", "method": method}
+        if var is not None:
+            action["var"] = var
+        if predicate is not None:
+            action["predicate"] = predicate
+        self.section.append(action)
+
+    @modify
+    def add_deny(
+        self,
+        predicate: Optional[str] = None,
+        with_status: Optional[int] = None,
+        with_headers: Optional[str] = None,
+        with_body: Optional[str] = None,
+    ):
+        """Add a deny action"""
+        action: Dict = {"type": "deny"}
+        if predicate is not None:
+            action["predicate"] = predicate
+        if with_status is not None:
+            action["withStatus"] = with_status
+        if with_headers is not None:
+            action["withHeaders"] = with_headers
+        if with_body is not None:
+            action["withBody"] = with_body
+        self.section.append(action)
+
+    @modify
+    def add_fail(self, log_message: str, predicate: Optional[str] = None):
+        """Add a fail action"""
+        action: Dict = {"type": "fail", "logMessage": log_message}
+        if predicate is not None:
+            action["predicate"] = predicate
+        self.section.append(action)
+
+    @modify
+    def add_headers(self, headers: List[List[str]], predicate: Optional[str] = None):
+        """Add an add_headers action"""
+        action: Dict = {"type": "add_headers", "headersToAdd": str(headers)}
+        if predicate is not None:
+            action["predicate"] = predicate
+        self.section.append(action)
+
+
+class PipelinePolicy(Policy):
+    """PipelinePolicy for defining declarative action pipelines (request/response actions) on routes"""
+
+    @classmethod
+    def create_instance(
+        cls,
+        cluster: KubernetesClient,
+        name: str,
+        target: Referencable,
+        labels: Dict[str, str] = None,
+        section_name: str = None,
+    ):
+        """Creates base instance"""
+        model: Dict = {
+            "apiVersion": "extensions.kuadrant.io/v1alpha1",
+            "kind": "PipelinePolicy",
+            "metadata": {"name": name, "namespace": cluster.project, "labels": labels},
+            "spec": {
+                "targetRef": target.reference,
+            },
+        }
+        if section_name:
+            model["spec"]["targetRef"]["sectionName"] = section_name
+
+        return cls(model, context=cluster.context)
+
+    @cached_property
+    def on_http_request(self) -> ActionSection:
+        """Gives access to request actions"""
+        return ActionSection(self, "request")
+
+    @cached_property
+    def on_http_response(self) -> ActionSection:
+        """Gives access to response actions"""
+        return ActionSection(self, "response")
+
+    @modify
+    def add_action_method(self, name: str, url: str, service: str, method: str, message_template: str):
+        """Add a gRPC upstream action method definition"""
+        self.model.spec.setdefault("actionMethods", []).append(
+            {
+                "name": name,
+                "url": url,
+                "service": service,
+                "method": method,
+                "messageTemplate": message_template,
+            }
+        )
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/__init__.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/conftest.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/conftest.py
new file mode 100644
index 000000000..3db89a902
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/conftest.py
@@ -0,0 +1,30 @@
+"""Shared fixtures for PipelinePolicy testing."""
+
+import pytest
+
+from openshift_client import OpenShiftPythonException
+
+from testsuite.kuadrant.extensions.pipeline_policy import PipelinePolicy
+
+
+@pytest.fixture(scope="session", autouse=True)
+def check_pipeline_policy_crd(cluster, skip_or_fail):
+    """Skip all PipelinePolicy tests if the CRD is not installed on the cluster."""
+    try:
+        cluster.do_action("get", "crd/pipelinepolicies.extensions.kuadrant.io")
+    except OpenShiftPythonException:
+        skip_or_fail("PipelinePolicy CRD is not installed on the cluster")
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(cluster, blame, route, module_label):
+    """PipelinePolicy targeting the test HTTPRoute"""
+    return PipelinePolicy.create_instance(cluster, blame("pipeline"), route, labels={"testRun": module_label})
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit(request, pipeline_policy):
+    """Commit and wait for PipelinePolicy to be ready."""
+    request.addfinalizer(pipeline_policy.delete)
+    pipeline_policy.commit()
+    pipeline_policy.wait_for_ready()
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/__init__.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/conftest.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/conftest.py
new file mode 100644
index 000000000..afab11236
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/conftest.py
@@ -0,0 +1,61 @@
+"""Shared fixtures for PipelinePolicy gRPC tests."""
+
+import pytest
+
+from testsuite.kubernetes import Selector
+from testsuite.kubernetes.deployment import Deployment
+from testsuite.kubernetes.service import Service, ServicePort
+from testsuite.utils.constants import HTTP_API_PORT
+
+
+@pytest.fixture(scope="module")
+def threat_assessment_service(request, cluster, blame, module_label, testconfig):
+    """Deploys the ThreatAssessmentService gRPC backend"""
+    testconfig.validators.validate(only="pipeline_policy_extension_service")
+    name = blame("threat")
+    match_labels = {"app": module_label, "deployment": name}
+
+    deployment = Deployment.create_instance(
+        cluster,
+        name,
+        container_name="threat-assessment",
+        image=testconfig["pipeline_policy_extension_service"]["image"],
+        ports={"grpc": HTTP_API_PORT},
+        selector=Selector(matchLabels=match_labels),
+        labels={"app": module_label},
+        readiness_probe={"grpc": {"port": HTTP_API_PORT}, "initialDelaySeconds": 3, "periodSeconds": 5},
+    )
+    request.addfinalizer(deployment.delete)
+    deployment.commit()
+    deployment.wait_for_ready()
+
+    service = Service.create_instance(
+        cluster,
+        name,
+        selector=match_labels,
+        ports=[ServicePort(name="grpc", port=HTTP_API_PORT, targetPort="grpc")],
+        labels={"app": module_label},
+    )
+    request.addfinalizer(service.delete)
+    service.commit()
+    return service
+
+
+@pytest.fixture(scope="module")
+def threat_service_url(threat_assessment_service):
+    """gRPC URL for the threat assessment service."""
+    svc = threat_assessment_service
+    return f"grpc://{svc.name()}.{svc.namespace()}.svc.cluster.local:{HTTP_API_PORT}"
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy, threat_service_url):
+    """PipelinePolicy with the threat assessment gRPC action method pre-registered."""
+    pipeline_policy.add_action_method(
+        name="assess",
+        url=threat_service_url,
+        service="threat.v1.ThreatAssessmentService",
+        method="AssessRequest",
+        message_template="threat.v1.ThreatRequest{uri: request.path}",
+    )
+    return pipeline_policy
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_basic.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_basic.py
new file mode 100644
index 000000000..32c2e81f2
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_basic.py
@@ -0,0 +1,50 @@
+"""Tests for PipelinePolicy grpc_method action: upstream calls and conditional execution."""
+
+import pytest
+
+from testsuite.utils.constants import THREAT_ASSESSMENT_THRESHOLD
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy):
+    """PipelinePolicy with conditional gRPC execution and threat-level deny."""
+    pipeline_policy.on_http_request.add_grpc_method(
+        method="assess",
+        var="threatResponse",
+        predicate='"x-assess-threat" in request.headers',
+    )
+    pipeline_policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    pipeline_policy.on_http_request.add_deny(
+        predicate=f"threatResponse.threat_level >= {THREAT_ASSESSMENT_THRESHOLD}",
+        with_status=403,
+    )
+
+    pipeline_policy.on_http_response.add_headers(
+        [["x-threat-assessed", "true"]],
+        predicate='"x-assess-threat" in request.headers',
+    )
+    pipeline_policy.on_http_response.add_headers(
+        [["x-threat-assessed", "false"]],
+        predicate='!("x-assess-threat" in request.headers)',
+    )
+    pipeline_policy.on_http_response.add_headers([["x-threat-threshold", str(THREAT_ASSESSMENT_THRESHOLD)]])
+
+    return pipeline_policy
+
+
+def test_basic_grpc_upstream_call(client):
+    """gRPC upstream is called when predicate matches, response var is available to subsequent actions."""
+    response = client.get("/get", headers={"x-assess-threat": "true"})
+    assert response.status_code == 200
+    assert response.headers.get("x-threat-assessed") == "true"
+    assert response.headers.get("x-threat-threshold") == str(THREAT_ASSESSMENT_THRESHOLD)
+
+
+def test_conditional_grpc_call_skipped(client):
+    """gRPC upstream is not called when predicate does not match."""
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-threat-assessed") == "false"
+    assert response.headers.get("x-threat-threshold") == str(THREAT_ASSESSMENT_THRESHOLD)
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_composition.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_composition.py
new file mode 100644
index 000000000..854d5a492
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_composition.py
@@ -0,0 +1,88 @@
+"""Tests for PipelinePolicy composition of gRPC actions with deny and fail."""
+
+import pytest
+
+from testsuite.kuadrant.extensions.pipeline_policy import PipelinePolicy
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit():
+    """No module-level policy; each test creates its own."""
+
+
+@pytest.fixture(scope="module")
+def create_grpc_policy(cluster, blame, route, threat_service_url, module_label):
+    """Factory for creating a PipelinePolicy with the assess gRPC action pre-registered."""
+
+    def _create(name):
+        policy = PipelinePolicy.create_instance(cluster, blame(name), route, labels={"testRun": module_label})
+        policy.add_action_method(
+            name="assess",
+            url=threat_service_url,
+            service="threat.v1.ThreatAssessmentService",
+            method="AssessRequest",
+            message_template="threat.v1.ThreatRequest{uri: request.path}",
+        )
+        policy.on_http_request.add_grpc_method(method="assess", var="threat")
+        return policy
+
+    return _create
+
+
+def test_fail_before_deny(request, create_grpc_policy, client):
+    """Fail action terminates the chain before the deny action when gRPC response triggers the fail predicate."""
+    policy = create_grpc_policy("failord")
+    policy.on_http_request.add_fail("threat too high", predicate="threat.threat_level >= 4")
+    policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/admin")
+    assert response.status_code == 500
+
+
+def test_deny_after_grpc_call(request, create_grpc_policy, client):
+    """Deny action after gRPC call works when the deny predicate matches."""
+    policy = create_grpc_policy("grpc-deny")
+    policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/blocked")
+    assert response.status_code == 403
+
+
+def test_deny_based_on_grpc_var(request, create_grpc_policy, client):
+    """Deny action using gRPC response variable denies requests when threat level is high."""
+    policy = create_grpc_policy("grpc-var-deny")
+    policy.on_http_request.add_deny(predicate="threat.threat_level >= 4", with_status=403)
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/admin")
+    assert response.status_code == 403
+
+    response = client.get("/get")
+    assert response.status_code == 200
+
+
+def test_deny_with_dynamic_body(request, create_grpc_policy, client):
+    """Deny action with CEL expression in withBody interpolates gRPC response variable."""
+    policy = create_grpc_policy("dyn-body")
+    policy.on_http_request.add_deny(
+        predicate="threat.threat_level >= 4",
+        with_status=403,
+        with_body="'blocked: threat level ' + string(threat.threat_level)",
+    )
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/admin")
+    assert response.status_code == 403
+    assert "blocked: threat level" in response.text
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_errors.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_errors.py
new file mode 100644
index 000000000..7471b63e3
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_errors.py
@@ -0,0 +1,80 @@
+"""Tests for PipelinePolicy gRPC error handling: unavailable services, wrong names, and fail actions."""
+
+import pytest
+
+from testsuite.kuadrant.extensions.pipeline_policy import PipelinePolicy
+from testsuite.kuadrant.policy import has_condition
+from testsuite.utils.constants import HTTP_API_PORT
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit():
+    """No module-level policy; each test creates its own PipelinePolicy."""
+
+
+def test_grpc_upstream_unavailable(request, cluster, blame, route, module_label):
+    """PipelinePolicy reports error when action method URL points to a non-existent service."""
+    policy = PipelinePolicy.create_instance(cluster, blame("bad-url"), route, labels={"testRun": module_label})
+    policy.add_action_method(
+        name="bad-method",
+        url=f"grpc://does-not-exist.default.svc.cluster.local:{HTTP_API_PORT}",
+        service="threat.v1.ThreatAssessmentService",
+        method="AssessRequest",
+        message_template="threat.v1.ThreatRequest{uri: request.path}",
+    )
+    policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    policy.on_http_request.add_grpc_method(method="bad-method")
+
+    request.addfinalizer(policy.delete)
+    policy.commit()
+
+    assert policy.wait_until(
+        has_condition("Accepted", "False", "Unknown", message="produced zero addresses"),
+        timelimit=60,
+    ), f"Policy did not reach expected error status, instead: {policy.refresh().model.status.conditions}"
+
+
+def test_grpc_wrong_service_name(request, cluster, blame, route, threat_service_url, module_label):
+    """PipelinePolicy reports error when action method references a non-existent gRPC service name."""
+    policy = PipelinePolicy.create_instance(cluster, blame("bad-svc"), route, labels={"testRun": module_label})
+    policy.add_action_method(
+        name="bad-service",
+        url=threat_service_url,
+        service="nonexistent.v1.FakeService",
+        method="DoSomething",
+        message_template="nonexistent.v1.FakeRequest{uri: request.path}",
+    )
+    policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    policy.on_http_request.add_grpc_method(method="bad-service")
+
+    request.addfinalizer(policy.delete)
+    policy.commit()
+
+    assert policy.wait_until(
+        has_condition("Accepted", "False", "Unknown", message="failed to fetch service"),
+        timelimit=60,
+    ), f"Policy did not reach expected error status, instead: {policy.refresh().model.status.conditions}"
+
+
+def test_grpc_wrong_method_name(request, cluster, blame, route, threat_service_url, module_label):
+    """PipelinePolicy reports error when action method references a non-existent gRPC method."""
+    policy = PipelinePolicy.create_instance(cluster, blame("bad-meth"), route, labels={"testRun": module_label})
+    policy.add_action_method(
+        name="wrong-method",
+        url=threat_service_url,
+        service="threat.v1.ThreatAssessmentService",
+        method="NonExistentMethod",
+        message_template="threat.v1.ThreatRequest{uri: request.path}",
+    )
+    policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    policy.on_http_request.add_grpc_method(method="wrong-method")
+
+    request.addfinalizer(policy.delete)
+    policy.commit()
+
+    assert policy.wait_until(
+        has_condition("Accepted", "False", "Unknown", message='method "NonExistentMethod" not found in service'),
+        timelimit=60,
+    ), f"Policy did not reach expected error status, instead: {policy.refresh().model.status.conditions}"
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_fail.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_fail.py
new file mode 100644
index 000000000..1975a539e
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_fail.py
@@ -0,0 +1,31 @@
+"""Tests for PipelinePolicy fail action triggered by gRPC response variables."""
+
+import pytest
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy):
+    """PipelinePolicy with gRPC call and fail action based on threat level."""
+    pipeline_policy.on_http_request.add_grpc_method(method="assess", var="threat")
+    pipeline_policy.on_http_request.add_fail(
+        "threat level too high",
+        predicate="threat.threat_level >= 4",
+    )
+    pipeline_policy.on_http_response.add_headers([["x-pipeline-active", "true"]])
+    return pipeline_policy
+
+
+def test_fail_triggers_on_high_threat(client):
+    """Fail action triggers when gRPC response threat_level meets the threshold and terminates the chain."""
+    response = client.get("/admin")
+    assert response.status_code == 500
+    assert response.headers.get("x-pipeline-active") is None
+
+
+def test_fail_does_not_trigger_on_low_threat(client):
+    """Fail action does not trigger when gRPC response threat_level is below the threshold."""
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-pipeline-active") == "true"
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/__init__.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/conftest.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/conftest.py
new file mode 100644
index 000000000..01958d124
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/conftest.py
@@ -0,0 +1,26 @@
+"""Fixtures for PipelinePolicy interaction tests with AuthPolicy and RateLimitPolicy."""
+
+import pytest
+
+from testsuite.httpx.auth import HttpxOidcClientAuth
+from testsuite.kuadrant.policy.rate_limit import Limit
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization, oidc_provider):
+    """AuthPolicy with OIDC identity verification."""
+    authorization.identity.add_oidc("default", oidc_provider.well_known["issuer"])
+    return authorization
+
+
+@pytest.fixture(scope="module")
+def auth(oidc_provider):
+    """Valid OIDC authentication for requests."""
+    return HttpxOidcClientAuth(oidc_provider.get_token, "authorization")
+
+
+@pytest.fixture(scope="module")
+def rate_limit(rate_limit):
+    """RateLimitPolicy with a low limit for testing."""
+    rate_limit.add_limit("basic", [Limit(3, "10s")])
+    return rate_limit
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/test_pipeline_policy_auth.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/test_pipeline_policy_auth.py
new file mode 100644
index 000000000..4ace7ebcd
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/test_pipeline_policy_auth.py
@@ -0,0 +1,42 @@
+"""Tests for PipelinePolicy interaction with AuthPolicy."""
+
+import pytest
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy):
+    """PipelinePolicy with deny action and response header."""
+    pipeline_policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    pipeline_policy.on_http_response.add_headers([["x-pipeline-policy", "active"]])
+    return pipeline_policy
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit(request, authorization, pipeline_policy):
+    """Commit AuthPolicy and PipelinePolicy."""
+    for component in [authorization, pipeline_policy]:
+        request.addfinalizer(component.delete)
+        component.commit()
+        component.wait_for_ready()
+
+
+def test_auth_and_pipeline_allowed(client, auth):
+    """Authenticated request to allowed path passes both AuthPolicy and PipelinePolicy."""
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 200
+    assert response.headers.get("x-pipeline-policy") == "active"
+
+
+def test_auth_and_pipeline_unauthorized(client):
+    """Unauthenticated request is rejected by AuthPolicy before PipelinePolicy runs."""
+    response = client.get("/get")
+    assert response.status_code == 401
+    assert response.headers.get("x-pipeline-policy") is None
+
+
+def test_auth_and_pipeline_blocked_path(client, auth):
+    """Authenticated request to blocked path is denied by PipelinePolicy deny action."""
+    response = client.get("/blocked", auth=auth)
+    assert response.status_code == 403
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/test_pipeline_policy_auth_ratelimit.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/test_pipeline_policy_auth_ratelimit.py
new file mode 100644
index 000000000..bb5f1bad5
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/test_pipeline_policy_auth_ratelimit.py
@@ -0,0 +1,34 @@
+"""Tests for PipelinePolicy interaction with both AuthPolicy and RateLimitPolicy."""
+
+import pytest
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy):
+    """PipelinePolicy with deny action and response header."""
+    pipeline_policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    pipeline_policy.on_http_response.add_headers([["x-pipeline-policy", "active"]])
+    return pipeline_policy
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit(request, authorization, rate_limit, pipeline_policy):
+    """Commit AuthPolicy, RateLimitPolicy, and PipelinePolicy."""
+    for component in [authorization, rate_limit, pipeline_policy]:
+        request.addfinalizer(component.delete)
+        component.commit()
+        component.wait_for_ready()
+
+
+@pytest.mark.flaky(reruns=3, reruns_delay=15)
+def test_all_policies_rate_limited(client, auth):
+    """Rate limit is enforced alongside AuthPolicy and PipelinePolicy."""
+    responses = client.get_many("/get", 3, auth=auth)
+    responses.assert_all(status_code=200)
+    for resp in responses:
+        assert resp.headers.get("x-pipeline-policy") == "active"
+
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 429
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/test_pipeline_policy_ratelimit.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/test_pipeline_policy_ratelimit.py
new file mode 100644
index 000000000..e4f7d42db
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/interactions/test_pipeline_policy_ratelimit.py
@@ -0,0 +1,33 @@
+"""Tests for PipelinePolicy interaction with RateLimitPolicy."""
+
+import pytest
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy):
+    """PipelinePolicy with response header."""
+    pipeline_policy.on_http_response.add_headers([["x-pipeline-policy", "active"]])
+    return pipeline_policy
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit(request, rate_limit, pipeline_policy):
+    """Commit RateLimitPolicy and PipelinePolicy."""
+    for component in [rate_limit, pipeline_policy]:
+        request.addfinalizer(component.delete)
+        component.commit()
+        component.wait_for_ready()
+
+
+@pytest.mark.flaky(reruns=3, reruns_delay=15)
+def test_rate_limit_and_pipeline(client):
+    """Rate limit is enforced alongside PipelinePolicy actions."""
+    responses = client.get_many("/get", 3)
+    responses.assert_all(status_code=200)
+    for resp in responses:
+        assert resp.headers.get("x-pipeline-policy") == "active"
+
+    response = client.get("/get")
+    assert response.status_code == 429
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_basic.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_basic.py
new file mode 100644
index 000000000..a1bc22b70
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_basic.py
@@ -0,0 +1,26 @@
+"""Basic happy-path tests for PipelinePolicy: deny action and response headers."""
+
+import pytest
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy):
+    """Configure PipelinePolicy with deny action and response headers."""
+    pipeline_policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    pipeline_policy.on_http_response.add_headers([["x-pipeline-policy", "active"]])
+    return pipeline_policy
+
+
+def test_allowed_path(client):
+    """Request to an allowed path returns 200 with the custom response header."""
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-pipeline-policy") == "active"
+
+
+def test_blocked_path(client):
+    """Request to /blocked is denied by the deny action."""
+    response = client.get("/blocked")
+    assert response.status_code == 403
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_composition.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_composition.py
new file mode 100644
index 000000000..bc927fd54
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_composition.py
@@ -0,0 +1,77 @@
+"""Tests for PipelinePolicy composition: action ordering, empty and partial pipelines."""
+
+import pytest
+
+from testsuite.kuadrant.extensions.pipeline_policy import PipelinePolicy
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit():
+    """No module-level policy; each test creates its own."""
+
+
+def test_first_deny_wins(request, cluster, blame, route, client, module_label):
+    """First deny action terminates the chain; the second deny with the same predicate never executes."""
+    policy = PipelinePolicy.create_instance(cluster, blame("order"), route, labels={"testRun": module_label})
+    policy.on_http_request.add_deny(predicate='request.url_path == "/order-test"', with_status=403)
+    policy.on_http_request.add_deny(predicate='request.url_path == "/order-test"', with_status=429)
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/order-test")
+    assert response.status_code == 403
+
+
+def test_response_action_ordering(request, cluster, blame, route, client, module_label):
+    """Response actions execute in spec order; both headers from separate actions are present."""
+    policy = PipelinePolicy.create_instance(cluster, blame("respord"), route, labels={"testRun": module_label})
+    policy.on_http_response.add_headers([["x-first", "alpha"]])
+    policy.on_http_response.add_headers([["x-second", "bravo"]])
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-first") == "alpha"
+    assert response.headers.get("x-second") == "bravo"
+
+
+def test_empty_pipeline(request, cluster, blame, route, client, module_label):
+    """PipelinePolicy with no actions passes requests through unmodified."""
+    policy = PipelinePolicy.create_instance(cluster, blame("empty"), route, labels={"testRun": module_label})
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-pipeline-policy") is None
+
+
+def test_request_only_pipeline(request, cluster, blame, route, client, module_label):
+    """PipelinePolicy with only request actions and no response section works correctly."""
+    policy = PipelinePolicy.create_instance(cluster, blame("reqonly"), route, labels={"testRun": module_label})
+    policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    assert client.get("/get").status_code == 200
+    assert client.get("/blocked").status_code == 403
+
+
+def test_response_only_pipeline(request, cluster, blame, route, client, module_label):
+    """PipelinePolicy with only response actions and no request section works correctly."""
+    policy = PipelinePolicy.create_instance(cluster, blame("resonly"), route, labels={"testRun": module_label})
+    policy.on_http_response.add_headers([["x-resp-only", "true"]])
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-resp-only") == "true"
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_deny.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_deny.py
new file mode 100644
index 000000000..1094c06fa
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_deny.py
@@ -0,0 +1,127 @@
+"""Tests for PipelinePolicy deny action variants in request and response phases."""
+
+import pytest
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy):
+    """PipelinePolicy with multiple deny configurations covering all deny variants."""
+    # Path-based deny
+    pipeline_policy.on_http_request.add_deny(predicate='request.url_path == "/deny-path"', with_status=403)
+    # Header-based deny
+    pipeline_policy.on_http_request.add_deny(predicate='"x-deny-me" in request.headers', with_status=401)
+    # Custom status code
+    pipeline_policy.on_http_request.add_deny(predicate='request.url_path == "/custom-status"', with_status=429)
+    # Custom headers
+    pipeline_policy.on_http_request.add_deny(
+        predicate='request.url_path == "/custom-headers"',
+        with_status=403,
+        with_headers='[["x-deny-reason", "blocked"]]',
+    )
+    # Custom body (CEL expression)
+    pipeline_policy.on_http_request.add_deny(
+        predicate='request.url_path == "/custom-body"',
+        with_status=403,
+        with_body="'Access denied'",
+    )
+    # All response fields (CEL expression in body)
+    pipeline_policy.on_http_request.add_deny(
+        predicate='request.url_path == "/custom-all"',
+        with_status=451,
+        with_headers='[["x-deny-reason", "full-custom"]]',
+        with_body="'Fully customized denial'",
+    )
+    # Response phase deny — status override
+    pipeline_policy.on_http_response.add_deny(
+        predicate='"x-override-code" in request.headers',
+        with_status=503,
+    )
+    # Response phase deny — all fields (CEL expression in body)
+    pipeline_policy.on_http_response.add_deny(
+        predicate='"x-resp-deny" in request.headers',
+        with_status=418,
+        with_headers='[["x-deny-phase", "response"]]',
+        with_body="'Teapot response'",
+    )
+    return pipeline_policy
+
+
+def test_path_based_deny(client):
+    """Request to a path matching the deny predicate is denied with 403."""
+    response = client.get("/deny-path")
+    assert response.status_code == 403
+
+
+def test_cel_predicate_does_not_deny(client):
+    """Request to a path not matching any deny predicate passes through."""
+    response = client.get("/get")
+    assert response.status_code == 200
+
+
+def test_header_based_deny(client):
+    """Request with a header matching the deny predicate is denied."""
+    response = client.get("/get", headers={"x-deny-me": "true"})
+    assert response.status_code == 401
+
+
+def test_header_based_deny_absent(client):
+    """Request without the blocked header passes through."""
+    response = client.get("/get")
+    assert response.status_code == 200
+
+
+def test_multiple_deny_actions_or_behavior(client):
+    """Multiple deny actions behave as OR: any matching predicate denies the request."""
+    assert client.get("/deny-path").status_code == 403
+    assert client.get("/get", headers={"x-deny-me": "true"}).status_code == 401
+    assert client.get("/get").status_code == 200
+
+
+def test_deny_custom_status_code(client):
+    """Deny with custom withStatus returns that status code."""
+    response = client.get("/custom-status")
+    assert response.status_code == 429
+
+
+def test_deny_custom_headers(client):
+    """Deny with withHeaders includes custom headers in the denied response."""
+    response = client.get("/custom-headers")
+    assert response.status_code == 403
+    assert response.headers.get("x-deny-reason") == "blocked"
+
+
+def test_deny_custom_body(client):
+    """Deny with withBody as CEL expression returns custom body text."""
+    response = client.get("/custom-body")
+    assert response.status_code == 403
+    assert response.text == "Access denied"
+
+
+def test_deny_all_response_fields(client):
+    """Deny with withStatus, withHeaders, and withBody as CEL expression returns all fields."""
+    response = client.get("/custom-all")
+    assert response.status_code == 451
+    assert response.headers.get("x-deny-reason") == "full-custom"
+    assert response.text == "Fully customized denial"
+
+
+def test_response_deny_override_status(client):
+    """Response deny overrides the backend response status code."""
+    response = client.get("/get", headers={"x-override-code": "true"})
+    assert response.status_code == 503
+
+
+def test_response_deny_no_override(client):
+    """Response without the override header passes through normally."""
+    response = client.get("/get")
+    assert response.status_code == 200
+
+
+def test_response_deny_with_headers_and_body(client):
+    """Response deny with all fields as CEL expressions replaces the backend response."""
+    response = client.get("/get", headers={"x-resp-deny": "true"})
+    assert response.status_code == 418
+    assert response.headers.get("x-deny-phase") == "response"
+    assert response.text == "Teapot response"
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_isolation.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_isolation.py
new file mode 100644
index 000000000..e869b945d
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_isolation.py
@@ -0,0 +1,112 @@
+"""Tests for PipelinePolicy isolation: policy should not leak to untargeted routes or gateways."""
+
+import time
+
+import pytest
+
+from testsuite.gateway import GatewayListener
+from testsuite.gateway.gateway_api.gateway import KuadrantGateway
+from testsuite.gateway.gateway_api.route import HTTPRoute
+from testsuite.utils.constants import EXTENSION_POLICY_PROPAGATION_WAIT
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def hostname2(gateway, exposer, blame):
+    """Second hostname for the unaffected route on the same gateway"""
+    return exposer.expose_hostname(blame("no-pp"), gateway)
+
+
+@pytest.fixture(scope="module")
+def route2(request, gateway, blame, hostname2, backend, module_label):
+    """Second HTTPRoute on the same gateway without any PipelinePolicy"""
+    route = HTTPRoute.create_instance(gateway.cluster, blame("route2"), gateway, {"app": module_label})
+    route.add_hostname(hostname2.hostname)
+    route.add_backend(backend)
+    request.addfinalizer(route.delete)
+    route.commit()
+    return route
+
+
+@pytest.fixture(scope="module")
+def client2(route2, hostname2):  # pylint: disable=unused-argument
+    """Client for the route without PipelinePolicy"""
+    client = hostname2.client()
+    yield client
+    client.close()
+
+
+@pytest.fixture(scope="module")
+def gateway2(request, cluster, blame, wildcard_domain, module_label):
+    """Second gateway without any PipelinePolicy"""
+    gw = KuadrantGateway.create_instance(cluster, blame("gw2"), {"app": module_label})
+    gw.add_listener(GatewayListener(hostname=wildcard_domain))
+    request.addfinalizer(gw.delete)
+    gw.commit()
+    gw.wait_for_ready()
+    return gw
+
+
+@pytest.fixture(scope="module")
+def hostname3(gateway2, exposer, blame):
+    """Hostname for the route on the second gateway"""
+    return exposer.expose_hostname(blame("no-pp-gw"), gateway2)
+
+
+@pytest.fixture(scope="module")
+def route3(request, gateway2, blame, hostname3, backend, module_label):
+    """HTTPRoute on the second gateway without any PipelinePolicy"""
+    route = HTTPRoute.create_instance(gateway2.cluster, blame("route3"), gateway2, {"app": module_label})
+    route.add_hostname(hostname3.hostname)
+    route.add_backend(backend)
+    request.addfinalizer(route.delete)
+    route.commit()
+    return route
+
+
+@pytest.fixture(scope="module")
+def client3(route3, hostname3):  # pylint: disable=unused-argument
+    """Client for the route on the second gateway"""
+    client = hostname3.client()
+    yield client
+    client.close()
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy):
+    """PipelinePolicy with response header targeting only the first route."""
+    pipeline_policy.on_http_response.add_headers([["x-pipeline-policy", "active"]])
+    return pipeline_policy
+
+
+def test_policy_does_not_affect_other_route(client, client2):
+    """Route without PipelinePolicy on the same gateway does not get the response header."""
+    # Wait for EnvoyFilter propagation after route2/client2 fixtures are created.
+    # Without this delay, a leak might not have propagated yet, causing the test to pass even when
+    # the policy is incorrectly leaking to the second route.
+    time.sleep(EXTENSION_POLICY_PROPAGATION_WAIT)
+
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-pipeline-policy") == "active"
+
+    response = client2.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-pipeline-policy") is None
+
+
+def test_policy_does_not_affect_other_gateway(client, client3):
+    """Route on a different gateway does not get the response header."""
+    # Wait for EnvoyFilter propagation after gateway2/route3/client3 fixtures are created.
+    # Without this delay, a leak might not have propagated yet, causing the test to pass even when
+    # the policy is incorrectly leaking to the second gateway.
+    time.sleep(EXTENSION_POLICY_PROPAGATION_WAIT)
+
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-pipeline-policy") == "active"
+
+    response = client3.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-pipeline-policy") is None
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_lifecycle.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_lifecycle.py
new file mode 100644
index 000000000..9f846d252
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_lifecycle.py
@@ -0,0 +1,60 @@
+"""Tests for PipelinePolicy lifecycle: update and delete."""
+
+import time
+
+import pytest
+
+from testsuite.kuadrant.extensions.pipeline_policy import PipelinePolicy
+from testsuite.utils.constants import EXTENSION_POLICY_PROPAGATION_WAIT
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit():
+    """No module-level policy; each test creates its own."""
+
+
+@pytest.mark.flaky(reruns=0)
+def test_update_policy(request, cluster, blame, route, client, module_label):
+    """Adding a new response header via policy update propagates to traffic."""
+    policy = PipelinePolicy.create_instance(cluster, blame("upd-pp"), route, labels={"testRun": module_label})
+    policy.on_http_response.add_headers([["x-update-test", "active"]])
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-update-test") == "active"
+    assert response.headers.get("x-update-new") is None
+
+    policy.on_http_response.add_headers([["x-update-new", "true"]])
+    policy.wait_for_ready()
+
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-update-test") == "active"
+    assert response.headers.get("x-update-new") == "true"
+
+
+@pytest.mark.flaky(reruns=0)
+def test_delete_policy(request, cluster, blame, route, client, module_label):
+    """After deleting the PipelinePolicy, the CR is removed and the actions stop being enforced."""
+    policy = PipelinePolicy.create_instance(cluster, blame("del-pp"), route, labels={"testRun": module_label})
+    policy.on_http_response.add_headers([["x-delete-test", "active"]])
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-delete-test") == "active"
+
+    policy.delete()
+    assert not policy.committed, "PipelinePolicy was not deleted"
+    time.sleep(EXTENSION_POLICY_PROPAGATION_WAIT)
+
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-delete-test") is None
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_response.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_response.py
new file mode 100644
index 000000000..7d6b5c4d1
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_response.py
@@ -0,0 +1,80 @@
+"""Tests for PipelinePolicy response add_headers actions."""
+
+import pytest
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy):
+    """PipelinePolicy with various add_headers response actions for testing."""
+    pipeline_policy.on_http_response.add_headers([["x-single", "one"]])
+    pipeline_policy.on_http_response.add_headers([["x-multi-a", "alpha"], ["x-multi-b", "bravo"]])
+    pipeline_policy.on_http_response.add_headers([["x-separate", "separate-value"]])
+
+    pipeline_policy.on_http_response.add_headers(
+        [["x-conditional", "present"]],
+        predicate='"x-trigger" in request.headers',
+    )
+
+    pipeline_policy.on_http_response.add_headers(
+        [["x-mode", "active"]],
+        predicate='"x-trigger" in request.headers',
+    )
+    pipeline_policy.on_http_response.add_headers(
+        [["x-mode", "inactive"]],
+        predicate='!("x-trigger" in request.headers)',
+    )
+
+    return pipeline_policy
+
+
+def test_single_response_header(client):
+    """Single response header is added to the response."""
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-single") == "one"
+
+
+def test_multiple_headers_in_one_action(client):
+    """Multiple headers added in a single add_headers action are all present."""
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-multi-a") == "alpha"
+    assert response.headers.get("x-multi-b") == "bravo"
+
+
+def test_separate_add_headers_actions(client):
+    """Headers from two separate add_headers actions are all present."""
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-single") == "one"
+    assert response.headers.get("x-separate") == "separate-value"
+
+
+def test_conditional_header_present(client):
+    """Header with predicate is added when the condition is met."""
+    response = client.get("/get", headers={"x-trigger": "true"})
+    assert response.status_code == 200
+    assert response.headers.get("x-conditional") == "present"
+
+
+def test_conditional_header_absent(client):
+    """Header with predicate is not added when the condition is not met."""
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-conditional") is None
+
+
+def test_mutually_exclusive_headers_active(client):
+    """Mutually exclusive headers: active mode when trigger header is present."""
+    response = client.get("/get", headers={"x-trigger": "true"})
+    assert response.status_code == 200
+    assert response.headers.get("x-mode") == "active"
+
+
+def test_mutually_exclusive_headers_inactive(client):
+    """Mutually exclusive headers: inactive mode when trigger header is absent."""
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-mode") == "inactive"
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_targeting.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_targeting.py
new file mode 100644
index 000000000..c04fb899a
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_targeting.py
@@ -0,0 +1,70 @@
+"""Tests for PipelinePolicy Gateway-level targeting.
+
+HTTPRoute targeting is the default and is covered by test_pipeline_policy_basic.py.
+"""
+
+import pytest
+
+from testsuite.gateway.gateway_api.route import HTTPRoute
+from testsuite.kuadrant.extensions.pipeline_policy import PipelinePolicy
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def hostname2(gateway, exposer, blame):
+    """Second hostname for the second route"""
+    return exposer.expose_hostname(blame("hostname2"), gateway)
+
+
+@pytest.fixture(scope="module")
+def route2(request, gateway, blame, hostname2, backend, module_label):
+    """Second HTTPRoute on the same gateway with a different hostname"""
+    route = HTTPRoute.create_instance(gateway.cluster, blame("route2"), gateway, {"app": module_label})
+    route.add_hostname(hostname2.hostname)
+    route.add_backend(backend)
+    request.addfinalizer(route.delete)
+    route.commit()
+    return route
+
+
+@pytest.fixture(scope="module")
+def client2(route2, hostname2):  # pylint: disable=unused-argument
+    """Client for the second route"""
+    client = hostname2.client()
+    yield client
+    client.close()
+
+
+@pytest.fixture(scope="module")
+def gateway_policy(cluster, blame, gateway, module_label):
+    """PipelinePolicy targeting the Gateway with deny and response headers."""
+    policy = PipelinePolicy.create_instance(cluster, blame("gw-pp"), gateway, labels={"testRun": module_label})
+    policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    policy.on_http_response.add_headers([["x-gateway-policy", "active"]])
+    return policy
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit(request, route2, gateway_policy):  # pylint: disable=unused-argument
+    """Commit gateway policy after route2 is ready."""
+    request.addfinalizer(gateway_policy.delete)
+    gateway_policy.commit()
+    gateway_policy.wait_for_ready()
+
+
+@pytest.mark.parametrize("client_fixture", ["client", "client2"])
+def test_gateway_target_affects_routes(request, client_fixture):
+    """Request to each route gets pipeline response header when policy targets the gateway."""
+    http_client = request.getfixturevalue(client_fixture)
+    response = http_client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-gateway-policy") == "active"
+
+
+@pytest.mark.parametrize("client_fixture", ["client", "client2"])
+def test_gateway_target_blocked_path(request, client_fixture):
+    """Blocked path is denied on both routes when policy targets the gateway."""
+    http_client = request.getfixturevalue(client_fixture)
+    response = http_client.get("/blocked")
+    assert response.status_code == 403
diff --git a/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_validation.py b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_validation.py
new file mode 100644
index 000000000..be3c90191
--- /dev/null
+++ b/testsuite/tests/singlecluster/extensions/pipeline_policy/test_pipeline_policy_validation.py
@@ -0,0 +1,95 @@
+"""Tests for PipelinePolicy validation: rejected configurations and error conditions."""
+
+import pytest
+
+from testsuite.gateway import CustomReference
+from testsuite.kuadrant.extensions.pipeline_policy import PipelinePolicy
+from testsuite.kuadrant.policy import has_condition
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit():
+    """No module-level policy; each test creates its own with bad configuration."""
+
+
+@pytest.mark.parametrize("kind", ["HTTPRoute", "Gateway"])
+def test_invalid_target_ref(request, cluster, blame, kind, module_label):
+    """PipelinePolicy targeting a non-existent resource does not reach Accepted state."""
+    target_name = "does-not-exist"
+    target = CustomReference(
+        group="gateway.networking.k8s.io",
+        kind=kind,
+        name=target_name,
+    )
+    policy = PipelinePolicy.create_instance(cluster, blame("bad-target"), target, labels={"testRun": module_label})
+    policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+
+    request.addfinalizer(policy.delete)
+    policy.commit()
+
+    assert policy.wait_until(
+        has_condition("Accepted", "False", message=f"targetRef {kind} {cluster.project}/{target_name} not found"),
+        timelimit=30,
+    ), f"Policy did not report target not found, status: {policy.refresh().model.status.conditions}"
+
+
+def test_top_level_fail_action(request, cluster, blame, route, module_label):
+    """PipelinePolicy with a top-level fail action (not inside gRPC onReply) should not be accepted."""
+    policy = PipelinePolicy.create_instance(cluster, blame("top-fail"), route, labels={"testRun": module_label})
+    policy.on_http_request.add_fail("top-level fail", predicate='request.url_path == "/fail"')
+
+    request.addfinalizer(policy.delete)
+    policy.commit()
+
+    assert policy.wait_until(
+        has_condition("Accepted", "False", message="fail action must reference a gRPC response variable"),
+        timelimit=30,
+    ), f"Policy with top-level fail was accepted, status: {policy.refresh().model.status.conditions}"
+
+
+def test_invalid_cel_expression(request, cluster, blame, route, module_label):
+    """PipelinePolicy with malformed CEL predicate fails to enforce."""
+    policy = PipelinePolicy.create_instance(cluster, blame("bad-cel"), route, labels={"testRun": module_label})
+    policy.on_http_request.add_deny(predicate="INVALID CEL !!!", with_status=403)
+
+    request.addfinalizer(policy.delete)
+    policy.commit()
+
+    assert policy.wait_until(
+        has_condition("Accepted", "False", message="invalid CEL expression"),
+        timelimit=30,
+    ), f"Policy did not report error for invalid CEL, status: {policy.refresh().model.status.conditions}"
+
+
+def test_variable_forward_reference(request, cluster, blame, route, module_label):
+    """PipelinePolicy referencing a variable before it is defined should fail validation."""
+    var_name = "threatResponse"
+    policy = PipelinePolicy.create_instance(cluster, blame("fwd-ref"), route, labels={"testRun": module_label})
+    policy.on_http_request.add_deny(predicate=f"{var_name}.threat_level >= 50", with_status=403)
+    policy.on_http_request.add_grpc_method(method="nonexistent-method", var=var_name)
+
+    request.addfinalizer(policy.delete)
+    policy.commit()
+
+    assert policy.wait_until(
+        has_condition("Accepted", "False", message=f'action references variable "{var_name}" before it is populated'),
+        timelimit=60,
+    ), f"Policy did not reject forward reference, status: {policy.refresh().model.status.conditions}"
+
+
+def test_duplicate_variable_name(request, cluster, blame, route, module_label):
+    """PipelinePolicy with two gRPC methods using the same variable name should fail validation."""
+    var_name = "dupVar"
+    policy = PipelinePolicy.create_instance(cluster, blame("dup-var"), route, labels={"testRun": module_label})
+    policy.on_http_request.add_grpc_method(method="method-a", var=var_name)
+    policy.on_http_request.add_grpc_method(method="method-b", var=var_name)
+
+    request.addfinalizer(policy.delete)
+    policy.commit()
+
+    assert policy.wait_until(
+        has_condition("Accepted", "False", message=f'duplicate variable name "{var_name}"'),
+        timelimit=60,
+    ), f"Policy did not reject duplicate var, status: {policy.refresh().model.status.conditions}"
diff --git a/testsuite/utils/constants.py b/testsuite/utils/constants.py
index 26c6a718d..87af044b0 100644
--- a/testsuite/utils/constants.py
+++ b/testsuite/utils/constants.py
@@ -165,6 +165,19 @@
 # DNS propagation wait.
 DNS_PROPAGATION_WAIT = 300  # 5 minutes
 
+# --- Threat Assessment Service ---
+
+# Threat level threshold for deny actions in PipelinePolicy gRPC tests.
+# Set intentionally high (max possible score is ~13) so the threat-level-based
+# deny never triggers, allowing tests to verify gRPC call execution and
+# response variable propagation without interference from the deny action.
+THREAT_ASSESSMENT_THRESHOLD = 50
+
+# --- Extension Policy ---
+
+# Wait for EnvoyFilter to propagate before checking isolation (testing absence of leaking).
+EXTENSION_POLICY_PROPAGATION_WAIT = 10
+
 # --- Miscellaneous Workarounds (seconds) ---
 
 # Workaround for https://github.com/Kuadrant/testsuite/issues/884 — remove when fixed