Improve robustness of subprocess text streaming (#6445)

Previously, this utility would call blocking writes to stderr and stdout directly. This appears to introduce the possibility of race conditions and blocked event loops when many processes are run concurrently. Since we use these utilities to launch flows in parallel processes from the agent, it is important that they are robust to concurrency. `wrap_file` _may_ not be thread safe, but this is still an improvement from where we were. There is not a clear suggested pattern for this, see extensive discussion at python-trio/trio#174.

src/prefect/utilities/processutils.py

@@ -8,7 +8,7 @@
 import anyio.abc
 from anyio.streams.text import TextReceiveStream, TextSendStream
 
-TextSink = Union[TextIO, TextSendStream]
+TextSink = Union[anyio.AsyncFile, TextIO, TextSendStream]
 
 
 @asynccontextmanager
@@ -101,9 +101,17 @@ async def consume_process_output(
 
 
 async def stream_text(source: TextReceiveStream, sink: Optional[TextSink]):
+    if isinstance(sink, TextIOBase):
+        # Convert the blocking sink to an async-compatible object
+        sink = anyio.wrap_file(sink)
+
     async for item in source:
         if isinstance(sink, TextSendStream):
             await sink.send(item)
-        elif isinstance(sink, TextIOBase):
-            sink.write(item)
-            sink.flush()
+        elif isinstance(sink, anyio.AsyncFile):
+            await sink.write(item)
+            await sink.flush()
+        elif sink is None:
+            pass  # Consume the item but perform no action
+        else:
+            raise TypeError(f"Unsupported sink type {type(sink).__name__}")