diff --git a/libs/localenv/pipeline.go b/libs/localenv/pipeline.go
new file mode 100644
index 00000000000..e3f152d7304
--- /dev/null
+++ b/libs/localenv/pipeline.go
@@ -0,0 +1,537 @@
+package localenv
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/databricks/cli/libs/log"
+	"github.com/hexops/gotextdiff"
+	"github.com/hexops/gotextdiff/myers"
+	"github.com/hexops/gotextdiff/span"
+)
+
+// Filenames and directories the pipeline reads and writes. venvDir is also the
+// virtualenv directory the package manager provisions. pyprojectFile lives in
+// detect.go, the first consumer of it.
+const (
+	backupFile = "pyproject.toml.bak"
+	venvDir    = ".venv"
+)
+
+// artifactSource values reported in --json resolved.artifactSource (spec §6).
+const (
+	artifactNetwork = "network"
+	artifactCache   = "cache"
+)
+
+// allPhases is the canonical phase order for the --json "phases" array. The
+// pipeline seeds every phase pending from this slice and flips each to ok/error
+// as the run progresses.
+var allPhases = []PhaseName{
+	PhasePreflight,
+	PhaseResolve,
+	PhaseFetch,
+	PhaseMerge,
+	PhaseProvision,
+	PhaseValidate,
+}
+
+// Pipeline orchestrates the dbconnect sync phases against a project directory.
+type Pipeline struct {
+	Mode              Mode
+	Check             bool
+	ProjectDir        string
+	ConstraintBaseURL string
+	CacheDir          string
+	Flags             TargetFlags
+	Bundle            BundleTarget
+	Compute           ComputeClient
+	PM                PackageManager
+
+	// res accumulates phase statuses and result fields as the run progresses.
+	res *Result
+}
+
+// Run executes all pipeline phases in order and returns a fully populated Result.
+// On a phase error, Result.Error is set and the same error is also returned. The
+// Result always carries the full canonical phase list: phases completed before a
+// failure are "ok", the failing phase is "error", and the rest are "pending".
+func (p *Pipeline) Run(ctx context.Context) (*Result, error) {
+	log.Debugf(ctx, "local-env: mode=%s check=%v project=%s cacheDir=%s constraintBaseURL=%s flags=%+v",
+		p.Mode,
+		p.Check,
+		filepath.ToSlash(p.ProjectDir),
+		filepath.ToSlash(p.CacheDir),
+		p.ConstraintBaseURL,
+		p.Flags,
+	)
+
+	// NewResult seeds Phases/Warnings to non-nil slices so --json always emits
+	// [] not null; override Phases with the canonical pending phase list.
+	p.res = NewResult()
+	p.res.SchemaVersion = SchemaVersion
+	p.res.Command = CommandName
+	p.res.Mode = p.Mode.String()
+	p.res.DryRun = p.Check
+	// Phases start as pending and flip to ok/error as the run progresses.
+	p.res.Phases = initialPhases()
+
+	if err := p.run(ctx); err != nil {
+		return p.res, err
+	}
+	p.res.OK = true
+	return p.res, nil
+}
+
+// run drives the phases and returns the first phase error. Result bookkeeping
+// (phase status, error object) is handled by fail / markOK.
+func (p *Pipeline) run(ctx context.Context) error {
+	// Phase: preflight — manager detection, writability, package-manager availability.
+	// P0 supports only uv; any other detected manager is a clean, non-blaming exit.
+	if m := detectManager(p.ProjectDir); m != managerUv {
+		return p.fail(PhasePreflight, false, NewError(ErrManagerUnsupported, nil, "%s", managerGuidance(m)))
+	}
+	// Under --check the pipeline only reads and reports a plan, so it must not
+	// mutate anything at preflight. Two preflight steps can write:
+	//   - ensureWritable creates and removes a temp file (and would fail a
+	//     read-only project the user only wants to inspect);
+	//   - PackageManager.EnsureAvailable may install the manager (uv) if missing.
+	// Both exist to fail fast before real writes, which --check never performs, so
+	// they are skipped in a dry run. Neither result is needed to compute the plan.
+	if p.Check {
+		p.markOK(PhasePreflight, "check")
+	} else {
+		if err := ensureWritable(p.ProjectDir); err != nil {
+			return p.fail(PhasePreflight, false, NewError(ErrNotWritable, err, "project directory %s is not writable", filepath.ToSlash(p.ProjectDir)))
+		}
+		version, err := p.PM.EnsureAvailable(ctx)
+		if err != nil {
+			return p.fail(PhasePreflight, false, asPipelineError(err, ErrUvMissing, "%s unavailable", p.PM.Name()))
+		}
+		p.markOK(PhasePreflight, p.PM.Name()+" "+version)
+	}
+
+	// Phase: resolve — compute target → environment key.
+	target, err := p.resolve(ctx)
+	if err != nil {
+		return err
+	}
+
+	// Phase: fetch — constraint artifact for the resolved env key.
+	c, err := p.fetch(ctx, target)
+	if err != nil {
+		return err
+	}
+
+	// Parse the required Python minor from the artifact; a failure here reflects a
+	// bad artifact, reported at the fetch phase.
+	pyMinor, err := PythonMinorFromRequires(c.RequiresPython)
+	if err != nil {
+		return p.fail(PhaseFetch, false, NewError(ErrFetch, err, "cannot parse python version from constraints %q", c.RequiresPython))
+	}
+
+	dbcPin := c.DatabricksConnect
+	if p.Mode == ModeConstraintsOnly {
+		// constraints-only stops *managing* the databricks-connect pin rather than
+		// removing it. Clearing dbcPin means the merge neither injects nor asserts a
+		// pin: greenfield renders dev = [] (no databricks-connect), while an existing
+		// project that already pins databricks-connect keeps its pin untouched (see
+		// mergeDatabricksConnect — an empty value is a no-op, not a deletion).
+		dbcPin = ""
+	}
+	p.res.Resolved = &ResolvedInfo{
+		PythonVersion:    pyMinor,
+		DBConnectVersion: versionFromPin(dbcPin),
+		ArtifactSource:   artifactSource(c.FromCache),
+	}
+
+	// Phase: merge — compute the merged pyproject.toml (in-memory, no writes yet).
+	mergedBytes, greenfield, err := p.mergePlan(ctx, pyMinor, c, dbcPin)
+	if err != nil {
+		return err
+	}
+	p.res.Greenfield = greenfield
+
+	// Check mode stops after planning — nothing below mutates disk.
+	if p.Check {
+		p.markOK(PhaseMerge, "")
+		p.markOK(PhaseProvision, "")
+		p.markOK(PhaseValidate, "")
+		return nil
+	}
+
+	// Apply the merged content to disk (backup first for existing projects).
+	if err := p.applyMerge(ctx, mergedBytes, greenfield); err != nil {
+		return err
+	}
+	p.markOK(PhaseMerge, "")
+
+	// Phase: provision — ensure Python, run uv sync, seed pip.
+	if err := p.provision(ctx, pyMinor); err != nil {
+		return err
+	}
+
+	// Phase: validate — assert the venv matches the target.
+	return p.validate(ctx, pyMinor, dbcPin)
+}
+
+// resolve runs ResolveTarget and records the resolve phase.
+func (p *Pipeline) resolve(ctx context.Context) (*TargetInfo, error) {
+	target, err := ResolveTarget(ctx, p.Flags, p.Compute, p.Bundle)
+	if err != nil {
+		return nil, p.fail(PhaseResolve, false, asPipelineError(err, ErrResolve, "target resolution failed"))
+	}
+	p.res.Target = target
+	p.markOK(PhaseResolve, fmt.Sprintf("source=%s envKey=%s", target.Source, target.EnvKey))
+	return target, nil
+}
+
+// fetch fetches constraints for the resolved target and records the fetch phase.
+func (p *Pipeline) fetch(ctx context.Context, target *TargetInfo) (*Constraints, error) {
+	c, err := FetchConstraints(ctx, p.ConstraintBaseURL, target.EnvKey, p.CacheDir)
+	if err != nil {
+		// FetchConstraints classifies the cause: E_ENV_UNSUPPORTED for a missing
+		// key (404) versus E_FETCH for transport failure with no cache. Both are
+		// discovered here, so both are reported at the fetch phase (the JSON keeps
+		// the failing phase and error.failurePhase in agreement).
+		return nil, p.fail(PhaseFetch, false, asPipelineError(err, ErrFetch, "fetch constraints failed"))
+	}
+	p.markOK(PhaseFetch, fmt.Sprintf("source=%s fromCache=%v", c.SourceURL, c.FromCache))
+	return c, nil
+}
+
+// pyprojectPath returns the path to pyproject.toml in the project directory.
+func (p *Pipeline) pyprojectPath() string {
+	return filepath.Join(p.ProjectDir, pyprojectFile)
+}
+
+// backupPath returns the path to the pyproject.toml backup file.
+func (p *Pipeline) backupPath() string {
+	return filepath.Join(p.ProjectDir, backupFile)
+}
+
+// mergePlan computes the merged pyproject.toml bytes (without writing to disk),
+// decides greenfield vs. existing, and builds the Plan (populated only under
+// --check). dbcPin is the databricks-connect pin to inject, or "" in
+// constraints-only mode.
+func (p *Pipeline) mergePlan(_ context.Context, pyMinor string, c *Constraints, dbcPin string) (merged []byte, greenfield bool, err error) {
+	pyproject := p.pyprojectPath()
+	backup := p.backupPath()
+
+	// The merge base is the live pyproject.toml, not the backup. MergeManaged
+	// rewrites only the three managed regions and preserves every other byte, and
+	// it is idempotent on its own output — so merging onto the current file yields
+	// managed regions identical to merging onto the pristine .bak, but without
+	// discarding edits the user made between syncs (add a dependency, then
+	// re-sync). The .bak stays a one-time safety copy of the pre-sync original, not
+	// the merge base.
+	var baseBytes []byte
+	data, rerr := os.ReadFile(pyproject)
+	switch {
+	case rerr == nil:
+		baseBytes = data
+	case !errors.Is(rerr, os.ErrNotExist):
+		// Only a genuine not-exist means greenfield. Any other read error on an
+		// existing pyproject.toml (permission change, transient I/O, delete race
+		// after detectManager saw it) must not be misread as greenfield — that
+		// would render a fresh file and overwrite the user's project with no
+		// backup. Fail instead of destroying unrecoverable state (invariant 2).
+		return nil, false, p.fail(PhaseMerge, false, NewError(ErrMerge, rerr, "read pyproject.toml %s failed", filepath.ToSlash(pyproject)))
+	}
+	greenfield = baseBytes == nil
+
+	// The artifact drives the merge; in constraints-only mode we clear the
+	// databricks-connect pin so it is neither written nor asserted.
+	effective := *c
+	effective.DatabricksConnect = dbcPin
+
+	var changedRegions []string
+	if greenfield {
+		// No existing pyproject.toml — render a fresh one. The project name comes
+		// from the directory name as a reasonable default.
+		projectName := filepath.Base(p.ProjectDir)
+		merged = RenderFreshPyproject(projectName, effective)
+		changedRegions = []string{regionRequiresPython, regionToolUv}
+		if dbcPin != "" {
+			changedRegions = append(changedRegions, regionDatabricksConnect)
+		}
+	} else {
+		merged, changedRegions, err = MergeManaged(baseBytes, effective)
+		if err != nil {
+			return nil, greenfield, p.fail(PhaseMerge, false, NewError(ErrMerge, err, "merge managed regions failed"))
+		}
+	}
+
+	// Under --check, build the plan (with a diff) for reporting. A real run does
+	// not need the diff.
+	if p.Check {
+		oldStr := ""
+		newStr := string(merged)
+		oldName := pyprojectFile
+		newName := pyprojectFile
+		if !greenfield {
+			oldStr = string(baseBytes)
+			newName = pyprojectFile + ".new"
+		}
+		edits := myers.ComputeEdits(span.URIFromPath(oldName), oldStr, newStr)
+		diff := fmt.Sprint(gotextdiff.ToUnified(oldName, newName, oldStr, edits))
+
+		plan := &Plan{
+			WouldWrite:         filepath.ToSlash(pyproject),
+			Diff:               diff,
+			ChangedRegions:     changedRegions,
+			WouldInstallPython: pyMinor,
+		}
+		// Report a backup only when a real run would actually write one: for an
+		// existing project with no .bak yet. On a re-run the .bak already exists and
+		// applyMerge keeps it (does not re-write), so claiming a backup here would
+		// describe a write that won't happen.
+		if !greenfield {
+			if _, statErr := os.Stat(backup); errors.Is(statErr, os.ErrNotExist) {
+				plan.WouldBackup = filepath.ToSlash(backup)
+			}
+		}
+		p.res.Plan = plan
+	}
+	return merged, greenfield, nil
+}
+
+// applyMerge writes the merged bytes to disk, backing up an existing
+// pyproject.toml first. From this point on, disk has been mutated.
+func (p *Pipeline) applyMerge(_ context.Context, mergedBytes []byte, greenfield bool) error {
+	pyproject := p.pyprojectPath()
+	backup := p.backupPath()
+
+	if !greenfield {
+		// Back up before modifying so the user's original is recoverable
+		// (invariant 2). Only create the backup when one does not already exist:
+		// on a re-run the existing .bak is the canonical original unmanaged state
+		// (mergePlan used it as the base), so overwriting it with the already-merged
+		// pyproject.toml would destroy that baseline.
+		_, statErr := os.Stat(backup)
+		switch {
+		case statErr == nil:
+			// Backup already exists — keep it as the canonical baseline.
+		case errors.Is(statErr, os.ErrNotExist):
+			// copyFile creates/truncates the backup path, so a failure mid-copy may
+			// leave a partial .bak: report disk as mutated.
+			if err := copyFile(pyproject, backup); err != nil {
+				return p.fail(PhaseMerge, true, NewError(ErrMerge, err, "backup pyproject.toml failed"))
+			}
+		default:
+			// An existing-but-unstattable backup must not be overwritten (that would
+			// destroy the recoverable original); fail before any write instead.
+			return p.fail(PhaseMerge, false, NewError(ErrMerge, statErr, "cannot stat backup %s", filepath.ToSlash(backup)))
+		}
+		p.res.BackupPath = filepath.ToSlash(backup)
+	}
+
+	if err := os.WriteFile(pyproject, mergedBytes, 0o644); err != nil {
+		code := ErrMerge
+		if greenfield {
+			code = ErrWrite
+		}
+		return p.fail(PhaseMerge, true, NewError(code, err, "write pyproject.toml failed"))
+	}
+	return nil
+}
+
+// provision ensures the required Python version is installed, runs uv sync, and
+// seeds pip. All three are reported under the provision phase.
+func (p *Pipeline) provision(ctx context.Context, pyMinor string) error {
+	if err := p.PM.EnsurePython(ctx, pyMinor); err != nil {
+		return p.fail(PhaseProvision, true, asPipelineError(err, ErrPythonInstall, "ensure python %s failed", pyMinor))
+	}
+	if err := p.PM.Provision(ctx, p.ProjectDir); err != nil {
+		return p.fail(PhaseProvision, true, asPipelineError(err, ErrProvision, "provision failed"))
+	}
+	if err := p.PM.PostProvision(ctx, p.ProjectDir); err != nil {
+		return p.fail(PhaseProvision, true, asPipelineError(err, ErrProvision, "post-provision failed"))
+	}
+	p.markOK(PhaseProvision, "")
+	return nil
+}
+
+// validate reads the Python and databricks-connect versions from the venv and
+// populates the venv path. dbcPin is "" in constraints-only mode, where the DB
+// Connect assertion is skipped.
+func (p *Pipeline) validate(ctx context.Context, expectedPyMinor, dbcPin string) error {
+	pyVer, dbcVer, err := p.PM.Validate(ctx, p.ProjectDir)
+	if err != nil {
+		return p.fail(PhaseValidate, true, asPipelineError(err, ErrValidate, "validation failed"))
+	}
+
+	// Assert the installed Python minor matches the target.
+	if pyVer != expectedPyMinor {
+		return p.fail(PhaseValidate, true, NewError(ErrValidate, nil,
+			"python version mismatch: want %s, got %s", expectedPyMinor, pyVer))
+	}
+
+	// In default mode, assert the installed databricks-connect major matches the
+	// pin's major. dbcPin is e.g. "databricks-connect~=17.2.0"; dbcVer is "17.2.0".
+	if dbcPin != "" {
+		pinMajor := dbcMajorFromPin(dbcPin)
+		if pinMajor == "" {
+			return p.fail(PhaseValidate, true, NewError(ErrValidate, nil,
+				"cannot determine databricks-connect major version from pin %q", dbcPin))
+		}
+		installedMajor := majorVersion(dbcVer)
+		if installedMajor == "" {
+			return p.fail(PhaseValidate, true, NewError(ErrValidate, nil,
+				"cannot determine installed databricks-connect major version from %q", dbcVer))
+		}
+		if pinMajor != installedMajor {
+			return p.fail(PhaseValidate, true, NewError(ErrValidate, nil,
+				"databricks-connect major version mismatch: want %s.x, got %s", pinMajor, dbcVer))
+		}
+	}
+
+	// Report the installed databricks-connect version only in default mode. In
+	// constraints-only mode databricks-connect is not a managed dependency, so the
+	// spec omits dbconnectVersion even if the package is present transitively.
+	defaultMode := dbcPin != ""
+
+	detail := "python=" + pyVer
+	if defaultMode && dbcVer != "" {
+		detail += " databricks-connect=" + dbcVer
+	}
+	p.markOK(PhaseValidate, detail)
+
+	p.res.VenvPath = filepath.ToSlash(filepath.Join(p.ProjectDir, venvDir))
+	if p.res.Resolved != nil {
+		if defaultMode {
+			p.res.Resolved.DBConnectVersion = dbcVer
+		} else {
+			p.res.Resolved.DBConnectVersion = ""
+		}
+	}
+	return nil
+}
+
+// initialPhases returns the canonical phase list with every phase pending.
+func initialPhases() []PhaseStatus {
+	phases := make([]PhaseStatus, len(allPhases))
+	for i, name := range allPhases {
+		phases[i] = PhaseStatus{Phase: name, Status: StatusPending}
+	}
+	return phases
+}
+
+// markOK marks a phase ok with an optional human-readable detail.
+func (p *Pipeline) markOK(name PhaseName, detail string) {
+	for i := range p.res.Phases {
+		if p.res.Phases[i].Phase == name {
+			p.res.Phases[i].Status = StatusOK
+			p.res.Phases[i].Detail = detail
+			return
+		}
+	}
+}
+
+// fail marks the given phase as errored, attaches the error (with its phase and
+// disk-mutation flag) to the Result, and returns it. Phases after the failing
+// one remain pending.
+func (p *Pipeline) fail(phase PhaseName, diskMutated bool, pe *PipelineError) error {
+	pe.FailurePhase = phase
+	pe.DiskMutated = diskMutated
+	for i := range p.res.Phases {
+		if p.res.Phases[i].Phase == phase {
+			p.res.Phases[i].Status = StatusError
+			p.res.Phases[i].Detail = pe.Error()
+			break
+		}
+	}
+	p.res.Error = pe
+	return pe
+}
+
+// asPipelineError returns err as a *PipelineError if it already is one, otherwise
+// wraps it with the fallback code and message.
+func asPipelineError(err error, fallback ErrorCode, format string, args ...any) *PipelineError {
+	if pe, ok := errors.AsType[*PipelineError](err); ok {
+		return pe
+	}
+	return NewError(fallback, err, format, args...)
+}
+
+// artifactSource maps the from-cache flag to the JSON artifactSource value.
+func artifactSource(fromCache bool) string {
+	if fromCache {
+		return artifactCache
+	}
+	return artifactNetwork
+}
+
+// versionFromPin extracts the bare version from a dependency pin such as
+// "databricks-connect~=17.2.0" → "17.2.0". Returns "" when pin is empty or has
+// no version component.
+func versionFromPin(pin string) string {
+	for i, c := range pin {
+		if c >= '0' && c <= '9' {
+			return pin[i:]
+		}
+	}
+	return ""
+}
+
+// dbcMajorFromPin extracts the major version number from a databricks-connect
+// pin string such as "databricks-connect~=17.2.0". Returns "" if unparseable.
+func dbcMajorFromPin(pin string) string {
+	return majorVersion(versionFromPin(pin))
+}
+
+// majorVersion returns the major portion of a version string (digits before the
+// first dot), e.g. "17" from "17.2.0". A bare integer like "17" returns "17".
+// Returns "" for an empty string or a non-numeric major component, so the
+// validate phase rejects a malformed version rather than comparing arbitrary
+// strings as major versions.
+func majorVersion(v string) string {
+	if v == "" {
+		return ""
+	}
+	before, _, ok := strings.Cut(v, ".")
+	if !ok {
+		before = v
+	}
+	if before == "" || !isAllDigits(before) {
+		return ""
+	}
+	return before
+}
+
+// isAllDigits reports whether s is non-empty and every rune is an ASCII digit.
+func isAllDigits(s string) bool {
+	if s == "" {
+		return false
+	}
+	for _, c := range s {
+		if c < '0' || c > '9' {
+			return false
+		}
+	}
+	return true
+}
+
+// copyFile copies src to dst, creating or overwriting dst. dst is created with
+// src's permission bits: the backup preserves a locked-down pyproject.toml
+// (e.g. 0o600 because it carries a private index URL) rather than widening it to
+// a hardcoded 0o644. os.WriteFile only applies the mode when it creates the
+// file, which is always the case for the freshly-created .bak.
+func copyFile(src, dst string) error {
+	info, err := os.Stat(src)
+	if err != nil {
+		return fmt.Errorf("stat %s: %w", src, err)
+	}
+	data, err := os.ReadFile(src)
+	if err != nil {
+		return fmt.Errorf("read %s: %w", src, err)
+	}
+	if err := os.WriteFile(dst, data, info.Mode().Perm()); err != nil {
+		return fmt.Errorf("write %s: %w", dst, err)
+	}
+	return nil
+}
diff --git a/libs/localenv/pipeline_test.go b/libs/localenv/pipeline_test.go
new file mode 100644
index 00000000000..1a318cd1872
--- /dev/null
+++ b/libs/localenv/pipeline_test.go
@@ -0,0 +1,599 @@
+package localenv
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"runtime"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type fakePM struct{ py, dbc string }
+
+func (fakePM) Name() string                                    { return "fake" }
+func (fakePM) EnsureAvailable(context.Context) (string, error) { return "fake 1.0", nil }
+func (fakePM) EnsurePython(context.Context, string) error      { return nil }
+func (fakePM) Provision(context.Context, string) error         { return nil }
+func (fakePM) PostProvision(context.Context, string) error     { return nil }
+func (f fakePM) Validate(context.Context, string) (string, string, error) {
+	return f.py, f.dbc, nil
+}
+
+// noProvisionPM fails any method that could touch the machine (install the
+// manager, install Python, sync, seed pip, validate). It asserts that --check
+// never reaches those write-side operations.
+type noProvisionPM struct{}
+
+func (noProvisionPM) Name() string { return "noprov" }
+func (noProvisionPM) EnsureAvailable(context.Context) (string, error) {
+	return "", errors.New("EnsureAvailable must not be called under --check")
+}
+
+func (noProvisionPM) EnsurePython(context.Context, string) error {
+	return errors.New("EnsurePython must not be called under --check")
+}
+
+func (noProvisionPM) Provision(context.Context, string) error {
+	return errors.New("Provision must not be called under --check")
+}
+
+func (noProvisionPM) PostProvision(context.Context, string) error {
+	return errors.New("PostProvision must not be called under --check")
+}
+
+func (noProvisionPM) Validate(context.Context, string) (string, string, error) {
+	return "", "", errors.New("Validate must not be called under --check")
+}
+
+// uvMissingPM fails EnsureAvailable, simulating a machine where the package
+// manager binary is absent and cannot be installed. The remaining methods
+// inherit fakePM but must never be reached, since preflight aborts first.
+type uvMissingPM struct{ fakePM }
+
+func (uvMissingPM) EnsureAvailable(context.Context) (string, error) {
+	return "", errors.New("uv not found and install failed")
+}
+
+func writeProject(t *testing.T) string {
+	dir := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml"), []byte(`[project]
+name = "demo"
+requires-python = ">=3.10"
+
+[dependency-groups]
+dev = ["databricks-connect~=16.0.0"]
+`), 0o644))
+	return dir
+}
+
+func newTestServer(t *testing.T) *httptest.Server {
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(sampleToml))
+	}))
+}
+
+func TestPipelineCheckMutatesNothing(t *testing.T) {
+	dir := writeProject(t)
+	before, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml"))
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, Check: true, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	require.NoError(t, err)
+	assert.True(t, res.DryRun)
+	assert.True(t, res.OK)
+	require.NotNil(t, res.Plan)
+	assert.Contains(t, res.Plan.Diff, "==3.12.*")
+	after, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml"))
+	assert.Equal(t, string(before), string(after)) // unchanged
+}
+
+func TestPipelineCheckReRunPlanMatchesRealRun(t *testing.T) {
+	// On a re-run where the .bak already exists and the live file already equals
+	// the merged output, --check must report a plan a real run would perform: no
+	// backup (the .bak is kept, not rewritten) and an empty diff (nothing changes).
+	dir := writeProject(t)
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	newPipe := func(check bool) *Pipeline {
+		return &Pipeline{
+			Mode: ModeDefault, Check: check, ProjectDir: dir,
+			ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+			Flags:   TargetFlags{Serverless: "v4"},
+			Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+		}
+	}
+
+	// First a real sync: writes the merged file and creates the .bak.
+	_, err := newPipe(false).Run(t.Context())
+	require.NoError(t, err)
+	require.FileExists(t, filepath.Join(dir, "pyproject.toml.bak"))
+
+	// Now --check on the already-synced project.
+	res, err := newPipe(true).Run(t.Context())
+	require.NoError(t, err)
+	require.NotNil(t, res.Plan)
+	assert.Empty(t, res.Plan.WouldBackup, "a re-run keeps the existing .bak, so --check must not claim a backup")
+	assert.Empty(t, res.Plan.Diff, "the live file already equals the merged output; the diff must be empty")
+}
+
+func TestPipelineCheckDoesNotProvision(t *testing.T) {
+	// --check must not call any PackageManager method that could mutate the
+	// machine (EnsureAvailable may install uv). noProvisionPM errors on all of
+	// them; the dry run must still succeed and produce a plan.
+	dir := writeProject(t)
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, Check: true, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: noProvisionPM{},
+	}
+	res, err := p.Run(t.Context())
+	require.NoError(t, err)
+	assert.True(t, res.OK)
+	require.NotNil(t, res.Plan)
+}
+
+func TestPipelineCheckWorksOnReadOnlyDir(t *testing.T) {
+	if runtime.GOOS == "windows" || os.Getuid() == 0 {
+		t.Skip("read-only-dir enforcement not available")
+	}
+	dir := writeProject(t)
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	// Make the project dir read-only: --check must still compute the plan without
+	// a writability probe (which would both mutate disk and fail here).
+	require.NoError(t, os.Chmod(dir, 0o555))
+	t.Cleanup(func() { _ = os.Chmod(dir, 0o755) })
+
+	p := &Pipeline{
+		Mode: ModeDefault, Check: true, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	require.NoError(t, err)
+	assert.True(t, res.OK)
+	require.NotNil(t, res.Plan)
+	// No writecheck temp file was left behind in the project dir.
+	entries, err := os.ReadDir(dir)
+	require.NoError(t, err)
+	for _, e := range entries {
+		assert.NotContains(t, e.Name(), "writecheck", "dry run must not create temp files")
+	}
+}
+
+func TestPipelineProvisionsAndValidatesExisting(t *testing.T) {
+	dir := writeProject(t)
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	require.NoError(t, err)
+	assert.True(t, res.OK)
+	assert.False(t, res.Greenfield)
+	require.NotNil(t, res.Resolved)
+	assert.Equal(t, "3.12", res.Resolved.PythonVersion)
+	assert.Equal(t, "17.2.0", res.Resolved.DBConnectVersion)
+	assert.Equal(t, ".venv", filepath.Base(res.VenvPath))
+	merged, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml"))
+	assert.Contains(t, string(merged), `"databricks-connect~=17.2.0"`)
+	assert.FileExists(t, filepath.Join(dir, "pyproject.toml.bak"))
+}
+
+func TestPipelineGreenfieldCreatesNewPyproject(t *testing.T) {
+	dir := t.TempDir()
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	require.NoError(t, err)
+	assert.True(t, res.OK)
+	assert.True(t, res.Greenfield)
+	data, readErr := os.ReadFile(filepath.Join(dir, "pyproject.toml"))
+	require.NoError(t, readErr)
+	assert.Contains(t, string(data), `"databricks-connect~=17.2.0",`)
+	// No backup created when pyproject.toml did not previously exist.
+	assert.NoFileExists(t, filepath.Join(dir, "pyproject.toml.bak"))
+}
+
+func TestPipelineExistingBacksUp(t *testing.T) {
+	dir := writeProject(t)
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	require.NoError(t, err)
+	assert.True(t, res.OK)
+	assert.FileExists(t, filepath.Join(dir, "pyproject.toml.bak"))
+	assert.Equal(t, "pyproject.toml.bak", filepath.Base(res.BackupPath))
+}
+
+func TestCopyFilePreservesMode(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		// Windows does not honor Unix permission bits.
+		t.Skip("permission-bit preservation is Unix-only")
+	}
+	// A locked-down pyproject.toml (e.g. 0o600 because it carries a private index
+	// URL) must not be widened when copied to the backup.
+	dir := t.TempDir()
+	src := filepath.Join(dir, "pyproject.toml")
+	require.NoError(t, os.WriteFile(src, []byte("[project]\n"), 0o600))
+	dst := filepath.Join(dir, "pyproject.toml.bak")
+	require.NoError(t, copyFile(src, dst))
+	info, err := os.Stat(dst)
+	require.NoError(t, err)
+	assert.Equal(t, os.FileMode(0o600), info.Mode().Perm())
+}
+
+// assertPreflightFailure checks the invariants shared by every preflight exit:
+// the run returns the error, preflight is attributed as the failing phase, no
+// disk mutation is claimed, and every later phase is left pending.
+func assertPreflightFailure(t *testing.T, res *Result, err error, wantCode ErrorCode) {
+	t.Helper()
+	var pe *PipelineError
+	require.ErrorAs(t, err, &pe)
+	assert.Equal(t, wantCode, pe.Code)
+	assert.Equal(t, PhasePreflight, pe.FailurePhase)
+	assert.False(t, pe.DiskMutated, "preflight fails before any write")
+	assert.False(t, res.OK)
+	for _, ph := range res.Phases {
+		switch ph.Phase {
+		case PhasePreflight:
+			assert.Equal(t, StatusError, ph.Status)
+		default:
+			assert.Equal(t, StatusPending, ph.Status, "phase %s must stay pending after a preflight failure", ph.Phase)
+		}
+	}
+}
+
+func TestPipelineManagerUnsupportedFailsAtPreflight(t *testing.T) {
+	// A non-uv project (environment.yml, no pyproject.toml) is detected as conda
+	// and must exit cleanly at preflight with E_MANAGER_UNSUPPORTED.
+	dir := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "environment.yml"), []byte("name: demo\n"), 0o644))
+
+	p := &Pipeline{
+		Mode: ModeDefault, ProjectDir: dir, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	assertPreflightFailure(t, res, err, ErrManagerUnsupported)
+	// Detection is pure-fs; no pyproject.toml was created.
+	assert.NoFileExists(t, filepath.Join(dir, "pyproject.toml"))
+}
+
+func TestPipelineUvMissingFailsAtPreflight(t *testing.T) {
+	// When the package manager can't be made available, preflight exits with
+	// E_UV_MISSING before resolve/fetch/merge run.
+	dir := writeProject(t)
+
+	p := &Pipeline{
+		Mode: ModeDefault, ProjectDir: dir, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: uvMissingPM{},
+	}
+	res, err := p.Run(t.Context())
+	assertPreflightFailure(t, res, err, ErrUvMissing)
+}
+
+func TestApplyMergeFailsOnUnstattableBackupWithoutOverwrite(t *testing.T) {
+	if runtime.GOOS == "windows" || os.Getuid() == 0 {
+		// chmod-based stat blocking does not apply for root or on Windows.
+		t.Skip("stat-permission enforcement not available")
+	}
+	// Both pyproject.toml and its .bak live in a project dir that is made
+	// unsearchable, so os.Stat of the backup fails with a permission error rather
+	// than not-exist — isolating applyMerge's "can't stat" branch. applyMerge is
+	// called directly, bypassing the writability preflight.
+	dir := filepath.Join(t.TempDir(), "proj")
+	require.NoError(t, os.Mkdir(dir, 0o755))
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml"), []byte("[project]\n"), 0o644))
+	backup := filepath.Join(dir, "pyproject.toml.bak")
+	require.NoError(t, os.WriteFile(backup, []byte("ORIGINAL BACKUP\n"), 0o644))
+	require.NoError(t, os.Chmod(dir, 0o000))
+	t.Cleanup(func() { _ = os.Chmod(dir, 0o755) })
+
+	p := &Pipeline{ProjectDir: dir, res: &Result{Phases: initialPhases()}}
+	err := p.applyMerge(t.Context(), []byte("merged"), false)
+	require.Error(t, err)
+	var pe *PipelineError
+	require.ErrorAs(t, err, &pe)
+	assert.Equal(t, PhaseMerge, pe.FailurePhase)
+	assert.False(t, pe.DiskMutated, "no write should have happened before the stat check")
+
+	// The original backup must be intact.
+	require.NoError(t, os.Chmod(dir, 0o755))
+	got, readErr := os.ReadFile(backup)
+	require.NoError(t, readErr)
+	assert.Equal(t, "ORIGINAL BACKUP\n", string(got), "the canonical backup must not be overwritten")
+}
+
+func TestPipelineConstraintsOnlyOmitsDBConnect(t *testing.T) {
+	dir := t.TempDir()
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	// Even though databricks-connect is present in the venv (fakePM reports a
+	// version), constraints-only must not assert it, not write the pin, and not
+	// report dbconnectVersion (spec §6 omits it in this mode).
+	p := &Pipeline{
+		Mode: ModeConstraintsOnly, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	require.NoError(t, err)
+	assert.True(t, res.OK)
+	assert.Equal(t, "constraints-only", res.Mode)
+	require.NotNil(t, res.Resolved)
+	assert.Empty(t, res.Resolved.DBConnectVersion, "constraints-only must omit dbconnectVersion")
+	data, readErr := os.ReadFile(filepath.Join(dir, "pyproject.toml"))
+	require.NoError(t, readErr)
+	assert.NotContains(t, string(data), "databricks-connect")
+	// The dev group is emitted empty, not with a blank-string entry.
+	assert.Contains(t, string(data), "dev = []")
+	assert.NotContains(t, string(data), `""`)
+}
+
+func TestPipelineNoTarget(t *testing.T) {
+	dir := writeProject(t)
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{},
+		Compute: stubCompute{}, PM: fakePM{},
+	}
+	res, err := p.Run(t.Context())
+	require.Error(t, err)
+	require.NotNil(t, res)
+	require.NotNil(t, res.Error)
+	assert.False(t, res.OK)
+	assert.Equal(t, ErrNoTarget, res.Error.Code)
+	// No-target is detected during target resolution, so it is reported at the
+	// resolve phase (the failing phase in phases[] and error.failurePhase agree).
+	assert.Equal(t, PhaseResolve, res.Error.FailurePhase)
+	// Preflight succeeded, resolve errored, everything after stays pending.
+	assert.Equal(t, StatusOK, phaseStatus(res, PhasePreflight))
+	assert.Equal(t, StatusError, phaseStatus(res, PhaseResolve))
+	assert.Equal(t, StatusPending, phaseStatus(res, PhaseProvision))
+}
+
+func TestPipelineMergesOnLiveFileNotBackup(t *testing.T) {
+	dir := t.TempDir()
+	// A pre-existing .bak holds the pristine pre-sync original (no user edit).
+	original := []byte(`[project]
+name = "demo"
+requires-python = ">=3.9"
+
+[dependency-groups]
+dev = ["databricks-connect~=15.0.0"]
+`)
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml.bak"), original, 0o644))
+	// The live pyproject.toml is a prior sync's output plus an edit the developer
+	// made afterwards: a non-managed dependency the .bak never saw.
+	live := []byte(`[project]
+name = "demo"
+requires-python = "==3.12.*"
+dependencies = ["rich"]
+
+[dependency-groups]
+dev = ["databricks-connect~=17.2.0"]
+`)
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "pyproject.toml"), live, 0o644))
+
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	require.NoError(t, err)
+	require.NotNil(t, res)
+	data, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml"))
+	// The between-sync user edit survives: the merge based on the live file, not
+	// the stale .bak, which would have discarded it.
+	assert.Contains(t, string(data), `dependencies = ["rich"]`)
+	// Managed regions are still applied.
+	assert.Contains(t, string(data), `"databricks-connect~=17.2.0"`)
+	assert.Contains(t, string(data), `requires-python = "==3.12.*"`)
+	// The .bak is left as the one-time original safety copy, not overwritten.
+	bak, _ := os.ReadFile(filepath.Join(dir, "pyproject.toml.bak"))
+	assert.Equal(t, string(original), string(bak))
+}
+
+func TestPipelineUnreadableExistingIsNotTreatedAsGreenfield(t *testing.T) {
+	if runtime.GOOS == "windows" || os.Getuid() == 0 {
+		// chmod 000 does not block reads for root or on Windows.
+		t.Skip("read-permission enforcement not available")
+	}
+	dir := t.TempDir()
+	pyproject := filepath.Join(dir, "pyproject.toml")
+	original := []byte(`[project]
+name = "demo"
+requires-python = ">=3.10"
+
+[dependency-groups]
+dev = ["databricks-connect~=16.0.0"]
+`)
+	require.NoError(t, os.WriteFile(pyproject, original, 0o644))
+	// Make it unreadable so os.ReadFile fails with a non-not-exist error after
+	// detectManager has already seen the file exists.
+	require.NoError(t, os.Chmod(pyproject, 0o000))
+	t.Cleanup(func() { _ = os.Chmod(pyproject, 0o644) })
+
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	// The run must fail at merge rather than silently overwrite the user's file.
+	require.Error(t, err)
+	var pe *PipelineError
+	require.ErrorAs(t, err, &pe)
+	assert.Equal(t, ErrMerge, pe.Code)
+	assert.Equal(t, PhaseMerge, pe.FailurePhase)
+	assert.False(t, res.Greenfield, "an unreadable existing file must not be treated as greenfield")
+
+	// The original file must be untouched (still unreadable, not overwritten).
+	require.NoError(t, os.Chmod(pyproject, 0o644))
+	after, readErr := os.ReadFile(pyproject)
+	require.NoError(t, readErr)
+	assert.Equal(t, string(original), string(after), "the user's pyproject.toml must not be overwritten")
+}
+
+func TestPipelineResultPopulatesResolved(t *testing.T) {
+	dir := writeProject(t)
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, Check: true, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	require.NoError(t, err)
+	require.NotNil(t, res.Resolved)
+	assert.Equal(t, "3.12", res.Resolved.PythonVersion)
+	assert.Equal(t, "17.2.0", res.Resolved.DBConnectVersion)
+	assert.Equal(t, "network", res.Resolved.ArtifactSource)
+}
+
+// newServerWithDBC returns a test server that serves a constraints TOML with the
+// given databricks-connect pin value in the dev dependency group.
+func newServerWithDBC(t *testing.T, dbcPin string) *httptest.Server {
+	t.Helper()
+	body := `[project]
+requires-python = "==3.12.*"
+
+[dependency-groups]
+dev = ["` + dbcPin + `"]
+
+[tool.uv]
+constraint-dependencies = []
+`
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(body))
+	}))
+}
+
+func TestPipelineValidateRejectsUnparseablePin(t *testing.T) {
+	dir := writeProject(t)
+	// Serve a TOML whose dev group has a malformed databricks-connect entry
+	// (no version digits after the package name).
+	srv := newServerWithDBC(t, "databricks-connect")
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: "17.2.0"},
+	}
+	res, err := p.Run(t.Context())
+	require.Error(t, err)
+	require.NotNil(t, res.Error)
+	assert.Equal(t, ErrValidate, res.Error.Code)
+	assert.Equal(t, PhaseValidate, res.Error.FailurePhase)
+}
+
+func TestPipelineValidateRejectsUnparseableInstalledVersion(t *testing.T) {
+	dir := writeProject(t)
+	// sampleToml has databricks-connect~=17.2.0 as the pin; use an empty installed
+	// version string to simulate an installed version that can't be parsed.
+	srv := newTestServer(t)
+	defer srv.Close()
+
+	p := &Pipeline{
+		Mode: ModeDefault, ProjectDir: dir,
+		ConstraintBaseURL: srv.URL, CacheDir: t.TempDir(),
+		Flags:   TargetFlags{Serverless: "v4"},
+		Compute: stubCompute{}, PM: fakePM{py: "3.12", dbc: ""},
+	}
+	res, err := p.Run(t.Context())
+	require.Error(t, err)
+	require.NotNil(t, res.Error)
+	assert.Equal(t, ErrValidate, res.Error.Code)
+}
+
+func TestMajorVersion(t *testing.T) {
+	cases := []struct {
+		input string
+		want  string
+	}{
+		{"17.2.0", "17"},
+		{"17", "17"},
+		{"", ""},
+		{"3.12", "3"},
+		// Non-numeric major components are rejected (empty) so validation does
+		// not compare arbitrary strings as versions.
+		{"bad.version", ""},
+		{"v17.2", ""},
+		{"abc", ""},
+	}
+	for _, tc := range cases {
+		assert.Equal(t, tc.want, majorVersion(tc.input), "input=%q", tc.input)
+	}
+}
+
+// phaseStatus returns the status recorded for the named phase in res.
+func phaseStatus(res *Result, name PhaseName) string {
+	for _, ph := range res.Phases {
+		if ph.Phase == name {
+			return ph.Status
+		}
+	}
+	return ""
+}