prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  When a flow has no storage attached, the agent should mark the run as failed
  
  It appears that if the Agent picks up a run with no storage, it will mark it as submitted. This run will then exist in limbo forever, as it can never be run. The agent should put it in a failed state if it can recognize it'll never run.
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `src/prefect/agent/local/agent.py`
  Content:
  ```
  1 from sys import platform
  2 from typing import Iterable
  3 
  4 import docker
  5 
  6 from prefect import config, context
  7 from prefect.agent import Agent
  8 from prefect.environments.storage import Docker
  9 from prefect.serialization.storage import StorageSchema
  10 from prefect.utilities.graphql import GraphQLResult
  11 
  12 
  13 class LocalAgent(Agent):
  14     """
  15     Agent which deploys flow runs locally as Docker containers. Information on using the
  16     Local Agent can be found at https://docs.prefect.io/cloud/agent/local.html
  17 
  18     Args:
  19         - name (str, optional): An optional name to give this agent. Can also be set through
  20             the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to "agent"
  21         - labels (List[str], optional): a list of labels, which are arbitrary string identifiers used by Prefect
  22             Agents when polling for work
  23         - base_url (str, optional): URL for a Docker daemon server. Defaults to
  24             `unix:///var/run/docker.sock` however other hosts such as
  25             `tcp://0.0.0.0:2375` can be provided
  26         - no_pull (bool, optional): Flag on whether or not to pull flow images.
  27             Defaults to `False` if not provided here or in context.
  28     """
  29 
  30     def __init__(
  31         self,
  32         name: str = None,
  33         labels: Iterable[str] = None,
  34         base_url: str = None,
  35         no_pull: bool = None,
  36     ) -> None:
  37         super().__init__(name=name, labels=labels)
  38 
  39         if platform == "win32":
  40             default_url = "npipe:////./pipe/docker_engine"
  41         else:
  42             default_url = "unix://var/run/docker.sock"
  43         self.logger.debug(
  44             "Platform {} and default docker daemon {}".format(platform, default_url)
  45         )
  46 
  47         # Determine Daemon URL
  48         self.base_url = base_url or context.get("base_url", default_url)
  49         self.logger.debug("Base docker daemon url {}".format(self.base_url))
  50 
  51         # Determine pull specification
  52         self.no_pull = no_pull or context.get("no_pull", False)
  53         self.logger.debug("no_pull set to {}".format(self.no_pull))
  54 
  55         self.docker_client = docker.APIClient(base_url=self.base_url, version="auto")
  56 
  57         # Ping Docker daemon for connection issues
  58         try:
  59             self.logger.debug("Pinging docker daemon")
  60             self.docker_client.ping()
  61         except Exception as exc:
  62             self.logger.exception(
  63                 "Issue connecting to the Docker daemon. Make sure it is running."
  64             )
  65             raise exc
  66 
  67     def deploy_flows(self, flow_runs: list) -> None:
  68         """
  69         Deploy flow runs on your local machine as Docker containers
  70 
  71         Args:
  72             - flow_runs (list): A list of GraphQLResult flow run objects
  73         """
  74         for flow_run in flow_runs:
  75             self.logger.info(
  76                 "Deploying flow run {}".format(flow_run.id)  # type: ignore
  77             )
  78 
  79             storage = StorageSchema().load(flow_run.flow.storage)
  80             if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):
  81                 self.logger.error(
  82                     "Storage for flow run {} is not of type Docker.".format(flow_run.id)
  83                 )
  84                 continue
  85 
  86             env_vars = self.populate_env_vars(flow_run=flow_run)
  87 
  88             if not self.no_pull and storage.registry_url:
  89                 self.logger.info("Pulling image {}...".format(storage.name))
  90                 try:
  91                     pull_output = self.docker_client.pull(
  92                         storage.name, stream=True, decode=True
  93                     )
  94                     for line in pull_output:
  95                         self.logger.debug(line)
  96                     self.logger.info(
  97                         "Successfully pulled image {}...".format(storage.name)
  98                     )
  99                 except docker.errors.APIError as exc:
  100                     self.logger.error("Issue pulling image {}".format(storage.name))
  101 
  102             # Create a container
  103             self.logger.debug("Creating Docker container {}".format(storage.name))
  104             container = self.docker_client.create_container(
  105                 storage.name, command="prefect execute cloud-flow", environment=env_vars
  106             )
  107 
  108             # Start the container
  109             self.logger.debug(
  110                 "Starting Docker container with ID {}".format(container.get("Id"))
  111             )
  112             self.docker_client.start(container=container.get("Id"))
  113 
  114     def populate_env_vars(self, flow_run: GraphQLResult) -> dict:
  115         """
  116         Populate metadata and variables in the environment variables for a flow run
  117 
  118         Args:
  119             - flow_run (GraphQLResult): A flow run object
  120 
  121         Returns:
  122             - dict: a dictionary representing the populated environment variables
  123         """
  124         return {
  125             "PREFECT__CLOUD__API": config.cloud.api,
  126             "PREFECT__CLOUD__AUTH_TOKEN": config.cloud.agent.auth_token,
  127             "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
  128             "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,  # type: ignore
  129             "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
  130             "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
  131             "PREFECT__LOGGING__LEVEL": "DEBUG",
  132             "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
  133             "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
  134         }
  135 
  136 
  137 if __name__ == "__main__":
  138     LocalAgent().start()
  139
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py
  --- a/src/prefect/agent/local/agent.py
  +++ b/src/prefect/agent/local/agent.py
  @@ -5,6 +5,7 @@
   
   from prefect import config, context
   from prefect.agent import Agent
  +from prefect.engine.state import Failed
   from prefect.environments.storage import Docker
   from prefect.serialization.storage import StorageSchema
   from prefect.utilities.graphql import GraphQLResult
  @@ -78,9 +79,14 @@
   
               storage = StorageSchema().load(flow_run.flow.storage)
               if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):
  -                self.logger.error(
  -                    "Storage for flow run {} is not of type Docker.".format(flow_run.id)
  +                msg = "Storage for flow run {} is not of type Docker.".format(
  +                    flow_run.id
                   )
  +                state_msg = "Agent {} failed to run flow: ".format(self.name) + msg
  +                self.client.set_flow_run_state(
  +                    flow_run.id, version=flow_run.version, state=Failed(state_msg)
  +                )
  +                self.logger.error(msg)
                   continue
   
               env_vars = self.populate_env_vars(flow_run=flow_run)
  @@ -97,7 +103,15 @@
                           "Successfully pulled image {}...".format(storage.name)
                       )
                   except docker.errors.APIError as exc:
  -                    self.logger.error("Issue pulling image {}".format(storage.name))
  +                    msg = "Issue pulling image {}".format(storage.name)
  +                    state_msg = (
  +                        "Agent {} failed to pull image for flow: ".format(self.name)
  +                        + msg
  +                    )
  +                    self.client.set_flow_run_state(
  +                        flow_run.id, version=flow_run.version, state=Failed(msg)
  +                    )
  +                    self.logger.error(msg)
   
               # Create a container
               self.logger.debug("Creating Docker container {}".format(storage.name))
  
verification_info: {"golden_diff": "diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py\n--- a/src/prefect/agent/local/agent.py\n+++ b/src/prefect/agent/local/agent.py\n@@ -5,6 +5,7 @@\n \n from prefect import config, context\n from prefect.agent import Agent\n+from prefect.engine.state import Failed\n from prefect.environments.storage import Docker\n from prefect.serialization.storage import StorageSchema\n from prefect.utilities.graphql import GraphQLResult\n@@ -78,9 +79,14 @@\n \n             storage = StorageSchema().load(flow_run.flow.storage)\n             if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):\n-                self.logger.error(\n-                    \"Storage for flow run {} is not of type Docker.\".format(flow_run.id)\n+                msg = \"Storage for flow run {} is not of type Docker.\".format(\n+                    flow_run.id\n                 )\n+                state_msg = \"Agent {} failed to run flow: \".format(self.name) + msg\n+                self.client.set_flow_run_state(\n+                    flow_run.id, version=flow_run.version, state=Failed(state_msg)\n+                )\n+                self.logger.error(msg)\n                 continue\n \n             env_vars = self.populate_env_vars(flow_run=flow_run)\n@@ -97,7 +103,15 @@\n                         \"Successfully pulled image {}...\".format(storage.name)\n                     )\n                 except docker.errors.APIError as exc:\n-                    self.logger.error(\"Issue pulling image {}\".format(storage.name))\n+                    msg = \"Issue pulling image {}\".format(storage.name)\n+                    state_msg = (\n+                        \"Agent {} failed to pull image for flow: \".format(self.name)\n+                        + msg\n+                    )\n+                    self.client.set_flow_run_state(\n+                        flow_run.id, version=flow_run.version, state=Failed(msg)\n+                    )\n+                    self.logger.error(msg)\n \n             # Create a container\n             self.logger.debug(\"Creating Docker container {}\".format(storage.name))\n", "issue": "When a flow has no storage attached, the agent should mark the run as failed\n\r\nIt appears that if the Agent picks up a run with no storage, it will mark it as submitted. This run will then exist in limbo forever, as it can never be run. The agent should put it in a failed state if it can recognize it'll never run.\n", "before_files": [{"content": "from sys import platform\nfrom typing import Iterable\n\nimport docker\n\nfrom prefect import config, context\nfrom prefect.agent import Agent\nfrom prefect.environments.storage import Docker\nfrom prefect.serialization.storage import StorageSchema\nfrom prefect.utilities.graphql import GraphQLResult\n\n\nclass LocalAgent(Agent):\n    \"\"\"\n    Agent which deploys flow runs locally as Docker containers. Information on using the\n    Local Agent can be found at https://docs.prefect.io/cloud/agent/local.html\n\n    Args:\n        - name (str, optional): An optional name to give this agent. Can also be set through\n            the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to \"agent\"\n        - labels (List[str], optional): a list of labels, which are arbitrary string identifiers used by Prefect\n            Agents when polling for work\n        - base_url (str, optional): URL for a Docker daemon server. Defaults to\n            `unix:///var/run/docker.sock` however other hosts such as\n            `tcp://0.0.0.0:2375` can be provided\n        - no_pull (bool, optional): Flag on whether or not to pull flow images.\n            Defaults to `False` if not provided here or in context.\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str = None,\n        labels: Iterable[str] = None,\n        base_url: str = None,\n        no_pull: bool = None,\n    ) -> None:\n        super().__init__(name=name, labels=labels)\n\n        if platform == \"win32\":\n            default_url = \"npipe:////./pipe/docker_engine\"\n        else:\n            default_url = \"unix://var/run/docker.sock\"\n        self.logger.debug(\n            \"Platform {} and default docker daemon {}\".format(platform, default_url)\n        )\n\n        # Determine Daemon URL\n        self.base_url = base_url or context.get(\"base_url\", default_url)\n        self.logger.debug(\"Base docker daemon url {}\".format(self.base_url))\n\n        # Determine pull specification\n        self.no_pull = no_pull or context.get(\"no_pull\", False)\n        self.logger.debug(\"no_pull set to {}\".format(self.no_pull))\n\n        self.docker_client = docker.APIClient(base_url=self.base_url, version=\"auto\")\n\n        # Ping Docker daemon for connection issues\n        try:\n            self.logger.debug(\"Pinging docker daemon\")\n            self.docker_client.ping()\n        except Exception as exc:\n            self.logger.exception(\n                \"Issue connecting to the Docker daemon. Make sure it is running.\"\n            )\n            raise exc\n\n    def deploy_flows(self, flow_runs: list) -> None:\n        \"\"\"\n        Deploy flow runs on your local machine as Docker containers\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        for flow_run in flow_runs:\n            self.logger.info(\n                \"Deploying flow run {}\".format(flow_run.id)  # type: ignore\n            )\n\n            storage = StorageSchema().load(flow_run.flow.storage)\n            if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):\n                self.logger.error(\n                    \"Storage for flow run {} is not of type Docker.\".format(flow_run.id)\n                )\n                continue\n\n            env_vars = self.populate_env_vars(flow_run=flow_run)\n\n            if not self.no_pull and storage.registry_url:\n                self.logger.info(\"Pulling image {}...\".format(storage.name))\n                try:\n                    pull_output = self.docker_client.pull(\n                        storage.name, stream=True, decode=True\n                    )\n                    for line in pull_output:\n                        self.logger.debug(line)\n                    self.logger.info(\n                        \"Successfully pulled image {}...\".format(storage.name)\n                    )\n                except docker.errors.APIError as exc:\n                    self.logger.error(\"Issue pulling image {}\".format(storage.name))\n\n            # Create a container\n            self.logger.debug(\"Creating Docker container {}\".format(storage.name))\n            container = self.docker_client.create_container(\n                storage.name, command=\"prefect execute cloud-flow\", environment=env_vars\n            )\n\n            # Start the container\n            self.logger.debug(\n                \"Starting Docker container with ID {}\".format(container.get(\"Id\"))\n            )\n            self.docker_client.start(container=container.get(\"Id\"))\n\n    def populate_env_vars(self, flow_run: GraphQLResult) -> dict:\n        \"\"\"\n        Populate metadata and variables in the environment variables for a flow run\n\n        Args:\n            - flow_run (GraphQLResult): A flow run object\n\n        Returns:\n            - dict: a dictionary representing the populated environment variables\n        \"\"\"\n        return {\n            \"PREFECT__CLOUD__API\": config.cloud.api,\n            \"PREFECT__CLOUD__AUTH_TOKEN\": config.cloud.agent.auth_token,\n            \"PREFECT__CLOUD__AGENT__LABELS\": str(self.labels),\n            \"PREFECT__CONTEXT__FLOW_RUN_ID\": flow_run.id,  # type: ignore\n            \"PREFECT__CLOUD__USE_LOCAL_SECRETS\": \"false\",\n            \"PREFECT__LOGGING__LOG_TO_CLOUD\": \"true\",\n            \"PREFECT__LOGGING__LEVEL\": \"DEBUG\",\n            \"PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudFlowRunner\",\n            \"PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudTaskRunner\",\n        }\n\n\nif __name__ == \"__main__\":\n    LocalAgent().start()\n", "path": "src/prefect/agent/local/agent.py"}], "after_files": [{"content": "from sys import platform\nfrom typing import Iterable\n\nimport docker\n\nfrom prefect import config, context\nfrom prefect.agent import Agent\nfrom prefect.engine.state import Failed\nfrom prefect.environments.storage import Docker\nfrom prefect.serialization.storage import StorageSchema\nfrom prefect.utilities.graphql import GraphQLResult\n\n\nclass LocalAgent(Agent):\n    \"\"\"\n    Agent which deploys flow runs locally as Docker containers. Information on using the\n    Local Agent can be found at https://docs.prefect.io/cloud/agent/local.html\n\n    Args:\n        - name (str, optional): An optional name to give this agent. Can also be set through\n            the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to \"agent\"\n        - labels (List[str], optional): a list of labels, which are arbitrary string identifiers used by Prefect\n            Agents when polling for work\n        - base_url (str, optional): URL for a Docker daemon server. Defaults to\n            `unix:///var/run/docker.sock` however other hosts such as\n            `tcp://0.0.0.0:2375` can be provided\n        - no_pull (bool, optional): Flag on whether or not to pull flow images.\n            Defaults to `False` if not provided here or in context.\n    \"\"\"\n\n    def __init__(\n        self,\n        name: str = None,\n        labels: Iterable[str] = None,\n        base_url: str = None,\n        no_pull: bool = None,\n    ) -> None:\n        super().__init__(name=name, labels=labels)\n\n        if platform == \"win32\":\n            default_url = \"npipe:////./pipe/docker_engine\"\n        else:\n            default_url = \"unix://var/run/docker.sock\"\n        self.logger.debug(\n            \"Platform {} and default docker daemon {}\".format(platform, default_url)\n        )\n\n        # Determine Daemon URL\n        self.base_url = base_url or context.get(\"base_url\", default_url)\n        self.logger.debug(\"Base docker daemon url {}\".format(self.base_url))\n\n        # Determine pull specification\n        self.no_pull = no_pull or context.get(\"no_pull\", False)\n        self.logger.debug(\"no_pull set to {}\".format(self.no_pull))\n\n        self.docker_client = docker.APIClient(base_url=self.base_url, version=\"auto\")\n\n        # Ping Docker daemon for connection issues\n        try:\n            self.logger.debug(\"Pinging docker daemon\")\n            self.docker_client.ping()\n        except Exception as exc:\n            self.logger.exception(\n                \"Issue connecting to the Docker daemon. Make sure it is running.\"\n            )\n            raise exc\n\n    def deploy_flows(self, flow_runs: list) -> None:\n        \"\"\"\n        Deploy flow runs on your local machine as Docker containers\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        for flow_run in flow_runs:\n            self.logger.info(\n                \"Deploying flow run {}\".format(flow_run.id)  # type: ignore\n            )\n\n            storage = StorageSchema().load(flow_run.flow.storage)\n            if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):\n                msg = \"Storage for flow run {} is not of type Docker.\".format(\n                    flow_run.id\n                )\n                state_msg = \"Agent {} failed to run flow: \".format(self.name) + msg\n                self.client.set_flow_run_state(\n                    flow_run.id, version=flow_run.version, state=Failed(state_msg)\n                )\n                self.logger.error(msg)\n                continue\n\n            env_vars = self.populate_env_vars(flow_run=flow_run)\n\n            if not self.no_pull and storage.registry_url:\n                self.logger.info(\"Pulling image {}...\".format(storage.name))\n                try:\n                    pull_output = self.docker_client.pull(\n                        storage.name, stream=True, decode=True\n                    )\n                    for line in pull_output:\n                        self.logger.debug(line)\n                    self.logger.info(\n                        \"Successfully pulled image {}...\".format(storage.name)\n                    )\n                except docker.errors.APIError as exc:\n                    msg = \"Issue pulling image {}\".format(storage.name)\n                    state_msg = (\n                        \"Agent {} failed to pull image for flow: \".format(self.name)\n                        + msg\n                    )\n                    self.client.set_flow_run_state(\n                        flow_run.id, version=flow_run.version, state=Failed(msg)\n                    )\n                    self.logger.error(msg)\n\n            # Create a container\n            self.logger.debug(\"Creating Docker container {}\".format(storage.name))\n            container = self.docker_client.create_container(\n                storage.name, command=\"prefect execute cloud-flow\", environment=env_vars\n            )\n\n            # Start the container\n            self.logger.debug(\n                \"Starting Docker container with ID {}\".format(container.get(\"Id\"))\n            )\n            self.docker_client.start(container=container.get(\"Id\"))\n\n    def populate_env_vars(self, flow_run: GraphQLResult) -> dict:\n        \"\"\"\n        Populate metadata and variables in the environment variables for a flow run\n\n        Args:\n            - flow_run (GraphQLResult): A flow run object\n\n        Returns:\n            - dict: a dictionary representing the populated environment variables\n        \"\"\"\n        return {\n            \"PREFECT__CLOUD__API\": config.cloud.api,\n            \"PREFECT__CLOUD__AUTH_TOKEN\": config.cloud.agent.auth_token,\n            \"PREFECT__CLOUD__AGENT__LABELS\": str(self.labels),\n            \"PREFECT__CONTEXT__FLOW_RUN_ID\": flow_run.id,  # type: ignore\n            \"PREFECT__CLOUD__USE_LOCAL_SECRETS\": \"false\",\n            \"PREFECT__LOGGING__LOG_TO_CLOUD\": \"true\",\n            \"PREFECT__LOGGING__LEVEL\": \"DEBUG\",\n            \"PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudFlowRunner\",\n            \"PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudTaskRunner\",\n        }\n\n\nif __name__ == \"__main__\":\n    LocalAgent().start()\n", "path": "src/prefect/agent/local/agent.py"}]}