✨ feat(node): add Stdio context API for host process stdio

node/mod.ts

@@ -1,2 +1,3 @@
 export * from "./stream.ts";
 export * from "./events.ts";
+export * from "./stdio.ts";

node/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/node",
   "description": "Node.js stream and event emitter adapters for Effection",
-  "version": "0.2.4",
+  "version": "0.3.0",
   "keywords": ["io", "streams"],
   "type": "module",
   "main": "./dist/mod.js",
@@ -26,6 +26,9 @@
       "default": "./dist/events.js"
     }
   },
+  "dependencies": {
+    "@effectionx/context-api": "workspace:*"
+  },
   "peerDependencies": {
     "effection": "^3 || ^4"
   },

node/stdio.test.ts

@@ -0,0 +1,118 @@
+import { describe, it } from "@effectionx/vitest";
+import { createSignal, each, scoped, type Stream } from "effection";
+import { expect } from "expect";
+
+import { Stdio, stderr, stdin, stdout } from "./stdio.ts";
+
+describe("Stdio middleware", () => {
+  it("captures stdout bytes via middleware", function* () {
+    const captured: Uint8Array[] = [];
+
+    yield* Stdio.around({
+      *stdout(args, next) {
+        captured.push(args[0]);
+        return yield* next(...args);
+      },
+    });
+
+    const bytes = new TextEncoder().encode("hello\n");
+    yield* stdout(bytes);
+
+    expect(captured.length).toBe(1);
+    expect(new TextDecoder().decode(captured[0])).toBe("hello\n");
+  });
+
+  it("captures stderr bytes via middleware", function* () {
+    const captured: Uint8Array[] = [];
+
+    yield* Stdio.around({
+      *stderr(args, next) {
+        captured.push(args[0]);
+        return yield* next(...args);
+      },
+    });
+
+    const bytes = new TextEncoder().encode("oops\n");
+    yield* stderr(bytes);
+
+    expect(captured.length).toBe(1);
+    expect(new TextDecoder().decode(captured[0])).toBe("oops\n");
+  });
+
+  it("can substitute stdin with a synthetic stream", function* () {
+    const signal = createSignal<Uint8Array, void>();
+    const synthetic: Stream<Uint8Array, void> = signal;
+
+    yield* Stdio.around({
+      *stdin(_args, _next) {
+        return synthetic;
+      },
+    });
+
+    const stream = yield* stdin();
+    const subscription = yield* stream;
+
+    const encoder = new TextEncoder();
+    signal.send(encoder.encode("one"));
+    signal.send(encoder.encode("two"));
+    signal.close();
+
+    const chunks: string[] = [];
+    const decoder = new TextDecoder();
+    let result = yield* subscription.next();
+    while (!result.done) {
+      chunks.push(decoder.decode(result.value));
+      result = yield* subscription.next();
+    }
+
+    expect(chunks).toEqual(["one", "two"]);
+  });
+
+  it("middleware is scoped and does not leak", function* () {
+    const outerCalls: string[] = [];
+
+    yield* Stdio.around({
+      *stdout(args, next) {
+        outerCalls.push("outer");
+        return yield* next(...args);
+      },
+    });
+
+    const bytes = new TextEncoder().encode("hi\n");
+    yield* stdout(bytes);
+    expect(outerCalls).toEqual(["outer"]);
+
+    yield* scoped(function* () {
+      const innerCalls: string[] = [];
+
+      yield* Stdio.around({
+        *stdout(args, next) {
+          innerCalls.push("inner");
+          return yield* next(...args);
+        },
+      });
+
+      yield* stdout(bytes);
+      expect(outerCalls).toEqual(["outer", "outer"]);
+      expect(innerCalls).toEqual(["inner"]);
+    });
+
+    outerCalls.length = 0;
+    yield* stdout(bytes);
+    expect(outerCalls).toEqual(["outer"]);
+  });
+});
+
+describe("Stdio defaults", () => {
+  it("reads from process.stdin by default (subscription acquires without error)", function* () {
+    yield* scoped(function* () {
+      // Default handler wraps process.stdin via fromReadable. We just
+      // verify that acquiring a subscription and then letting the scope
+      // tear it down does not throw; we can't easily assert on real
+      // host stdin bytes in a unit test.
+      const stream = yield* stdin();
+      const _sub = yield* stream;
+      expect(true).toBe(true);
+    });
+  });
+});

node/stdio.ts

@@ -0,0 +1,66 @@
+import process from "node:process";
+import { type Api, createApi } from "@effectionx/context-api";
+import type { Operation, Stream } from "effection";
+import { fromReadable } from "./stream.ts";
+
+/**
+ * Middleware-capable shape for host-process stdio.
+ *
+ * `stdin` yields a readable byte stream sourced from the host's standard
+ * input. `stdout` and `stderr` take bytes and write them to the host's
+ * corresponding output streams.
+ */
+export interface StdioApi {
+  stdin(): Operation<Stream<Uint8Array, void>>;
+  stdout(bytes: Uint8Array): Operation<void>;
+  stderr(bytes: Uint8Array): Operation<void>;
+}
+
+/**
+ * Context API used to observe or customize the host process's stdio.
+ *
+ * By default, `stdin` reads from `process.stdin`, and `stdout` / `stderr`
+ * write to `process.stdout` / `process.stderr`. Middleware can wrap this API
+ * via `Stdio.around(...)` to capture, transform, or redirect bytes — useful
+ * for tests that assert what was written to stdout, or harnesses that feed
+ * synthesized stdin.
+ *
+ * This is distinct from `@effectionx/process`'s `Stdio`, which governs child
+ * process stdio.
+ *
+ * @example
+ * ```ts
+ * import { main } from "effection";
+ * import { Stdio, stdout } from "@effectionx/node";
+ *
+ * await main(function* () {
+ *   const captured: Uint8Array[] = [];
+ *
+ *   yield* Stdio.around({
+ *     *stdout(args, next) {
+ *       captured.push(args[0]);
+ *       return yield* next(...args);
+ *     },
+ *   });
+ *
+ *   yield* stdout(new TextEncoder().encode("hello\n"));
+ *   // bytes flow into `captured` instead of the terminal
+ * });
+ * ```
+ */
+export const Stdio: Api<StdioApi> = createApi<StdioApi>(
+  "@effectionx/node/stdio",
+  {
+    *stdin() {
+      return fromReadable(process.stdin);
+    },
+    *stdout(bytes) {
+      process.stdout.write(bytes);
+    },
+    *stderr(bytes) {
+      process.stderr.write(bytes);
+    },
+  },
+);
+
+export const { stdin, stdout, stderr } = Stdio.operations;

node/tsconfig.json

@@ -9,6 +9,9 @@
   "references": [
     {
       "path": "../vitest"
+    },
+    {
+      "path": "../context-api"
     }
   ]
 }