We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Include the `task-run` as a related resources when emitting events.
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar request and didn't find it.
- [X] I searched the Prefect documentation for this feature.

### Prefect Version

2.x

### Describe the current behavior

Currently task-runs aren't included as related resources when emitting events.

### Describe the proposed behavior

The current task run in the TaskRunContext should be included in any event fired while a task is running.

### Example Use

_No response_

### Additional context

_No response_

--- 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/events/related.py`
Content:
```
1 import asyncio
2 import pendulum
3 from typing import (
4     TYPE_CHECKING,
5     Any,
6     Awaitable,
7     Callable,
8     Dict,
9     Iterable,
10     List,
11     Optional,
12     Set,
13     Tuple,
14     Union,
15 )
16 from uuid import UUID
17 from pendulum.datetime import DateTime
18 
19 from .schemas import RelatedResource
20 
21 if TYPE_CHECKING:
22     from prefect._internal.schemas.bases import ObjectBaseModel
23 
24 ResourceCacheEntry = Dict[str, Union[str, "ObjectBaseModel", None]]
25 RelatedResourceCache = Dict[str, Tuple[ResourceCacheEntry, DateTime]]
26 
27 MAX_CACHE_SIZE = 100
28 RESOURCE_CACHE: RelatedResourceCache = {}
29 
30 
31 def tags_as_related_resources(tags: Iterable[str]) -> List[RelatedResource]:
32     return [
33         RelatedResource(
34             __root__={
35                 "prefect.resource.id": f"prefect.tag.{tag}",
36                 "prefect.resource.role": "tag",
37             }
38         )
39         for tag in sorted(tags)
40     ]
41 
42 
43 def object_as_related_resource(kind: str, role: str, object: Any) -> RelatedResource:
44     resource_id = f"prefect.{kind}.{object.id}"
45 
46     return RelatedResource(
47         __root__={
48             "prefect.resource.id": resource_id,
49             "prefect.resource.role": role,
50             "prefect.resource.name": object.name,
51         }
52     )
53 
54 
55 async def related_resources_from_run_context(
56     exclude: Optional[Set[str]] = None,
57 ) -> List[RelatedResource]:
58     from prefect.client.orchestration import get_client
59     from prefect.context import FlowRunContext, TaskRunContext
60 
61     if exclude is None:
62         exclude = set()
63 
64     flow_run_context = FlowRunContext.get()
65     task_run_context = TaskRunContext.get()
66 
67     if not flow_run_context and not task_run_context:
68         return []
69 
70     flow_run_id: UUID = (
71         flow_run_context.flow_run.id
72         if flow_run_context
73         else task_run_context.task_run.flow_run_id
74     )
75 
76     related_objects: list[ResourceCacheEntry] = []
77 
78     async with get_client() as client:
79 
80         async def dummy_read():
81             return {}
82 
83         related_objects = [
84             await _get_and_cache_related_object(
85                 kind="flow-run",
86                 role="flow-run",
87                 client_method=client.read_flow_run,
88                 obj_id=flow_run_id,
89                 cache=RESOURCE_CACHE,
90             )
91         ]
92 
93         flow_run = related_objects[0]["object"]
94 
95         if flow_run:
96             related_objects += list(
97                 await asyncio.gather(
98                     _get_and_cache_related_object(
99                         kind="flow",
100                         role="flow",
101                         client_method=client.read_flow,
102                         obj_id=flow_run.flow_id,
103                         cache=RESOURCE_CACHE,
104                     ),
105                     (
106                         _get_and_cache_related_object(
107                             kind="deployment",
108                             role="deployment",
109                             client_method=client.read_deployment,
110                             obj_id=flow_run.deployment_id,
111                             cache=RESOURCE_CACHE,
112                         )
113                         if flow_run.deployment_id
114                         else dummy_read()
115                     ),
116                     (
117                         _get_and_cache_related_object(
118                             kind="work-queue",
119                             role="work-queue",
120                             client_method=client.read_work_queue,
121                             obj_id=flow_run.work_queue_id,
122                             cache=RESOURCE_CACHE,
123                         )
124                         if flow_run.work_queue_id
125                         else dummy_read()
126                     ),
127                     (
128                         _get_and_cache_related_object(
129                             kind="work-pool",
130                             role="work-pool",
131                             client_method=client.read_work_pool,
132                             obj_id=flow_run.work_pool_name,
133                             cache=RESOURCE_CACHE,
134                         )
135                         if flow_run.work_pool_name
136                         else dummy_read()
137                     ),
138                 )
139             )
140 
141     related = []
142     tags = set()
143 
144     for entry in related_objects:
145         obj_ = entry.get("object")
146         if obj_ is None:
147             continue
148 
149         assert isinstance(entry["kind"], str) and isinstance(entry["role"], str)
150 
151         resource = object_as_related_resource(
152             kind=entry["kind"], role=entry["kind"], object=obj_
153         )
154 
155         if resource.id in exclude:
156             continue
157 
158         related.append(resource)
159         if hasattr(obj_, "tags"):
160             tags |= set(obj_.tags)
161 
162     related += [
163         resource
164         for resource in tags_as_related_resources(tags)
165         if resource.id not in exclude
166     ]
167 
168     return related
169 
170 
171 async def _get_and_cache_related_object(
172     kind: str,
173     role: str,
174     client_method: Callable[[Union[UUID, str]], Awaitable[Optional["ObjectBaseModel"]]],
175     obj_id: Union[UUID, str],
176     cache: RelatedResourceCache,
177 ) -> ResourceCacheEntry:
178     cache_key = f"{kind}.{obj_id}"
179     entry = None
180 
181     if cache_key in cache:
182         entry, _ = cache[cache_key]
183     else:
184         obj_ = await client_method(obj_id)
185         entry = {
186             "kind": kind,
187             "object": obj_,
188         }
189 
190     cache[cache_key] = (entry, pendulum.now("UTC"))
191 
192     # In the case of a worker or agent this cache could be long-lived. To keep
193     # from running out of memory only keep `MAX_CACHE_SIZE` entries in the
194     # cache.
195     if len(cache) > MAX_CACHE_SIZE:
196         oldest_key = sorted(
197             [(key, timestamp) for key, (_, timestamp) in cache.items()],
198             key=lambda k: k[1],
199         )[0][0]
200 
201         if oldest_key:
202             del cache[oldest_key]
203 
204     # Because the role is event specific and can change depending on the
205     # type of event being emitted, this adds the role from the args to the
206     # entry before returning it rather than storing it in the cache.
207     entry["role"] = role
208     return entry
209
```
--- 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/events/related.py b/src/prefect/events/related.py
--- a/src/prefect/events/related.py
+++ b/src/prefect/events/related.py
@@ -80,15 +80,33 @@
         async def dummy_read():
             return {}
 
-        related_objects = [
-            await _get_and_cache_related_object(
-                kind="flow-run",
-                role="flow-run",
-                client_method=client.read_flow_run,
-                obj_id=flow_run_id,
-                cache=RESOURCE_CACHE,
+        if flow_run_context:
+            related_objects.append(
+                {
+                    "kind": "flow-run",
+                    "role": "flow-run",
+                    "object": flow_run_context.flow_run,
+                },
+            )
+        else:
+            related_objects.append(
+                await _get_and_cache_related_object(
+                    kind="flow-run",
+                    role="flow-run",
+                    client_method=client.read_flow_run,
+                    obj_id=flow_run_id,
+                    cache=RESOURCE_CACHE,
+                )
+            )
+
+        if task_run_context:
+            related_objects.append(
+                {
+                    "kind": "task-run",
+                    "role": "task-run",
+                    "object": task_run_context.task_run,
+                },
             )
-        ]
 
         flow_run = related_objects[0]["object"]

{"golden_diff": "diff --git a/src/prefect/events/related.py b/src/prefect/events/related.py\n--- a/src/prefect/events/related.py\n+++ b/src/prefect/events/related.py\n@@ -80,15 +80,33 @@\n         async def dummy_read():\n             return {}\n \n-        related_objects = [\n-            await _get_and_cache_related_object(\n-                kind=\"flow-run\",\n-                role=\"flow-run\",\n-                client_method=client.read_flow_run,\n-                obj_id=flow_run_id,\n-                cache=RESOURCE_CACHE,\n+        if flow_run_context:\n+            related_objects.append(\n+                {\n+                    \"kind\": \"flow-run\",\n+                    \"role\": \"flow-run\",\n+                    \"object\": flow_run_context.flow_run,\n+                },\n+            )\n+        else:\n+            related_objects.append(\n+                await _get_and_cache_related_object(\n+                    kind=\"flow-run\",\n+                    role=\"flow-run\",\n+                    client_method=client.read_flow_run,\n+                    obj_id=flow_run_id,\n+                    cache=RESOURCE_CACHE,\n+                )\n+            )\n+\n+        if task_run_context:\n+            related_objects.append(\n+                {\n+                    \"kind\": \"task-run\",\n+                    \"role\": \"task-run\",\n+                    \"object\": task_run_context.task_run,\n+                },\n             )\n-        ]\n \n         flow_run = related_objects[0][\"object\"]\n", "issue": "Include the `task-run` as a related resources when emitting events.\n### First check\n\n- [X] I added a descriptive title to this issue.\n- [X] I used the GitHub search to find a similar request and didn't find it.\n- [X] I searched the Prefect documentation for this feature.\n\n### Prefect Version\n\n2.x\n\n### Describe the current behavior\n\nCurrently task-runs aren't included as related resources when emitting events.\n\n### Describe the proposed behavior\n\nThe current task run in the TaskRunContext should be included in any event fired while a task is running.\n\n### Example Use\n\n_No response_\n\n### Additional context\n\n_No response_\n", "before_files": [{"content": "import asyncio\nimport pendulum\nfrom typing import (\n    TYPE_CHECKING,\n    Any,\n    Awaitable,\n    Callable,\n    Dict,\n    Iterable,\n    List,\n    Optional,\n    Set,\n    Tuple,\n    Union,\n)\nfrom uuid import UUID\nfrom pendulum.datetime import DateTime\n\nfrom .schemas import RelatedResource\n\nif TYPE_CHECKING:\n    from prefect._internal.schemas.bases import ObjectBaseModel\n\nResourceCacheEntry = Dict[str, Union[str, \"ObjectBaseModel\", None]]\nRelatedResourceCache = Dict[str, Tuple[ResourceCacheEntry, DateTime]]\n\nMAX_CACHE_SIZE = 100\nRESOURCE_CACHE: RelatedResourceCache = {}\n\n\ndef tags_as_related_resources(tags: Iterable[str]) -> List[RelatedResource]:\n    return [\n        RelatedResource(\n            __root__={\n                \"prefect.resource.id\": f\"prefect.tag.{tag}\",\n                \"prefect.resource.role\": \"tag\",\n            }\n        )\n        for tag in sorted(tags)\n    ]\n\n\ndef object_as_related_resource(kind: str, role: str, object: Any) -> RelatedResource:\n    resource_id = f\"prefect.{kind}.{object.id}\"\n\n    return RelatedResource(\n        __root__={\n            \"prefect.resource.id\": resource_id,\n            \"prefect.resource.role\": role,\n            \"prefect.resource.name\": object.name,\n        }\n    )\n\n\nasync def related_resources_from_run_context(\n    exclude: Optional[Set[str]] = None,\n) -> List[RelatedResource]:\n    from prefect.client.orchestration import get_client\n    from prefect.context import FlowRunContext, TaskRunContext\n\n    if exclude is None:\n        exclude = set()\n\n    flow_run_context = FlowRunContext.get()\n    task_run_context = TaskRunContext.get()\n\n    if not flow_run_context and not task_run_context:\n        return []\n\n    flow_run_id: UUID = (\n        flow_run_context.flow_run.id\n        if flow_run_context\n        else task_run_context.task_run.flow_run_id\n    )\n\n    related_objects: list[ResourceCacheEntry] = []\n\n    async with get_client() as client:\n\n        async def dummy_read():\n            return {}\n\n        related_objects = [\n            await _get_and_cache_related_object(\n                kind=\"flow-run\",\n                role=\"flow-run\",\n                client_method=client.read_flow_run,\n                obj_id=flow_run_id,\n                cache=RESOURCE_CACHE,\n            )\n        ]\n\n        flow_run = related_objects[0][\"object\"]\n\n        if flow_run:\n            related_objects += list(\n                await asyncio.gather(\n                    _get_and_cache_related_object(\n                        kind=\"flow\",\n                        role=\"flow\",\n                        client_method=client.read_flow,\n                        obj_id=flow_run.flow_id,\n                        cache=RESOURCE_CACHE,\n                    ),\n                    (\n                        _get_and_cache_related_object(\n                            kind=\"deployment\",\n                            role=\"deployment\",\n                            client_method=client.read_deployment,\n                            obj_id=flow_run.deployment_id,\n                            cache=RESOURCE_CACHE,\n                        )\n                        if flow_run.deployment_id\n                        else dummy_read()\n                    ),\n                    (\n                        _get_and_cache_related_object(\n                            kind=\"work-queue\",\n                            role=\"work-queue\",\n                            client_method=client.read_work_queue,\n                            obj_id=flow_run.work_queue_id,\n                            cache=RESOURCE_CACHE,\n                        )\n                        if flow_run.work_queue_id\n                        else dummy_read()\n                    ),\n                    (\n                        _get_and_cache_related_object(\n                            kind=\"work-pool\",\n                            role=\"work-pool\",\n                            client_method=client.read_work_pool,\n                            obj_id=flow_run.work_pool_name,\n                            cache=RESOURCE_CACHE,\n                        )\n                        if flow_run.work_pool_name\n                        else dummy_read()\n                    ),\n                )\n            )\n\n    related = []\n    tags = set()\n\n    for entry in related_objects:\n        obj_ = entry.get(\"object\")\n        if obj_ is None:\n            continue\n\n        assert isinstance(entry[\"kind\"], str) and isinstance(entry[\"role\"], str)\n\n        resource = object_as_related_resource(\n            kind=entry[\"kind\"], role=entry[\"kind\"], object=obj_\n        )\n\n        if resource.id in exclude:\n            continue\n\n        related.append(resource)\n        if hasattr(obj_, \"tags\"):\n            tags |= set(obj_.tags)\n\n    related += [\n        resource\n        for resource in tags_as_related_resources(tags)\n        if resource.id not in exclude\n    ]\n\n    return related\n\n\nasync def _get_and_cache_related_object(\n    kind: str,\n    role: str,\n    client_method: Callable[[Union[UUID, str]], Awaitable[Optional[\"ObjectBaseModel\"]]],\n    obj_id: Union[UUID, str],\n    cache: RelatedResourceCache,\n) -> ResourceCacheEntry:\n    cache_key = f\"{kind}.{obj_id}\"\n    entry = None\n\n    if cache_key in cache:\n        entry, _ = cache[cache_key]\n    else:\n        obj_ = await client_method(obj_id)\n        entry = {\n            \"kind\": kind,\n            \"object\": obj_,\n        }\n\n    cache[cache_key] = (entry, pendulum.now(\"UTC\"))\n\n    # In the case of a worker or agent this cache could be long-lived. To keep\n    # from running out of memory only keep `MAX_CACHE_SIZE` entries in the\n    # cache.\n    if len(cache) > MAX_CACHE_SIZE:\n        oldest_key = sorted(\n            [(key, timestamp) for key, (_, timestamp) in cache.items()],\n            key=lambda k: k[1],\n        )[0][0]\n\n        if oldest_key:\n            del cache[oldest_key]\n\n    # Because the role is event specific and can change depending on the\n    # type of event being emitted, this adds the role from the args to the\n    # entry before returning it rather than storing it in the cache.\n    entry[\"role\"] = role\n    return entry\n", "path": "src/prefect/events/related.py"}], "after_files": [{"content": "import asyncio\nimport pendulum\nfrom typing import (\n    TYPE_CHECKING,\n    Any,\n    Awaitable,\n    Callable,\n    Dict,\n    Iterable,\n    List,\n    Optional,\n    Set,\n    Tuple,\n    Union,\n)\nfrom uuid import UUID\nfrom pendulum.datetime import DateTime\n\nfrom .schemas import RelatedResource\n\nif TYPE_CHECKING:\n    from prefect._internal.schemas.bases import ObjectBaseModel\n\nResourceCacheEntry = Dict[str, Union[str, \"ObjectBaseModel\", None]]\nRelatedResourceCache = Dict[str, Tuple[ResourceCacheEntry, DateTime]]\n\nMAX_CACHE_SIZE = 100\nRESOURCE_CACHE: RelatedResourceCache = {}\n\n\ndef tags_as_related_resources(tags: Iterable[str]) -> List[RelatedResource]:\n    return [\n        RelatedResource(\n            __root__={\n                \"prefect.resource.id\": f\"prefect.tag.{tag}\",\n                \"prefect.resource.role\": \"tag\",\n            }\n        )\n        for tag in sorted(tags)\n    ]\n\n\ndef object_as_related_resource(kind: str, role: str, object: Any) -> RelatedResource:\n    resource_id = f\"prefect.{kind}.{object.id}\"\n\n    return RelatedResource(\n        __root__={\n            \"prefect.resource.id\": resource_id,\n            \"prefect.resource.role\": role,\n            \"prefect.resource.name\": object.name,\n        }\n    )\n\n\nasync def related_resources_from_run_context(\n    exclude: Optional[Set[str]] = None,\n) -> List[RelatedResource]:\n    from prefect.client.orchestration import get_client\n    from prefect.context import FlowRunContext, TaskRunContext\n\n    if exclude is None:\n        exclude = set()\n\n    flow_run_context = FlowRunContext.get()\n    task_run_context = TaskRunContext.get()\n\n    if not flow_run_context and not task_run_context:\n        return []\n\n    flow_run_id: UUID = (\n        flow_run_context.flow_run.id\n        if flow_run_context\n        else task_run_context.task_run.flow_run_id\n    )\n\n    related_objects: list[ResourceCacheEntry] = []\n\n    async with get_client() as client:\n\n        async def dummy_read():\n            return {}\n\n        if flow_run_context:\n            related_objects.append(\n                {\n                    \"kind\": \"flow-run\",\n                    \"role\": \"flow-run\",\n                    \"object\": flow_run_context.flow_run,\n                },\n            )\n        else:\n            related_objects.append(\n                await _get_and_cache_related_object(\n                    kind=\"flow-run\",\n                    role=\"flow-run\",\n                    client_method=client.read_flow_run,\n                    obj_id=flow_run_id,\n                    cache=RESOURCE_CACHE,\n                )\n            )\n\n        if task_run_context:\n            related_objects.append(\n                {\n                    \"kind\": \"task-run\",\n                    \"role\": \"task-run\",\n                    \"object\": task_run_context.task_run,\n                },\n            )\n\n        flow_run = related_objects[0][\"object\"]\n\n        if flow_run:\n            related_objects += list(\n                await asyncio.gather(\n                    _get_and_cache_related_object(\n                        kind=\"flow\",\n                        role=\"flow\",\n                        client_method=client.read_flow,\n                        obj_id=flow_run.flow_id,\n                        cache=RESOURCE_CACHE,\n                    ),\n                    (\n                        _get_and_cache_related_object(\n                            kind=\"deployment\",\n                            role=\"deployment\",\n                            client_method=client.read_deployment,\n                            obj_id=flow_run.deployment_id,\n                            cache=RESOURCE_CACHE,\n                        )\n                        if flow_run.deployment_id\n                        else dummy_read()\n                    ),\n                    (\n                        _get_and_cache_related_object(\n                            kind=\"work-queue\",\n                            role=\"work-queue\",\n                            client_method=client.read_work_queue,\n                            obj_id=flow_run.work_queue_id,\n                            cache=RESOURCE_CACHE,\n                        )\n                        if flow_run.work_queue_id\n                        else dummy_read()\n                    ),\n                    (\n                        _get_and_cache_related_object(\n                            kind=\"work-pool\",\n                            role=\"work-pool\",\n                            client_method=client.read_work_pool,\n                            obj_id=flow_run.work_pool_name,\n                            cache=RESOURCE_CACHE,\n                        )\n                        if flow_run.work_pool_name\n                        else dummy_read()\n                    ),\n                )\n            )\n\n    related = []\n    tags = set()\n\n    for entry in related_objects:\n        obj_ = entry.get(\"object\")\n        if obj_ is None:\n            continue\n\n        assert isinstance(entry[\"kind\"], str) and isinstance(entry[\"role\"], str)\n\n        resource = object_as_related_resource(\n            kind=entry[\"kind\"], role=entry[\"kind\"], object=obj_\n        )\n\n        if resource.id in exclude:\n            continue\n\n        related.append(resource)\n        if hasattr(obj_, \"tags\"):\n            tags |= set(obj_.tags)\n\n    related += [\n        resource\n        for resource in tags_as_related_resources(tags)\n        if resource.id not in exclude\n    ]\n\n    return related\n\n\nasync def _get_and_cache_related_object(\n    kind: str,\n    role: str,\n    client_method: Callable[[Union[UUID, str]], Awaitable[Optional[\"ObjectBaseModel\"]]],\n    obj_id: Union[UUID, str],\n    cache: RelatedResourceCache,\n) -> ResourceCacheEntry:\n    cache_key = f\"{kind}.{obj_id}\"\n    entry = None\n\n    if cache_key in cache:\n        entry, _ = cache[cache_key]\n    else:\n        obj_ = await client_method(obj_id)\n        entry = {\n            \"kind\": kind,\n            \"object\": obj_,\n        }\n\n    cache[cache_key] = (entry, pendulum.now(\"UTC\"))\n\n    # In the case of a worker or agent this cache could be long-lived. To keep\n    # from running out of memory only keep `MAX_CACHE_SIZE` entries in the\n    # cache.\n    if len(cache) > MAX_CACHE_SIZE:\n        oldest_key = sorted(\n            [(key, timestamp) for key, (_, timestamp) in cache.items()],\n            key=lambda k: k[1],\n        )[0][0]\n\n        if oldest_key:\n            del cache[oldest_key]\n\n    # Because the role is event specific and can change depending on the\n    # type of event being emitted, this adds the role from the args to the\n    # entry before returning it rather than storing it in the cache.\n    entry[\"role\"] = role\n    return entry\n", "path": "src/prefect/events/related.py"}]}