We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add Parameter Specification to run CLI command
## Current behavior
Currently the CLI command doesn't allow for the specification of parameters for a flow run



## Proposed behavior
Couple options we could take here:

Multiple --parameter NAME=VALUE options (e.g. each --parameter similar to how docker build does --build-arg)

Single --parameters {name: value} that takes a dictionary

Also we should have a file option where you can provide a json file

--- 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/cli/run.py`
Content:
```
1 import time
2 
3 import click
4 from tabulate import tabulate
5 
6 from prefect.client import Client
7 from prefect.utilities.graphql import EnumValue, with_args
8 
9 
10 @click.group(hidden=True)
11 def run():
12     """
13     Run Prefect flows.
14 
15     \b
16     Usage:
17         $ prefect run [STORAGE/PLATFORM]
18 
19     \b
20     Arguments:
21         cloud   Run flows in Prefect Cloud
22 
23     \b
24     Examples:
25         $ prefect run cloud --name Test-Flow --project My-Project
26         Flow Run ID: 2ba3rrfd-411c-4d99-bb2a-f64a6dea78f9
27 
28     \b
29         $ prefect run cloud --name Test-Flow --project My-Project --watch
30         Flow Run ID: 2ba3rrfd-411c-4d99-bb2a-f64a6dea78f9
31         Scheduled -> Submitted -> Running -> Success
32     """
33     pass
34 
35 
36 @run.command(hidden=True)
37 @click.option(
38     "--name", "-n", required=True, help="The name of a flow to run.", hidden=True
39 )
40 @click.option(
41     "--project",
42     "-p",
43     required=True,
44     help="The project that contains the flow.",
45     hidden=True,
46 )
47 @click.option("--version", "-v", type=int, help="A flow version to run.", hidden=True)
48 @click.option(
49     "--watch",
50     "-w",
51     is_flag=True,
52     help="Watch current state of the flow run.",
53     hidden=True,
54 )
55 @click.option(
56     "--logs", "-l", is_flag=True, help="Live logs of the flow run.", hidden=True
57 )
58 def cloud(name, project, version, watch, logs):
59     """
60     Run a deployed flow in Prefect Cloud.
61 
62     \b
63     Options:
64         --name, -n      TEXT    The name of a flow to run                                       [required]
65         --project, -p   TEXT    The name of a project that contains the flow                    [required]
66         --version, -v   INTEGER A flow version to run
67         --watch, -w             Watch current state of the flow run, stream output to stdout
68         --logs, -l              Get logs of the flow run, stream output to stdout
69     """
70 
71     if watch and logs:
72         click.secho(
73             "Streaming state and logs not currently supported together.", fg="red"
74         )
75         return
76 
77     query = {
78         "query": {
79             with_args(
80                 "flow",
81                 {
82                     "where": {
83                         "_and": {
84                             "name": {"_eq": name},
85                             "version": {"_eq": version},
86                             "project": {"name": {"_eq": project}},
87                         }
88                     },
89                     "order_by": {
90                         "name": EnumValue("asc"),
91                         "version": EnumValue("desc"),
92                     },
93                     "distinct_on": EnumValue("name"),
94                 },
95             ): {"id": True}
96         }
97     }
98 
99     client = Client()
100     result = client.graphql(query)
101 
102     flow_data = result.data.flow
103 
104     if flow_data:
105         flow_id = flow_data[0].id
106     else:
107         click.secho("{} not found".format(name), fg="red")
108         return
109 
110     flow_run_id = client.create_flow_run(flow_id=flow_id)
111     click.echo("Flow Run ID: {}".format(flow_run_id))
112 
113     if watch:
114         current_states = []
115         while True:
116             query = {
117                 "query": {
118                     with_args("flow_run_by_pk", {"id": flow_run_id}): {
119                         with_args(
120                             "states",
121                             {"order_by": {EnumValue("timestamp"): EnumValue("asc")}},
122                         ): {"state": True, "timestamp": True}
123                     }
124                 }
125             }
126 
127             result = client.graphql(query)
128 
129             # Filter through retrieved states and output in order
130             for state_index in result.data.flow_run_by_pk.states:
131                 state = state_index.state
132                 if state not in current_states:
133                     if state != "Success" and state != "Failed":
134                         click.echo("{} -> ".format(state), nl=False)
135                     else:
136                         click.echo(state)
137                         return
138 
139                     current_states.append(state)
140 
141             time.sleep(3)
142 
143     if logs:
144         all_logs = []
145 
146         log_query = {
147             with_args(
148                 "logs", {"order_by": {EnumValue("timestamp"): EnumValue("asc")}}
149             ): {"timestamp": True, "message": True, "level": True},
150             "start_time": True,
151         }
152 
153         query = {
154             "query": {
155                 with_args(
156                     "flow_run",
157                     {
158                         "where": {"id": {"_eq": flow_run_id}},
159                         "order_by": {EnumValue("start_time"): EnumValue("desc")},
160                     },
161                 ): log_query
162             }
163         }
164 
165         while True:
166             result = Client().graphql(query)
167 
168             flow_run = result.data.flow_run
169             if not flow_run:
170                 click.secho("{} not found".format(flow_run_id), fg="red")
171                 return
172 
173             new_run = flow_run[0]
174             logs = new_run.logs
175             output = []
176 
177             for i in logs:
178                 if [i.timestamp, i.level, i.message] not in all_logs:
179 
180                     if not len(all_logs):
181                         click.echo(
182                             tabulate(
183                                 [[i.timestamp, i.level, i.message]],
184                                 headers=["TIMESTAMP", "LEVEL", "MESSAGE"],
185                                 tablefmt="plain",
186                                 numalign="left",
187                                 stralign="left",
188                             )
189                         )
190                         all_logs.append([i.timestamp, i.level, i.message])
191                         continue
192 
193                     output.append([i.timestamp, i.level, i.message])
194                     all_logs.append([i.timestamp, i.level, i.message])
195 
196             if output:
197                 click.echo(
198                     tabulate(output, tablefmt="plain", numalign="left", stralign="left")
199                 )
200 
201             # Check if state is either Success or Failed, exit if it is
202             pk_query = {
203                 "query": {
204                     with_args("flow_run_by_pk", {"id": flow_run_id}): {"state": True}
205                 }
206             }
207             result = client.graphql(pk_query)
208 
209             if (
210                 result.data.flow_run_by_pk.state == "Success"
211                 or result.data.flow_run_by_pk.state == "Failed"
212             ):
213                 return
214 
215             time.sleep(3)
216
```
--- 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()

```

diff --git a/src/prefect/cli/run.py b/src/prefect/cli/run.py
--- a/src/prefect/cli/run.py
+++ b/src/prefect/cli/run.py
@@ -1,3 +1,4 @@
+import json
 import time
 
 import click
@@ -45,6 +46,16 @@
     hidden=True,
 )
 @click.option("--version", "-v", type=int, help="A flow version to run.", hidden=True)
+@click.option(
+    "--parameters-file",
+    "-pf",
+    help="A parameters JSON file.",
+    hidden=True,
+    type=click.Path(exists=True),
+)
+@click.option(
+    "--parameters-string", "--ps", help="A parameters JSON string.", hidden=True
+)
 @click.option(
     "--watch",
     "-w",
@@ -55,17 +66,29 @@
 @click.option(
     "--logs", "-l", is_flag=True, help="Live logs of the flow run.", hidden=True
 )
-def cloud(name, project, version, watch, logs):
+def cloud(name, project, version, parameters_file, parameters_string, watch, logs):
     """
     Run a deployed flow in Prefect Cloud.
 
     \b
     Options:
-        --name, -n      TEXT    The name of a flow to run                                       [required]
-        --project, -p   TEXT    The name of a project that contains the flow                    [required]
-        --version, -v   INTEGER A flow version to run
-        --watch, -w             Watch current state of the flow run, stream output to stdout
-        --logs, -l              Get logs of the flow run, stream output to stdout
+        --name, -n                  TEXT        The name of a flow to run                                       [required]
+        --project, -p               TEXT        The name of a project that contains the flow                    [required]
+        --version, -v               INTEGER     A flow version to run
+        --parameters-file, -pf      FILE PATH   A filepath of a JSON file containing parameters
+        --parameters-string, -ps    TEXT        A string of JSON parameters
+        --watch, -w                             Watch current state of the flow run, stream output to stdout
+        --logs, -l                              Get logs of the flow run, stream output to stdout
+
+    \b
+    If both `--parameters-file` and `--parameters-string` are provided then the values passed
+    in through the string will override the values provided from the file.
+
+    \b
+    e.g.
+    File contains:  {"a": 1, "b": 2}
+    String:         '{"a": 3}'
+    Parameters passed to the flow run: {"a": 3, "b": 2}
     """
 
     if watch and logs:
@@ -107,7 +130,20 @@
         click.secho("{} not found".format(name), fg="red")
         return
 
-    flow_run_id = client.create_flow_run(flow_id=flow_id)
+    # Load parameters from file if provided
+    file_params = {}
+    if parameters_file:
+        with open(parameters_file) as params_file:
+            file_params = json.load(params_file)
+
+    # Load parameters from string if provided
+    string_params = {}
+    if parameters_string:
+        string_params = json.loads(parameters_string)
+
+    flow_run_id = client.create_flow_run(
+        flow_id=flow_id, parameters={**file_params, **string_params}
+    )
     click.echo("Flow Run ID: {}".format(flow_run_id))
 
     if watch:

{"golden_diff": "diff --git a/src/prefect/cli/run.py b/src/prefect/cli/run.py\n--- a/src/prefect/cli/run.py\n+++ b/src/prefect/cli/run.py\n@@ -1,3 +1,4 @@\n+import json\n import time\n \n import click\n@@ -45,6 +46,16 @@\n     hidden=True,\n )\n @click.option(\"--version\", \"-v\", type=int, help=\"A flow version to run.\", hidden=True)\n+@click.option(\n+    \"--parameters-file\",\n+    \"-pf\",\n+    help=\"A parameters JSON file.\",\n+    hidden=True,\n+    type=click.Path(exists=True),\n+)\n+@click.option(\n+    \"--parameters-string\", \"--ps\", help=\"A parameters JSON string.\", hidden=True\n+)\n @click.option(\n     \"--watch\",\n     \"-w\",\n@@ -55,17 +66,29 @@\n @click.option(\n     \"--logs\", \"-l\", is_flag=True, help=\"Live logs of the flow run.\", hidden=True\n )\n-def cloud(name, project, version, watch, logs):\n+def cloud(name, project, version, parameters_file, parameters_string, watch, logs):\n     \"\"\"\n     Run a deployed flow in Prefect Cloud.\n \n     \\b\n     Options:\n-        --name, -n      TEXT    The name of a flow to run                                       [required]\n-        --project, -p   TEXT    The name of a project that contains the flow                    [required]\n-        --version, -v   INTEGER A flow version to run\n-        --watch, -w             Watch current state of the flow run, stream output to stdout\n-        --logs, -l              Get logs of the flow run, stream output to stdout\n+        --name, -n                  TEXT        The name of a flow to run                                       [required]\n+        --project, -p               TEXT        The name of a project that contains the flow                    [required]\n+        --version, -v               INTEGER     A flow version to run\n+        --parameters-file, -pf      FILE PATH   A filepath of a JSON file containing parameters\n+        --parameters-string, -ps    TEXT        A string of JSON parameters\n+        --watch, -w                             Watch current state of the flow run, stream output to stdout\n+        --logs, -l                              Get logs of the flow run, stream output to stdout\n+\n+    \\b\n+    If both `--parameters-file` and `--parameters-string` are provided then the values passed\n+    in through the string will override the values provided from the file.\n+\n+    \\b\n+    e.g.\n+    File contains:  {\"a\": 1, \"b\": 2}\n+    String:         '{\"a\": 3}'\n+    Parameters passed to the flow run: {\"a\": 3, \"b\": 2}\n     \"\"\"\n \n     if watch and logs:\n@@ -107,7 +130,20 @@\n         click.secho(\"{} not found\".format(name), fg=\"red\")\n         return\n \n-    flow_run_id = client.create_flow_run(flow_id=flow_id)\n+    # Load parameters from file if provided\n+    file_params = {}\n+    if parameters_file:\n+        with open(parameters_file) as params_file:\n+            file_params = json.load(params_file)\n+\n+    # Load parameters from string if provided\n+    string_params = {}\n+    if parameters_string:\n+        string_params = json.loads(parameters_string)\n+\n+    flow_run_id = client.create_flow_run(\n+        flow_id=flow_id, parameters={**file_params, **string_params}\n+    )\n     click.echo(\"Flow Run ID: {}\".format(flow_run_id))\n \n     if watch:\n", "issue": "Add Parameter Specification to run CLI command\n## Current behavior\r\nCurrently the CLI command doesn't allow for the specification of parameters for a flow run\r\n\r\n\r\n\r\n## Proposed behavior\r\nCouple options we could take here:\r\n\r\nMultiple --parameter NAME=VALUE options (e.g. each --parameter similar to how docker build does --build-arg)\r\n\r\nSingle --parameters {name: value} that takes a dictionary\r\n\r\nAlso we should have a file option where you can provide a json file\n", "before_files": [{"content": "import time\n\nimport click\nfrom tabulate import tabulate\n\nfrom prefect.client import Client\nfrom prefect.utilities.graphql import EnumValue, with_args\n\n\n@click.group(hidden=True)\ndef run():\n    \"\"\"\n    Run Prefect flows.\n\n    \\b\n    Usage:\n        $ prefect run [STORAGE/PLATFORM]\n\n    \\b\n    Arguments:\n        cloud   Run flows in Prefect Cloud\n\n    \\b\n    Examples:\n        $ prefect run cloud --name Test-Flow --project My-Project\n        Flow Run ID: 2ba3rrfd-411c-4d99-bb2a-f64a6dea78f9\n\n    \\b\n        $ prefect run cloud --name Test-Flow --project My-Project --watch\n        Flow Run ID: 2ba3rrfd-411c-4d99-bb2a-f64a6dea78f9\n        Scheduled -> Submitted -> Running -> Success\n    \"\"\"\n    pass\n\n\n@run.command(hidden=True)\n@click.option(\n    \"--name\", \"-n\", required=True, help=\"The name of a flow to run.\", hidden=True\n)\n@click.option(\n    \"--project\",\n    \"-p\",\n    required=True,\n    help=\"The project that contains the flow.\",\n    hidden=True,\n)\n@click.option(\"--version\", \"-v\", type=int, help=\"A flow version to run.\", hidden=True)\n@click.option(\n    \"--watch\",\n    \"-w\",\n    is_flag=True,\n    help=\"Watch current state of the flow run.\",\n    hidden=True,\n)\n@click.option(\n    \"--logs\", \"-l\", is_flag=True, help=\"Live logs of the flow run.\", hidden=True\n)\ndef cloud(name, project, version, watch, logs):\n    \"\"\"\n    Run a deployed flow in Prefect Cloud.\n\n    \\b\n    Options:\n        --name, -n      TEXT    The name of a flow to run                                       [required]\n        --project, -p   TEXT    The name of a project that contains the flow                    [required]\n        --version, -v   INTEGER A flow version to run\n        --watch, -w             Watch current state of the flow run, stream output to stdout\n        --logs, -l              Get logs of the flow run, stream output to stdout\n    \"\"\"\n\n    if watch and logs:\n        click.secho(\n            \"Streaming state and logs not currently supported together.\", fg=\"red\"\n        )\n        return\n\n    query = {\n        \"query\": {\n            with_args(\n                \"flow\",\n                {\n                    \"where\": {\n                        \"_and\": {\n                            \"name\": {\"_eq\": name},\n                            \"version\": {\"_eq\": version},\n                            \"project\": {\"name\": {\"_eq\": project}},\n                        }\n                    },\n                    \"order_by\": {\n                        \"name\": EnumValue(\"asc\"),\n                        \"version\": EnumValue(\"desc\"),\n                    },\n                    \"distinct_on\": EnumValue(\"name\"),\n                },\n            ): {\"id\": True}\n        }\n    }\n\n    client = Client()\n    result = client.graphql(query)\n\n    flow_data = result.data.flow\n\n    if flow_data:\n        flow_id = flow_data[0].id\n    else:\n        click.secho(\"{} not found\".format(name), fg=\"red\")\n        return\n\n    flow_run_id = client.create_flow_run(flow_id=flow_id)\n    click.echo(\"Flow Run ID: {}\".format(flow_run_id))\n\n    if watch:\n        current_states = []\n        while True:\n            query = {\n                \"query\": {\n                    with_args(\"flow_run_by_pk\", {\"id\": flow_run_id}): {\n                        with_args(\n                            \"states\",\n                            {\"order_by\": {EnumValue(\"timestamp\"): EnumValue(\"asc\")}},\n                        ): {\"state\": True, \"timestamp\": True}\n                    }\n                }\n            }\n\n            result = client.graphql(query)\n\n            # Filter through retrieved states and output in order\n            for state_index in result.data.flow_run_by_pk.states:\n                state = state_index.state\n                if state not in current_states:\n                    if state != \"Success\" and state != \"Failed\":\n                        click.echo(\"{} -> \".format(state), nl=False)\n                    else:\n                        click.echo(state)\n                        return\n\n                    current_states.append(state)\n\n            time.sleep(3)\n\n    if logs:\n        all_logs = []\n\n        log_query = {\n            with_args(\n                \"logs\", {\"order_by\": {EnumValue(\"timestamp\"): EnumValue(\"asc\")}}\n            ): {\"timestamp\": True, \"message\": True, \"level\": True},\n            \"start_time\": True,\n        }\n\n        query = {\n            \"query\": {\n                with_args(\n                    \"flow_run\",\n                    {\n                        \"where\": {\"id\": {\"_eq\": flow_run_id}},\n                        \"order_by\": {EnumValue(\"start_time\"): EnumValue(\"desc\")},\n                    },\n                ): log_query\n            }\n        }\n\n        while True:\n            result = Client().graphql(query)\n\n            flow_run = result.data.flow_run\n            if not flow_run:\n                click.secho(\"{} not found\".format(flow_run_id), fg=\"red\")\n                return\n\n            new_run = flow_run[0]\n            logs = new_run.logs\n            output = []\n\n            for i in logs:\n                if [i.timestamp, i.level, i.message] not in all_logs:\n\n                    if not len(all_logs):\n                        click.echo(\n                            tabulate(\n                                [[i.timestamp, i.level, i.message]],\n                                headers=[\"TIMESTAMP\", \"LEVEL\", \"MESSAGE\"],\n                                tablefmt=\"plain\",\n                                numalign=\"left\",\n                                stralign=\"left\",\n                            )\n                        )\n                        all_logs.append([i.timestamp, i.level, i.message])\n                        continue\n\n                    output.append([i.timestamp, i.level, i.message])\n                    all_logs.append([i.timestamp, i.level, i.message])\n\n            if output:\n                click.echo(\n                    tabulate(output, tablefmt=\"plain\", numalign=\"left\", stralign=\"left\")\n                )\n\n            # Check if state is either Success or Failed, exit if it is\n            pk_query = {\n                \"query\": {\n                    with_args(\"flow_run_by_pk\", {\"id\": flow_run_id}): {\"state\": True}\n                }\n            }\n            result = client.graphql(pk_query)\n\n            if (\n                result.data.flow_run_by_pk.state == \"Success\"\n                or result.data.flow_run_by_pk.state == \"Failed\"\n            ):\n                return\n\n            time.sleep(3)\n", "path": "src/prefect/cli/run.py"}], "after_files": [{"content": "import json\nimport time\n\nimport click\nfrom tabulate import tabulate\n\nfrom prefect.client import Client\nfrom prefect.utilities.graphql import EnumValue, with_args\n\n\n@click.group(hidden=True)\ndef run():\n    \"\"\"\n    Run Prefect flows.\n\n    \\b\n    Usage:\n        $ prefect run [STORAGE/PLATFORM]\n\n    \\b\n    Arguments:\n        cloud   Run flows in Prefect Cloud\n\n    \\b\n    Examples:\n        $ prefect run cloud --name Test-Flow --project My-Project\n        Flow Run ID: 2ba3rrfd-411c-4d99-bb2a-f64a6dea78f9\n\n    \\b\n        $ prefect run cloud --name Test-Flow --project My-Project --watch\n        Flow Run ID: 2ba3rrfd-411c-4d99-bb2a-f64a6dea78f9\n        Scheduled -> Submitted -> Running -> Success\n    \"\"\"\n    pass\n\n\n@run.command(hidden=True)\n@click.option(\n    \"--name\", \"-n\", required=True, help=\"The name of a flow to run.\", hidden=True\n)\n@click.option(\n    \"--project\",\n    \"-p\",\n    required=True,\n    help=\"The project that contains the flow.\",\n    hidden=True,\n)\n@click.option(\"--version\", \"-v\", type=int, help=\"A flow version to run.\", hidden=True)\n@click.option(\n    \"--parameters-file\",\n    \"-pf\",\n    help=\"A parameters JSON file.\",\n    hidden=True,\n    type=click.Path(exists=True),\n)\n@click.option(\n    \"--parameters-string\", \"--ps\", help=\"A parameters JSON string.\", hidden=True\n)\n@click.option(\n    \"--watch\",\n    \"-w\",\n    is_flag=True,\n    help=\"Watch current state of the flow run.\",\n    hidden=True,\n)\n@click.option(\n    \"--logs\", \"-l\", is_flag=True, help=\"Live logs of the flow run.\", hidden=True\n)\ndef cloud(name, project, version, parameters_file, parameters_string, watch, logs):\n    \"\"\"\n    Run a deployed flow in Prefect Cloud.\n\n    \\b\n    Options:\n        --name, -n                  TEXT        The name of a flow to run                                       [required]\n        --project, -p               TEXT        The name of a project that contains the flow                    [required]\n        --version, -v               INTEGER     A flow version to run\n        --parameters-file, -pf      FILE PATH   A filepath of a JSON file containing parameters\n        --parameters-string, -ps    TEXT        A string of JSON parameters\n        --watch, -w                             Watch current state of the flow run, stream output to stdout\n        --logs, -l                              Get logs of the flow run, stream output to stdout\n\n    \\b\n    If both `--parameters-file` and `--parameters-string` are provided then the values passed\n    in through the string will override the values provided from the file.\n\n    \\b\n    e.g.\n    File contains:  {\"a\": 1, \"b\": 2}\n    String:         '{\"a\": 3}'\n    Parameters passed to the flow run: {\"a\": 3, \"b\": 2}\n    \"\"\"\n\n    if watch and logs:\n        click.secho(\n            \"Streaming state and logs not currently supported together.\", fg=\"red\"\n        )\n        return\n\n    query = {\n        \"query\": {\n            with_args(\n                \"flow\",\n                {\n                    \"where\": {\n                        \"_and\": {\n                            \"name\": {\"_eq\": name},\n                            \"version\": {\"_eq\": version},\n                            \"project\": {\"name\": {\"_eq\": project}},\n                        }\n                    },\n                    \"order_by\": {\n                        \"name\": EnumValue(\"asc\"),\n                        \"version\": EnumValue(\"desc\"),\n                    },\n                    \"distinct_on\": EnumValue(\"name\"),\n                },\n            ): {\"id\": True}\n        }\n    }\n\n    client = Client()\n    result = client.graphql(query)\n\n    flow_data = result.data.flow\n\n    if flow_data:\n        flow_id = flow_data[0].id\n    else:\n        click.secho(\"{} not found\".format(name), fg=\"red\")\n        return\n\n    # Load parameters from file if provided\n    file_params = {}\n    if parameters_file:\n        with open(parameters_file) as params_file:\n            file_params = json.load(params_file)\n\n    # Load parameters from string if provided\n    string_params = {}\n    if parameters_string:\n        string_params = json.loads(parameters_string)\n\n    flow_run_id = client.create_flow_run(\n        flow_id=flow_id, parameters={**file_params, **string_params}\n    )\n    click.echo(\"Flow Run ID: {}\".format(flow_run_id))\n\n    if watch:\n        current_states = []\n        while True:\n            query = {\n                \"query\": {\n                    with_args(\"flow_run_by_pk\", {\"id\": flow_run_id}): {\n                        with_args(\n                            \"states\",\n                            {\"order_by\": {EnumValue(\"timestamp\"): EnumValue(\"asc\")}},\n                        ): {\"state\": True, \"timestamp\": True}\n                    }\n                }\n            }\n\n            result = client.graphql(query)\n\n            # Filter through retrieved states and output in order\n            for state_index in result.data.flow_run_by_pk.states:\n                state = state_index.state\n                if state not in current_states:\n                    if state != \"Success\" and state != \"Failed\":\n                        click.echo(\"{} -> \".format(state), nl=False)\n                    else:\n                        click.echo(state)\n                        return\n\n                    current_states.append(state)\n\n            time.sleep(3)\n\n    if logs:\n        all_logs = []\n\n        log_query = {\n            with_args(\n                \"logs\", {\"order_by\": {EnumValue(\"timestamp\"): EnumValue(\"asc\")}}\n            ): {\"timestamp\": True, \"message\": True, \"level\": True},\n            \"start_time\": True,\n        }\n\n        query = {\n            \"query\": {\n                with_args(\n                    \"flow_run\",\n                    {\n                        \"where\": {\"id\": {\"_eq\": flow_run_id}},\n                        \"order_by\": {EnumValue(\"start_time\"): EnumValue(\"desc\")},\n                    },\n                ): log_query\n            }\n        }\n\n        while True:\n            result = Client().graphql(query)\n\n            flow_run = result.data.flow_run\n            if not flow_run:\n                click.secho(\"{} not found\".format(flow_run_id), fg=\"red\")\n                return\n\n            new_run = flow_run[0]\n            logs = new_run.logs\n            output = []\n\n            for i in logs:\n                if [i.timestamp, i.level, i.message] not in all_logs:\n\n                    if not len(all_logs):\n                        click.echo(\n                            tabulate(\n                                [[i.timestamp, i.level, i.message]],\n                                headers=[\"TIMESTAMP\", \"LEVEL\", \"MESSAGE\"],\n                                tablefmt=\"plain\",\n                                numalign=\"left\",\n                                stralign=\"left\",\n                            )\n                        )\n                        all_logs.append([i.timestamp, i.level, i.message])\n                        continue\n\n                    output.append([i.timestamp, i.level, i.message])\n                    all_logs.append([i.timestamp, i.level, i.message])\n\n            if output:\n                click.echo(\n                    tabulate(output, tablefmt=\"plain\", numalign=\"left\", stralign=\"left\")\n                )\n\n            # Check if state is either Success or Failed, exit if it is\n            pk_query = {\n                \"query\": {\n                    with_args(\"flow_run_by_pk\", {\"id\": flow_run_id}): {\"state\": True}\n                }\n            }\n            result = client.graphql(pk_query)\n\n            if (\n                result.data.flow_run_by_pk.state == \"Success\"\n                or result.data.flow_run_by_pk.state == \"Failed\"\n            ):\n                return\n\n            time.sleep(3)\n", "path": "src/prefect/cli/run.py"}]}