process logging middleware

process/README.md

@@ -48,7 +48,7 @@ await main(function* () {
 
   // Stream stdout in real-time
   yield* spawn(function* () {
-    for (let chunk of yield* each(yield* process.stdout)) {
+    for (let chunk of yield* each(process.stdout)) {
       console.log(chunk);
       yield* each.next();
     }
@@ -59,6 +59,33 @@ await main(function* () {
 });
 ```
 
+### Handling Process Output With Middleware
+
+By default, we log the output, but you can remove or add additional handling of output lines per `stdout` and `stderr`.
+
+```typescript
+import { each, main, spawn } from "effection";
+import { exec } from "@effectionx/process";
+
+await main(function* () {
+  let process = yield* exec("npm install");
+
+  yield* process.around({
+    *stdout(line) {
+      // it does this by default
+      process.stdout.write(line);
+    },
+    *stderr(line) {
+      // it does this by default
+      process.stderr.write(line);
+    },
+  });
+
+  // Wait for the process to complete
+  yield* process.expect();
+});
+```
+
 ### Sending Input to stdin
 
 Write to a process's stdin:

process/mod.ts

@@ -1,2 +1,3 @@
 export * from "./src/exec.ts";
 export { type Daemon, daemon } from "./src/daemon.ts";
+export * from "./src/api.ts";

process/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/process",
   "description": "Spawn and manage child processes with structured concurrency",
-  "version": "0.7.4",
+  "version": "0.8.0",
   "keywords": ["process"],
   "type": "module",
   "main": "./dist/mod.js",
@@ -28,7 +28,9 @@
   },
   "sideEffects": false,
   "dependencies": {
+    "@effectionx/context-api": "workspace:*",
     "@effectionx/node": "workspace:*",
+    "@effectionx/scope-eval": "workspace:*",
     "cross-spawn": "^7",
     "ctrlc-windows": "^2",
     "shellwords-ts": "^3.0.1"

process/src/api.ts

@@ -0,0 +1,48 @@
+import { createApi } from "@effectionx/context-api";
+import type { StdioApi } from "./exec/types.ts";
+
+/**
+ * Context API used to observe or customize process stdio handling.
+ *
+ * By default, `stdout` and `stderr` are written directly to the host process
+ * streams. Middleware can wrap this API via `Stdio.around(...)` to capture,
+ * transform, or redirect child process output.
+ *
+ * @example
+ * ```ts
+ * import { main } from "effection";
+ * import { Stdio, exec } from "@effectionx/process";
+ *
+ * await main(function* () {
+ *   let outputStdout: Uint8Array[] = [];
+ *   let outputStderr: Uint8Array[] = [];
+ *
+ *   // affects child processes in this scope
+ *   // and all child scopes unless overridden
+ *   yield* Stdio.around({
+ *     *stdout(line, next) {
+ *       const [bytes] = line;
+ *       outputStdout.push(bytes);
+ *       return yield* next(line);
+ *     },
+ *     *stderr(line, next) {
+ *       const [bytes] = line;
+ *       outputStderr.push(bytes);
+ *       return yield* next(line);
+ *     },
+ *   });
+ *
+ *   yield* exec("node ./fixtures/hello-world.js", {
+ *     cwd: import.meta.dirname,
+ *   }).expect();
+ * });
+ * ```
+ */
+export const Stdio = createApi<StdioApi>("process:io", {
+  *stdout(line) {
+    process.stdout.write(line);
+  },
+  *stderr(line) {
+    process.stderr.write(line);
+  },
+});

process/src/exec.ts

@@ -7,15 +7,22 @@ import type {
   ExitStatus,
   Process,
   ProcessResult,
-} from "./exec/api.ts";
+} from "./exec/types.ts";
 import { createPosixProcess } from "./exec/posix.ts";
 import { createWin32Process, isWin32 } from "./exec/win32.ts";
 
-export * from "./exec/api.ts";
+export * from "./exec/types.ts";
 export * from "./exec/error.ts";
 
 export interface Exec extends Operation<Process> {
+  /**
+   * Wait for process completion and return exit status plus captured output.
+   */
   join(): Operation<ProcessResult>;
+
+  /**
+   * Like `join()`, but throws if the process exits unsuccessfully.
+   */
   expect(): Operation<ProcessResult>;
 }
 
@@ -31,6 +38,21 @@ const createProcess: CreateOSProcess = (cmd, opts) => {
  * that have a finite lifetime and on which you may wish to synchronize on the
  * exit status. If you want to start a process like a server that spins up and runs
  * forever, consider using `daemon()`
+ *
+ * @example
+ * ```ts
+ * import { main } from "effection";
+ * import { exec } from "@effectionx/process";
+ *
+ * await main(function* () {
+ *   let process = yield* exec("node ./fixtures/hello-world.js", {
+ *     cwd: import.meta.dirname,
+ *   })
+ *   let result = yield* process.expect();
+ *
+ *   console.log(result.code); // 0
+ * });
+ * ```
  */
 export function exec(command: string, options: ExecOptions = {}): Exec {
   let [cmd, ...args] = options.shell ? [command] : shellwords.split(command);

process/src/exec/error.ts

@@ -1,4 +1,4 @@
-import type { ExecOptions, ExitStatus } from "./api.ts";
+import type { ExecOptions, ExitStatus } from "./types.ts";
 
 export class ExecError extends Error {
   status: ExitStatus;

process/src/exec/posix.ts

@@ -1,26 +1,39 @@
 import { spawn as spawnProcess } from "node:child_process";
 import process from "node:process";
-import { once } from "@effectionx/node/events";
-import { fromReadable } from "@effectionx/node/stream";
 import {
+  type Result,
+  type Operation,
+  type Yielded,
   Err,
   Ok,
-  type Result,
   all,
   createSignal,
-  resource,
+  ensure,
   spawn,
   withResolvers,
 } from "effection";
-import type { CreateOSProcess, ExitStatus, Writable } from "./api.ts";
+import { unbox, useEvalScope } from "@effectionx/scope-eval";
+import { once } from "@effectionx/node/events";
+import { fromReadable } from "@effectionx/node/stream";
+import type {
+  CreateOSProcess,
+  ExecOptions,
+  ExitStatus,
+  Process,
+  Writable,
+} from "./types.ts";
+import { Stdio } from "../api.ts";
 import { ExecError } from "./error.ts";
 
 type ProcessResultValue = [number?, string?];
 
-export const createPosixProcess: CreateOSProcess = (command, options) => {
-  return resource(function* (provide) {
-    let processResult = withResolvers<Result<ProcessResultValue>>();
-
+export function* createPosixProcess(
+  command: string,
+  options: ExecOptions,
+): Operation<Process> {
+  let processResult = withResolvers<Result<ProcessResultValue>>();
+  const evalScope = yield* useEvalScope();
+  const result = yield* evalScope.eval(function* () {
     // Killing all child processes started by this command is surprisingly
     // tricky. If a process spawns another processes and we kill the parent,
     // then the child process is NOT automatically killed. Instead we're using
@@ -58,6 +71,7 @@ export const createPosixProcess: CreateOSProcess = (command, options) => {
     yield* spawn(function* () {
       let next = yield* io.stdout.next();
       while (!next.done) {
+        yield* Stdio.operations.stdout(next.value);
         stdout.send(next.value);
         next = yield* io.stdout.next();
       }
@@ -68,6 +82,7 @@ export const createPosixProcess: CreateOSProcess = (command, options) => {
     yield* spawn(function* () {
       let next = yield* io.stderr.next();
       while (!next.done) {
+        yield* Stdio.operations.stderr(next.value);
         stderr.send(next.value);
         next = yield* io.stderr.next();
       }
@@ -108,26 +123,32 @@ export const createPosixProcess: CreateOSProcess = (command, options) => {
       return status;
     }
 
-    try {
-      yield* provide({
-        pid: pid as number,
-        stdin,
-        stdout,
-        stderr,
-        join,
-        expect,
-      });
-    } finally {
+    yield* ensure(function* () {
       try {
         if (typeof childProcess.pid === "undefined") {
-          // biome-ignore lint/correctness/noUnsafeFinally: Intentional error for missing PID
           throw new Error("no pid for childProcess");
         }
         process.kill(-childProcess.pid, "SIGTERM");
         yield* all([io.stdoutDone.operation, io.stderrDone.operation]);
       } catch (_e) {
         // do nothing, process is probably already dead
       }
-    }
+    });
+
+    return {
+      pid: pid as number,
+      *around(
+        ...args: Parameters<typeof Stdio.around>
+      ): ReturnType<typeof Stdio.around> {
+        const result = yield* evalScope.eval(() => Stdio.around(...args));
+        return unbox(result);
+      },
+      stdin,
+      stdout,
+      stderr,
+      join,
+      expect,
+    } satisfies Yielded<ReturnType<CreateOSProcess>>;
   });
-};
+  return unbox(result);
+}

process/src/exec/api.ts → process/src/exec/types.ts

@@ -1,7 +1,10 @@
 import type { Operation } from "effection";
 import type { OutputStream } from "../helpers.ts";
+import type { Api } from "@effectionx/context-api";
 
-// TODO: import from subscription package once #236 is merged
+/**
+ * Writable handle used for process stdin.
+ */
 export interface Writable<T> {
   send(message: T): void;
 }
@@ -11,6 +14,7 @@ export interface Writable<T> {
  * standard io handles, and methods for synchronizing on return.
  */
 export interface Process extends StdIO {
+  /** Child process id as reported by the operating system. */
   readonly pid: number;
 
   /**
@@ -24,8 +28,42 @@ export interface Process extends StdIO {
    * not complete successfully, it will raise an ExecError.
    */
   expect(): Operation<ExitStatus>;
+
+  /**
+   * Middleware entrypoint for wrapping stdio behavior using `Stdio` middleware API.
+   *
+   * @example
+   * ```ts
+   * import { main } from "effection";
+   * import { exec } from "@effectionx/process";
+   *
+   * await main(function* () {
+   *   let proc = yield* exec("node ./fixtures/hello-world.js", {
+   *     cwd: import.meta.dirname,
+   *   });
+   *
+   *   let chunks: Uint8Array[] = [];
+   *   yield* proc.around({
+   *     *stdout(line, next) {
+   *       // handle bytes as required
+   *       const [bytes] = line;
+   *       chunks.push(bytes);
+   *       // optionally continue with next middleware
+   *       return yield* next(line);
+   *     },
+   *   });
+   *
+   *   yield* proc.expect();
+   *   console.log(chunks.toString());
+   * });
+   * ```
+   */
+  around: Api<StdioApi>["around"];
 }
 
+/**
+ * Options for spawning a child process.
+ */
 export interface ExecOptions {
   /**
    * When not using passing the `shell` option all arguments must be passed
@@ -54,15 +92,24 @@ export interface ExecOptions {
 }
 
 export interface StdIO {
+  /** Stream of bytes written by the process to standard output. */
   stdout: OutputStream;
+
+  /** Stream of bytes written by the process to standard error. */
   stderr: OutputStream;
+
+  /** Writable interface for sending data to process standard input. */
   stdin: Writable<string>;
 }
 
+export interface StdioApi {
+  stdout(bytes: Uint8Array): Operation<void>;
+  stderr(bytes: Uint8Array): Operation<void>;
+}
+
 export interface ExitStatus {
   /**
    * exit code
-   * //TODO: is this pertinent on Windows? Do we need an 'OK' flag
    */
   code?: number;
 
@@ -74,7 +121,10 @@ export interface ExitStatus {
 }
 
 export interface ProcessResult extends ExitStatus {
+  /** Collected stdout text from process execution helpers. */
   stdout: string;
+
+  /** Collected stderr text from process execution helpers. */
   stderr: string;
 }
 export type CreateOSProcess = (