Repository: firecracker-microvm/firecracker-containerd
Pull request #164 (2019-04-15 19:19:04)

Support VM Agent managing multiple containers.
Signed-off-by: Erik Sipsma <sipsma@amazon.com>

- [x] Use dynamic vsock ports for stdio (host decides and tells the vm what to use via protobuf msg)
- [x] Refactor agent to use one runc shim instance per-Container 
- [x] Refactor shared code between runtime+agent into common struct (called `VMTask` for now)
- [x] Working e2e test (updated `TestMultipleVMs_Isolated` to spin up 3 VMs each with 3 containers)
- [x] More doc strings + log statements all over the place
- [x] commented TODOs for error handling
- [x] fix race condition where container runs and exits before IO can be setup
- [x] further refactoring to get rid of duplicate code between runtime+agent
- [x] fix failure in TestShimExitsUponContainerKill_Isolated e2e test 
- [x] unit tests that need updating and a few new unit tests for the new code

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.


sipsma: Just FYI, the e2e tests are passing the vast majority of the time, but I did see a one-off failure that appears to be an issue from before this PR. In the interest of not expanding this PR further I created [a separate issue I'll follow up on here](https://github.com/firecracker-microvm/firecracker-containerd/issues/169).

Patch:
diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml
index 3e6da28..6125a59 100644
--- a/.buildkite/pipeline.yml
+++ b/.buildkite/pipeline.yml
@@ -63,7 +63,7 @@ steps:
       - "runtime/logs/*"
     command: 'cd runtime && make test-docker-isolated'
 
-  - label: ":fencer: agent tests"
+  - label: ":fencer: agent unit tests"
     agents:
       queue: "${BUILDKITE_AGENT_META_DATA_QUEUE:-default}"
     env:
@@ -71,6 +71,14 @@ steps:
       EXTRAGOARGS: "-v -count=1"
     command: 'cd agent && make test-docker-unit'
 
+  - label: ":gear: vm task unit tests"
+    agents:
+      queue: "${BUILDKITE_AGENT_META_DATA_QUEUE:-default}"
+    env:
+      DOCKER_IMAGE_TAG: "$BUILDKITE_BUILD_NUMBER"
+      EXTRAGOARGS: "-v -count=1"
+    command: 'cd internal/vm && make test-docker-unit'
+
   - label: ":exclamation: event tests"
     agents:
       queue: "${BUILDKITE_AGENT_META_DATA_QUEUE:-default}"
diff --git a/agent/main.go b/agent/main.go
index 9f3f5de..f4235c7 100644
--- a/agent/main.go
+++ b/agent/main.go
@@ -24,9 +24,8 @@ import (
 	"github.com/containerd/containerd/events/exchange"
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/namespaces"
-	"github.com/containerd/containerd/runtime/v2/runc"
 	"github.com/containerd/containerd/runtime/v2/shim"
-	shimapi "github.com/containerd/containerd/runtime/v2/task"
+	taskAPI "github.com/containerd/containerd/runtime/v2/task"
 	"github.com/containerd/containerd/sys"
 	"github.com/containerd/ttrpc"
 	"github.com/mdlayher/vsock"
@@ -88,19 +87,14 @@ func main() {
 	log.G(ctx).WithField("id", id).Info("creating runc shim")
 
 	eventExchange := exchange.NewExchange()
-	runcTaskService, err := runc.New(ctx, id, eventExchange)
-	if err != nil {
-		log.G(ctx).WithError(err).Fatal("failed to create runc shim")
-	}
-
-	taskService := NewTaskService(runcTaskService, cancel)
+	taskService := NewTaskService(eventExchange, cancel)
 
 	server, err := ttrpc.NewServer()
 	if err != nil {
 		log.G(ctx).WithError(err).Fatal("failed to create ttrpc server")
 	}
 
-	shimapi.RegisterTaskService(server, taskService)
+	taskAPI.RegisterTaskService(server, taskService)
 	eventbridge.RegisterGetterService(server, eventbridge.NewGetterService(ctx, eventExchange))
 
 	// Run ttrpc over vsock
@@ -141,12 +135,11 @@ func main() {
 		}
 	})
 
-	if err := group.Wait(); err != nil {
-		log.G(ctx).WithError(err).Warn("shim error")
-	}
-
+	err = group.Wait()
 	log.G(ctx).Info("shutting down agent")
-	if _, err := runcTaskService.Shutdown(ctx, &shimapi.ShutdownRequest{ID: id, Now: true}); err != nil {
-		log.G(ctx).WithError(err).Error("runc shutdown error")
+
+	if err != nil {
+		log.G(ctx).WithError(err).Error("shim error")
+		os.Exit(1)
 	}
 }
diff --git a/agent/service.go b/agent/service.go
index 2d8488c..26d4ef4 100644
--- a/agent/service.go
+++ b/agent/service.go
@@ -15,30 +15,29 @@ package main
 
 import (
 	"context"
-	"io"
-	"io/ioutil"
 	"net"
-	"os"
 	"path/filepath"
-	"syscall"
+	"runtime/debug"
+	"time"
 
 	"github.com/containerd/containerd/cio"
+	"github.com/containerd/containerd/events/exchange"
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/namespaces"
-	"github.com/containerd/containerd/runtime/v2/shim"
-	shimapi "github.com/containerd/containerd/runtime/v2/task"
-	"github.com/containerd/fifo"
+	"github.com/containerd/containerd/runtime/v2/runc"
+	taskAPI "github.com/containerd/containerd/runtime/v2/task"
 	"github.com/gogo/protobuf/types"
 	"github.com/mdlayher/vsock"
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 
-	"github.com/firecracker-microvm/firecracker-containerd/internal"
+	"github.com/firecracker-microvm/firecracker-containerd/internal/bundle"
+	"github.com/firecracker-microvm/firecracker-containerd/internal/vm"
 	proto "github.com/firecracker-microvm/firecracker-containerd/proto/grpc"
 )
 
 const (
-	bundleMountPath  = "/container"
-	defaultStdioPath = "/container/fifo"
+	containerRootDir = "/container"
 )
 
 // TaskService represents inner shim wrapper over runc in order to:
@@ -46,134 +45,177 @@ const (
 // - Add debug logging to simplify debugging
 // - Make place for future extensions as needed
 type TaskService struct {
-	runc    shim.Shim
-	cancels []context.CancelFunc
-	io      *cio.FIFOSet
+	taskManager vm.TaskManager
+
+	eventExchange *exchange.Exchange
+	cancel        func()
 }
 
 // NewTaskService creates new runc shim wrapper
-func NewTaskService(runc shim.Shim, cancel context.CancelFunc) shimapi.TaskService {
+func NewTaskService(eventExchange *exchange.Exchange, cancel context.CancelFunc) taskAPI.TaskService {
 	return &TaskService{
-		runc:    runc,
-		cancels: []context.CancelFunc{cancel},
+		taskManager: vm.NewTaskManager(),
+
+		eventExchange: eventExchange,
+		cancel:        cancel,
 	}
 }
 
-// Create creates a new initial process and container using runc
-func (ts *TaskService) Create(ctx context.Context, req *shimapi.CreateTaskRequest) (*shimapi.CreateTaskResponse, error) {
-	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "bundle": req.Bundle}).Info("create")
+func logPanicAndDie(logger *logrus.Entry) {
+	if err := recover(); err != nil {
+		logger.WithError(err.(error)).Fatalf("panic: %s", string(debug.Stack()))
+	}
+}
 
-	// Passthrough runcOptions
-	opts, err := unpackBundle(filepath.Join(bundleMountPath, "config.json"), req.Options)
+func unmarshalExtraData(marshalled *types.Any) (*proto.ExtraData, error) {
+	// get json bytes from task request
+	extraData := &proto.ExtraData{}
+	err := types.UnmarshalAny(marshalled, extraData)
 	if err != nil {
 		return nil, err
 	}
-	req.Options = opts
-	// Use mount path instead of bundle path inside the VM
-	req.Bundle = bundleMountPath
 
-	// Do not pass any mounts to runc, everything is already mounted for us
-	req.Rootfs = nil
-	// handle STDIO
-	ts.io, err = cio.NewFIFOSetInDir(defaultStdioPath, req.ID, req.Terminal)
+	return extraData, nil
+}
+
+// Create creates a new initial process and container using runc
+func (ts *TaskService) Create(ctx context.Context, req *taskAPI.CreateTaskRequest) (*taskAPI.CreateTaskResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+	logger := log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "bundle": req.Bundle})
+	logger.Info("create")
+
+	extraData, err := unmarshalExtraData(req.Options)
 	if err != nil {
-		log.G(ctx).WithError(err).Error("error proxying io")
 		return nil, err
 	}
-	req.Stdin = ts.io.Stdin
-	req.Stderr = ts.io.Stderr
-	req.Stdout = ts.io.Stdout
-	ioctx, cancel := context.WithCancel(ctx)
-	ts.cancels = append(ts.cancels, cancel)
-	ts.proxyStdio(ioctx, req.Stdin, req.Stdout, req.Stderr)
-	// before create call ensure we remove any existing .init.pid file
-	// We can ignore errors since it's valid for the file to not be present
-	os.Remove(".init.pid")
-	log.G(ctx).Debug("calling runc create")
-	resp, err := ts.runc.Create(ctx, req)
 
+	bundleDir := bundle.Dir(filepath.Join(containerRootDir, req.ID))
+	err = bundleDir.Create()
 	if err != nil {
-		log.G(ctx).WithError(err).Error("error creating container")
 		return nil, err
 	}
 
-	log.G(ctx).WithField("pid", resp.Pid).Debugf("create succeeded")
-	return resp, nil
-}
-
-func (ts *TaskService) proxyStdio(ctx context.Context, stdin, stdout, stderr string) {
-	go proxyIO(ctx, stdin, internal.StdinPort, true)
-	go proxyIO(ctx, stdout, internal.StdoutPort, false)
-	go proxyIO(ctx, stderr, internal.StderrPort, false)
-}
+	err = bundleDir.OCIConfig().Write(extraData.JsonSpec)
+	if err != nil {
+		return nil, err
+	}
 
-func proxyIO(ctx context.Context, path string, port uint32, in bool) {
-	if path == "" {
-		return
+	// TODO replace with proper drive mounting once that PR is merged. Right now, all containers in
+	// this VM start up with the same rootfs image no matter their configuration
+	err = bundleDir.MountRootfs("/dev/vdb", "ext4", nil)
+	if err != nil {
+		return nil, err
 	}
-	log.G(ctx).Debug("setting up IO for " + path)
-	f, err := fifo.OpenFifo(ctx, path, syscall.O_RDWR|syscall.O_NONBLOCK|syscall.O_CREAT, 0700)
+
+	// Create a runc shim to manage this task
+	// TODO if we update to the v2 runc implementation in containerd, we can use a single
+	// runc service instance to manage all tasks instead of creating a new one for each
+	runcService, err := runc.New(ctx, req.ID, ts.eventExchange)
 	if err != nil {
-		log.G(ctx).WithError(err).Error("error opening fifo")
-		return
+		return nil, err
 	}
-	listener, err := vsock.Listen(port)
+
+	// Override the incoming stdio FIFOs, which have paths from the host that we can't use
+	fifoSet, err := cio.NewFIFOSetInDir(bundleDir.RootPath(), req.ID, req.Terminal)
 	if err != nil {
-		log.G(ctx).WithError(err).Error("unable to listen on vsock")
-		f.Close()
-		return
+		log.G(ctx).WithError(err).Error("failed opening stdio FIFOs")
+		return nil, err
 	}
 
-	var conn net.Conn
-	for {
-		// accept is non-blocking so try to accept until we get
-		// a connection
-		// TODO: investigate if there is a way to distinguish
-		// transient errors from permanent ones.
-		conn, err = listener.Accept()
-		if err != nil {
-			continue
-		}
-		break
+	// Don't try to connect any io streams that weren't requested by the client
+	if req.Stdin == "" {
+		fifoSet.Stdin = ""
+	}
+
+	if req.Stdout == "" {
+		fifoSet.Stdout = ""
+	}
+
+	if req.Stderr == "" {
+		fifoSet.Stderr = ""
 	}
-	go func() {
-		<-ctx.Done()
-		conn.Close()
-		f.Close()
-	}()
-	log.G(ctx).Debug("begin copying io")
-	buf := make([]byte, internal.DefaultBufferSize)
-	if in {
-		_, err = io.CopyBuffer(f, conn, buf)
-	} else {
-		_, err = io.CopyBuffer(conn, f, buf)
+
+	task, err := ts.taskManager.AddTask(ctx, req.ID, runcService, bundleDir, extraData, fifoSet)
+	if err != nil {
+		return nil, err
 	}
+
+	logger.Debug("calling runc create")
+
+	// Override some of the incoming paths, which were set on the Host and thus not valid here in the Guest
+	req.Bundle = bundleDir.RootPath()
+	req.Rootfs = nil
+	req.Stdin = fifoSet.Stdin
+	req.Stdout = fifoSet.Stdout
+	req.Stderr = fifoSet.Stderr
+
+	// Just provide runc the options it knows about, not our wrapper
+	req.Options = task.ExtraData().GetRuncOptions()
+
+	// Start the io proxy and wait for initialization to complete before starting
+	// the task to ensure we capture all task output
+	err = <-task.StartStdioProxy(ctx, vm.VSockToFIFO, acceptVSock)
 	if err != nil {
-		log.G(ctx).WithError(err).Error("error with stdio")
+		return nil, err
 	}
-}
 
-func unpackBundle(path string, bundle *types.Any) (*types.Any, error) {
-	// get json bytes from task request
-	extraData := &proto.ExtraData{}
-	err := types.UnmarshalAny(bundle, extraData)
+	resp, err := task.Create(ctx, req)
 	if err != nil {
+		logger.WithError(err).Error("error creating container")
 		return nil, err
 	}
-	// write bundle/config.json bytes
-	err = ioutil.WriteFile(path, extraData.JsonSpec, 0644)
+
+	logger.WithField("pid", resp.Pid).Debugf("create succeeded")
+	return resp, nil
+}
+
+func isTemporaryNetErr(err error) bool {
+	terr, ok := err.(interface {
+		Temporary() bool
+	})
+
+	return err != nil && ok && terr.Temporary()
+}
+
+var _ vm.VSockConnector = acceptVSock
+
+func acceptVSock(ctx context.Context, port uint32) (net.Conn, error) {
+	listener, err := vsock.Listen(port)
 	if err != nil {
+		return nil, errors.Wrap(err, "unable to listen on vsock")
+	}
+
+	for range time.Tick(10 * time.Millisecond) {
+		// accept is non-blocking so try to accept until we get a connection
+		conn, err := listener.Accept()
+		if err == nil {
+			return conn, nil
+		}
+
+		if isTemporaryNetErr(err) {
+			log.G(ctx).WithError(err).Debug("temporary stdio vsock accept failure")
+			continue
+		}
+
+		log.G(ctx).WithError(err).Error("non-temporary stdio vsock accept failure")
 		return nil, err
 	}
-	return extraData.RuncOptions, nil
+
+	panic("unreachable code") // appeases the compiler, which doesn't know the for loop is infinite
 }
 
 // State returns process state information
-func (ts *TaskService) State(ctx context.Context, req *shimapi.StateRequest) (*shimapi.StateResponse, error) {
+func (ts *TaskService) State(ctx context.Context, req *taskAPI.StateRequest) (*taskAPI.StateResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("state")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.State(ctx, req)
+	resp, err := task.State(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("state failed")
 		return nil, err
@@ -189,11 +231,17 @@ func (ts *TaskService) State(ctx context.Context, req *shimapi.StateRequest) (*s
 }
 
 // Start starts a process
-func (ts *TaskService) Start(ctx context.Context, req *shimapi.StartRequest) (*shimapi.StartResponse, error) {
+func (ts *TaskService) Start(ctx context.Context, req *taskAPI.StartRequest) (*taskAPI.StartResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("start")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Start(ctx, req)
+	resp, err := task.Start(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("start failed")
 		return nil, err
@@ -204,11 +252,17 @@ func (ts *TaskService) Start(ctx context.Context, req *shimapi.StartRequest) (*s
 }
 
 // Delete deletes the initial process and container
-func (ts *TaskService) Delete(ctx context.Context, req *shimapi.DeleteRequest) (*shimapi.DeleteResponse, error) {
+func (ts *TaskService) Delete(ctx context.Context, req *taskAPI.DeleteRequest) (*taskAPI.DeleteResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("delete")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Delete(ctx, req)
+	resp, err := task.Delete(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("delete failed")
 		return nil, err
@@ -222,11 +276,17 @@ func (ts *TaskService) Delete(ctx context.Context, req *shimapi.DeleteRequest) (
 }
 
 // Pids returns all pids inside the container
-func (ts *TaskService) Pids(ctx context.Context, req *shimapi.PidsRequest) (*shimapi.PidsResponse, error) {
+func (ts *TaskService) Pids(ctx context.Context, req *taskAPI.PidsRequest) (*taskAPI.PidsResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("pids")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Pids(ctx, req)
+	resp, err := task.Pids(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("pids failed")
 		return nil, err
@@ -237,11 +297,17 @@ func (ts *TaskService) Pids(ctx context.Context, req *shimapi.PidsRequest) (*shi
 }
 
 // Pause pauses the container
-func (ts *TaskService) Pause(ctx context.Context, req *shimapi.PauseRequest) (*types.Empty, error) {
+func (ts *TaskService) Pause(ctx context.Context, req *taskAPI.PauseRequest) (*types.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("pause")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Pause(ctx, req)
+	resp, err := task.Pause(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("pause failed")
 		return nil, err
@@ -252,11 +318,17 @@ func (ts *TaskService) Pause(ctx context.Context, req *shimapi.PauseRequest) (*t
 }
 
 // Resume resumes the container
-func (ts *TaskService) Resume(ctx context.Context, req *shimapi.ResumeRequest) (*types.Empty, error) {
+func (ts *TaskService) Resume(ctx context.Context, req *taskAPI.ResumeRequest) (*types.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("resume")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Resume(ctx, req)
+	resp, err := task.Resume(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Debug("resume failed")
 		return nil, err
@@ -267,26 +339,38 @@ func (ts *TaskService) Resume(ctx context.Context, req *shimapi.ResumeRequest) (
 }
 
 // Checkpoint saves the state of the container instance
-func (ts *TaskService) Checkpoint(ctx context.Context, req *shimapi.CheckpointTaskRequest) (*types.Empty, error) {
+func (ts *TaskService) Checkpoint(ctx context.Context, req *taskAPI.CheckpointTaskRequest) (*types.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "path": req.Path}).Info("checkpoint")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Checkpoint(ctx, req)
+	resp, err := task.Checkpoint(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("checkout failed")
 		return nil, err
 	}
 
-	log.G(ctx).Debug("checkpoint succeeded")
+	log.G(ctx).Debug("checkpoint sutaskAPI")
 	return resp, nil
 }
 
 // Kill kills a process with the provided signal
-func (ts *TaskService) Kill(ctx context.Context, req *shimapi.KillRequest) (*types.Empty, error) {
+func (ts *TaskService) Kill(ctx context.Context, req *taskAPI.KillRequest) (*types.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("kill")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Kill(ctx, req)
+	resp, err := task.Kill(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("kill failed")
 		return nil, err
@@ -297,11 +381,17 @@ func (ts *TaskService) Kill(ctx context.Context, req *shimapi.KillRequest) (*typ
 }
 
 // Exec runs an additional process inside the container
-func (ts *TaskService) Exec(ctx context.Context, req *shimapi.ExecProcessRequest) (*types.Empty, error) {
+func (ts *TaskService) Exec(ctx context.Context, req *taskAPI.ExecProcessRequest) (*types.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("exec")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Exec(ctx, req)
+	resp, err := task.Exec(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("exec failed")
 		return nil, err
@@ -312,11 +402,17 @@ func (ts *TaskService) Exec(ctx context.Context, req *shimapi.ExecProcessRequest
 }
 
 // ResizePty resizes pty
-func (ts *TaskService) ResizePty(ctx context.Context, req *shimapi.ResizePtyRequest) (*types.Empty, error) {
+func (ts *TaskService) ResizePty(ctx context.Context, req *taskAPI.ResizePtyRequest) (*types.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("resize_pty")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.ResizePty(ctx, req)
+	resp, err := task.ResizePty(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("resize_pty failed")
 		return nil, err
@@ -327,11 +423,17 @@ func (ts *TaskService) ResizePty(ctx context.Context, req *shimapi.ResizePtyRequ
 }
 
 // CloseIO closes all IO inside container
-func (ts *TaskService) CloseIO(ctx context.Context, req *shimapi.CloseIORequest) (*types.Empty, error) {
+func (ts *TaskService) CloseIO(ctx context.Context, req *taskAPI.CloseIORequest) (*types.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("close_io")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.CloseIO(ctx, req)
+	resp, err := task.CloseIO(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("close io failed")
 		return nil, err
@@ -342,11 +444,17 @@ func (ts *TaskService) CloseIO(ctx context.Context, req *shimapi.CloseIORequest)
 }
 
 // Update updates running container
-func (ts *TaskService) Update(ctx context.Context, req *shimapi.UpdateTaskRequest) (*types.Empty, error) {
+func (ts *TaskService) Update(ctx context.Context, req *taskAPI.UpdateTaskRequest) (*types.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("update")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Update(ctx, req)
+	resp, err := task.Update(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("update failed")
 		return nil, err
@@ -357,11 +465,17 @@ func (ts *TaskService) Update(ctx context.Context, req *shimapi.UpdateTaskReques
 }
 
 // Wait waits for a process to exit
-func (ts *TaskService) Wait(ctx context.Context, req *shimapi.WaitRequest) (*shimapi.WaitResponse, error) {
+func (ts *TaskService) Wait(ctx context.Context, req *taskAPI.WaitRequest) (*taskAPI.WaitResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("wait")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Wait(ctx, req)
+	resp, err := task.Wait(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("wait failed")
 		return nil, err
@@ -372,11 +486,17 @@ func (ts *TaskService) Wait(ctx context.Context, req *shimapi.WaitRequest) (*shi
 }
 
 // Stats returns a process stats
-func (ts *TaskService) Stats(ctx context.Context, req *shimapi.StatsRequest) (*shimapi.StatsResponse, error) {
+func (ts *TaskService) Stats(ctx context.Context, req *taskAPI.StatsRequest) (*taskAPI.StatsResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("stats")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Stats(ctx, req)
+	resp, err := task.Stats(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("stats failed")
 		return nil, err
@@ -387,11 +507,17 @@ func (ts *TaskService) Stats(ctx context.Context, req *shimapi.StatsRequest) (*s
 }
 
 // Connect returns shim information such as the shim's pid
-func (ts *TaskService) Connect(ctx context.Context, req *shimapi.ConnectRequest) (*shimapi.ConnectResponse, error) {
+func (ts *TaskService) Connect(ctx context.Context, req *taskAPI.ConnectRequest) (*taskAPI.ConnectResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("connect")
+	task, err := ts.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
 
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
-	resp, err := ts.runc.Connect(ctx, req)
+	resp, err := task.Connect(ctx, req)
 	if err != nil {
 		log.G(ctx).WithError(err).Error("connect failed")
 		return nil, err
@@ -406,27 +532,17 @@ func (ts *TaskService) Connect(ctx context.Context, req *shimapi.ConnectRequest)
 }
 
 // Shutdown cleanups runc resources ans gracefully shutdowns ttrpc server
-func (ts *TaskService) Shutdown(ctx context.Context, req *shimapi.ShutdownRequest) (*types.Empty, error) {
+func (ts *TaskService) Shutdown(ctx context.Context, req *taskAPI.ShutdownRequest) (*types.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "now": req.Now}).Debug("shutdown")
 	ctx = namespaces.WithNamespace(ctx, defaultNamespace)
 
-	_, err := ts.runc.Cleanup(ctx)
-	if err != nil {
-		log.G(ctx).WithError(err).Warn("error cleaning up")
-	}
-
 	// We don't want to call runc.Shutdown here as it just os.Exits behind.
 	// calling all cancels here for graceful shutdown instead and call runc Shutdown at end.
-	ts.cancelAll()
+	ts.taskManager.RemoveAll(ctx)
+	ts.cancel()
 
 	log.G(ctx).Debug("going to gracefully shutdown agent")
 	return &types.Empty{}, nil
 }
-
-func (ts *TaskService) cancelAll() {
-	// cancel LIFO order
-	for i := len(ts.cancels) - 1; i >= 0; i-- {
-		log.G(context.Background()).Debug("Cancelling ", i)
-		ts.cancels[i]()
-	}
-}
diff --git a/go.mod b/go.mod
index 9325a8e..b1c06f5 100644
--- a/go.mod
+++ b/go.mod
@@ -39,12 +39,11 @@ require (
 	github.com/prometheus/client_golang v0.9.2 // indirect
 	github.com/shirou/gopsutil v2.18.12+incompatible
 	github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 // indirect
-	github.com/sirupsen/logrus v1.2.0
+	github.com/sirupsen/logrus v1.4.1
 	github.com/stretchr/testify v1.2.2
 	github.com/syndtr/gocapability v0.0.0-20180916011248-d98352740cb2 // indirect
 	github.com/urfave/cli v1.20.0 // indirect
 	go.etcd.io/bbolt v1.3.1-etcd.8
-	golang.org/x/crypto v0.0.0-20190325154230-a5d413f7728c // indirect
 	golang.org/x/net v0.0.0-20190328230028-74de082e2cca
 	golang.org/x/sync v0.0.0-20181108010431-42b317875d0f
 	golang.org/x/sys v0.0.0-20190329044733-9eb1bfa1ce65
diff --git a/go.sum b/go.sum
index 1642b1c..61c87c9 100644
--- a/go.sum
+++ b/go.sum
@@ -145,8 +145,8 @@ github.com/shirou/gopsutil v2.18.12+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMT
 github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 h1:udFKJ0aHUL60LboW/A+DfgoHVedieIzIXE8uylPue0U=
 github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4/go.mod h1:qsXQc7+bwAM3Q1u/4XEfrquwF8Lw7D7y5cD8CuHnfIc=
 github.com/sirupsen/logrus v1.1.1/go.mod h1:zrgwTnHtNr00buQ1vSptGe8m1f/BbgsPukg8qsT7A+A=
-github.com/sirupsen/logrus v1.2.0 h1:juTguoYk5qI21pwyTXY3B3Y5cOTH3ZUyZCg1v/mihuo=
-github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
+github.com/sirupsen/logrus v1.4.1 h1:GL2rEmy6nsikmW0r8opw9JIRScdMF5hA8cOYLH7In1k=
+github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
@@ -159,14 +159,13 @@ go.etcd.io/bbolt v1.3.1-etcd.8/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XK
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793 h1:u+LnwYTOOW7Ukr/fppxEb1Nwz0AtPflrblfvUudpo+I=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20190325154230-a5d413f7728c h1:Vj5n4GlwjmQteupaxJ9+0FNOmBrHfq7vN4btdGoDZgI=
-golang.org/x/crypto v0.0.0-20190325154230-a5d413f7728c/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181005035420-146acd28ed58/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181106065722-10aee1819953/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190327214358-63eda1eb0650/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190328230028-74de082e2cca h1:hyA6yiAgbUwuWqtscNvWAI7U1CtlaD1KilQ6iudt1aI=
 golang.org/x/net v0.0.0-20190328230028-74de082e2cca/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
diff --git a/internal/bundle/bundle.go b/internal/bundle/bundle.go
new file mode 100644
index 0000000..b770fc3
--- /dev/null
+++ b/internal/bundle/bundle.go
@@ -0,0 +1,143 @@
+// Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may
+// not use this file except in compliance with the License. A copy of the
+// License is located at
+//
+//	http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file is distributed
+// on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+// express or implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package bundle
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+
+	"github.com/containerd/containerd/mount"
+
+	"github.com/firecracker-microvm/firecracker-containerd/internal"
+	"github.com/firecracker-microvm/firecracker-containerd/runtime/firecrackeroci"
+	"github.com/pkg/errors"
+)
+
+// Dir represents the root of a container bundle dir. It's just a type wrapper
+// around a string, where the string is the path of the bundle dir.
+type Dir string
+
+// RootPath returns the top-level directory of the bundle dir
+func (d Dir) RootPath() string {
+	return string(d)
+}
+
+// Create will mkdir the bundle RootPath with correct permissions (or no-op if it
+// already exists)
+func (d Dir) Create() error {
+	return os.MkdirAll(d.RootPath(), 0700)
+}
+
+// AddrFilePath is the path to the address file as found in the bundleDir.
+// Even though the shim address is set per-VM, not per-container, containerd expects
+// to find the shim addr file in the bundle dir, so we still have to create it or
+// symlink it to the shimAddrFilePath
+func (d Dir) AddrFilePath() string {
+	return filepath.Join(d.RootPath(), internal.ShimAddrFileName)
+}
+
+// LogFifoPath is a path to a FIFO for writing shim logs as found in the bundleDir.
+// It is the path created by containerd for us, the shimLogFifoPath is just a symlink to one.
+func (d Dir) LogFifoPath() string {
+	return filepath.Join(d.RootPath(), internal.ShimLogFifoName)
+}
+
+// RootfsPath returns the path to the "rootfs" dir of the bundle
+func (d Dir) RootfsPath() string {
+	return filepath.Join(d.RootPath(), internal.BundleRootfsName)
+}
+
+// MountRootfs creates the RootfsPath and mounts the provided sourcePath there.
+func (d Dir) MountRootfs(sourcePath string, mountType string, mountOpts []string) error {
+	err := os.Mkdir(d.RootfsPath(), 0700)
+	if err != nil {
+		return err
+	}
+
+	return mount.All([]mount.Mount{{
+		Source:  sourcePath,
+		Type:    mountType,
+		Options: mountOpts,
+	}}, d.RootfsPath())
+}
+
+// OCIConfigPath returns the path to the bundle's config.json
+func (d Dir) OCIConfigPath() string {
+	return filepath.Join(d.RootPath(), internal.OCIConfigName)
+}
+
+// OCIConfig returns a OCIConfig object wrapper around the bundle
+// config.json
+func (d Dir) OCIConfig() *OCIConfig {
+	return &OCIConfig{path: d.OCIConfigPath()}
+}
+
+// OCIConfig is wrapper around a bundle's config.json that provided
+// basic file operations on it with appropriate permissions.
+type OCIConfig struct {
+	path string
+}
+
+// File opens the config.json as read-only
+func (c *OCIConfig) File() (*os.File, error) {
+	f, err := os.Open(c.path)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to open OCI config file %s", c.path)
+	}
+
+	return f, nil
+}
+
+// Bytes returns the bytes of config.json
+func (c *OCIConfig) Bytes() ([]byte, error) {
+	f, err := ioutil.ReadFile(c.path)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to read OCI config file %s", c.path)
+	}
+
+	return f, nil
+}
+
+// Write will create or overwrite the config.json with the provided bytes
+func (c *OCIConfig) Write(contents []byte) error {
+	err := ioutil.WriteFile(c.path, contents, 0700)
+	if err != nil {
+		return errors.Wrapf(err, "failed to write OCI config file %s", c.path)
+	}
+
+	return nil
+}
+
+// VMID returns the firecracker VM ID set by the client in the OCI config Annotations section, if any.
+func (c *OCIConfig) VMID() (string, error) {
+	ociConfigFile, err := c.File()
+	if err != nil {
+		return "", err
+	}
+
+	defer ociConfigFile.Close()
+	var ociConfig struct {
+		Annotations map[string]string `json:"annotations,omitempty"`
+	}
+
+	if err := json.NewDecoder(ociConfigFile).Decode(&ociConfig); err != nil {
+		return "", errors.Wrapf(err, "failed to parse Annotations section of OCI config file %s", c.path)
+	}
+
+	// This will return empty string if the key is not present in the OCI config, which the caller can decide
+	// how to deal with
+	return ociConfig.Annotations[firecrackeroci.VMIDAnnotationKey], nil
+}
diff --git a/internal/common.go b/internal/common.go
index 7b3d965..be68769 100644
--- a/internal/common.go
+++ b/internal/common.go
@@ -22,4 +22,28 @@ const (
 	StderrPort = 11002
 	// DefaultBufferSize represents buffer size in bytes to used for IO between runtime and agent
 	DefaultBufferSize = 1024
+
+	// FirecrackerSockName is the name of the Firecracker VMM API socket
+	FirecrackerSockName = "firecracker.sock"
+	// FirecrackerLogFifoName is the name of the Firecracker VMM log FIFO
+	FirecrackerLogFifoName = "fc-logs.fifo"
+	// FirecrackerMetricsFifoName is the name of the Firecracker VMM metrics FIFO
+	FirecrackerMetricsFifoName = "fc-metrics.fifo"
+
+	// TODO these strings are hardcoded throughout the containerd codebase, it may
+	// be worth sending them a PR to define them as constants accessible to shim
+	// implementations like our own
+
+	// ShimAddrFileName is the name of the file in which a shim's API address can be found, used by
+	// containerd to reconnect after it restarts
+	ShimAddrFileName = "address"
+
+	// ShimLogFifoName is the name of the FIFO created by containerd for a shim to write its logs to
+	ShimLogFifoName = "log"
+
+	// OCIConfigName is the name of the OCI bundle's config field
+	OCIConfigName = "config.json"
+
+	// BundleRootfsName is the name of the bundle's directory for holding the container's rootfs
+	BundleRootfsName = "rootfs"
 )
diff --git a/internal/vm/Makefile b/internal/vm/Makefile
new file mode 100644
index 0000000..22a7f0f
--- /dev/null
+++ b/internal/vm/Makefile
@@ -0,0 +1,37 @@
+# Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You may
+# not use this file except in compliance with the License. A copy of the
+# License is located at
+#
+# 	http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is distributed
+# on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+# express or implied. See the License for the specific language governing
+# permissions and limitations under the License.
+
+EXTRAGOARGS:=
+
+GOMOD := $(shell go env GOMOD)
+GOSUM := $(GOMOD:.mod=.sum)
+SOURCES:=$(shell find . -name '*.go')
+DOCKER_IMAGE_TAG?=latest
+
+all:
+
+test: $(SOURCES) $(GOMOD) $(GOSUM)
+	go test ./... $(EXTRAGOARGS)
+
+test-docker-unit: $(SOURCES) $(GOMOD) $(GOSUM)
+	docker run --rm \
+		--env EXTRAGOARGS="${EXTRAGOARGS}" \
+		--workdir="/firecracker-containerd/internal/vm" \
+		localhost/firecracker-containerd-unittest-nonroot:${DOCKER_IMAGE_TAG} \
+		"make test"
+
+clean:
+
+install:
+
+.PHONY: clean all install test
diff --git a/internal/vm/dir.go b/internal/vm/dir.go
new file mode 100644
index 0000000..decf96a
--- /dev/null
+++ b/internal/vm/dir.go
@@ -0,0 +1,106 @@
+// Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may
+// not use this file except in compliance with the License. A copy of the
+// License is located at
+//
+//	http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file is distributed
+// on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+// express or implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package vm
+
+import (
+	"os"
+	"path/filepath"
+
+	"github.com/containerd/containerd/runtime/v2/shim"
+	"github.com/firecracker-microvm/firecracker-containerd/internal"
+	"github.com/firecracker-microvm/firecracker-containerd/internal/bundle"
+	"github.com/pkg/errors"
+)
+
+// Dir represents the root of a firecracker-containerd VM directory, which
+// holds various files, sockets and FIFOs used during VM runtime.
+type Dir string
+
+// RootPath returns the top-level directory of the VM dir
+func (d Dir) RootPath() string {
+	return string(d)
+}
+
+// Create will mkdir the RootPath with correct permissions, or no-op if it
+// already exists
+func (d Dir) Create() error {
+	return os.MkdirAll(d.RootPath(), 0700)
+}
+
+// AddrFilePath returns the path to the shim address file as found in the vmDir. The
+// address file holds the abstract unix socket path at which the shim serves its API.
+func (d Dir) AddrFilePath() string {
+	return filepath.Join(d.RootPath(), internal.ShimAddrFileName)
+}
+
+// LogFifoPath returns the path to the FIFO for writing shim logs
+func (d Dir) LogFifoPath() string {
+	return filepath.Join(d.RootPath(), internal.ShimLogFifoName)
+}
+
+// FirecrackerSockPath returns the path to the unix socket at which the firecracker VMM
+// services its API
+func (d Dir) FirecrackerSockPath() string {
+	return filepath.Join(d.RootPath(), internal.FirecrackerSockName)
+}
+
+// FirecrackerLogFifoPath returns the path to the FIFO at which the firecracker VMM writes
+// its logs
+func (d Dir) FirecrackerLogFifoPath() string {
+	return filepath.Join(d.RootPath(), internal.FirecrackerLogFifoName)
+}
+
+// FirecrackerMetricsFifoPath returns the path to the FIFO at which the firecracker VMM writes
+// metrics
+func (d Dir) FirecrackerMetricsFifoPath() string {
+	return filepath.Join(d.RootPath(), internal.FirecrackerMetricsFifoName)
+}
+
+// BundleLink returns the path to the symlink to the bundle dir for a given container running
+// inside the VM of this vm dir.
+func (d Dir) BundleLink(containerID string) bundle.Dir {
+	return bundle.Dir(filepath.Join(d.RootPath(), containerID))
+}
+
+// CreateBundleLink creates the BundleLink by symlinking to the provided bundle dir
+func (d Dir) CreateBundleLink(containerID string, bundleDir bundle.Dir) error {
+	return createSymlink(bundleDir.RootPath(), d.BundleLink(containerID).RootPath(), "bundle")
+}
+
+// CreateAddressLink creates a symlink from the VM dir to the bundle dir for the shim address file.
+// This symlink is read by containerd.
+// CreateAddressLink assumes that CreateBundleLink has been called.
+func (d Dir) CreateAddressLink(containerID string) error {
+	return createSymlink(d.AddrFilePath(), d.BundleLink(containerID).AddrFilePath(), "shim address file")
+}
+
+// CreateShimLogFifoLink creates a symlink from the bundleDir of the provided container.
+// CreateAddressLink assumes that CreateBundleLink has been called.
+func (d Dir) CreateShimLogFifoLink(containerID string) error {
+	return createSymlink(d.BundleLink(containerID).LogFifoPath(), d.LogFifoPath(), "shim log fifo")
+}
+
+// WriteAddress will write the actual address file in the VM dir.
+func (d Dir) WriteAddress(shimSocketAddress string) error {
+	return shim.WriteAddress(d.AddrFilePath(), shimSocketAddress)
+}
+
+func createSymlink(oldPath, newPath string, errMsgName string) error {
+	err := os.Symlink(oldPath, newPath)
+	if err != nil {
+		return errors.Wrapf(err, `failed to create %s symlink from "%s"->"%s"`, errMsgName, newPath, oldPath)
+	}
+
+	return nil
+}
diff --git a/internal/vm/task.go b/internal/vm/task.go
new file mode 100644
index 0000000..d203ebc
--- /dev/null
+++ b/internal/vm/task.go
@@ -0,0 +1,339 @@
+// Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may
+// not use this file except in compliance with the License. A copy of the
+// License is located at
+//
+//	http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file is distributed
+// on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+// express or implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package vm
+
+import (
+	"context"
+	"io"
+	"net"
+	"sync"
+	"syscall"
+
+	"github.com/containerd/containerd/cio"
+	"github.com/containerd/containerd/log"
+	taskAPI "github.com/containerd/containerd/runtime/v2/task"
+	"github.com/containerd/fifo"
+	ptypes "github.com/gogo/protobuf/types"
+	"github.com/hashicorp/go-multierror"
+	"github.com/pkg/errors"
+
+	"github.com/firecracker-microvm/firecracker-containerd/internal"
+	"github.com/firecracker-microvm/firecracker-containerd/internal/bundle"
+	proto "github.com/firecracker-microvm/firecracker-containerd/proto/grpc"
+)
+
+// IODirection represents the direction in which container related I/O should
+// flow between a FIFO and a vsock connection (the possible directions being
+// from FIFO to vsock or from vsock to FIFO). This is needed to parameterize
+// functions that operate on either Host or Guest (in which case I/O directions
+// for the same stream are inverses) and functions that operate on stdin vs.
+// stdout/stderr (which again have inverse directions).
+type IODirection bool
+
+const (
+	// FIFOtoVSock represents I/O that should be copied from a FIFO to a VSock.
+	FIFOtoVSock IODirection = true
+
+	// VSockToFIFO represents I/O that should be copied from a VSock to a FIFO.
+	VSockToFIFO IODirection = false
+)
+
+func (d IODirection) opposite() IODirection {
+	return IODirection(!bool(d))
+}
+
+// VSockConnector is a function used as a callback for obtaining a vsock connection.
+// Implementations may include one that dials a remote vsock listener and one that
+// listens and accepts incoming connections.
+type VSockConnector func(ctx context.Context, port uint32) (net.Conn, error)
+
+// TaskManager manages a mapping of containerIDs to metadata for containerd tasks
+// being executed via a firecracker-containerd runtime. It's intended to be
+// abstracted over whether it's being executed on the Host or inside a VM Guest.
+type TaskManager interface {
+	AddTask(context.Context, string, taskAPI.TaskService, bundle.Dir, *proto.ExtraData, *cio.FIFOSet) (*Task, error)
+	Task(string) (*Task, error)
+	TaskCount() uint
+	Remove(context.Context, string)
+	RemoveAll(context.Context)
+}
+
+// NewTaskManager initializes a new TaskManager
+func NewTaskManager() TaskManager {
+	return &taskManager{
+		tasks: make(map[string]*Task),
+	}
+}
+
+type taskManager struct {
+	mu    sync.RWMutex
+	tasks map[string]*Task
+}
+
+// AddTask registers a task with the provided metadata with the taskManager.
+// taskService should implement the TaskService API for the task (i.e. Create, Kill, Exec, etc.).
+func (m *taskManager) AddTask(
+	ctx context.Context,
+	containerID string,
+	taskService taskAPI.TaskService,
+	bundleDir bundle.Dir,
+	extraData *proto.ExtraData,
+	fifoSet *cio.FIFOSet,
+) (*Task, error) {
+	err := bundleDir.Create() // no-op if it already exists
+	if err != nil {
+		return nil, err
+	}
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if _, ok := m.tasks[containerID]; ok {
+		return nil, errors.Errorf("cannot add duplicate task with container ID %q", containerID)
+	}
+
+	task := &Task{
+		TaskService: taskService,
+		ID:          containerID,
+
+		extraData: extraData,
+		bundleDir: bundleDir,
+		fifoSet:   fifoSet,
+
+		cancelCh: make(chan struct{}),
+	}
+
+	m.tasks[task.ID] = task
+	return task, nil
+}
+
+func (m *taskManager) get(containerID string) (*Task, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	task, ok := m.tasks[containerID]
+	if !ok {
+		return nil, errors.Errorf("task with containerID %q not found ", containerID)
+	}
+
+	return task, nil
+}
+
+// assumes caller is holding m.mu lock
+func (m *taskManager) remove(ctx context.Context, containerID string, task *Task) {
+	delete(m.tasks, containerID)
+	err := task.CancelIO()
+	if err != nil {
+		log.G(ctx).WithField("id", containerID).Error("failed to cancel task io")
+	}
+}
+
+// Task returns the TaskService API for the task with the provided containerID, allowing
+// the caller to perform APIs like Create, Kill, Exec, etc. on the task.
+func (m *taskManager) Task(containerID string) (*Task, error) {
+	return m.get(containerID)
+}
+
+// TaskCount returns the number of tasks registered with the taskManager.
+func (m *taskManager) TaskCount() uint {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	return uint(len(m.tasks))
+}
+
+// Remove unregisters the task with the provided containerID with the taskManager and cancels
+// any ongoing I/O proxying for the task.
+func (m *taskManager) Remove(ctx context.Context, containerID string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	task, ok := m.tasks[containerID]
+	if !ok {
+		// the task was already removed or just never existed, no-op
+		return
+	}
+
+	m.remove(ctx, containerID, task)
+}
+
+// RemoveAll calls Remove() on all tasks registered with the taskManager.
+func (m *taskManager) RemoveAll(ctx context.Context) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	for containerID, task := range m.tasks {
+		m.remove(ctx, containerID, task)
+	}
+}
+
+// Task represents a containerd Task under the control of a taskManager
+type Task struct {
+	taskAPI.TaskService
+	ID string
+
+	extraData *proto.ExtraData
+	bundleDir bundle.Dir
+	fifoSet   *cio.FIFOSet
+
+	cancelCh chan struct{}
+}
+
+// ExtraData returns the extra data the VM Agent needs to execute this task
+func (task *Task) ExtraData() *proto.ExtraData {
+	return task.extraData
+}
+
+// MarshalExtraData serializes the task in preparation for transport over vsock
+func (task *Task) MarshalExtraData() (*ptypes.Any, error) {
+	return ptypes.MarshalAny(task.extraData)
+}
+
+// BundleDir returns the bundle dir for the task
+func (task *Task) BundleDir() bundle.Dir {
+	return task.bundleDir
+}
+
+// CancelIO will stop any ongoing IO proxying. It returns an error to be compatible
+// with a cio interface but at this time should only return nil in practice.
+func (task *Task) CancelIO() error {
+	close(task.cancelCh)
+	return nil
+}
+
+// StartStdioProxy sets up stdin, stdout and stderr streams for the task. It returns a channel that will be closed once
+// *initialization* of the streams has completed (or, if there was an error, the error will be published and then the chan
+// will be closed). Initialization consists of opening the FIFOs and establishing the vsock connections. After initialization,
+// data will be copied asynchronously until an error is encountered or task.Cancel() is called.
+//
+// Doing the setup asynchronously allows the caller to perform other actions that may be necessary for the initialization to
+// complete (such as sending a request over vsock to the other side of the Host/Guest barrier).
+//
+// inputDirection is the IODirection that should be used for *stdin*. Stdout/stderr will, by definition, go in the opposite direction.
+// vsockConnector is a callback for establishing a vsock connection.
+func (task *Task) StartStdioProxy(ctx context.Context, inputDirection IODirection, vsockConnector VSockConnector) <-chan error {
+	logger := log.G(ctx).WithField("id", task.ID)
+	returnCh := make(chan error)
+
+	var wg sync.WaitGroup
+	errCh := make(chan error, 3)
+
+	// For each of stdin, stdout and stderr, if a FIFO has been set (i.e. the client requested that stream to be setup), go
+	// open the FIFO and connect the vsock.
+	if task.fifoSet.Stdin != "" {
+		wg.Add(1)
+		go func() {
+			errCh <- task.proxyIO(ctx, task.fifoSet.Stdin, task.extraData.StdinPort, inputDirection, vsockConnector)
+			wg.Done()
+		}()
+	} else {
+		logger.Info("skipping proxy io for unset stdin")
+	}
+
+	if task.fifoSet.Stdout != "" {
+		wg.Add(1)
+		go func() {
+			errCh <- task.proxyIO(ctx, task.fifoSet.Stdout, task.extraData.StdoutPort, inputDirection.opposite(), vsockConnector)
+			wg.Done()
+		}()
+	} else {
+		logger.Info("skipping proxy io for unset stdout")
+	}
+
+	if task.fifoSet.Stderr != "" {
+		wg.Add(1)
+		go func() {
+			errCh <- task.proxyIO(ctx, task.fifoSet.Stderr, task.extraData.StderrPort, inputDirection.opposite(), vsockConnector)
+			wg.Done()
+		}()
+	} else {
+		logger.Info("skipping proxy io for unset stderr")
+	}
+
+	// Once the streams have been initialized, gather any errors they returned and let the caller know by publishing
+	// on the return chan
+	go func() {
+		wg.Wait()
+		close(errCh)
+
+		var finalError *multierror.Error
+		for err := range errCh {
+			finalError = multierror.Append(finalError, err)
+		}
+
+		returnCh <- finalError.ErrorOrNil()
+		close(returnCh)
+	}()
+
+	return returnCh
+}
+
+func (task *Task) proxyIO(ctx context.Context, fifoPath string, port uint32, ioDirection IODirection, vsockConnector VSockConnector) error {
+	logger := log.G(ctx).WithField("path", fifoPath).WithField("port", port)
+
+	fifoFile, err := fifo.OpenFifo(ctx, fifoPath, syscall.O_RDWR|syscall.O_CREAT|syscall.O_NONBLOCK, 0700)
+	if err != nil {
+		logger.WithError(err).Error("failed to open fifo")
+		return err
+	}
+
+	sock, err := vsockConnector(ctx, port)
+	if err != nil {
+		fifoFile.Close()
+		logger.WithError(err).Error("failed to establish vsock connection")
+		return err
+	}
+
+	// the FIFO and vsock have been opened, so go off proxying between them asynchronously
+	// until Cancel is called.
+	go func() {
+		buf := make([]byte, internal.DefaultBufferSize)
+
+		logger.Debug("begin copying io")
+
+		switch ioDirection {
+		case FIFOtoVSock:
+			go func() {
+				<-task.cancelCh
+				// If we are copying from FIFO to vsock, be sure to first close just the FIFO,
+				// allowing any buffered data to be flushed into the vsock first.
+				fifoFile.Close()
+			}()
+
+			_, err = io.CopyBuffer(sock, fifoFile, buf)
+			sock.Close()
+
+		case VSockToFIFO:
+			go func() {
+				<-task.cancelCh
+				// If we are copying from vsock to FIFO, be sure to first close just the vsock,
+				// allowing any buffered data to be flushed into the FIFO first.
+				sock.Close()
+			}()
+
+			_, err = io.CopyBuffer(fifoFile, sock, buf)
+			fifoFile.Close()
+
+		default:
+			logger.Fatalf("undefined io direction") // should not be possible, direction is a bool
+		}
+
+		// All FIFO+vsocks should be closed by this point in the previous switch cases.
+		if err != nil {
+			logger.WithError(err).Error("error with stdio")
+		}
+	}()
+
+	return nil
+}
diff --git a/proto/grpc/types.pb.go b/proto/grpc/types.pb.go
index 6a4f4e8..a4266df 100644
--- a/proto/grpc/types.pb.go
+++ b/proto/grpc/types.pb.go
@@ -25,6 +25,9 @@ const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package
 type ExtraData struct {
 	JsonSpec             []byte     `protobuf:"bytes,1,opt,name=JsonSpec,json=jsonSpec,proto3" json:"JsonSpec,omitempty"`
 	RuncOptions          *types.Any `protobuf:"bytes,2,opt,name=RuncOptions,json=runcOptions,proto3" json:"RuncOptions,omitempty"`
+	StdinPort            uint32     `protobuf:"varint,3,opt,name=StdinPort,json=stdinPort,proto3" json:"StdinPort,omitempty"`
+	StdoutPort           uint32     `protobuf:"varint,4,opt,name=StdoutPort,json=stdoutPort,proto3" json:"StdoutPort,omitempty"`
+	StderrPort           uint32     `protobuf:"varint,5,opt,name=StderrPort,json=stderrPort,proto3" json:"StderrPort,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}   `json:"-"`
 	XXX_unrecognized     []byte     `json:"-"`
 	XXX_sizecache        int32      `json:"-"`
@@ -68,6 +71,27 @@ func (m *ExtraData) GetRuncOptions() *types.Any {
 	return nil
 }
 
+func (m *ExtraData) GetStdinPort() uint32 {
+	if m != nil {
+		return m.StdinPort
+	}
+	return 0
+}
+
+func (m *ExtraData) GetStdoutPort() uint32 {
+	if m != nil {
+		return m.StdoutPort
+	}
+	return 0
+}
+
+func (m *ExtraData) GetStderrPort() uint32 {
+	if m != nil {
+		return m.StderrPort
+	}
+	return 0
+}
+
 // Message to specify the various Firecracker VM configurations for a task
 type FirecrackerConfig struct {
 	RuncOptions       *types.Any                       `protobuf:"bytes,1,opt,name=RuncOptions,json=runcOptions,proto3" json:"RuncOptions,omitempty"`
@@ -482,48 +506,50 @@ func init() {
 func init() { proto.RegisterFile("types.proto", fileDescriptor_d938547f84707355) }
 
 var fileDescriptor_d938547f84707355 = []byte{
-	// 679 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x54, 0xdf, 0x6e, 0xd3, 0x3e,
-	0x14, 0x56, 0x7e, 0xf9, 0x6d, 0x6b, 0x9c, 0x76, 0x5b, 0x2d, 0x04, 0x65, 0x4c, 0x50, 0xe5, 0xaa,
-	0xdc, 0x74, 0xd2, 0x10, 0x48, 0x5c, 0x4c, 0xd0, 0x3f, 0x43, 0xeb, 0xa0, 0xeb, 0xe4, 0x0d, 0x2e,
-	0xb8, 0x41, 0x6e, 0x72, 0xda, 0x9a, 0x26, 0x76, 0xe4, 0x38, 0xdd, 0xba, 0xd7, 0xe0, 0x01, 0x78,
-	0x22, 0x1e, 0x09, 0x09, 0xd9, 0x4d, 0xba, 0xb4, 0x1a, 0x03, 0xae, 0x22, 0x7f, 0xe7, 0x3b, 0x9f,
-	0xcf, 0xf9, 0xce, 0x89, 0x91, 0xab, 0xe6, 0x31, 0x24, 0xcd, 0x58, 0x0a, 0x25, 0xf6, 0x1e, 0x8f,
-	0x85, 0x18, 0x87, 0x70, 0x60, 0x4e, 0xc3, 0x74, 0x74, 0x40, 0xf9, 0x7c, 0x11, 0xf2, 0xbe, 0x20,
-	0xe7, 0xf8, 0x5a, 0x49, 0xda, 0xa5, 0x8a, 0xe2, 0x3d, 0x54, 0x3a, 0x4d, 0x04, 0xbf, 0x88, 0xc1,
-	0xaf, 0x59, 0x75, 0xab, 0x51, 0x26, 0xa5, 0xaf, 0xd9, 0x19, 0xbf, 0x42, 0x2e, 0x49, 0xb9, 0x3f,
-	0x88, 0x15, 0x13, 0x3c, 0xa9, 0xfd, 0x57, 0xb7, 0x1a, 0xee, 0xe1, 0x83, 0xe6, 0x42, 0xb9, 0x99,
-	0x2b, 0x37, 0x5b, 0x7c, 0x4e, 0x5c, 0x79, 0x4b, 0xf4, 0xbe, 0xd9, 0xa8, 0xfa, 0x8e, 0x49, 0xf0,
-	0x25, 0xf5, 0xa7, 0x20, 0x3b, 0x82, 0x8f, 0xd8, 0x78, 0x5d, 0xcd, 0xfa, 0x4b, 0x35, 0x7c, 0x8a,
-	0xaa, 0x67, 0xa0, 0xae, 0x84, 0x9c, 0xf6, 0xb8, 0x02, 0x39, 0xa2, 0x3e, 0xe8, 0x5a, 0xec, 0x86,
-	0x7b, 0xb8, 0xdf, 0x2c, 0x5c, 0xb3, 0x4e, 0x22, 0x55, 0xbe, 0x9e, 0x86, 0x1b, 0x68, 0xe7, 0x3d,
-	0x48, 0x0e, 0x61, 0x2f, 0xa2, 0x63, 0x38, 0xa7, 0x6a, 0x52, 0xb3, 0xeb, 0x56, 0xc3, 0x21, 0x3b,
-	0xd3, 0x55, 0x18, 0x3f, 0x45, 0x68, 0xc1, 0x6c, 0xc9, 0x71, 0x52, 0xfb, 0xdf, 0x90, 0xd0, 0x74,
-	0x89, 0xe0, 0xb7, 0x08, 0xf5, 0xa9, 0x3f, 0x61, 0x1c, 0x3a, 0xa3, 0x71, 0x6d, 0xc3, 0x34, 0x53,
-	0x2f, 0x96, 0x93, 0x47, 0x4d, 0xf3, 0xa9, 0xa4, 0xba, 0x19, 0x82, 0xa2, 0x65, 0x0e, 0x3e, 0x40,
-	0x0e, 0x11, 0x42, 0x75, 0x25, 0x9b, 0x41, 0x6d, 0xd3, 0x08, 0x54, 0x8b, 0x02, 0x26, 0x40, 0x1c,
-	0x99, 0x73, 0xf0, 0x11, 0xda, 0x6d, 0x05, 0x01, 0xd3, 0x42, 0x34, 0x34, 0x50, 0x52, 0xdb, 0x32,
-	0x3e, 0xdc, 0x91, 0xb7, 0x4b, 0xd7, 0xa8, 0xde, 0x4f, 0x0b, 0x3d, 0xb9, 0xc7, 0x2e, 0xdd, 0x71,
-	0x9f, 0xfa, 0xad, 0x20, 0x90, 0x90, 0x2c, 0xc6, 0xe3, 0x98, 0x7a, 0x33, 0x04, 0xd7, 0x91, 0x7b,
-	0x22, 0x12, 0xd5, 0x85, 0xd9, 0x19, 0x8d, 0xc0, 0x6c, 0x83, 0x43, 0xdc, 0xc9, 0x2d, 0x84, 0xf7,
-	0x91, 0xd3, 0x0a, 0x43, 0x71, 0xd5, 0xef, 0x77, 0x2f, 0x8c, 0xaf, 0x25, 0xe2, 0xd0, 0x1c, 0xc0,
-	0x47, 0xa8, 0xd2, 0xe3, 0x84, 0x2a, 0xf8, 0xc0, 0x22, 0xa6, 0x40, 0x1a, 0x53, 0xdd, 0xc3, 0x47,
-	0xc5, 0xda, 0x0b, 0x61, 0x52, 0x61, 0x45, 0x36, 0x7e, 0x83, 0xb6, 0x07, 0xa9, 0x2a, 0xe6, 0x6f,
-	0xdc, 0x9f, 0xbf, 0x2d, 0x56, 0xe8, 0xde, 0x77, 0x0b, 0x3d, 0xfb, 0xc3, 0x7c, 0x74, 0x8f, 0x9d,
-	0xf3, 0x8f, 0x97, 0x10, 0xc5, 0x21, 0x55, 0x90, 0x99, 0xe0, 0xfa, 0xb7, 0x90, 0xee, 0xf1, 0x44,
-	0x1d, 0x73, 0x3a, 0x0c, 0x21, 0x30, 0x1e, 0x94, 0x88, 0x33, 0xc9, 0x01, 0xe3, 0x21, 0x44, 0x17,
-	0xec, 0x06, 0xfa, 0x6c, 0x68, 0x2c, 0xa8, 0x10, 0x14, 0x2d, 0x11, 0x9d, 0xfd, 0xc9, 0x8f, 0xd3,
-	0x8e, 0x48, 0xb9, 0x32, 0xfd, 0x57, 0x88, 0x33, 0xcb, 0x01, 0xef, 0x87, 0x85, 0x76, 0xd7, 0x07,
-	0xa9, 0x25, 0x7b, 0x09, 0x01, 0x1a, 0x0c, 0x78, 0x38, 0x37, 0x15, 0x95, 0x08, 0x62, 0x4b, 0x04,
-	0x7b, 0xa8, 0xdc, 0x4b, 0xcc, 0x22, 0xc1, 0x8c, 0xf9, 0x90, 0xd5, 0x54, 0x66, 0x05, 0x4c, 0xff,
-	0xe4, 0xe7, 0x54, 0xaa, 0x34, 0x65, 0x41, 0xb6, 0xef, 0xa5, 0x38, 0x3b, 0x6b, 0x7d, 0xbd, 0xf0,
-	0x03, 0xae, 0x87, 0x9b, 0x2f, 0x7a, 0xbc, 0x44, 0xf0, 0x6b, 0xe4, 0xfe, 0x83, 0xe9, 0xae, 0x2c,
-	0x38, 0x7e, 0x83, 0x1e, 0xde, 0x4d, 0xc3, 0x2f, 0x91, 0xd3, 0xa6, 0x3c, 0xb8, 0x62, 0x81, 0x9a,
-	0x64, 0x2f, 0xc1, 0x8a, 0xe4, 0xa5, 0x98, 0x02, 0x6f, 0xa7, 0xfe, 0x14, 0x14, 0x71, 0x86, 0x39,
-	0x13, 0x3f, 0x47, 0xf6, 0x20, 0xce, 0x1f, 0xa2, 0xdf, 0x26, 0xd8, 0x22, 0x4e, 0xbc, 0xeb, 0x95,
-	0xbb, 0x0b, 0x61, 0x6d, 0xd8, 0x80, 0xc3, 0x25, 0x8b, 0xa0, 0x9d, 0xca, 0x44, 0x99, 0xeb, 0x6d,
-	0x52, 0x16, 0x05, 0x4c, 0x9b, 0x42, 0x60, 0xc4, 0xc2, 0x50, 0x43, 0xe6, 0x3e, 0x9b, 0x20, 0xb9,
-	0x44, 0xb4, 0xa1, 0x1d, 0x1a, 0x53, 0x9f, 0xa9, 0xb9, 0x31, 0xd4, 0x26, 0x25, 0x3f, 0x3b, 0xb7,
-	0xb7, 0x3e, 0x6f, 0x2c, 0x1e, 0xb3, 0x4d, 0xf3, 0x79, 0xf1, 0x2b, 0x00, 0x00, 0xff, 0xff, 0x73,
-	0x09, 0x59, 0xef, 0x98, 0x05, 0x00, 0x00,
+	// 718 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x94, 0xcd, 0x6e, 0xe3, 0x36,
+	0x10, 0xc7, 0xa1, 0xaa, 0x49, 0xac, 0x91, 0xf3, 0x61, 0xa1, 0x68, 0xdd, 0x34, 0x68, 0x0d, 0x9f,
+	0xdc, 0x8b, 0x03, 0xb8, 0x68, 0x81, 0x1e, 0x82, 0xd6, 0x1f, 0x29, 0xe2, 0xb4, 0x8e, 0x0d, 0x3a,
+	0xed, 0xa1, 0x37, 0x5a, 0x1a, 0xdb, 0xac, 0x25, 0x52, 0xa0, 0x28, 0x27, 0xce, 0x6b, 0xf4, 0x01,
+	0xfa, 0x24, 0x7d, 0x84, 0x7d, 0xa4, 0x05, 0x16, 0xa4, 0x25, 0x45, 0x36, 0xb2, 0xd9, 0xdd, 0x93,
+	0xc0, 0xdf, 0x7f, 0x66, 0x38, 0xf3, 0x27, 0x45, 0x70, 0xd5, 0x26, 0xc6, 0xa4, 0x1d, 0x4b, 0xa1,
+	0xc4, 0xf9, 0xd7, 0x0b, 0x21, 0x16, 0x21, 0x5e, 0x9a, 0xd5, 0x2c, 0x9d, 0x5f, 0x52, 0xbe, 0xd9,
+	0x4a, 0xcd, 0xff, 0x2d, 0x70, 0xae, 0x1f, 0x95, 0xa4, 0x03, 0xaa, 0xa8, 0x77, 0x0e, 0x95, 0xdb,
+	0x44, 0xf0, 0x69, 0x8c, 0x7e, 0xdd, 0x6a, 0x58, 0xad, 0x2a, 0xa9, 0xfc, 0x93, 0xad, 0xbd, 0x9f,
+	0xc0, 0x25, 0x29, 0xf7, 0xc7, 0xb1, 0x62, 0x82, 0x27, 0xf5, 0xcf, 0x1a, 0x56, 0xcb, 0xed, 0x7c,
+	0xd1, 0xde, 0x96, 0x6e, 0xe7, 0xa5, 0xdb, 0x5d, 0xbe, 0x21, 0xae, 0x7c, 0x0e, 0xf4, 0x2e, 0xc0,
+	0x99, 0xaa, 0x80, 0xf1, 0x89, 0x90, 0xaa, 0x6e, 0x37, 0xac, 0xd6, 0x31, 0x71, 0x92, 0x1c, 0x78,
+	0xdf, 0x02, 0x4c, 0x55, 0x20, 0x52, 0x65, 0xe4, 0xcf, 0x8d, 0x0c, 0x49, 0x41, 0x32, 0x1d, 0xa5,
+	0x34, 0xfa, 0x41, 0xa1, 0x67, 0xa4, 0xf9, 0xaf, 0x0d, 0xb5, 0xdf, 0x98, 0x44, 0x5f, 0x52, 0x7f,
+	0x85, 0xb2, 0x2f, 0xf8, 0x9c, 0x2d, 0xf6, 0x7b, 0xb5, 0x3e, 0xb6, 0xd7, 0x5b, 0xa8, 0xdd, 0xa1,
+	0x7a, 0x10, 0x72, 0x35, 0xe4, 0x0a, 0xe5, 0x9c, 0xfa, 0xa8, 0x27, 0xb5, 0x5b, 0x6e, 0xe7, 0xa2,
+	0x5d, 0xda, 0x66, 0x3f, 0x88, 0xd4, 0xf8, 0x7e, 0x9a, 0xd7, 0x82, 0xd3, 0xdf, 0x51, 0x72, 0x0c,
+	0x87, 0x11, 0x5d, 0xe0, 0x84, 0xaa, 0xa5, 0x99, 0xde, 0x21, 0xa7, 0xab, 0x5d, 0xac, 0x67, 0xdc,
+	0x46, 0x76, 0xe5, 0x22, 0x31, 0x1e, 0x38, 0x04, 0x56, 0x05, 0xf1, 0x7e, 0x05, 0x18, 0x51, 0x7f,
+	0xc9, 0x38, 0xf6, 0xe7, 0x0b, 0xe3, 0x81, 0xdb, 0x69, 0x94, 0xdb, 0xc9, 0x55, 0x33, 0x7c, 0x2a,
+	0xa9, 0x1e, 0x86, 0x40, 0x54, 0xe4, 0x78, 0x97, 0xe0, 0x10, 0x21, 0xd4, 0x40, 0xb2, 0x35, 0xd6,
+	0x0f, 0x4d, 0x81, 0x5a, 0xb9, 0x80, 0x11, 0x88, 0x23, 0xf3, 0x18, 0xef, 0x0a, 0xce, 0xba, 0x41,
+	0xc0, 0x74, 0x21, 0x1a, 0x1a, 0x94, 0xd4, 0x8f, 0x8c, 0x0f, 0x2f, 0xe4, 0x9d, 0xd1, 0xbd, 0xd0,
+	0xe6, 0x5b, 0x0b, 0xbe, 0x79, 0xc5, 0x2e, 0x3d, 0xf1, 0x88, 0xfa, 0xdd, 0x20, 0x90, 0x98, 0x6c,
+	0x8f, 0xc7, 0x31, 0xfd, 0x66, 0xc4, 0x6b, 0x80, 0x7b, 0x23, 0x12, 0x35, 0xc0, 0xf5, 0x1d, 0x8d,
+	0xd0, 0xdc, 0x35, 0x87, 0xb8, 0xcb, 0x67, 0xa4, 0x6f, 0x55, 0x37, 0x0c, 0xc5, 0xc3, 0x68, 0x34,
+	0x98, 0x1a, 0x5f, 0x2b, 0xc4, 0xa1, 0x39, 0xf0, 0xae, 0xe0, 0x78, 0xc8, 0x09, 0x55, 0xf8, 0x07,
+	0x8b, 0x98, 0x42, 0x69, 0x4c, 0x75, 0x3b, 0x5f, 0x95, 0x7b, 0x2f, 0xc9, 0xe4, 0x98, 0x95, 0xa3,
+	0xbd, 0x5f, 0xe0, 0x64, 0x9c, 0xaa, 0x72, 0xfe, 0xc1, 0xeb, 0xf9, 0x27, 0x62, 0x27, 0xbc, 0xf9,
+	0x9f, 0x05, 0xdf, 0x7d, 0xe0, 0x7c, 0xf4, 0x8c, 0xfd, 0xc9, 0x9f, 0xf7, 0x18, 0xc5, 0x21, 0x55,
+	0x98, 0x99, 0xe0, 0xfa, 0xcf, 0x48, 0xcf, 0x78, 0xa3, 0xae, 0x39, 0x9d, 0x85, 0x18, 0x18, 0x0f,
+	0x2a, 0xc4, 0x59, 0xe6, 0xc0, 0x78, 0x88, 0xd1, 0x94, 0x3d, 0xe1, 0x88, 0xcd, 0xb2, 0x1f, 0x0b,
+	0xa2, 0x82, 0xe8, 0xec, 0xbf, 0xfc, 0x38, 0xed, 0x8b, 0x94, 0xe7, 0x3f, 0x96, 0xb3, 0xce, 0x41,
+	0xf3, 0x8d, 0x05, 0x67, 0xfb, 0x07, 0xa9, 0x4b, 0x0e, 0x13, 0x82, 0x34, 0x18, 0xf3, 0x70, 0x63,
+	0x3a, 0xaa, 0x10, 0x60, 0x05, 0xf1, 0x9a, 0x50, 0x1d, 0x26, 0xe6, 0x22, 0xe1, 0x9a, 0xf9, 0x98,
+	0xf5, 0x54, 0x65, 0x25, 0xa6, 0x9f, 0x90, 0x09, 0x95, 0x2a, 0x4d, 0x59, 0x90, 0xdd, 0xf7, 0x4a,
+	0x9c, 0xad, 0x75, 0x7d, 0x7d, 0xe1, 0xc7, 0x5c, 0x1f, 0x6e, 0x7e, 0xd1, 0xe3, 0x82, 0x78, 0x3f,
+	0x83, 0xfb, 0x09, 0xa6, 0xbb, 0xb2, 0xe4, 0xf8, 0x13, 0x7c, 0xf9, 0x72, 0x98, 0xf7, 0x23, 0x38,
+	0x3d, 0xca, 0x83, 0x07, 0x16, 0xa8, 0x65, 0xf6, 0x12, 0xec, 0x94, 0xbc, 0x17, 0x2b, 0xe4, 0xbd,
+	0xd4, 0x5f, 0xa1, 0x22, 0xce, 0x2c, 0x8f, 0xf4, 0xbe, 0x07, 0x7b, 0x1c, 0xe7, 0xcf, 0xdc, 0x7b,
+	0x13, 0x6c, 0x11, 0x27, 0xcd, 0xc7, 0x9d, 0xbd, 0x4b, 0xb2, 0x36, 0x6c, 0xcc, 0xf1, 0x9e, 0x45,
+	0xd8, 0x4b, 0x65, 0xa2, 0xcc, 0xf6, 0x36, 0xa9, 0x8a, 0x12, 0xd3, 0xa6, 0x10, 0x9c, 0xb3, 0x30,
+	0xd4, 0xc8, 0xec, 0x67, 0x13, 0x90, 0x05, 0xd1, 0x86, 0xf6, 0x69, 0x4c, 0x7d, 0xa6, 0x36, 0xc6,
+	0x50, 0x9b, 0x54, 0xfc, 0x6c, 0xdd, 0x3b, 0xfa, 0xfb, 0x60, 0xfb, 0x98, 0x1d, 0x9a, 0xcf, 0x0f,
+	0xef, 0x02, 0x00, 0x00, 0xff, 0xff, 0xfc, 0x4d, 0xe2, 0x2e, 0xf7, 0x05, 0x00, 0x00,
 }
diff --git a/proto/ttrpc/types.pb.go b/proto/ttrpc/types.pb.go
index d2ac5e4..050578f 100644
--- a/proto/ttrpc/types.pb.go
+++ b/proto/ttrpc/types.pb.go
@@ -28,6 +28,9 @@ const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package
 type ExtraData struct {
 	JsonSpec             []byte     `protobuf:"bytes,1,opt,name=JsonSpec,json=jsonSpec,proto3" json:"JsonSpec,omitempty"`
 	RuncOptions          *types.Any `protobuf:"bytes,2,opt,name=RuncOptions,json=runcOptions,proto3" json:"RuncOptions,omitempty"`
+	StdinPort            uint32     `protobuf:"varint,3,opt,name=StdinPort,json=stdinPort,proto3" json:"StdinPort,omitempty"`
+	StdoutPort           uint32     `protobuf:"varint,4,opt,name=StdoutPort,json=stdoutPort,proto3" json:"StdoutPort,omitempty"`
+	StderrPort           uint32     `protobuf:"varint,5,opt,name=StderrPort,json=stderrPort,proto3" json:"StderrPort,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}   `json:"-"`
 	XXX_unrecognized     []byte     `json:"-"`
 	XXX_sizecache        int32      `json:"-"`
@@ -345,52 +348,54 @@ func init() {
 func init() { proto.RegisterFile("types.proto", fileDescriptor_d938547f84707355) }
 
 var fileDescriptor_d938547f84707355 = []byte{
-	// 715 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x54, 0xcd, 0x6e, 0xe3, 0x36,
-	0x10, 0x8e, 0xaa, 0x26, 0xb5, 0x28, 0x3b, 0x89, 0x89, 0xa2, 0x75, 0xd3, 0xc0, 0x31, 0x7c, 0x72,
-	0x2f, 0x0e, 0x90, 0xa2, 0x05, 0x7a, 0x08, 0x5a, 0xff, 0xa4, 0x88, 0xd3, 0x3a, 0x0e, 0x98, 0xb4,
-	0x87, 0x5e, 0x0a, 0x5a, 0x1a, 0xdb, 0xac, 0x25, 0x52, 0xa0, 0x28, 0x27, 0xce, 0xa9, 0xef, 0xd0,
-	0x07, 0xe8, 0xeb, 0xe4, 0x52, 0xa0, 0xc7, 0x3d, 0x6e, 0xfc, 0x1e, 0x0b, 0x2c, 0x48, 0x4b, 0x8e,
-	0x6c, 0x64, 0xb3, 0xbb, 0x27, 0x81, 0xdf, 0x7c, 0xf3, 0x71, 0xe6, 0x9b, 0x11, 0x91, 0xab, 0xe6,
-	0x11, 0xc4, 0xcd, 0x48, 0x0a, 0x25, 0x0e, 0xbe, 0x1a, 0x0b, 0x31, 0x0e, 0xe0, 0xd8, 0x9c, 0x86,
-	0xc9, 0xe8, 0x98, 0xf2, 0xf9, 0x32, 0x54, 0xff, 0x13, 0x39, 0x67, 0x77, 0x4a, 0xd2, 0x2e, 0x55,
-	0x14, 0x1f, 0xa0, 0xc2, 0x45, 0x2c, 0xf8, 0x75, 0x04, 0x5e, 0xc5, 0xaa, 0x59, 0x8d, 0x22, 0x29,
-	0xfc, 0x95, 0x9e, 0xf1, 0xf7, 0xc8, 0x25, 0x09, 0xf7, 0x06, 0x91, 0x62, 0x82, 0xc7, 0x95, 0x4f,
-	0x6a, 0x56, 0xc3, 0x3d, 0xf9, 0xbc, 0xb9, 0x54, 0x6e, 0x66, 0xca, 0xcd, 0x16, 0x9f, 0x13, 0x57,
-	0x3e, 0x11, 0xeb, 0xff, 0xd8, 0xa8, 0xfc, 0x33, 0x93, 0xe0, 0x49, 0xea, 0x4d, 0x41, 0x76, 0x04,
-	0x1f, 0xb1, 0xf1, 0xa6, 0x9a, 0xf5, 0x81, 0x6a, 0xf8, 0x02, 0x95, 0x2f, 0x41, 0xdd, 0x0a, 0x39,
-	0xed, 0x71, 0x05, 0x72, 0x44, 0x3d, 0xd0, 0xb5, 0xd8, 0x0d, 0xf7, 0xe4, 0xb0, 0x99, 0xbb, 0x66,
-	0x93, 0x44, 0xca, 0x7c, 0x33, 0x0d, 0x37, 0xd0, 0xde, 0x2f, 0x20, 0x39, 0x04, 0xbd, 0x90, 0x8e,
-	0xe1, 0x8a, 0xaa, 0x49, 0xc5, 0xae, 0x59, 0x0d, 0x87, 0xec, 0x4d, 0xd7, 0x61, 0x5c, 0x45, 0x68,
-	0xc9, 0x6c, 0xc9, 0x71, 0x5c, 0xf9, 0xd4, 0x90, 0xd0, 0x74, 0x85, 0xe0, 0x9f, 0x10, 0xea, 0x53,
-	0x6f, 0xc2, 0x38, 0x74, 0x46, 0xe3, 0xca, 0xb6, 0x69, 0xa6, 0x96, 0x2f, 0x27, 0x8b, 0x9a, 0xe6,
-	0x13, 0x49, 0x75, 0x33, 0x04, 0x85, 0xab, 0x1c, 0x7c, 0x8c, 0x1c, 0x22, 0x84, 0xea, 0x4a, 0x36,
-	0x83, 0xca, 0x8e, 0x11, 0x28, 0xe7, 0x05, 0x4c, 0x80, 0x38, 0x32, 0xe3, 0xe0, 0x53, 0xb4, 0xdf,
-	0xf2, 0x7d, 0xa6, 0x85, 0x68, 0x60, 0xa0, 0xb8, 0xf2, 0x99, 0xf1, 0xe1, 0x99, 0xbc, 0x7d, 0xba,
-	0x41, 0xad, 0xbf, 0xb1, 0xd0, 0xd7, 0x2f, 0xd8, 0xa5, 0x3b, 0xee, 0x53, 0xaf, 0xe5, 0xfb, 0x12,
-	0xe2, 0xe5, 0x78, 0x1c, 0x53, 0x6f, 0x8a, 0xe0, 0x1a, 0x72, 0xcf, 0x45, 0xac, 0xba, 0x30, 0xbb,
-	0xa4, 0x21, 0x98, 0x6d, 0x70, 0x88, 0x3b, 0x79, 0x82, 0xf0, 0x21, 0x72, 0x5a, 0x41, 0x20, 0x6e,
-	0xfb, 0xfd, 0xee, 0xb5, 0xf1, 0xb5, 0x40, 0x1c, 0x9a, 0x01, 0xf8, 0x14, 0x95, 0x7a, 0x9c, 0x50,
-	0x05, 0xbf, 0xb2, 0x90, 0x29, 0x90, 0xc6, 0x54, 0xf7, 0xe4, 0xcb, 0x7c, 0xed, 0xb9, 0x30, 0x29,
-	0xb1, 0x3c, 0x1b, 0xff, 0x88, 0x76, 0x07, 0x89, 0xca, 0xe7, 0x6f, 0xbf, 0x9c, 0xbf, 0x2b, 0xd6,
-	0xe8, 0xf5, 0x7f, 0x2d, 0x74, 0xf4, 0x9e, 0xf9, 0xe8, 0x1e, 0x3b, 0x57, 0xbf, 0xdd, 0x40, 0x18,
-	0x05, 0x54, 0x41, 0x6a, 0x82, 0xeb, 0x3d, 0x41, 0xba, 0xc7, 0x73, 0x75, 0xc6, 0xe9, 0x30, 0x00,
-	0xdf, 0x78, 0x50, 0x20, 0xce, 0x24, 0x03, 0x8c, 0x87, 0x10, 0x5e, 0xb3, 0x7b, 0xe8, 0xb3, 0xa1,
-	0xb1, 0xa0, 0x44, 0x50, 0xb8, 0x42, 0x74, 0xf6, 0xef, 0x5e, 0x94, 0x74, 0x44, 0xc2, 0x95, 0xe9,
-	0xbf, 0x44, 0x9c, 0x59, 0x06, 0xd4, 0xff, 0xb3, 0xd0, 0xfe, 0xe6, 0x20, 0xb5, 0x64, 0x2f, 0x26,
-	0x40, 0xfd, 0x01, 0x0f, 0xe6, 0xa6, 0xa2, 0x02, 0x41, 0x6c, 0x85, 0xe0, 0x3a, 0x2a, 0xf6, 0x62,
-	0xb3, 0x48, 0x30, 0x63, 0x1e, 0xa4, 0x35, 0x15, 0x59, 0x0e, 0xd3, 0x3f, 0xf9, 0x15, 0x95, 0x2a,
-	0x49, 0x98, 0x9f, 0xee, 0x7b, 0x21, 0x4a, 0xcf, 0x5a, 0x5f, 0x2f, 0xfc, 0x80, 0xeb, 0xe1, 0x66,
-	0x8b, 0x1e, 0xad, 0x10, 0xfc, 0x03, 0x72, 0x3f, 0xc2, 0x74, 0x57, 0xe6, 0x1c, 0xbf, 0x47, 0x5f,
-	0x3c, 0x4f, 0xc3, 0xdf, 0x21, 0xa7, 0x4d, 0xb9, 0x7f, 0xcb, 0x7c, 0x35, 0x49, 0x5f, 0x82, 0x35,
-	0xc9, 0x1b, 0x31, 0x05, 0xde, 0x4e, 0xbc, 0x29, 0x28, 0xe2, 0x0c, 0x33, 0x26, 0xfe, 0x06, 0xd9,
-	0x83, 0x28, 0x7b, 0x88, 0xde, 0x99, 0x60, 0x8b, 0x28, 0xae, 0xdf, 0xad, 0xdd, 0x9d, 0x0b, 0x6b,
-	0xc3, 0x06, 0x1c, 0x6e, 0x58, 0x08, 0xed, 0x44, 0xc6, 0xca, 0x5c, 0x6f, 0x93, 0xa2, 0xc8, 0x61,
-	0xda, 0x14, 0x02, 0x23, 0x16, 0x04, 0x1a, 0x32, 0xf7, 0xd9, 0x04, 0xc9, 0x15, 0xa2, 0x0d, 0xed,
-	0xd0, 0x88, 0x7a, 0x4c, 0xcd, 0x8d, 0xa1, 0x36, 0x29, 0x78, 0xe9, 0xb9, 0x7d, 0xf4, 0xf0, 0x58,
-	0xdd, 0x7a, 0xf5, 0x58, 0xdd, 0xfa, 0x7b, 0x51, 0xb5, 0x1e, 0x16, 0x55, 0xeb, 0xff, 0x45, 0xd5,
-	0x7a, 0xbd, 0xa8, 0x5a, 0x7f, 0x6c, 0x2f, 0x1f, 0xb9, 0x1d, 0xf3, 0xf9, 0xf6, 0x6d, 0x00, 0x00,
-	0x00, 0xff, 0xff, 0x17, 0x24, 0xbf, 0xb1, 0xb0, 0x05, 0x00, 0x00,
+	// 750 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x94, 0xcd, 0x6e, 0xe3, 0x36,
+	0x10, 0xc7, 0xa3, 0xaa, 0x49, 0x2d, 0xca, 0xf9, 0xb0, 0x50, 0xb4, 0x6e, 0x1a, 0x38, 0x86, 0x4f,
+	0xee, 0xc5, 0x01, 0x5c, 0xb4, 0x40, 0x0f, 0x41, 0xeb, 0x8f, 0x14, 0x71, 0x5a, 0xc7, 0x06, 0x9d,
+	0xf6, 0xd0, 0x1b, 0x2d, 0x8d, 0x6d, 0xd6, 0x12, 0x29, 0x50, 0x94, 0x13, 0xe7, 0xd4, 0x77, 0xe8,
+	0x03, 0xf4, 0x49, 0x7a, 0xcf, 0xa5, 0x40, 0x8f, 0x7b, 0xdc, 0xf8, 0x3d, 0x16, 0x58, 0x90, 0x96,
+	0x14, 0xd9, 0xc8, 0x66, 0x77, 0x4f, 0x02, 0x7f, 0xff, 0x99, 0xe1, 0xcc, 0x9f, 0x14, 0x91, 0x2d,
+	0x97, 0x21, 0x44, 0x8d, 0x50, 0x70, 0xc9, 0x8f, 0xbf, 0x9a, 0x72, 0x3e, 0xf5, 0xe1, 0x4c, 0xaf,
+	0xc6, 0xf1, 0xe4, 0x8c, 0xb0, 0xe5, 0x5a, 0xaa, 0xfd, 0x6b, 0x20, 0xeb, 0xe2, 0x4e, 0x0a, 0xd2,
+	0x25, 0x92, 0x38, 0xc7, 0xa8, 0x70, 0x15, 0x71, 0x36, 0x0a, 0xc1, 0x2d, 0x1b, 0x55, 0xa3, 0x5e,
+	0xc4, 0x85, 0x3f, 0x93, 0xb5, 0xf3, 0x3d, 0xb2, 0x71, 0xcc, 0xdc, 0x41, 0x28, 0x29, 0x67, 0x51,
+	0xf9, 0x93, 0xaa, 0x51, 0xb7, 0x9b, 0x9f, 0x37, 0xd6, 0xa5, 0x1b, 0x69, 0xe9, 0x46, 0x8b, 0x2d,
+	0xb1, 0x2d, 0x9e, 0x02, 0x9d, 0x13, 0x64, 0x8d, 0xa4, 0x47, 0xd9, 0x90, 0x0b, 0x59, 0x36, 0xab,
+	0x46, 0x7d, 0x1f, 0x5b, 0x51, 0x0a, 0x9c, 0x0a, 0x42, 0x23, 0xe9, 0xf1, 0x58, 0x6a, 0xf9, 0x53,
+	0x2d, 0xa3, 0x28, 0x23, 0x89, 0x0e, 0x42, 0x68, 0x7d, 0x37, 0xd3, 0x13, 0x52, 0xfb, 0xdb, 0x44,
+	0xa5, 0x9f, 0xa9, 0x00, 0x57, 0x10, 0x77, 0x0e, 0xa2, 0xc3, 0xd9, 0x84, 0x4e, 0xb7, 0x7b, 0x35,
+	0x3e, 0xb4, 0xd7, 0x2b, 0x54, 0xba, 0x06, 0x79, 0xcb, 0xc5, 0xbc, 0xc7, 0x24, 0x88, 0x09, 0x71,
+	0x41, 0x4d, 0x6a, 0xd6, 0xed, 0xe6, 0x49, 0x23, 0xb7, 0xcd, 0x76, 0x10, 0x2e, 0xb1, 0xed, 0x34,
+	0xa7, 0x8e, 0x0e, 0x7f, 0x01, 0xc1, 0xc0, 0xef, 0x05, 0x64, 0x0a, 0x43, 0x22, 0x67, 0x7a, 0x7a,
+	0x0b, 0x1f, 0xce, 0x37, 0xb1, 0x9a, 0x71, 0x1d, 0xd9, 0x12, 0xd3, 0x48, 0x7b, 0x60, 0x61, 0x34,
+	0xcf, 0x88, 0xf3, 0x13, 0x42, 0x7d, 0xe2, 0xce, 0x28, 0x83, 0xce, 0x64, 0xaa, 0x3d, 0xb0, 0x9b,
+	0xd5, 0x7c, 0x3b, 0xa9, 0xaa, 0x87, 0x8f, 0x05, 0x51, 0xc3, 0x60, 0x14, 0x64, 0x39, 0xce, 0x19,
+	0xb2, 0x30, 0xe7, 0xb2, 0x2b, 0xe8, 0x02, 0xca, 0x7b, 0xba, 0x40, 0x29, 0x5f, 0x40, 0x0b, 0xd8,
+	0x12, 0x69, 0x8c, 0x73, 0x8e, 0x8e, 0x5a, 0x9e, 0x47, 0x55, 0x21, 0xe2, 0x6b, 0x14, 0x95, 0x3f,
+	0xd3, 0x3e, 0x3c, 0x93, 0x77, 0x44, 0xb6, 0x42, 0x6b, 0x6f, 0x0c, 0xf4, 0xf5, 0x0b, 0x76, 0xa9,
+	0x89, 0xfb, 0xc4, 0x6d, 0x79, 0x9e, 0x80, 0x68, 0x7d, 0x3c, 0x96, 0xee, 0x37, 0x21, 0x4e, 0x15,
+	0xd9, 0x97, 0x3c, 0x92, 0x5d, 0x58, 0x5c, 0x93, 0x00, 0xf4, 0x5d, 0xb3, 0xb0, 0x3d, 0x7b, 0x42,
+	0xea, 0x56, 0xb5, 0x7c, 0x9f, 0xdf, 0xf6, 0xfb, 0xdd, 0x91, 0xf6, 0xb5, 0x80, 0x2d, 0x92, 0x02,
+	0xe7, 0x1c, 0xed, 0xf7, 0x18, 0x26, 0x12, 0x7e, 0xa5, 0x01, 0x95, 0x20, 0xb4, 0xa9, 0x76, 0xf3,
+	0xcb, 0x7c, 0xef, 0x39, 0x19, 0xef, 0xd3, 0x7c, 0xb4, 0xf3, 0x23, 0x3a, 0x18, 0xc4, 0x32, 0x9f,
+	0xbf, 0xfb, 0x72, 0xfe, 0x01, 0xdf, 0x08, 0xaf, 0xfd, 0x63, 0xa0, 0xd3, 0xf7, 0x9c, 0x8f, 0x9a,
+	0xb1, 0x33, 0xfc, 0xed, 0x06, 0x82, 0xd0, 0x27, 0x12, 0x12, 0x13, 0x6c, 0xf7, 0x09, 0xa9, 0x19,
+	0x2f, 0xe5, 0x05, 0x23, 0x63, 0x1f, 0x3c, 0xed, 0x41, 0x01, 0x5b, 0xb3, 0x14, 0x68, 0x0f, 0x21,
+	0x18, 0xd1, 0x7b, 0xe8, 0xd3, 0x71, 0xf2, 0x63, 0xa1, 0x20, 0x23, 0x2a, 0xfb, 0x77, 0x37, 0x8c,
+	0x3b, 0x3c, 0x66, 0xe9, 0x8f, 0x65, 0x2d, 0x52, 0x50, 0xfb, 0xcf, 0x40, 0x47, 0xdb, 0x07, 0xa9,
+	0x4a, 0xf6, 0x22, 0x0c, 0xc4, 0x1b, 0x30, 0x7f, 0xa9, 0x3b, 0x2a, 0x60, 0x44, 0x33, 0xe2, 0xd4,
+	0x50, 0xb1, 0x17, 0xe9, 0x8b, 0x04, 0x0b, 0xea, 0x42, 0xd2, 0x53, 0x91, 0xe6, 0x98, 0x7a, 0x42,
+	0x86, 0x44, 0xc8, 0x38, 0xa6, 0x5e, 0x72, 0xdf, 0x0b, 0x61, 0xb2, 0x56, 0xf5, 0xd5, 0x85, 0x1f,
+	0x30, 0x75, 0xb8, 0xe9, 0x45, 0x0f, 0x33, 0xe2, 0xfc, 0x80, 0xec, 0x8f, 0x30, 0xdd, 0x16, 0x39,
+	0xc7, 0xef, 0xd1, 0x17, 0xcf, 0x87, 0x39, 0xdf, 0x21, 0xab, 0x4d, 0x98, 0x77, 0x4b, 0x3d, 0x39,
+	0x4b, 0x5e, 0x82, 0x8d, 0x92, 0x37, 0x7c, 0x0e, 0xac, 0x1d, 0xbb, 0x73, 0x90, 0xd8, 0x1a, 0xa7,
+	0x91, 0xce, 0x37, 0xc8, 0x1c, 0x84, 0xe9, 0x33, 0xf7, 0xce, 0x04, 0x93, 0x87, 0x51, 0xed, 0x6e,
+	0x63, 0xef, 0x9c, 0xac, 0x0c, 0x1b, 0x30, 0xb8, 0xa1, 0x01, 0xb4, 0x63, 0x11, 0x49, 0xbd, 0xbd,
+	0x89, 0x8b, 0x3c, 0xc7, 0x94, 0x29, 0x18, 0x26, 0xd4, 0xf7, 0x15, 0xd2, 0xfb, 0x99, 0x18, 0x89,
+	0x8c, 0x28, 0x43, 0x3b, 0x24, 0x24, 0x2e, 0x95, 0x4b, 0x6d, 0xa8, 0x89, 0x0b, 0x6e, 0xb2, 0x6e,
+	0x9f, 0x3e, 0x3c, 0x56, 0x76, 0x5e, 0x3d, 0x56, 0x76, 0xfe, 0x5a, 0x55, 0x8c, 0x87, 0x55, 0xc5,
+	0xf8, 0x7f, 0x55, 0x31, 0x5e, 0xaf, 0x2a, 0xc6, 0x1f, 0xbb, 0xeb, 0x47, 0x6e, 0x4f, 0x7f, 0xbe,
+	0x7d, 0x1b, 0x00, 0x00, 0xff, 0xff, 0xd6, 0x23, 0xb3, 0xad, 0x0f, 0x06, 0x00, 0x00,
 }
 
 func (m *ExtraData) Marshal() (dAtA []byte, err error) {
@@ -424,6 +429,21 @@ func (m *ExtraData) MarshalTo(dAtA []byte) (int, error) {
 		}
 		i += n1
 	}
+	if m.StdinPort != 0 {
+		dAtA[i] = 0x18
+		i++
+		i = encodeVarintTypes(dAtA, i, uint64(m.StdinPort))
+	}
+	if m.StdoutPort != 0 {
+		dAtA[i] = 0x20
+		i++
+		i = encodeVarintTypes(dAtA, i, uint64(m.StdoutPort))
+	}
+	if m.StderrPort != 0 {
+		dAtA[i] = 0x28
+		i++
+		i = encodeVarintTypes(dAtA, i, uint64(m.StderrPort))
+	}
 	if m.XXX_unrecognized != nil {
 		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
@@ -790,6 +810,15 @@ func (m *ExtraData) Size() (n int) {
 		l = m.RuncOptions.Size()
 		n += 1 + l + sovTypes(uint64(l))
 	}
+	if m.StdinPort != 0 {
+		n += 1 + sovTypes(uint64(m.StdinPort))
+	}
+	if m.StdoutPort != 0 {
+		n += 1 + sovTypes(uint64(m.StdoutPort))
+	}
+	if m.StderrPort != 0 {
+		n += 1 + sovTypes(uint64(m.StderrPort))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -987,6 +1016,9 @@ func (this *ExtraData) String() string {
 	s := strings.Join([]string{`&ExtraData{`,
 		`JsonSpec:` + fmt.Sprintf("%v", this.JsonSpec) + `,`,
 		`RuncOptions:` + strings.Replace(fmt.Sprintf("%v", this.RuncOptions), "Any", "types.Any", 1) + `,`,
+		`StdinPort:` + fmt.Sprintf("%v", this.StdinPort) + `,`,
+		`StdoutPort:` + fmt.Sprintf("%v", this.StdoutPort) + `,`,
+		`StderrPort:` + fmt.Sprintf("%v", this.StderrPort) + `,`,
 		`XXX_unrecognized:` + fmt.Sprintf("%v", this.XXX_unrecognized) + `,`,
 		`}`,
 	}, "")
@@ -1195,6 +1227,63 @@ func (m *ExtraData) Unmarshal(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field StdinPort", wireType)
+			}
+			m.StdinPort = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowTypes
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.StdinPort |= uint32(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 4:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field StdoutPort", wireType)
+			}
+			m.StdoutPort = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowTypes
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.StdoutPort |= uint32(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 5:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field StderrPort", wireType)
+			}
+			m.StderrPort = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowTypes
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.StderrPort |= uint32(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipTypes(dAtA[iNdEx:])
diff --git a/proto/types.proto b/proto/types.proto
index bc11f0b..6fab535 100644
--- a/proto/types.proto
+++ b/proto/types.proto
@@ -8,6 +8,9 @@ option go_package = "proto";
 message ExtraData {
 	bytes JsonSpec = 1;
 	google.protobuf.Any RuncOptions = 2;
+	uint32 StdinPort = 3;
+	uint32 StdoutPort = 4;
+	uint32 StderrPort = 5;
 }
 
 // Message to specify the various Firecracker VM configurations for a task
diff --git a/runtime/service.go b/runtime/service.go
index 0ff0b0c..1f669be 100644
--- a/runtime/service.go
+++ b/runtime/service.go
@@ -15,27 +15,26 @@ package main
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
-	"io"
-	"io/ioutil"
 	"math"
-	"strings"
-
-	// disable gosec check for math/rand. We just need a random starting
-	// place to start looking for CIDs; no need for cryptographically
-	// secure randomness
-	"math/rand" // #nosec
-
 	"net"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"runtime/debug"
+	"strings"
 	"sync"
 	"syscall"
 	"time"
 	"unsafe"
 
+	// disable gosec check for math/rand. We just need a random starting
+	// place to start looking for CIDs; no need for cryptographically
+	// secure randomness
+	"math/rand" // #nosec
+
+	eventsAPI "github.com/containerd/containerd/api/events"
+	"github.com/containerd/containerd/cio"
 	"github.com/containerd/containerd/events"
 	"github.com/containerd/containerd/events/exchange"
 	"github.com/containerd/containerd/log"
@@ -43,7 +42,6 @@ import (
 	"github.com/containerd/containerd/runtime"
 	"github.com/containerd/containerd/runtime/v2/shim"
 	taskAPI "github.com/containerd/containerd/runtime/v2/task"
-	"github.com/containerd/fifo"
 	"github.com/containerd/ttrpc"
 	"github.com/containerd/typeurl"
 	"github.com/firecracker-microvm/firecracker-go-sdk"
@@ -56,9 +54,9 @@ import (
 	"golang.org/x/sys/unix"
 
 	"github.com/firecracker-microvm/firecracker-containerd/eventbridge"
-	"github.com/firecracker-microvm/firecracker-containerd/internal"
+	"github.com/firecracker-microvm/firecracker-containerd/internal/bundle"
+	"github.com/firecracker-microvm/firecracker-containerd/internal/vm"
 	proto "github.com/firecracker-microvm/firecracker-containerd/proto/grpc"
-	"github.com/firecracker-microvm/firecracker-containerd/runtime/firecrackeroci"
 )
 
 func init() {
@@ -67,45 +65,43 @@ func init() {
 
 const (
 	defaultVsockPort     = 10789
+	minVsockIOPort       = uint32(11000)
 	supportedMountFSType = "ext4"
 
 	vmIDEnvVarKey = "FIRECRACKER_VM_ID"
 
-	varRunDir                  = "/var/run/firecracker-containerd"
-	firecrackerSockName        = "firecracker.sock"
-	firecrackerLogFifoName     = "fc-logs.fifo"
-	firecrackerMetricsFifoName = "fc-metrics.fifo"
-
-	// TODO these strings are hardcoded throughout the containerd codebase, it may
-	// be worth sending them a PR to define them as constants accessible to shim
-	// implementations like our own
-	shimAddrFileName = "address"
-	shimLogFifoName  = "log"
-	ociConfigName    = "config.json"
+	varRunDir = "/var/run/firecracker-containerd"
+)
+
+var (
+	// type assertions
+	_ taskAPI.TaskService = &service{}
+	_ shim.Init           = NewService
+
+	sysCall = syscall.Syscall
 )
 
 // implements shimapi
 type service struct {
+	taskManager vm.TaskManager
+
 	server        *ttrpc.Server
-	id            string
-	vmID          string
 	publish       events.Publisher
 	eventExchange *exchange.Exchange
+	namespace     string
+
+	vmID   string
+	config *Config
 
 	startVMMutex sync.Mutex
 	agentStarted bool
 	agentClient  taskAPI.TaskService
-	config       *Config
-	machine      *firecracker.Machine
-	machineCID   uint32
-	ctx          context.Context
-	cancel       context.CancelFunc
-}
 
-var (
-	_       = (taskAPI.TaskService)(&service{})
-	sysCall = syscall.Syscall
-)
+	machine          *firecracker.Machine
+	machineCID       uint32
+	vsockIOPortCount uint32
+	vsockPortMu      sync.Mutex
+}
 
 func shimOpts(ctx context.Context) (*shim.Opts, error) {
 	opts, ok := ctx.Value(shim.OptsKey{}).(shim.Opts)
@@ -116,68 +112,6 @@ func shimOpts(ctx context.Context) (*shim.Opts, error) {
 	return &opts, nil
 }
 
-// The bundle dir for the container can be provided by a containerd-defined flag, or, if that's
-// not set, it's assumed to be the current working directory of this process (as set by the
-// containerd parent process).
-func resolveBundleDir(ctx context.Context) (string, error) {
-	// Check to see if it was set by command line flag
-	opts, err := shimOpts(ctx)
-	if err != nil {
-		return "", err
-	}
-
-	if opts.BundlePath != "" {
-		return opts.BundlePath, nil
-	}
-
-	// If no command line flag, this must be a shim start routine, so we can assume it's the
-	// current working directory
-	cwd, err := os.Getwd()
-	if err != nil {
-		return "", errors.Wrap(err, "failed to read bundle dir path")
-	}
-
-	return cwd, nil
-}
-
-// If the VMID was provided by the client, we'll find it in the OCI config Annotations section.
-// If, however, no VMID was provided in the OCI config, we it may have been set by a "shim start"
-// parent process as an env var. resolveVMID checks those two locations and returns the VMID
-// if found in either.
-func resolveVMID(ctx context.Context) (string, error) {
-	// If the vmID is provided via env var, use that
-	if vmID := os.Getenv(vmIDEnvVarKey); vmID != "" {
-		return vmID, nil
-	}
-
-	// if no env var, check the OCI config annotation section
-	bundleDir, err := resolveBundleDir(ctx)
-	if err != nil {
-		return "", err
-	}
-
-	specPath := filepath.Join(bundleDir, ociConfigName)
-	ociConfigFile, err := os.Open(specPath)
-	if err != nil {
-		return "", errors.Wrapf(err, "failed to open OCI config file %s", specPath)
-	}
-
-	defer ociConfigFile.Close()
-	var ociConfig struct {
-		Annotations map[string]string `json:"annotations,omitempty"`
-	}
-
-	if err := json.NewDecoder(ociConfigFile).Decode(&ociConfig); err != nil {
-		return "", errors.Wrapf(err, "failed to parse Annotations section of OCI config file %s", specPath)
-	}
-
-	// This will return empty string if the key is not present in the OCI config, which the caller can decided
-	// how to deal with
-	return ociConfig.Annotations[firecrackeroci.VMIDAnnotationKey], nil
-}
-
-var _ shim.Init = NewService
-
 // NewService creates new runtime shim.
 func NewService(ctx context.Context, id string, remotePublisher events.Publisher) (shim.Shim, error) {
 	server, err := newServer()
@@ -199,165 +133,164 @@ func NewService(ctx context.Context, id string, remotePublisher events.Publisher
 		config.Debug = opts.Debug
 	}
 
+	namespace, ok := namespaces.Namespace(ctx)
+	if !ok {
+		namespace = namespaces.Default
+	}
+
 	eventExchange := exchange.NewExchange()
 
 	// Republish each event received on our exchange to the provided remote publisher.
-	// TODO ideally we would be forwarding events instead of re-publishing them, which would preserve the events'
-	// original timestamps and namespaces. However, as of this writing, the containerd v2 runtime model only provides a
-	// shim with a publisher, not a forwarder, so we have to republish for now.
+	// TODO ideally we would be forwarding events instead of re-publishing them, which would
+	// preserve the events' original timestamps and namespaces. However, as of this writing,
+	// the containerd v2 runtime model only provides a shim with a publisher, not a forwarder,
+	// so we have to republish for now.
 	go func() {
 		if err := <-eventbridge.Republish(ctx, eventExchange, remotePublisher); err != nil && err != context.Canceled {
 			log.G(ctx).WithError(err).Error("error while republishing events")
 		}
 	}()
 
-	vmID, err := resolveVMID(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	// If no VMID was provided by the client or already auto-generated by a parent "shim start", generate a new random one.
-	// This results in a default behavior of running each container in its own VM.
-	if vmID == "" {
-		uuid, err := uuid.NewV4()
-		if err != nil {
-			return nil, errors.Wrap(err, "failed to generate UUID for VMID")
-		}
-		vmID = uuid.String()
-	}
-
 	s := &service{
+		taskManager: vm.NewTaskManager(),
+
 		server:        server,
-		id:            id,
-		vmID:          vmID,
 		publish:       remotePublisher,
 		eventExchange: eventExchange,
+		namespace:     namespace,
 
+		vmID:   os.Getenv(vmIDEnvVarKey),
 		config: config,
 	}
 
 	return s, nil
 }
 
-// vmDir holds files, sockets and FIFOs scoped to a single given VM
-func (s *service) vmDir() string {
-	return filepath.Join(varRunDir, s.vmID)
+// vmDir holds files, sockets and FIFOs scoped to a single given VM.
+// It is unique per-VM and containerd namespace.
+func (s *service) vmDir() vm.Dir {
+	return vm.Dir(filepath.Join(varRunDir, s.namespace, s.vmID))
 }
 
-// shimAddrFilePath is the path to the shim address file as found in the vmDir. The
-// address file holds the abstract unix socket path at which the shim serves its API.
-// It's read by containerd to find the shim.
-func (s *service) shimAddrFilePath() string {
-	return filepath.Join(s.vmDir(), shimAddrFileName)
+// shimSocketAddress is the abstract unix socket path at which our shim serves its API
+// It is unique per-VM and containerd namespace.
+func (s *service) shimSocketAddress(ctx context.Context) (string, error) {
+	return shim.SocketAddress(ctx, s.vmID)
 }
 
-// bundleAddrFilePath is the path to the address file as found in the bundleDir.
-// Even though the shim address is set per-VM, not per-container, containerd expects
-// to find the shim addr file in the bundle dir, so we still have to create it or
-// symlink it to the shimAddrFilePath
-func (s *service) bundleAddrFilePath(ctx context.Context) (string, error) {
-	bundleDir, err := resolveBundleDir(ctx)
+func (s *service) newShim(ctx context.Context, containerdBinary, containerdAddress string, shimSocket *net.UnixListener) (*exec.Cmd, error) {
+	ns, err := namespaces.NamespaceRequired(ctx)
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 
-	return filepath.Join(bundleDir, shimAddrFileName), nil
-}
-
-// shimLogFifoPath is a path to a FIFO for writing shim logs as found in the vmDir.
-func (s *service) shimLogFifoPath() string {
-	return filepath.Join(s.vmDir(), shimLogFifoName)
-}
-
-// bundleLogFifoPath is a path to a FIFO for writing shim logs as found in the bundleDir.
-// It is the path created by containerd for us, the shimLogFifoPath is just a symlink to one.
-func (s *service) bundleLogFifoPath(ctx context.Context) (string, error) {
-	bundleDir, err := resolveBundleDir(ctx)
+	self, err := os.Executable()
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 
-	return filepath.Join(bundleDir, shimLogFifoName), nil
-}
+	args := []string{
+		"-namespace", ns,
+		"-address", containerdAddress,
+		"-publish-binary", containerdBinary,
+	}
 
-// shimSocketAddress is the abstract unix socket path at which our shim serves its API
-// It is unique per-VM.
-func (s *service) shimSocketAddress(ctx context.Context) (string, error) {
-	return shim.SocketAddress(ctx, s.vmID)
-}
+	if s.config.Debug {
+		args = append(args, "-debug")
+	}
 
-// firecrackerSockPath is the path to the firecracker VMM's API socket
-func (s *service) firecrackerSockPath() string {
-	return filepath.Join(s.vmDir(), firecrackerSockName)
-}
+	cmd := exec.Command(self, args...)
 
-// firecrackerLogFifoPath is the path to the firecracker VMM's log fifo
-func (s *service) firecrackerLogFifoPath() string {
-	return filepath.Join(s.vmDir(), firecrackerLogFifoName)
-}
+	cmd.Dir = s.vmDir().RootPath()
 
-// firecrackerMetricsFifoPath is the path to the firecracker VMM's metrics fifo
-func (s *service) firecrackerMetricsFifoPath() string {
-	return filepath.Join(s.vmDir(), firecrackerMetricsFifoName)
-}
+	cmd.Env = append(os.Environ(),
+		fmt.Sprintf("%s=%s", vmIDEnvVarKey, s.vmID))
 
-// containerd expects there to be a file named "address" in the container bundle directory (it contains the
-// unix abstract address at which the shim managing that container can be found). We create that file in the
-// shim's VM directory and then symlink to it from each container's bundle dir.
-func (s *service) createAddressSymlink(ctx context.Context) error {
-	bundleAddrFilePath, err := s.bundleAddrFilePath(ctx)
+	shimSocketFile, err := shimSocket.File()
 	if err != nil {
-		return err
+		return nil, err
 	}
+	cmd.ExtraFiles = append(cmd.ExtraFiles, shimSocketFile)
 
-	err = os.Symlink(s.shimAddrFilePath(), bundleAddrFilePath)
-	if err != nil {
-		return errors.Wrapf(err, `failed to create shim address file symlink from "%s"->"%s"`, bundleAddrFilePath, s.shimAddrFilePath())
+	cmd.SysProcAttr = &syscall.SysProcAttr{
+		Setpgid: true,
 	}
 
-	return nil
-}
-
-// containerd creates a fifo for writing shim logs every time it spins up a container in the container's bundle dir.
-// Code defined as part containerd packages also assume there to be a fifo name "log" in the current working directory
-// of the shim. Our shim's working directory after initialization is the vmID dir, so we create a symlink from
-// <vmdir>/log to the log fifo of the first container spun up for the shim.
-func (s *service) createShimLogFifoSymlink(ctx context.Context) error {
-	bundleLogFifoPath, err := s.bundleLogFifoPath(ctx)
+	err = cmd.Start()
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	err = os.Symlink(bundleLogFifoPath, s.shimLogFifoPath())
+	// make sure to wait after start
+	go func() {
+		log.G(ctx).WithField("vmID", s.vmID).Debug("waiting on shim process")
+		waitErr := cmd.Wait()
+		log.G(ctx).WithError(waitErr).WithField("vmID", s.vmID).Debug("completed waiting on shim process")
+	}()
+
+	err = shim.SetScore(cmd.Process.Pid)
 	if err != nil {
-		return errors.Wrapf(err, `failed to create shim log fifo symlink from "%s"->"%s"`, s.shimLogFifoPath(), bundleLogFifoPath)
+		log.G(ctx).WithError(err).WithField("vmID", s.vmID).Error("failed to set OOM score on shim")
+		return nil, errors.Wrap(err, "failed to set OOM Score on shim")
 	}
 
-	return nil
+	return cmd, nil
 }
 
 func isEADDRINUSE(err error) bool {
 	return err != nil && strings.Contains(err.Error(), "address already in use")
 }
 
-func (s *service) StartShim(ctx context.Context, id, containerdBinary, containerdAddress string) (string, error) {
-	log.G(ctx).WithField("id", id).Debug("StartShim")
+func (s *service) StartShim(ctx context.Context, containerID, containerdBinary, containerdAddress string) (string, error) {
+	log.G(ctx).WithField("id", containerID).Debug("StartShim")
 
-	shimSocketAddress, err := s.shimSocketAddress(ctx)
+	// If we are running a shim start routine, we can safely assume our current working
+	// directory is the bundle directory
+	cwd, err := os.Getwd()
+	if err != nil {
+		return "", errors.Wrap(err, "failed to get current working directory")
+	}
+	bundleDir := bundle.Dir(cwd)
+
+	// Since we're running a shim start routine, we need to determine the vmID for the incoming
+	// container. Start by looking at the container's OCI annotations
+	s.vmID, err = bundleDir.OCIConfig().VMID()
 	if err != nil {
 		return "", err
 	}
 
+	if s.vmID == "" {
+		// If here, no VMID has been provided by the client for this container, so auto-generate a new one.
+		// This results in a default behavior of running each container in its own VM if not otherwise
+		// specified by the client.
+		uuid, err := uuid.NewV4()
+		if err != nil {
+			return "", errors.Wrap(err, "failed to generate UUID for VMID")
+		}
+
+		s.vmID = uuid.String()
+	}
+
 	// We determine if there is already a shim managing a VM with the current VMID by attempting
 	// to listen on the abstract socket address (which is parameterized by VMID). If we get
 	// EADDRINUSE, then we assume there is already a shim for the VM and just hand off to that
 	// one. This is in line with the approach used by containerd's reference runC shim v2
 	// implementation (which is also designed to manage multiple containers from a single shim
 	// process)
+	shimSocketAddress, err := s.shimSocketAddress(ctx)
+	if err != nil {
+		return "", err
+	}
+
 	shimSocket, err := shim.NewSocket(shimSocketAddress)
 	if isEADDRINUSE(err) {
 		// There's already a shim for this VMID, so just hand off to it
-		err = s.createAddressSymlink(ctx)
+		err = s.vmDir().CreateBundleLink(containerID, bundleDir)
+		if err != nil {
+			return "", err
+		}
+
+		err = s.vmDir().CreateAddressLink(containerID)
 		if err != nil {
 			return "", err
 		}
@@ -369,75 +302,125 @@ func (s *service) StartShim(ctx context.Context, id, containerdBinary, container
 
 	// If we're here, there is no pre-existing shim for this VMID, so we spawn a new one
 	defer func() {
-		log.G(ctx).WithField("id", id).Debug("closing shim socket")
+		log.G(ctx).WithField("id", containerID).Debug("closing shim socket")
 		shimSocket.Close()
 	}()
 
-	err = os.MkdirAll(s.vmDir(), 0700)
+	err = s.vmDir().Create()
 	if err != nil {
 		return "", err
 	}
 
-	shimSocketFile, err := shimSocket.File()
+	err = s.vmDir().CreateBundleLink(containerID, bundleDir)
 	if err != nil {
 		return "", err
 	}
-	defer func() {
-		log.G(ctx).WithField("id", id).Debug("closing shim socket file")
-		shimSocketFile.Close()
-	}()
 
-	cmd, err := s.newCommand(ctx, id, containerdBinary, containerdAddress, shimSocketFile)
+	cmd, err := s.newShim(ctx, containerdBinary, containerdAddress, shimSocket)
 	if err != nil {
-		return "", err
-	}
-
-	err = cmd.Start()
-	if err != nil {
-		log.G(ctx).WithError(err).WithField("id", id).Error("Failed to Start()")
+		log.G(ctx).WithError(err).WithField("id", containerID).Error("Failed to start new shim")
 		return "", err
 	}
 
 	defer func() {
 		if err != nil {
-			log.G(ctx).WithError(err).WithField("id", id).Error("killing shim process")
+			log.G(ctx).WithError(err).WithField("id", containerID).Error("killing shim process after error")
 			cmd.Process.Kill()
 		}
 	}()
 
-	// make sure to wait after start
-	go func() {
-		log.G(ctx).WithField("id", id).Debug("waiting on shim process")
-		waitErr := cmd.Wait()
-		log.G(ctx).WithError(waitErr).WithField("id", id).Debug("completed waiting on shim process")
-	}()
-
-	err = shim.WriteAddress(s.shimAddrFilePath(), shimSocketAddress)
+	err = s.vmDir().WriteAddress(shimSocketAddress)
 	if err != nil {
-		log.G(ctx).WithError(err).WithField("id", id).Error("failed to write address")
+		log.G(ctx).WithError(err).WithField("id", containerID).Error("failed to write address")
 		return "", err
 	}
 
-	err = s.createAddressSymlink(ctx)
+	err = s.vmDir().CreateAddressLink(containerID)
 	if err != nil {
 		return "", err
 	}
 
-	err = s.createShimLogFifoSymlink(ctx)
+	err = s.vmDir().CreateShimLogFifoLink(containerID)
 	if err != nil {
 		return "", err
 	}
 
-	err = shim.SetScore(cmd.Process.Pid)
+	return shimSocketAddress, nil
+}
+
+func logPanicAndDie(logger *logrus.Entry) {
+	if err := recover(); err != nil {
+		logger.WithError(err.(error)).Fatalf("panic: %s", string(debug.Stack()))
+	}
+}
+
+func parseCreateTaskOpts(ctx context.Context, opts *ptypes.Any) (*proto.FirecrackerConfig, *ptypes.Any, error) {
+	cfg, err := typeurl.UnmarshalAny(opts)
 	if err != nil {
-		log.G(ctx).WithError(err).WithField("id", id).Error("failed to set OOM score on shim")
-		return "", errors.Wrap(err, "failed to set OOM Score on shim")
+		return nil, nil, errors.Wrapf(err, "unmarshaling task create request options")
+	}
+	// We've verified that this is a valid prototype at this point of time.
+	// Check if it's the FirecrackerConfig type
+	firecrackerConfig, ok := cfg.(*proto.FirecrackerConfig)
+	if ok {
+		// We've verified that the proto message was of FirecrackerConfig type.
+		// Get runc options based on what was set in FirecrackerConfig.
+		return firecrackerConfig, firecrackerConfig.RuncOptions, nil
 	}
+	// This is a valid proto message, but is not FirecrackerConfig type.
+	// Treat the message as runc opts
+	return nil, opts, nil
+}
 
-	return shimSocketAddress, nil
+func (s *service) generateExtraData(bundleDir bundle.Dir, options *ptypes.Any) (*proto.ExtraData, error) {
+	// Add the bundle/config.json to the request so it can be recreated
+	// inside the vm:
+	// Read bundle json
+	jsonBytes, err := bundleDir.OCIConfig().Bytes()
+	if err != nil {
+		return nil, err
+	}
+
+	var opts *ptypes.Any
+	if options != nil {
+		// Copy values of existing options over
+		valCopy := make([]byte, len(options.Value))
+		copy(valCopy, options.Value)
+		opts = &ptypes.Any{
+			TypeUrl: options.TypeUrl,
+			Value:   valCopy,
+		}
+	}
+
+	return &proto.ExtraData{
+		JsonSpec:    jsonBytes,
+		RuncOptions: opts,
+		StdinPort:   s.nextVSockPort(),
+		StdoutPort:  s.nextVSockPort(),
+		StderrPort:  s.nextVSockPort(),
+	}, nil
+}
+
+// assumes caller has s.startVMMutex
+func (s *service) nextVSockPort() uint32 {
+	s.vsockPortMu.Lock()
+	defer s.vsockPortMu.Unlock()
+
+	port := minVsockIOPort + s.vsockIOPortCount
+	if port == math.MaxUint32 {
+		// given we use 3 ports per container, there would need to
+		// be about 1431652098 containers spawned in this VM for
+		// this to actually happen in practice.
+		panic("overflow of vsock ports")
+	}
+
+	s.vsockIOPortCount++
+	return port
 }
 
 func (s *service) Create(ctx context.Context, request *taskAPI.CreateTaskRequest) (*taskAPI.CreateTaskResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{
 		"id":         request.ID,
 		"bundle":     request.Bundle,
@@ -448,8 +431,8 @@ func (s *service) Create(ctx context.Context, request *taskAPI.CreateTaskRequest
 		"checkpoint": request.Checkpoint,
 	}).Debug("creating task")
 
-	var firecrackerConfig *proto.FirecrackerConfig
 	var err error
+	var firecrackerConfig *proto.FirecrackerConfig
 	var runcOpts *ptypes.Any
 
 	if request.Options != nil {
@@ -464,7 +447,6 @@ func (s *service) Create(ctx context.Context, request *taskAPI.CreateTaskRequest
 		}
 	}
 
-	// TODO: handle case where VM is pre-created
 	s.startVMMutex.Lock()
 	if !s.agentStarted {
 		log.G(ctx).WithField("id", request.ID).Debug("calling startVM")
@@ -482,91 +464,102 @@ func (s *service) Create(ctx context.Context, request *taskAPI.CreateTaskRequest
 
 	log.G(ctx).WithField("id", request.ID).Info("creating task")
 
-	// Generate new anyData with bundle/config.json packed inside
-	anyData, err := packBundle(filepath.Join(request.Bundle, "config.json"), runcOpts)
+	bundleDir := bundle.Dir(request.Bundle)
+	extraData, err := s.generateExtraData(bundleDir, runcOpts)
+	if err != nil {
+		return nil, err
+	}
+
+	task, err := s.taskManager.AddTask(ctx, request.ID, s.agentClient, bundleDir, extraData, cio.NewFIFOSet(cio.Config{
+		Stdin:    request.Stdin,
+		Stdout:   request.Stdout,
+		Stderr:   request.Stderr,
+		Terminal: request.Terminal,
+	}, nil))
 	if err != nil {
 		return nil, err
 	}
 
-	request.Options = anyData
+	defer func() {
+		if err != nil {
+			s.taskManager.Remove(ctx, request.ID)
+		}
+	}()
+
+	// Begin initializing stdio, but don't block on the initialization so we can send the Create
+	// call (which will allow the stdio initialization to complete)
+	stdioReadyCh := task.StartStdioProxy(ctx, vm.FIFOtoVSock, cidDialer(s.machineCID))
 
-	resp, err := s.agentClient.Create(ctx, request)
+	// Override the original request options with ExtraData needed by the VM Agent before sending it off
+	request.Options, err = task.MarshalExtraData()
 	if err != nil {
-		log.G(ctx).WithError(err).Error("create failed")
 		return nil, err
 	}
-	s.ctx, s.cancel = context.WithCancel(ctx)
-	s.proxyStdio(s.ctx, request.Stdin, request.Stdout, request.Stderr, s.machineCID)
-	log.G(ctx).WithField("id", request.ID).WithField("pid", resp.Pid).Info("successfully created task")
 
-	return resp, nil
-}
+	resp, err := task.Create(ctx, request)
+	if err != nil {
+		log.G(ctx).WithError(err).Error("create failed")
+		return nil, err
+	}
 
-func (s *service) Start(ctx context.Context, req *taskAPI.StartRequest) (*taskAPI.StartResponse, error) {
-	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("start")
-	resp, err := s.agentClient.Start(ctx, req)
+	// make sure stdio was initialized successfully
+	err = <-stdioReadyCh
 	if err != nil {
 		return nil, err
 	}
 
+	log.G(ctx).WithField("id", request.ID).WithField("pid", resp.Pid).Info("successfully created task")
 	return resp, nil
 }
 
-func (s *service) proxyStdio(ctx context.Context, stdin, stdout, stderr string, cid uint32) {
-	go proxyIO(ctx, stdin, cid, internal.StdinPort, true)
-	go proxyIO(ctx, stdout, cid, internal.StdoutPort, false)
-	go proxyIO(ctx, stderr, cid, internal.StderrPort, false)
-}
+func (s *service) Start(ctx context.Context, req *taskAPI.StartRequest) (*taskAPI.StartResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
 
-func proxyIO(ctx context.Context, path string, cid, port uint32, in bool) {
-	if path == "" {
-		log.G(ctx).WithField("cid", cid).WithField("port", port).Warn("skipping IO, path is empty")
-		return
-	}
-	log.G(ctx).WithField("path", path).WithField("cid", cid).WithField("port", port).Debug("setting up IO")
-	f, err := fifo.OpenFifo(ctx, path, syscall.O_RDWR|syscall.O_NONBLOCK, 0700)
-	if err != nil {
-		log.G(ctx).WithError(err).WithField("path", path).Error("error opening fifo")
-		return
-	}
-	conn, err := vsock.Dial(cid, port)
+	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("start")
+	task, err := s.taskManager.Task(req.ID)
 	if err != nil {
-		log.G(ctx).WithError(err).WithField("path", path).Error("unable to dial agent vsock IO port")
-		f.Close()
-		return
-	}
-	go func() {
-		<-ctx.Done()
-		conn.Close()
-		f.Close()
-	}()
-	log.G(ctx).WithField("path", path).Debug("begin copying io")
-	buf := make([]byte, internal.DefaultBufferSize)
-	if in {
-		_, err = io.CopyBuffer(conn, f, buf)
-	} else {
-		_, err = io.CopyBuffer(f, conn, buf)
+		return nil, err
 	}
+
+	resp, err := task.Start(ctx, req)
 	if err != nil {
-		log.G(ctx).WithError(err).WithField("path", path).Error("error with stdio")
+		return nil, err
 	}
+
+	return resp, nil
 }
 
-// Delete the initial process and container
 func (s *service) Delete(ctx context.Context, req *taskAPI.DeleteRequest) (*taskAPI.DeleteResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("delete")
-	resp, err := s.agentClient.Delete(ctx, req)
+
+	task, err := s.taskManager.Task(req.ID)
 	if err != nil {
 		return nil, err
 	}
 
+	resp, err := task.Delete(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	s.taskManager.Remove(ctx, req.ID)
+
 	return resp, nil
 }
 
 // Exec an additional process inside the container
 func (s *service) Exec(ctx context.Context, req *taskAPI.ExecProcessRequest) (*ptypes.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("exec")
-	resp, err := s.agentClient.Exec(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.Exec(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -576,8 +569,15 @@ func (s *service) Exec(ctx context.Context, req *taskAPI.ExecProcessRequest) (*p
 
 // ResizePty of a process
 func (s *service) ResizePty(ctx context.Context, req *taskAPI.ResizePtyRequest) (*ptypes.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("resize_pty")
-	resp, err := s.agentClient.ResizePty(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.ResizePty(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -587,8 +587,15 @@ func (s *service) ResizePty(ctx context.Context, req *taskAPI.ResizePtyRequest)
 
 // State returns runtime state information for a process
 func (s *service) State(ctx context.Context, req *taskAPI.StateRequest) (*taskAPI.StateResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("state")
-	resp, err := s.agentClient.State(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.State(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -598,8 +605,15 @@ func (s *service) State(ctx context.Context, req *taskAPI.StateRequest) (*taskAP
 
 // Pause the container
 func (s *service) Pause(ctx context.Context, req *taskAPI.PauseRequest) (*ptypes.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("pause")
-	resp, err := s.agentClient.Pause(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.Pause(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -609,8 +623,15 @@ func (s *service) Pause(ctx context.Context, req *taskAPI.PauseRequest) (*ptypes
 
 // Resume the container
 func (s *service) Resume(ctx context.Context, req *taskAPI.ResumeRequest) (*ptypes.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("resume")
-	resp, err := s.agentClient.Resume(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.Resume(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -620,8 +641,15 @@ func (s *service) Resume(ctx context.Context, req *taskAPI.ResumeRequest) (*ptyp
 
 // Kill a process with the provided signal
 func (s *service) Kill(ctx context.Context, req *taskAPI.KillRequest) (*ptypes.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("kill")
-	resp, err := s.agentClient.Kill(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.Kill(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -630,8 +658,15 @@ func (s *service) Kill(ctx context.Context, req *taskAPI.KillRequest) (*ptypes.E
 
 // Pids returns all pids inside the container
 func (s *service) Pids(ctx context.Context, req *taskAPI.PidsRequest) (*taskAPI.PidsResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("pids")
-	resp, err := s.agentClient.Pids(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.Pids(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -641,8 +676,15 @@ func (s *service) Pids(ctx context.Context, req *taskAPI.PidsRequest) (*taskAPI.
 
 // CloseIO of a process
 func (s *service) CloseIO(ctx context.Context, req *taskAPI.CloseIORequest) (*ptypes.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("close_io")
-	resp, err := s.agentClient.CloseIO(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.CloseIO(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -652,8 +694,15 @@ func (s *service) CloseIO(ctx context.Context, req *taskAPI.CloseIORequest) (*pt
 
 // Checkpoint the container
 func (s *service) Checkpoint(ctx context.Context, req *taskAPI.CheckpointTaskRequest) (*ptypes.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "path": req.Path}).Info("checkpoint")
-	resp, err := s.agentClient.Checkpoint(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.Checkpoint(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -663,8 +712,15 @@ func (s *service) Checkpoint(ctx context.Context, req *taskAPI.CheckpointTaskReq
 
 // Connect returns shim information such as the shim's pid
 func (s *service) Connect(ctx context.Context, req *taskAPI.ConnectRequest) (*taskAPI.ConnectResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("connect")
-	resp, err := s.agentClient.Connect(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.Connect(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -673,9 +729,18 @@ func (s *service) Connect(ctx context.Context, req *taskAPI.ConnectRequest) (*ta
 }
 
 func (s *service) Shutdown(ctx context.Context, req *taskAPI.ShutdownRequest) (*ptypes.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "now": req.Now}).Debug("shutdown")
-	if _, err := s.agentClient.Shutdown(ctx, req); err != nil {
-		log.G(ctx).WithError(err).Error("failed to shutdown agent")
+
+	// If we are still managing containers, don't shutdown
+	if s.taskManager.TaskCount() > 0 {
+		return &ptypes.Empty{}, nil
+	}
+
+	_, err := s.agentClient.Shutdown(ctx, req)
+	if err != nil {
+		return nil, err
 	}
 
 	log.G(ctx).Debug("stopping VM")
@@ -684,12 +749,11 @@ func (s *service) Shutdown(ctx context.Context, req *taskAPI.ShutdownRequest) (*
 		return nil, err
 	}
 
-	if err := os.RemoveAll(s.vmDir()); err != nil {
-		log.G(ctx).WithError(err).Errorf("failed to remove VM dir %s", s.vmDir())
-		return nil, err
+	err = os.RemoveAll(s.vmDir().RootPath())
+	if err != nil {
+		log.G(ctx).WithField("path", s.vmDir().RootPath()).WithError(err).Error("failed to remove VM dir during shutdown")
 	}
 
-	s.cancel()
 	// Exit to avoid 'zombie' shim processes
 	defer os.Exit(0)
 	log.G(ctx).Debug("stopping runtime")
@@ -697,8 +761,15 @@ func (s *service) Shutdown(ctx context.Context, req *taskAPI.ShutdownRequest) (*
 }
 
 func (s *service) Stats(ctx context.Context, req *taskAPI.StatsRequest) (*taskAPI.StatsResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("stats")
-	resp, err := s.agentClient.Stats(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.Stats(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -708,8 +779,15 @@ func (s *service) Stats(ctx context.Context, req *taskAPI.StatsRequest) (*taskAP
 
 // Update a running container
 func (s *service) Update(ctx context.Context, req *taskAPI.UpdateTaskRequest) (*ptypes.Empty, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithField("id", req.ID).Debug("update")
-	resp, err := s.agentClient.Update(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.Update(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -719,8 +797,15 @@ func (s *service) Update(ctx context.Context, req *taskAPI.UpdateTaskRequest) (*
 
 // Wait for a process to exit
 func (s *service) Wait(ctx context.Context, req *taskAPI.WaitRequest) (*taskAPI.WaitResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).WithFields(logrus.Fields{"id": req.ID, "exec_id": req.ExecID}).Debug("wait")
-	resp, err := s.agentClient.Wait(ctx, req)
+	task, err := s.taskManager.Task(req.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := task.Wait(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -729,6 +814,8 @@ func (s *service) Wait(ctx context.Context, req *taskAPI.WaitRequest) (*taskAPI.
 }
 
 func (s *service) Cleanup(ctx context.Context) (*taskAPI.DeleteResponse, error) {
+	defer logPanicAndDie(log.G(ctx))
+
 	log.G(ctx).Debug("cleanup")
 	// Destroy VM/etc here?
 	// copied from runcs impl, nothing to cleanup atm
@@ -738,55 +825,14 @@ func (s *service) Cleanup(ctx context.Context) (*taskAPI.DeleteResponse, error)
 	}, nil
 }
 
-func (s *service) newCommand(ctx context.Context, id, containerdBinary, containerdAddress string, shimSocketFile *os.File) (*exec.Cmd, error) {
-	ns, err := namespaces.NamespaceRequired(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	self, err := os.Executable()
-	if err != nil {
-		return nil, err
-	}
-
-	bundleDir, err := resolveBundleDir(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	args := []string{
-		"-namespace", ns,
-		"-bundle", bundleDir,
-		"-id", id,
-		"-address", containerdAddress,
-		"-publish-binary", containerdBinary,
-	}
-
-	if s.config.Debug {
-		args = append(args, "-debug")
-	}
-
-	cmd := exec.Command(self, args...)
-	cmd.Dir = s.vmDir()
-	cmd.Env = append(os.Environ(),
-		"GOMAXPROCS=2",
-		fmt.Sprintf("%s=%s", vmIDEnvVarKey, s.vmID))
-	cmd.ExtraFiles = append(cmd.ExtraFiles, shimSocketFile)
-	cmd.SysProcAttr = &syscall.SysProcAttr{
-		Setpgid: true,
-	}
-
-	return cmd, nil
-}
-
 func newServer() (*ttrpc.Server, error) {
 	return ttrpc.NewServer(ttrpc.WithServerHandshaker(ttrpc.UnixSocketRequireSameUser()))
 }
 
-func dialVsock(ctx context.Context, contextID uint32, port uint32) (net.Conn, error) {
-	// VM should start within 200ms, vsock dial will make retries at 100ms, 200ms, 400ms, 800ms and 1.6s
+func dialVsock(ctx context.Context, contextID, port uint32) (net.Conn, error) {
+	// VM should start within 200ms, vsock dial will make retries at 100ms, 200ms, 400ms, 800ms, 1.6s, 3.2s, 6.4s
 	const (
-		retryCount      = 5
+		retryCount      = 7
 		initialDelay    = 100 * time.Millisecond
 		delayMultiplier = 2
 	)
@@ -811,6 +857,12 @@ func dialVsock(ctx context.Context, contextID uint32, port uint32) (net.Conn, er
 	return nil, lastErr
 }
 
+func cidDialer(cid uint32) vm.VSockConnector {
+	return func(ctx context.Context, port uint32) (net.Conn, error) {
+		return dialVsock(ctx, cid, port)
+	}
+}
+
 // findNextAvailableVsockCID finds first available vsock context ID.
 // It uses VHOST_VSOCK_SET_GUEST_CID ioctl which allows some CID ranges to be statically reserved in advance.
 // The ioctl fails with EADDRINUSE if cid is already taken and with EINVAL if the CID is invalid.
@@ -863,24 +915,6 @@ func findNextAvailableVsockCID(ctx context.Context) (uint32, error) {
 	return 0, errors.New("couldn't find any available vsock context id")
 }
 
-func parseCreateTaskOpts(ctx context.Context, opts *ptypes.Any) (*proto.FirecrackerConfig, *ptypes.Any, error) {
-	cfg, err := typeurl.UnmarshalAny(opts)
-	if err != nil {
-		return nil, nil, errors.Wrapf(err, "unmarshaling task create request options")
-	}
-	// We've verified that this is a valid prototype at this point of time.
-	// Check if it's the FirecrackerConfig type
-	firecrackerConfig, ok := cfg.(*proto.FirecrackerConfig)
-	if ok {
-		// We've verified that the proto message was of FirecrackerConfig type.
-		// Get runc options based on what was set in FirecrackerConfig.
-		return firecrackerConfig, firecrackerConfig.RuncOptions, nil
-	}
-	// This is a valid proto message, but is not FirecrackerConfig type.
-	// Treat the message as runc opts
-	return nil, opts, nil
-}
-
 // TODO: replace startVM with calls to the FC-control plugin
 func (s *service) startVM(ctx context.Context,
 	request *taskAPI.CreateTaskRequest,
@@ -894,7 +928,7 @@ func (s *service) startVM(ctx context.Context,
 	}
 
 	cfg := firecracker.Config{
-		SocketPath:      s.firecrackerSockPath(),
+		SocketPath:      s.vmDir().FirecrackerSockPath(),
 		VsockDevices:    []firecracker.VsockDevice{{Path: "root", CID: cid}},
 		KernelImagePath: s.config.KernelImagePath,
 		KernelArgs:      s.config.KernelArgs,
@@ -903,9 +937,9 @@ func (s *service) startVM(ctx context.Context,
 			CPUTemplate: models.CPUTemplate(s.config.CPUTemplate),
 			MemSizeMib:  256,
 		},
-		LogFifo:     s.firecrackerLogFifoPath(),
+		LogFifo:     s.vmDir().FirecrackerLogFifoPath(),
 		LogLevel:    s.config.LogLevel,
-		MetricsFifo: s.firecrackerMetricsFifoPath(),
+		MetricsFifo: s.vmDir().FirecrackerMetricsFifoPath(),
 		Debug:       s.config.Debug,
 	}
 
@@ -931,7 +965,7 @@ func (s *service) startVM(ctx context.Context,
 	cfg.Drives = driveBuilder.Build()
 	cmd := firecracker.VMCommandBuilder{}.
 		WithBin(s.config.FirecrackerBinaryPath).
-		WithSocketPath(s.firecrackerSockPath()).
+		WithSocketPath(s.vmDir().FirecrackerSockPath()).
 		Build(ctx)
 	machineOpts := []firecracker.Opt{
 		firecracker.WithProcessRunner(cmd),
@@ -965,8 +999,9 @@ func (s *service) startVM(ctx context.Context,
 	eventBridgeClient := eventbridge.NewGetterClient(rpcClient)
 
 	go func() {
-		// Connect the agent's event exchange to our own own event exchange using the eventbridge client. All events
-		// that are published on the agent's exchange will also be published on our own
+		// Connect the agent's event exchange to our own own event exchange
+		// using the eventbridge client. All events that are published on the
+		// agent's exchange will also be published on our own
 		if err := <-eventbridge.Attach(ctx, eventBridgeClient, s.eventExchange); err != nil && err != context.Canceled {
 			log.G(ctx).WithError(err).Error("error while forwarding events from VM agent")
 		}
@@ -977,55 +1012,52 @@ func (s *service) startVM(ctx context.Context,
 	return apiClient, nil
 }
 
+func (s *service) stopVM() error {
+	return s.machine.StopVMM()
+}
+
 func (s *service) monitorTaskExit(ctx context.Context) {
+	logger := log.G(ctx)
 	exitEvents, exitEventErrs := s.eventExchange.Subscribe(ctx, fmt.Sprintf(`topic=="%s"`, runtime.TaskExitEventTopic))
 
 	var err error
 	defer func() {
 		if err != nil && err != context.Canceled {
-			log.G(ctx).WithError(err).Error("error while waiting for task exit events")
+			logger.WithError(err).Error("error while waiting for task exit events")
 		}
 	}()
 
-	select {
-	case <-exitEvents:
-		// If the task exits, we shut down the VM and exit this shim. This behavior may change in the future when
-		// we support multiple containers per VM.
-		s.Shutdown(ctx, &taskAPI.ShutdownRequest{ID: s.id})
-	case err = <-exitEventErrs:
-	case <-ctx.Done():
-		err = ctx.Err()
-	}
-}
+	for {
+		select {
+		case envelope := <-exitEvents:
+			unmarshaledEvent, err := typeurl.UnmarshalAny(envelope.Event)
+			if err != nil {
+				logger.WithError(err).Error("error unmarshaling event")
+				continue
+			}
 
-func (s *service) stopVM() error {
-	return s.machine.StopVMM()
-}
+			switch event := unmarshaledEvent.(type) {
+			case *eventsAPI.TaskExit:
+				s.taskManager.Remove(ctx, event.ContainerID)
 
-func packBundle(path string, options *ptypes.Any) (*ptypes.Any, error) {
-	// Add the bundle/config.json to the request so it can be recreated
-	// inside the vm:
-	// Read bundle json
-	jsonBytes, err := ioutil.ReadFile(path)
-	if err != nil {
-		return nil, err
-	}
+				// If we have no more containers, shutdown. If we still have containers left,
+				// this will be a no-op
+				_, err = s.Shutdown(ctx, &taskAPI.ShutdownRequest{})
+				if err != nil {
+					logger.WithError(err).WithField("id", event.ContainerID).Fatal("failed to shutdown after container exit")
+				}
+
+			default:
+				logger.Error("unexpected non-exit event type published on exit event channel")
+			}
+
+		case err = <-exitEventErrs:
+			logger.WithError(err).Error("event error channel published to")
+
+		case <-ctx.Done():
+			err = ctx.Err()
+			return
 
-	var opts *ptypes.Any
-	if options != nil {
-		// Copy values of existing options over
-		valCopy := make([]byte, len(options.Value))
-		copy(valCopy, options.Value)
-		opts = &ptypes.Any{
-			TypeUrl: options.TypeUrl,
-			Value:   valCopy,
 		}
 	}
-	// add it to a type
-	// Convert to any
-	extraData := &proto.ExtraData{
-		JsonSpec:    jsonBytes,
-		RuncOptions: opts,
-	}
-	return ptypes.MarshalAny(extraData)
 }
diff --git a/tools/docker/fc-agent.start b/tools/docker/fc-agent.start
index 9ecfbc7..cb01321 100755
--- a/tools/docker/fc-agent.start
+++ b/tools/docker/fc-agent.start
@@ -3,7 +3,7 @@ mkdir -p /container
 exec > /container/agent-debug.log # Debug logs from the agent
 exec 2>&1
 touch /container/runtime
-mkdir /container/rootfs
-mount -t auto -o rw /dev/vdb /container/rootfs
+
 cd /container
-/usr/local/bin/agent -id 1 -debug &
+
+exec /usr/local/bin/agent -id 1 -debug


Test patch:
diff --git a/agent/service_test.go b/internal/bundle/doc.go
similarity index 75%
rename from agent/service_test.go
rename to internal/bundle/doc.go
index 3583798..9cc3694 100644
--- a/agent/service_test.go
+++ b/internal/bundle/doc.go
@@ -11,20 +11,8 @@
 // express or implied. See the License for the specific language governing
 // permissions and limitations under the License.
 
-package main
-
-import (
-	"context"
-	"testing"
-)
-
-func TestCancelAll(t *testing.T) {
-	ts := &TaskService{
-		cancels: []context.CancelFunc{
-			func() {},
-			func() {},
-		},
-	}
-
-	ts.cancelAll()
-}
+/*
+Package bundle implements some helper functions for firecracker-containerd's interaction
+with bundle dirs, both inside the VM and outside on the host.
+*/
+package bundle
diff --git a/internal/vm/task_test.go b/internal/vm/task_test.go
new file mode 100644
index 0000000..e843a9a
--- /dev/null
+++ b/internal/vm/task_test.go
@@ -0,0 +1,318 @@
+// Copyright 2018-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may
+// not use this file except in compliance with the License. A copy of the
+// License is located at
+//
+//	http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file is distributed
+// on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+// express or implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package vm
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/containerd/containerd/cio"
+	"github.com/containerd/containerd/runtime/v2/task"
+	"github.com/firecracker-microvm/firecracker-containerd/internal/bundle"
+	proto "github.com/firecracker-microvm/firecracker-containerd/proto/grpc"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	mockBundleDir = bundle.Dir("/tmp")
+
+	mockExtraData = &proto.ExtraData{
+		JsonSpec:    []byte{},
+		RuncOptions: nil,
+		StdinPort:   1,
+		StdoutPort:  2,
+		StderrPort:  3,
+	}
+
+	mockFIFOSet = cio.NewFIFOSet(cio.Config{
+		Stdin:  "/dev/null",
+		Stdout: "/dev/null",
+		Stderr: "/dev/null",
+	}, nil)
+)
+
+type mockTaskService struct {
+	task.TaskService
+}
+
+func defaultMockTaskArgs(id string) addTaskArgs {
+	return addTaskArgs{
+		id:        fmt.Sprintf("container-%s", id),
+		ts:        &mockTaskService{},
+		bundleDir: mockBundleDir,
+		extraData: mockExtraData,
+		fifoSet:   mockFIFOSet,
+	}
+}
+
+type addTaskArgs struct {
+	id        string
+	ts        task.TaskService
+	bundleDir bundle.Dir
+	extraData *proto.ExtraData
+	fifoSet   *cio.FIFOSet
+}
+
+func addTaskFromArgs(tm TaskManager, args addTaskArgs) (*Task, error) {
+	return tm.AddTask(context.TODO(), args.id, args.ts, args.bundleDir, args.extraData, args.fifoSet)
+}
+
+func TestTaskManager_AddRemoveTask(t *testing.T) {
+	tm := NewTaskManager()
+
+	taskAArgs := defaultMockTaskArgs("A")
+	taskBArgs := defaultMockTaskArgs("B")
+	taskCArgs := defaultMockTaskArgs("C")
+
+	assertExpectedTask := func(args addTaskArgs, createdTask *Task) {
+		require.Equal(t, args.ts, createdTask.TaskService, "AddTask sets expected task service")
+		require.Equal(t, args.id, createdTask.ID, "AddTask sets expected container ID")
+		require.Equal(t, args.extraData, createdTask.extraData, "AddTask sets expected container ID")
+		require.Equal(t, args.bundleDir, createdTask.bundleDir, "AddTask sets expected container ID")
+		require.Equal(t, args.fifoSet, createdTask.fifoSet, "AddTask sets expected container ID")
+	}
+
+	// Basic adding of tasks
+	taskA, err := addTaskFromArgs(tm, taskAArgs)
+	require.NoError(t, err, "add first task")
+	assertExpectedTask(taskAArgs, taskA)
+
+	taskB, err := addTaskFromArgs(tm, taskBArgs)
+	require.NoError(t, err, "add second task")
+	assertExpectedTask(taskBArgs, taskB)
+
+	taskC, err := addTaskFromArgs(tm, taskCArgs)
+	require.NoError(t, err, "add third task")
+	assertExpectedTask(taskCArgs, taskC)
+
+	_, err = addTaskFromArgs(tm, taskAArgs)
+	require.Error(t, err, "add duplicate task should fail")
+
+	require.EqualValues(t, 3, tm.TaskCount(), "task count should return correct number of tasks")
+
+	// We should be able to get return the tasks we added
+	supposedlyTaskA, err := tm.Task(taskAArgs.id)
+	require.NoError(t, err, "get task")
+	require.Exactly(t, taskA, supposedlyTaskA, "get should return expected task")
+
+	supposedlyTaskB, err := tm.Task(taskBArgs.id)
+	require.NoError(t, err, "get task")
+	require.Exactly(t, taskB, supposedlyTaskB, "get should return expected task")
+
+	supposedlyTaskC, err := tm.Task(taskCArgs.id)
+	require.NoError(t, err, "get task")
+	require.Exactly(t, taskC, supposedlyTaskC, "get should return expected task")
+
+	_, err = tm.Task("never-existed")
+	require.Error(t, err, "get non-existent task should fail")
+
+	// Removing a task should remove the task requested but have no effect on the others
+	tm.Remove(context.TODO(), taskBArgs.id)
+	require.EqualValues(t, 2, tm.TaskCount(), "task count should return correct number of tasks after remove")
+
+	supposedlyTaskA, err = tm.Task(taskAArgs.id)
+	require.NoError(t, err, "get task")
+	require.Exactly(t, taskA, supposedlyTaskA, "get should return expected task")
+
+	_, err = tm.Task(taskBArgs.id)
+	require.Error(t, err, "get removed task should fail")
+
+	supposedlyTaskC, err = tm.Task(taskCArgs.id)
+	require.NoError(t, err, "get task")
+	require.Exactly(t, taskC, supposedlyTaskC, "get should return expected task")
+
+	// Remove all should remove all the tasks
+	tm.RemoveAll(context.TODO())
+	require.EqualValues(t, 0, tm.TaskCount(), "task count should return correct number of tasks after remove all")
+
+	_, err = tm.Task(taskAArgs.id)
+	require.Error(t, err, "get removed task should fail")
+
+	_, err = tm.Task(taskCArgs.id)
+	require.Error(t, err, "get removed task should fail")
+
+	tm.RemoveAll(context.TODO())
+	require.EqualValues(t, 0, tm.TaskCount(), "remove all on empty task manager should have no effect")
+	tm.Remove(context.TODO(), taskAArgs.id)
+	require.EqualValues(t, 0, tm.TaskCount(), "remove on non-existent task should have no effect")
+
+	// Tasks should still be able to be added after removes
+	taskC, err = addTaskFromArgs(tm, taskCArgs)
+	require.NoError(t, err, "re-add task after deletion")
+	assertExpectedTask(taskCArgs, taskC)
+
+	supposedlyTaskC, err = tm.Task(taskCArgs.id)
+	require.NoError(t, err, "get re-added task")
+	require.Exactly(t, taskC, supposedlyTaskC, "get should return expected task")
+
+	require.EqualValues(t, 1, tm.TaskCount(), "re-add task should set task count")
+}
+
+func pathOfFile(f *os.File) string {
+	return fmt.Sprintf("/proc/self/fd/%d", f.Fd())
+}
+
+func TestTaskManager_StartStdioProxy_FIFOtoVSock(t *testing.T) {
+	testStdioProxy(t, FIFOtoVSock)
+}
+
+func TestTaskManager_StartStdioProxy_VSockToFIFO(t *testing.T) {
+	testStdioProxy(t, VSockToFIFO)
+}
+
+func testStdioProxy(t *testing.T, inputDirection IODirection) {
+	tm := NewTaskManager()
+
+	taskArgs := defaultMockTaskArgs("A")
+
+	// Use pipes to serve as fake FIFOs. We provide the io proxy
+	// paths to their /proc/self/fd
+	stdinR, stdinW, err := os.Pipe()
+	require.NoError(t, err, "get os pipe FDs")
+
+	stdoutR, stdoutW, err := os.Pipe()
+	require.NoError(t, err, "get os pipe FDs")
+
+	stderrR, stderrW, err := os.Pipe()
+	require.NoError(t, err, "get os pipe FDs")
+
+	switch inputDirection {
+	case FIFOtoVSock:
+		taskArgs.fifoSet = cio.NewFIFOSet(cio.Config{
+			Stdin:  pathOfFile(stdinR),
+			Stdout: pathOfFile(stdoutW),
+			Stderr: pathOfFile(stderrW),
+		}, nil)
+
+	case VSockToFIFO:
+		taskArgs.fifoSet = cio.NewFIFOSet(cio.Config{
+			Stdin:  pathOfFile(stdinW),
+			Stdout: pathOfFile(stdoutR),
+			Stderr: pathOfFile(stderrR),
+		}, nil)
+	}
+
+	// Use net.Conn as fake vsocks. *Local represents the side the io proxy
+	// will interact with. We will fake remote data by reading/writing from/to
+	// the *Remote conns in the test case code itself.
+	stdinSockLocal, stdinSockRemote := net.Pipe()
+	stdoutSockLocal, stdoutSockRemote := net.Pipe()
+	stderrSockLocal, stderrSockRemote := net.Pipe()
+
+	vsockConnector := func(ctx context.Context, port uint32) (net.Conn, error) {
+		switch port {
+		case taskArgs.extraData.StdinPort:
+			return stdinSockLocal, nil
+		case taskArgs.extraData.StdoutPort:
+			return stdoutSockLocal, nil
+		case taskArgs.extraData.StderrPort:
+			return stderrSockLocal, nil
+		default:
+			require.Failf(t, "unexpected vsock port", "port: %d", port)
+			return nil, nil // just here to appease the compiler, Failf exits
+		}
+	}
+
+	// add the task and start the io proxy
+	task, err := addTaskFromArgs(tm, taskArgs)
+	require.NoError(t, err, "add first task")
+
+	readyCh := task.StartStdioProxy(context.TODO(), inputDirection, vsockConnector)
+	select {
+	case err := <-readyCh:
+		require.NoError(t, err, "initialize stdio")
+	case <-time.After(10 * time.Second):
+		require.Fail(t, "timed out waiting for stdio initialization to finish")
+	}
+
+	// stdin
+	var stdinWriter io.WriteCloser
+	var stdinReader io.ReadCloser
+	switch inputDirection {
+	case FIFOtoVSock:
+		stdinWriter = stdinW
+		stdinReader = stdinSockRemote
+	case VSockToFIFO:
+		stdinWriter = stdinSockRemote
+		stdinReader = stdinR
+	}
+
+	writtenStdinBytes := []byte("stdin")
+	readStdinBytes := make([]byte, len(writtenStdinBytes))
+
+	_, err = stdinWriter.Write(writtenStdinBytes)
+	require.NoError(t, err, "write to stdin")
+
+	err = stdinWriter.Close()
+	require.NoError(t, err, "close stdin socket")
+
+	_, err = stdinReader.Read(readStdinBytes)
+	require.NoError(t, err, "read stdout")
+	require.Equal(t, writtenStdinBytes, readStdinBytes, "stdin proxied transparently")
+
+	// stdout
+	var stdoutWriter io.WriteCloser
+	var stdoutReader io.ReadCloser
+	switch inputDirection {
+	case FIFOtoVSock:
+		stdoutWriter = stdoutSockRemote
+		stdoutReader = stdoutR
+	case VSockToFIFO:
+		stdoutWriter = stdoutW
+		stdoutReader = stdoutSockRemote
+	}
+
+	writtenStdoutBytes := []byte("stdout")
+	readStdoutBytes := make([]byte, len(writtenStdoutBytes))
+
+	stdoutWriter.Write(writtenStdoutBytes)
+	require.NoError(t, err, "write stdout")
+
+	err = stdoutWriter.Close()
+	require.NoError(t, err, "close stdout socket")
+
+	_, err = stdoutReader.Read(readStdoutBytes)
+	require.NoError(t, err, "read stdout")
+	require.Equal(t, writtenStdoutBytes, readStdoutBytes, "stdout proxied transparently")
+
+	// stderr
+	var stderrWriter io.WriteCloser
+	var stderrReader io.ReadCloser
+	switch inputDirection {
+	case FIFOtoVSock:
+		stderrWriter = stderrSockRemote
+		stderrReader = stderrR
+	case VSockToFIFO:
+		stderrWriter = stderrW
+		stderrReader = stderrSockRemote
+	}
+
+	writtenStderrBytes := []byte("stderr")
+	readStderrBytes := make([]byte, len(writtenStderrBytes))
+
+	stderrWriter.Write(writtenStderrBytes)
+	require.NoError(t, err, "write stderr")
+
+	err = stderrWriter.Close()
+	require.NoError(t, err, "close stderr socket")
+
+	_, err = stderrReader.Read(readStderrBytes)
+	require.NoError(t, err, "read stderr")
+	require.Equal(t, writtenStderrBytes, readStderrBytes, "stderr proxied transparently")
+}
diff --git a/runtime/service_integ_test.go b/runtime/service_integ_test.go
index c905203..bf1ee2d 100644
--- a/runtime/service_integ_test.go
+++ b/runtime/service_integ_test.go
@@ -60,7 +60,7 @@ const (
 func TestShimExitsUponContainerKill_Isolated(t *testing.T) {
 	internal.RequiresIsolation(t)
 
-	ctx := namespaces.WithNamespace(context.Background(), "default")
+	ctx := namespaces.WithNamespace(context.Background(), namespaces.Default)
 
 	client, err := containerd.New(containerdSockPath)
 	require.NoError(t, err, "unable to create client to containerd service at %s, is containerd running?", containerdSockPath)
@@ -130,10 +130,10 @@ func TestShimExitsUponContainerKill_Isolated(t *testing.T) {
 		err = internal.WaitForPidToExit(testCtx, time.Second, shimProcess.Pid)
 		require.NoError(t, err, "failed waiting for shim process \"%s\" to exit", shimProcessName)
 
-		varRunFCContents, err := ioutil.ReadDir(varRunDir)
-		require.NoError(t, err, `failed to list directory "%s"`, varRunDir)
-		// there should just be one file left in the directory for the naive snapshotter socket
-		require.Len(t, varRunFCContents, 1, "expect %s to be cleared after shims shutdown", varRunDir)
+		namespaceVarRunDir := filepath.Join(varRunDir, namespaces.Default)
+		varRunFCContents, err := ioutil.ReadDir(namespaceVarRunDir)
+		require.NoError(t, err, `failed to list directory "%s"`, namespaceVarRunDir)
+		require.Len(t, varRunFCContents, 0, "expect %s to be cleared after shims shutdown", namespaceVarRunDir)
 	case err = <-exitEventErrCh:
 		require.Fail(t, "unexpected error", "unexpectedly received on task exit error channel: %s", err.Error())
 	case <-testCtx.Done():
@@ -179,9 +179,8 @@ func TestMultipleVMs_Isolated(t *testing.T) {
 	internal.RequiresIsolation(t)
 
 	const (
-		numVMs = 5
-		// TODO update with multiple containers per-VM once that PR is merged
-		containersPerVM = 1
+		numVMs          = 3
+		containersPerVM = 3
 	)
 
 	ctx := namespaces.WithNamespace(context.Background(), "default")
@@ -229,7 +228,10 @@ func TestMultipleVMs_Isolated(t *testing.T) {
 						containerd.WithSnapshotter(naiveSnapshotterName),
 						containerd.WithNewSnapshot(snapshotName, image),
 						containerd.WithNewSpec(
-							oci.WithProcessArgs("cat", fmt.Sprintf("/sys/class/net/%s/address", defaultVMNetDevName)),
+							oci.WithProcessArgs("sh", "-c", fmt.Sprintf("%s && %s",
+								fmt.Sprintf("cat /sys/class/net/%s/address", defaultVMNetDevName),
+								fmt.Sprintf("sleep %d", 10),
+							)),
 							oci.WithHostNamespace(specs.NetworkNamespace),
 							firecrackeroci.WithVMID(strconv.Itoa(vmID)),
 						),
@@ -259,12 +261,12 @@ func TestMultipleVMs_Isolated(t *testing.T) {
 						})
 					require.NoError(t, err, "failed to create task for container %s", containerName)
 
-					err = newTask.Start(ctx)
-					require.NoError(t, err, "failed to start task for container %s", containerName)
-
 					exitCh, err := newTask.Wait(ctx)
 					require.NoError(t, err, "failed to wait on task for container %s", containerName)
 
+					err = newTask.Start(ctx)
+					require.NoError(t, err, "failed to start task for container %s", containerName)
+
 					select {
 					case exitStatus := <-exitCh:
 						// if there was anything on stderr, print it to assist debugging
diff --git a/runtime/service_test.go b/runtime/service_test.go
index 6e6383e..3829201 100644
--- a/runtime/service_test.go
+++ b/runtime/service_test.go
@@ -20,8 +20,8 @@ import (
 
 	"github.com/containerd/typeurl"
 	proto "github.com/firecracker-microvm/firecracker-containerd/proto/grpc"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"gotest.tools/assert"
 )
 
 const (