diff --git a/akka-actor-tests/src/test/scala/akka/actor/ActorRefSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/ActorRefSpec.scala
index b485aa0931..9edf60b57f 100644
--- a/akka-actor-tests/src/test/scala/akka/actor/ActorRefSpec.scala
+++ b/akka-actor-tests/src/test/scala/akka/actor/ActorRefSpec.scala
@@ -278,32 +278,19 @@ class ActorRefSpec extends AkkaSpec {
         " Use akka.serialization.Serialization.app.withValue(akkaApplication) { ... }"
     }
 
-    "must throw exception on deserialize if not present in local registry and remoting is not enabled" in {
-      val latch = new CountDownLatch(1)
-      val a = actorOf(new InnerActor {
-        override def postStop {
-          // app.registry.unregister(self)
-          latch.countDown
-        }
-      })
-
-      val inetAddress = app.defaultAddress
-
-      val expectedSerializedRepresentation = new SerializedActorRef(a.address, inetAddress)
-
+    "must throw exception on deserialize if not present in actor hierarchy (and remoting is not enabled)" in {
       import java.io._
 
       val baos = new ByteArrayOutputStream(8192 * 32)
       val out = new ObjectOutputStream(baos)
 
-      out.writeObject(a)
+      val serialized = SerializedActorRef(app.hostname, app.port, "/this/path/does/not/exist")
+
+      out.writeObject(serialized)
 
       out.flush
       out.close
 
-      a.stop()
-      latch.await(5, TimeUnit.SECONDS) must be === true
-
       Serialization.app.withValue(app) {
         val in = new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray))
         (intercept[java.lang.IllegalStateException] {
diff --git a/akka-actor-tests/src/test/scala/akka/actor/FSMActorSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/FSMActorSpec.scala
index 05716ea04b..5bf3fcf9d7 100644
--- a/akka-actor-tests/src/test/scala/akka/actor/FSMActorSpec.scala
+++ b/akka-actor-tests/src/test/scala/akka/actor/FSMActorSpec.scala
@@ -217,13 +217,13 @@ class FSMActorSpec extends AkkaSpec(Configuration("akka.actor.debug.fsm" -> true
           app.mainbus.subscribe(testActor, classOf[Logging.Debug])
           fsm ! "go"
           expectMsgPF(1 second, hint = "processing Event(go,null)") {
-            case Logging.Debug(`fsm`, s: String) if s.startsWith("processing Event(go,null) from Actor[testActor") ⇒ true
+            case Logging.Debug(`fsm`, s: String) if s.startsWith("processing Event(go,null) from Actor[" + app.address + "/sys/testActor") ⇒ true
           }
           expectMsg(1 second, Logging.Debug(fsm, "setting timer 't'/1500 milliseconds: Shutdown"))
           expectMsg(1 second, Logging.Debug(fsm, "transition 1 -> 2"))
           fsm ! "stop"
           expectMsgPF(1 second, hint = "processing Event(stop,null)") {
-            case Logging.Debug(`fsm`, s: String) if s.startsWith("processing Event(stop,null) from Actor[testActor") ⇒ true
+            case Logging.Debug(`fsm`, s: String) if s.startsWith("processing Event(stop,null) from Actor[" + app.address + "/sys/testActor") ⇒ true
           }
           expectMsgAllOf(1 second, Logging.Debug(fsm, "canceling timer 't'"), Normal)
           expectNoMsg(1 second)
diff --git a/akka-actor-tests/src/test/scala/akka/actor/LoggingReceiveSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/LoggingReceiveSpec.scala
index 891b09ce5b..a4115fce2b 100644
--- a/akka-actor-tests/src/test/scala/akka/actor/LoggingReceiveSpec.scala
+++ b/akka-actor-tests/src/test/scala/akka/actor/LoggingReceiveSpec.scala
@@ -139,11 +139,15 @@ class LoggingReceiveSpec extends WordSpec with BeforeAndAfterEach with BeforeAnd
         app.mainbus.subscribe(testActor, classOf[Logging.Debug])
         app.mainbus.subscribe(testActor, classOf[Logging.Error])
         within(3 seconds) {
+          val lifecycleGuardian = appLifecycle.guardian
           val supervisor = TestActorRef[TestLogActor](Props[TestLogActor].withFaultHandler(OneForOneStrategy(List(classOf[Throwable]), 5, 5000)))
 
-          expectMsgPF() {
-            case Logging.Debug(`supervisor`, msg: String) if msg startsWith "started" ⇒
-          }
+          val supervisorSet = receiveWhile(messages = 2) {
+            case Logging.Debug(`lifecycleGuardian`, msg: String) if msg startsWith "now supervising" ⇒ 1
+            case Logging.Debug(`supervisor`, msg: String) if msg startsWith "started"                ⇒ 2
+          }.toSet
+          expectNoMsg(Duration.Zero)
+          assert(supervisorSet == Set(1, 2), supervisorSet + " was not Set(1, 2)")
 
           val actor = new TestActorRef[TestLogActor](app, Props[TestLogActor], supervisor, "none")
 
diff --git a/akka-actor/src/main/scala/akka/AkkaApplication.scala b/akka-actor/src/main/scala/akka/AkkaApplication.scala
index 7ec88972fe..8eee358ee9 100644
--- a/akka-actor/src/main/scala/akka/AkkaApplication.scala
+++ b/akka-actor/src/main/scala/akka/AkkaApplication.scala
@@ -167,6 +167,8 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
 
   def port: Int = defaultAddress.getPort
 
+  def address: String = hostname + ":" + port.toString
+
   // this provides basic logging (to stdout) until .start() is called below
   val mainbus = new MainBus(DebugMainBus)
   mainbus.startStdoutLogger(AkkaConfig)
@@ -180,6 +182,11 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
   // TODO think about memory consistency effects when doing funky stuff inside constructor
   val reflective = new ReflectiveAccess(this)
 
+  /**
+   * The root actor path for this application.
+   */
+  val root: ActorPath = new RootActorPath(this)
+
   // TODO think about memory consistency effects when doing funky stuff inside constructor
   val provider: ActorRefProvider = reflective.createProvider
 
@@ -205,14 +212,14 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
   }
   private val guardianProps = Props(new Guardian).withFaultHandler(guardianFaultHandlingStrategy)
 
-  private val guardianInChief: ActorRef =
-    provider.actorOf(guardianProps, provider.theOneWhoWalksTheBubblesOfSpaceTime, "GuardianInChief", true)
+  private val rootGuardian: ActorRef =
+    provider.actorOf(guardianProps, provider.theOneWhoWalksTheBubblesOfSpaceTime, root, true)
 
   protected[akka] val guardian: ActorRef =
-    provider.actorOf(guardianProps, guardianInChief, "ApplicationSupervisor", true)
+    provider.actorOf(guardianProps, rootGuardian, "app", true)
 
   protected[akka] val systemGuardian: ActorRef =
-    provider.actorOf(guardianProps.withCreator(new SystemGuardian), guardianInChief, "SystemSupervisor", true)
+    provider.actorOf(guardianProps.withCreator(new SystemGuardian), rootGuardian, "sys", true)
 
   // TODO think about memory consistency effects when doing funky stuff inside constructor
   val deadLetters = new DeadLetterActorRef(this)
@@ -221,7 +228,7 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
 
   // chain death watchers so that killing guardian stops the application
   deathWatch.subscribe(systemGuardian, guardian)
-  deathWatch.subscribe(guardianInChief, systemGuardian)
+  deathWatch.subscribe(rootGuardian, systemGuardian)
 
   // this starts the reaper actor and the user-configured logging subscribers, which are also actors
   mainbus.start(this)
@@ -239,6 +246,11 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
   val scheduler = new DefaultScheduler
   terminationFuture.onComplete(_ ⇒ scheduler.shutdown())
 
+  /**
+   * Create an actor path under the application supervisor (/app).
+   */
+  def /(actorName: String): ActorPath = guardian.path / actorName
+
   // TODO shutdown all that other stuff, whatever that may be
   def stop(): Unit = {
     guardian.stop()
diff --git a/akka-actor/src/main/scala/akka/actor/Actor.scala b/akka-actor/src/main/scala/akka/actor/Actor.scala
index c7f93b493f..a557e11877 100644
--- a/akka-actor/src/main/scala/akka/actor/Actor.scala
+++ b/akka-actor/src/main/scala/akka/actor/Actor.scala
@@ -213,7 +213,7 @@ trait Actor {
    * Stores the context for this actor, including self, sender, and hotswap.
    */
   @transient
-  private[akka] implicit val context: ActorContext = {
+  protected[akka] implicit val context: ActorContext = {
     val contextStack = ActorCell.contextStack.get
 
     def noContextError =
diff --git a/akka-actor/src/main/scala/akka/actor/ActorCell.scala b/akka-actor/src/main/scala/akka/actor/ActorCell.scala
index 16f1ec7535..7e1e5e5aca 100644
--- a/akka-actor/src/main/scala/akka/actor/ActorCell.scala
+++ b/akka-actor/src/main/scala/akka/actor/ActorCell.scala
@@ -18,7 +18,7 @@ import akka.event.Logging.{ Debug, Warning, Error }
  * Exposes contextual information for the actor and the current message.
  * TODO: everything here for current compatibility - could be limited more
  */
-private[akka] trait ActorContext extends ActorRefFactory with TypedActorFactory {
+trait ActorContext extends ActorRefFactory with TypedActorFactory {
 
   def self: ActorRef with ScalaActorRef
 
@@ -56,7 +56,9 @@ private[akka] object ActorCell {
     override def initialValue = Stack[ActorContext]()
   }
 
-  val emptyChildren = TreeMap[ActorRef, ChildRestartStats]()
+  val emptyChildrenRefs = TreeMap[String, ActorRef]()
+
+  val emptyChildrenStats = TreeMap[ActorRef, ChildRestartStats]()
 }
 
 //vars don't need volatile since it's protected with the mailbox status
@@ -79,7 +81,9 @@ private[akka] class ActorCell(
 
   var futureTimeout: Option[ScheduledFuture[AnyRef]] = None
 
-  var _children = emptyChildren //Reuse same empty instance to avoid allocating new instance of the Ordering and the actual empty instance for every actor
+  var childrenRefs = emptyChildrenRefs
+
+  var childrenStats = emptyChildrenStats
 
   var currentMessage: Envelope = null
 
@@ -125,7 +129,13 @@ private[akka] class ActorCell(
     subject
   }
 
-  final def children: Iterable[ActorRef] = _children.keys
+  final def children: Iterable[ActorRef] = childrenStats.keys
+
+  final def getChild(name: String): Option[ActorRef] = {
+    val isClosed = mailbox.isClosed // fence plus volatile read
+    if (isClosed) None
+    else childrenRefs.get(name)
+  }
 
   final def postMessageToMailbox(message: Any, sender: ActorRef): Unit = dispatcher.dispatch(this, Envelope(message, sender))
 
@@ -222,7 +232,8 @@ private[akka] class ActorCell(
           //Stop supervised actors
           val c = children
           if (c.nonEmpty) {
-            _children = TreeMap.empty
+            childrenRefs = emptyChildrenRefs
+            childrenStats = emptyChildrenStats
             for (child ← c) child.stop()
           }
         }
@@ -238,9 +249,10 @@ private[akka] class ActorCell(
     }
 
     def supervise(child: ActorRef): Unit = {
-      val links = _children
-      if (!links.contains(child)) {
-        _children = _children.updated(child, ChildRestartStats())
+      val stats = childrenStats
+      if (!stats.contains(child)) {
+        childrenRefs = childrenRefs.updated(child.name, child)
+        childrenStats = childrenStats.updated(child, ChildRestartStats())
         if (app.AkkaConfig.DebugLifecycle) app.mainbus.publish(Debug(self, "now supervising " + child))
       } else app.mainbus.publish(Warning(self, "Already supervising " + child))
     }
@@ -311,13 +323,14 @@ private[akka] class ActorCell(
     }
   }
 
-  final def handleFailure(fail: Failed): Unit = _children.get(fail.actor) match {
-    case Some(stats) ⇒ if (!props.faultHandler.handleFailure(fail, stats, _children)) throw fail.cause
+  final def handleFailure(fail: Failed): Unit = childrenStats.get(fail.actor) match {
+    case Some(stats) ⇒ if (!props.faultHandler.handleFailure(fail, stats, childrenStats)) throw fail.cause
     case None        ⇒ app.mainbus.publish(Warning(self, "dropping " + fail + " from unknown child"))
   }
 
   final def handleChildTerminated(child: ActorRef): Unit = {
-    _children -= child
+    childrenRefs -= child.name
+    childrenStats -= child
     props.faultHandler.handleChildTerminated(child, children)
   }
 
diff --git a/akka-actor/src/main/scala/akka/actor/ActorPath.scala b/akka-actor/src/main/scala/akka/actor/ActorPath.scala
new file mode 100644
index 0000000000..062793dcdf
--- /dev/null
+++ b/akka-actor/src/main/scala/akka/actor/ActorPath.scala
@@ -0,0 +1,119 @@
+/**
+ *  Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package akka.actor
+
+import akka.AkkaApplication
+
+object ActorPath {
+  final val separator = "/"
+
+  /**
+   * Create an actor path from a string.
+   */
+  def apply(app: AkkaApplication, path: String): ActorPath =
+    apply(app, split(path))
+
+  /**
+   * Create an actor path from an iterable.
+   */
+  def apply(app: AkkaApplication, path: Iterable[String]): ActorPath =
+    path.foldLeft(app.root)(_ / _)
+
+  /**
+   * Split a string path into an iterable.
+   */
+  def split(path: String): Iterable[String] =
+    if (path.startsWith(separator))
+      path.substring(1).split(separator)
+    else
+      path.split(separator)
+
+  /**
+   * Join an iterable path into a string.
+   */
+  def join(path: Iterable[String]): String =
+    path.mkString(separator, separator, "")
+}
+
+/**
+ * Actor path is a unique path to an actor that shows the creation path
+ * up through the actor tree to the root actor.
+ */
+trait ActorPath {
+  /**
+   * The akka application for this path.
+   */
+  def app: AkkaApplication
+
+  /**
+   * The name of the actor that this path refers to.
+   */
+  def name: String
+
+  /**
+   * The path for the parent actor.
+   */
+  def parent: ActorPath
+
+  /**
+   * Create a new child actor path.
+   */
+  def /(child: String): ActorPath
+
+  /**
+   * Find the ActorRef for this path.
+   */
+  def ref: Option[ActorRef]
+
+  /**
+   * String representation of this path. Different from toString for root path.
+   */
+  def string: String
+
+  /**
+   * Sequence of names for this path.
+   */
+  def path: Iterable[String]
+
+  /**
+   * Is this the root path?
+   */
+  def isRoot: Boolean
+}
+
+class RootActorPath(val app: AkkaApplication) extends ActorPath {
+
+  def name: String = "/"
+
+  def parent: ActorPath = this
+
+  def /(child: String): ActorPath = new ChildActorPath(app, this, child)
+
+  def ref: Option[ActorRef] = app.actorFor(path)
+
+  def string: String = ""
+
+  def path: Iterable[String] = Iterable.empty
+
+  def isRoot: Boolean = true
+
+  override def toString = ActorPath.separator
+}
+
+class ChildActorPath(val app: AkkaApplication, val parent: ActorPath, val name: String) extends ActorPath {
+
+  def /(child: String): ActorPath = new ChildActorPath(app, this, child)
+
+  def ref: Option[ActorRef] = app.actorFor(path)
+
+  def string: String = parent.string + ActorPath.separator + name
+
+  def path: Iterable[String] = parent.path ++ Iterable(name)
+
+  def isRoot: Boolean = false
+
+  override def toString = string
+}
+
diff --git a/akka-actor/src/main/scala/akka/actor/ActorRef.scala b/akka-actor/src/main/scala/akka/actor/ActorRef.scala
index 6a3392f910..ab1ad1fea6 100644
--- a/akka-actor/src/main/scala/akka/actor/ActorRef.scala
+++ b/akka-actor/src/main/scala/akka/actor/ActorRef.scala
@@ -49,7 +49,17 @@ abstract class ActorRef extends java.lang.Comparable[ActorRef] with Serializable
   // Only mutable for RemoteServer in order to maintain identity across nodes
 
   /**
-   * Returns the address for the actor.
+   * Returns the name for this actor. Locally unique (across siblings).
+   */
+  def name: String
+
+  /**
+   * Returns the path for this actor (from this actor up to the root actor).
+   */
+  def path: ActorPath
+
+  /**
+   * Returns the absolute address for this actor in the form hostname:port/path/to/actor.
    */
   def address: String
 
@@ -154,17 +164,15 @@ class LocalActorRef private[akka] (
   _app: AkkaApplication,
   props: Props,
   _supervisor: ActorRef,
-  _givenAddress: String,
+  val path: ActorPath,
   val systemService: Boolean = false,
-  private[akka] val uuid: Uuid = newUuid,
   receiveTimeout: Option[Long] = None,
   hotswap: Stack[PartialFunction[Any, Unit]] = Props.noHotSwap)
   extends ActorRef with ScalaActorRef {
 
-  final val address: String = _givenAddress match {
-    case null | Props.randomAddress ⇒ uuid.toString
-    case other                      ⇒ other
-  }
+  def name = path.name
+
+  def address: String = _app.address + path.toString
 
   private[this] val actorCell = new ActorCell(_app, this, props, _supervisor, receiveTimeout, hotswap)
   actorCell.start()
@@ -283,10 +291,10 @@ trait ScalaActorRef { ref: ActorRef ⇒
  * Memento pattern for serializing ActorRefs transparently
  */
 
-case class SerializedActorRef(address: String, hostname: String, port: Int) {
+case class SerializedActorRef(hostname: String, port: Int, path: String) {
   import akka.serialization.Serialization.app
 
-  def this(address: String, inet: InetSocketAddress) = this(address, inet.getAddress.getHostAddress, inet.getPort)
+  def this(inet: InetSocketAddress, path: String) = this(inet.getAddress.getHostAddress, inet.getPort, path)
 
   @throws(classOf[java.io.ObjectStreamException])
   def readResolve(): AnyRef = {
@@ -331,7 +339,7 @@ trait UnsupportedActorRef extends ActorRef with ScalaActorRef {
 trait MinimalActorRef extends ActorRef with ScalaActorRef {
 
   private[akka] val uuid: Uuid = newUuid()
-  def address = uuid.toString
+  def name: String = uuid.toString
 
   def startsMonitoring(actorRef: ActorRef): ActorRef = actorRef
   def stopsMonitoring(actorRef: ActorRef): ActorRef = actorRef
@@ -365,7 +373,13 @@ object DeadLetterActorRef {
 
 class DeadLetterActorRef(val app: AkkaApplication) extends MinimalActorRef {
   val brokenPromise = new KeptPromise[Any](Left(new ActorKilledException("In DeadLetterActorRef, promises are always broken.")))(app.dispatcher)
-  override val address: String = "akka:internal:DeadLetterActorRef"
+
+  override val name: String = "dead-letter"
+
+  // FIXME (actor path): put this under the sys guardian supervisor
+  val path: ActorPath = app.root / "sys" / name
+
+  def address: String = app.address + path.toString
 
   override def isShutdown(): Boolean = true
 
@@ -384,6 +398,11 @@ class DeadLetterActorRef(val app: AkkaApplication) extends MinimalActorRef {
 abstract class AskActorRef(protected val app: AkkaApplication)(timeout: Timeout = app.AkkaConfig.ActorTimeout, dispatcher: MessageDispatcher = app.dispatcher) extends MinimalActorRef {
   final val result = new DefaultPromise[Any](timeout)(dispatcher)
 
+  // FIXME (actor path): put this under the tmp guardian supervisor
+  val path: ActorPath = app.root / "tmp" / name
+
+  def address: String = app.address + path.toString
+
   {
     val callback: Future[Any] ⇒ Unit = { _ ⇒ app.deathWatch.publish(Terminated(AskActorRef.this)); whenDone() }
     result onComplete callback
diff --git a/akka-actor/src/main/scala/akka/actor/ActorRefProvider.scala b/akka-actor/src/main/scala/akka/actor/ActorRefProvider.scala
index c7c23ef76c..f262c3b8f1 100644
--- a/akka-actor/src/main/scala/akka/actor/ActorRefProvider.scala
+++ b/akka-actor/src/main/scala/akka/actor/ActorRefProvider.scala
@@ -13,26 +13,29 @@ import com.eaio.uuid.UUID
 import akka.AkkaException
 import akka.event.{ ActorClassification, DeathWatch, Logging }
 import akka.dispatch._
+import scala.annotation.tailrec
 
 /**
  * Interface for all ActorRef providers to implement.
  */
 trait ActorRefProvider {
 
-  def actorOf(props: Props, supervisor: ActorRef, address: String): ActorRef = actorOf(props, supervisor, address, false)
+  def actorOf(props: Props, supervisor: ActorRef, name: String): ActorRef = actorOf(props, supervisor, name, false)
 
-  def actorOf(props: RoutedProps, supervisor: ActorRef, address: String): ActorRef
+  def actorOf(props: RoutedProps, supervisor: ActorRef, name: String): ActorRef
 
-  def actorFor(address: String): Option[ActorRef]
+  def actorFor(path: Iterable[String]): Option[ActorRef]
 
   /**
    * What deployer will be used to resolve deployment configuration?
    */
   private[akka] def deployer: Deployer
 
-  private[akka] def actorOf(props: Props, supervisor: ActorRef, address: String, systemService: Boolean): ActorRef
+  private[akka] def actorOf(props: Props, supervisor: ActorRef, name: String, systemService: Boolean): ActorRef
 
-  private[akka] def evict(address: String): Boolean
+  private[akka] def actorOf(props: Props, supervisor: ActorRef, path: ActorPath, systemService: Boolean): ActorRef
+
+  private[akka] def evict(path: String): Boolean
 
   private[akka] def deserialize(actor: SerializedActorRef): Option[ActorRef]
 
@@ -69,12 +72,12 @@ trait ActorRefFactory {
    * the same address can race on the cluster, and then you never know which
    * implementation wins
    */
-  def actorOf(props: Props, address: String): ActorRef = provider.actorOf(props, guardian, address, false)
+  def actorOf(props: Props, name: String): ActorRef = provider.actorOf(props, guardian, name, false)
 
   def actorOf[T <: Actor](implicit m: Manifest[T]): ActorRef = actorOf(Props(m.erasure.asInstanceOf[Class[_ <: Actor]]))
 
-  def actorOf[T <: Actor](address: String)(implicit m: Manifest[T]): ActorRef =
-    actorOf(Props(m.erasure.asInstanceOf[Class[_ <: Actor]]), address)
+  def actorOf[T <: Actor](name: String)(implicit m: Manifest[T]): ActorRef =
+    actorOf(Props(m.erasure.asInstanceOf[Class[_ <: Actor]]), name)
 
   def actorOf[T <: Actor](clazz: Class[T]): ActorRef = actorOf(Props(clazz))
 
@@ -84,10 +87,11 @@ trait ActorRefFactory {
 
   def actorOf(props: RoutedProps): ActorRef = actorOf(props, Props.randomAddress)
 
-  def actorOf(props: RoutedProps, address: String): ActorRef = provider.actorOf(props, guardian, address)
+  def actorOf(props: RoutedProps, name: String): ActorRef = provider.actorOf(props, guardian, name)
 
-  def actorFor(address: String): Option[ActorRef] = provider.actorFor(address)
+  def actorFor(path: String): Option[ActorRef] = actorFor(ActorPath.split(path))
 
+  def actorFor(path: Iterable[String]): Option[ActorRef] = provider.actorFor(path)
 }
 
 class ActorRefProviderException(message: String) extends AkkaException(message)
@@ -110,9 +114,14 @@ class LocalActorRefProvider(val app: AkkaApplication) extends ActorRefProvider {
     @volatile
     var stopped = false
 
-    override def address = app.name + ":BubbleWalker"
+    val name = app.name + "-bubble-walker"
 
-    override def toString = address
+    // FIXME (actor path): move the root path to the new root guardian
+    val path = app.root
+
+    val address = app.address + path.toString
+
+    override def toString = name
 
     def stop() = stopped = true
 
@@ -134,9 +143,26 @@ class LocalActorRefProvider(val app: AkkaApplication) extends ActorRefProvider {
     }
   }
 
+  // FIXME (actor path): this could become a cache for the new tree traversal actorFor
+  // currently still used for tmp actors (e.g. ask actor refs)
   private val actors = new ConcurrentHashMap[String, AnyRef]
 
-  def actorFor(address: String): Option[ActorRef] = actors.get(address) match {
+  // FIXME (actor path): should start at the new root guardian, and not use the tail (just to avoid the expected "app" name for now)
+  def actorFor(path: Iterable[String]): Option[ActorRef] = findInCache(ActorPath.join(path)) orElse findInTree(Some(app.guardian), path.tail)
+
+  @tailrec
+  private def findInTree(start: Option[ActorRef], path: Iterable[String]): Option[ActorRef] = {
+    if (path.isEmpty) start
+    else {
+      val child = start match {
+        case Some(local: LocalActorRef) ⇒ local.underlying.getChild(path.head)
+        case _                          ⇒ None
+      }
+      findInTree(child, path.tail)
+    }
+  }
+
+  private def findInCache(path: String): Option[ActorRef] = actors.get(path) match {
     case null              ⇒ None
     case actor: ActorRef   ⇒ Some(actor)
     case future: Future[_] ⇒ Some(future.get.asInstanceOf[ActorRef])
@@ -145,26 +171,33 @@ class LocalActorRefProvider(val app: AkkaApplication) extends ActorRefProvider {
   /**
    * Returns true if the actor was in the provider's cache and evicted successfully, else false.
    */
-  private[akka] def evict(address: String): Boolean = actors.remove(address) ne null
+  private[akka] def evict(path: String): Boolean = actors.remove(path) ne null
 
-  private[akka] def actorOf(props: Props, supervisor: ActorRef, address: String, systemService: Boolean): ActorRef = {
-    if ((address eq null) || address == Props.randomAddress) {
-      val actor = new LocalActorRef(app, props, supervisor, address, systemService = true)
-      actors.putIfAbsent(actor.address, actor) match {
+  private[akka] def actorOf(props: Props, supervisor: ActorRef, name: String, systemService: Boolean): ActorRef =
+    actorOf(props, supervisor, supervisor.path / name, systemService)
+
+  private[akka] def actorOf(props: Props, supervisor: ActorRef, path: ActorPath, systemService: Boolean): ActorRef = {
+    val name = path.name
+    if ((name eq null) || name == Props.randomAddress) {
+      val randomName: String = newUuid.toString
+      val newPath = path.parent / randomName
+      val actor = new LocalActorRef(app, props, supervisor, newPath, systemService = true)
+      actors.putIfAbsent(newPath.toString, actor) match {
         case null  ⇒ actor
-        case other ⇒ throw new IllegalStateException("Same uuid generated twice for: " + actor + " and " + other)
+        case other ⇒ throw new IllegalStateException("Same path generated twice for: " + actor + " and " + other)
       }
     } else {
       val newFuture = Promise[ActorRef](5000)(app.dispatcher) // FIXME is this proper timeout?
 
-      actors.putIfAbsent(address, newFuture) match {
+      actors.putIfAbsent(path.toString, newFuture) match {
         case null ⇒
           val actor: ActorRef = try {
-            (if (systemService) None else deployer.lookupDeployment(address)) match { // see if the deployment already exists, if so use it, if not create actor
+            // FIXME (actor path): lookup should be by path
+            (if (systemService) None else deployer.lookupDeployment(name)) match { // see if the deployment already exists, if so use it, if not create actor
 
               // create a local actor
               case None | Some(DeploymentConfig.Deploy(_, _, DeploymentConfig.Direct, _, DeploymentConfig.LocalScope)) ⇒
-                new LocalActorRef(app, props, supervisor, address, systemService) // create a local actor
+                new LocalActorRef(app, props, supervisor, path, systemService) // create a local actor
 
               // create a routed actor ref
               case deploy @ Some(DeploymentConfig.Deploy(_, _, routerType, nrOfInstances, DeploymentConfig.LocalScope)) ⇒
@@ -181,10 +214,12 @@ class LocalActorRefProvider(val app: AkkaApplication) extends ActorRefProvider {
                   case RouterType.Custom(implClass) ⇒ () ⇒ Routing.createCustomRouter(implClass)
                 }
 
-                val connections: Iterable[ActorRef] =
-                  if (nrOfInstances.factor > 0) Vector.fill(nrOfInstances.factor)(new LocalActorRef(app, props, supervisor, "", systemService)) else Nil
+                val connections: Iterable[ActorRef] = (1 to nrOfInstances.factor) map { i ⇒
+                  val routedPath = path.parent / (path.name + ":" + i)
+                  new LocalActorRef(app, props, supervisor, routedPath, systemService)
+                }
 
-                actorOf(RoutedProps(routerFactory = routerFactory, connectionManager = new LocalConnectionManager(connections)), supervisor, address)
+                actorOf(RoutedProps(routerFactory = routerFactory, connectionManager = new LocalConnectionManager(connections)), supervisor, path.toString)
 
               case unknown ⇒ throw new Exception("Don't know how to create this actor ref! Why? Got: " + unknown)
             }
@@ -196,7 +231,7 @@ class LocalActorRefProvider(val app: AkkaApplication) extends ActorRefProvider {
           }
 
           newFuture completeWithResult actor
-          actors.replace(address, newFuture, actor)
+          actors.replace(path.toString, newFuture, actor)
           actor
         case actor: ActorRef ⇒
           actor
@@ -210,7 +245,7 @@ class LocalActorRefProvider(val app: AkkaApplication) extends ActorRefProvider {
   /**
    * Creates (or fetches) a routed actor reference, configured by the 'props: RoutedProps' configuration.
    */
-  def actorOf(props: RoutedProps, supervisor: ActorRef, address: String): ActorRef = {
+  def actorOf(props: RoutedProps, supervisor: ActorRef, name: String): ActorRef = {
     // FIXME: this needs to take supervision into account!
 
     //FIXME clustering should be implemented by cluster actor ref provider
@@ -218,16 +253,16 @@ class LocalActorRefProvider(val app: AkkaApplication) extends ActorRefProvider {
     //TODO If address matches an already created actor (Ahead-of-time deployed) return that actor
     //TODO If address exists in config, it will override the specified Props (should we attempt to merge?)
     //TODO If the actor deployed uses a different config, then ignore or throw exception?
-    if (props.connectionManager.isEmpty) throw new ConfigurationException("RoutedProps used for creating actor [" + address + "] has zero connections configured; can't create a router")
+    if (props.connectionManager.isEmpty) throw new ConfigurationException("RoutedProps used for creating actor [" + name + "] has zero connections configured; can't create a router")
     // val clusteringEnabled = ReflectiveAccess.ClusterModule.isEnabled
     // val localOnly = props.localOnly
     // if (clusteringEnabled && !props.localOnly) ReflectiveAccess.ClusterModule.newClusteredActorRef(props)
     // else new RoutedActorRef(props, address)
-    new RoutedActorRef(app, props, address)
+    new RoutedActorRef(app, props, supervisor, name)
   }
 
-  private[akka] def deserialize(actor: SerializedActorRef): Option[ActorRef] = actorFor(actor.address)
-  private[akka] def serialize(actor: ActorRef): SerializedActorRef = new SerializedActorRef(actor.address, app.defaultAddress)
+  private[akka] def deserialize(actor: SerializedActorRef): Option[ActorRef] = actorFor(ActorPath.split(actor.path))
+  private[akka] def serialize(actor: ActorRef): SerializedActorRef = new SerializedActorRef(app.defaultAddress, actor.path.toString)
 
   private[akka] def createDeathWatch(): DeathWatch = new LocalDeathWatch
 
@@ -237,7 +272,7 @@ class LocalActorRefProvider(val app: AkkaApplication) extends ActorRefProvider {
       case t if t.duration.length <= 0 ⇒ new DefaultPromise[Any](0)(app.dispatcher) //Abort early if nonsensical timeout
       case t ⇒
         val a = new AskActorRef(app)(timeout = t) { def whenDone() = actors.remove(this) }
-        assert(actors.putIfAbsent(a.address, a) eq null) //If this fails, we're in deep trouble
+        assert(actors.putIfAbsent(a.path.toString, a) eq null) //If this fails, we're in deep trouble
         recipient.tell(message, a)
         a.result
     }
diff --git a/akka-actor/src/main/scala/akka/actor/package.scala b/akka-actor/src/main/scala/akka/actor/package.scala
index 0178db875f..49d4a2d660 100644
--- a/akka-actor/src/main/scala/akka/actor/package.scala
+++ b/akka-actor/src/main/scala/akka/actor/package.scala
@@ -8,6 +8,10 @@ package object actor {
   implicit def actorRef2Scala(ref: ActorRef): ScalaActorRef = ref.asInstanceOf[ScalaActorRef]
   implicit def scala2ActorRef(ref: ScalaActorRef): ActorRef = ref.asInstanceOf[ActorRef]
 
+  // actor path can be used as an actor ref (note: does a lookup in the app using path.ref)
+  implicit def actorPath2Ref(path: ActorPath): ActorRef = path.ref.getOrElse(path.app.deadLetters)
+  implicit def actorPath2ScalaRef(path: ActorPath): ScalaActorRef = actorPath2Ref(path).asInstanceOf[ScalaActorRef]
+
   type Uuid = com.eaio.uuid.UUID
 
   def newUuid(): Uuid = new Uuid()
diff --git a/akka-actor/src/main/scala/akka/event/Logging.scala b/akka-actor/src/main/scala/akka/event/Logging.scala
index 4fd9b86325..2cd1903948 100644
--- a/akka-actor/src/main/scala/akka/event/Logging.scala
+++ b/akka-actor/src/main/scala/akka/event/Logging.scala
@@ -3,7 +3,7 @@
  */
 package akka.event
 
-import akka.actor.{ Actor, ActorRef, MinimalActorRef, LocalActorRef, Props }
+import akka.actor.{ Actor, ActorPath, ActorRef, MinimalActorRef, LocalActorRef, Props }
 import akka.{ AkkaException, AkkaApplication }
 import akka.AkkaApplication.AkkaConfig
 import akka.util.ReflectiveAccess
@@ -340,6 +340,9 @@ object Logging {
    * <code>akka.stdout-loglevel</code> in <code>akka.conf</code>.
    */
   class StandardOutLogger extends MinimalActorRef with StdOutLogger {
+    override val name: String = "standard-out-logger"
+    val path: ActorPath = null // pathless
+    val address: String = name
     override val toString = "StandardOutLogger"
     override def postMessageToMailbox(obj: Any, sender: ActorRef) { print(obj) }
   }
diff --git a/akka-actor/src/main/scala/akka/routing/Routing.scala b/akka-actor/src/main/scala/akka/routing/Routing.scala
index 88026249bb..e78b99e572 100644
--- a/akka-actor/src/main/scala/akka/routing/Routing.scala
+++ b/akka-actor/src/main/scala/akka/routing/Routing.scala
@@ -104,7 +104,12 @@ abstract private[akka] class AbstractRoutedActorRef(val app: AkkaApplication, va
  * A RoutedActorRef is an ActorRef that has a set of connected ActorRef and it uses a Router to send a message to
  * on (or more) of these actors.
  */
-private[akka] class RoutedActorRef(app: AkkaApplication, val routedProps: RoutedProps, override val address: String) extends AbstractRoutedActorRef(app, routedProps) {
+private[akka] class RoutedActorRef(app: AkkaApplication, val routedProps: RoutedProps, val supervisor: ActorRef, override val name: String) extends AbstractRoutedActorRef(app, routedProps) {
+
+  val path = supervisor.path / name
+
+  // FIXME (actor path): address normally has host and port, what about routed actor ref?
+  def address = "routed:/" + path.toString
 
   @volatile
   private var running: Boolean = true
diff --git a/akka-remote/src/main/scala/akka/remote/Gossiper.scala b/akka-remote/src/main/scala/akka/remote/Gossiper.scala
index 4a5efc4f40..715230893d 100644
--- a/akka-remote/src/main/scala/akka/remote/Gossiper.scala
+++ b/akka-remote/src/main/scala/akka/remote/Gossiper.scala
@@ -157,7 +157,7 @@ class Gossiper(remote: Remote) {
           node ← oldAvailableNodes
           if connectionManager.connectionFor(node).isEmpty
         } {
-          val connectionFactory = () ⇒ RemoteActorRef(remote.server, gossipingNode, remote.remoteDaemonServiceName, None)
+          val connectionFactory = () ⇒ RemoteActorRef(remote.server, gossipingNode, remote.remoteDaemon.path, None)
           connectionManager.putIfAbsent(node, connectionFactory) // create a new remote connection to the new node
           oldState.nodeMembershipChangeListeners foreach (_ nodeConnected node) // notify listeners about the new nodes
         }
@@ -310,7 +310,7 @@ class Gossiper(remote: Remote) {
 
     RemoteSystemDaemonMessageProtocol.newBuilder
       .setMessageType(GOSSIP)
-      .setActorAddress(remote.remoteDaemonServiceName)
+      .setActorAddress(remote.remoteDaemon.path.toString)
       .setPayload(ByteString.copyFrom(gossipAsBytes))
       .build()
   }
diff --git a/akka-remote/src/main/scala/akka/remote/Remote.scala b/akka-remote/src/main/scala/akka/remote/Remote.scala
index 11c3371802..5b5026bd30 100644
--- a/akka-remote/src/main/scala/akka/remote/Remote.scala
+++ b/akka-remote/src/main/scala/akka/remote/Remote.scala
@@ -51,7 +51,7 @@ class Remote(val app: AkkaApplication) {
   val computeGridDispatcher = dispatcherFactory.newDispatcher("akka:compute-grid").build
 
   private[remote] lazy val remoteDaemonSupervisor = app.actorOf(Props(
-    OneForOneStrategy(List(classOf[Exception]), None, None))) // is infinite restart what we want?
+    OneForOneStrategy(List(classOf[Exception]), None, None)), "akka-system-remote-supervisor") // is infinite restart what we want?
 
   private[remote] lazy val remoteDaemon =
     app.provider.actorOf(
@@ -140,7 +140,15 @@ class RemoteSystemDaemon(remote: Remote) extends Actor {
             case Right(instance) ⇒ instance.asInstanceOf[() ⇒ Actor]
           }
 
-        app.actorOf(Props(creator = actorFactory), message.getActorAddress)
+        val actorPath = ActorPath(remote.app, message.getActorAddress)
+        val parent = actorPath.parent.ref
+
+        if (parent.isDefined) {
+          app.provider.actorOf(Props(creator = actorFactory), parent.get, actorPath.name)
+        } else {
+          log.error("Parent actor does not exist, ignoring remote system daemon command [{}]", message)
+        }
+
       } else {
         log.error("Actor 'address' for actor to instantiate is not defined, ignoring remote system daemon command [{}]", message)
       }
@@ -180,7 +188,7 @@ class RemoteSystemDaemon(remote: Remote) extends Actor {
       Props(
         context ⇒ {
           case f: Function0[_] ⇒ try { f() } finally { context.self.stop() }
-        }).copy(dispatcher = computeGridDispatcher), app.guardian, Props.randomAddress, systemService = true) ! payloadFor(message, classOf[Function0[Unit]])
+        }).copy(dispatcher = computeGridDispatcher), app.guardian, app.guardian.path / Props.randomAddress, systemService = true) ! payloadFor(message, classOf[Function0[Unit]])
   }
 
   // FIXME: handle real remote supervision
@@ -189,7 +197,7 @@ class RemoteSystemDaemon(remote: Remote) extends Actor {
       Props(
         context ⇒ {
           case f: Function0[_] ⇒ try { sender ! f() } finally { context.self.stop() }
-        }).copy(dispatcher = computeGridDispatcher), app.guardian, Props.randomAddress, systemService = true) forward payloadFor(message, classOf[Function0[Any]])
+        }).copy(dispatcher = computeGridDispatcher), app.guardian, app.guardian.path / Props.randomAddress, systemService = true) forward payloadFor(message, classOf[Function0[Any]])
   }
 
   // FIXME: handle real remote supervision
@@ -198,7 +206,7 @@ class RemoteSystemDaemon(remote: Remote) extends Actor {
       Props(
         context ⇒ {
           case (fun: Function[_, _], param: Any) ⇒ try { fun.asInstanceOf[Any ⇒ Unit].apply(param) } finally { context.self.stop() }
-        }).copy(dispatcher = computeGridDispatcher), app.guardian, Props.randomAddress, systemService = true) ! payloadFor(message, classOf[Tuple2[Function1[Any, Unit], Any]])
+        }).copy(dispatcher = computeGridDispatcher), app.guardian, app.guardian.path / Props.randomAddress, systemService = true) ! payloadFor(message, classOf[Tuple2[Function1[Any, Unit], Any]])
   }
 
   // FIXME: handle real remote supervision
@@ -207,7 +215,7 @@ class RemoteSystemDaemon(remote: Remote) extends Actor {
       Props(
         context ⇒ {
           case (fun: Function[_, _], param: Any) ⇒ try { sender ! fun.asInstanceOf[Any ⇒ Any](param) } finally { context.self.stop() }
-        }).copy(dispatcher = computeGridDispatcher), app.guardian, Props.randomAddress, systemService = true) forward payloadFor(message, classOf[Tuple2[Function1[Any, Any], Any]])
+        }).copy(dispatcher = computeGridDispatcher), app.guardian, app.guardian.path / Props.randomAddress, systemService = true) forward payloadFor(message, classOf[Tuple2[Function1[Any, Any], Any]])
   }
 
   def handleFailover(message: RemoteSystemDaemonMessageProtocol) {
@@ -227,9 +235,10 @@ class RemoteMessage(input: RemoteMessageProtocol, remote: RemoteSupport, classLo
   lazy val sender: ActorRef =
     if (input.hasSender)
       remote.app.provider.deserialize(
-        SerializedActorRef(input.getSender.getAddress, input.getSender.getHost, input.getSender.getPort)).getOrElse(throw new IllegalStateException("OHNOES"))
+        SerializedActorRef(input.getSender.getHost, input.getSender.getPort, input.getSender.getAddress)).getOrElse(throw new IllegalStateException("OHNOES"))
     else
       remote.app.deadLetters
+
   lazy val recipient: ActorRef = remote.app.actorFor(input.getRecipient.getAddress).getOrElse(remote.app.deadLetters)
 
   lazy val payload: Either[Throwable, AnyRef] =
@@ -276,7 +285,7 @@ trait RemoteMarshallingOps {
    */
   def toRemoteActorRefProtocol(actor: ActorRef): ActorRefProtocol = {
     val rep = app.provider.serialize(actor)
-    ActorRefProtocol.newBuilder.setAddress(rep.address).setHost(rep.hostname).setPort(rep.port).build
+    ActorRefProtocol.newBuilder.setHost(rep.hostname).setPort(rep.port).setAddress(rep.path).build
   }
 
   def createRemoteMessageProtocolBuilder(
diff --git a/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala b/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
index eca7372f3f..d4a0066883 100644
--- a/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
+++ b/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
@@ -51,15 +51,20 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
   def defaultDispatcher = app.dispatcher
   def defaultTimeout = app.AkkaConfig.ActorTimeout
 
-  def actorOf(props: Props, supervisor: ActorRef, address: String, systemService: Boolean): ActorRef =
-    if (systemService) local.actorOf(props, supervisor, address, systemService)
+  private[akka] def actorOf(props: Props, supervisor: ActorRef, name: String, systemService: Boolean): ActorRef =
+    actorOf(props, supervisor, supervisor.path / name, systemService)
+
+  private[akka] def actorOf(props: Props, supervisor: ActorRef, path: ActorPath, systemService: Boolean): ActorRef =
+    if (systemService) local.actorOf(props, supervisor, path, systemService)
     else {
+      val name = path.name
       val newFuture = Promise[ActorRef](5000)(defaultDispatcher) // FIXME is this proper timeout?
 
-      actors.putIfAbsent(address, newFuture) match { // we won the race -- create the actor and resolve the future
+      actors.putIfAbsent(path.toString, newFuture) match { // we won the race -- create the actor and resolve the future
         case null ⇒
           val actor: ActorRef = try {
-            deployer.lookupDeploymentFor(address) match {
+            // FIXME (actor path): lookup should be by path
+            deployer.lookupDeploymentFor(name) match {
               case Some(DeploymentConfig.Deploy(_, _, routerType, nrOfInstances, DeploymentConfig.RemoteScope(remoteAddresses))) ⇒
 
                 // FIXME move to AccrualFailureDetector as soon as we have the Gossiper up and running and remove the option to select impl in the akka.conf file since we only have one
@@ -76,7 +81,7 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
 
                 if (isReplicaNode) {
                   // we are on one of the replica node for this remote actor
-                  local.actorOf(props, supervisor, address, true) //FIXME systemService = true here to bypass Deploy, should be fixed when create-or-get is replaced by get-or-create
+                  local.actorOf(props, supervisor, name, true) //FIXME systemService = true here to bypass Deploy, should be fixed when create-or-get is replaced by get-or-create
                 } else {
 
                   // we are on the single "reference" node uses the remote actors on the replica nodes
@@ -84,25 +89,25 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
                     case RouterType.Direct ⇒
                       if (remoteAddresses.size != 1) throw new ConfigurationException(
                         "Actor [%s] configured with Direct router must have exactly 1 remote node configured. Found [%s]"
-                          .format(address, remoteAddresses.mkString(", ")))
+                          .format(name, remoteAddresses.mkString(", ")))
                       () ⇒ new DirectRouter
 
                     case RouterType.Random ⇒
                       if (remoteAddresses.size < 1) throw new ConfigurationException(
                         "Actor [%s] configured with Random router must have at least 1 remote node configured. Found [%s]"
-                          .format(address, remoteAddresses.mkString(", ")))
+                          .format(name, remoteAddresses.mkString(", ")))
                       () ⇒ new RandomRouter
 
                     case RouterType.RoundRobin ⇒
                       if (remoteAddresses.size < 1) throw new ConfigurationException(
                         "Actor [%s] configured with RoundRobin router must have at least 1 remote node configured. Found [%s]"
-                          .format(address, remoteAddresses.mkString(", ")))
+                          .format(name, remoteAddresses.mkString(", ")))
                       () ⇒ new RoundRobinRouter
 
                     case RouterType.ScatterGather ⇒
                       if (remoteAddresses.size < 1) throw new ConfigurationException(
                         "Actor [%s] configured with ScatterGather router must have at least 1 remote node configured. Found [%s]"
-                          .format(address, remoteAddresses.mkString(", ")))
+                          .format(name, remoteAddresses.mkString(", ")))
                       () ⇒ new ScatterGatherFirstCompletedRouter()(defaultDispatcher, defaultTimeout)
 
                     case RouterType.LeastCPU          ⇒ sys.error("Router LeastCPU not supported yet")
@@ -113,17 +118,17 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
 
                   val connections = (Map.empty[InetSocketAddress, ActorRef] /: remoteAddresses) { (conns, a) ⇒
                     val inetAddr = new InetSocketAddress(a.hostname, a.port)
-                    conns + (inetAddr -> RemoteActorRef(remote.server, inetAddr, address, None))
+                    conns + (inetAddr -> RemoteActorRef(remote.server, inetAddr, path, None))
                   }
 
                   val connectionManager = new RemoteConnectionManager(app, remote, connections)
 
-                  connections.keys foreach { useActorOnNode(_, address, props.creator) }
+                  connections.keys foreach { useActorOnNode(_, path.toString, props.creator) }
 
-                  actorOf(RoutedProps(routerFactory = routerFactory, connectionManager = connectionManager), supervisor, address)
+                  actorOf(RoutedProps(routerFactory = routerFactory, connectionManager = connectionManager), supervisor, name)
                 }
 
-              case deploy ⇒ local.actorOf(props, supervisor, address, systemService)
+              case deploy ⇒ local.actorOf(props, supervisor, name, systemService)
             }
           } catch {
             case e: Exception ⇒
@@ -134,7 +139,7 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
           // actor foreach app.registry.register // only for ActorRegistry backward compat, will be removed later
 
           newFuture completeWithResult actor
-          actors.replace(address, newFuture, actor)
+          actors.replace(path.toString, newFuture, actor)
           actor
         case actor: ActorRef   ⇒ actor
         case future: Future[_] ⇒ future.get.asInstanceOf[ActorRef]
@@ -145,13 +150,13 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
    * Copied from LocalActorRefProvider...
    */
   // FIXME: implement supervision
-  def actorOf(props: RoutedProps, supervisor: ActorRef, address: String): ActorRef = {
-    if (props.connectionManager.isEmpty) throw new ConfigurationException("RoutedProps used for creating actor [" + address + "] has zero connections configured; can't create a router")
-    new RoutedActorRef(app, props, address)
+  def actorOf(props: RoutedProps, supervisor: ActorRef, name: String): ActorRef = {
+    if (props.connectionManager.isEmpty) throw new ConfigurationException("RoutedProps used for creating actor [" + name + "] has zero connections configured; can't create a router")
+    new RoutedActorRef(app, props, supervisor, name)
   }
 
-  def actorFor(address: String): Option[ActorRef] = actors.get(address) match {
-    case null              ⇒ local.actorFor(address)
+  def actorFor(path: Iterable[String]): Option[ActorRef] = actors.get(ActorPath.join(path)) match {
+    case null              ⇒ local.actorFor(path)
     case actor: ActorRef   ⇒ Some(actor)
     case future: Future[_] ⇒ Some(future.get.asInstanceOf[ActorRef])
   }
@@ -162,27 +167,28 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
   /**
    * Returns true if the actor was in the provider's cache and evicted successfully, else false.
    */
-  private[akka] def evict(address: String): Boolean = actors.remove(address) ne null
+  private[akka] def evict(path: String): Boolean = actors.remove(path) ne null
+
   private[akka] def serialize(actor: ActorRef): SerializedActorRef = actor match {
-    case r: RemoteActorRef ⇒ new SerializedActorRef(actor.address, r.remoteAddress)
+    case r: RemoteActorRef ⇒ new SerializedActorRef(r.remoteAddress, actor.path.toString)
     case other             ⇒ local.serialize(actor)
   }
 
   private[akka] def deserialize(actor: SerializedActorRef): Option[ActorRef] = {
     if (optimizeLocalScoped_? && (actor.hostname == app.hostname || actor.hostname == app.defaultAddress.getHostName) && actor.port == app.port) {
-      local.actorFor(actor.address)
+      local.actorFor(ActorPath.split(actor.path))
     } else {
       val remoteInetSocketAddress = new InetSocketAddress(actor.hostname, actor.port) //FIXME Drop the InetSocketAddresses and use RemoteAddress
-      log.debug("{}: Creating RemoteActorRef with address [{}] connected to [{}]", app.defaultAddress, actor.address, remoteInetSocketAddress)
-      Some(RemoteActorRef(remote.server, remoteInetSocketAddress, actor.address, None)) //Should it be None here
+      log.debug("{}: Creating RemoteActorRef with address [{}] connected to [{}]", app.defaultAddress, actor.path, remoteInetSocketAddress)
+      Some(RemoteActorRef(remote.server, remoteInetSocketAddress, ActorPath(app, actor.path), None)) //Should it be None here
     }
   }
 
   /**
    * Using (checking out) actor on a specific node.
    */
-  def useActorOnNode(remoteAddress: InetSocketAddress, actorAddress: String, actorFactory: () ⇒ Actor) {
-    log.debug("[{}] Instantiating Actor [{}] on node [{}]", app.defaultAddress, actorAddress, remoteAddress)
+  def useActorOnNode(remoteAddress: InetSocketAddress, actorPath: String, actorFactory: () ⇒ Actor) {
+    log.debug("[{}] Instantiating Actor [{}] on node [{}]", app.defaultAddress, actorPath, remoteAddress)
 
     val actorFactoryBytes =
       app.serialization.serialize(actorFactory) match {
@@ -192,11 +198,11 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
 
     val command = RemoteSystemDaemonMessageProtocol.newBuilder
       .setMessageType(USE)
-      .setActorAddress(actorAddress)
+      .setActorAddress(actorPath)
       .setPayload(ByteString.copyFrom(actorFactoryBytes))
       .build()
 
-    val connectionFactory = () ⇒ deserialize(new SerializedActorRef(remote.remoteDaemonServiceName, remoteAddress)).get
+    val connectionFactory = () ⇒ deserialize(new SerializedActorRef(remoteAddress, remote.remoteDaemon.path.toString)).get
 
     // try to get the connection for the remote address, if not already there then create it
     val connection = remoteDaemonConnectionManager.putIfAbsent(remoteAddress, connectionFactory)
@@ -245,12 +251,17 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
 private[akka] case class RemoteActorRef private[akka] (
   remote: RemoteSupport,
   remoteAddress: InetSocketAddress,
-  address: String,
+  path: ActorPath,
   loader: Option[ClassLoader])
   extends ActorRef with ScalaActorRef {
+
   @volatile
   private var running: Boolean = true
 
+  def name = path.name
+
+  def address = remoteAddress.getAddress.getHostAddress + ":" + remoteAddress.getPort + path.toString
+
   def isShutdown: Boolean = !running
 
   protected[akka] def sendSystemMessage(message: SystemMessage): Unit = unsupported
diff --git a/akka-remote/src/main/scala/akka/remote/RemoteConnectionManager.scala b/akka-remote/src/main/scala/akka/remote/RemoteConnectionManager.scala
index f76f9d072a..f088cebbca 100644
--- a/akka-remote/src/main/scala/akka/remote/RemoteConnectionManager.scala
+++ b/akka-remote/src/main/scala/akka/remote/RemoteConnectionManager.scala
@@ -74,7 +74,7 @@ class RemoteConnectionManager(
       case (`from`, actorRef) ⇒
         changed = true
         //actorRef.stop()
-        (to, newConnection(actorRef.address, to))
+        (to, newConnection(to, actorRef.path))
       case other ⇒ other
     }
 
@@ -149,7 +149,7 @@ class RemoteConnectionManager(
     }
   }
 
-  private[remote] def newConnection(actorAddress: String, inetSocketAddress: InetSocketAddress) = {
-    RemoteActorRef(remote.server, inetSocketAddress, actorAddress, None)
+  private[remote] def newConnection(inetSocketAddress: InetSocketAddress, actorPath: ActorPath) = {
+    RemoteActorRef(remote.server, inetSocketAddress, actorPath, None)
   }
 }
diff --git a/akka-testkit/src/main/scala/akka/testkit/TestActorRef.scala b/akka-testkit/src/main/scala/akka/testkit/TestActorRef.scala
index c6827a3547..ca2fb13fc8 100644
--- a/akka-testkit/src/main/scala/akka/testkit/TestActorRef.scala
+++ b/akka-testkit/src/main/scala/akka/testkit/TestActorRef.scala
@@ -18,8 +18,8 @@ import akka.AkkaApplication
  * @author Roland Kuhn
  * @since 1.1
  */
-class TestActorRef[T <: Actor](_app: AkkaApplication, _props: Props, _supervisor: ActorRef, address: String)
-  extends LocalActorRef(_app, _props.withDispatcher(new CallingThreadDispatcher(_app)), _supervisor, address, false) {
+class TestActorRef[T <: Actor](_app: AkkaApplication, _props: Props, _supervisor: ActorRef, name: String)
+  extends LocalActorRef(_app, _props.withDispatcher(new CallingThreadDispatcher(_app)), _supervisor, _supervisor.path / name, false) {
   /**
    * Directly inject messages into actor receive behavior. Any exceptions
    * thrown will be available to you, while still being able to use
@@ -34,9 +34,9 @@ class TestActorRef[T <: Actor](_app: AkkaApplication, _props: Props, _supervisor
    */
   def underlyingActor: T = underlyingActorInstance.asInstanceOf[T]
 
-  override def toString = "TestActor[" + address + ":" + uuid + "]"
+  override def toString = "TestActor[" + address + "]"
 
-  override def equals(other: Any) = other.isInstanceOf[TestActorRef[_]] && other.asInstanceOf[TestActorRef[_]].uuid == uuid
+  override def equals(other: Any) = other.isInstanceOf[TestActorRef[_]] && other.asInstanceOf[TestActorRef[_]].address == address
 }
 
 object TestActorRef {
@@ -49,8 +49,13 @@ object TestActorRef {
 
   def apply[T <: Actor](props: Props, address: String)(implicit app: AkkaApplication): TestActorRef[T] = apply[T](props, app.guardian, address)
 
-  def apply[T <: Actor](props: Props, supervisor: ActorRef, address: String)(implicit app: AkkaApplication): TestActorRef[T] =
-    new TestActorRef(app, props, supervisor, address)
+  def apply[T <: Actor](props: Props, supervisor: ActorRef, address: String)(implicit app: AkkaApplication): TestActorRef[T] = {
+    val name: String = address match {
+      case null | Props.randomAddress ⇒ newUuid.toString
+      case given                      ⇒ given
+    }
+    new TestActorRef(app, props, supervisor, name)
+  }
 
   def apply[T <: Actor](implicit m: Manifest[T], app: AkkaApplication): TestActorRef[T] = apply[T](Props.randomAddress)