We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
AttributeError: 'ClientPlayback' object has no attribute 'options'
#### Problem Description
Client playback does not seem to work anymore, since 10.2.0. 
On 10.1.6 it still works.

I am getting the following error:
>Traceback (most recent call last):
  File "mitmproxy/addons/clientplayback.py", line 177, in playback
AttributeError: 'ClientPlayback' object has no attribute 'options'

#### Steps to reproduce the behavior:
1. Save a recorded flow
2. Replay it using `mitmdump -C stored.flow` 
3. See error

#### System Information
Mitmproxy: 10.2.1 binary
Python:    3.12.1
OpenSSL:   OpenSSL 3.1.4 24 Oct 2023
Platform:  macOS-14.2.1-x86_64-i386-64bit


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mitmproxy/addons/clientplayback.py`
Content:
```
1 from __future__ import annotations
2 
3 import asyncio
4 import logging
5 import time
6 from collections.abc import Sequence
7 from types import TracebackType
8 from typing import cast
9 from typing import Literal
10 
11 import mitmproxy.types
12 from mitmproxy import command
13 from mitmproxy import ctx
14 from mitmproxy import exceptions
15 from mitmproxy import flow
16 from mitmproxy import http
17 from mitmproxy import io
18 from mitmproxy.connection import ConnectionState
19 from mitmproxy.connection import Server
20 from mitmproxy.hooks import UpdateHook
21 from mitmproxy.log import ALERT
22 from mitmproxy.options import Options
23 from mitmproxy.proxy import commands
24 from mitmproxy.proxy import events
25 from mitmproxy.proxy import layers
26 from mitmproxy.proxy import server
27 from mitmproxy.proxy.context import Context
28 from mitmproxy.proxy.layer import CommandGenerator
29 from mitmproxy.proxy.layers.http import HTTPMode
30 from mitmproxy.proxy.mode_specs import UpstreamMode
31 from mitmproxy.utils import asyncio_utils
32 
33 logger = logging.getLogger(__name__)
34 
35 
36 class MockServer(layers.http.HttpConnection):
37     """
38     A mock HTTP "server" that just pretends it received a full HTTP request,
39     which is then processed by the proxy core.
40     """
41 
42     flow: http.HTTPFlow
43 
44     def __init__(self, flow: http.HTTPFlow, context: Context):
45         super().__init__(context, context.client)
46         self.flow = flow
47 
48     def _handle_event(self, event: events.Event) -> CommandGenerator[None]:
49         if isinstance(event, events.Start):
50             content = self.flow.request.raw_content
51             self.flow.request.timestamp_start = (
52                 self.flow.request.timestamp_end
53             ) = time.time()
54             yield layers.http.ReceiveHttp(
55                 layers.http.RequestHeaders(
56                     1,
57                     self.flow.request,
58                     end_stream=not (content or self.flow.request.trailers),
59                     replay_flow=self.flow,
60                 )
61             )
62             if content:
63                 yield layers.http.ReceiveHttp(layers.http.RequestData(1, content))
64             if self.flow.request.trailers:  # pragma: no cover
65                 # TODO: Cover this once we support HTTP/1 trailers.
66                 yield layers.http.ReceiveHttp(
67                     layers.http.RequestTrailers(1, self.flow.request.trailers)
68                 )
69             yield layers.http.ReceiveHttp(layers.http.RequestEndOfMessage(1))
70         elif isinstance(
71             event,
72             (
73                 layers.http.ResponseHeaders,
74                 layers.http.ResponseData,
75                 layers.http.ResponseTrailers,
76                 layers.http.ResponseEndOfMessage,
77                 layers.http.ResponseProtocolError,
78             ),
79         ):
80             pass
81         else:  # pragma: no cover
82             logger.warning(f"Unexpected event during replay: {event}")
83 
84 
85 class ReplayHandler(server.ConnectionHandler):
86     layer: layers.HttpLayer
87 
88     def __init__(self, flow: http.HTTPFlow, options: Options) -> None:
89         client = flow.client_conn.copy()
90         client.state = ConnectionState.OPEN
91 
92         context = Context(client, options)
93         context.server = Server(address=(flow.request.host, flow.request.port))
94         if flow.request.scheme == "https":
95             context.server.tls = True
96             context.server.sni = flow.request.pretty_host
97         if options.mode and options.mode[0].startswith("upstream:"):
98             mode = UpstreamMode.parse(options.mode[0])
99             assert isinstance(mode, UpstreamMode)  # remove once mypy supports Self.
100             context.server.via = flow.server_conn.via = (mode.scheme, mode.address)
101 
102         super().__init__(context)
103 
104         if options.mode and options.mode[0].startswith("upstream:"):
105             self.layer = layers.HttpLayer(context, HTTPMode.upstream)
106         else:
107             self.layer = layers.HttpLayer(context, HTTPMode.transparent)
108         self.layer.connections[client] = MockServer(flow, context.fork())
109         self.flow = flow
110         self.done = asyncio.Event()
111 
112     async def replay(self) -> None:
113         self.server_event(events.Start())
114         await self.done.wait()
115 
116     def log(
117         self,
118         message: str,
119         level: int = logging.INFO,
120         exc_info: Literal[True]
121         | tuple[type[BaseException] | None, BaseException | None, TracebackType | None]
122         | None = None,
123     ) -> None:
124         assert isinstance(level, int)
125         logger.log(level=level, msg=f"[replay] {message}")
126 
127     async def handle_hook(self, hook: commands.StartHook) -> None:
128         (data,) = hook.args()
129         await ctx.master.addons.handle_lifecycle(hook)
130         if isinstance(data, flow.Flow):
131             await data.wait_for_resume()
132         if isinstance(hook, (layers.http.HttpResponseHook, layers.http.HttpErrorHook)):
133             if self.transports:
134                 # close server connections
135                 for x in self.transports.values():
136                     if x.handler:
137                         x.handler.cancel()
138                 await asyncio.wait(
139                     [x.handler for x in self.transports.values() if x.handler]
140                 )
141             # signal completion
142             self.done.set()
143 
144 
145 class ClientPlayback:
146     playback_task: asyncio.Task | None = None
147     inflight: http.HTTPFlow | None
148     queue: asyncio.Queue
149     options: Options
150     replay_tasks: set[asyncio.Task]
151 
152     def __init__(self):
153         self.queue = asyncio.Queue()
154         self.inflight = None
155         self.task = None
156         self.replay_tasks = set()
157 
158     def running(self):
159         self.playback_task = asyncio_utils.create_task(
160             self.playback(), name="client playback"
161         )
162         self.options = ctx.options
163 
164     async def done(self):
165         if self.playback_task:
166             self.playback_task.cancel()
167             try:
168                 await self.playback_task
169             except asyncio.CancelledError:
170                 pass
171 
172     async def playback(self):
173         while True:
174             self.inflight = await self.queue.get()
175             try:
176                 assert self.inflight
177                 h = ReplayHandler(self.inflight, self.options)
178                 if ctx.options.client_replay_concurrency == -1:
179                     t = asyncio_utils.create_task(
180                         h.replay(), name="client playback awaiting response"
181                     )
182                     # keep a reference so this is not garbage collected
183                     self.replay_tasks.add(t)
184                     t.add_done_callback(self.replay_tasks.remove)
185                 else:
186                     await h.replay()
187             except Exception:
188                 logger.exception(f"Client replay has crashed!")
189             self.queue.task_done()
190             self.inflight = None
191 
192     def check(self, f: flow.Flow) -> str | None:
193         if f.live or f == self.inflight:
194             return "Can't replay live flow."
195         if f.intercepted:
196             return "Can't replay intercepted flow."
197         if isinstance(f, http.HTTPFlow):
198             if not f.request:
199                 return "Can't replay flow with missing request."
200             if f.request.raw_content is None:
201                 return "Can't replay flow with missing content."
202             if f.websocket is not None:
203                 return "Can't replay WebSocket flows."
204         else:
205             return "Can only replay HTTP flows."
206         return None
207 
208     def load(self, loader):
209         loader.add_option(
210             "client_replay",
211             Sequence[str],
212             [],
213             "Replay client requests from a saved file.",
214         )
215         loader.add_option(
216             "client_replay_concurrency",
217             int,
218             1,
219             "Concurrency limit on in-flight client replay requests. Currently the only valid values are 1 and -1 (no limit).",
220         )
221 
222     def configure(self, updated):
223         if "client_replay" in updated and ctx.options.client_replay:
224             try:
225                 flows = io.read_flows_from_paths(ctx.options.client_replay)
226             except exceptions.FlowReadException as e:
227                 raise exceptions.OptionsError(str(e))
228             self.start_replay(flows)
229 
230         if "client_replay_concurrency" in updated:
231             if ctx.options.client_replay_concurrency not in [-1, 1]:
232                 raise exceptions.OptionsError(
233                     "Currently the only valid client_replay_concurrency values are -1 and 1."
234                 )
235 
236     @command.command("replay.client.count")
237     def count(self) -> int:
238         """
239         Approximate number of flows queued for replay.
240         """
241         return self.queue.qsize() + int(bool(self.inflight))
242 
243     @command.command("replay.client.stop")
244     def stop_replay(self) -> None:
245         """
246         Clear the replay queue.
247         """
248         updated = []
249         while True:
250             try:
251                 f = self.queue.get_nowait()
252             except asyncio.QueueEmpty:
253                 break
254             else:
255                 self.queue.task_done()
256                 f.revert()
257                 updated.append(f)
258 
259         ctx.master.addons.trigger(UpdateHook(updated))
260         logger.log(ALERT, "Client replay queue cleared.")
261 
262     @command.command("replay.client")
263     def start_replay(self, flows: Sequence[flow.Flow]) -> None:
264         """
265         Add flows to the replay queue, skipping flows that can't be replayed.
266         """
267         updated: list[http.HTTPFlow] = []
268         for f in flows:
269             err = self.check(f)
270             if err:
271                 logger.warning(err)
272                 continue
273 
274             http_flow = cast(http.HTTPFlow, f)
275 
276             # Prepare the flow for replay
277             http_flow.backup()
278             http_flow.is_replay = "request"
279             http_flow.response = None
280             http_flow.error = None
281             self.queue.put_nowait(http_flow)
282             updated.append(http_flow)
283         ctx.master.addons.trigger(UpdateHook(updated))
284 
285     @command.command("replay.client.file")
286     def load_file(self, path: mitmproxy.types.Path) -> None:
287         """
288         Load flows from file, and add them to the replay queue.
289         """
290         try:
291             flows = io.read_flows_from_paths([path])
292         except exceptions.FlowReadException as e:
293             raise exceptions.CommandError(str(e))
294         self.start_replay(flows)
295
```
--- 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/mitmproxy/addons/clientplayback.py b/mitmproxy/addons/clientplayback.py
--- a/mitmproxy/addons/clientplayback.py
+++ b/mitmproxy/addons/clientplayback.py
@@ -156,10 +156,10 @@
         self.replay_tasks = set()
 
     def running(self):
+        self.options = ctx.options
         self.playback_task = asyncio_utils.create_task(
             self.playback(), name="client playback"
         )
-        self.options = ctx.options
 
     async def done(self):
         if self.playback_task:

{"golden_diff": "diff --git a/mitmproxy/addons/clientplayback.py b/mitmproxy/addons/clientplayback.py\n--- a/mitmproxy/addons/clientplayback.py\n+++ b/mitmproxy/addons/clientplayback.py\n@@ -156,10 +156,10 @@\n         self.replay_tasks = set()\n \n     def running(self):\n+        self.options = ctx.options\n         self.playback_task = asyncio_utils.create_task(\n             self.playback(), name=\"client playback\"\n         )\n-        self.options = ctx.options\n \n     async def done(self):\n         if self.playback_task:\n", "issue": "AttributeError: 'ClientPlayback' object has no attribute 'options'\n#### Problem Description\r\nClient playback does not seem to work anymore, since 10.2.0. \r\nOn 10.1.6 it still works.\r\n\r\nI am getting the following error:\r\n>Traceback (most recent call last):\r\n  File \"mitmproxy/addons/clientplayback.py\", line 177, in playback\r\nAttributeError: 'ClientPlayback' object has no attribute 'options'\r\n\r\n#### Steps to reproduce the behavior:\r\n1. Save a recorded flow\r\n2. Replay it using `mitmdump -C stored.flow` \r\n3. See error\r\n\r\n#### System Information\r\nMitmproxy: 10.2.1 binary\r\nPython:    3.12.1\r\nOpenSSL:   OpenSSL 3.1.4 24 Oct 2023\r\nPlatform:  macOS-14.2.1-x86_64-i386-64bit\r\n\n", "before_files": [{"content": "from __future__ import annotations\n\nimport asyncio\nimport logging\nimport time\nfrom collections.abc import Sequence\nfrom types import TracebackType\nfrom typing import cast\nfrom typing import Literal\n\nimport mitmproxy.types\nfrom mitmproxy import command\nfrom mitmproxy import ctx\nfrom mitmproxy import exceptions\nfrom mitmproxy import flow\nfrom mitmproxy import http\nfrom mitmproxy import io\nfrom mitmproxy.connection import ConnectionState\nfrom mitmproxy.connection import Server\nfrom mitmproxy.hooks import UpdateHook\nfrom mitmproxy.log import ALERT\nfrom mitmproxy.options import Options\nfrom mitmproxy.proxy import commands\nfrom mitmproxy.proxy import events\nfrom mitmproxy.proxy import layers\nfrom mitmproxy.proxy import server\nfrom mitmproxy.proxy.context import Context\nfrom mitmproxy.proxy.layer import CommandGenerator\nfrom mitmproxy.proxy.layers.http import HTTPMode\nfrom mitmproxy.proxy.mode_specs import UpstreamMode\nfrom mitmproxy.utils import asyncio_utils\n\nlogger = logging.getLogger(__name__)\n\n\nclass MockServer(layers.http.HttpConnection):\n    \"\"\"\n    A mock HTTP \"server\" that just pretends it received a full HTTP request,\n    which is then processed by the proxy core.\n    \"\"\"\n\n    flow: http.HTTPFlow\n\n    def __init__(self, flow: http.HTTPFlow, context: Context):\n        super().__init__(context, context.client)\n        self.flow = flow\n\n    def _handle_event(self, event: events.Event) -> CommandGenerator[None]:\n        if isinstance(event, events.Start):\n            content = self.flow.request.raw_content\n            self.flow.request.timestamp_start = (\n                self.flow.request.timestamp_end\n            ) = time.time()\n            yield layers.http.ReceiveHttp(\n                layers.http.RequestHeaders(\n                    1,\n                    self.flow.request,\n                    end_stream=not (content or self.flow.request.trailers),\n                    replay_flow=self.flow,\n                )\n            )\n            if content:\n                yield layers.http.ReceiveHttp(layers.http.RequestData(1, content))\n            if self.flow.request.trailers:  # pragma: no cover\n                # TODO: Cover this once we support HTTP/1 trailers.\n                yield layers.http.ReceiveHttp(\n                    layers.http.RequestTrailers(1, self.flow.request.trailers)\n                )\n            yield layers.http.ReceiveHttp(layers.http.RequestEndOfMessage(1))\n        elif isinstance(\n            event,\n            (\n                layers.http.ResponseHeaders,\n                layers.http.ResponseData,\n                layers.http.ResponseTrailers,\n                layers.http.ResponseEndOfMessage,\n                layers.http.ResponseProtocolError,\n            ),\n        ):\n            pass\n        else:  # pragma: no cover\n            logger.warning(f\"Unexpected event during replay: {event}\")\n\n\nclass ReplayHandler(server.ConnectionHandler):\n    layer: layers.HttpLayer\n\n    def __init__(self, flow: http.HTTPFlow, options: Options) -> None:\n        client = flow.client_conn.copy()\n        client.state = ConnectionState.OPEN\n\n        context = Context(client, options)\n        context.server = Server(address=(flow.request.host, flow.request.port))\n        if flow.request.scheme == \"https\":\n            context.server.tls = True\n            context.server.sni = flow.request.pretty_host\n        if options.mode and options.mode[0].startswith(\"upstream:\"):\n            mode = UpstreamMode.parse(options.mode[0])\n            assert isinstance(mode, UpstreamMode)  # remove once mypy supports Self.\n            context.server.via = flow.server_conn.via = (mode.scheme, mode.address)\n\n        super().__init__(context)\n\n        if options.mode and options.mode[0].startswith(\"upstream:\"):\n            self.layer = layers.HttpLayer(context, HTTPMode.upstream)\n        else:\n            self.layer = layers.HttpLayer(context, HTTPMode.transparent)\n        self.layer.connections[client] = MockServer(flow, context.fork())\n        self.flow = flow\n        self.done = asyncio.Event()\n\n    async def replay(self) -> None:\n        self.server_event(events.Start())\n        await self.done.wait()\n\n    def log(\n        self,\n        message: str,\n        level: int = logging.INFO,\n        exc_info: Literal[True]\n        | tuple[type[BaseException] | None, BaseException | None, TracebackType | None]\n        | None = None,\n    ) -> None:\n        assert isinstance(level, int)\n        logger.log(level=level, msg=f\"[replay] {message}\")\n\n    async def handle_hook(self, hook: commands.StartHook) -> None:\n        (data,) = hook.args()\n        await ctx.master.addons.handle_lifecycle(hook)\n        if isinstance(data, flow.Flow):\n            await data.wait_for_resume()\n        if isinstance(hook, (layers.http.HttpResponseHook, layers.http.HttpErrorHook)):\n            if self.transports:\n                # close server connections\n                for x in self.transports.values():\n                    if x.handler:\n                        x.handler.cancel()\n                await asyncio.wait(\n                    [x.handler for x in self.transports.values() if x.handler]\n                )\n            # signal completion\n            self.done.set()\n\n\nclass ClientPlayback:\n    playback_task: asyncio.Task | None = None\n    inflight: http.HTTPFlow | None\n    queue: asyncio.Queue\n    options: Options\n    replay_tasks: set[asyncio.Task]\n\n    def __init__(self):\n        self.queue = asyncio.Queue()\n        self.inflight = None\n        self.task = None\n        self.replay_tasks = set()\n\n    def running(self):\n        self.playback_task = asyncio_utils.create_task(\n            self.playback(), name=\"client playback\"\n        )\n        self.options = ctx.options\n\n    async def done(self):\n        if self.playback_task:\n            self.playback_task.cancel()\n            try:\n                await self.playback_task\n            except asyncio.CancelledError:\n                pass\n\n    async def playback(self):\n        while True:\n            self.inflight = await self.queue.get()\n            try:\n                assert self.inflight\n                h = ReplayHandler(self.inflight, self.options)\n                if ctx.options.client_replay_concurrency == -1:\n                    t = asyncio_utils.create_task(\n                        h.replay(), name=\"client playback awaiting response\"\n                    )\n                    # keep a reference so this is not garbage collected\n                    self.replay_tasks.add(t)\n                    t.add_done_callback(self.replay_tasks.remove)\n                else:\n                    await h.replay()\n            except Exception:\n                logger.exception(f\"Client replay has crashed!\")\n            self.queue.task_done()\n            self.inflight = None\n\n    def check(self, f: flow.Flow) -> str | None:\n        if f.live or f == self.inflight:\n            return \"Can't replay live flow.\"\n        if f.intercepted:\n            return \"Can't replay intercepted flow.\"\n        if isinstance(f, http.HTTPFlow):\n            if not f.request:\n                return \"Can't replay flow with missing request.\"\n            if f.request.raw_content is None:\n                return \"Can't replay flow with missing content.\"\n            if f.websocket is not None:\n                return \"Can't replay WebSocket flows.\"\n        else:\n            return \"Can only replay HTTP flows.\"\n        return None\n\n    def load(self, loader):\n        loader.add_option(\n            \"client_replay\",\n            Sequence[str],\n            [],\n            \"Replay client requests from a saved file.\",\n        )\n        loader.add_option(\n            \"client_replay_concurrency\",\n            int,\n            1,\n            \"Concurrency limit on in-flight client replay requests. Currently the only valid values are 1 and -1 (no limit).\",\n        )\n\n    def configure(self, updated):\n        if \"client_replay\" in updated and ctx.options.client_replay:\n            try:\n                flows = io.read_flows_from_paths(ctx.options.client_replay)\n            except exceptions.FlowReadException as e:\n                raise exceptions.OptionsError(str(e))\n            self.start_replay(flows)\n\n        if \"client_replay_concurrency\" in updated:\n            if ctx.options.client_replay_concurrency not in [-1, 1]:\n                raise exceptions.OptionsError(\n                    \"Currently the only valid client_replay_concurrency values are -1 and 1.\"\n                )\n\n    @command.command(\"replay.client.count\")\n    def count(self) -> int:\n        \"\"\"\n        Approximate number of flows queued for replay.\n        \"\"\"\n        return self.queue.qsize() + int(bool(self.inflight))\n\n    @command.command(\"replay.client.stop\")\n    def stop_replay(self) -> None:\n        \"\"\"\n        Clear the replay queue.\n        \"\"\"\n        updated = []\n        while True:\n            try:\n                f = self.queue.get_nowait()\n            except asyncio.QueueEmpty:\n                break\n            else:\n                self.queue.task_done()\n                f.revert()\n                updated.append(f)\n\n        ctx.master.addons.trigger(UpdateHook(updated))\n        logger.log(ALERT, \"Client replay queue cleared.\")\n\n    @command.command(\"replay.client\")\n    def start_replay(self, flows: Sequence[flow.Flow]) -> None:\n        \"\"\"\n        Add flows to the replay queue, skipping flows that can't be replayed.\n        \"\"\"\n        updated: list[http.HTTPFlow] = []\n        for f in flows:\n            err = self.check(f)\n            if err:\n                logger.warning(err)\n                continue\n\n            http_flow = cast(http.HTTPFlow, f)\n\n            # Prepare the flow for replay\n            http_flow.backup()\n            http_flow.is_replay = \"request\"\n            http_flow.response = None\n            http_flow.error = None\n            self.queue.put_nowait(http_flow)\n            updated.append(http_flow)\n        ctx.master.addons.trigger(UpdateHook(updated))\n\n    @command.command(\"replay.client.file\")\n    def load_file(self, path: mitmproxy.types.Path) -> None:\n        \"\"\"\n        Load flows from file, and add them to the replay queue.\n        \"\"\"\n        try:\n            flows = io.read_flows_from_paths([path])\n        except exceptions.FlowReadException as e:\n            raise exceptions.CommandError(str(e))\n        self.start_replay(flows)\n", "path": "mitmproxy/addons/clientplayback.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport asyncio\nimport logging\nimport time\nfrom collections.abc import Sequence\nfrom types import TracebackType\nfrom typing import cast\nfrom typing import Literal\n\nimport mitmproxy.types\nfrom mitmproxy import command\nfrom mitmproxy import ctx\nfrom mitmproxy import exceptions\nfrom mitmproxy import flow\nfrom mitmproxy import http\nfrom mitmproxy import io\nfrom mitmproxy.connection import ConnectionState\nfrom mitmproxy.connection import Server\nfrom mitmproxy.hooks import UpdateHook\nfrom mitmproxy.log import ALERT\nfrom mitmproxy.options import Options\nfrom mitmproxy.proxy import commands\nfrom mitmproxy.proxy import events\nfrom mitmproxy.proxy import layers\nfrom mitmproxy.proxy import server\nfrom mitmproxy.proxy.context import Context\nfrom mitmproxy.proxy.layer import CommandGenerator\nfrom mitmproxy.proxy.layers.http import HTTPMode\nfrom mitmproxy.proxy.mode_specs import UpstreamMode\nfrom mitmproxy.utils import asyncio_utils\n\nlogger = logging.getLogger(__name__)\n\n\nclass MockServer(layers.http.HttpConnection):\n    \"\"\"\n    A mock HTTP \"server\" that just pretends it received a full HTTP request,\n    which is then processed by the proxy core.\n    \"\"\"\n\n    flow: http.HTTPFlow\n\n    def __init__(self, flow: http.HTTPFlow, context: Context):\n        super().__init__(context, context.client)\n        self.flow = flow\n\n    def _handle_event(self, event: events.Event) -> CommandGenerator[None]:\n        if isinstance(event, events.Start):\n            content = self.flow.request.raw_content\n            self.flow.request.timestamp_start = (\n                self.flow.request.timestamp_end\n            ) = time.time()\n            yield layers.http.ReceiveHttp(\n                layers.http.RequestHeaders(\n                    1,\n                    self.flow.request,\n                    end_stream=not (content or self.flow.request.trailers),\n                    replay_flow=self.flow,\n                )\n            )\n            if content:\n                yield layers.http.ReceiveHttp(layers.http.RequestData(1, content))\n            if self.flow.request.trailers:  # pragma: no cover\n                # TODO: Cover this once we support HTTP/1 trailers.\n                yield layers.http.ReceiveHttp(\n                    layers.http.RequestTrailers(1, self.flow.request.trailers)\n                )\n            yield layers.http.ReceiveHttp(layers.http.RequestEndOfMessage(1))\n        elif isinstance(\n            event,\n            (\n                layers.http.ResponseHeaders,\n                layers.http.ResponseData,\n                layers.http.ResponseTrailers,\n                layers.http.ResponseEndOfMessage,\n                layers.http.ResponseProtocolError,\n            ),\n        ):\n            pass\n        else:  # pragma: no cover\n            logger.warning(f\"Unexpected event during replay: {event}\")\n\n\nclass ReplayHandler(server.ConnectionHandler):\n    layer: layers.HttpLayer\n\n    def __init__(self, flow: http.HTTPFlow, options: Options) -> None:\n        client = flow.client_conn.copy()\n        client.state = ConnectionState.OPEN\n\n        context = Context(client, options)\n        context.server = Server(address=(flow.request.host, flow.request.port))\n        if flow.request.scheme == \"https\":\n            context.server.tls = True\n            context.server.sni = flow.request.pretty_host\n        if options.mode and options.mode[0].startswith(\"upstream:\"):\n            mode = UpstreamMode.parse(options.mode[0])\n            assert isinstance(mode, UpstreamMode)  # remove once mypy supports Self.\n            context.server.via = flow.server_conn.via = (mode.scheme, mode.address)\n\n        super().__init__(context)\n\n        if options.mode and options.mode[0].startswith(\"upstream:\"):\n            self.layer = layers.HttpLayer(context, HTTPMode.upstream)\n        else:\n            self.layer = layers.HttpLayer(context, HTTPMode.transparent)\n        self.layer.connections[client] = MockServer(flow, context.fork())\n        self.flow = flow\n        self.done = asyncio.Event()\n\n    async def replay(self) -> None:\n        self.server_event(events.Start())\n        await self.done.wait()\n\n    def log(\n        self,\n        message: str,\n        level: int = logging.INFO,\n        exc_info: Literal[True]\n        | tuple[type[BaseException] | None, BaseException | None, TracebackType | None]\n        | None = None,\n    ) -> None:\n        assert isinstance(level, int)\n        logger.log(level=level, msg=f\"[replay] {message}\")\n\n    async def handle_hook(self, hook: commands.StartHook) -> None:\n        (data,) = hook.args()\n        await ctx.master.addons.handle_lifecycle(hook)\n        if isinstance(data, flow.Flow):\n            await data.wait_for_resume()\n        if isinstance(hook, (layers.http.HttpResponseHook, layers.http.HttpErrorHook)):\n            if self.transports:\n                # close server connections\n                for x in self.transports.values():\n                    if x.handler:\n                        x.handler.cancel()\n                await asyncio.wait(\n                    [x.handler for x in self.transports.values() if x.handler]\n                )\n            # signal completion\n            self.done.set()\n\n\nclass ClientPlayback:\n    playback_task: asyncio.Task | None = None\n    inflight: http.HTTPFlow | None\n    queue: asyncio.Queue\n    options: Options\n    replay_tasks: set[asyncio.Task]\n\n    def __init__(self):\n        self.queue = asyncio.Queue()\n        self.inflight = None\n        self.task = None\n        self.replay_tasks = set()\n\n    def running(self):\n        self.options = ctx.options\n        self.playback_task = asyncio_utils.create_task(\n            self.playback(), name=\"client playback\"\n        )\n\n    async def done(self):\n        if self.playback_task:\n            self.playback_task.cancel()\n            try:\n                await self.playback_task\n            except asyncio.CancelledError:\n                pass\n\n    async def playback(self):\n        while True:\n            self.inflight = await self.queue.get()\n            try:\n                assert self.inflight\n                h = ReplayHandler(self.inflight, self.options)\n                if ctx.options.client_replay_concurrency == -1:\n                    t = asyncio_utils.create_task(\n                        h.replay(), name=\"client playback awaiting response\"\n                    )\n                    # keep a reference so this is not garbage collected\n                    self.replay_tasks.add(t)\n                    t.add_done_callback(self.replay_tasks.remove)\n                else:\n                    await h.replay()\n            except Exception:\n                logger.exception(f\"Client replay has crashed!\")\n            self.queue.task_done()\n            self.inflight = None\n\n    def check(self, f: flow.Flow) -> str | None:\n        if f.live or f == self.inflight:\n            return \"Can't replay live flow.\"\n        if f.intercepted:\n            return \"Can't replay intercepted flow.\"\n        if isinstance(f, http.HTTPFlow):\n            if not f.request:\n                return \"Can't replay flow with missing request.\"\n            if f.request.raw_content is None:\n                return \"Can't replay flow with missing content.\"\n            if f.websocket is not None:\n                return \"Can't replay WebSocket flows.\"\n        else:\n            return \"Can only replay HTTP flows.\"\n        return None\n\n    def load(self, loader):\n        loader.add_option(\n            \"client_replay\",\n            Sequence[str],\n            [],\n            \"Replay client requests from a saved file.\",\n        )\n        loader.add_option(\n            \"client_replay_concurrency\",\n            int,\n            1,\n            \"Concurrency limit on in-flight client replay requests. Currently the only valid values are 1 and -1 (no limit).\",\n        )\n\n    def configure(self, updated):\n        if \"client_replay\" in updated and ctx.options.client_replay:\n            try:\n                flows = io.read_flows_from_paths(ctx.options.client_replay)\n            except exceptions.FlowReadException as e:\n                raise exceptions.OptionsError(str(e))\n            self.start_replay(flows)\n\n        if \"client_replay_concurrency\" in updated:\n            if ctx.options.client_replay_concurrency not in [-1, 1]:\n                raise exceptions.OptionsError(\n                    \"Currently the only valid client_replay_concurrency values are -1 and 1.\"\n                )\n\n    @command.command(\"replay.client.count\")\n    def count(self) -> int:\n        \"\"\"\n        Approximate number of flows queued for replay.\n        \"\"\"\n        return self.queue.qsize() + int(bool(self.inflight))\n\n    @command.command(\"replay.client.stop\")\n    def stop_replay(self) -> None:\n        \"\"\"\n        Clear the replay queue.\n        \"\"\"\n        updated = []\n        while True:\n            try:\n                f = self.queue.get_nowait()\n            except asyncio.QueueEmpty:\n                break\n            else:\n                self.queue.task_done()\n                f.revert()\n                updated.append(f)\n\n        ctx.master.addons.trigger(UpdateHook(updated))\n        logger.log(ALERT, \"Client replay queue cleared.\")\n\n    @command.command(\"replay.client\")\n    def start_replay(self, flows: Sequence[flow.Flow]) -> None:\n        \"\"\"\n        Add flows to the replay queue, skipping flows that can't be replayed.\n        \"\"\"\n        updated: list[http.HTTPFlow] = []\n        for f in flows:\n            err = self.check(f)\n            if err:\n                logger.warning(err)\n                continue\n\n            http_flow = cast(http.HTTPFlow, f)\n\n            # Prepare the flow for replay\n            http_flow.backup()\n            http_flow.is_replay = \"request\"\n            http_flow.response = None\n            http_flow.error = None\n            self.queue.put_nowait(http_flow)\n            updated.append(http_flow)\n        ctx.master.addons.trigger(UpdateHook(updated))\n\n    @command.command(\"replay.client.file\")\n    def load_file(self, path: mitmproxy.types.Path) -> None:\n        \"\"\"\n        Load flows from file, and add them to the replay queue.\n        \"\"\"\n        try:\n            flows = io.read_flows_from_paths([path])\n        except exceptions.FlowReadException as e:\n            raise exceptions.CommandError(str(e))\n        self.start_replay(flows)\n", "path": "mitmproxy/addons/clientplayback.py"}]}