Repository: PrefectHQ/prefect

Issue:
DockerAgent on Prefect Server tries to pull secrets from API
## Description
<!-- A clear description of the bug -->

If a secret is not available in the context the API is queried for the secret (when the backend is `cloud`) or an error is raised (when the backend is `server`). The configured backend is not passed from the agent environment into the docker container so a confusing graphql error is raised instead of the nicer 'Local secret not found' when you're using the `server` backend.


```python
# secrets.py
        try:
            value = secrets[self.name]
        except KeyError:
           # This is the code path we want to hit
            if prefect.config.backend != "cloud":
                raise ValueError(
                    'Local Secret "{}" was not found.'.format(self.name)
                ) from None
           # Query the API...
```


## Expected Behavior
<!-- What did you expect to happen instead? -->

The backend is passed through to the container so the helpful error is raised

## Reproduction
<!-- A minimal example that exhibits the behavior. -->

A `DockerRun` flow with a secret that is not passed to the agent environment explicitly


Discussion:

Patch:
diff --git a/changes/pr4076.yaml b/changes/pr4076.yaml
new file mode 100644
index 0000000000..fc9b3e8244
--- /dev/null
+++ b/changes/pr4076.yaml
@@ -0,0 +1,2 @@
+fix:
+  - "Forward Prefect backend type to deployed flow runs, ensuring backend-specific logic functions properly - [#4076](https://github.com/PrefectHQ/prefect/pull/4076)"
diff --git a/src/prefect/agent/docker/agent.py b/src/prefect/agent/docker/agent.py
index 4b231bc6e7..218fd74449 100644
--- a/src/prefect/agent/docker/agent.py
+++ b/src/prefect/agent/docker/agent.py
@@ -529,6 +529,7 @@ class DockerAgent(Agent):
         # 4. Non-overrideable required env vars
         env.update(
             {
+                "PREFECT__BACKEND": config.backend,
                 "PREFECT__CLOUD__API": api,
                 "PREFECT__CLOUD__AUTH_TOKEN": config.cloud.agent.auth_token,
                 "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
diff --git a/src/prefect/agent/ecs/agent.py b/src/prefect/agent/ecs/agent.py
index 36c0c46aac..9caf518b3a 100644
--- a/src/prefect/agent/ecs/agent.py
+++ b/src/prefect/agent/ecs/agent.py
@@ -518,6 +518,7 @@ class ECSAgent(Agent):
             env.update(run_config.env)
         env.update(
             {
+                "PREFECT__BACKEND": config.backend,
                 "PREFECT__CLOUD__API": config.cloud.api,
                 "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,
                 "PREFECT__CONTEXT__FLOW_ID": flow_run.flow.id,
diff --git a/src/prefect/agent/fargate/agent.py b/src/prefect/agent/fargate/agent.py
index afce33e802..ebd21c19ce 100644
--- a/src/prefect/agent/fargate/agent.py
+++ b/src/prefect/agent/fargate/agent.py
@@ -611,6 +611,10 @@ class FargateAgent(Agent):
                 "image": image,
                 "command": ["/bin/sh", "-c", flow_run_command],
                 "environment": [
+                    {
+                        "name": "PREFECT__BACKEND",
+                        "value": config.backend,
+                    },
                     {
                         "name": "PREFECT__CLOUD__API",
                         "value": config.cloud.api or "https://api.prefect.io",
diff --git a/src/prefect/agent/kubernetes/agent.py b/src/prefect/agent/kubernetes/agent.py
index 223cf30bfa..ddf4c03506 100644
--- a/src/prefect/agent/kubernetes/agent.py
+++ b/src/prefect/agent/kubernetes/agent.py
@@ -667,6 +667,7 @@ class KubernetesAgent(Agent):
             env.update(run_config.env)
         env.update(
             {
+                "PREFECT__BACKEND": config.backend,
                 "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
                 "PREFECT__CLOUD__API": config.cloud.api,
                 "PREFECT__CLOUD__AUTH_TOKEN": config.cloud.agent.auth_token,
diff --git a/src/prefect/agent/kubernetes/job_spec.yaml b/src/prefect/agent/kubernetes/job_spec.yaml
index a45e5c423f..96c2dae2b0 100644
--- a/src/prefect/agent/kubernetes/job_spec.yaml
+++ b/src/prefect/agent/kubernetes/job_spec.yaml
@@ -48,4 +48,4 @@ spec:
       imagePullSecrets:
       - name: ""
       volumes:
-        - name: ""
\ No newline at end of file
+        - name: ""
diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py
index 9b2ab0efcb..358270efda 100644
--- a/src/prefect/agent/local/agent.py
+++ b/src/prefect/agent/local/agent.py
@@ -216,6 +216,7 @@ class LocalAgent(Agent):
         # 6. Non-overrideable required env vars
         env.update(
             {
+                "PREFECT__BACKEND": config.backend,
                 "PREFECT__CLOUD__API": config.cloud.api,
                 "PREFECT__CLOUD__AUTH_TOKEN": self.client._api_token,
                 "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),


Test patch:
diff --git a/tests/agent/test_docker_agent.py b/tests/agent/test_docker_agent.py
index f7b56a601c..92bdad2372 100644
--- a/tests/agent/test_docker_agent.py
+++ b/tests/agent/test_docker_agent.py
@@ -3,6 +3,7 @@ from unittest.mock import MagicMock
 
 import pytest
 
+import prefect
 from prefect import context
 from prefect.agent.docker.agent import DockerAgent, _stream_container_logs
 from prefect.environments import LocalEnvironment
@@ -116,15 +117,21 @@ def test_populate_env_vars_from_agent_config(api):
     assert env_vars["AUTH_THING"] == "foo"
 
 
-def test_populate_env_vars(api):
+def test_populate_env_vars(api, backend):
     agent = DockerAgent()
 
     env_vars = agent.populate_env_vars(
         GraphQLResult({"id": "id", "name": "name", "flow": {"id": "foo"}}), "test-image"
     )
 
+    if backend == "server":
+        cloud_api = "http://host.docker.internal:4200"
+    else:
+        cloud_api = prefect.config.cloud.api
+
     expected_vars = {
-        "PREFECT__CLOUD__API": "https://api.prefect.io",
+        "PREFECT__BACKEND": backend,
+        "PREFECT__CLOUD__API": cloud_api,
         "PREFECT__CLOUD__AUTH_TOKEN": "TEST_TOKEN",
         "PREFECT__CLOUD__AGENT__LABELS": "[]",
         "PREFECT__CONTEXT__FLOW_RUN_ID": "id",
diff --git a/tests/agent/test_ecs_agent.py b/tests/agent/test_ecs_agent.py
index 5a44496438..86ec5ff827 100644
--- a/tests/agent/test_ecs_agent.py
+++ b/tests/agent/test_ecs_agent.py
@@ -4,6 +4,7 @@ import box
 import pytest
 import yaml
 
+import prefect
 from prefect.agent.ecs.agent import (
     merge_run_task_kwargs,
     ECSAgent,
@@ -484,7 +485,7 @@ class TestGetRunTaskKwargs:
             "overrides": {"cpu": "2048", "memory": "2048", "taskRoleArn": "testing"},
         }
 
-    def test_get_run_task_kwargs_environment(self, tmpdir):
+    def test_get_run_task_kwargs_environment(self, tmpdir, backend):
         path = str(tmpdir.join("kwargs.yaml"))
         with open(path, "w") as f:
             yaml.safe_dump(
@@ -512,7 +513,8 @@ class TestGetRunTaskKwargs:
         env_list = kwargs["overrides"]["containerOverrides"][0]["environment"]
         env = {item["name"]: item["value"] for item in env_list}
         assert env == {
-            "PREFECT__CLOUD__API": "https://api.prefect.io",
+            "PREFECT__BACKEND": backend,
+            "PREFECT__CLOUD__API": prefect.config.cloud.api,
             "PREFECT__CLOUD__AUTH_TOKEN": "TEST_TOKEN",
             "PREFECT__CLOUD__AGENT__LABELS": "[]",
             "PREFECT__CONTEXT__FLOW_RUN_ID": "flow-run-id",
diff --git a/tests/agent/test_fargate_agent.py b/tests/agent/test_fargate_agent.py
index 9431364782..e97af08c51 100644
--- a/tests/agent/test_fargate_agent.py
+++ b/tests/agent/test_fargate_agent.py
@@ -2,6 +2,7 @@ from unittest.mock import MagicMock
 
 import pytest
 
+import prefect
 from prefect.agent.fargate import FargateAgent
 from prefect.environments import LocalEnvironment
 from prefect.storage import Docker, Local
@@ -931,7 +932,10 @@ def test_deploy_flow_register_task_definition_uses_user_env_vars(
     ],
 )
 def test_deploy_flow_register_task_definition_all_args(
-    core_version, command, monkeypatch, cloud_api
+    core_version,
+    command,
+    monkeypatch,
+    backend,
 ):
     boto3_client = MagicMock()
 
@@ -1002,14 +1006,13 @@ def test_deploy_flow_register_task_definition_all_args(
         ],
     }
 
-    with set_temporary_config({"logging.log_to_cloud": True}):
-        agent = FargateAgent(
-            aws_access_key_id="id",
-            aws_secret_access_key="secret",
-            aws_session_token="token",
-            region_name="region",
-            **kwarg_dict
-        )
+    agent = FargateAgent(
+        aws_access_key_id="id",
+        aws_secret_access_key="secret",
+        aws_session_token="token",
+        region_name="region",
+        **kwarg_dict
+    )
     agent.deploy_flow(
         flow_run=GraphQLResult(
             {
@@ -1042,7 +1045,8 @@ def test_deploy_flow_register_task_definition_all_args(
             "image": "test/name:tag",
             "command": ["/bin/sh", "-c", command],
             "environment": [
-                {"name": "PREFECT__CLOUD__API", "value": "https://api.prefect.io"},
+                {"name": "PREFECT__BACKEND", "value": backend},
+                {"name": "PREFECT__CLOUD__API", "value": prefect.config.cloud.api},
                 {"name": "PREFECT__CLOUD__AGENT__LABELS", "value": "[]"},
                 {"name": "PREFECT__CLOUD__USE_LOCAL_SECRETS", "value": "false"},
                 {"name": "PREFECT__LOGGING__LOG_TO_CLOUD", "value": "true"},
@@ -1177,7 +1181,8 @@ def test_deploy_flows_includes_agent_labels_in_environment(
             "image": "test/name:tag",
             "command": ["/bin/sh", "-c", "prefect execute flow-run"],
             "environment": [
-                {"name": "PREFECT__CLOUD__API", "value": "https://api.prefect.io"},
+                {"name": "PREFECT__BACKEND", "value": "cloud"},
+                {"name": "PREFECT__CLOUD__API", "value": prefect.config.cloud.api},
                 {
                     "name": "PREFECT__CLOUD__AGENT__LABELS",
                     "value": "['aws', 'staging']",
@@ -1281,7 +1286,8 @@ def test_deploy_flows_enable_task_revisions_no_tags(monkeypatch, cloud_api):
                 "image": "test/name:tag",
                 "command": ["/bin/sh", "-c", "prefect execute flow-run"],
                 "environment": [
-                    {"name": "PREFECT__CLOUD__API", "value": "https://api.prefect.io"},
+                    {"name": "PREFECT__BACKEND", "value": "cloud"},
+                    {"name": "PREFECT__CLOUD__API", "value": prefect.config.cloud.api},
                     {"name": "PREFECT__CLOUD__AGENT__LABELS", "value": "[]"},
                     {"name": "PREFECT__CLOUD__USE_LOCAL_SECRETS", "value": "false"},
                     {"name": "PREFECT__LOGGING__LOG_TO_CLOUD", "value": "true"},
@@ -1659,7 +1665,8 @@ def test_deploy_flows_enable_task_revisions_with_external_kwargs(
                 "image": "test/name:tag",
                 "command": ["/bin/sh", "-c", "prefect execute flow-run"],
                 "environment": [
-                    {"name": "PREFECT__CLOUD__API", "value": "https://api.prefect.io"},
+                    {"name": "PREFECT__BACKEND", "value": "cloud"},
+                    {"name": "PREFECT__CLOUD__API", "value": prefect.config.cloud.api},
                     {
                         "name": "PREFECT__CLOUD__AGENT__LABELS",
                         "value": "['aws', 'staging']",
diff --git a/tests/agent/test_k8s_agent.py b/tests/agent/test_k8s_agent.py
index a47de6f606..7bb1825bd1 100644
--- a/tests/agent/test_k8s_agent.py
+++ b/tests/agent/test_k8s_agent.py
@@ -1314,7 +1314,7 @@ class TestK8sAgentRunConfig:
         args = job["spec"]["template"]["spec"]["containers"][0]["args"]
         assert args == expected.split()
 
-    def test_generate_job_spec_environment_variables(self, tmpdir):
+    def test_generate_job_spec_environment_variables(self, tmpdir, backend):
         """Check that environment variables are set in precedence order
 
         - CUSTOM1 & CUSTOM2 are set on the template
@@ -1346,6 +1346,7 @@ class TestK8sAgentRunConfig:
         env_list = job["spec"]["template"]["spec"]["containers"][0]["env"]
         env = {item["name"]: item["value"] for item in env_list}
         assert env == {
+            "PREFECT__BACKEND": backend,
             "PREFECT__CLOUD__AGENT__LABELS": "[]",
             "PREFECT__CLOUD__API": prefect.config.cloud.api,
             "PREFECT__CLOUD__AUTH_TOKEN": prefect.config.cloud.agent.auth_token,
diff --git a/tests/agent/test_local_agent.py b/tests/agent/test_local_agent.py
index 88e2418bd8..a2b2783b87 100644
--- a/tests/agent/test_local_agent.py
+++ b/tests/agent/test_local_agent.py
@@ -8,6 +8,7 @@ from marshmallow.exceptions import ValidationError
 from testfixtures.popen import MockPopen
 from testfixtures import compare, LogCapture
 
+import prefect
 from prefect.agent.local import LocalAgent
 from prefect.storage import (
     Docker,
@@ -81,7 +82,7 @@ def test_local_agent_uses_ip_if_dockerdesktop_hostname(monkeypatch):
     assert "IP" in agent.labels
 
 
-def test_populate_env_vars(monkeypatch):
+def test_populate_env_vars(monkeypatch, backend):
     agent = LocalAgent()
 
     # The python path may be a single item and we want to ensure the correct separator
@@ -97,7 +98,8 @@ def test_populate_env_vars(monkeypatch):
     expected.update(
         {
             "PYTHONPATH": os.getcwd() + os.pathsep + expected.get("PYTHONPATH", ""),
-            "PREFECT__CLOUD__API": "https://api.prefect.io",
+            "PREFECT__BACKEND": backend,
+            "PREFECT__CLOUD__API": prefect.config.cloud.api,
             "PREFECT__CLOUD__AUTH_TOKEN": "TEST_TOKEN",
             "PREFECT__CLOUD__AGENT__LABELS": str(DEFAULT_AGENT_LABELS),
             "PREFECT__CONTEXT__FLOW_RUN_ID": "id",
diff --git a/tests/conftest.py b/tests/conftest.py
index 33fffdf3e7..6888da556f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -181,11 +181,22 @@ def cloud_api():
 @pytest.fixture()
 def server_api():
     with prefect.utilities.configuration.set_temporary_config(
-        {"cloud.api": "https:/localhost:4200", "backend": "server"}
+        {"cloud.api": "https://localhost:4200", "backend": "server"}
     ):
         yield
 
 
+@pytest.fixture(
+    params=[("cloud", "https://api.prefect.io"), ("server", "https://localhost:4200")]
+)
+def backend(request):
+    backend, api = request.param
+    with prefect.utilities.configuration.set_temporary_config(
+        {"cloud.api": api, "backend": backend}
+    ):
+        yield backend
+
+
 @pytest.fixture()
 def running_with_backend():
     with prefect.context({"running_with_backend": True}):