effect(core-transport): add Stream-shaped factories alongside legacy classes

Introduce TransportEffect interface and *TransportEffect factories
(memory, stdio, websocket) that expose inbound frames as
Stream<WireFrame, TaggedTransportError> and Effects for send/close.
Legacy MemoryTransport / StdioTransport / WebSocketTransport classes
are unchanged — the new factories adapt them, preserving the published
Transport contract.

Adds TaggedTransportError (Schema.TaggedError, cause + optional kind)
since the §12 ARCP error catalog does not cover the transport seam.

Closes #41
Closes #21

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Nick Ficano

packages/core/src/errors-tagged.ts

@@ -163,6 +163,27 @@ export class TaggedAgentVersionNotAvailable extends Schema.TaggedError<TaggedAge
   readonly code = "AGENT_VERSION_NOT_AVAILABLE" as const;
 }
 
+/**
+ * Transport-layer failure surfaced through Effect's typed-error channel.
+ *
+ * Not part of the §12 ARCP error catalog — `Transport` is the seam below
+ * protocol logic, so its failures are categorically distinct from the
+ * `TaggedSdkError` union. Effect-shaped transports (`memoryTransportEffect`,
+ * `stdioTransportEffect`, `websocketTransportEffect`) fail their `incoming`
+ * stream and `send` Effect with this error.
+ *
+ * `kind` is a free-form, opt-in tag for upstream pattern matching
+ * (`"send"`, `"receive"`, `"parse"`, `"closed"`, etc.). `cause` carries the
+ * underlying defect (typically a Node `Error` from `ws` / `readline`).
+ */
+export class TaggedTransportError extends Schema.TaggedError<TaggedTransportError>()(
+  "TransportError",
+  {
+    cause: Schema.Defect,
+    kind: Schema.optional(Schema.String),
+  },
+) {}
+
 /**
  * Discriminated union of every Effect-native ARCP error. Mirrors `SdkError`
  * but in the typed-error channel of an `Effect`.

packages/core/src/index.ts

@@ -68,6 +68,7 @@ export {
   TaggedResumeWindowExpired,
   type TaggedSdkError,
   TaggedTimeout,
+  TaggedTransportError,
   TaggedUnauthenticated,
 } from "./errors-tagged.js";
 export {

packages/core/src/transport/index.ts

@@ -1,10 +1,20 @@
-export { MemoryTransport, pairMemoryTransports } from "./memory.js";
-export { StdioTransport } from "./stdio.js";
+export {
+  MemoryTransport,
+  memoryTransportEffect,
+  pairMemoryTransports,
+  pairMemoryTransportsEffect,
+} from "./memory.js";
+export { StdioTransport, stdioTransportEffect } from "./stdio.js";
 export type {
   FrameHandler,
   SendableFrame,
   Transport,
+  TransportEffect,
   WebSocketServerHandle,
   WireFrame,
 } from "./types.js";
-export { startWebSocketServer, WebSocketTransport } from "./websocket.js";
+export {
+  startWebSocketServer,
+  WebSocketTransport,
+  websocketTransportEffect,
+} from "./websocket.js";

packages/core/src/transport/memory.ts

@@ -1,6 +1,15 @@
+import { Effect, Stream } from "effect";
+
+import { TaggedTransportError } from "../errors-tagged.js";
 import { InvalidRequestError } from "../errors.js";
 
-import type { FrameHandler, SendableFrame, Transport } from "./types.js";
+import type {
+  FrameHandler,
+  SendableFrame,
+  Transport,
+  TransportEffect,
+  WireFrame,
+} from "./types.js";
 
 /**
  * Two transports sharing a Promise-coupled queue. Used by tests to drive a
@@ -99,3 +108,55 @@ export function pairMemoryTransports(): [MemoryTransport, MemoryTransport] {
   b.connect(a);
   return [a, b];
 }
+
+/**
+ * Effect-shaped factory that wraps a legacy {@link MemoryTransport} as a
+ * {@link TransportEffect}. The legacy class API is unchanged; this adapter
+ * exposes an `incoming` {@link Stream.Stream} and an `Effect`-returning
+ * `send`/`close` for Effect-native call sites.
+ *
+ * Use {@link pairMemoryTransportsEffect} to get a wired pair. This factory
+ * is exposed standalone for callers that already own a `MemoryTransport`.
+ */
+export function memoryTransportEffect(
+  transport: MemoryTransport,
+): TransportEffect {
+  const incoming = Stream.async<WireFrame, TaggedTransportError>((emit) => {
+    transport.onFrame((frame) => {
+      void emit.single(frame);
+    });
+    transport.onClose((err) => {
+      if (err === undefined) {
+        void emit.end();
+      } else {
+        void emit.fail(
+          new TaggedTransportError({ cause: err, kind: "closed" }),
+        );
+      }
+    });
+  });
+  const send = (frame: SendableFrame): Effect.Effect<void, TaggedTransportError> =>
+    Effect.tryPromise({
+      try: () => transport.send(frame),
+      catch: (cause) => new TaggedTransportError({ cause, kind: "send" }),
+    });
+  return {
+    incoming,
+    send,
+    close: Effect.promise(() => transport.close()),
+    isClosed: () => transport.closed,
+  };
+}
+
+/**
+ * Effect-shaped twin of {@link pairMemoryTransports}. Returns
+ * `[clientSideEffect, serverSideEffect]` — two cross-wired
+ * {@link TransportEffect}s suitable for Effect-native tests.
+ */
+export function pairMemoryTransportsEffect(): readonly [
+  TransportEffect,
+  TransportEffect,
+] {
+  const [a, b] = pairMemoryTransports();
+  return [memoryTransportEffect(a), memoryTransportEffect(b)];
+}

packages/core/src/transport/stdio.ts

@@ -1,12 +1,16 @@
 import { createInterface, type Interface } from "node:readline";
 import type { Readable, Writable } from "node:stream";
 
+import { Effect, Stream } from "effect";
+
+import { TaggedTransportError } from "../errors-tagged.js";
 import { InvalidRequestError } from "../errors.js";
 
 import type {
   FrameHandler,
   SendableFrame,
   Transport,
+  TransportEffect,
   WireFrame,
 } from "./types.js";
 
@@ -130,3 +134,39 @@ export class StdioTransport implements Transport {
     }
   }
 }
+
+/**
+ * Effect-shaped factory that adapts a {@link StdioTransport} to a
+ * {@link TransportEffect}. The legacy class is unchanged; this wrapper
+ * exposes `incoming` as a {@link Stream.Stream} and returns `Effect`s for
+ * `send` and `close` so Effect-native pipelines can consume stdio frames.
+ */
+export function stdioTransportEffect(
+  transport: StdioTransport,
+): TransportEffect {
+  const incoming = Stream.async<WireFrame, TaggedTransportError>((emit) => {
+    transport.onFrame((frame) => {
+      void emit.single(frame);
+    });
+    transport.onClose((err) => {
+      if (err === undefined) {
+        void emit.end();
+      } else {
+        void emit.fail(
+          new TaggedTransportError({ cause: err, kind: "closed" }),
+        );
+      }
+    });
+  });
+  const send = (frame: SendableFrame): Effect.Effect<void, TaggedTransportError> =>
+    Effect.tryPromise({
+      try: () => transport.send(frame),
+      catch: (cause) => new TaggedTransportError({ cause, kind: "send" }),
+    });
+  return {
+    incoming,
+    send,
+    close: Effect.promise(() => transport.close()),
+    isClosed: () => transport.closed,
+  };
+}

packages/core/src/transport/types.ts

@@ -11,7 +11,10 @@
  * guarantee the protocol relies on.
  */
 
+import type { Effect, Stream } from "effect";
+
 import type { BaseEnvelope } from "../envelope.js";
+import type { TaggedTransportError } from "../errors-tagged.js";
 
 /** Raw message frame: a JSON-encodable object. */
 export type WireFrame = Record<string, unknown>;
@@ -54,6 +57,33 @@ export interface Transport {
   readonly closed: boolean;
 }
 
+/**
+ * Effect-shaped twin of {@link Transport}. Returned by the
+ * `*TransportEffect` factories (`memoryTransportEffect`,
+ * `stdioTransportEffect`, `websocketTransportEffect`).
+ *
+ * The legacy {@link Transport} class API is preserved for downstream
+ * consumers; this interface exists alongside it so Effect-native call sites
+ * can compose `incoming` with the rest of an Effect pipeline without an
+ * intermediary callback.
+ *
+ * `incoming` MUST emit frames in receive order. Backpressure is delegated
+ * to Stream consumers (each `Stream.async` factory uses an unbounded queue
+ * by default, matching the legacy class's buffering behavior).
+ */
+export interface TransportEffect {
+  /** Stream of inbound frames; terminates on peer close. */
+  readonly incoming: Stream.Stream<WireFrame, TaggedTransportError>;
+  /** Send a frame to the peer. Fails with `TaggedTransportError` on I/O error or after close. */
+  readonly send: (
+    frame: SendableFrame,
+  ) => Effect.Effect<void, TaggedTransportError>;
+  /** Close the transport. Idempotent; never fails. */
+  readonly close: Effect.Effect<void>;
+  /** Synchronous closed-state check, mirroring the legacy `closed` getter. */
+  readonly isClosed: () => boolean;
+}
+
 /** Resolved handle returned by `startWebSocketServer`. */
 export interface WebSocketServerHandle {
   /** Resolved port the server bound to. */

packages/core/src/transport/websocket.ts

@@ -1,11 +1,14 @@
+import { Effect, Stream } from "effect";
 import { WebSocket, WebSocketServer } from "ws";
 
+import { TaggedTransportError } from "../errors-tagged.js";
 import { InternalError, InvalidRequestError } from "../errors.js";
 
 import type {
   FrameHandler,
   SendableFrame,
   Transport,
+  TransportEffect,
   WebSocketServerHandle,
   WireFrame,
 } from "./types.js";
@@ -210,3 +213,79 @@ export async function startWebSocketServer(args: {
       }),
   };
 }
+
+/**
+ * Effect-shaped factory for a WebSocket transport. Wraps a raw `ws` socket
+ * (NOT a {@link WebSocketTransport} instance) so the inbound side is a
+ * `Stream<WireFrame, TaggedTransportError>` instead of a callback. The
+ * legacy {@link WebSocketTransport} class is unchanged and is the path
+ * currently used by the runtime / client.
+ *
+ * The socket is expected to be OPEN before this factory is called — see
+ * {@link WebSocketTransport.connect} for the open-handshake helper used on
+ * the client side.
+ */
+export function websocketTransportEffect(socket: WebSocket): TransportEffect {
+  const state = { closed: false };
+  return {
+    incoming: makeIncoming(socket, state),
+    send: (frame) => makeSend(socket, frame),
+    close: Effect.sync(() => {
+      if (state.closed) return;
+      state.closed = true;
+      if (
+        socket.readyState === WebSocket.OPEN ||
+        socket.readyState === WebSocket.CONNECTING
+      ) {
+        socket.close();
+      }
+    }),
+    isClosed: () => state.closed || socket.readyState === WebSocket.CLOSED,
+  };
+}
+
+function makeIncoming(
+  socket: WebSocket,
+  state: { closed: boolean },
+): Stream.Stream<WireFrame, TaggedTransportError> {
+  return Stream.async<WireFrame, TaggedTransportError>((emit) => {
+    const onMessage = (data: unknown, isBinary: boolean): void => {
+      if (isBinary) return;
+      const frame = parseWireFrame(data);
+      if (frame === null) return;
+      void emit.single(frame);
+    };
+    const onClose = (): void => {
+      state.closed = true;
+      void emit.end();
+    };
+    const onError = (err: Error): void => {
+      state.closed = true;
+      void emit.fail(new TaggedTransportError({ cause: err, kind: "receive" }));
+    };
+    socket.on("message", onMessage);
+    socket.on("close", onClose);
+    socket.on("error", onError);
+    return Effect.sync(() => {
+      socket.off("message", onMessage);
+      socket.off("close", onClose);
+      socket.off("error", onError);
+    });
+  });
+}
+
+function makeSend(
+  socket: WebSocket,
+  frame: SendableFrame,
+): Effect.Effect<void, TaggedTransportError> {
+  return Effect.tryPromise({
+    try: () =>
+      new Promise<void>((resolve, reject) => {
+        socket.send(JSON.stringify(frame), (err) => {
+          if (err) reject(err);
+          else resolve();
+        });
+      }),
+    catch: (cause) => new TaggedTransportError({ cause, kind: "send" }),
+  });
+}

packages/core/test/transport-memory.test.ts

@@ -1,6 +1,10 @@
+import { Effect, Stream } from "effect";
 import { describe, expect, it } from "vitest";
 
-import { pairMemoryTransports } from "@arcp/core";
+import {
+  pairMemoryTransports,
+  pairMemoryTransportsEffect,
+} from "@arcp/core";
 
 describe("pairMemoryTransports", () => {
   it("delivers frames in FIFO order from a to b", async () => {
@@ -46,3 +50,36 @@ describe("pairMemoryTransports", () => {
     await expect(a.send({})).rejects.toThrow();
   });
 });
+
+describe("pairMemoryTransportsEffect", () => {
+  it("delivers 10 frames a→b in order via Stream.take", async () => {
+    const [a, b] = pairMemoryTransportsEffect();
+    const take = Effect.runPromise(
+      Stream.runCollect(Stream.take(b.incoming, 10)),
+    );
+    for (let i = 0; i < 10; i++) {
+      await Effect.runPromise(a.send({ n: i }));
+    }
+    const chunk = await take;
+    const ns = [...chunk].map((f) => f["n"] as number);
+    expect(ns).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+  });
+
+  it("close on one side ends the peer's incoming stream", async () => {
+    const [a, b] = pairMemoryTransportsEffect();
+    const collected = Effect.runPromise(Stream.runCollect(b.incoming));
+    await Effect.runPromise(a.send({ k: "v" }));
+    await Effect.runPromise(a.close);
+    const chunk = await collected;
+    expect([...chunk]).toEqual([{ k: "v" }]);
+    expect(b.isClosed()).toBe(true);
+  });
+
+  it("send after close fails with TaggedTransportError", async () => {
+    const [a, b] = pairMemoryTransportsEffect();
+    void b;
+    await Effect.runPromise(a.close);
+    const exit = await Effect.runPromiseExit(a.send({ n: 1 }));
+    expect(exit._tag).toBe("Failure");
+  });
+});