diff --git a/cmd/skaffold/app/cmd/runner.go b/cmd/skaffold/app/cmd/runner.go
index 1e0102f06eb..f15d6fdca15 100644
--- a/cmd/skaffold/app/cmd/runner.go
+++ b/cmd/skaffold/app/cmd/runner.go
@@ -35,6 +35,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/parser"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
+	v1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/v1"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/defaults"
 	latest_v1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/validation"
@@ -64,7 +65,7 @@ func createNewRunner(out io.Writer, opts config.SkaffoldOptions) (runner.Runner,
 	}
 
 	instrumentation.InitMeterFromConfig(configs, opts.User)
-	runner, err := runner.NewForConfig(runCtx)
+	runner, err := v1.NewForConfig(runCtx)
 	if err != nil {
 		event.InititializationFailed(err)
 		return nil, nil, nil, fmt.Errorf("creating runner: %w", err)
diff --git a/pkg/skaffold/runner/build_deploy.go b/pkg/skaffold/runner/build_deploy.go
index 4431e1da3b3..17c61177cdc 100644
--- a/pkg/skaffold/runner/build_deploy.go
+++ b/pkg/skaffold/runner/build_deploy.go
@@ -39,11 +39,22 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
 )
 
+func NewBuilder(builder build.Builder, tagger tag.Tagger, cache cache.Cache, podSelector *kubernetes.ImageList,
+	runCtx *runcontext.RunContext) *Builder {
+	return &Builder{
+		Builder:     builder,
+		tagger:      tagger,
+		cache:       cache,
+		podSelector: podSelector,
+		runCtx:      runCtx,
+	}
+}
+
 type Builder struct {
-	builder build.Builder
+	Builder build.Builder
 	tagger  tag.Tagger
 	cache   cache.Cache
-	builds  []graph.Artifact
+	Builds  []graph.Artifact
 
 	// podSelector is used to determine relevant pods for logging and portForwarding
 	podSelector *kubernetes.ImageList
@@ -52,16 +63,21 @@ type Builder struct {
 	runCtx   *runcontext.RunContext
 }
 
+// GetBuilds returns the builds value.
+func (r *Builder) GetBuilds() []graph.Artifact {
+	return r.Builds
+}
+
 // Build builds a list of artifacts.
 func (r *Builder) Build(ctx context.Context, out io.Writer, artifacts []*latest_v1.Artifact) ([]graph.Artifact, error) {
 	eventV2.TaskInProgress(constants.Build)
 
 	// Use tags directly from the Kubernetes manifests.
-	if r.runCtx.DigestSource() == noneDigestSource {
+	if r.runCtx.DigestSource() == NoneDigestSource {
 		return []graph.Artifact{}, nil
 	}
 
-	if err := checkWorkspaces(artifacts); err != nil {
+	if err := CheckWorkspaces(artifacts); err != nil {
 		eventV2.TaskFailed(constants.Build, err)
 		return nil, err
 	}
@@ -77,10 +93,10 @@ func (r *Builder) Build(ctx context.Context, out io.Writer, artifacts []*latest_
 	case r.runCtx.DryRun():
 		color.Yellow.Fprintln(out, "Skipping build phase since --dry-run=true")
 		return artifactsWithTags(tags, artifacts), nil
-	case r.runCtx.RenderOnly() && r.runCtx.DigestSource() == remoteDigestSource:
+	case r.runCtx.RenderOnly() && r.runCtx.DigestSource() == RemoteDigestSource:
 		color.Yellow.Fprintln(out, "Skipping build phase since --digest-source=remote")
 		return artifactsWithTags(tags, artifacts), nil
-	case r.runCtx.RenderOnly() && r.runCtx.DigestSource() == tagDigestSource:
+	case r.runCtx.RenderOnly() && r.runCtx.DigestSource() == TagDigestSource:
 		color.Yellow.Fprintln(out, "Skipping build phase since --digest-source=tag")
 		return artifactsWithTags(tags, artifacts), nil
 	default:
@@ -93,7 +109,7 @@ func (r *Builder) Build(ctx context.Context, out io.Writer, artifacts []*latest_
 
 		r.hasBuilt = true
 
-		bRes, err := r.builder.Build(ctx, out, tags, artifacts)
+		bRes, err := r.Builder.Build(ctx, out, tags, artifacts)
 		if err != nil {
 			return nil, err
 		}
@@ -106,10 +122,10 @@ func (r *Builder) Build(ctx context.Context, out io.Writer, artifacts []*latest_
 	}
 
 	// Update which images are logged.
-	r.addTagsToPodSelector(bRes)
+	r.AddTagsToPodSelector(bRes)
 
 	// Make sure all artifacts are redeployed. Not only those that were just built.
-	r.builds = build.MergeWithPreviousBuilds(bRes, r.builds)
+	r.Builds = build.MergeWithPreviousBuilds(bRes, r.Builds)
 
 	eventV2.TaskSucceeded(constants.Build)
 	return bRes, nil
@@ -133,7 +149,7 @@ func artifactsWithTags(tags tag.ImageTags, artifacts []*latest_v1.Artifact) []gr
 }
 
 // Update which images are logged.
-func (r *Builder) addTagsToPodSelector(artifacts []graph.Artifact) {
+func (r *Builder) AddTagsToPodSelector(artifacts []graph.Artifact) {
 	for _, artifact := range artifacts {
 		r.podSelector.Add(artifact.Tag)
 	}
@@ -161,8 +177,8 @@ func (r *Builder) imageTags(ctx context.Context, out io.Writer, artifacts []*lat
 
 		i := i
 		go func() {
-			tag, err := tag.GenerateFullyQualifiedImageName(r.tagger, *artifacts[i])
-			tagErrs[i] <- tagErr{tag: tag, err: err}
+			_tag, err := tag.GenerateFullyQualifiedImageName(r.tagger, *artifacts[i])
+			tagErrs[i] <- tagErr{tag: _tag, err: err}
 		}()
 	}
 
@@ -191,13 +207,13 @@ func (r *Builder) imageTags(ctx context.Context, out io.Writer, artifacts []*lat
 				showWarning = true
 			}
 
-			tag, err := r.ApplyDefaultRepo(t.tag)
+			_tag, err := r.ApplyDefaultRepo(t.tag)
 			if err != nil {
 				return nil, err
 			}
 
-			fmt.Fprintln(out, tag)
-			imageTags[imageName] = tag
+			fmt.Fprintln(out, _tag)
+			imageTags[imageName] = _tag
 		}
 	}
 
@@ -209,7 +225,7 @@ func (r *Builder) imageTags(ctx context.Context, out io.Writer, artifacts []*lat
 	return imageTags, nil
 }
 
-func checkWorkspaces(artifacts []*latest_v1.Artifact) error {
+func CheckWorkspaces(artifacts []*latest_v1.Artifact) error {
 	for _, a := range artifacts {
 		if a.Workspace != "" {
 			if info, err := os.Stat(a.Workspace); err != nil {
diff --git a/pkg/skaffold/runner/builder.go b/pkg/skaffold/runner/builder.go
index d5916b9498a..efc2523579d 100644
--- a/pkg/skaffold/runner/builder.go
+++ b/pkg/skaffold/runner/builder.go
@@ -45,8 +45,8 @@ func (b *builderCtx) SourceDependenciesResolver() graph.SourceDependenciesCache
 	return b.sourceDependenciesCache
 }
 
-// getBuilder creates a builder from a given RunContext and build pipeline type.
-func getBuilder(r *runcontext.RunContext, s build.ArtifactStore, d graph.SourceDependenciesCache, p latest_v1.Pipeline) (build.PipelineBuilder, error) {
+// GetBuilder creates a builder from a given RunContext and build pipeline type.
+func GetBuilder(r *runcontext.RunContext, s build.ArtifactStore, d graph.SourceDependenciesCache, p latest_v1.Pipeline) (build.PipelineBuilder, error) {
 	bCtx := &builderCtx{artifactStore: s, sourceDependenciesCache: d, RunContext: r}
 	switch {
 	case p.Build.LocalBuild != nil:
diff --git a/pkg/skaffold/runner/changeset.go b/pkg/skaffold/runner/changeset.go
index 316d523086a..567737355e4 100644
--- a/pkg/skaffold/runner/changeset.go
+++ b/pkg/skaffold/runner/changeset.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2019 The Skaffold Authors
+Copyright 2021 The Skaffold Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -31,11 +31,35 @@ type ChangeSet struct {
 	needsReload    bool
 }
 
+// NeedsRebuild gets the value of needsRebuild, which itself is not expected to be changed outside ChangeSet
+func (c *ChangeSet) NeedsRebuild() []*latest_v1.Artifact {
+	return c.needsRebuild
+}
+
+// NeedsResync gets the value of needsResync, which itself is not expected to be changed outside ChangeSet
+func (c *ChangeSet) NeedsResync() []*sync.Item {
+	return c.needsResync
+}
+
+// NeedsRedeploy gets the value of needsRedeploy, which itself is not expected to be changed outside ChangeSet
+func (c *ChangeSet) NeedsRedeploy() bool {
+	return c.needsRedeploy
+}
+
+// NeedsRetest gets the value of needsRetest, which itself is not expected to be changed outside ChangeSet
+func (c *ChangeSet) NeedsRetest() map[string]bool {
+	return c.needsRetest
+}
+
+// NeedsReload gets the value of needsReload, which itself is not expected to be changed outside ChangeSet
+func (c *ChangeSet) NeedsReload() bool {
+	return c.needsReload
+}
+
 func (c *ChangeSet) AddRebuild(a *latest_v1.Artifact) {
 	if _, ok := c.rebuildTracker[a.ImageName]; ok {
 		return
 	}
-
 	if c.rebuildTracker == nil {
 		c.rebuildTracker = map[string]*latest_v1.Artifact{}
 	}
@@ -54,7 +78,6 @@ func (c *ChangeSet) AddResync(s *sync.Item) {
 	if _, ok := c.resyncTracker[s.Image]; ok {
 		return
 	}
-
 	if c.resyncTracker == nil {
 		c.resyncTracker = map[string]*sync.Item{}
 	}
@@ -62,20 +85,30 @@ func (c *ChangeSet) AddResync(s *sync.Item) {
 	c.needsResync = append(c.needsResync, s)
 }
 
-func (c *ChangeSet) resetBuild() {
+func (c *ChangeSet) ResetBuild() {
 	c.rebuildTracker = make(map[string]*latest_v1.Artifact)
 	c.needsRebuild = nil
 }
 
-func (c *ChangeSet) resetSync() {
+func (c *ChangeSet) ResetSync() {
 	c.resyncTracker = make(map[string]*sync.Item)
 	c.needsResync = nil
 }
 
-func (c *ChangeSet) resetDeploy() {
+func (c *ChangeSet) ResetDeploy() {
 	c.needsRedeploy = false
 }
 
-func (c *ChangeSet) resetTest() {
+// Redeploy marks that deploy is expected to happen.
+func (c *ChangeSet) Redeploy() {
+	c.needsRedeploy = true
+}
+
+// Reload marks that reload is expected to happen.
+func (c *ChangeSet) Reload() {
+	c.needsReload = true
+}
+
+func (c *ChangeSet) ResetTest() {
 	c.needsRetest = make(map[string]bool)
 }
diff --git a/pkg/skaffold/runner/intent.go b/pkg/skaffold/runner/intent.go
index 601355f0f81..0e2c17b1fda 100644
--- a/pkg/skaffold/runner/intent.go
+++ b/pkg/skaffold/runner/intent.go
@@ -29,7 +29,7 @@ type Intents struct {
 	lock sync.Mutex
 }
 
-func newIntents(autoBuild, autoSync, autoDeploy bool) *Intents {
+func NewIntents(autoBuild, autoSync, autoDeploy bool) *Intents {
 	i := &Intents{
 		autoBuild:  autoBuild,
 		autoSync:   autoSync,
@@ -39,7 +39,12 @@ func newIntents(autoBuild, autoSync, autoDeploy bool) *Intents {
 	return i
 }
 
-func (i *Intents) reset() {
+// GetIntentsAttrs returns the intent attributes for testing only.
+func (i *Intents) GetIntentsAttrs() (bool, bool, bool) {
+	return i.build, i.sync, i.deploy
+}
+
+func (i *Intents) Reset() {
 	i.lock.Lock()
 	i.build = i.autoBuild
 	i.sync = i.autoSync
@@ -47,73 +52,73 @@ func (i *Intents) reset() {
 	i.lock.Unlock()
 }
 
-func (i *Intents) resetBuild() {
+func (i *Intents) ResetBuild() {
 	i.lock.Lock()
 	i.build = i.autoBuild
 	i.lock.Unlock()
 }
 
-func (i *Intents) resetSync() {
+func (i *Intents) ResetSync() {
 	i.lock.Lock()
 	i.sync = i.autoSync
 	i.lock.Unlock()
 }
 
-func (i *Intents) resetDeploy() {
+func (i *Intents) ResetDeploy() {
 	i.lock.Lock()
 	i.deploy = i.autoDeploy
 	i.lock.Unlock()
 }
 
-func (i *Intents) setBuild(val bool) {
+func (i *Intents) SetBuild(val bool) {
 	i.lock.Lock()
 	i.build = val
 	i.lock.Unlock()
 }
 
-func (i *Intents) setSync(val bool) {
+func (i *Intents) SetSync(val bool) {
 	i.lock.Lock()
 	i.sync = val
 	i.lock.Unlock()
 }
 
-func (i *Intents) setDeploy(val bool) {
+func (i *Intents) SetDeploy(val bool) {
 	i.lock.Lock()
 	i.deploy = val
 	i.lock.Unlock()
 }
 
-func (i *Intents) getAutoBuild() bool {
+func (i *Intents) GetAutoBuild() bool {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 	return i.autoBuild
 }
 
-func (i *Intents) getAutoSync() bool {
+func (i *Intents) GetAutoSync() bool {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 	return i.autoSync
 }
 
-func (i *Intents) getAutoDeploy() bool {
+func (i *Intents) GetAutoDeploy() bool {
 	i.lock.Lock()
 	defer i.lock.Unlock()
 	return i.autoDeploy
 }
 
-func (i *Intents) setAutoBuild(val bool) {
+func (i *Intents) SetAutoBuild(val bool) {
 	i.lock.Lock()
 	i.autoBuild = val
 	i.lock.Unlock()
 }
 
-func (i *Intents) setAutoSync(val bool) {
+func (i *Intents) SetAutoSync(val bool) {
 	i.lock.Lock()
 	i.autoSync = val
 	i.lock.Unlock()
 }
 
-func (i *Intents) setAutoDeploy(val bool) {
+func (i *Intents) SetAutoDeploy(val bool) {
 	i.lock.Lock()
 	i.autoDeploy = val
 	i.lock.Unlock()
diff --git a/pkg/skaffold/runner/listen.go b/pkg/skaffold/runner/listen.go
index 4b8d96113b8..4d57758c8c9 100644
--- a/pkg/skaffold/runner/listen.go
+++ b/pkg/skaffold/runner/listen.go
@@ -29,6 +29,16 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/trigger"
 )
 
+func NewSkaffoldListener(monitor filemon.Monitor, trigger trigger.Trigger, cache graph.SourceDependenciesCache,
+	intentChan <-chan bool) *SkaffoldListener {
+	return &SkaffoldListener{
+		Monitor:                 monitor,
+		Trigger:                 trigger,
+		sourceDependenciesCache: cache,
+		intentChan:              intentChan,
+	}
+}
+
 type Listener interface {
 	WatchForChanges(context.Context, io.Writer, func() error) error
 	LogWatchToUser(io.Writer)
diff --git a/pkg/skaffold/runner/runner.go b/pkg/skaffold/runner/runner.go
index 3445fce690c..9d920bcc42b 100644
--- a/pkg/skaffold/runner/runner.go
+++ b/pkg/skaffold/runner/runner.go
@@ -18,28 +18,23 @@ package runner
 
 import (
 	"context"
+	"errors"
 	"io"
 
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/build"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/build/cache"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy/label"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy/status"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/filemon"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
 	latest_v1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/sync"
 )
 
 const (
-	remoteDigestSource = "remote"
-	noneDigestSource   = "none"
-	tagDigestSource    = "tag"
+	RemoteDigestSource = "remote"
+	NoneDigestSource   = "none"
+	TagDigestSource    = "tag"
 )
 
+// ErrorConfigurationChanged is a special error that's returned when the skaffold configuration was changed.
+var ErrorConfigurationChanged = errors.New("configuration changed")
+
 // Runner is responsible for running the skaffold build, test and deploy config.
 type Runner interface {
 	Apply(context.Context, io.Writer) error
@@ -57,39 +52,7 @@ type Runner interface {
 	Test(context.Context, io.Writer, []graph.Artifact) error
 }
 
-// SkaffoldRunner is responsible for running the skaffold build, test and deploy config.
-type SkaffoldRunner struct {
-	Builder
-	Pruner
-	Tester
-
-	deployer deploy.Deployer
-	syncer   sync.Syncer
-	monitor  filemon.Monitor
-	listener Listener
-
-	kubectlCLI         *kubectl.CLI
-	cache              cache.Cache
-	changeSet          ChangeSet
-	runCtx             *runcontext.RunContext
-	labeller           *label.DefaultLabeller
-	artifactStore      build.ArtifactStore
-	sourceDependencies graph.SourceDependenciesCache
-	// podSelector is used to determine relevant pods for logging and portForwarding
-	podSelector *kubernetes.ImageList
-
-	devIteration int
-	isLocalImage func(imageName string) (bool, error)
-	hasDeployed  bool
-	intents      *Intents
-}
-
 // for testing
 var (
-	newStatusCheck = status.NewStatusChecker
+	NewStatusCheck = status.NewStatusChecker
 )
-
-// HasDeployed returns true if this runner has deployed something.
-func (r *SkaffoldRunner) HasDeployed() bool {
-	return r.hasDeployed
-}
diff --git a/pkg/skaffold/runner/tester.go b/pkg/skaffold/runner/tester.go
deleted file mode 100644
index 9147c071b16..00000000000
--- a/pkg/skaffold/runner/tester.go
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
-Copyright 2021 The Skaffold Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License 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 runner
-
-import (
-	"context"
-	"io"
-
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/constants"
-	eventV2 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/event/v2"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/test"
-)
-
-type Tester struct {
-	tester test.Tester
-}
-
-// Test tests a list of already built artifacts.
-func (r *Tester) Test(ctx context.Context, out io.Writer, artifacts []graph.Artifact) error {
-	eventV2.TaskInProgress(constants.Test)
-
-	if err := r.tester.Test(ctx, out, artifacts); err != nil {
-		eventV2.TaskFailed(constants.Test, err)
-		return err
-	}
-
-	eventV2.TaskSucceeded(constants.Test)
-	return nil
-}
diff --git a/pkg/skaffold/runner/apply.go b/pkg/skaffold/runner/v1/apply.go
similarity index 99%
rename from pkg/skaffold/runner/apply.go
rename to pkg/skaffold/runner/v1/apply.go
index 19ecdf1c41c..75059f24bad 100644
--- a/pkg/skaffold/runner/apply.go
+++ b/pkg/skaffold/runner/v1/apply.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
diff --git a/pkg/skaffold/runner/build_deploy_test.go b/pkg/skaffold/runner/v1/build_deploy_test.go
similarity index 98%
rename from pkg/skaffold/runner/build_deploy_test.go
rename to pkg/skaffold/runner/v1/build_deploy_test.go
index ae0a0d4425f..3173831ffae 100644
--- a/pkg/skaffold/runner/build_deploy_test.go
+++ b/pkg/skaffold/runner/v1/build_deploy_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
@@ -27,6 +27,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/config"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/client"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
 	latest_v1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
 	"github.com/GoogleContainerTools/skaffold/testutil"
@@ -284,7 +285,7 @@ func TestCheckWorkspaces(t *testing.T) {
 	}
 	for _, test := range tests {
 		testutil.Run(t, test.description, func(t *testutil.T) {
-			err := checkWorkspaces(test.artifacts)
+			err := runner.CheckWorkspaces(test.artifacts)
 			t.CheckError(test.shouldErr, err)
 		})
 	}
diff --git a/pkg/skaffold/runner/cleanup.go b/pkg/skaffold/runner/v1/cleanup.go
similarity index 97%
rename from pkg/skaffold/runner/cleanup.go
rename to pkg/skaffold/runner/v1/cleanup.go
index a14d2b23794..996e38e43f5 100644
--- a/pkg/skaffold/runner/cleanup.go
+++ b/pkg/skaffold/runner/v1/cleanup.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
diff --git a/pkg/skaffold/runner/debugging.go b/pkg/skaffold/runner/v1/debugging.go
similarity index 98%
rename from pkg/skaffold/runner/debugging.go
rename to pkg/skaffold/runner/v1/debugging.go
index e9bfa9ecd59..bbb0c0d75b9 100644
--- a/pkg/skaffold/runner/debugging.go
+++ b/pkg/skaffold/runner/v1/debugging.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/config"
diff --git a/pkg/skaffold/runner/deploy.go b/pkg/skaffold/runner/v1/deploy.go
similarity index 97%
rename from pkg/skaffold/runner/deploy.go
rename to pkg/skaffold/runner/v1/deploy.go
index 411af3baf7f..86447b6f6a8 100644
--- a/pkg/skaffold/runner/deploy.go
+++ b/pkg/skaffold/runner/v1/deploy.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
@@ -34,6 +34,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	kubernetesclient "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/client"
 	kubectx "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/context"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
 )
 
@@ -42,7 +43,7 @@ func (r *SkaffoldRunner) DeployAndLog(ctx context.Context, out io.Writer, artifa
 	eventV2.TaskInProgress(constants.Deploy)
 
 	// Update which images are logged.
-	r.addTagsToPodSelector(artifacts)
+	r.AddTagsToPodSelector(artifacts)
 
 	logger := r.createLogger(out, artifacts)
 	defer logger.Stop()
@@ -213,7 +214,7 @@ func (r *SkaffoldRunner) performStatusCheck(ctx context.Context, out io.Writer)
 	start := time.Now()
 	color.Default.Fprintln(out, "Waiting for deployments to stabilize...")
 
-	s := newStatusCheck(r.runCtx, r.labeller)
+	s := runner.NewStatusCheck(r.runCtx, r.labeller)
 	if err := s.Check(ctx, out); err != nil {
 		eventV2.TaskFailed(constants.StatusCheck, err)
 		return err
diff --git a/pkg/skaffold/runner/deploy_test.go b/pkg/skaffold/runner/v1/deploy_test.go
similarity index 87%
rename from pkg/skaffold/runner/deploy_test.go
rename to pkg/skaffold/runner/v1/deploy_test.go
index 7622182a84c..20aa3ad0374 100644
--- a/pkg/skaffold/runner/deploy_test.go
+++ b/pkg/skaffold/runner/v1/deploy_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"bytes"
@@ -33,6 +33,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/client"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
 	latest_v1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
@@ -114,16 +115,16 @@ func TestDeploy(t *testing.T) {
 		testutil.Run(t, test.description, func(t *testutil.T) {
 			t.SetupFakeKubernetesContext(api.Config{CurrentContext: "cluster1"})
 			t.Override(&client.Client, mockK8sClient)
-			t.Override(&newStatusCheck, func(status.Config, *label.DefaultLabeller) status.Checker {
+			t.Override(&runner.NewStatusCheck, func(status.Config, *label.DefaultLabeller) status.Checker {
 				return dummyStatusChecker{}
 			})
 
-			runner := createRunner(t, test.testBench, nil, []*latest_v1.Artifact{{ImageName: "img1"}, {ImageName: "img2"}}, nil)
-			runner.runCtx.Opts.StatusCheck = config.NewBoolOrUndefined(test.statusCheckFlag)
-			runner.runCtx.Pipelines.All()[0].Deploy.StatusCheck = test.statusCheckConfig
+			r := createRunner(t, test.testBench, nil, []*latest_v1.Artifact{{ImageName: "img1"}, {ImageName: "img2"}}, nil)
+			r.runCtx.Opts.StatusCheck = config.NewBoolOrUndefined(test.statusCheckFlag)
+			r.runCtx.Pipelines.All()[0].Deploy.StatusCheck = test.statusCheckConfig
 			out := new(bytes.Buffer)
 
-			err := runner.Deploy(context.Background(), out, []graph.Artifact{
+			err := r.Deploy(context.Background(), out, []graph.Artifact{
 				{ImageName: "img1", Tag: "img1:tag1"},
 				{ImageName: "img2", Tag: "img2:tag2"},
 			})
@@ -165,19 +166,19 @@ func TestDeployNamespace(t *testing.T) {
 		testutil.Run(t, test.description, func(t *testutil.T) {
 			t.SetupFakeKubernetesContext(api.Config{CurrentContext: "cluster1"})
 			t.Override(&client.Client, mockK8sClient)
-			t.Override(&newStatusCheck, func(status.Config, *label.DefaultLabeller) status.Checker {
+			t.Override(&runner.NewStatusCheck, func(status.Config, *label.DefaultLabeller) status.Checker {
 				return dummyStatusChecker{}
 			})
 
-			runner := createRunner(t, test.testBench, nil, []*latest_v1.Artifact{{ImageName: "img1"}, {ImageName: "img2"}}, nil)
-			runner.runCtx.Namespaces = test.Namespaces
+			r := createRunner(t, test.testBench, nil, []*latest_v1.Artifact{{ImageName: "img1"}, {ImageName: "img2"}}, nil)
+			r.runCtx.Namespaces = test.Namespaces
 
-			runner.Deploy(context.Background(), ioutil.Discard, []graph.Artifact{
+			err := r.Deploy(context.Background(), ioutil.Discard, []graph.Artifact{
 				{ImageName: "img1", Tag: "img1:tag1"},
 				{ImageName: "img2", Tag: "img2:tag2"},
 			})
-
-			t.CheckDeepEqual(test.expected, runner.runCtx.GetNamespaces())
+			t.CheckNoError(err)
+			t.CheckDeepEqual(test.expected, r.runCtx.GetNamespaces())
 		})
 	}
 }
diff --git a/pkg/skaffold/runner/dev.go b/pkg/skaffold/runner/v1/dev.go
similarity index 88%
rename from pkg/skaffold/runner/dev.go
rename to pkg/skaffold/runner/v1/dev.go
index 00645becad2..21e2962fb47 100644
--- a/pkg/skaffold/runner/dev.go
+++ b/pkg/skaffold/runner/v1/dev.go
@@ -14,11 +14,10 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"time"
@@ -34,15 +33,13 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/instrumentation"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/logger"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/portforward"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
 	latest_v1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/sync"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
 	"github.com/GoogleContainerTools/skaffold/proto/v1"
 )
 
-// ErrorConfigurationChanged is a special error that's returned when the skaffold configuration was changed.
-var ErrorConfigurationChanged = errors.New("configuration changed")
-
 var (
 	// For testing
 	fileSyncInProgress = event.FileSyncInProgress
@@ -52,18 +49,18 @@ var (
 
 func (r *SkaffoldRunner) doDev(ctx context.Context, out io.Writer, logger *logger.LogAggregator, forwarderManager portforward.Forwarder) error {
 	// never queue intents from user, even if they're not used
-	defer r.intents.reset()
+	defer r.intents.Reset()
 
-	if r.changeSet.needsReload {
-		return ErrorConfigurationChanged
+	if r.changeSet.NeedsReload() {
+		return runner.ErrorConfigurationChanged
 	}
 
 	buildIntent, syncIntent, deployIntent := r.intents.GetIntents()
 	logrus.Tracef("dev intents: build %t, sync %t, deploy %t\n", buildIntent, syncIntent, deployIntent)
-	needsSync := syncIntent && len(r.changeSet.needsResync) > 0
-	needsBuild := buildIntent && len(r.changeSet.needsRebuild) > 0
-	needsTest := len(r.changeSet.needsRetest) > 0
-	needsDeploy := deployIntent && r.changeSet.needsRedeploy
+	needsSync := syncIntent && len(r.changeSet.NeedsResync()) > 0
+	needsBuild := buildIntent && len(r.changeSet.NeedsRebuild()) > 0
+	needsTest := len(r.changeSet.NeedsRetest()) > 0
+	needsDeploy := deployIntent && r.changeSet.NeedsRedeploy()
 	if !needsSync && !needsBuild && !needsTest && !needsDeploy {
 		return nil
 	}
@@ -79,12 +76,12 @@ func (r *SkaffoldRunner) doDev(ctx context.Context, out io.Writer, logger *logge
 	meterUpdated := false
 	if needsSync {
 		defer func() {
-			r.changeSet.resetSync()
-			r.intents.resetSync()
+			r.changeSet.ResetSync()
+			r.intents.ResetSync()
 		}()
 		instrumentation.AddDevIteration("sync")
 		meterUpdated = true
-		for _, s := range r.changeSet.needsResync {
+		for _, s := range r.changeSet.NeedsResync() {
 			fileCount := len(s.Copy) + len(s.Delete)
 			color.Default.Fprintf(out, "Syncing %d files for %s\n", fileCount, s.Image)
 			fileSyncInProgress(fileCount, s.Image)
@@ -105,8 +102,8 @@ func (r *SkaffoldRunner) doDev(ctx context.Context, out io.Writer, logger *logge
 	if needsBuild {
 		event.ResetStateOnBuild()
 		defer func() {
-			r.changeSet.resetBuild()
-			r.intents.resetBuild()
+			r.changeSet.ResetBuild()
+			r.intents.ResetBuild()
 		}()
 		if !meterUpdated {
 			instrumentation.AddDevIteration("build")
@@ -114,14 +111,14 @@ func (r *SkaffoldRunner) doDev(ctx context.Context, out io.Writer, logger *logge
 		}
 
 		var err error
-		bRes, err = r.Build(ctx, out, r.changeSet.needsRebuild)
+		bRes, err = r.Build(ctx, out, r.changeSet.NeedsRebuild())
 		if err != nil {
 			logrus.Warnln("Skipping test and deploy due to build error:", err)
 			event.DevLoopFailedInPhase(r.devIteration, constants.Build, err)
 			eventV2.TaskFailed(constants.DevLoop, err)
 			return nil
 		}
-		r.changeSet.needsRedeploy = true
+		r.changeSet.Redeploy()
 		needsDeploy = deployIntent
 	}
 
@@ -129,13 +126,13 @@ func (r *SkaffoldRunner) doDev(ctx context.Context, out io.Writer, logger *logge
 	if (len(bRes) > 0 || needsTest) && !r.runCtx.SkipTests() {
 		event.ResetStateOnTest()
 		defer func() {
-			r.changeSet.resetTest()
+			r.changeSet.ResetTest()
 		}()
 		for _, a := range bRes {
-			delete(r.changeSet.needsRetest, a.ImageName)
+			delete(r.changeSet.NeedsRetest(), a.ImageName)
 		}
-		for _, a := range r.builds {
-			if r.changeSet.needsRetest[a.ImageName] {
+		for _, a := range r.Builds {
+			if r.changeSet.NeedsRetest()[a.ImageName] {
 				bRes = append(bRes, a)
 			}
 		}
@@ -152,15 +149,15 @@ func (r *SkaffoldRunner) doDev(ctx context.Context, out io.Writer, logger *logge
 	if needsDeploy {
 		event.ResetStateOnDeploy()
 		defer func() {
-			r.changeSet.resetDeploy()
-			r.intents.resetDeploy()
+			r.changeSet.ResetDeploy()
+			r.intents.ResetDeploy()
 		}()
 
 		forwarderManager.Stop()
 		if !meterUpdated {
 			instrumentation.AddDevIteration("deploy")
 		}
-		if err := r.Deploy(ctx, out, r.builds); err != nil {
+		if err := r.Deploy(ctx, out, r.Builds); err != nil {
 			logrus.Warnln("Skipping deploy due to error:", err)
 			event.DevLoopFailedInPhase(r.devIteration, constants.Deploy, err)
 			eventV2.TaskFailed(constants.DevLoop, err)
@@ -204,7 +201,7 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*la
 					return r.sourceDependencies.TransitiveArtifactDependencies(ctx, artifact)
 				},
 				func(e filemon.Events) {
-					s, err := sync.NewItem(ctx, artifact, e, r.builds, r.runCtx, len(g[artifact.ImageName]))
+					s, err := sync.NewItem(ctx, artifact, e, r.Builds, r.runCtx, len(g[artifact.ImageName]))
 					switch {
 					case err != nil:
 						logrus.Warnf("error adding dirty artifact to changeset: %s", err.Error())
@@ -226,7 +223,7 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*la
 	for i := range artifacts {
 		artifact := artifacts[i]
 		if err := r.monitor.Register(
-			func() ([]string, error) { return r.tester.TestDependencies(artifact) },
+			func() ([]string, error) { return r.Tester.TestDependencies(artifact) },
 			func(filemon.Events) { r.changeSet.AddRetest(artifact) },
 		); err != nil {
 			event.DevLoopFailedWithErrorCode(r.devIteration, proto.StatusCode_DEVINIT_REGISTER_TEST_DEPS, err)
@@ -238,7 +235,7 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*la
 	// Watch deployment configuration
 	if err := r.monitor.Register(
 		r.deployer.Dependencies,
-		func(filemon.Events) { r.changeSet.needsRedeploy = true },
+		func(filemon.Events) { r.changeSet.Redeploy() },
 	); err != nil {
 		event.DevLoopFailedWithErrorCode(r.devIteration, proto.StatusCode_DEVINIT_REGISTER_DEPLOY_DEPS, err)
 		eventV2.TaskFailed(constants.DevLoop, err)
@@ -248,7 +245,7 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*la
 	// Watch Skaffold configuration
 	if err := r.monitor.Register(
 		func() ([]string, error) { return []string{r.runCtx.ConfigurationFile()}, nil },
-		func(filemon.Events) { r.changeSet.needsReload = true },
+		func(filemon.Events) { r.changeSet.Reload() },
 	); err != nil {
 		event.DevLoopFailedWithErrorCode(r.devIteration, proto.StatusCode_DEVINIT_REGISTER_CONFIG_DEP, err)
 		eventV2.TaskFailed(constants.DevLoop, err)
@@ -290,7 +287,7 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*la
 	logger.SetSince(time.Now())
 
 	// First deploy
-	if err := r.Deploy(ctx, out, r.builds); err != nil {
+	if err := r.Deploy(ctx, out, r.Builds); err != nil {
 		event.DevLoopFailedInPhase(r.devIteration, constants.Deploy, err)
 		eventV2.TaskFailed(constants.DevLoop, err)
 		return fmt.Errorf("exiting dev mode because first deploy failed: %w", err)
diff --git a/pkg/skaffold/runner/dev_test.go b/pkg/skaffold/runner/v1/dev_test.go
similarity index 92%
rename from pkg/skaffold/runner/dev_test.go
rename to pkg/skaffold/runner/v1/dev_test.go
index e78acf3d056..98ef35741fc 100644
--- a/pkg/skaffold/runner/dev_test.go
+++ b/pkg/skaffold/runner/v1/dev_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
@@ -30,6 +30,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/docker"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/filemon"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/client"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
 	latest_v1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/sync"
 	"github.com/GoogleContainerTools/skaffold/testutil"
@@ -146,10 +147,10 @@ func TestDevFailFirstCycle(t *testing.T) {
 			artifacts := []*latest_v1.Artifact{{
 				ImageName: "img",
 			}}
-			runner := createRunner(t, test.testBench, test.monitor, artifacts, nil)
+			r := createRunner(t, test.testBench, test.monitor, artifacts, nil)
 			test.testBench.firstMonitor = test.monitor.Run
 
-			err := runner.Dev(context.Background(), ioutil.Discard, artifacts)
+			err := r.Dev(context.Background(), ioutil.Discard, artifacts)
 
 			t.CheckErrorAndDeepEqual(true, err, test.expectedActions, test.testBench.Actions())
 		})
@@ -278,12 +279,12 @@ func TestDev(t *testing.T) {
 				{ImageName: "img1"},
 				{ImageName: "img2"},
 			}
-			runner := createRunner(t, test.testBench, &TestMonitor{
+			r := createRunner(t, test.testBench, &TestMonitor{
 				events:    test.watchEvents,
 				testBench: test.testBench,
 			}, artifacts, nil)
 
-			err := runner.Dev(context.Background(), ioutil.Discard, artifacts)
+			err := r.Dev(context.Background(), ioutil.Discard, artifacts)
 
 			t.CheckNoError(err)
 			t.CheckDeepEqual(test.expectedActions, test.testBench.Actions())
@@ -298,7 +299,7 @@ func TestDevAutoTriggers(t *testing.T) {
 		expectedActions []Actions
 		autoTriggers    triggerState // the state of auto triggers
 		singleTriggers  triggerState // the state of single intent triggers at the end of dev loop
-		userIntents     []func(i *Intents)
+		userIntents     []func(i *runner.Intents)
 	}{
 		{
 			description: "build on; sync on; deploy on",
@@ -368,9 +369,9 @@ func TestDevAutoTriggers(t *testing.T) {
 			autoTriggers:    triggerState{false, false, false},
 			singleTriggers:  triggerState{false, false, false},
 			expectedActions: []Actions{},
-			userIntents: []func(i *Intents){
-				func(i *Intents) {
-					i.setBuild(true)
+			userIntents: []func(i *runner.Intents){
+				func(i *runner.Intents) {
+					i.SetBuild(true)
 				},
 			},
 		},
@@ -380,10 +381,10 @@ func TestDevAutoTriggers(t *testing.T) {
 			autoTriggers:    triggerState{false, false, false},
 			singleTriggers:  triggerState{false, false, false},
 			expectedActions: []Actions{},
-			userIntents: []func(i *Intents){
-				func(i *Intents) {
-					i.setBuild(true)
-					i.setSync(true)
+			userIntents: []func(i *runner.Intents){
+				func(i *runner.Intents) {
+					i.SetBuild(true)
+					i.SetSync(true)
 				},
 			},
 		},
@@ -393,12 +394,12 @@ func TestDevAutoTriggers(t *testing.T) {
 			autoTriggers:    triggerState{false, false, false},
 			singleTriggers:  triggerState{false, false, false},
 			expectedActions: []Actions{},
-			userIntents: []func(i *Intents){
-				func(i *Intents) {
-					i.setBuild(true)
+			userIntents: []func(i *runner.Intents){
+				func(i *runner.Intents) {
+					i.SetBuild(true)
 				},
-				func(i *Intents) {
-					i.setSync(true)
+				func(i *runner.Intents) {
+					i.SetSync(true)
 				},
 			},
 		},
@@ -429,22 +430,23 @@ func TestDevAutoTriggers(t *testing.T) {
 					ImageName: "img2",
 				},
 			}
-			runner := createRunner(t, testBench, &TestMonitor{
+			r := createRunner(t, testBench, &TestMonitor{
 				events:    test.watchEvents,
 				testBench: testBench,
 			}, artifacts, &test.autoTriggers)
 
-			testBench.intents = runner.intents
+			testBench.intents = r.intents
 
-			err := runner.Dev(context.Background(), ioutil.Discard, artifacts)
+			err := r.Dev(context.Background(), ioutil.Discard, artifacts)
 
 			t.CheckNoError(err)
 			t.CheckDeepEqual(append([]Actions{firstAction}, test.expectedActions...), testBench.Actions())
 
+			build, _sync, deploy := r.intents.GetIntentsAttrs()
 			singleTriggers := triggerState{
-				build:  runner.intents.build,
-				sync:   runner.intents.sync,
-				deploy: runner.intents.deploy,
+				build:  build,
+				sync:   _sync,
+				deploy: deploy,
 			}
 			t.CheckDeepEqual(test.singleTriggers, singleTriggers, cmp.AllowUnexported(triggerState{}))
 		})
@@ -535,12 +537,12 @@ func TestDevSync(t *testing.T) {
 					ImageName: "img2",
 				},
 			}
-			runner := createRunner(t, test.testBench, &TestMonitor{
+			r := createRunner(t, test.testBench, &TestMonitor{
 				events:    test.watchEvents,
 				testBench: test.testBench,
 			}, artifacts, nil)
 
-			err := runner.Dev(context.Background(), ioutil.Discard, artifacts)
+			err := r.Dev(context.Background(), ioutil.Discard, artifacts)
 
 			t.CheckNoError(err)
 			t.CheckDeepEqual(test.expectedActions, test.testBench.Actions())
diff --git a/pkg/skaffold/runner/generate_pipeline.go b/pkg/skaffold/runner/v1/generate_pipeline.go
similarity index 99%
rename from pkg/skaffold/runner/generate_pipeline.go
rename to pkg/skaffold/runner/v1/generate_pipeline.go
index ad0b72b7ad1..e015e5045ec 100644
--- a/pkg/skaffold/runner/generate_pipeline.go
+++ b/pkg/skaffold/runner/v1/generate_pipeline.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
diff --git a/pkg/skaffold/runner/load_images.go b/pkg/skaffold/runner/v1/load_images.go
similarity index 98%
rename from pkg/skaffold/runner/load_images.go
rename to pkg/skaffold/runner/v1/load_images.go
index 13ddebe3d7b..88ec046ab20 100644
--- a/pkg/skaffold/runner/load_images.go
+++ b/pkg/skaffold/runner/v1/load_images.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
@@ -101,7 +101,7 @@ func findKnownImages(ctx context.Context, cli *kubectl.CLI) ([]string, error) {
 }
 
 func (r *SkaffoldRunner) wasBuilt(tag string) bool {
-	for _, built := range r.builds {
+	for _, built := range r.Builds {
 		if built.Tag == tag {
 			return true
 		}
diff --git a/pkg/skaffold/runner/load_images_test.go b/pkg/skaffold/runner/v1/load_images_test.go
similarity index 98%
rename from pkg/skaffold/runner/load_images_test.go
rename to pkg/skaffold/runner/v1/load_images_test.go
index f86c4437802..9c84bd411a3 100644
--- a/pkg/skaffold/runner/load_images_test.go
+++ b/pkg/skaffold/runner/v1/load_images_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
@@ -25,6 +25,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/config"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
 	"github.com/GoogleContainerTools/skaffold/testutil"
@@ -201,9 +202,7 @@ func runImageLoadingTests(t *testing.T, tests []ImageLoadingTest, loadingFunc fu
 			r := &SkaffoldRunner{
 				runCtx:     runCtx,
 				kubectlCLI: kubectl.NewCLI(runCtx, ""),
-				Builder: Builder{
-					builds: test.built,
-				},
+				Builder:    runner.Builder{Builds: test.built},
 			}
 			err := loadingFunc(r, test)
 
diff --git a/pkg/skaffold/runner/logger.go b/pkg/skaffold/runner/v1/logger.go
similarity index 98%
rename from pkg/skaffold/runner/logger.go
rename to pkg/skaffold/runner/v1/logger.go
index cd2cb82f044..f5a8a09c805 100644
--- a/pkg/skaffold/runner/logger.go
+++ b/pkg/skaffold/runner/v1/logger.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"io"
diff --git a/pkg/skaffold/runner/new.go b/pkg/skaffold/runner/v1/new.go
similarity index 89%
rename from pkg/skaffold/runner/new.go
rename to pkg/skaffold/runner/v1/new.go
index ac22ce9ac9e..1f379b78189 100644
--- a/pkg/skaffold/runner/new.go
+++ b/pkg/skaffold/runner/v1/new.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
@@ -39,6 +39,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	pkgkubectl "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
 	latest_v1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/server"
@@ -67,7 +68,7 @@ func NewForConfig(runCtx *runcontext.RunContext) (*SkaffoldRunner, error) {
 
 	var builder build.Builder
 	builder, err = build.NewBuilderMux(runCtx, store, func(p latest_v1.Pipeline) (build.PipelineBuilder, error) {
-		return getBuilder(runCtx, store, sourceDependencies, p)
+		return runner.GetBuilder(runCtx, store, sourceDependencies, p)
 	})
 	if err != nil {
 		return nil, fmt.Errorf("creating builder: %w", err)
@@ -106,42 +107,29 @@ func NewForConfig(runCtx *runcontext.RunContext) (*SkaffoldRunner, error) {
 		return nil, fmt.Errorf("initializing cache: %w", err)
 	}
 
-	builder, tester, deployer = WithTimings(builder, tester, deployer, runCtx.CacheArtifacts())
+	builder, tester, deployer = runner.WithTimings(builder, tester, deployer, runCtx.CacheArtifacts())
 	if runCtx.Notification() {
-		deployer = WithNotification(deployer)
+		deployer = runner.WithNotification(deployer)
 	}
 
 	monitor := filemon.NewMonitor()
 	intents, intentChan := setupIntents(runCtx)
-	trigger, err := trigger.NewTrigger(runCtx, intents.IsAnyAutoEnabled)
+	rtrigger, err := trigger.NewTrigger(runCtx, intents.IsAnyAutoEnabled)
 	if err != nil {
 		return nil, fmt.Errorf("creating watch trigger: %w", err)
 	}
 
 	podSelectors := kubernetes.NewImageList()
+
+	rbuilder := runner.NewBuilder(builder, tagger, artifactCache, podSelectors, runCtx)
 	return &SkaffoldRunner{
-		Builder: Builder{
-			builder:     builder,
-			tagger:      tagger,
-			cache:       artifactCache,
-			podSelector: podSelectors,
-			runCtx:      runCtx,
-		},
-		Pruner: Pruner{
-			builder,
-		},
-		Tester: Tester{
-			tester: tester,
-		},
-		deployer: deployer,
-		syncer:   syncer,
-		monitor:  monitor,
-		listener: &SkaffoldListener{
-			Monitor:                 monitor,
-			Trigger:                 trigger,
-			intentChan:              intentChan,
-			sourceDependenciesCache: sourceDependencies,
-		},
+		Builder:            *rbuilder,
+		Pruner:             runner.Pruner{Builder: builder},
+		Tester:             tester,
+		deployer:           deployer,
+		syncer:             syncer,
+		monitor:            monitor,
+		listener:           runner.NewSkaffoldListener(monitor, rtrigger, sourceDependencies, intentChan),
 		artifactStore:      store,
 		sourceDependencies: sourceDependencies,
 		kubectlCLI:         kubectlCLI,
@@ -154,13 +142,13 @@ func NewForConfig(runCtx *runcontext.RunContext) (*SkaffoldRunner, error) {
 	}, nil
 }
 
-func setupIntents(runCtx *runcontext.RunContext) (*Intents, chan bool) {
-	intents := newIntents(runCtx.AutoBuild(), runCtx.AutoSync(), runCtx.AutoDeploy())
+func setupIntents(runCtx *runcontext.RunContext) (*runner.Intents, chan bool) {
+	intents := runner.NewIntents(runCtx.AutoBuild(), runCtx.AutoSync(), runCtx.AutoDeploy())
 
 	intentChan := make(chan bool, 1)
-	setupTrigger("build", intents.setBuild, intents.setAutoBuild, intents.getAutoBuild, server.SetBuildCallback, server.SetAutoBuildCallback, intentChan)
-	setupTrigger("sync", intents.setSync, intents.setAutoSync, intents.getAutoSync, server.SetSyncCallback, server.SetAutoSyncCallback, intentChan)
-	setupTrigger("deploy", intents.setDeploy, intents.setAutoDeploy, intents.getAutoDeploy, server.SetDeployCallback, server.SetAutoDeployCallback, intentChan)
+	setupTrigger("build", intents.SetBuild, intents.SetAutoBuild, intents.GetAutoBuild, server.SetBuildCallback, server.SetAutoBuildCallback, intentChan)
+	setupTrigger("sync", intents.SetSync, intents.SetAutoSync, intents.GetAutoSync, server.SetSyncCallback, server.SetAutoSyncCallback, intentChan)
+	setupTrigger("deploy", intents.SetDeploy, intents.SetAutoDeploy, intents.GetAutoDeploy, server.SetDeployCallback, server.SetAutoDeployCallback, intentChan)
 
 	return intents, intentChan
 }
diff --git a/pkg/skaffold/runner/new_test.go b/pkg/skaffold/runner/v1/new_test.go
similarity index 99%
rename from pkg/skaffold/runner/new_test.go
rename to pkg/skaffold/runner/v1/new_test.go
index dbb2e368801..8b150f23f56 100644
--- a/pkg/skaffold/runner/new_test.go
+++ b/pkg/skaffold/runner/v1/new_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"reflect"
diff --git a/pkg/skaffold/runner/portforwarder.go b/pkg/skaffold/runner/v1/portforwarder.go
similarity index 98%
rename from pkg/skaffold/runner/portforwarder.go
rename to pkg/skaffold/runner/v1/portforwarder.go
index 36a9e41118d..afc71abf836 100644
--- a/pkg/skaffold/runner/portforwarder.go
+++ b/pkg/skaffold/runner/v1/portforwarder.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"io"
diff --git a/pkg/skaffold/runner/render.go b/pkg/skaffold/runner/v1/render.go
similarity index 88%
rename from pkg/skaffold/runner/render.go
rename to pkg/skaffold/runner/v1/render.go
index 55e4694f907..4d9e61b8005 100644
--- a/pkg/skaffold/runner/render.go
+++ b/pkg/skaffold/runner/v1/render.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
@@ -25,11 +25,12 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/color"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/docker"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
 )
 
 func (r *SkaffoldRunner) Render(ctx context.Context, out io.Writer, builds []graph.Artifact, offline bool, filepath string) error {
 	// Fetch the digest and append it to the tag with the format of "tag@digest"
-	if r.runCtx.DigestSource() == remoteDigestSource {
+	if r.runCtx.DigestSource() == runner.RemoteDigestSource {
 		for i, a := range builds {
 			digest, err := docker.RemoteDigest(a.Tag, r.runCtx)
 			if err != nil {
@@ -38,7 +39,7 @@ func (r *SkaffoldRunner) Render(ctx context.Context, out io.Writer, builds []gra
 			builds[i].Tag = build.TagWithDigest(a.Tag, digest)
 		}
 	}
-	if r.runCtx.DigestSource() == noneDigestSource {
+	if r.runCtx.DigestSource() == runner.NoneDigestSource {
 		color.Default.Fprintln(out, "--digest-source set to 'none', tags listed in Kubernetes manifests will be used for render")
 	}
 	return r.deployer.Render(ctx, out, builds, offline, filepath)
diff --git a/pkg/skaffold/runner/v1/runner.go b/pkg/skaffold/runner/v1/runner.go
new file mode 100644
index 00000000000..6dc8c140bc0
--- /dev/null
+++ b/pkg/skaffold/runner/v1/runner.go
@@ -0,0 +1,63 @@
+/*
+Copyright 2021 The Skaffold Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License 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 v1
+
+import (
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/build"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/build/cache"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy/label"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/filemon"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/sync"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/test"
+)
+
+// SkaffoldRunner is responsible for running the skaffold build, test and deploy config.
+type SkaffoldRunner struct {
+	runner.Builder
+	runner.Pruner
+	test.Tester
+
+	deployer deploy.Deployer
+	syncer   sync.Syncer
+	monitor  filemon.Monitor
+	listener runner.Listener
+
+	kubectlCLI         *kubectl.CLI
+	cache              cache.Cache
+	changeSet          runner.ChangeSet
+	runCtx             *runcontext.RunContext
+	labeller           *label.DefaultLabeller
+	artifactStore      build.ArtifactStore
+	sourceDependencies graph.SourceDependenciesCache
+	// podSelector is used to determine relevant pods for logging and portForwarding
+	podSelector *kubernetes.ImageList
+
+	devIteration int
+	isLocalImage func(imageName string) (bool, error)
+	hasDeployed  bool
+	intents      *runner.Intents
+}
+
+// HasDeployed returns true if this runner has deployed something.
+func (r *SkaffoldRunner) HasDeployed() bool {
+	return r.hasDeployed
+}
diff --git a/pkg/skaffold/runner/runner_test.go b/pkg/skaffold/runner/v1/runner_test.go
similarity index 94%
rename from pkg/skaffold/runner/runner_test.go
rename to pkg/skaffold/runner/v1/runner_test.go
index 4aeb61d2650..608fa023c22 100644
--- a/pkg/skaffold/runner/runner_test.go
+++ b/pkg/skaffold/runner/v1/runner_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package runner
+package v1
 
 import (
 	"context"
@@ -35,6 +35,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy/kustomize"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/filemon"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/defaults"
 	latest_v1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
@@ -58,8 +59,8 @@ type TestBench struct {
 	testErrors    []error
 	deployErrors  []error
 	namespaces    []string
-	userIntents   []func(*Intents)
-	intents       *Intents
+	userIntents   []func(*runner.Intents)
+	intents       *runner.Intents
 	intentTrigger bool
 
 	devLoop        func(context.Context, io.Writer, func() error) error
@@ -257,9 +258,10 @@ func createRunner(t *testutil.T, testBench *TestBench, monitor filemon.Monitor,
 	runner, err := NewForConfig(runCtx)
 	t.CheckNoError(err)
 
-	runner.builder = testBench
+	// TODO(yuwenma):builder.builder looks weird. Avoid the nested struct.
+	runner.Builder.Builder = testBench
 	runner.syncer = testBench
-	runner.tester = testBench
+	runner.Tester = testBench
 	runner.deployer = testBench
 	runner.listener = testBench
 	runner.monitor = monitor
@@ -427,18 +429,15 @@ func TestNewForConfig(t *testing.T) {
 			}
 
 			cfg, err := NewForConfig(runCtx)
-			t.CheckError(test.shouldErr, err)
-
 			t.CheckError(test.shouldErr, err)
 			if cfg != nil {
-				b, _t, d := WithTimings(&test.expectedBuilder, test.expectedTester, test.expectedDeployer, test.cacheArtifacts)
-
+				b, _t, d := runner.WithTimings(&test.expectedBuilder, test.expectedTester, test.expectedDeployer, test.cacheArtifacts)
 				if test.shouldErr {
 					t.CheckError(true, err)
 				} else {
 					t.CheckNoError(err)
-					t.CheckTypeEquality(b, cfg.builder)
-					t.CheckTypeEquality(_t, cfg.tester)
+					t.CheckTypeEquality(b, cfg.Pruner.Builder)
+					t.CheckTypeEquality(_t, cfg.Tester)
 					t.CheckTypeEquality(d, cfg.deployer)
 				}
 			}
@@ -521,13 +520,14 @@ func TestTriggerCallbackAndIntents(t *testing.T) {
 				Pipelines: runcontext.NewPipelines([]latest_v1.Pipeline{pipeline}),
 			})
 
-			r.intents.resetBuild()
-			r.intents.resetSync()
-			r.intents.resetDeploy()
+			r.intents.ResetBuild()
+			r.intents.ResetSync()
+			r.intents.ResetDeploy()
 
-			t.CheckDeepEqual(test.expectedBuildIntent, r.intents.build)
-			t.CheckDeepEqual(test.expectedSyncIntent, r.intents.sync)
-			t.CheckDeepEqual(test.expectedDeployIntent, r.intents.deploy)
+			build, sync, deploy := r.intents.GetIntentsAttrs()
+			t.CheckDeepEqual(test.expectedBuildIntent, build)
+			t.CheckDeepEqual(test.expectedSyncIntent, sync)
+			t.CheckDeepEqual(test.expectedDeployIntent, deploy)
 		})
 	}
 }
diff --git a/pkg/skaffold/runner/v2/runner.go b/pkg/skaffold/runner/v2/runner.go
index eaf480be5e3..1cdfc457fba 100644
--- a/pkg/skaffold/runner/v2/runner.go
+++ b/pkg/skaffold/runner/v2/runner.go
@@ -17,12 +17,13 @@ package v2
 
 import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/test"
 )
 
 type SkaffoldRunner struct {
 	runner.Builder
 	runner.Pruner
-	runner.Tester
+	test.Tester
 }
 
 func (r *SkaffoldRunner) HasDeployed() bool { return true }