fix(util): share streamed stderr consumption (#45716)

packages/core/test/process/process.test.ts

@@ -306,6 +306,34 @@ describe("AppProcess", () => {
       }),
     )
 
+    it.live(
+      "includes stderr in output while retaining capped failure diagnostics",
+      Effect.gen(function* () {
+        const svc = yield* AppProcess.Service
+        const lines: string[] = []
+        const exit = yield* Effect.exit(
+          svc
+            .runStream(
+              cmd(
+                "-e",
+                "console.log('stdout-line'); console.error('stderr-line'); console.error('diagnostic-tail'); process.exit(2)",
+              ),
+              { includeStderr: true, maxErrorBytes: 20, okExitCodes: [0] },
+            )
+            .pipe(Stream.runForEach((line) => Effect.sync(() => lines.push(line)))),
+        )
+
+        expect(lines.toSorted()).toEqual(["diagnostic-tail", "stderr-line", "stdout-line"])
+        expect(Exit.isFailure(exit)).toBe(true)
+        if (!Exit.isFailure(exit)) return
+        const reason = exit.cause.reasons[0]
+        expect(reason?._tag).toBe("Fail")
+        if (!reason || reason._tag !== "Fail") return
+        expect(reason.error).toBeInstanceOf(AppProcess.AppProcessError)
+        expect(reason.error.stderr).toBe("stderr-line\ndiagnost")
+      }),
+    )
+
     it.live(
       "without okExitCodes, never fails on exit code",
       Effect.gen(function* () {

packages/util/src/process.ts

@@ -220,11 +220,22 @@ const layer = Layer.effect(
       const built: Stream.Stream<string, AppProcessError | PlatformError> = Stream.unwrap(
         Effect.gen(function* () {
           const handle = yield* spawner.spawn(command)
+          const streams =
+            options?.includeStderr === true
+              ? yield* handle.stderr.pipe(
+                  Stream.broadcastN({ n: 2, capacity: 16 }),
+                  Effect.map((copies) => ({
+                    source: Stream.merge(handle.stdout, copies[0]),
+                    diagnostics: copies[1],
+                  })),
+                )
+              : { source: handle.stdout, diagnostics: handle.stderr }
           const stderrFiber = yield* Effect.forkScoped(
-            collectStream(handle.stderr, options?.maxErrorBytes).pipe(Effect.map((x) => x.buffer.toString("utf8"))),
+            collectStream(streams.diagnostics, options?.maxErrorBytes).pipe(
+              Effect.map((x) => x.buffer.toString("utf8")),
+            ),
           )
-          const source = options?.includeStderr === true ? handle.all : handle.stdout
-          const lines = source.pipe(
+          const lines = streams.source.pipe(
             Stream.decodeText,
             Stream.splitLines,
             Stream.filter((line) => line.length > 0),