diff --git a/runtime/internal/flow/conn/conn.go b/runtime/internal/flow/conn/conn.go
index 5f78b77..d463ef2 100644
--- a/runtime/internal/flow/conn/conn.go
+++ b/runtime/internal/flow/conn/conn.go
@@ -62,8 +62,8 @@
 	lastUsedTime   time.Time
 }
 
-// Ensure that *Conn implements flow.Conn.
-var _ flow.Conn = &Conn{}
+// Ensure that *Conn implements flow.ManagedConn.
+var _ flow.ManagedConn = &Conn{}
 
 // NewDialed dials a new Conn on the given conn.
 func NewDialed(
diff --git a/runtime/internal/flow/conn/flow.go b/runtime/internal/flow/conn/flow.go
index ccea002..a3207dc 100644
--- a/runtime/internal/flow/conn/flow.go
+++ b/runtime/internal/flow/conn/flow.go
@@ -242,7 +242,7 @@
 }
 
 // Conn returns the connection the flow is multiplexed on.
-func (f *flw) Conn() flow.Conn {
+func (f *flw) Conn() flow.ManagedConn {
 	return f.conn
 }
 
diff --git a/runtime/internal/lib/tcputil/tcputil.go b/runtime/internal/lib/tcputil/tcputil.go
index 9a4aeb5..39a167a 100644
--- a/runtime/internal/lib/tcputil/tcputil.go
+++ b/runtime/internal/lib/tcputil/tcputil.go
@@ -41,7 +41,7 @@
 type TCP struct{}
 
 // Dial dials a net.Conn to a the specific address and adds framing to the connection.
-func (TCP) Dial(ctx *context.T, network, address string, timeout time.Duration) (flow.MsgReadWriteCloser, error) {
+func (TCP) Dial(ctx *context.T, network, address string, timeout time.Duration) (flow.Conn, error) {
 	conn, err := net.DialTimeout(network, address, timeout)
 	if err != nil {
 		return nil, err
@@ -49,7 +49,7 @@
 	if err := EnableTCPKeepAlive(conn); err != nil {
 		return nil, err
 	}
-	return framer.New(conn), nil
+	return NewTCPConn(conn), nil
 }
 
 // Resolve performs a DNS resolution on the provided network and address.
@@ -63,7 +63,7 @@
 
 // Listen returns a listener that sets KeepAlive on all accepted connections.
 // Connections returned from the listener will be framed.
-func (TCP) Listen(ctx *context.T, network, address string) (flow.MsgListener, error) {
+func (TCP) Listen(ctx *context.T, network, address string) (flow.Listener, error) {
 	ln, err := net.Listen(network, address)
 	if err != nil {
 		return nil, err
@@ -72,12 +72,12 @@
 }
 
 // tcpListener is a wrapper around net.Listener that sets KeepAlive on all
-// accepted connections and returns framed flow.MsgReadWriteClosers.
+// accepted connections and returns framed flow.Conns.
 type tcpListener struct {
 	netLn net.Listener
 }
 
-func (ln *tcpListener) Accept(ctx *context.T) (flow.MsgReadWriteCloser, error) {
+func (ln *tcpListener) Accept(ctx *context.T) (flow.Conn, error) {
 	conn, err := ln.netLn.Accept()
 	if err != nil {
 		return nil, err
@@ -85,9 +85,22 @@
 	if err := EnableTCPKeepAlive(conn); err != nil {
 		return nil, err
 	}
-	return framer.New(conn), nil
+	return NewTCPConn(conn), nil
 }
 
 func (ln *tcpListener) Addr() net.Addr {
 	return ln.netLn.Addr()
 }
+
+func NewTCPConn(c net.Conn) flow.Conn {
+	return tcpConn{framer.New(c), c.LocalAddr()}
+}
+
+type tcpConn struct {
+	flow.MsgReadWriteCloser
+	localAddr net.Addr
+}
+
+func (c tcpConn) LocalAddr() net.Addr {
+	return c.localAddr
+}
diff --git a/runtime/internal/lib/xwebsocket/conn.go b/runtime/internal/lib/xwebsocket/conn.go
index 7efc3e7..7d2c248 100644
--- a/runtime/internal/lib/xwebsocket/conn.go
+++ b/runtime/internal/lib/xwebsocket/conn.go
@@ -18,12 +18,12 @@
 	"v.io/v23/flow"
 )
 
-// WebsocketConn provides a flow.MsgReadWriteCloser interface for a websocket connection.
-func WebsocketConn(ws *websocket.Conn) flow.MsgReadWriteCloser {
+// WebsocketConn provides a flow.Conn interface for a websocket connection.
+func WebsocketConn(ws *websocket.Conn) flow.Conn {
 	return &wrappedConn{ws: ws}
 }
 
-// wrappedConn provides a flow.MsgReadWriteCloser interface to a websocket.
+// wrappedConn provides a flow.Conn interface to a websocket.
 // The underlying websocket connection needs regular calls to Read to make sure
 // websocket control messages (such as pings) are processed by the websocket
 // library.
@@ -82,6 +82,10 @@
 	return c.ws.Close()
 }
 
+func (c *wrappedConn) LocalAddr() net.Addr {
+	return c.ws.LocalAddr()
+}
+
 // hybridConn is used by the 'hybrid' protocol that can accept
 // either 'tcp' or 'websocket' connections. In particular, it allows
 // for the reader to peek and buffer the first n bytes of a stream
diff --git a/runtime/internal/lib/xwebsocket/conn_test.go b/runtime/internal/lib/xwebsocket/conn_test.go
index cf7a631..c8c6357 100644
--- a/runtime/internal/lib/xwebsocket/conn_test.go
+++ b/runtime/internal/lib/xwebsocket/conn_test.go
@@ -20,14 +20,14 @@
 	"v.io/v23/flow"
 )
 
-func writer(c flow.MsgReadWriteCloser, data []byte, times int, wg *sync.WaitGroup) {
+func writer(c flow.Conn, data []byte, times int, wg *sync.WaitGroup) {
 	defer wg.Done()
 	for i := 0; i < times; i++ {
 		c.WriteMsg(data)
 	}
 }
 
-func reader(t *testing.T, c flow.MsgReadWriteCloser, expected []byte, totalWrites int) {
+func reader(t *testing.T, c flow.Conn, expected []byte, totalWrites int) {
 	totalReads := 0
 	for buf, err := c.ReadMsg(); err == nil; buf, err = c.ReadMsg() {
 		totalReads++
diff --git a/runtime/internal/lib/xwebsocket/hybrid.go b/runtime/internal/lib/xwebsocket/hybrid.go
index ac23c24..c69348a 100644
--- a/runtime/internal/lib/xwebsocket/hybrid.go
+++ b/runtime/internal/lib/xwebsocket/hybrid.go
@@ -8,7 +8,6 @@
 	"net"
 	"time"
 
-	"v.io/x/ref/runtime/internal/lib/framer"
 	"v.io/x/ref/runtime/internal/lib/tcputil"
 
 	"v.io/v23/context"
@@ -17,11 +16,11 @@
 
 type WSH struct{}
 
-// Dial returns flow.MsgReadWriteCloser that can be used with a
+// Dial returns flow.Conn that can be used with a
 // HybridListener but always uses tcp. A client must specifically elect to use
 // websockets by calling websocket.Dialer. The returned net.Conn will report
 // 'tcp' as its Network.
-func (WSH) Dial(ctx *context.T, network, address string, timeout time.Duration) (flow.MsgReadWriteCloser, error) {
+func (WSH) Dial(ctx *context.T, network, address string, timeout time.Duration) (flow.Conn, error) {
 	tcp := mapWebSocketToTCP[network]
 	conn, err := net.DialTimeout(tcp, address, timeout)
 	if err != nil {
@@ -30,7 +29,7 @@
 	if err := tcputil.EnableTCPKeepAlive(conn); err != nil {
 		return nil, err
 	}
-	return framer.New(conn), nil
+	return tcputil.NewTCPConn(conn), nil
 }
 
 // Resolve performs a DNS resolution on the network, address and always
@@ -44,7 +43,7 @@
 	return tcp, tcpAddr.String(), nil
 }
 
-// Listener returns a flow.MsgReadWriteCloser that supports both tcp and
+// Listener returns a flow.Conn that supports both tcp and
 // websockets over the same, single, port. A listen address of
 // --v23.tcp.protocol=wsh --v23.tcp.address=127.0.0.1:8101 means
 // that port 8101 can accept connections that use either tcp or websocket.
@@ -52,6 +51,6 @@
 // to decide if it's a websocket protocol or not. These must be 'GET ' for
 // websockets, all other protocols must guarantee to not send 'GET ' as the
 // first four bytes of the payload.
-func (WSH) Listen(ctx *context.T, protocol, address string) (flow.MsgListener, error) {
+func (WSH) Listen(ctx *context.T, protocol, address string) (flow.Listener, error) {
 	return listener(protocol, address, true)
 }
diff --git a/runtime/internal/lib/xwebsocket/listener.go b/runtime/internal/lib/xwebsocket/listener.go
index 9a3511d..3e6c48b 100644
--- a/runtime/internal/lib/xwebsocket/listener.go
+++ b/runtime/internal/lib/xwebsocket/listener.go
@@ -16,7 +16,6 @@
 	"github.com/gorilla/websocket"
 
 	"v.io/x/ref/internal/logger"
-	"v.io/x/ref/runtime/internal/lib/framer"
 	"v.io/x/ref/runtime/internal/lib/tcputil"
 
 	"v.io/v23/context"
@@ -30,14 +29,14 @@
 	closed bool // GUARDED_BY(mu)
 	mu     sync.Mutex
 
-	acceptQ chan interface{} // flow.MsgReadWriteCloser or error returned by netLn.Accept
+	acceptQ chan interface{} // flow.Conn or error returned by netLn.Accept
 	httpQ   chan net.Conn    // Candidates for websocket upgrades before being added to acceptQ
 	netLn   net.Listener     // The underlying listener
 	httpReq sync.WaitGroup   // Number of active HTTP requests
 	hybrid  bool             // true if running in 'hybrid' mode
 }
 
-func listener(protocol, address string, hybrid bool) (flow.MsgListener, error) {
+func listener(protocol, address string, hybrid bool) (flow.Listener, error) {
 	netLn, err := net.Listen(mapWebSocketToTCP[protocol], address)
 	if err != nil {
 		return nil, err
@@ -54,14 +53,14 @@
 	return ln, nil
 }
 
-func (ln *wsTCPListener) Accept(ctx *context.T) (flow.MsgReadWriteCloser, error) {
+func (ln *wsTCPListener) Accept(ctx *context.T) (flow.Conn, error) {
 	for {
 		item, ok := <-ln.acceptQ
 		if !ok {
 			return nil, NewErrListenerClosed(ctx)
 		}
 		switch v := item.(type) {
-		case flow.MsgReadWriteCloser:
+		case flow.Conn:
 			return v, nil
 		case error:
 			return nil, v
@@ -159,7 +158,7 @@
 		ln.httpQ <- conn
 		return
 	}
-	ln.acceptQ <- framer.New(conn)
+	ln.acceptQ <- tcputil.NewTCPConn(conn)
 }
 
 func (ln *wsTCPListener) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -213,7 +212,7 @@
 		if !ok {
 			return
 		}
-		if conn, ok := item.(flow.MsgReadWriteCloser); ok {
+		if conn, ok := item.(flow.Conn); ok {
 			conn.Close()
 		}
 	}
diff --git a/runtime/internal/lib/xwebsocket/protocol.go b/runtime/internal/lib/xwebsocket/protocol.go
index 1fcd641..0429cc8 100644
--- a/runtime/internal/lib/xwebsocket/protocol.go
+++ b/runtime/internal/lib/xwebsocket/protocol.go
@@ -27,7 +27,7 @@
 
 type WS struct{}
 
-func (WS) Dial(ctx *context.T, protocol, address string, timeout time.Duration) (flow.MsgReadWriteCloser, error) {
+func (WS) Dial(ctx *context.T, protocol, address string, timeout time.Duration) (flow.Conn, error) {
 	var deadline time.Time
 	if timeout > 0 {
 		deadline = time.Now().Add(timeout)
@@ -63,6 +63,6 @@
 	return "ws", tcpAddr.String(), nil
 }
 
-func (WS) Listen(ctx *context.T, protocol, address string) (flow.MsgListener, error) {
+func (WS) Listen(ctx *context.T, protocol, address string) (flow.Listener, error) {
 	return listener(protocol, address, false)
 }
diff --git a/runtime/internal/lib/xwebsocket/ws_test.go b/runtime/internal/lib/xwebsocket/ws_test.go
index 6cfd8d6..2411030 100644
--- a/runtime/internal/lib/xwebsocket/ws_test.go
+++ b/runtime/internal/lib/xwebsocket/ws_test.go
@@ -51,7 +51,7 @@
 	ctx, _ := context.RootContext()
 	address := "127.0.0.1:0"
 	timeout := time.Second
-	acceptCh := make(chan flow.MsgReadWriteCloser)
+	acceptCh := make(chan flow.Conn)
 
 	ln, err := listenObj.Listen(ctx, listenP, address)
 	if err != nil {
@@ -76,7 +76,7 @@
 	go readData(t, accepted, randData)
 }
 
-func writeData(t *testing.T, c flow.MsgReadWriteCloser, data []byte) {
+func writeData(t *testing.T, c flow.Conn, data []byte) {
 	for i := 0; i < numChunks; i++ {
 		if _, err := c.WriteMsg(data[:chunkSize]); err != nil {
 			t.Fatal(err)
@@ -85,7 +85,7 @@
 	}
 }
 
-func readData(t *testing.T, c flow.MsgReadWriteCloser, expected []byte) {
+func readData(t *testing.T, c flow.Conn, expected []byte) {
 	read := make([]byte, len(expected))
 	read = read[:0]
 	for i := 0; i < numChunks; i++ {