diff --git a/packages/kernel-node-runtime/CHANGELOG.md b/packages/kernel-node-runtime/CHANGELOG.md
index 934b2c4f7d..5a75a45781 100644
--- a/packages/kernel-node-runtime/CHANGELOG.md
+++ b/packages/kernel-node-runtime/CHANGELOG.md
@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **BREAKING:** `makeIOChannelFactory` is now `makeIOListenerFactory`, and `makeSocketIOChannel` is now `makeSocketIOListener`. The Unix-socket server hands each connection to `accept()` as its own `IOChannel`, whose receive buffer, decoder, line queue, and reader queue are local to that connection, so any number of peers can be served concurrently. Connections arriving before `accept()` is called are queued rather than dropped. Gone with the single-client design: the shared `currentSocket`, the session-boundary latch, the merged line queue, and the `socket.destroy()` that rejected every second connection ([#1007](https://github.com/MetaMask/ocap-kernel/pull/1007))
 - **BREAKING:** Drop `platformOptions.fetch` from `makeNodeJsVatSupervisor` ([#942](https://github.com/MetaMask/ocap-kernel/pull/942))
   - `fetch` is now a vat endowment; stub `globalThis.fetch` directly if needed
 
diff --git a/packages/kernel-node-runtime/src/index.ts b/packages/kernel-node-runtime/src/index.ts
index 1a1eeb323c..a05ad038f8 100644
--- a/packages/kernel-node-runtime/src/index.ts
+++ b/packages/kernel-node-runtime/src/index.ts
@@ -2,4 +2,4 @@ export { NodejsPlatformServices } from './kernel/PlatformServices.ts';
 export { makeKernel } from './kernel/make-kernel.ts';
 export type { MakeKernelResult } from './kernel/make-kernel.ts';
 export { makeNodeJsVatSupervisor } from './vat/make-supervisor.ts';
-export { makeIOChannelFactory, makeSocketIOChannel } from './io/index.ts';
+export { makeIOListenerFactory, makeSocketIOListener } from './io/index.ts';
diff --git a/packages/kernel-node-runtime/src/io/index.ts b/packages/kernel-node-runtime/src/io/index.ts
index 739aa7757d..2bdfd38e2e 100644
--- a/packages/kernel-node-runtime/src/io/index.ts
+++ b/packages/kernel-node-runtime/src/io/index.ts
@@ -1,23 +1,23 @@
-import type { IOChannelFactory, IOConfig } from '@metamask/ocap-kernel';
+import type { IOListenerFactory, IOConfig } from '@metamask/ocap-kernel';
 
-import { makeSocketIOChannel } from './socket-channel.ts';
+import { makeSocketIOListener } from './socket-listener.ts';
 
-export { makeSocketIOChannel } from './socket-channel.ts';
+export { makeSocketIOListener } from './socket-listener.ts';
 
 /**
- * Create an IOChannelFactory for the Node.js environment.
- * Dispatches on `config.type` to the appropriate channel implementation.
+ * Create an IOListenerFactory for the Node.js environment.
+ * Dispatches on `config.type` to the appropriate listener implementation.
  *
- * @returns An IOChannelFactory.
+ * @returns An IOListenerFactory.
  */
-export function makeIOChannelFactory(): IOChannelFactory {
+export function makeIOListenerFactory(): IOListenerFactory {
   return async (name: string, config: IOConfig) => {
     switch (config.type) {
       case 'socket':
-        return makeSocketIOChannel(name, config.path);
+        return makeSocketIOListener(name, config.path);
       default:
         throw new Error(
-          `Unsupported IO channel type "${config.type}" for channel "${name}"`,
+          `Unsupported IO listener type "${config.type}" for listener "${name}"`,
         );
     }
   };
diff --git a/packages/kernel-node-runtime/src/io/socket-channel.test.ts b/packages/kernel-node-runtime/src/io/socket-channel.test.ts
deleted file mode 100644
index fe8bf982c9..0000000000
--- a/packages/kernel-node-runtime/src/io/socket-channel.test.ts
+++ /dev/null
@@ -1,301 +0,0 @@
-import type { IOChannel } from '@metamask/ocap-kernel';
-import fs from 'node:fs/promises';
-import * as net from 'node:net';
-import * as os from 'node:os';
-import * as path from 'node:path';
-import { describe, it, expect, afterEach } from 'vitest';
-
-import { makeSocketIOChannel } from './socket-channel.ts';
-
-function tempSocketPath(): string {
-  return path.join(
-    os.tmpdir(),
-    `io-test-${Date.now()}-${Math.random().toString(36).slice(2)}.sock`,
-  );
-}
-
-async function connectToSocket(socketPath: string): Promise<net.Socket> {
-  return new Promise((resolve, reject) => {
-    const client = net.createConnection(socketPath, () => {
-      client.removeListener('error', reject);
-      resolve(client);
-    });
-    client.on('error', reject);
-  });
-}
-
-async function writeLine(socket: net.Socket, line: string): Promise<void> {
-  return new Promise((resolve, reject) => {
-    socket.write(`${line}\n`, (error) => {
-      if (error) {
-        reject(error);
-      } else {
-        resolve();
-      }
-    });
-  });
-}
-
-async function readLine(socket: net.Socket): Promise<string> {
-  return new Promise((resolve) => {
-    let buffer = '';
-    const onData = (data: Buffer): void => {
-      buffer += data.toString();
-      const idx = buffer.indexOf('\n');
-      if (idx !== -1) {
-        socket.removeListener('data', onData);
-        resolve(buffer.slice(0, idx));
-      }
-    };
-    socket.on('data', onData);
-  });
-}
-
-async function fileExists(filePath: string): Promise<boolean> {
-  try {
-    await fs.access(filePath);
-    return true;
-  } catch {
-    return false;
-  }
-}
-
-describe('makeSocketIOChannel', () => {
-  const channels: IOChannel[] = [];
-  const clients: net.Socket[] = [];
-
-  afterEach(async () => {
-    for (const client of clients) {
-      client.destroy();
-    }
-    clients.length = 0;
-    for (const channel of channels) {
-      await channel.close();
-    }
-    channels.length = 0;
-  });
-
-  it('creates a listening socket', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel);
-
-    expect(await fileExists(socketPath)).toBe(true);
-  });
-
-  it('reads lines from a connected client', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel);
-
-    const client = await connectToSocket(socketPath);
-    clients.push(client);
-
-    await writeLine(client, 'hello');
-    await writeLine(client, 'world');
-
-    const line1 = await channel.read();
-    const line2 = await channel.read();
-
-    expect(line1).toBe('hello');
-    expect(line2).toBe('world');
-  });
-
-  it('writes lines to a connected client', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel);
-
-    const client = await connectToSocket(socketPath);
-    clients.push(client);
-
-    // Small delay for connection to be established
-    await new Promise((resolve) => setTimeout(resolve, 10));
-
-    const linePromise = readLine(client);
-    await channel.write('output');
-    const received = await linePromise;
-
-    expect(received).toBe('output');
-  });
-
-  it('returns null on client disconnect', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel);
-
-    const client = await connectToSocket(socketPath);
-
-    // Start a read that will block
-    const readPromise = channel.read();
-    client.destroy();
-
-    const result = await readPromise;
-    expect(result).toBeNull();
-  });
-
-  it('blocks read until a client connects and sends data', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel);
-
-    // Start read before any client connects — should block
-    const readPromise = channel.read();
-
-    // Connect and send data
-    const client = await connectToSocket(socketPath);
-    clients.push(client);
-    await writeLine(client, 'hello');
-
-    const result = await readPromise;
-    expect(result).toBe('hello');
-  });
-
-  it('throws on write when no client is connected', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel);
-
-    await expect(channel.write('data')).rejects.toThrow(
-      'has no connected client',
-    );
-  });
-
-  it('queues lines before read is called', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel);
-
-    const client = await connectToSocket(socketPath);
-    clients.push(client);
-
-    // Send lines before any reads
-    await writeLine(client, 'a');
-    await writeLine(client, 'b');
-
-    // Small delay for data to arrive
-    await new Promise((resolve) => setTimeout(resolve, 20));
-
-    expect(await channel.read()).toBe('a');
-    expect(await channel.read()).toBe('b');
-  });
-
-  it('rejects second connection', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel);
-
-    const client1 = await connectToSocket(socketPath);
-    clients.push(client1);
-
-    const client2 = await connectToSocket(socketPath);
-
-    // Second client should be destroyed
-    await new Promise<void>((resolve) => {
-      client2.on('close', () => resolve());
-    });
-    expect(client2.destroyed).toBe(true);
-  });
-
-  it('cleans up socket file on close', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-
-    expect(await fileExists(socketPath)).toBe(true);
-    await channel.close();
-    expect(await fileExists(socketPath)).toBe(false);
-  });
-
-  it('returns null after close', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-
-    await channel.close();
-
-    const result = await channel.read();
-    expect(result).toBeNull();
-  });
-
-  it('throws on write after close', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    const client = await connectToSocket(socketPath);
-    clients.push(client);
-
-    await channel.close();
-
-    await expect(channel.write('data')).rejects.toThrow('is closed');
-  });
-
-  it('drains stale lineQueue when a new client connects', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel);
-
-    // First client sends lines that are not read
-    const client1 = await connectToSocket(socketPath);
-    await writeLine(client1, 'stale-line');
-    await new Promise((resolve) => setTimeout(resolve, 20));
-
-    // Disconnect first client
-    client1.destroy();
-    await new Promise((resolve) => setTimeout(resolve, 20));
-
-    // Second client connects — stale lines should be gone
-    const client2 = await connectToSocket(socketPath);
-    clients.push(client2);
-
-    await writeLine(client2, 'fresh-line');
-    await new Promise((resolve) => setTimeout(resolve, 20));
-
-    expect(await channel.read()).toBe('fresh-line');
-  });
-
-  it('handles multi-byte UTF-8 split across TCP chunks', async () => {
-    const socketPath = tempSocketPath();
-    const channel = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel);
-
-    const client = await connectToSocket(socketPath);
-    clients.push(client);
-
-    // U+1F600 (😀) is 4 bytes: f0 9f 98 80
-    const emoji = '\u{1F600}';
-    const fullMessage = `hello ${emoji} world\n`;
-    const encoded = Buffer.from(fullMessage, 'utf8');
-
-    // Split in the middle of the emoji (after first 2 bytes of the 4-byte sequence)
-    const splitPoint = Buffer.from('hello ', 'utf8').length + 2;
-    const chunk1 = encoded.subarray(0, splitPoint);
-    const chunk2 = encoded.subarray(splitPoint);
-
-    // Send the two chunks separately
-    await new Promise<void>((resolve, reject) => {
-      client.write(chunk1, (error) => (error ? reject(error) : resolve()));
-    });
-    await new Promise((resolve) => setTimeout(resolve, 10));
-    await new Promise<void>((resolve, reject) => {
-      client.write(chunk2, (error) => (error ? reject(error) : resolve()));
-    });
-    await new Promise((resolve) => setTimeout(resolve, 10));
-
-    expect(await channel.read()).toBe(`hello ${emoji} world`);
-  });
-
-  it('removes stale socket file on creation', async () => {
-    const socketPath = tempSocketPath();
-
-    // Create the first channel
-    const channel1 = await makeSocketIOChannel('test', socketPath);
-    await channel1.close();
-
-    // Recreate a stale file
-    await fs.writeFile(socketPath, '');
-
-    // Should succeed despite the stale file
-    const channel2 = await makeSocketIOChannel('test', socketPath);
-    channels.push(channel2);
-
-    expect(await fileExists(socketPath)).toBe(true);
-  });
-});
diff --git a/packages/kernel-node-runtime/src/io/socket-channel.ts b/packages/kernel-node-runtime/src/io/socket-channel.ts
deleted file mode 100644
index 3acda62f58..0000000000
--- a/packages/kernel-node-runtime/src/io/socket-channel.ts
+++ /dev/null
@@ -1,191 +0,0 @@
-import type { IOChannel } from '@metamask/ocap-kernel';
-import fs from 'node:fs/promises';
-import * as net from 'node:net';
-import { StringDecoder } from 'node:string_decoder';
-
-type PendingReader = {
-  resolve: (value: string | null) => void;
-};
-
-/**
- * Create an IOChannel backed by a Unix domain socket.
- *
- * Creates a `net.Server` listening on the configured socket path.
- * Accepts one connection at a time. Lines are `\n`-delimited.
- *
- * @param name - The channel name (for diagnostics).
- * @param socketPath - The file path for the Unix domain socket.
- * @returns A promise for the IOChannel, resolved once the server is listening.
- */
-export async function makeSocketIOChannel(
-  name: string,
-  socketPath: string,
-): Promise<IOChannel> {
-  const lineQueue: string[] = [];
-  const readerQueue: PendingReader[] = [];
-  let currentSocket: net.Socket | null = null;
-  let decoder = new StringDecoder('utf8');
-  let buffer = '';
-  let closed = false;
-
-  /**
-   * Deliver a line to a pending reader or enqueue it.
-   *
-   * @param line - The line to deliver.
-   */
-  function deliverLine(line: string): void {
-    const reader = readerQueue.shift();
-    if (reader) {
-      reader.resolve(line);
-    } else {
-      lineQueue.push(line);
-    }
-  }
-
-  /**
-   * Handle the end of the input stream.
-   */
-  function deliverEOF(): void {
-    while (readerQueue.length > 0) {
-      const reader = readerQueue.shift();
-      reader?.resolve(null);
-    }
-  }
-
-  /**
-   * Handle incoming data by splitting on newlines.
-   *
-   * @param data - The raw data buffer from the socket.
-   */
-  function handleData(data: Buffer): void {
-    buffer += decoder.write(data);
-    let newlineIndex = buffer.indexOf('\n');
-    while (newlineIndex !== -1) {
-      const line = buffer.slice(0, newlineIndex);
-      buffer = buffer.slice(newlineIndex + 1);
-      deliverLine(line);
-      newlineIndex = buffer.indexOf('\n');
-    }
-  }
-
-  /**
-   * Handle the channel disconnecting.
-   *
-   * @param socket - The socket that disconnected.
-   */
-  function handleDisconnect(socket: net.Socket): void {
-    if (currentSocket !== socket) {
-      return;
-    }
-    // Flush any incomplete multi-byte sequence from the decoder
-    buffer += decoder.end();
-    // Deliver any remaining buffered data as a final line
-    if (buffer.length > 0) {
-      deliverLine(buffer);
-      buffer = '';
-    }
-    currentSocket = null;
-    deliverEOF();
-  }
-
-  const server = net.createServer((socket) => {
-    if (currentSocket) {
-      if (currentSocket.readableEnded || currentSocket.destroyed) {
-        // Old connection is dead but events haven't been fully processed;
-        // clean it up and accept the new connection.
-        currentSocket.removeAllListeners();
-        currentSocket.destroy();
-        currentSocket = null;
-      } else {
-        // Existing active client — reject the new connection
-        socket.destroy();
-        return;
-      }
-    }
-    // Drain stale data from any previous connection, but keep pending
-    // readers alive so they can receive data from the new connection.
-    lineQueue.length = 0;
-
-    currentSocket = socket;
-    decoder = new StringDecoder('utf8');
-    buffer = '';
-
-    socket.on('data', handleData);
-    socket.on('end', () => handleDisconnect(socket));
-    socket.on('error', () => handleDisconnect(socket));
-    socket.on('close', () => handleDisconnect(socket));
-  });
-
-  // Remove stale socket file if it exists
-  try {
-    await fs.unlink(socketPath);
-  } catch {
-    // Ignore if it doesn't exist
-  }
-
-  await new Promise<void>((resolve, reject) => {
-    server.on('error', reject);
-    server.listen(socketPath, () => {
-      server.removeListener('error', reject);
-      resolve();
-    });
-  });
-
-  const channel: IOChannel = {
-    async read(): Promise<string | null> {
-      if (closed) {
-        return null;
-      }
-      const queued = lineQueue.shift();
-      if (queued !== undefined) {
-        return queued;
-      }
-      // Block until data arrives (from a current or future client connection)
-      return new Promise<string | null>((resolve) => {
-        readerQueue.push({ resolve });
-      });
-    },
-
-    async write(data: string): Promise<void> {
-      if (closed) {
-        throw new Error(`IO channel "${name}" is closed`);
-      }
-      if (!currentSocket) {
-        throw new Error(`IO channel "${name}" has no connected client`);
-      }
-      const socket = currentSocket;
-      return new Promise<void>((resolve, reject) => {
-        socket.write(`${data}\n`, (error) => {
-          if (error) {
-            reject(error);
-          } else {
-            resolve();
-          }
-        });
-      });
-    },
-
-    async close(): Promise<void> {
-      if (closed) {
-        return;
-      }
-      closed = true;
-      deliverEOF();
-      if (currentSocket) {
-        currentSocket.destroy();
-        currentSocket = null;
-      }
-      await new Promise<void>((resolve) => {
-        server.close(() => resolve());
-      });
-      // Clean up socket file
-      try {
-        await fs.unlink(socketPath);
-      } catch {
-        // Ignore
-      }
-    },
-  };
-
-  return channel;
-}
diff --git a/packages/kernel-node-runtime/src/io/socket-listener.test.ts b/packages/kernel-node-runtime/src/io/socket-listener.test.ts
new file mode 100644
index 0000000000..101289f1a1
--- /dev/null
+++ b/packages/kernel-node-runtime/src/io/socket-listener.test.ts
@@ -0,0 +1,532 @@
+import type { IOChannel, IOListener } from '@metamask/ocap-kernel';
+import { EventEmitter } from 'node:events';
+import fs from 'node:fs/promises';
+import * as net from 'node:net';
+import * as os from 'node:os';
+import * as path from 'node:path';
+import { describe, it, expect, afterEach, vi } from 'vitest';
+
+import {
+  makeConnectionChannel,
+  makeSocketIOListener,
+} from './socket-listener.ts';
+
+function tempSocketPath(): string {
+  return path.join(
+    os.tmpdir(),
+    `io-test-${Date.now()}-${Math.random().toString(36).slice(2)}.sock`,
+  );
+}
+
+async function connectToSocket(socketPath: string): Promise<net.Socket> {
+  return new Promise((resolve, reject) => {
+    const client = net.createConnection(socketPath, () => {
+      client.removeListener('error', reject);
+      resolve(client);
+    });
+    client.on('error', reject);
+  });
+}
+
+async function writeLine(socket: net.Socket, line: string): Promise<void> {
+  return new Promise((resolve, reject) => {
+    socket.write(`${line}\n`, (error) => {
+      if (error) {
+        reject(error);
+      } else {
+        resolve();
+      }
+    });
+  });
+}
+
+async function readLine(socket: net.Socket): Promise<string> {
+  return new Promise((resolve) => {
+    let buffer = '';
+    const onData = (data: Buffer): void => {
+      buffer += data.toString();
+      const idx = buffer.indexOf('\n');
+      if (idx !== -1) {
+        socket.removeListener('data', onData);
+        resolve(buffer.slice(0, idx));
+      }
+    };
+    socket.on('data', onData);
+  });
+}
+
+async function fileExists(filePath: string): Promise<boolean> {
+  try {
+    await fs.access(filePath);
+    return true;
+  } catch {
+    return false;
+  }
+}
+
+const settle = async (ms = 20): Promise<void> =>
+  new Promise((resolve) => setTimeout(resolve, ms));
+
+describe('makeSocketIOListener', () => {
+  const listeners: IOListener[] = [];
+  const clients: net.Socket[] = [];
+
+  afterEach(async () => {
+    for (const client of clients) {
+      client.destroy();
+    }
+    clients.length = 0;
+    for (const listener of listeners) {
+      await listener.close();
+    }
+    listeners.length = 0;
+  });
+
+  /**
+   * Create a listener that is torn down after the test.
+   *
+   * @param socketPath - Path for the Unix domain socket.
+   * @returns The listener.
+   */
+  async function makeTracked(socketPath: string): Promise<IOListener> {
+    const listener = await makeSocketIOListener('test', socketPath);
+    listeners.push(listener);
+    return listener;
+  }
+
+  /**
+   * Connect a client that is destroyed after the test.
+   *
+   * @param socketPath - Path for the Unix domain socket.
+   * @returns The connected client socket.
+   */
+  async function connectTracked(socketPath: string): Promise<net.Socket> {
+    const client = await connectToSocket(socketPath);
+    clients.push(client);
+    return client;
+  }
+
+  it('creates a listening socket', async () => {
+    const socketPath = tempSocketPath();
+    await makeTracked(socketPath);
+
+    expect(await fileExists(socketPath)).toBe(true);
+  });
+
+  describe('accept()', () => {
+    it('yields a channel for a connecting peer', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const acceptPromise = listener.accept();
+      const client = await connectTracked(socketPath);
+      await writeLine(client, 'hello');
+
+      const channel = await acceptPromise;
+      expect(await channel?.read()).toBe('hello');
+    });
+
+    it('queues peers that connect before accept is called', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const client = await connectTracked(socketPath);
+      await writeLine(client, 'early');
+      await settle();
+
+      const channel = await listener.accept();
+      expect(await channel?.read()).toBe('early');
+    });
+
+    it('yields one channel per peer, in connection order', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const first = await connectTracked(socketPath);
+      await settle();
+      const second = await connectTracked(socketPath);
+      await settle();
+
+      const channelA = await listener.accept();
+      const channelB = await listener.accept();
+      await writeLine(first, 'from-first');
+      await writeLine(second, 'from-second');
+
+      expect(await channelA?.read()).toBe('from-first');
+      expect(await channelB?.read()).toBe('from-second');
+    });
+
+    it('returns null once the listener is closed', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeSocketIOListener('test', socketPath);
+
+      await listener.close();
+
+      expect(await listener.accept()).toBeNull();
+    });
+
+    it('releases a pending accept when the listener closes', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeSocketIOListener('test', socketPath);
+
+      const acceptPromise = listener.accept();
+      await listener.close();
+
+      expect(await acceptPromise).toBeNull();
+    });
+  });
+
+  describe('concurrent connections', () => {
+    it('serves several peers at once without mixing their data', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const alice = await connectTracked(socketPath);
+      await settle();
+      const bob = await connectTracked(socketPath);
+      await settle();
+
+      const aliceChannel = (await listener.accept()) as IOChannel;
+      const bobChannel = (await listener.accept()) as IOChannel;
+
+      // Interleave traffic from both peers.
+      await writeLine(alice, 'alice-1');
+      await writeLine(bob, 'bob-1');
+      await writeLine(alice, 'alice-2');
+      await writeLine(bob, 'bob-2');
+      await settle();
+
+      expect(await aliceChannel.read()).toBe('alice-1');
+      expect(await aliceChannel.read()).toBe('alice-2');
+      expect(await bobChannel.read()).toBe('bob-1');
+      expect(await bobChannel.read()).toBe('bob-2');
+    });
+
+    it('routes each write back to its own peer', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const alice = await connectTracked(socketPath);
+      await settle();
+      const bob = await connectTracked(socketPath);
+      await settle();
+
+      const aliceChannel = (await listener.accept()) as IOChannel;
+      const bobChannel = (await listener.accept()) as IOChannel;
+
+      const aliceHeard = readLine(alice);
+      const bobHeard = readLine(bob);
+      await aliceChannel.write('for-alice');
+      await bobChannel.write('for-bob');
+
+      expect(await aliceHeard).toBe('for-alice');
+      expect(await bobHeard).toBe('for-bob');
+    });
+
+    it('leaves one peer unaffected when another disconnects', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const doomed = await connectToSocket(socketPath);
+      await settle();
+      const survivor = await connectTracked(socketPath);
+      await settle();
+
+      const doomedChannel = (await listener.accept()) as IOChannel;
+      const survivorChannel = (await listener.accept()) as IOChannel;
+
+      doomed.destroy();
+      await settle();
+
+      expect(await doomedChannel.read()).toBeNull();
+      await writeLine(survivor, 'still-here');
+      expect(await survivorChannel.read()).toBe('still-here');
+    });
+  });
+
+  describe('connection channels', () => {
+    it('writes lines to its peer', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const client = await connectTracked(socketPath);
+      const channel = (await listener.accept()) as IOChannel;
+
+      const linePromise = readLine(client);
+      await channel.write('output');
+
+      expect(await linePromise).toBe('output');
+    });
+
+    it('queues lines that arrive before read is called', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const client = await connectTracked(socketPath);
+      const channel = (await listener.accept()) as IOChannel;
+
+      await writeLine(client, 'a');
+      await writeLine(client, 'b');
+      await settle();
+
+      expect(await channel.read()).toBe('a');
+      expect(await channel.read()).toBe('b');
+    });
+
+    it('returns null to a pending read when the peer disconnects', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const client = await connectToSocket(socketPath);
+      const channel = (await listener.accept()) as IOChannel;
+
+      const readPromise = channel.read();
+      client.destroy();
+
+      expect(await readPromise).toBeNull();
+    });
+
+    it('delivers buffered lines before reporting EOF', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const client = await connectToSocket(socketPath);
+      const channel = (await listener.accept()) as IOChannel;
+
+      await writeLine(client, 'last-words');
+      await settle();
+      client.destroy();
+      await settle();
+
+      // Data the peer sent before going away is not lost.
+      expect(await channel.read()).toBe('last-words');
+      expect(await channel.read()).toBeNull();
+    });
+
+    it('returns null after the channel is closed', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      await connectTracked(socketPath);
+      const channel = (await listener.accept()) as IOChannel;
+      await channel.close();
+
+      expect(await channel.read()).toBeNull();
+    });
+
+    it('discards buffered data on close rather than delivering it after EOF', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const client = await connectTracked(socketPath);
+      const channel = (await listener.accept()) as IOChannel;
+
+      // A complete line plus a trailing fragment with no newline.
+      await writeLine(client, 'buffered');
+      await new Promise<void>((resolve, reject) => {
+        client.write('partial-no-newline', (error) =>
+          error ? reject(error) : resolve(),
+        );
+      });
+      await settle();
+
+      await channel.close();
+
+      // Closing means the holder is done reading. Neither the queued line
+      // nor the trailing fragment may surface after EOF was signalled.
+      expect(await channel.read()).toBeNull();
+      expect(await channel.read()).toBeNull();
+    });
+
+    it('throws on write after the channel is closed', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      await connectTracked(socketPath);
+      const channel = (await listener.accept()) as IOChannel;
+      await channel.close();
+
+      await expect(channel.write('data')).rejects.toThrow('is closed');
+    });
+
+    it('handles multi-byte UTF-8 split across TCP chunks', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeTracked(socketPath);
+
+      const client = await connectTracked(socketPath);
+      const channel = (await listener.accept()) as IOChannel;
+
+      // U+1F600 (😀) is 4 bytes: f0 9f 98 80
+      const emoji = '\u{1F600}';
+      const encoded = Buffer.from(`hello ${emoji} world\n`, 'utf8');
+
+      // Split in the middle of the emoji (after 2 of its 4 bytes)
+      const splitPoint = Buffer.from('hello ', 'utf8').length + 2;
+
+      await new Promise<void>((resolve, reject) => {
+        client.write(encoded.subarray(0, splitPoint), (error) =>
+          error ? reject(error) : resolve(),
+        );
+      });
+      await settle(10);
+      await new Promise<void>((resolve, reject) => {
+        client.write(encoded.subarray(splitPoint), (error) =>
+          error ? reject(error) : resolve(),
+        );
+      });
+      await settle(10);
+
+      expect(await channel.read()).toBe(`hello ${emoji} world`);
+    });
+  });
+
+  describe('close()', () => {
+    it('cleans up the socket file', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeSocketIOListener('test', socketPath);
+
+      expect(await fileExists(socketPath)).toBe(true);
+      await listener.close();
+      expect(await fileExists(socketPath)).toBe(false);
+    });
+
+    it('closes the connections it handed out', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeSocketIOListener('test', socketPath);
+
+      await connectTracked(socketPath);
+      const channel = (await listener.accept()) as IOChannel;
+
+      await listener.close();
+
+      expect(await channel.read()).toBeNull();
+      await expect(channel.write('data')).rejects.toThrow('is closed');
+    });
+
+    it('is idempotent', async () => {
+      const socketPath = tempSocketPath();
+      const listener = await makeSocketIOListener('test', socketPath);
+
+      await listener.close();
+      expect(await listener.close()).toBeUndefined();
+    });
+  });
+
+  it('removes a stale socket file on creation', async () => {
+    const socketPath = tempSocketPath();
+
+    const first = await makeSocketIOListener('test', socketPath);
+    await first.close();
+
+    // Recreate a stale file
+    await fs.writeFile(socketPath, '');
+
+    // Should succeed despite the stale file
+    await makeTracked(socketPath);
+
+    expect(await fileExists(socketPath)).toBe(true);
+  });
+});
+
+describe('makeConnectionChannel', () => {
+  /**
+   * A minimal stand-in for a connected socket: enough of the surface for
+   * the channel to attach handlers, and emitters so a test can drive the
+   * peer side directly.
+   *
+   * @returns The fake socket.
+   */
+  function makeFakeSocket(): net.Socket {
+    const emitter = new EventEmitter();
+    return Object.assign(emitter, {
+      destroy: () => undefined,
+      write: () => true,
+    }) as unknown as net.Socket;
+  }
+
+  it('reports the connection closed when the holder closes it', async () => {
+    const onClosed = vi.fn();
+    const channel = makeConnectionChannel('c1', makeFakeSocket(), onClosed);
+
+    await channel.close();
+
+    // Without this the listener keeps the channel registered forever, so a
+    // long-lived listener accumulates every session it ever served.
+    expect(onClosed).toHaveBeenCalledOnce();
+  });
+
+  it('reports the connection closed when the peer ends it', () => {
+    const onClosed = vi.fn();
+    const socket = makeFakeSocket();
+    makeConnectionChannel('c1', socket, onClosed);
+
+    socket.emit('end');
+
+    expect(onClosed).toHaveBeenCalledOnce();
+  });
+
+  it('reports closed only once across peer end and holder close', async () => {
+    const onClosed = vi.fn();
+    const socket = makeFakeSocket();
+    const channel = makeConnectionChannel('c1', socket, onClosed);
+
+    socket.emit('end');
+    await channel.close();
+    socket.emit('close');
+
+    expect(onClosed).toHaveBeenCalledOnce();
+  });
+
+  it('flushes a trailing partial line when the peer ends', async () => {
+    const channel = makeConnectionChannel(
+      'c1',
+      (() => {
+        const peer = makeFakeSocket();
+        setImmediate(() => {
+          peer.emit('data', Buffer.from('no-newline-here'));
+          peer.emit('end');
+        });
+        return peer;
+      })(),
+      vi.fn(),
+    );
+
+    // Data that arrived before the peer went away is still owed to the reader.
+    expect(await channel.read()).toBe('no-newline-here');
+    expect(await channel.read()).toBeNull();
+  });
+
+  it('ignores data that arrives after close', async () => {
+    const socket = makeFakeSocket();
+    const channel = makeConnectionChannel('c1', socket, () => undefined);
+
+    await channel.close();
+    // Node can still emit 'data' after destroy(); a late chunk must not
+    // refill the queue that close() cleared.
+    socket.emit('data', Buffer.from('too-late\n'));
+
+    expect(await channel.read()).toBeNull();
+  });
+
+  it('still delivers data that arrived before a peer end', async () => {
+    const socket = makeFakeSocket();
+    const channel = makeConnectionChannel('c1', socket, () => undefined);
+
+    socket.emit('data', Buffer.from('in-time\n'));
+    socket.emit('end');
+    socket.emit('data', Buffer.from('too-late\n'));
+
+    expect(await channel.read()).toBe('in-time');
+    expect(await channel.read()).toBeNull();
+  });
+
+  it('discards a trailing partial line when the holder closes', async () => {
+    const socket = makeFakeSocket();
+    const channel = makeConnectionChannel('c1', socket, vi.fn());
+
+    socket.emit('data', Buffer.from('no-newline-here'));
+    await channel.close();
+
+    expect(await channel.read()).toBeNull();
+  });
+});
diff --git a/packages/kernel-node-runtime/src/io/socket-listener.ts b/packages/kernel-node-runtime/src/io/socket-listener.ts
new file mode 100644
index 0000000000..163590953d
--- /dev/null
+++ b/packages/kernel-node-runtime/src/io/socket-listener.ts
@@ -0,0 +1,273 @@
+import type { IOChannel, IOListener } from '@metamask/ocap-kernel';
+import fs from 'node:fs/promises';
+import * as net from 'node:net';
+import { StringDecoder } from 'node:string_decoder';
+
+type PendingReader = {
+  resolve: (value: string | null) => void;
+};
+
+type PendingAcceptor = {
+  resolve: (value: IOChannel | null) => void;
+};
+
+/**
+ * Wrap a connected socket as an `IOChannel`.
+ *
+ * All of the channel's state — receive buffer, decoder, queued lines, and
+ * pending readers — is local to this function, so concurrent connections
+ * cannot interfere with one another. This is the reason the listener can
+ * serve many peers at once where a single shared channel could not.
+ *
+ * @param name - The connection name (for diagnostics).
+ * @param socket - The connected socket.
+ * @param onClosed - Invoked once when the connection is finished, whether
+ * because the peer went away or because `close()` was called.
+ * @returns The channel for this connection.
+ */
+export function makeConnectionChannel(
+  name: string,
+  socket: net.Socket,
+  onClosed: () => void,
+): IOChannel {
+  const lineQueue: string[] = [];
+  const readerQueue: PendingReader[] = [];
+  const decoder = new StringDecoder('utf8');
+  let buffer = '';
+  let ended = false;
+  let closed = false;
+
+  /**
+   * Deliver a line to a pending reader or enqueue it for a future read.
+   *
+   * @param line - The line to deliver.
+   */
+  function deliverLine(line: string): void {
+    const reader = readerQueue.shift();
+    if (reader) {
+      reader.resolve(line);
+    } else {
+      lineQueue.push(line);
+    }
+  }
+
+  /**
+   * Resolve every waiting reader with EOF.
+   */
+  function deliverEOF(): void {
+    while (readerQueue.length > 0) {
+      readerQueue.shift()?.resolve(null);
+    }
+  }
+
+  /**
+   * Split incoming bytes into `\n`-delimited lines.
+   *
+   * @param data - The raw data from the socket.
+   */
+  function handleData(data: Buffer): void {
+    if (ended || closed) {
+      // Node can still emit 'data' after `socket.destroy()`. Accepting a
+      // late chunk would refill the queue that `close()` just cleared, and
+      // a subsequent `read()` would hand it out even though EOF has already
+      // been reported. Data that arrived *before* the end is unaffected: it
+      // is already queued and stays readable, which is what a peer-initiated
+      // end owes its reader.
+      return;
+    }
+    buffer += decoder.write(data);
+    let newlineIndex = buffer.indexOf('\n');
+    while (newlineIndex !== -1) {
+      const line = buffer.slice(0, newlineIndex);
+      buffer = buffer.slice(newlineIndex + 1);
+      deliverLine(line);
+      newlineIndex = buffer.indexOf('\n');
+    }
+  }
+
+  /**
+   * Handle the channel finishing, from either end. Unlike a shared channel
+   * there is no ambiguity about whose session ended: this channel serves
+   * exactly one peer, so the end of the socket is the end of the channel.
+   *
+   * A trailing partial line is flushed only when the peer ended things,
+   * because that data arrived before the peer went away. When the holder
+   * called `close()` it is discarded instead — EOF has already been
+   * reported, and handing data over afterwards would contradict it.
+   */
+  function handleEnd(): void {
+    if (ended) {
+      return;
+    }
+    ended = true;
+    const trailing = buffer + decoder.end();
+    buffer = '';
+    if (!closed && trailing.length > 0) {
+      deliverLine(trailing);
+    }
+    deliverEOF();
+    onClosed();
+  }
+
+  socket.on('data', handleData);
+  socket.on('end', handleEnd);
+  socket.on('error', handleEnd);
+  socket.on('close', handleEnd);
+
+  return {
+    async read(): Promise<string | null> {
+      const queued = lineQueue.shift();
+      if (queued !== undefined) {
+        return queued;
+      }
+      if (ended || closed) {
+        return null;
+      }
+      return new Promise<string | null>((resolve) => {
+        readerQueue.push({ resolve });
+      });
+    },
+
+    async write(data: string): Promise<void> {
+      if (closed || ended) {
+        throw new Error(`IO connection "${name}" is closed`);
+      }
+      return new Promise<void>((resolve, reject) => {
+        socket.write(`${data}\n`, (error) => {
+          if (error) {
+            reject(error);
+          } else {
+            resolve();
+          }
+        });
+      });
+    },
+
+    async close(): Promise<void> {
+      if (closed) {
+        return;
+      }
+      closed = true;
+      // Drop lines already queued: the holder is done reading, so nothing
+      // buffered may surface from a later read() once EOF is reported.
+      // `handleEnd` discards the trailing fragment for the same reason,
+      // keyed on `closed`. Deliberately not setting `ended` here — that
+      // would make `handleEnd` return early and skip `onClosed`, leaving
+      // this channel registered with the listener for good.
+      lineQueue.length = 0;
+      deliverEOF();
+      socket.destroy();
+      // `close` on the socket will fire handleEnd, but call it directly so
+      // the caller's `onClosed` bookkeeping is done by the time close()
+      // resolves rather than a turn later.
+      handleEnd();
+    },
+  };
+}
+
+/**
+ * Create an `IOListener` backed by a Unix domain socket.
+ *
+ * Creates a `net.Server` on the configured path. Every connection that
+ * arrives becomes its own `IOChannel`, handed out by `accept()`, so any
+ * number of peers can be served concurrently. Lines are `\n`-delimited.
+ *
+ * Connections that arrive before anyone calls `accept()` are queued, so a
+ * peer connecting during startup is not dropped.
+ *
+ * @param name - The listener name (for diagnostics).
+ * @param socketPath - The file path for the Unix domain socket.
+ * @returns A promise for the IOListener, resolved once the server is listening.
+ */
+export async function makeSocketIOListener(
+  name: string,
+  socketPath: string,
+): Promise<IOListener> {
+  /** Connections that have arrived but not yet been accepted. */
+  const readyQueue: IOChannel[] = [];
+  /** Callers waiting in `accept()` for a connection to arrive. */
+  const acceptorQueue: PendingAcceptor[] = [];
+  /** Live connections, so `close()` can tear them all down. */
+  const liveChannels = new Set<IOChannel>();
+  let closed = false;
+  let nextConnectionId = 0;
+
+  const server = net.createServer((socket) => {
+    if (closed) {
+      socket.destroy();
+      return;
+    }
+    nextConnectionId += 1;
+    const connectionName = `${name}:${nextConnectionId}`;
+    const channel: IOChannel = makeConnectionChannel(
+      connectionName,
+      socket,
+      () => {
+        liveChannels.delete(channel);
+      },
+    );
+    liveChannels.add(channel);
+
+    const acceptor = acceptorQueue.shift();
+    if (acceptor) {
+      acceptor.resolve(channel);
+    } else {
+      readyQueue.push(channel);
+    }
+  });
+
+  // Remove stale socket file if it exists
+  try {
+    await fs.unlink(socketPath);
+  } catch {
+    // Ignore if it doesn't exist
+  }
+
+  await new Promise<void>((resolve, reject) => {
+    server.on('error', reject);
+    server.listen(socketPath, () => {
+      server.removeListener('error', reject);
+      resolve();
+    });
+  });
+
+  return {
+    async accept(): Promise<IOChannel | null> {
+      const ready = readyQueue.shift();
+      if (ready) {
+        return ready;
+      }
+      if (closed) {
+        return null;
+      }
+      return new Promise<IOChannel | null>((resolve) => {
+        acceptorQueue.push({ resolve });
+      });
+    },
+
+    async close(): Promise<void> {
+      if (closed) {
+        return;
+      }
+      closed = true;
+      // Release anyone parked in accept() so their loops can exit.
+      while (acceptorQueue.length > 0) {
+        acceptorQueue.shift()?.resolve(null);
+      }
+      readyQueue.length = 0;
+      for (const channel of [...liveChannels]) {
+        await channel.close();
+      }
+      liveChannels.clear();
+      await new Promise<void>((resolve) => {
+        server.close(() => resolve());
+      });
+      // Clean up socket file
+      try {
+        await fs.unlink(socketPath);
+      } catch {
+        // Ignore
+      }
+    },
+  };
+}
diff --git a/packages/kernel-node-runtime/src/kernel/make-kernel.ts b/packages/kernel-node-runtime/src/kernel/make-kernel.ts
index 4cbda735b3..901e79982a 100644
--- a/packages/kernel-node-runtime/src/kernel/make-kernel.ts
+++ b/packages/kernel-node-runtime/src/kernel/make-kernel.ts
@@ -3,13 +3,13 @@ import { makeSQLKernelDatabase } from '@metamask/kernel-store/sqlite/nodejs';
 import { Logger } from '@metamask/logger';
 import { Kernel } from '@metamask/ocap-kernel';
 import type {
-  IOChannelFactory,
+  IOListenerFactory,
   OnRunLoopFailure,
   SystemSubclusterConfig,
 } from '@metamask/ocap-kernel';
 
 import { NodejsPlatformServices } from './PlatformServices.ts';
-import { makeIOChannelFactory } from '../io/index.ts';
+import { makeIOListenerFactory } from '../io/index.ts';
 
 /**
  * Result of {@link makeKernel}.
@@ -28,7 +28,7 @@ export type MakeKernelResult = {
  * @param options.dbFilename - The filename of the SQLite database file.
  * @param options.logger - The logger to use for the kernel.
  * @param options.keySeed - Optional seed for libp2p key generation.
- * @param options.ioChannelFactory - Optional factory for creating IO channels.
+ * @param options.ioListenerFactory - Optional factory for creating IO listeners.
  * @param options.systemSubclusters - Optional system subcluster configurations.
  * @param options.onRunLoopFailure - Optional handler called if the kernel's run
  * loop dies, after which the kernel must be restarted.
@@ -40,7 +40,7 @@ export async function makeKernel({
   dbFilename,
   logger,
   keySeed,
-  ioChannelFactory,
+  ioListenerFactory,
   systemSubclusters,
   onRunLoopFailure,
 }: {
@@ -49,7 +49,7 @@ export async function makeKernel({
   dbFilename?: string;
   logger?: Logger;
   keySeed?: string | undefined;
-  ioChannelFactory?: IOChannelFactory;
+  ioListenerFactory?: IOListenerFactory;
   systemSubclusters?: SystemSubclusterConfig[];
   onRunLoopFailure?: OnRunLoopFailure;
 }): Promise<MakeKernelResult> {
@@ -67,7 +67,7 @@ export async function makeKernel({
     resetStorage,
     logger: rootLogger.subLogger({ tags: ['kernel'] }),
     keySeed,
-    ioChannelFactory: ioChannelFactory ?? makeIOChannelFactory(),
+    ioListenerFactory: ioListenerFactory ?? makeIOListenerFactory(),
     ...(systemSubclusters ? { systemSubclusters } : {}),
     ...(onRunLoopFailure ? { onRunLoopFailure } : {}),
   });
diff --git a/packages/kernel-node-runtime/test/helpers/kernel.ts b/packages/kernel-node-runtime/test/helpers/kernel.ts
index 1595897c12..313604f830 100644
--- a/packages/kernel-node-runtime/test/helpers/kernel.ts
+++ b/packages/kernel-node-runtime/test/helpers/kernel.ts
@@ -4,7 +4,7 @@ import { Logger } from '@metamask/logger';
 import { Kernel, kunser } from '@metamask/ocap-kernel';
 import type {
   ClusterConfig,
-  IOChannelFactory,
+  IOListenerFactory,
   SystemSubclusterConfig,
 } from '@metamask/ocap-kernel';
 
@@ -14,7 +14,7 @@ type MakeTestKernelOptions = {
   resetStorage?: boolean;
   mnemonic?: string;
   systemSubclusters?: SystemSubclusterConfig[];
-  ioChannelFactory?: IOChannelFactory;
+  ioListenerFactory?: IOListenerFactory;
 };
 
 /**
@@ -26,7 +26,7 @@ type MakeTestKernelOptions = {
  * @param options.resetStorage - Whether to reset the storage (default: true).
  * @param options.mnemonic - Optional BIP39 mnemonic string.
  * @param options.systemSubclusters - Optional system subcluster configurations.
- * @param options.ioChannelFactory - Optional IO channel factory.
+ * @param options.ioListenerFactory - Optional IO listener factory.
  * @returns The kernel.
  */
 export async function makeTestKernel(
@@ -37,7 +37,7 @@ export async function makeTestKernel(
     resetStorage = true,
     mnemonic,
     systemSubclusters,
-    ioChannelFactory,
+    ioListenerFactory,
   } = options;
 
   const logger = new Logger('test-kernel');
@@ -48,7 +48,7 @@ export async function makeTestKernel(
     resetStorage,
     mnemonic,
     systemSubclusters,
-    ioChannelFactory,
+    ioListenerFactory,
     logger: logger.subLogger({ tags: ['kernel'] }),
   });
 
diff --git a/packages/kernel-test/src/io.test.ts b/packages/kernel-test/src/io.test.ts
index 62ccc7f5a7..df94edb03a 100644
--- a/packages/kernel-test/src/io.test.ts
+++ b/packages/kernel-test/src/io.test.ts
@@ -1,7 +1,6 @@
 import { makeSQLKernelDatabase } from '@metamask/kernel-store/sqlite/nodejs';
 import { waitUntilQuiescent } from '@metamask/kernel-utils';
 import { Kernel } from '@metamask/ocap-kernel';
-import type { IOChannel, IOConfig } from '@metamask/ocap-kernel';
 import * as net from 'node:net';
 import * as os from 'node:os';
 import * as path from 'node:path';
@@ -53,125 +52,56 @@ async function readLine(socket: net.Socket): Promise<string> {
   });
 }
 
-async function makeTestSocketChannel(
-  _name: string,
+/**
+ * Stand up a kernel wired to the real Node listener factory, with an
+ * io-vat subcluster listening on `socketPath`.
+ *
+ * @param socketPath - Path for the listener's Unix domain socket.
+ * @returns The kernel and the io-vat's root kref.
+ */
+async function makeIoKernel(
   socketPath: string,
-): Promise<IOChannel> {
-  const fsPromises = await import('node:fs/promises');
-  const lineQueue: string[] = [];
-  const readerQueue: { resolve: (value: string | null) => void }[] = [];
-  let currentSocket: net.Socket | null = null;
-  let lineBuffer = '';
-  let closed = false;
-
-  function deliverLine(line: string): void {
-    const reader = readerQueue.shift();
-    if (reader) {
-      reader.resolve(line);
-    } else {
-      lineQueue.push(line);
-    }
-  }
-
-  function deliverEOF(): void {
-    while (readerQueue.length > 0) {
-      readerQueue.shift()?.resolve(null);
-    }
-  }
-
-  const server = net.createServer((socket) => {
-    if (currentSocket) {
-      socket.destroy();
-      return;
-    }
-    currentSocket = socket;
-    lineBuffer = '';
-    socket.on('data', (data: Buffer) => {
-      lineBuffer += data.toString();
-      let idx = lineBuffer.indexOf('\n');
-      while (idx !== -1) {
-        deliverLine(lineBuffer.slice(0, idx));
-        lineBuffer = lineBuffer.slice(idx + 1);
-        idx = lineBuffer.indexOf('\n');
-      }
-    });
-    socket.on('end', () => {
-      if (lineBuffer.length > 0) {
-        deliverLine(lineBuffer);
-        lineBuffer = '';
-      }
-      currentSocket = null;
-      deliverEOF();
-    });
-    socket.on('error', () => {
-      currentSocket = null;
-      deliverEOF();
-    });
-  });
-
-  try {
-    await fsPromises.unlink(socketPath);
-  } catch {
-    // ignore
-  }
-
-  await new Promise<void>((resolve, reject) => {
-    server.on('error', reject);
-    server.listen(socketPath, () => {
-      server.removeListener('error', reject);
-      resolve();
-    });
+): Promise<{ kernel: Kernel; rootKref: string }> {
+  const kernelDatabase = await makeSQLKernelDatabase({
+    dbFilename: ':memory:',
   });
+  const { logger } = makeTestLogger();
 
-  return {
-    async read() {
-      if (closed) {
-        return null;
-      }
-      const queued = lineQueue.shift();
-      if (queued !== undefined) {
-        return queued;
-      }
-      if (!currentSocket) {
-        return null;
-      }
-      return new Promise<string | null>((resolve) => {
-        readerQueue.push({ resolve });
-      });
+  const { NodejsPlatformServices, makeIOListenerFactory } = await import(
+    '@metamask/kernel-node-runtime'
+  );
+  const kernel = await Kernel.make(
+    new NodejsPlatformServices({
+      logger: logger.subLogger({ tags: ['platform'] }),
+    }),
+    kernelDatabase,
+    {
+      resetStorage: true,
+      logger,
+      ioListenerFactory: makeIOListenerFactory(),
     },
-    async write(data: string) {
-      if (!currentSocket) {
-        throw new Error('no connected client');
-      }
-      const socket = currentSocket;
-      return new Promise<void>((resolve, reject) => {
-        socket.write(`${data}\n`, (error) => {
-          if (error) {
-            reject(error);
-          } else {
-            resolve();
-          }
-        });
-      });
+  );
+
+  const { rootKref } = await kernel.launchSubcluster({
+    bootstrap: 'io',
+    forceReset: true,
+    io: {
+      repl: {
+        type: 'socket' as const,
+        path: socketPath,
+      },
     },
-    async close() {
-      if (closed) {
-        return;
-      }
-      closed = true;
-      deliverEOF();
-      currentSocket?.destroy();
-      currentSocket = null;
-      await new Promise<void>((resolve) => {
-        server.close(() => resolve());
-      });
-      try {
-        await fsPromises.unlink(socketPath);
-      } catch {
-        // ignore
-      }
+    services: ['repl'],
+    vats: {
+      io: {
+        bundleSpec: getBundleSpec('io-vat'),
+        parameters: { name: 'io' },
+      },
     },
-  };
+  });
+  await waitUntilQuiescent();
+
+  return { kernel, rootKref };
 }
 
 describe('IO kernel service', () => {
@@ -184,65 +114,20 @@ describe('IO kernel service', () => {
     clients.length = 0;
   });
 
-  it('reads and writes through an IO channel', async () => {
+  it('reads and writes through an accepted connection', async () => {
     const socketPath = tempSocketPath();
-    const kernelDatabase = await makeSQLKernelDatabase({
-      dbFilename: ':memory:',
-    });
-    const { logger } = makeTestLogger();
+    const { kernel, rootKref } = await makeIoKernel(socketPath);
 
-    const { NodejsPlatformServices } = await import(
-      '@metamask/kernel-node-runtime'
-    );
-    const kernel = await Kernel.make(
-      new NodejsPlatformServices({
-        logger: logger.subLogger({ tags: ['platform'] }),
-      }),
-      kernelDatabase,
-      {
-        resetStorage: true,
-        logger,
-        ioChannelFactory: async (name: string, config: IOConfig) => {
-          if (config.type !== 'socket') {
-            throw new Error(`unsupported: ${config.type}`);
-          }
-          return makeTestSocketChannel(name, config.path);
-        },
-      },
-    );
-
-    const config = {
-      bootstrap: 'io',
-      forceReset: true,
-      io: {
-        repl: {
-          type: 'socket' as const,
-          path: socketPath,
-        },
-      },
-      services: ['repl'],
-      vats: {
-        io: {
-          bundleSpec: getBundleSpec('io-vat'),
-          parameters: { name: 'io' },
-        },
-      },
-    };
-
-    const { rootKref } = await kernel.launchSubcluster(config);
-    await waitUntilQuiescent();
-
-    // Connect to the socket
     const client = await connectToSocket(socketPath);
     clients.push(client);
-
-    // Small delay for connection setup
     await new Promise((resolve) => setTimeout(resolve, 20));
 
+    await kernel.queueMessage(rootKref, 'doAccept', []);
+    await waitUntilQuiescent(100);
+
     // Send a line from the test to the vat
     await writeLine(client, 'hello from test');
 
-    // Trigger the vat to read and verify it received the data
     await kernel.queueMessage(rootKref, 'doRead', []);
     await waitUntilQuiescent(100);
 
@@ -259,7 +144,63 @@ describe('IO kernel service', () => {
     await kernel.queueMessage(rootKref, 'doWrite', ['hello from vat']);
     await waitUntilQuiescent(100);
 
-    const received = await linePromise;
-    expect(received).toBe('hello from vat');
+    expect(await linePromise).toBe('hello from vat');
+  });
+
+  it('serves two concurrent peers without crossing their traffic', async () => {
+    const socketPath = tempSocketPath();
+    const { kernel, rootKref } = await makeIoKernel(socketPath);
+
+    const alice = await connectToSocket(socketPath);
+    clients.push(alice);
+    await new Promise((resolve) => setTimeout(resolve, 20));
+    const bob = await connectToSocket(socketPath);
+    clients.push(bob);
+    await new Promise((resolve) => setTimeout(resolve, 20));
+
+    // Both peers are accepted — under the old single-client channel the
+    // second connection would have been destroyed outright.
+    await kernel.queueMessage(rootKref, 'doAccept', []);
+    await waitUntilQuiescent(100);
+    await kernel.queueMessage(rootKref, 'doAccept', []);
+    await waitUntilQuiescent(100);
+
+    const countResult = await kernel.queueMessage(
+      rootKref,
+      'getConnectionCount',
+      [],
+    );
+    await waitUntilQuiescent(100);
+    expect(countResult.body).toContain('2');
+
+    // Each peer's line arrives on its own connection.
+    await writeLine(alice, 'from-alice');
+    await writeLine(bob, 'from-bob');
+    await new Promise((resolve) => setTimeout(resolve, 20));
+
+    await kernel.queueMessage(rootKref, 'doRead', [0]);
+    await waitUntilQuiescent(100);
+    await kernel.queueMessage(rootKref, 'doRead', [1]);
+    await waitUntilQuiescent(100);
+
+    const bufferResult = await kernel.queueMessage(
+      rootKref,
+      'getReadBuffer',
+      [],
+    );
+    await waitUntilQuiescent(100);
+    expect(bufferResult.body).toContain('from-alice');
+    expect(bufferResult.body).toContain('from-bob');
+
+    // And each write goes back to the right peer.
+    const aliceHeard = readLine(alice);
+    const bobHeard = readLine(bob);
+    await kernel.queueMessage(rootKref, 'doWrite', ['for-alice', 0]);
+    await waitUntilQuiescent(100);
+    await kernel.queueMessage(rootKref, 'doWrite', ['for-bob', 1]);
+    await waitUntilQuiescent(100);
+
+    expect(await aliceHeard).toBe('for-alice');
+    expect(await bobHeard).toBe('for-bob');
   });
 });
diff --git a/packages/kernel-test/src/vats/io-vat.ts b/packages/kernel-test/src/vats/io-vat.ts
index 04b582fde1..0be770d45e 100644
--- a/packages/kernel-test/src/vats/io-vat.ts
+++ b/packages/kernel-test/src/vats/io-vat.ts
@@ -7,6 +7,10 @@ import type { TestPowers } from '../test-powers.ts';
 /**
  * Build function for testing IO kernel services.
  *
+ * The `repl` endowment is an `IOListener`, so the vat accepts connections
+ * from it and keeps each one separately. `doRead`/`doWrite` name a
+ * connection by index so a test can drive several peers independently.
+ *
  * @param vatPowers - Special powers granted to this vat.
  * @param parameters - Initialization parameters from the vat's config object.
  * @param parameters.name - The name of the vat.
@@ -19,26 +23,46 @@ export function buildRootObject(
 ) {
   const name = parameters?.name ?? 'io-vat';
   const tlog = unwrapTestLogger(vatPowers, name);
-  let ioService: unknown;
+  let listener: unknown;
+  const connections: unknown[] = [];
   const readBuffer: string[] = [];
 
   return makeDefaultExo('root', {
     async bootstrap(_vats: unknown, services: { repl: unknown }) {
       tlog('bootstrap');
-      ioService = services.repl;
+      listener = services.repl;
+    },
+    /**
+     * Accept the next waiting connection, appending it to the list.
+     *
+     * @returns The index of the accepted connection, or -1 on EOF.
+     */
+    async doAccept() {
+      const connection = await E(listener).accept();
+      if (!connection) {
+        tlog('accept: listener closed');
+        return -1;
+      }
+      connections.push(connection);
+      const index = connections.length - 1;
+      tlog(`accepted connection ${index}`);
+      return index;
     },
-    async doRead() {
-      const line = await E(ioService).read();
-      tlog(`read: ${line}`);
+    async doRead(index = 0) {
+      const line = await E(connections[index]).read();
+      tlog(`read[${index}]: ${line}`);
       readBuffer.push(String(line));
       return line;
     },
-    async doWrite(data: string) {
-      await E(ioService).write(data);
-      tlog(`wrote: ${data}`);
+    async doWrite(data: string, index = 0) {
+      await E(connections[index]).write(data);
+      tlog(`wrote[${index}]: ${data}`);
     },
     async getReadBuffer() {
       return [...readBuffer];
     },
+    async getConnectionCount() {
+      return connections.length;
+    },
   });
 }
diff --git a/packages/kernel-utils/CHANGELOG.md b/packages/kernel-utils/CHANGELOG.md
index 88f885e05e..c84728c27b 100644
--- a/packages/kernel-utils/CHANGELOG.md
+++ b/packages/kernel-utils/CHANGELOG.md
@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Add an `interface` variant to `JsonSchema` — `{ type: 'interface', description?, methods }` — describing an object whose methods can be invoked, so a method that returns an object reference can declare that object's API inline and a client need not make a second round-trip to discover it. The `methods` field is recursive, so a returned interface can itself return interfaces. The variant describes an _interface_; whether the reference to the object is unforgeable is a property of the reference plumbing, not of the description ([#1007](https://github.com/MetaMask/ocap-kernel/pull/1007))
 - Add a `./described` export with a combinator namespace `S` (`S.string`/`S.number`/`S.boolean`/`S.arrayOf`/`S.record`/`S.object`/`S.nothing` leaves, plus `S.arg`/`S.method`/`S.interface`) that authors an `@endo/patterns` interface guard and a matching `MethodSchema` from a single source, so a discoverable exo's enforced shape and its `__getDescription__` hint cannot drift ([#958](https://github.com/MetaMask/ocap-kernel/pull/958))
 - Add an optional `required` field to `MethodSchema` (mirroring `required` on object `JsonSchema`) naming which arguments are required, and a `{ required }` option on `methodArgsToStruct` that validates unlisted arguments as optional, so a method's argument schema can faithfully represent the optional trailing arguments its guard already allows ([#958](https://github.com/MetaMask/ocap-kernel/pull/958))
 - Add `getLibp2pRelayHome()` to the `./nodejs` exports, returning the libp2p relay's bookkeeping directory (default `~/.libp2p-relay`, overridable via `$LIBP2P_RELAY_HOME`) — kept separate from `$OCAP_HOME` so one relay can serve daemons with different OCAP_HOMEs ([#952](https://github.com/MetaMask/ocap-kernel/pull/952))
diff --git a/packages/kernel-utils/src/discoverable.test.ts b/packages/kernel-utils/src/discoverable.test.ts
index 3075c09b01..1a328368d7 100644
--- a/packages/kernel-utils/src/discoverable.test.ts
+++ b/packages/kernel-utils/src/discoverable.test.ts
@@ -143,6 +143,38 @@ describe('makeDiscoverableExo', () => {
     );
   });
 
+  it('accepts a return schema describing an interface', () => {
+    const factorySchema: MethodSchema = {
+      description: 'Return a counter object',
+      args: {},
+      returns: {
+        type: 'interface',
+        description: 'A stateful counter',
+        methods: {
+          increment: {
+            description: 'Bump the counter and return the new value',
+            args: {},
+            returns: { type: 'number', description: 'The new count' },
+          },
+          reset: {
+            description: 'Reset the counter to zero',
+            args: {},
+          },
+        },
+      },
+    };
+    const methods = { makeCounter: () => ({}) };
+    const schema: Record<keyof typeof methods, MethodSchema> = {
+      makeCounter: factorySchema,
+    };
+
+    const exo = makeDiscoverableExo('CounterFactory', methods, schema);
+
+    expect(exo[GET_DESCRIPTION]()).toStrictEqual({
+      makeCounter: factorySchema,
+    });
+  });
+
   it('re-throws errors from makeExo that are not about describe key', () => {
     const testError = new Error('Some other error from makeExo');
     makeExoMock.mockImplementation(() => {
diff --git a/packages/kernel-utils/src/json-schema-to-struct.test.ts b/packages/kernel-utils/src/json-schema-to-struct.test.ts
index 8530c21c15..f6be0a50d8 100644
--- a/packages/kernel-utils/src/json-schema-to-struct.test.ts
+++ b/packages/kernel-utils/src/json-schema-to-struct.test.ts
@@ -51,6 +51,39 @@ describe('jsonSchemaToStruct', () => {
     });
     assert({ a: 1, extra: 'ignored' }, struct);
   });
+
+  describe('interface', () => {
+    const interfaceSchema = {
+      type: 'interface',
+      description: 'a reviser',
+      methods: {
+        revise: { description: 'revise it', args: {} },
+      },
+    } as const;
+
+    it('accepts any object reference without introspecting its methods', () => {
+      const struct = jsonSchemaToStruct(interfaceSchema);
+      // The declared `methods` are a description for the caller, not a
+      // shape to enforce here: whether the object honours them is only
+      // discoverable by invoking it, which is the receiver's business.
+      assert({}, struct);
+      assert({ revise: () => undefined }, struct);
+      assert({ somethingElse: 1 }, struct);
+    });
+
+    it.each([
+      ['a string', 'not an object'],
+      ['a number', 42],
+      ['a boolean', true],
+      ['null', null],
+      ['undefined', undefined],
+    ])('rejects %s', (_label, value) => {
+      const struct = jsonSchemaToStruct(interfaceSchema);
+      expect(() => assert(value, struct)).toThrow(
+        /Expected an object reference/u,
+      );
+    });
+  });
 });
 
 describe('methodArgsToStruct', () => {
diff --git a/packages/kernel-utils/src/json-schema-to-struct.ts b/packages/kernel-utils/src/json-schema-to-struct.ts
index 944cb0dfe1..59d16ced92 100644
--- a/packages/kernel-utils/src/json-schema-to-struct.ts
+++ b/packages/kernel-utils/src/json-schema-to-struct.ts
@@ -87,6 +87,18 @@ export function jsonSchemaToStruct(schema: JsonSchema): Struct<unknown> {
       }
       return looseObjectStruct(schema);
     }
+    case 'interface': {
+      // An interface reference: at runtime the value is an object (possibly
+      // an exo, remotable, or presence). We can't introspect its methods
+      // here — that's the receiver's responsibility on invocation. Validate
+      // that it's a non-null object and pass through.
+      return define('JsonSchemaInterface', (value) => {
+        if (typeof value !== 'object' || value === null) {
+          return 'Expected an object reference';
+        }
+        return true;
+      }) as Struct<unknown>;
+    }
     default: {
       const _never: never = schema;
       throw new TypeError(`Unsupported JSON schema: ${String(_never)}`);
diff --git a/packages/kernel-utils/src/schema.ts b/packages/kernel-utils/src/schema.ts
index b3534b8e13..6f79d3752e 100644
--- a/packages/kernel-utils/src/schema.ts
+++ b/packages/kernel-utils/src/schema.ts
@@ -1,11 +1,13 @@
 /**
- * JSON Schema type for describing values. Supports primitives, arrays, and objects
+ * JSON Schema type for describing values. Supports primitives, arrays, objects,
+ * and object interfaces (i.e. an object with methods you can invoke),
  * with recursive definitions.
  */
 export type JsonSchema =
   | PrimitiveJsonSchema
   | ArrayJsonSchema
-  | ObjectJsonSchema;
+  | ObjectJsonSchema
+  | InterfaceJsonSchema;
 
 /**
  * Primitive JSON Schema types (string, number, boolean).
@@ -37,6 +39,30 @@ type ObjectJsonSchema = {
   additionalProperties?: boolean;
 };
 
+/**
+ * Schema describing an object interface — a reference to an object whose
+ * methods can be invoked. Used as the return-type schema for methods that
+ * hand back an object reference (whether local or across a boundary), so
+ * a client can learn the returned object's API inline from the parent
+ * description without an extra round-trip.
+ *
+ * The `methods` field is recursive: any method here can itself return an
+ * interface, and so on.
+ *
+ * Naming note: this schema describes an object interface. Whether the
+ * reference to that object is unforgeable (i.e. an ocap in the strict
+ * sense) is a property of the reference plumbing (which vat holds it,
+ * whether it crossed a CapTP boundary, etc.), not of the interface
+ * description itself. Same schema either way.
+ */
+type InterfaceJsonSchema = {
+  type: 'interface';
+  description?: string;
+  methods: {
+    [key: string]: MethodSchema;
+  };
+};
+
 /**
  * Schema describing a method, including its purpose, arguments, and return value.
  */
diff --git a/packages/ocap-kernel/CHANGELOG.md b/packages/ocap-kernel/CHANGELOG.md
index 87dcabb833..588c44df69 100644
--- a/packages/ocap-kernel/CHANGELOG.md
+++ b/packages/ocap-kernel/CHANGELOG.md
@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Add `IOListener`, an endpoint peers connect to that yields one `IOChannel` per connection via `accept()`, replacing the previous one-client-at-a-time channel. Each accepted connection is a distinct object, so holding one conveys no way to reach another, and `direction` is enforced per connection. `accept()` resolves `null` once the listener is closed so an accept loop can terminate rather than hang ([#1007](https://github.com/MetaMask/ocap-kernel/pull/1007))
+- Anonymous kernel-hosted objects are recorded persistently and swept at kernel init, so one abandoned by a previous incarnation is not left pinned forever, accumulating with every restart — an anonymous object has no name to be re-registered under on boot, unlike a named service. The sweep unpins but cannot delete an object a vat import or queued message still references, so it does not by itself make a delivery to a survivor safe; the `invokeKernelService` fix below is what does ([#1007](https://github.com/MetaMask/ocap-kernel/pull/1007))
+- Add `KernelServiceManager.registerAnonymousKernelObject()` / `releaseAnonymousKernelObject()`, which make a kernel-hosted object routable by kref without entering it in the service-name index, so it has no name in the global service namespace and cannot be requested via a cluster config's `services` list. Used to host accepted IO connections, whose authority comes from holding the reference ([#1007](https://github.com/MetaMask/ocap-kernel/pull/1007))
 - Report run loop health in `KernelStatus.runLoop` (`{ state: 'idle' | 'running' }` or `{ state: 'failed', error, detail }`), exporting `RunLoopStatus`, `RunLoopStatusStruct`, and `OnRunLoopFailure` ([#1005](https://github.com/MetaMask/ocap-kernel/pull/1005))
   - `idle` means never started; a loop parked on an empty queue reports `running`
   - `error` is the failure's message and `detail` its whole cause chain, because only strings cross the wire: when a crank dies and its rollback then fails, the message names the rollback and only the chain names what killed the kernel
@@ -34,6 +37,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **BREAKING:** `Kernel.make`'s `ioChannelFactory` option is now `ioListenerFactory`, and the exported `IOChannelFactory` type is replaced by `IOListener` and `IOListenerFactory`. A cluster config's `io` entries now create listeners; vats call `accept()` to obtain a channel instead of reading and writing the endowment directly ([#1007](https://github.com/MetaMask/ocap-kernel/pull/1007))
 - Attribute a failed subcluster vat launch to the specific vat by kernel id and `ClusterConfig` name (e.g. `Failed to launch vat v3 (bob)`), preserving the original error as the `cause` ([#975](https://github.com/MetaMask/ocap-kernel/pull/975))
 - **BREAKING:** Remove `VatConfig.platformConfig.fetch` — migrate to `globals: ['fetch', ...]` + `network.allowedHosts` ([#942](https://github.com/MetaMask/ocap-kernel/pull/942))
 - **BREAKING:** `MakeAllowedGlobals` now takes a `{ logger }` options bag ([#942](https://github.com/MetaMask/ocap-kernel/pull/942))
@@ -43,6 +47,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- A message delivered to a kernel-owned kref with no registered service now rejects the caller with `ENDPOINT_UNREACHABLE` instead of throwing, which escaped the crank and killed the run loop — turning one unreachable reference into a dead kernel ([#1007](https://github.com/MetaMask/ocap-kernel/pull/1007))
+  - Reachable without any kernel bug: an anonymous kernel object hosts something that cannot outlive the process, such as an accepted socket connection, so a vat holding one across a restart or a message to one still queued from the previous incarnation lands here. Kernel objects are born with a `(1, 1)` refcount, so the init sweep cannot delete such an object and its `kernel` owner survives (see [#1006](https://github.com/MetaMask/ocap-kernel/issues/1006))
+  - Matches what `KernelRouter` already does for a delivery whose endpoint has vanished. A message sent with no result promise has nobody to report to, so it is logged instead
+- The kernel run queue no longer strands messages, going quiet with no error, no log, and no crash. `runQueueLengthCache` uses a negative value to mean "unknown, re-read from the database", but `enqueueRun`/`dequeueRun` adjusted it arithmetically without materializing it first — so an enqueue while the cache held `-1` produced `0` for a queue that actually held an item, and because `0` is not negative it was never re-read again. The run loop then saw an empty queue, went to sleep, and stranded everything behind it. Two paths reach that `-1`: kernel startup, and `rollbackCrank`, which invalidates the cache because a rollback may have restored dequeued items. The rollback path is the more likely of the two in practice, since a rollback is normally followed immediately by enqueueing an error or termination message. The run loop is also now woken by any non-empty queue rather than only by the empty-to-one transition, so a drifted count cannot silently lose the wakeup either ([#1007](https://github.com/MetaMask/ocap-kernel/pull/1007))
 - Stop reporting a healthy kernel after the run loop dies ([#1005](https://github.com/MetaMask/ocap-kernel/pull/1005))
   - The error was logged and swallowed, so `getStatus` kept returning its healthy-looking record while nothing on the run queue was processed and every `queueMessage` hung forever. Results in flight now reject with the killing error as their `cause`, later calls reject immediately, and `getStatus` answers without waiting on a crank that may never end
 - Roll back the crank the run loop died in instead of committing it, so a restart resumes from a consistent boundary ([#1005](https://github.com/MetaMask/ocap-kernel/pull/1005))
diff --git a/packages/ocap-kernel/src/Kernel.ts b/packages/ocap-kernel/src/Kernel.ts
index 912c1f6929..a312e60e65 100644
--- a/packages/ocap-kernel/src/Kernel.ts
+++ b/packages/ocap-kernel/src/Kernel.ts
@@ -4,7 +4,7 @@ import { isCapData } from '@metamask/kernel-utils';
 import { Logger } from '@metamask/logger';
 
 import { IOManager } from './io/IOManager.ts';
-import type { IOChannelFactory } from './io/types.ts';
+import type { IOListenerFactory } from './io/types.ts';
 import { makeKernelFacet } from './kernel-facet.ts';
 import type { KernelFacet } from './kernel-facet.ts';
 import { KernelQueue } from './KernelQueue.ts';
@@ -106,7 +106,7 @@ export class Kernel {
    * @param options.logger - Optional logger for error and diagnostic output.
    * @param options.keySeed - Optional seed for libp2p key generation.
    * @param options.mnemonic - Optional BIP39 mnemonic for deriving the kernel identity.
-   * @param options.ioChannelFactory - Optional factory for creating IO channels.
+   * @param options.ioListenerFactory - Optional factory for creating IO listeners.
    * @param options.allowedGlobalNames - Optional list of allowed global names for vat endowments.
    * @param options.onRunLoopFailure - Optional handler called if the run loop dies.
    */
@@ -119,7 +119,7 @@ export class Kernel {
       logger?: Logger;
       keySeed?: string | undefined;
       mnemonic?: string | undefined;
-      ioChannelFactory?: IOChannelFactory;
+      ioListenerFactory?: IOListenerFactory;
       allowedGlobalNames?: AllowedGlobalName[];
       onRunLoopFailure?: OnRunLoopFailure;
     } = {},
@@ -176,9 +176,9 @@ export class Kernel {
       logger: this.#logger.subLogger({ tags: ['KernelServiceManager'] }),
     });
 
-    if (options.ioChannelFactory) {
+    if (options.ioListenerFactory) {
       this.#ioManager = new IOManager({
-        factory: options.ioChannelFactory,
+        factory: options.ioListenerFactory,
         registerService:
           this.#kernelServiceManager.registerKernelServiceObject.bind(
             this.#kernelServiceManager,
@@ -187,6 +187,14 @@ export class Kernel {
           this.#kernelServiceManager.unregisterKernelServiceObject.bind(
             this.#kernelServiceManager,
           ),
+        registerAnonymous:
+          this.#kernelServiceManager.registerAnonymousKernelObject.bind(
+            this.#kernelServiceManager,
+          ),
+        releaseAnonymous:
+          this.#kernelServiceManager.releaseAnonymousKernelObject.bind(
+            this.#kernelServiceManager,
+          ),
         logger: this.#logger.subLogger({ tags: ['IOManager'] }),
       });
     }
@@ -237,7 +245,7 @@ export class Kernel {
    * @param options.logger - Optional logger for error and diagnostic output.
    * @param options.keySeed - Optional seed for libp2p key generation.
    * @param options.mnemonic - Optional BIP39 mnemonic for deriving the kernel identity.
-   * @param options.ioChannelFactory - Optional factory for creating IO channels.
+   * @param options.ioListenerFactory - Optional factory for creating IO listeners.
    * @param options.systemSubclusters - Optional array of system subcluster configurations.
    * @param options.allowedGlobalNames - Optional list of allowed global names for vat endowments. When set, only these names from the `VatSupervisor`'s configured endowments (see `createDefaultEndowments`) are available to vats.
    * @param options.onRunLoopFailure - Optional handler called if the run loop dies. The kernel must be restarted after that, so an embedder that outlives it (e.g. a daemon) should use this to terminate or restart.
@@ -251,7 +259,7 @@ export class Kernel {
       logger?: Logger;
       keySeed?: string | undefined;
       mnemonic?: string | undefined;
-      ioChannelFactory?: IOChannelFactory;
+      ioListenerFactory?: IOListenerFactory;
       systemSubclusters?: SystemSubclusterConfig[];
       allowedGlobalNames?: AllowedGlobalName[];
       onRunLoopFailure?: OnRunLoopFailure;
@@ -286,6 +294,26 @@ export class Kernel {
     // the run queue has no selective removal capability.
     this.provideFacet();
 
+    // Discard anonymous kernel objects from a previous incarnation. They
+    // host things that cannot outlive the process — an accepted socket
+    // connection, say — and unlike a named service there is no name to
+    // re-register one under, so a survivor is unreachable but still pinned,
+    // accumulating with every restart.
+    //
+    // This unpins; it does not by itself make a delivery to a survivor safe,
+    // because the object outlives the sweep whenever a vat import or queued
+    // message still references it. `invokeKernelService` is what makes that
+    // case survivable, by rejecting the caller instead of throwing. Swept
+    // before the queue starts regardless, so the unreachable objects are gone
+    // before anything can address them.
+    const abandoned =
+      this.#kernelServiceManager.releaseAbandonedAnonymousKernelObjects();
+    if (abandoned > 0) {
+      this.#logger.info(
+        `Released ${abandoned} anonymous kernel object(s) abandoned by a previous incarnation`,
+      );
+    }
+
     // Restore persisted system subclusters and delete ones that no
     // longer have a config, to ensure that orphaned vats aren't started
     this.#subclusterManager.initSystemSubclusters(configs);
diff --git a/packages/ocap-kernel/src/KernelQueue.ts b/packages/ocap-kernel/src/KernelQueue.ts
index 4148300ff7..3465e93cde 100644
--- a/packages/ocap-kernel/src/KernelQueue.ts
+++ b/packages/ocap-kernel/src/KernelQueue.ts
@@ -353,7 +353,11 @@ export class KernelQueue {
    */
   #enqueueRun(item: RunQueueItem): void {
     this.#kernelStore.enqueueRun(item);
-    if (this.#kernelStore.runQueueLength() === 1 && this.#wakeUpTheRunQueue) {
+    // Wake on any non-empty queue rather than only on the empty->1
+    // transition. A sleeping run loop plus a non-empty queue is a
+    // permanent wedge, so err towards a spurious wake: the resolver is
+    // cleared as it fires, and the loop re-checks the queue on waking.
+    if (this.#kernelStore.runQueueLength() > 0 && this.#wakeUpTheRunQueue) {
       const wakeUpTheRunQueue = this.#wakeUpTheRunQueue;
       this.#wakeUpTheRunQueue = null;
       wakeUpTheRunQueue();
diff --git a/packages/ocap-kernel/src/KernelServiceManager.test.ts b/packages/ocap-kernel/src/KernelServiceManager.test.ts
index 4d82d8af15..a3efd11bf9 100644
--- a/packages/ocap-kernel/src/KernelServiceManager.test.ts
+++ b/packages/ocap-kernel/src/KernelServiceManager.test.ts
@@ -4,7 +4,7 @@ import { describe, it, expect, beforeEach, vi } from 'vitest';
 
 import type { KernelQueue } from './KernelQueue.ts';
 import { KernelServiceManager } from './KernelServiceManager.ts';
-import { kser, makeKernelError } from './liveslots/kernel-marshal.ts';
+import { kser, kslot, makeKernelError } from './liveslots/kernel-marshal.ts';
 import { makeKernelStore } from './store/index.ts';
 import type { KernelMessage } from './types.ts';
 import { makeMapKernelDatabase } from '../test/storage.ts';
@@ -386,14 +386,44 @@ describe('KernelServiceManager', () => {
       expect(mockKernelQueue.resolvePromises).not.toHaveBeenCalled();
     });
 
-    it('throws error for non-existent service', () => {
+    it('rejects the caller for a non-existent service', () => {
       const message: KernelMessage = {
         methargs: kser(['testMethod', []]),
+        result: 'kp1',
       };
 
+      // Must not throw: a throw here escapes the crank and kills the run
+      // loop, and this is reachable whenever an anonymous kernel object
+      // outlives the process that hosted it.
       expect(() =>
         serviceManager.invokeKernelService('ko999', message),
-      ).toThrow('No registered service for ko999');
+      ).not.toThrow();
+      expect(mockKernelQueue.resolvePromises).toHaveBeenCalledWith('kernel', [
+        [
+          'kp1',
+          true,
+          makeKernelError(
+            'ENDPOINT_UNREACHABLE',
+            'No registered service for ko999',
+          ),
+        ],
+      ]);
+    });
+
+    it('logs for a non-existent service when the message has no result', () => {
+      const loggerErrorSpy = vi.spyOn(logger, 'error');
+      const message: KernelMessage = {
+        methargs: kser(['testMethod', []]),
+      };
+
+      expect(() =>
+        serviceManager.invokeKernelService('ko999', message),
+      ).not.toThrow();
+      expect(loggerErrorSpy).toHaveBeenCalledWith(
+        'Error in kernel service method:',
+        'No registered service for ko999',
+      );
+      expect(mockKernelQueue.resolvePromises).not.toHaveBeenCalled();
     });
 
     it('handles unknown method with result', async () => {
@@ -537,4 +567,152 @@ describe('KernelServiceManager', () => {
       ]);
     });
   });
+
+  describe('registerAnonymousKernelObject', () => {
+    it('hosts the object for routing without naming it', () => {
+      const kref = serviceManager.registerAnonymousKernelObject(
+        { ping: () => 'pong' },
+        'io-connection',
+      );
+
+      expect(serviceManager.isKernelService(kref)).toBe(true);
+      expect(kernelStore.getOwner(kref)).toBe('kernel');
+      expect(kernelStore.pinObject).toHaveBeenCalledWith(kref);
+      // The whole point: absent from the global name namespace, so no
+      // string can be used to ask for it.
+      expect(serviceManager.getKernelService('io-connection')).toBeUndefined();
+      // Recorded persistently so a later incarnation can sweep it.
+      expect(kernelStore.getAnonymousKernelObjects()).toStrictEqual([kref]);
+    });
+
+    it('allows the same label for distinct objects', () => {
+      const first = serviceManager.registerAnonymousKernelObject(
+        { which: () => 'first' },
+        'io-connection',
+      );
+      const second = serviceManager.registerAnonymousKernelObject(
+        { which: () => 'second' },
+        'io-connection',
+      );
+
+      expect(second).not.toBe(first);
+      expect(serviceManager.isKernelService(first)).toBe(true);
+      expect(serviceManager.isKernelService(second)).toBe(true);
+    });
+
+    it('round-trips through kslot/kser so a service method can return one', () => {
+      // This is what makes `accept()` possible: `invokeKernelService`
+      // passes a method's return value through `kser`, whose val-to-slot
+      // step only accepts standins minted by `kslot`.
+      const kref = serviceManager.registerAnonymousKernelObject(
+        { ping: () => 'pong' },
+        'io-connection',
+      );
+
+      expect(kser(kslot(kref))).toStrictEqual({
+        body: expect.any(String),
+        slots: [kref],
+      });
+    });
+
+    it('delivers messages to the hosted object', async () => {
+      const { method: ping, calls } = makeTrackableMethod(() => 'pong');
+      const kref = serviceManager.registerAnonymousKernelObject(
+        { ping },
+        'io-connection',
+      );
+
+      serviceManager.invokeKernelService(kref, {
+        methargs: kser(['ping', ['hello']]),
+        result: 'kp200',
+      });
+      await delay();
+
+      expect(calls).toStrictEqual([['hello']]);
+      expect(mockKernelQueue.resolvePromises).toHaveBeenCalledWith('kernel', [
+        ['kp200', false, kser('pong')],
+      ]);
+    });
+  });
+
+  describe('releaseAbandonedAnonymousKernelObjects', () => {
+    it('discards objects recorded by a previous incarnation', () => {
+      // Simulate a restart: the krefs are still recorded in the store, but
+      // the in-memory routing table starts empty.
+      const stale = kernelStore.initKernelObject('kernel');
+      kernelStore.pinObject(stale);
+      kernelStore.addAnonymousKernelObject(stale);
+
+      const fresh = new KernelServiceManager({
+        kernelStore,
+        kernelQueue: mockKernelQueue,
+        logger,
+      });
+
+      expect(fresh.releaseAbandonedAnonymousKernelObjects()).toBe(1);
+      expect(kernelStore.isObjectPinned(stale)).toBe(false);
+      expect(kernelStore.getAnonymousKernelObjects()).toStrictEqual([]);
+    });
+
+    it('leaves objects hosted by the current incarnation alone', () => {
+      const live = serviceManager.registerAnonymousKernelObject(
+        { ping: () => 'pong' },
+        'io-connection',
+      );
+
+      expect(serviceManager.releaseAbandonedAnonymousKernelObjects()).toBe(0);
+      expect(serviceManager.isKernelService(live)).toBe(true);
+      expect(kernelStore.isObjectPinned(live)).toBe(true);
+    });
+
+    it('reports nothing to do when none were recorded', () => {
+      expect(serviceManager.releaseAbandonedAnonymousKernelObjects()).toBe(0);
+    });
+  });
+
+  describe('releaseAnonymousKernelObject', () => {
+    it('removes the object from routing and unpins it', () => {
+      const kref = serviceManager.registerAnonymousKernelObject(
+        { ping: () => 'pong' },
+        'io-connection',
+      );
+
+      serviceManager.releaseAnonymousKernelObject(kref);
+
+      expect(serviceManager.isKernelService(kref)).toBe(false);
+      expect(serviceManager.getKernelServiceByKref(kref)).toBeUndefined();
+      expect(kernelStore.isObjectPinned(kref)).toBe(false);
+    });
+
+    it('leaves other hosted objects alone', () => {
+      const kept = serviceManager.registerAnonymousKernelObject(
+        { which: () => 'kept' },
+        'io-connection',
+      );
+      const dropped = serviceManager.registerAnonymousKernelObject(
+        { which: () => 'dropped' },
+        'io-connection',
+      );
+
+      serviceManager.releaseAnonymousKernelObject(dropped);
+
+      expect(serviceManager.isKernelService(kept)).toBe(true);
+      expect(serviceManager.isKernelService(dropped)).toBe(false);
+    });
+
+    it('is idempotent and tolerates an unregistered kref', () => {
+      const kref = serviceManager.registerAnonymousKernelObject(
+        { ping: () => 'pong' },
+        'io-connection',
+      );
+
+      serviceManager.releaseAnonymousKernelObject(kref);
+      expect(() =>
+        serviceManager.releaseAnonymousKernelObject(kref),
+      ).not.toThrow();
+      expect(() =>
+        serviceManager.releaseAnonymousKernelObject('ko9999'),
+      ).not.toThrow();
+    });
+  });
 });
diff --git a/packages/ocap-kernel/src/KernelServiceManager.ts b/packages/ocap-kernel/src/KernelServiceManager.ts
index c907f110b7..bc236e6113 100644
--- a/packages/ocap-kernel/src/KernelServiceManager.ts
+++ b/packages/ocap-kernel/src/KernelServiceManager.ts
@@ -1,4 +1,5 @@
 import { E } from '@endo/eventual-send';
+import type { ExpectedKernelErrorCode } from '@metamask/kernel-errors';
 import type { Logger } from '@metamask/logger';
 
 import type { KernelQueue } from './KernelQueue.ts';
@@ -8,6 +9,14 @@ import type { KRef, KernelMessage } from './types.ts';
 import { assert } from './utils/assert.ts';
 
 export type KernelService = {
+  /**
+   * The service's name. For services registered with
+   * `registerKernelServiceObject` this is the key vats name in their
+   * cluster config's `services` list, and it is unique. For objects
+   * registered with `registerAnonymousKernelObject` it is only a
+   * diagnostic label: those objects are deliberately absent from the
+   * name index and need not be unique.
+   */
   name: string;
   kref: KRef;
   service: object;
@@ -103,6 +112,108 @@ export class KernelServiceManager {
     this.#kernelStore.deleteKernelServiceKref(name);
   }
 
+  /**
+   * Register a kernel-hosted object reachable *only* by reference.
+   *
+   * Unlike `registerKernelServiceObject`, this enters the object in the
+   * by-kref routing table but deliberately not in the name index, so it
+   * has no name in the global service namespace and cannot be requested
+   * via a cluster config's `services` list. The only way to obtain one
+   * is to be handed the reference, which is what makes it suitable for
+   * per-session objects such as an accepted IO connection: authority is
+   * conveyed by an unforgeable reference rather than by a string that
+   * anything able to name it could use.
+   *
+   * The returned kref is meant to be passed to `kslot()` so a kernel
+   * service method can return the object to a vat, which receives it as
+   * an ordinary Presence.
+   *
+   * The object is pinned, so it stays alive until
+   * `releaseAnonymousKernelObject` is called; the registrar owns that
+   * lifetime.
+   *
+   * @param service - The object to host.
+   * @param label - A diagnostic label. Need not be unique; it is never
+   * used for lookup.
+   * @returns The kref of the newly hosted object.
+   */
+  registerAnonymousKernelObject(service: object, label: string): KRef {
+    const kref = this.#kernelStore.initKernelObject('kernel');
+    this.#kernelStore.pinObject(kref);
+    // Recorded persistently so `releaseAbandonedAnonymousKernelObjects` can
+    // find it after a restart. The routing entry below is in-memory only,
+    // and an anonymous object has no name to be re-registered under, so one
+    // that outlives its incarnation is unreachable yet still pinned.
+    this.#kernelStore.addAnonymousKernelObject(kref);
+    this.#kernelServicesByObject.set(kref, {
+      name: label,
+      kref,
+      service,
+      systemOnly: false,
+    });
+    return kref;
+  }
+
+  /**
+   * Discard anonymous kernel objects left behind by a previous incarnation.
+   *
+   * These exist to host things that cannot outlive the process — an accepted
+   * socket connection, say — so any that survived a restart are garbage, and
+   * harmful if left: still pinned, so they accumulate with every restart.
+   *
+   * Note what this does *not* guarantee. The kernel object is deleted only
+   * once nothing references it, which with the current `(1, 1)` refcount
+   * baseline (see #1006) is never; a survivor therefore keeps its `'kernel'`
+   * owner, and a delivery to it still routes to `invokeKernelService`. That
+   * case is made survivable there, by rejecting the caller's promise rather
+   * than throwing, and not here.
+   *
+   * Runs before the run queue starts, so the unpinning is complete before
+   * anything can address one of these krefs.
+   *
+   * @returns The number of objects discarded.
+   */
+  releaseAbandonedAnonymousKernelObjects(): number {
+    const abandoned = this.#kernelStore
+      .getAnonymousKernelObjects()
+      .filter((kref) => !this.#kernelServicesByObject.has(kref));
+    for (const kref of abandoned) {
+      this.#kernelStore.unpinObject(kref);
+      const { reachable, recognizable } =
+        this.#kernelStore.getObjectRefCount(kref);
+      if (reachable === 0 && recognizable === 0) {
+        this.#kernelStore.deleteKernelObject(kref);
+      }
+      this.#kernelStore.removeAnonymousKernelObject(kref);
+    }
+    return abandoned.length;
+  }
+
+  /**
+   * Release an object registered with `registerAnonymousKernelObject`,
+   * unpinning it and removing it from the routing table. Idempotent, and
+   * safe to call for a kref that was never registered.
+   *
+   * The kernel object itself is deleted here once nothing references it,
+   * rather than being left to `collectGarbage`, which skips kernel-owned
+   * objects. With the current refcount baseline this branch does not fire;
+   * it is the correct place for the deletion once that changes (see #1006).
+   *
+   * @param kref - The kref of the object to release.
+   */
+  releaseAnonymousKernelObject(kref: KRef): void {
+    if (!this.#kernelServicesByObject.delete(kref)) {
+      return;
+    }
+    this.#kernelStore.unpinObject(kref);
+    const { reachable, recognizable } =
+      this.#kernelStore.getObjectRefCount(kref);
+    if (reachable === 0 && recognizable === 0) {
+      this.#kernelStore.deleteKernelObject(kref);
+    }
+    this.#kernelStore.removeAnonymousKernelObject(kref);
+  }
+
   /**
    * Get a kernel service by name.
    *
@@ -148,7 +259,25 @@ export class KernelServiceManager {
   invokeKernelService(target: KRef, message: KernelMessage): void {
     const kernelService = this.#kernelServicesByObject.get(target);
     if (!kernelService) {
-      throw Error(`No registered service for ${target}`);
+      // Reachable, and not necessarily a kernel bug: an anonymous kernel
+      // object hosts something that cannot outlive the process, such as an
+      // accepted socket connection. A vat holding one across a restart, or a
+      // message to one still sitting in the run queue from the previous
+      // incarnation, arrives here with nothing registered.
+      //
+      // Rejecting rather than throwing is the point. A throw escapes the
+      // crank and takes the run loop with it, so one unreachable reference
+      // becomes a dead kernel — and the sweep in `Kernel.#init` cannot
+      // prevent that on its own, since the object survives with its `kernel`
+      // owner intact whenever a vat import or queued message still
+      // references it. This mirrors what `KernelRouter` already does for a
+      // delivery whose endpoint has vanished.
+      this.#failMessage(
+        message.result,
+        'ENDPOINT_UNREACHABLE',
+        `No registered service for ${target}`,
+      );
+      return;
     }
     const { methargs, result } = message;
     const [method, args] = kunser(methargs) as [string, unknown[]];
@@ -176,27 +305,36 @@ export class KernelServiceManager {
           return undefined;
         })
         .catch((problem: unknown) => {
-          if (result) {
-            const detail =
-              problem instanceof Error ? problem.message : String(problem);
-            this.#kernelQueue.resolvePromises('kernel', [
-              [result, true, makeKernelError('DELIVERY_FAILED', detail)],
-            ]);
-          } else {
-            this.#logger?.error('Error in kernel service method:', problem);
-          }
+          this.#failMessage(result, 'DELIVERY_FAILED', problem);
         });
     } catch (syncError) {
       // Handle synchronous errors thrown before returning a Promise
-      if (result) {
-        const detail =
-          syncError instanceof Error ? syncError.message : String(syncError);
-        this.#kernelQueue.resolvePromises('kernel', [
-          [result, true, makeKernelError('DELIVERY_FAILED', detail)],
-        ]);
-      } else {
-        this.#logger?.error('Error in kernel service method:', syncError);
-      }
+      this.#failMessage(result, 'DELIVERY_FAILED', syncError);
+    }
+  }
+
+  /**
+   * Report a failed kernel service message by rejecting the caller's result
+   * promise. A message sent with no result promise has nobody to report to,
+   * so the problem is logged instead.
+   *
+   * @param result - The kref of the message's result promise, if it has one.
+   * @param code - The kernel error code to report to the caller.
+   * @param problem - The error or description of what went wrong.
+   */
+  #failMessage(
+    result: KRef | null | undefined,
+    code: ExpectedKernelErrorCode,
+    problem: unknown,
+  ): void {
+    if (result) {
+      const detail =
+        problem instanceof Error ? problem.message : String(problem);
+      this.#kernelQueue.resolvePromises('kernel', [
+        [result, true, makeKernelError(code, detail)],
+      ]);
+    } else {
+      this.#logger?.error('Error in kernel service method:', problem);
     }
   }
 }
diff --git a/packages/ocap-kernel/src/index.ts b/packages/ocap-kernel/src/index.ts
index 41a7e4c831..6ea9594f3f 100644
--- a/packages/ocap-kernel/src/index.ts
+++ b/packages/ocap-kernel/src/index.ts
@@ -8,7 +8,7 @@ export type {
   VatEndowments,
 } from './vats/endowments.ts';
 export { initTransport } from './remotes/platform/transport.ts';
-export type { IOChannel, IOChannelFactory } from './io/types.ts';
+export type { IOChannel, IOListener, IOListenerFactory } from './io/types.ts';
 export type {
   Baggage,
   ClusterConfig,
diff --git a/packages/ocap-kernel/src/io/IOManager.test.ts b/packages/ocap-kernel/src/io/IOManager.test.ts
index 31fe62712d..592452fd2b 100644
--- a/packages/ocap-kernel/src/io/IOManager.test.ts
+++ b/packages/ocap-kernel/src/io/IOManager.test.ts
@@ -2,9 +2,9 @@ import { Logger } from '@metamask/logger';
 import { describe, it, expect, vi, beforeEach } from 'vitest';
 
 import { IOManager } from './IOManager.ts';
-import type { IOChannel, IOChannelFactory } from './types.ts';
+import type { IOChannel, IOListener, IOListenerFactory } from './types.ts';
 import type { KernelService } from '../KernelServiceManager.ts';
-import type { IOConfig } from '../types.ts';
+import type { IOConfig, KRef } from '../types.ts';
 
 const makeChannel = (): IOChannel => ({
   read: vi.fn().mockResolvedValue('data'),
@@ -12,43 +12,61 @@ const makeChannel = (): IOChannel => ({
   close: vi.fn().mockResolvedValue(undefined),
 });
 
+const makeListener = (): IOListener => ({
+  accept: vi.fn().mockImplementation(async () => makeChannel()),
+  close: vi.fn().mockResolvedValue(undefined),
+});
+
+/** The vat-facing shape of a listener service, for driving it in tests. */
+type ListenerFacet = { accept: () => Promise<unknown> };
+
 describe('IOManager', () => {
-  let factory: IOChannelFactory;
+  let factory: IOListenerFactory;
   let registerService: ReturnType<typeof vi.fn>;
   let unregisterService: ReturnType<typeof vi.fn>;
+  let registerAnonymous: ReturnType<typeof vi.fn>;
+  let releaseAnonymous: ReturnType<typeof vi.fn>;
   let logger: Logger;
   let manager: IOManager;
-  let channels: IOChannel[];
+  let listeners: IOListener[];
+  let registeredServices: Map<string, ListenerFacet>;
+  let nextAnonymousId: number;
 
   beforeEach(() => {
-    channels = [];
+    listeners = [];
+    registeredServices = new Map();
+    nextAnonymousId = 0;
+
     factory = vi.fn(async () => {
-      const ch = makeChannel();
-      channels.push(ch);
-      return ch;
-    }) as unknown as IOChannelFactory;
-
-    registerService = vi.fn(
-      (name: string): KernelService => ({
-        name,
-        kref: `ko${name}`,
-        service: {},
-        systemOnly: false,
-      }),
-    );
+      const listener = makeListener();
+      listeners.push(listener);
+      return listener;
+    }) as unknown as IOListenerFactory;
+
+    registerService = vi.fn((name: string, service: object): KernelService => {
+      registeredServices.set(name, service as unknown as ListenerFacet);
+      return { name, kref: `ko${name}`, service, systemOnly: false };
+    });
     unregisterService = vi.fn();
+    registerAnonymous = vi.fn((): KRef => {
+      nextAnonymousId += 1;
+      return `ko${900 + nextAnonymousId}` as KRef;
+    });
+    releaseAnonymous = vi.fn();
     logger = new Logger('test');
 
     manager = new IOManager({
       factory,
       registerService,
       unregisterService,
+      registerAnonymous,
+      releaseAnonymous,
       logger,
     });
   });
 
   describe('createChannels', () => {
-    it('creates channels and registers services', async () => {
+    it('creates listeners and registers services', async () => {
       const ioConfig: Record<string, IOConfig> = {
         repl: { type: 'socket', path: '/tmp/repl.sock' } as IOConfig,
       };
@@ -62,7 +80,7 @@ describe('IOManager', () => {
       );
     });
 
-    it('creates multiple channels', async () => {
+    it('creates multiple listeners', async () => {
       const ioConfig: Record<string, IOConfig> = {
         input: { type: 'socket', path: '/tmp/in.sock' } as IOConfig,
         output: { type: 'socket', path: '/tmp/out.sock' } as IOConfig,
@@ -74,21 +92,36 @@ describe('IOManager', () => {
       expect(registerService).toHaveBeenCalledTimes(2);
     });
 
+    it('hosts connections accepted through the registered service', async () => {
+      await manager.createChannels('s1', {
+        repl: { type: 'socket', path: '/tmp/repl.sock' } as IOConfig,
+      });
+
+      await registeredServices.get('io:s1:repl')?.accept();
+
+      expect(registerAnonymous).toHaveBeenCalledWith(
+        expect.any(Object),
+        'io:s1:repl:c1',
+      );
+    });
+
     it('cleans up on factory failure', async () => {
-      const successChannel = makeChannel();
+      const successListener = makeListener();
       let callCount = 0;
       const failingFactory = vi.fn(async () => {
         callCount += 1;
         if (callCount === 2) {
           throw new Error('factory error');
         }
-        return successChannel;
-      }) as unknown as IOChannelFactory;
+        return successListener;
+      }) as unknown as IOListenerFactory;
 
       const mgr = new IOManager({
         factory: failingFactory,
         registerService,
         unregisterService,
+        registerAnonymous,
+        releaseAnonymous,
         logger,
       });
 
@@ -101,20 +134,52 @@ describe('IOManager', () => {
         'factory error',
       );
 
-      expect(successChannel.close).toHaveBeenCalledOnce();
+      expect(successListener.close).toHaveBeenCalledOnce();
       expect(unregisterService).toHaveBeenCalledWith('io:s1:first');
     });
 
+    it('releases already-accepted connections on factory failure', async () => {
+      let callCount = 0;
+      const failingFactory = vi.fn(async () => {
+        callCount += 1;
+        if (callCount === 2) {
+          // Accept a connection on the first listener before the second
+          // listener's creation blows up, so rollback has something to undo.
+          await registeredServices.get('io:s1:first')?.accept();
+          throw new Error('factory error');
+        }
+        return makeListener();
+      }) as unknown as IOListenerFactory;
+
+      const mgr = new IOManager({
+        factory: failingFactory,
+        registerService,
+        unregisterService,
+        registerAnonymous,
+        releaseAnonymous,
+        logger,
+      });
+
+      await expect(
+        mgr.createChannels('s1', {
+          first: { type: 'socket', path: '/tmp/a.sock' } as IOConfig,
+          second: { type: 'socket', path: '/tmp/b.sock' } as IOConfig,
+        }),
+      ).rejects.toThrow('factory error');
+
+      expect(releaseAnonymous).toHaveBeenCalledWith('ko901');
+    });
+
     it('does not mask factory error when unregister fails during rollback', async () => {
-      const successChannel = makeChannel();
+      const successListener = makeListener();
       let callCount = 0;
       const failingFactory = vi.fn(async () => {
         callCount += 1;
         if (callCount === 2) {
           throw new Error('factory error');
         }
-        return successChannel;
-      }) as unknown as IOChannelFactory;
+        return successListener;
+      }) as unknown as IOListenerFactory;
 
       const failingUnregister = vi.fn(() => {
         throw new Error('unregister boom');
@@ -125,6 +190,8 @@ describe('IOManager', () => {
         factory: failingFactory,
         registerService,
         unregisterService: failingUnregister,
+        registerAnonymous,
+        releaseAnonymous,
         logger,
       });
 
@@ -142,12 +209,12 @@ describe('IOManager', () => {
         'Error unregistering IO service "io:s1:first":',
         expect.any(Error),
       );
-      expect(successChannel.close).toHaveBeenCalledOnce();
+      expect(successListener.close).toHaveBeenCalledOnce();
     });
   });
 
   describe('destroyChannels', () => {
-    it('closes channels and unregisters services', async () => {
+    it('closes listeners and unregisters services', async () => {
       const ioConfig: Record<string, IOConfig> = {
         repl: { type: 'socket', path: '/tmp/repl.sock' } as IOConfig,
       };
@@ -155,10 +222,24 @@ describe('IOManager', () => {
       await manager.createChannels('s1', ioConfig);
       await manager.destroyChannels('s1');
 
-      expect(channels[0]?.close).toHaveBeenCalledOnce();
+      expect(listeners[0]?.close).toHaveBeenCalledOnce();
       expect(unregisterService).toHaveBeenCalledWith('io:s1:repl');
     });
 
+    it('releases connections still accepted from the listener', async () => {
+      await manager.createChannels('s1', {
+        repl: { type: 'socket', path: '/tmp/repl.sock' } as IOConfig,
+      });
+      const service = registeredServices.get('io:s1:repl');
+      await service?.accept();
+      await service?.accept();
+
+      await manager.destroyChannels('s1');
+
+      expect(releaseAnonymous).toHaveBeenCalledWith('ko901');
+      expect(releaseAnonymous).toHaveBeenCalledWith('ko902');
+    });
+
     it('is idempotent for unknown subcluster', async () => {
       expect(await manager.destroyChannels('nonexistent')).toBeUndefined();
     });
@@ -173,6 +254,8 @@ describe('IOManager', () => {
         factory,
         registerService,
         unregisterService: failingUnregister,
+        registerAnonymous,
+        releaseAnonymous,
         logger,
       });
 
@@ -185,25 +268,27 @@ describe('IOManager', () => {
         'Error unregistering IO service "io:s1:ch":',
         expect.any(Error),
       );
-      // Channel should still be closed despite unregister failure
-      expect(channels[0]?.close).toHaveBeenCalledOnce();
+      // Listener should still be closed despite unregister failure
+      expect(listeners[0]?.close).toHaveBeenCalledOnce();
     });
 
     it('handles close errors gracefully', async () => {
-      const errorChannel = makeChannel();
-      (errorChannel.close as ReturnType<typeof vi.fn>).mockRejectedValue(
+      const errorListener = makeListener();
+      (errorListener.close as ReturnType<typeof vi.fn>).mockRejectedValue(
         new Error('close failed'),
       );
 
       const errorFactory = vi.fn(
-        async () => errorChannel,
-      ) as unknown as IOChannelFactory;
+        async () => errorListener,
+      ) as unknown as IOListenerFactory;
       const errorSpy = vi.spyOn(logger, 'error');
 
       const mgr = new IOManager({
         factory: errorFactory,
         registerService,
         unregisterService,
+        registerAnonymous,
+        releaseAnonymous,
         logger,
       });
 
@@ -213,14 +298,14 @@ describe('IOManager', () => {
       await mgr.destroyChannels('s1');
 
       expect(errorSpy).toHaveBeenCalledWith(
-        'Error closing IO channel "ch":',
+        'Error closing IO listener "ch":',
         expect.any(Error),
       );
     });
   });
 
   describe('destroyAllChannels', () => {
-    it('destroys channels for all subclusters', async () => {
+    it('destroys listeners for all subclusters', async () => {
       await manager.createChannels('s1', {
         a: { type: 'socket', path: '/tmp/a.sock' } as IOConfig,
       });
@@ -230,13 +315,13 @@ describe('IOManager', () => {
 
       await manager.destroyAllChannels();
 
-      expect(channels[0]?.close).toHaveBeenCalledOnce();
-      expect(channels[1]?.close).toHaveBeenCalledOnce();
+      expect(listeners[0]?.close).toHaveBeenCalledOnce();
+      expect(listeners[1]?.close).toHaveBeenCalledOnce();
       expect(unregisterService).toHaveBeenCalledWith('io:s1:a');
       expect(unregisterService).toHaveBeenCalledWith('io:s2:b');
     });
 
-    it('is safe to call when no channels exist', async () => {
+    it('is safe to call when no listeners exist', async () => {
       expect(await manager.destroyAllChannels()).toBeUndefined();
     });
   });
diff --git a/packages/ocap-kernel/src/io/IOManager.ts b/packages/ocap-kernel/src/io/IOManager.ts
index cd3cdfdc6e..e445c670f7 100644
--- a/packages/ocap-kernel/src/io/IOManager.ts
+++ b/packages/ocap-kernel/src/io/IOManager.ts
@@ -1,9 +1,9 @@
 import type { Logger } from '@metamask/logger';
 
-import { makeIOService } from './io-service.ts';
-import type { IOChannel, IOChannelFactory } from './types.ts';
+import { makeIOListenerService } from './io-service.ts';
+import type { IOListener, IOListenerFactory } from './types.ts';
 import type { KernelService } from '../KernelServiceManager.ts';
-import type { IOConfig } from '../types.ts';
+import type { IOConfig, KRef } from '../types.ts';
 
 type RegisterService = (
   name: string,
@@ -11,30 +11,41 @@ type RegisterService = (
   options?: { systemOnly?: boolean },
 ) => KernelService;
 type UnregisterService = (name: string) => void;
+type RegisterAnonymous = (service: object, label: string) => KRef;
+type ReleaseAnonymous = (kref: KRef) => void;
 
 type IOManagerOptions = {
-  factory: IOChannelFactory;
+  factory: IOListenerFactory;
   registerService: RegisterService;
   unregisterService: UnregisterService;
+  registerAnonymous: RegisterAnonymous;
+  releaseAnonymous: ReleaseAnonymous;
   logger?: Logger;
 };
 
 type SubclusterIOState = {
-  channels: Map<string, IOChannel>;
+  listeners: Map<string, IOListener>;
   serviceNames: string[];
+  /** Krefs of connections accepted from this subcluster's listeners. */
+  connectionKrefs: Set<KRef>;
 };
 
 /**
- * Manages IO channel lifecycle, creating channels at subcluster launch
- * and destroying them at termination.
+ * Manages IO listener lifecycle, creating listeners at subcluster launch
+ * and destroying them — along with any connections accepted from them — at
+ * termination.
  */
 export class IOManager {
-  readonly #factory: IOChannelFactory;
+  readonly #factory: IOListenerFactory;
 
   readonly #registerService: RegisterService;
 
   readonly #unregisterService: UnregisterService;
 
+  readonly #registerAnonymous: RegisterAnonymous;
+
+  readonly #releaseAnonymous: ReleaseAnonymous;
+
   readonly #logger: Logger | undefined;
 
   /** IO state indexed by subcluster ID */
@@ -44,53 +55,75 @@ export class IOManager {
    * Creates a new IOManager instance.
    *
    * @param options - Constructor options.
-   * @param options.factory - Factory for creating IO channels.
+   * @param options.factory - Factory for creating IO listeners.
    * @param options.registerService - Function to register a kernel service.
    * @param options.unregisterService - Function to unregister a kernel service.
+   * @param options.registerAnonymous - Function to host an accepted
+   * connection as a kernel object reachable only by reference.
+   * @param options.releaseAnonymous - Function to release a hosted connection.
    * @param options.logger - Optional logger for diagnostics.
    */
   constructor({
     factory,
     registerService,
     unregisterService,
+    registerAnonymous,
+    releaseAnonymous,
     logger,
   }: IOManagerOptions) {
     this.#factory = factory;
     this.#registerService = registerService;
     this.#unregisterService = unregisterService;
+    this.#registerAnonymous = registerAnonymous;
+    this.#releaseAnonymous = releaseAnonymous;
     this.#logger = logger;
     harden(this);
   }
 
   /**
-   * Create IO channels for a subcluster and register them as kernel services.
+   * Create IO listeners for a subcluster and register them as kernel services.
    *
    * @param subclusterId - The ID of the subcluster.
-   * @param ioConfig - The IO configuration map from channel names to configs.
+   * @param ioConfig - The IO configuration map from listener names to configs.
    */
   async createChannels(
     subclusterId: string,
     ioConfig: Record<string, IOConfig>,
   ): Promise<void> {
-    const channels = new Map<string, IOChannel>();
+    const listeners = new Map<string, IOListener>();
     const serviceNames: string[] = [];
+    const connectionKrefs = new Set<KRef>();
 
     for (const [name, config] of Object.entries(ioConfig)) {
       const serviceName = `io:${subclusterId}:${name}`;
       try {
-        const channel = await this.#factory(name, config);
-        channels.set(name, channel);
-
-        const service = makeIOService(serviceName, channel, config);
+        const listener = await this.#factory(name, config);
+        listeners.set(name, listener);
+
+        const service = makeIOListenerService(serviceName, listener, config, {
+          register: (connection, label) => {
+            const kref = this.#registerAnonymous(connection, label);
+            connectionKrefs.add(kref);
+            return kref;
+          },
+          release: (kref) => {
+            connectionKrefs.delete(kref);
+            this.#releaseAnonymous(kref);
+          },
+        });
         this.#registerService(serviceName, service);
         serviceNames.push(serviceName);
 
         this.#logger?.info(
-          `Created IO channel "${name}" for subcluster ${subclusterId}`,
+          `Created IO listener "${name}" for subcluster ${subclusterId}`,
         );
       } catch (error) {
-        // Clean up any channels we already created before re-throwing
-        await this.#closeChannels(channels);
+        // Clean up anything we already created before re-throwing
+        await this.#closeListeners(listeners);
+        for (const kref of connectionKrefs) {
+          this.#releaseAnonymous(kref);
+        }
+        connectionKrefs.clear();
         for (const registeredName of serviceNames) {
           try {
             this.#unregisterService(registeredName);
@@ -105,11 +138,16 @@ export class IOManager {
       }
     }
 
-    this.#subclusters.set(subclusterId, { channels, serviceNames });
+    this.#subclusters.set(subclusterId, {
+      listeners,
+      serviceNames,
+      connectionKrefs,
+    });
   }
 
   /**
-   * Destroy IO channels for a subcluster and unregister their services.
+   * Destroy IO listeners for a subcluster, unregister their services, and
+   * release any connections still accepted from them.
    *
    * @param subclusterId - The ID of the subcluster.
    */
@@ -127,15 +165,21 @@ export class IOManager {
       }
     }
 
-    await this.#closeChannels(state.channels);
+    // Closing a listener closes its connections at the transport level;
+    // stop hosting them so their krefs don't outlive the subcluster.
+    await this.#closeListeners(state.listeners);
+    for (const kref of state.connectionKrefs) {
+      this.#releaseAnonymous(kref);
+    }
+    state.connectionKrefs.clear();
     this.#subclusters.delete(subclusterId);
 
-    this.#logger?.info(`Destroyed IO channels for subcluster ${subclusterId}`);
+    this.#logger?.info(`Destroyed IO listeners for subcluster ${subclusterId}`);
   }
 
   /**
-   * Destroy all IO channels across all subclusters.
-   * Used during kernel reset to ensure no channels are leaked.
+   * Destroy all IO listeners across all subclusters.
+   * Used during kernel reset to ensure nothing is leaked.
    */
   async destroyAllChannels(): Promise<void> {
     for (const subclusterId of [...this.#subclusters.keys()]) {
@@ -144,16 +188,16 @@ export class IOManager {
   }
 
   /**
-   * Close all channels in a map, logging errors.
+   * Close all listeners in a map, logging errors.
    *
-   * @param channels - The channels to close.
+   * @param listeners - The listeners to close.
    */
-  async #closeChannels(channels: Map<string, IOChannel>): Promise<void> {
-    for (const [name, channel] of channels) {
+  async #closeListeners(listeners: Map<string, IOListener>): Promise<void> {
+    for (const [name, listener] of listeners) {
       try {
-        await channel.close();
+        await listener.close();
       } catch (error) {
-        this.#logger?.error(`Error closing IO channel "${name}":`, error);
+        this.#logger?.error(`Error closing IO listener "${name}":`, error);
       }
     }
   }
diff --git a/packages/ocap-kernel/src/io/index.ts b/packages/ocap-kernel/src/io/index.ts
index a132c8a31a..033e1fe200 100644
--- a/packages/ocap-kernel/src/io/index.ts
+++ b/packages/ocap-kernel/src/io/index.ts
@@ -1,2 +1,2 @@
 export { IOManager } from './IOManager.ts';
-export type { IOChannel, IOChannelFactory } from './types.ts';
+export type { IOChannel, IOListener, IOListenerFactory } from './types.ts';
diff --git a/packages/ocap-kernel/src/io/io-service.test.ts b/packages/ocap-kernel/src/io/io-service.test.ts
index 14c8c0a574..d3a4e9848a 100644
--- a/packages/ocap-kernel/src/io/io-service.test.ts
+++ b/packages/ocap-kernel/src/io/io-service.test.ts
@@ -1,8 +1,25 @@
 import { describe, it, expect, vi } from 'vitest';
 
-import { makeIOService } from './io-service.ts';
-import type { IOChannel } from './types.ts';
-import type { IOConfig } from '../types.ts';
+import {
+  makeIOConnectionService,
+  makeIOListenerService,
+} from './io-service.ts';
+import type { ConnectionHost } from './io-service.ts';
+import type { IOChannel, IOListener } from './types.ts';
+import { krefOf } from '../liveslots/kernel-marshal.ts';
+import type { SlotValue } from '../liveslots/kernel-marshal.ts';
+import type { IOConfig, KRef } from '../types.ts';
+
+type ConnectionFacet = {
+  read: () => Promise<string | null>;
+  write: (data: string) => Promise<void>;
+  close: () => Promise<void>;
+};
+
+type ListenerFacet = {
+  accept: () => Promise<unknown>;
+  close: () => Promise<void>;
+};
 
 const makeChannel = (): IOChannel => ({
   read: vi.fn().mockResolvedValue('hello'),
@@ -17,34 +34,75 @@ const makeConfig = (overrides: Partial<IOConfig> = {}): IOConfig =>
     ...overrides,
   }) as IOConfig;
 
-describe('makeIOService', () => {
+/**
+ * Build a listener that hands out the supplied channels in order, then
+ * reports EOF by resolving `null`.
+ *
+ * @param channels - The channels to yield from successive `accept()` calls.
+ * @returns The listener plus its close spy.
+ */
+function makeListener(channels: IOChannel[]): IOListener {
+  const queue = [...channels];
+  return {
+    accept: vi.fn().mockImplementation(async () => queue.shift() ?? null),
+    close: vi.fn().mockResolvedValue(undefined),
+  };
+}
+
+/**
+ * Build a connection host that allocates sequential fake krefs and records
+ * what was registered and released.
+ *
+ * @returns The host plus its bookkeeping.
+ */
+function makeHost(): ConnectionHost & {
+  registered: { kref: KRef; label: string; connection: object }[];
+  released: KRef[];
+} {
+  const registered: { kref: KRef; label: string; connection: object }[] = [];
+  const released: KRef[] = [];
+  let next = 0;
+  return {
+    registered,
+    released,
+    register: (connection: object, label: string): KRef => {
+      next += 1;
+      const kref = `ko${next}` as KRef;
+      registered.push({ kref, label, connection });
+      return kref;
+    },
+    release: (kref: KRef): void => {
+      released.push(kref);
+    },
+  };
+}
+
+describe('makeIOConnectionService', () => {
   describe('read()', () => {
     it('delegates to the channel', async () => {
       const channel = makeChannel();
-      const service = makeIOService(
-        'io:subclusterFoo:test',
+      const connection = makeIOConnectionService(
+        'io:subclusterFoo:test:c1',
         channel,
         makeConfig(),
-      ) as {
-        read: () => Promise<string | null>;
-      };
-
-      const result = await service.read();
+        vi.fn(),
+      ) as ConnectionFacet;
 
-      expect(result).toBe('hello');
+      expect(await connection.read()).toBe('hello');
       expect(channel.read).toHaveBeenCalledOnce();
     });
 
-    it('throws on write-only channel', async () => {
+    it('throws on a write-only connection', async () => {
       const channel = makeChannel();
-      const service = makeIOService(
-        'io:subclusterFoo:test',
+      const connection = makeIOConnectionService(
+        'io:subclusterFoo:test:c1',
         channel,
         makeConfig({ direction: 'out' }),
-      ) as { read: () => Promise<string | null> };
+        vi.fn(),
+      ) as ConnectionFacet;
 
-      await expect(service.read()).rejects.toThrow(
-        'IO channel "io:subclusterFoo:test" is write-only',
+      await expect(connection.read()).rejects.toThrow(
+        'IO connection "io:subclusterFoo:test:c1" is write-only',
       );
       expect(channel.read).not.toHaveBeenCalled();
     });
@@ -52,14 +110,14 @@ describe('makeIOService', () => {
     it.each(['in', 'inout'] as const)(
       'allows read on direction=%s',
       async (direction) => {
-        const channel = makeChannel();
-        const service = makeIOService(
-          'io:subclusterFoo:test',
-          channel,
+        const connection = makeIOConnectionService(
+          'io:subclusterFoo:test:c1',
+          makeChannel(),
           makeConfig({ direction }),
-        ) as { read: () => Promise<string | null> };
+          vi.fn(),
+        ) as ConnectionFacet;
 
-        expect(await service.read()).toBe('hello');
+        expect(await connection.read()).toBe('hello');
       },
     );
   });
@@ -67,29 +125,29 @@ describe('makeIOService', () => {
   describe('write()', () => {
     it('delegates to the channel', async () => {
       const channel = makeChannel();
-      const service = makeIOService(
-        'io:subclusterFoo:test',
+      const connection = makeIOConnectionService(
+        'io:subclusterFoo:test:c1',
         channel,
         makeConfig(),
-      ) as {
-        write: (data: string) => Promise<void>;
-      };
+        vi.fn(),
+      ) as ConnectionFacet;
 
-      await service.write('world');
+      await connection.write('world');
 
       expect(channel.write).toHaveBeenCalledWith('world');
     });
 
-    it('throws on read-only channel', async () => {
+    it('throws on a read-only connection', async () => {
       const channel = makeChannel();
-      const service = makeIOService(
-        'io:subclusterFoo:test',
+      const connection = makeIOConnectionService(
+        'io:subclusterFoo:test:c1',
         channel,
         makeConfig({ direction: 'in' }),
-      ) as { write: (data: string) => Promise<void> };
+        vi.fn(),
+      ) as ConnectionFacet;
 
-      await expect(service.write('data')).rejects.toThrow(
-        'IO channel "io:subclusterFoo:test" is read-only',
+      await expect(connection.write('data')).rejects.toThrow(
+        'IO connection "io:subclusterFoo:test:c1" is read-only',
       );
       expect(channel.write).not.toHaveBeenCalled();
     });
@@ -97,32 +155,230 @@ describe('makeIOService', () => {
     it.each(['out', 'inout'] as const)(
       'allows write on direction=%s',
       async (direction) => {
-        const channel = makeChannel();
-        const service = makeIOService(
-          'io:subclusterFoo:test',
-          channel,
+        const connection = makeIOConnectionService(
+          'io:subclusterFoo:test:c1',
+          makeChannel(),
           makeConfig({ direction }),
-        ) as { write: (data: string) => Promise<void> };
+          vi.fn(),
+        ) as ConnectionFacet;
 
-        expect(await service.write('data')).toBeUndefined();
+        expect(await connection.write('data')).toBeUndefined();
       },
     );
   });
 
-  describe('direction defaults', () => {
-    it('defaults to inout when direction is not specified', async () => {
+  describe('close()', () => {
+    it('closes the channel and notifies the host', async () => {
       const channel = makeChannel();
-      const service = makeIOService(
-        'io:subclusterFoo:test',
+      const onClose = vi.fn();
+      const connection = makeIOConnectionService(
+        'io:subclusterFoo:test:c1',
         channel,
         makeConfig(),
-      ) as {
-        read: () => Promise<string | null>;
-        write: (data: string) => Promise<void>;
-      };
+        onClose,
+      ) as ConnectionFacet;
 
-      expect(await service.read()).toBe('hello');
-      expect(await service.write('data')).toBeUndefined();
+      await connection.close();
+
+      expect(channel.close).toHaveBeenCalledOnce();
+      expect(onClose).toHaveBeenCalledOnce();
     });
+
+    it('is idempotent', async () => {
+      const channel = makeChannel();
+      const onClose = vi.fn();
+      const connection = makeIOConnectionService(
+        'io:subclusterFoo:test:c1',
+        channel,
+        makeConfig(),
+        onClose,
+      ) as ConnectionFacet;
+
+      await connection.close();
+      await connection.close();
+
+      expect(channel.close).toHaveBeenCalledOnce();
+      expect(onClose).toHaveBeenCalledOnce();
+    });
+
+    it('notifies the host even when the channel close fails', async () => {
+      const channel = makeChannel();
+      (channel.close as unknown as ReturnType<typeof vi.fn>).mockRejectedValue(
+        new Error('boom'),
+      );
+      const onClose = vi.fn();
+      const connection = makeIOConnectionService(
+        'io:subclusterFoo:test:c1',
+        channel,
+        makeConfig(),
+        onClose,
+      ) as ConnectionFacet;
+
+      await expect(connection.close()).rejects.toThrow('boom');
+      expect(onClose).toHaveBeenCalledOnce();
+    });
+  });
+
+  describe('direction defaults', () => {
+    it('defaults to inout when direction is not specified', async () => {
+      const connection = makeIOConnectionService(
+        'io:subclusterFoo:test:c1',
+        makeChannel(),
+        makeConfig(),
+        vi.fn(),
+      ) as ConnectionFacet;
+
+      expect(await connection.read()).toBe('hello');
+      expect(await connection.write('data')).toBeUndefined();
+    });
+  });
+});
+
+describe('makeIOListenerService', () => {
+  it('hosts each accepted connection and returns a reference to it', async () => {
+    const channel = makeChannel();
+    const host = makeHost();
+    const listener = makeIOListenerService(
+      'io:s1:repl',
+      makeListener([channel]),
+      makeConfig(),
+      host,
+    ) as ListenerFacet;
+
+    const result = await listener.accept();
+
+    expect(host.registered).toHaveLength(1);
+    expect(host.registered[0]?.label).toBe('io:s1:repl:c1');
+    // The vat receives a reference, never a raw name it could forge.
+    expect(krefOf(result as SlotValue)).toBe('ko1');
+  });
+
+  it('gives each connection a distinct identity', async () => {
+    const host = makeHost();
+    const listener = makeIOListenerService(
+      'io:s1:repl',
+      makeListener([makeChannel(), makeChannel()]),
+      makeConfig(),
+      host,
+    ) as ListenerFacet;
+
+    const first = await listener.accept();
+    const second = await listener.accept();
+
+    expect(krefOf(first as SlotValue)).toBe('ko1');
+    expect(krefOf(second as SlotValue)).toBe('ko2');
+    expect(host.registered.map((entry) => entry.label)).toStrictEqual([
+      'io:s1:repl:c1',
+      'io:s1:repl:c2',
+    ]);
+  });
+
+  it('isolates connections: each reads only its own channel', async () => {
+    const first = makeChannel();
+    const second = makeChannel();
+    (first.read as unknown as ReturnType<typeof vi.fn>).mockResolvedValue(
+      'from-first',
+    );
+    (second.read as unknown as ReturnType<typeof vi.fn>).mockResolvedValue(
+      'from-second',
+    );
+    const host = makeHost();
+    const listener = makeIOListenerService(
+      'io:s1:repl',
+      makeListener([first, second]),
+      makeConfig(),
+      host,
+    ) as ListenerFacet;
+
+    await listener.accept();
+    await listener.accept();
+
+    const facets = host.registered.map(
+      (entry) => entry.connection as unknown as ConnectionFacet,
+    );
+    expect(await facets[0]?.read()).toBe('from-first');
+    expect(await facets[1]?.read()).toBe('from-second');
+  });
+
+  it('releases a connection from the host when it is closed', async () => {
+    const host = makeHost();
+    const listener = makeIOListenerService(
+      'io:s1:repl',
+      makeListener([makeChannel()]),
+      makeConfig(),
+      host,
+    ) as ListenerFacet;
+
+    await listener.accept();
+    const connection = host.registered[0]
+      ?.connection as unknown as ConnectionFacet;
+    await connection.close();
+
+    expect(host.released).toStrictEqual(['ko1']);
+  });
+
+  it('returns null once the listener is exhausted', async () => {
+    const host = makeHost();
+    const listener = makeIOListenerService(
+      'io:s1:repl',
+      makeListener([]),
+      makeConfig(),
+      host,
+    ) as ListenerFacet;
+
+    expect(await listener.accept()).toBeNull();
+    expect(host.registered).toHaveLength(0);
+  });
+
+  it('delegates close() to the listener', async () => {
+    const underlying = makeListener([]);
+    const listener = makeIOListenerService(
+      'io:s1:repl',
+      underlying,
+      makeConfig(),
+      makeHost(),
+    ) as ListenerFacet;
+
+    await listener.close();
+
+    expect(underlying.close).toHaveBeenCalledOnce();
+  });
+
+  it('stops hosting outstanding connections when the listener closes', async () => {
+    const host = makeHost();
+    const listener = makeIOListenerService(
+      'io:s1:repl',
+      makeListener([makeChannel(), makeChannel()]),
+      makeConfig(),
+      host,
+    ) as ListenerFacet;
+
+    await listener.accept();
+    await listener.accept();
+    // Neither connection was closed by its holder; closing the listener
+    // must still release them, or their krefs stay pinned for the life of
+    // the subcluster.
+    await listener.close();
+
+    expect(host.released).toStrictEqual(['ko1', 'ko2']);
+  });
+
+  it('does not release a connection twice when it was already closed', async () => {
+    const host = makeHost();
+    const listener = makeIOListenerService(
+      'io:s1:repl',
+      makeListener([makeChannel(), makeChannel()]),
+      makeConfig(),
+      host,
+    ) as ListenerFacet;
+
+    await listener.accept();
+    await listener.accept();
+    const first = host.registered[0]?.connection as unknown as ConnectionFacet;
+    await first.close();
+
+    await listener.close();
+
+    expect(host.released).toStrictEqual(['ko1', 'ko2']);
   });
 });
diff --git a/packages/ocap-kernel/src/io/io-service.ts b/packages/ocap-kernel/src/io/io-service.ts
index 0c7d86468e..9fb37b4931 100644
--- a/packages/ocap-kernel/src/io/io-service.ts
+++ b/packages/ocap-kernel/src/io/io-service.ts
@@ -1,37 +1,155 @@
 import { makeDefaultExo } from '@metamask/kernel-utils/exo';
 
-import type { IOChannel } from './types.ts';
+import type { IOChannel, IOListener } from './types.ts';
+import { kslot } from '../liveslots/kernel-marshal.ts';
+import type { KRef } from '../types.ts';
 import type { IOConfig } from '../types.ts';
 
 /**
- * Create a kernel service exo that wraps an IOChannel.
+ * Hooks the listener service uses to host each accepted connection as a
+ * kernel object reachable only by reference.
+ */
+export type ConnectionHost = {
+  /** Host `connection` and return its kref. */
+  register: (connection: object, label: string) => KRef;
+  /** Release a previously hosted connection. */
+  release: (kref: KRef) => void;
+};
+
+/**
+ * Create a kernel service exo wrapping an `IOChannel` for one accepted
+ * connection.
  *
- * @param name - The scoped service name (e.g. `io:s1:repl`).
- * @param channel - The underlying IOChannel to delegate to.
- * @param config - The IO configuration for this channel.
- * @returns A remotable service object with `read()` and `write()` methods.
+ * `direction` is enforced here rather than on the listener, since it is a
+ * property of the data flow rather than of the point of contact.
+ *
+ * Lifetime: the holder must `close()` a connection when finished with it.
+ * A peer disconnecting ends the underlying transport and makes `read()`
+ * report EOF, but does *not* by itself stop the kernel hosting this
+ * object, because the holder still has a live reference to it. Releasing
+ * on EOF instead would be worse than leaking: the vat's c-list still names
+ * the kref, so a subsequent call on the dropped reference would route to
+ * `invokeKernelService` and find nothing registered, failing the caller
+ * with `ENDPOINT_UNREACHABLE` for a connection it never released. Until a
+ * vat dropping the reference is itself observable (see #1006), unreleased
+ * connections are
+ * bounded by their listener's lifetime: `close()` on the listener
+ * releases whatever it handed out, and `IOManager` releases the rest when
+ * the subcluster goes away.
+ *
+ * @param name - The scoped connection name, used as the exo's interface
+ * name (e.g. `io:s1:repl:c3`).
+ * @param channel - The channel for this connection.
+ * @param config - The IO configuration for the owning listener.
+ * @param onClose - Invoked after the channel closes, so the host can stop
+ * hosting this connection.
+ * @returns A remotable with `read()`, `write()`, and `close()`.
  */
-export function makeIOService(
+export function makeIOConnectionService(
   name: string,
   channel: IOChannel,
   config: IOConfig,
+  onClose: () => void,
 ): object {
   const direction = config.direction ?? 'inout';
+  let closed = false;
 
   return makeDefaultExo(name, {
     async read(): Promise<string | null> {
       if (direction === 'out') {
-        throw new Error(`IO channel "${name}" is write-only`);
+        throw new Error(`IO connection "${name}" is write-only`);
       }
       return channel.read();
     },
 
     async write(data: string): Promise<void> {
       if (direction === 'in') {
-        throw new Error(`IO channel "${name}" is read-only`);
+        throw new Error(`IO connection "${name}" is read-only`);
       }
       return channel.write(data);
     },
+
+    async close(): Promise<void> {
+      if (closed) {
+        return;
+      }
+      closed = true;
+      try {
+        await channel.close();
+      } finally {
+        onClose();
+      }
+    },
+  });
+}
+harden(makeIOConnectionService);
+
+/**
+ * Create a kernel service exo that wraps an `IOListener`.
+ *
+ * `accept()` waits for the next peer, wraps that connection in its own exo,
+ * hosts it as an anonymous kernel object, and returns a `kslot` standin so
+ * the calling vat receives it as an ordinary Presence. Each connection is
+ * therefore a distinct object with its own state, and holding one conveys
+ * no access to any other.
+ *
+ * @param name - The scoped service name (e.g. `io:s1:repl`).
+ * @param listener - The underlying listener to delegate to.
+ * @param config - The IO configuration for this listener.
+ * @param host - Hooks for hosting accepted connections as kernel objects.
+ * @returns A remotable service object with `accept()` and `close()`.
+ */
+export function makeIOListenerService(
+  name: string,
+  listener: IOListener,
+  config: IOConfig,
+  host: ConnectionHost,
+): object {
+  let nextConnectionId = 0;
+  /**
+   * Krefs of connections handed out and not yet released, so closing the
+   * listener stops hosting them too. Without this, closing a listener
+   * dropped its sockets but left every accepted connection's kref pinned.
+   */
+  const hostedConnections = new Set<KRef>();
+
+  return makeDefaultExo(name, {
+    async accept(): Promise<unknown> {
+      const channel = await listener.accept();
+      if (!channel) {
+        // Listener closed; report EOF rather than leaving the caller's
+        // accept loop hanging forever.
+        return null;
+      }
+      nextConnectionId += 1;
+      const connectionName = `${name}:c${nextConnectionId}`;
+      // Hosting needs the connection object, but the connection's close
+      // handler needs the resulting kref, so the kref is shared through a
+      // holder that is filled in immediately after registration.
+      const hosted: { kref?: KRef } = {};
+      const connection = makeIOConnectionService(
+        connectionName,
+        channel,
+        config,
+        () => {
+          if (hosted.kref) {
+            hostedConnections.delete(hosted.kref);
+            host.release(hosted.kref);
+          }
+        },
+      );
+      hosted.kref = host.register(connection, connectionName);
+      hostedConnections.add(hosted.kref);
+      return kslot(hosted.kref, connectionName);
+    },
+
+    async close(): Promise<void> {
+      await listener.close();
+      for (const kref of [...hostedConnections]) {
+        hostedConnections.delete(kref);
+        host.release(kref);
+      }
+    },
   });
 }
-harden(makeIOService);
+harden(makeIOListenerService);
diff --git a/packages/ocap-kernel/src/io/types.ts b/packages/ocap-kernel/src/io/types.ts
index f08f00dbd6..44e316b90d 100644
--- a/packages/ocap-kernel/src/io/types.ts
+++ b/packages/ocap-kernel/src/io/types.ts
@@ -3,6 +3,10 @@ import type { IOConfig } from '../types.ts';
 /**
  * A platform-agnostic IO channel that supports reading and writing data.
  * Implementations are platform-specific (e.g., Unix domain sockets in Node.js).
+ *
+ * A channel represents a *single* connection: one bidirectional stream of
+ * data with one peer. Serving several peers concurrently means holding
+ * several channels, one per peer, obtained from an `IOListener`.
  */
 export type IOChannel = {
   /** Read the next unit of data, or `null` on EOF/disconnect. */
@@ -14,14 +18,39 @@ export type IOChannel = {
 };
 
 /**
- * Factory function that creates an IOChannel for a given configuration.
+ * A platform-agnostic endpoint that peers connect to, yielding one
+ * `IOChannel` per connection.
+ *
+ * This is the BSD listen/accept split: the listener is the stable,
+ * configured point of contact (one socket path, one entry in a cluster
+ * config's `io` map), while each accepted connection is a separate object
+ * with its own state. Sessions are isolated because they are distinct
+ * objects, so a holder of one connection has no way to reach another.
+ */
+export type IOListener = {
+  /**
+   * Wait for the next peer to connect and return a channel for it.
+   *
+   * Resolves to `null` once the listener has been closed, so an accept
+   * loop can terminate rather than hang.
+   */
+  accept(): Promise<IOChannel | null>;
+  /**
+   * Stop listening and close every connection accepted from this
+   * listener.
+   */
+  close(): Promise<void>;
+};
+
+/**
+ * Factory function that creates an IOListener for a given configuration.
  * Injected from the host environment (e.g., Node.js) into the kernel.
  *
- * @param name - The name of the IO channel (from the cluster config key).
- * @param config - The IO configuration describing the channel type and options.
- * @returns A promise for the created IOChannel.
+ * @param name - The name of the IO listener (from the cluster config key).
+ * @param config - The IO configuration describing the listener type and options.
+ * @returns A promise for the created IOListener.
  */
-export type IOChannelFactory = (
+export type IOListenerFactory = (
   name: string,
   config: IOConfig,
-) => Promise<IOChannel>;
+) => Promise<IOListener>;
diff --git a/packages/ocap-kernel/src/store/index.test.ts b/packages/ocap-kernel/src/store/index.test.ts
index 3db43c3e31..58fefc80c3 100644
--- a/packages/ocap-kernel/src/store/index.test.ts
+++ b/packages/ocap-kernel/src/store/index.test.ts
@@ -40,6 +40,7 @@ describe('kernel store', () => {
     it('has all the expected parts', () => {
       const ks = makeKernelStore(mockKernelDatabase);
       expect(Object.keys(ks).sort()).toStrictEqual([
+        'addAnonymousKernelObject',
         'addCListEntry',
         'addGCActions',
         'addPromiseSubscriber',
@@ -86,6 +87,7 @@ describe('kernel store', () => {
         'getAllRemoteRecords',
         'getAllSystemSubclusterMappings',
         'getAllVatRecords',
+        'getAnonymousKernelObjects',
         'getGCActions',
         'getImporters',
         'getKernelPromise',
@@ -149,6 +151,7 @@ describe('kernel store', () => {
         'recordLastActiveTime',
         'releaseAllSavepoints',
         'releaseSavepoint',
+        'removeAnonymousKernelObject',
         'removeVatFromSubcluster',
         'reset',
         'resolveKernelPromise',
diff --git a/packages/ocap-kernel/src/store/index.ts b/packages/ocap-kernel/src/store/index.ts
index 144b3a6c48..5c8f49fc3d 100644
--- a/packages/ocap-kernel/src/store/index.ts
+++ b/packages/ocap-kernel/src/store/index.ts
@@ -334,6 +334,35 @@ export function makeKernelStore(kdb: KernelDatabase, logger?: Logger) {
     deleteKernelServiceKref(name: string): void {
       kv.delete(`kernelService.${name}`);
     },
+
+    // Anonymous kernel-hosted objects
+    //
+    // Recorded so they can be swept at kernel init. Unlike a named service,
+    // an anonymous object has no name to be re-registered under on boot, so
+    // one surviving a restart is unreachable but still pinned — and a
+    // delivery to it would find nothing registered and kill the run loop.
+    // They are used for things like accepted IO connections, which cannot
+    // outlive the process anyway.
+    getAnonymousKernelObjects(): KRef[] {
+      const raw = kv.get('anonymousKernelObjects');
+      return raw ? (raw.split(',') as KRef[]) : [];
+    },
+    addAnonymousKernelObject(kref: KRef): void {
+      const krefs = new Set(this.getAnonymousKernelObjects());
+      krefs.add(kref);
+      kv.set('anonymousKernelObjects', [...krefs].sort().join(','));
+    },
+    removeAnonymousKernelObject(kref: KRef): void {
+      const krefs = new Set(this.getAnonymousKernelObjects());
+      if (!krefs.delete(kref)) {
+        return;
+      }
+      if (krefs.size === 0) {
+        kv.delete('anonymousKernelObjects');
+      } else {
+        kv.set('anonymousKernelObjects', [...krefs].sort().join(','));
+      }
+    },
   });
 }
 
diff --git a/packages/ocap-kernel/src/store/methods/queue.test.ts b/packages/ocap-kernel/src/store/methods/queue.test.ts
index d44a3dffb8..39c1ed623c 100644
--- a/packages/ocap-kernel/src/store/methods/queue.test.ts
+++ b/packages/ocap-kernel/src/store/methods/queue.test.ts
@@ -117,6 +117,26 @@ describe('queue store methods', () => {
       expect(mockRunQueue.enqueue).toHaveBeenNthCalledWith(1, message1);
       expect(mockRunQueue.enqueue).toHaveBeenNthCalledWith(2, message2);
     });
+
+    it('resolves an invalidated cache from the database before incrementing', () => {
+      // A negative cache means "length unknown, re-read from the DB".
+      // Incrementing it blindly would yield 0 for a queue that already
+      // holds an item, and since 0 is not negative the stale value would
+      // never be re-read — stranding queued items and losing the run
+      // loop's wakeup.
+      const message: RunQueueItem = {
+        type: 'message',
+        data: { some: 'data' },
+      } as unknown as RunQueueItem;
+      context.runQueueLengthCache = -1;
+      mockKV.set('queue.run.head', '38');
+      mockKV.set('queue.run.tail', '37');
+
+      queueMethods.enqueueRun(message);
+
+      expect(context.runQueueLengthCache).toBe(2);
+      expect(queueMethods.runQueueLength()).toBe(2);
+    });
   });
 
   describe('dequeueRun', () => {
@@ -172,6 +192,21 @@ describe('queue store methods', () => {
       expect(queueMethods.dequeueRun()).toBeUndefined();
       expect(context.runQueueLengthCache).toBe(0);
     });
+
+    it('resolves an invalidated cache from the database before decrementing', () => {
+      const message: RunQueueItem = {
+        type: 'message',
+        data: { some: 'data' },
+      } as unknown as RunQueueItem;
+      mockRunQueue.dequeue.mockReturnValue(message);
+      context.runQueueLengthCache = -1;
+      mockKV.set('queue.run.head', '39');
+      mockKV.set('queue.run.tail', '37');
+
+      expect(queueMethods.dequeueRun()).toStrictEqual(message);
+
+      expect(context.runQueueLengthCache).toBe(1);
+    });
   });
 
   describe('runQueueLength', () => {
diff --git a/packages/ocap-kernel/src/store/methods/queue.ts b/packages/ocap-kernel/src/store/methods/queue.ts
index 54693580e8..01ed054822 100644
--- a/packages/ocap-kernel/src/store/methods/queue.ts
+++ b/packages/ocap-kernel/src/store/methods/queue.ts
@@ -33,6 +33,11 @@ export function getQueueMethods(ctx: StoreContext) {
    * @param message - The message to enqueue.
    */
   function enqueueRun(message: RunQueueItem): void {
+    // Materialize the cache from the database before adjusting it. A
+    // negative cache means "unknown"; incrementing it blindly would turn
+    // that sentinel into a concrete (and wrong) count, and since the
+    // result is no longer negative it would never be re-read.
+    runQueueLength();
     ctx.runQueueLengthCache += 1;
     ctx.runQueue.enqueue(message);
   }
@@ -44,6 +49,9 @@ export function getQueueMethods(ctx: StoreContext) {
    * empty.
    */
   function dequeueRun(): RunQueueItem | undefined {
+    // Materialize the cache before adjusting it, for the same reason as
+    // in `enqueueRun`.
+    runQueueLength();
     ctx.runQueueLengthCache -= 1;
     return ctx.runQueue.dequeue() as RunQueueItem | undefined;
   }
diff --git a/packages/service-discovery-types/CHANGELOG.md b/packages/service-discovery-types/CHANGELOG.md
index e774a6d5af..5dfaa4e353 100644
--- a/packages/service-discovery-types/CHANGELOG.md
+++ b/packages/service-discovery-types/CHANGELOG.md
@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Add `interfaceJsonSchemaToRemotableSpec`, and teach the `JsonSchema` converter to translate the new `interface` variant to a `RemotableSpec`, so `remotable` is no longer among the kinds `JsonSchema` cannot express ([#1007](https://github.com/MetaMask/ocap-kernel/pull/1007))
 - `methodSchemaToMethodSpec` marks any parameter absent from the source `MethodSchema.required` as `optional` on its emitted `ValueSpec`, instead of treating every parameter as required ([#958](https://github.com/MetaMask/ocap-kernel/pull/958))
 
 [Unreleased]: https://github.com/MetaMask/ocap-kernel/
diff --git a/packages/service-discovery-types/src/index.test.ts b/packages/service-discovery-types/src/index.test.ts
index 3f2402ca27..fd876ed6af 100644
--- a/packages/service-discovery-types/src/index.test.ts
+++ b/packages/service-discovery-types/src/index.test.ts
@@ -308,4 +308,97 @@ describe('methodsToRemotableSpec', () => {
       },
     });
   });
+
+  it('translates interface-typed returns into RemotableTypeSpec', () => {
+    const result: RemotableSpec = methodsToRemotableSpec({
+      methods: {
+        makeCounter: {
+          description: 'make a counter',
+          args: {},
+          returns: {
+            type: 'interface',
+            description: 'a stateful counter',
+            methods: {
+              increment: {
+                description: 'bump and return',
+                args: {},
+                returns: { type: 'number' },
+              },
+              reset: {
+                description: 'reset to zero',
+                args: {},
+              },
+            },
+          },
+        },
+      },
+    });
+    expect(result.methods.makeCounter?.returnType).toStrictEqual({
+      kind: 'remotable',
+      spec: {
+        description: 'a stateful counter',
+        methods: {
+          increment: {
+            description: 'bump and return',
+            parameters: [],
+            returnType: { kind: 'number' },
+          },
+          reset: {
+            description: 'reset to zero',
+            parameters: [],
+            returnType: { kind: 'void' },
+          },
+        },
+      },
+    });
+  });
+
+  it('supports interfaces nested inside object returns', () => {
+    const result: RemotableSpec = methodsToRemotableSpec({
+      methods: {
+        buy: {
+          description: 'buy something',
+          args: {},
+          returns: {
+            type: 'object',
+            description: 'purchase result',
+            properties: {
+              handle: { type: 'string' },
+              reviser: {
+                type: 'interface',
+                description: 'follow-up reviser',
+                methods: {
+                  revise: {
+                    description: 'produce next revision',
+                    args: { feedback: { type: 'string' } },
+                    returns: { type: 'string' },
+                  },
+                },
+              },
+            },
+            required: ['handle', 'reviser'],
+          },
+        },
+      },
+    });
+    const returnType = result.methods.buy?.returnType;
+    expect(returnType?.kind).toBe('object');
+    const objectReturn = returnType as Extract<
+      typeof returnType,
+      { kind: 'object' }
+    >;
+    expect(objectReturn.spec.properties.reviser?.type).toStrictEqual({
+      kind: 'remotable',
+      spec: {
+        description: 'follow-up reviser',
+        methods: {
+          revise: {
+            description: 'produce next revision',
+            parameters: [{ description: 'feedback', type: { kind: 'string' } }],
+            returnType: { kind: 'string' },
+          },
+        },
+      },
+    });
+  });
 });
diff --git a/packages/service-discovery-types/src/method-schema-convert.ts b/packages/service-discovery-types/src/method-schema-convert.ts
index e2900d68d9..efcf2ce24e 100644
--- a/packages/service-discovery-types/src/method-schema-convert.ts
+++ b/packages/service-discovery-types/src/method-schema-convert.ts
@@ -11,8 +11,10 @@
  *   use the iteration order of the args record, and we drop the names. The
  *   names are preserved as `ValueSpec.description` if no description was
  *   otherwise present, so they remain human-readable.
- * - `JsonSchema` has no notion of `remotable`, `null`, `void`, `bigint`,
- *   `unknown`, or `union`. The converter never emits those kinds.
+ * - `JsonSchema` has no notion of `null`, `void`, `bigint`, `unknown`, or
+ *   `union`. The converter never emits those kinds. Interfaces
+ *   (`type: 'interface'`) do have a `JsonSchema` counterpart and translate
+ *   to `RemotableTypeSpec`.
  */
 
 import type { JsonSchema, MethodSchema } from '@metamask/kernel-utils';
@@ -47,6 +49,11 @@ export function jsonSchemaToTypeSpec(schema: JsonSchema): TypeSpec {
         kind: 'object',
         spec: jsonSchemaToObjectSpec(schema),
       };
+    case 'interface':
+      return {
+        kind: 'remotable',
+        spec: interfaceJsonSchemaToRemotableSpec(schema),
+      };
     default: {
       // Exhaustive: JsonSchema is a closed union.
       const unreachable: never = schema;
@@ -55,6 +62,28 @@ export function jsonSchemaToTypeSpec(schema: JsonSchema): TypeSpec {
   }
 }
 
+/**
+ * Convert an interface-typed `JsonSchema` to a `RemotableSpec`. The
+ * schema describes an object with methods; each method is converted
+ * recursively via {@link methodSchemaToMethodSpec}.
+ *
+ * @param schema - The source interface-typed JsonSchema.
+ * @returns The equivalent RemotableSpec.
+ */
+export function interfaceJsonSchemaToRemotableSpec(
+  schema: Extract<JsonSchema, { type: 'interface' }>,
+): RemotableSpec {
+  const methods: Record<string, MethodSpec> = {};
+  for (const [name, methodSchema] of Object.entries(schema.methods)) {
+    methods[name] = methodSchemaToMethodSpec(methodSchema);
+  }
+  const out: RemotableSpec = { methods };
+  if (schema.description !== undefined) {
+    out.description = schema.description;
+  }
+  return out;
+}
+
 /**
  * Convert an object-typed `JsonSchema` to an `ObjectSpec`.
  *