diff --git a/experimental/ssh/cmd/agent_shim.go b/experimental/ssh/cmd/agent_shim.go
new file mode 100644
index 00000000000..4eb97189201
--- /dev/null
+++ b/experimental/ssh/cmd/agent_shim.go
@@ -0,0 +1,43 @@
+package ssh
+
+import (
+	"github.com/databricks/cli/cmd/root"
+	"github.com/databricks/cli/experimental/ssh/internal/client"
+	"github.com/databricks/cli/libs/cmdctx"
+	"github.com/spf13/cobra"
+)
+
+func newAgentShimCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:    "agent-shim",
+		Short:  "Launch a coding agent configured against Unity AI Gateway (invoked on the remote)",
+		Hidden: true,
+	}
+	for _, agent := range client.SupportedAgentNames() {
+		cmd.AddCommand(newAgentShimAgentCommand(agent))
+	}
+	return cmd
+}
+
+func newAgentShimAgentCommand(agent string) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   agent,
+		Short: "Launch " + agent + " configured against Unity AI Gateway",
+		// Disable flag parsing: forward everything after the agent name to the agent verbatim.
+		DisableFlagParsing: true,
+	}
+
+	cmd.PreRunE = func(cmd *cobra.Command, args []string) error {
+		// Runs on the driver with injected env auth; no bundle, no prompt.
+		cmd.SetContext(root.SkipLoadBundle(cmd.Context()))
+		cmd.SetContext(root.SkipPrompt(cmd.Context()))
+		return root.MustWorkspaceClient(cmd, args)
+	}
+
+	cmd.RunE = func(cmd *cobra.Command, args []string) error {
+		ctx := cmd.Context()
+		return client.RunAgentShim(ctx, cmdctx.WorkspaceClient(ctx), agent, args)
+	}
+
+	return cmd
+}
diff --git a/experimental/ssh/cmd/ssh.go b/experimental/ssh/cmd/ssh.go
index 9939d2f7050..19517cd6204 100644
--- a/experimental/ssh/cmd/ssh.go
+++ b/experimental/ssh/cmd/ssh.go
@@ -22,6 +22,7 @@ Use ` + "`databricks ssh connect --help`" + ` to see all available flags.`,
 	cmd.AddCommand(newSetupCommand())
 	cmd.AddCommand(newConnectCommand())
 	cmd.AddCommand(newServerCommand())
+	cmd.AddCommand(newAgentShimCommand())
 
 	return cmd
 }
diff --git a/experimental/ssh/internal/client/agentshim.go b/experimental/ssh/internal/client/agentshim.go
new file mode 100644
index 00000000000..ead57dc3ebe
--- /dev/null
+++ b/experimental/ssh/internal/client/agentshim.go
@@ -0,0 +1,657 @@
+package client
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"runtime"
+	"strings"
+	"time"
+
+	"github.com/databricks/cli/libs/auth"
+	"github.com/databricks/cli/libs/cmdio"
+	"github.com/databricks/cli/libs/env"
+	"github.com/databricks/cli/libs/log"
+	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/apierr"
+	"github.com/databricks/databricks-sdk-go/client"
+	"github.com/databricks/databricks-sdk-go/config"
+	"github.com/databricks/databricks-sdk-go/httpclient"
+	"github.com/databricks/databricks-sdk-go/listing"
+	"github.com/databricks/databricks-sdk-go/service/catalog"
+)
+
+const (
+	// home-relative directory to hold files/dependencies for the shim.
+	agentDir = ".agent-shim"
+
+	// directory which holds the per-agent wrappers; must match remoteShimDir in client.go.
+	binDir = agentDir + "/bin"
+
+	// the GitHub repo the shim installs the Unity Gateway CLI from.
+	ugRepo = "databricks/unity-gateway"
+
+	// pins the Unity Gateway CLI release tag the shim installs.
+	ugVersion = "v0.1.0"
+
+	// environment variable used to pass the user's workspace home to the shim for the agent context.
+	workspaceHomeEnv = "DATABRICKS_WORKSPACE_HOME"
+
+	// directory which holds toolchain deps (Node/npm) fetched when the image ships none.
+	depsDir = agentDir + "/deps"
+
+	// the file holding the Databricks session context.
+	contextFile = "agent-system-context.md"
+
+	// lock file to serialize first-run toolchain setup across
+	// concurrent SSH clients sharing this driver's $HOME.
+	setupLockName = ".setup.lock"
+
+	// timeout to reclaim a setup lock left behind by a process that died
+	// mid-install. It is deliberately generous: a cold first run can download uv,
+	// ug, and Node
+	setupLockStaleAfter = 2 * time.Minute
+)
+
+type agentSpec struct {
+	name string
+	// contextFlag passes the context file via this CLI flag (Claude's --append-system-prompt-file).
+	contextFlag string
+	// contextHomeFile writes the context to this $HOME instructions file (codex's AGENTS.md).
+	contextHomeFile string
+}
+
+// Currently limited to the agents whose Unity Gateway CLI command supports --workspace
+var supportedAgents = []agentSpec{
+	{name: "claude", contextFlag: "--append-system-prompt-file"},
+	{name: "codex", contextHomeFile: ".codex/AGENTS.md"},
+}
+
+// SupportedAgentNames lists the agents the ssh command registers a subcommand for.
+func SupportedAgentNames() []string {
+	names := make([]string, len(supportedAgents))
+	for i, a := range supportedAgents {
+		names[i] = a.name
+	}
+	return names
+}
+
+func agentByName(name string) (agentSpec, bool) {
+	for _, a := range supportedAgents {
+		if a.name == name {
+			return a, true
+		}
+	}
+	return agentSpec{}, false
+}
+
+func RunAgentShim(ctx context.Context, client *databricks.WorkspaceClient, agentName string, agentArgs []string) error {
+	agent, ok := agentByName(agentName)
+	if !ok {
+		return fmt.Errorf("unsupported agent %q", agentName)
+	}
+	// Probe first: fail fast before the slow first-run bootstrap if the gateway is off.
+	if err := probeAIGateway(ctx, client); err != nil {
+		return err
+	}
+	workspace := strings.TrimRight(client.Config.Host, "/")
+	return bootstrapAndLaunchAgent(ctx, agent, workspace, agentArgs)
+}
+
+// --- Unity AI Gateway preflight ---
+//
+// Kept in lockstep with Unity Gateway CLI's probe_unity_gateway_capabilities
+// (https://github.com/databricks/unity-gateway/blob/main/src/ucode/databricks.py)
+// Probe the Unity Catalog model-services API (v3) first, paging through it, then
+// fall back to the legacy AI Gateway endpoints API (v2). A path counts only when
+// its JSON body actually lists a usable resource — a 200 with an empty collection
+// is "reachable but empty", not "enabled". We fail fast with the same actionable
+// guidance Unity Gateway CLI surfaces so the shim's preflight matches what Unity
+// Gateway CLI itself checks.
+
+const (
+	legacyEndpointsPath       = "/api/ai-gateway/v2/endpoints"
+	modelServiceProbePageSize = 50
+	modelServiceProbeMaxPages = 20
+	modelServiceProbeMaxItems = modelServiceProbePageSize * modelServiceProbeMaxPages
+	aiGatewayDocsURL          = "https://docs.databricks.com/aws/en/ai-gateway/overview-beta"
+)
+
+type gatewayProbe struct {
+	reachable bool
+	err       error
+	// true if the API returned at least one usable resource
+	resourceAvailable bool
+}
+
+func probeAIGateway(ctx context.Context, wsclient *databricks.WorkspaceClient) error {
+	host := strings.TrimRight(wsclient.Config.Host, "/")
+
+	// build a client with retries disabled so a transient 429/503 fails fast
+	// instead of looping on the SDK's default 5-minute retry budget before the
+	// slow first-run bootstrap even starts.
+	apiClient, err := newProbeAPIClient(wsclient.Config)
+	if err != nil {
+		// this probe is best effort, so don't block
+		log.Debugf(ctx, "Skipping Unity AI Gateway check: %s", err)
+		return nil
+	}
+
+	modelSvc := probeModelServices(ctx, apiClient)
+	// A 401 (or a 400 "invalid token") can't be rescued by trying another API,
+	// so surface it before the fallback probe.
+	if !modelSvc.reachable && looksLikeDefinitiveAuthFailure(modelSvc.err) {
+		return aiGatewayAuthError(host, modelSvc.err.Error())
+	}
+	if modelSvc.resourceAvailable {
+		return nil
+	}
+
+	legacy := probeLegacyEndpoints(ctx, apiClient)
+	switch {
+	case legacy.reachable:
+		// The legacy endpoints API answered, so the gateway is enabled even if no
+		// model services are visible to this caller.
+		if modelSvc.err != nil {
+			log.Warnf(ctx, "Unity AI Gateway model service check: %s", modelSvc.err)
+		}
+		return nil
+	case modelSvc.reachable:
+		// v3 answered; give it the benefit of the doubt rather than blocking on an
+		// unconfirmed "empty".
+		return nil
+	case looksLikeDefinitiveAuthFailure(legacy.err):
+		return aiGatewayAuthError(host, legacy.err.Error())
+	case looksLikeScopeFailure(modelSvc.err):
+		return aiGatewayScopeError(host, modelSvc.err.Error())
+	case looksLikeScopeFailure(legacy.err):
+		return aiGatewayScopeError(host, legacy.err.Error())
+	case looksLikeTransient(modelSvc.err) || looksLikeTransient(legacy.err):
+		// A rate-limit/5xx/network blip is not "disabled" — tell the user to retry
+		// rather than sending them to the enablement docs.
+		return versionNeutralGatewayError(
+			fmt.Sprintf("could not verify the Databricks Unity AI Gateway on %s: the probe hit a transient error (model services: %s; legacy endpoints: %s). Retry in a moment", host, modelSvc.err.Error(), legacy.err.Error()),
+		)
+	case errors.Is(modelSvc.err, apierr.ErrPermissionDenied):
+		return versionNeutralGatewayError(
+			fmt.Sprintf("model service access could not be verified on %s (%s). The legacy endpoint fallback also failed (%s). The model service probe requires permission to list Unity Catalog model services. Verify USE CATALOG on `system`, and USE SCHEMA and EXECUTE on `system.ai`", host, modelSvc.err.Error(), legacy.err.Error()),
+		)
+	case errors.Is(legacy.err, apierr.ErrPermissionDenied):
+		return versionNeutralGatewayError(
+			fmt.Sprintf("legacy endpoint access could not be verified on %s (%s). The model service probe also failed (%s). Verify the caller's workspace permissions for the legacy endpoints listing", host, legacy.err.Error(), modelSvc.err.Error()),
+		)
+	default:
+		return versionNeutralGatewayError(
+			fmt.Sprintf("the Databricks Unity AI Gateway is not enabled on this workspace (%s): neither model services (%s) nor legacy endpoints (%s) are available. See %s", host, modelSvc.err.Error(), legacy.err.Error(), aiGatewayDocsURL),
+		)
+	}
+}
+
+// newProbeAPIClient builds a low-level client from cfg with the SDK's retries
+// turned off, so the gateway preflight is one-shot. Clearing ErrorRetriable and
+// TransientErrors makes a retriable status (429, 503, ...) surface on the first
+// attempt rather than looping until the retry budget expires; it mirrors
+// client.New, which builds the retrying client the rest of the CLI uses.
+func newProbeAPIClient(cfg *config.Config) (*client.DatabricksClient, error) {
+	clientCfg, err := config.HTTPClientConfigFromConfig(cfg)
+	if err != nil {
+		return nil, err
+	}
+	clientCfg.ErrorRetriable = func(context.Context, error) bool { return false }
+	clientCfg.TransientErrors = nil
+	return client.NewWithClient(cfg, httpclient.NewApiClient(clientCfg))
+}
+
+func probeModelServices(ctx context.Context, apiClient *client.DatabricksClient) gatewayProbe {
+	it := catalog.NewAiGateway(apiClient).ListModelServices(ctx, catalog.ListModelServicesRequest{PageSize: modelServiceProbePageSize})
+	_, err := it.Next(ctx)
+	if err != nil {
+		if err == listing.ErrNoMoreItems {
+			// endpoint is reachable but there are no results
+			return gatewayProbe{reachable: true}
+		}
+		return gatewayProbe{err: err}
+	}
+	return gatewayProbe{reachable: true, resourceAvailable: true}
+}
+
+type legacyEndpointsResponse struct {
+	Endpoints []any `json:"endpoints"`
+}
+
+func probeLegacyEndpoints(ctx context.Context, apiClient *client.DatabricksClient) gatewayProbe {
+	var resp legacyEndpointsResponse
+	err := apiClient.Do(ctx, http.MethodGet, legacyEndpointsPath, auth.WorkspaceIDHeaders(apiClient.Config), map[string]any{
+		"page_size": "1",
+	}, nil, &resp)
+	if err != nil {
+		return gatewayProbe{err: err}
+	}
+	if len(resp.Endpoints) > 0 {
+		return gatewayProbe{reachable: true, resourceAvailable: true}
+	}
+	return gatewayProbe{reachable: true}
+}
+
+var (
+	gatewayDetailV3 = regexp.MustCompile(`(?i)\bv3\b`)
+	gatewayDetailV2 = regexp.MustCompile(`(?i)\bv2\b`)
+)
+
+// rewrite the internal v2/v3 API labels in a failure reason into user-facing terms
+func versionNeutralGatewayError(reason string) error {
+	reason = gatewayDetailV3.ReplaceAllString(reason, "model service")
+	reason = gatewayDetailV2.ReplaceAllString(reason, "legacy endpoint")
+	return errors.New(reason)
+}
+
+// returns true when retrying another workspace API can't rescue the token.
+// A bare 403 is left out: it can be endpoint-specific authorization, so the
+// preflight still tries the fallback before giving up.
+func looksLikeDefinitiveAuthFailure(err error) bool {
+	// HTTP 401
+	if errors.Is(err, apierr.ErrUnauthenticated) {
+		return true
+	}
+	return errors.Is(err, apierr.ErrBadRequest) && strings.Contains(strings.ToLower(err.Error()), "invalid token")
+}
+
+// matches a 403 that reports the OAuth token is missing a required scope (which
+// re-login can fix), as opposed to a plain permission 403.
+func looksLikeScopeFailure(err error) bool {
+	reason := strings.ToLower(err.Error())
+	return errors.Is(err, apierr.ErrPermissionDenied) && strings.Contains(reason, "oauth token") && strings.Contains(reason, "required scopes")
+}
+
+func looksLikeTransient(err error) bool {
+	return strings.HasPrefix(err.Error(), "network error") ||
+		errors.Is(err, apierr.ErrTooManyRequests) || // HTTP 429
+		errors.Is(err, apierr.ErrInternalError) || // HTTP 500
+		errors.Is(err, apierr.ErrTemporarilyUnavailable) || // HTTP 503
+		errors.Is(err, apierr.ErrDeadlineExceeded) // HTTP 504
+}
+
+func aiGatewayAuthError(host, reason string) error {
+	return versionNeutralGatewayError(
+		fmt.Sprintf("the Databricks workspace %s rejected the access token (%s). Try:\n  databricks auth logout --host %s\n  databricks auth login --host %s", host, reason, host, host),
+	)
+}
+
+func aiGatewayScopeError(host, reason string) error {
+	return versionNeutralGatewayError(
+		fmt.Sprintf("the access token for %s is missing an OAuth scope required by the AI Gateway APIs (%s). Re-authenticate to mint a token with the needed scopes:\n  databricks auth login --host %s", host, reason, host),
+	)
+}
+
+// --- end Unity AI Gateway preflight ---
+
+func bootstrapAndLaunchAgent(ctx context.Context, agent agentSpec, workspace string, agentArgs []string) error {
+	home, err := env.UserHomeDir(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to resolve home directory: %w", err)
+	}
+
+	// Reconstruct the PATH tooling runs under on every launch rather than caching
+	// it: every entry is a known location, so deriving it can't restore a stale
+	// path (e.g. a versioned CLI dir from an older session). Drop the shim dir so
+	// Unity Gateway CLI execs the real agent rather than this wrapper, then prepend
+	// uv/ug's bin, this databricks CLI's own dir (so it's used by ug/the spawned agent),
+	// and the Node bin ensureNode installs into.
+	removePath(ctx, filepath.Join(home, binDir))
+	prependPath(ctx, filepath.Join(home, ".local", "bin")) // uv
+	if self, err := os.Executable(); err == nil {
+		prependPath(ctx, filepath.Dir(self))
+	}
+	prependPath(ctx, filepath.Join(home, depsDir, "node", "bin"))
+
+	if err := ensureToolchain(ctx, home); err != nil {
+		return err
+	}
+
+	// npm is on PATH now (shipped or just installed); silence its update-notifier
+	// box so it doesn't clutter the agent session.
+	disableNpmUpdateNotifier(ctx)
+
+	// Put npm's global bin on PATH so an npm-installed agent binary resolves after
+	// Unity Gateway CLI installs it.
+	if prefix := npmGlobalPrefix(ctx); prefix != "" {
+		prependPath(ctx, filepath.Join(prefix, "bin"))
+	}
+
+	return launchAgent(ctx, home, agent, workspace, agentArgs)
+}
+
+func toolchainReady() bool {
+	_, ugErr := exec.LookPath("ucode")
+	_, npmErr := exec.LookPath("npm")
+	return ugErr == nil && npmErr == nil
+}
+
+func ensureToolchain(ctx context.Context, home string) error {
+	if toolchainReady() {
+		return nil
+	}
+	unlock, err := acquireSetupLock(ctx, home)
+	if err != nil {
+		return err
+	}
+	defer unlock()
+	// Another client may have finished the install while we waited for the lock.
+	if toolchainReady() {
+		return nil
+	}
+
+	// 1. uv (installs into ~/.local/bin).
+	if _, err := exec.LookPath("uv"); err != nil {
+		cmdio.LogString(ctx, "Installing uv...")
+		if err := runShell(ctx, "curl -LsSf https://astral.sh/uv/install.sh | sh"); err != nil {
+			return fmt.Errorf("failed to install uv: %w", err)
+		}
+	}
+
+	// 2. Unity Gateway CLI (pinned stock upstream release).
+	if _, err := exec.LookPath("ucode"); err != nil {
+		cmdio.LogString(ctx, "Installing Unity Gateway CLI...")
+		if err := runCommand(ctx, "uv", "tool", "install", "git+https://github.com/"+ugRepo+"@"+ugVersion); err != nil {
+			return fmt.Errorf("failed to install Unity Gateway CLI: %w", err)
+		}
+	}
+
+	// 3. Node/npm (installs into depsDir/node/bin)
+	if _, err := exec.LookPath("npm"); err != nil {
+		cmdio.LogString(ctx, "Installing npm...")
+		if _, err := ensureNode(ctx, home); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// take a machine-local lock (an O_EXCL sentinel file) serializing first-run setup
+// across SSH clients. It blocks until the lock is free, reclaiming one left behind
+// by a dead process after setupLockStaleAfter. The returned func releases the lock.
+func acquireSetupLock(ctx context.Context, home string) (func(), error) {
+	lockPath := filepath.Join(home, agentDir, setupLockName)
+	if err := os.MkdirAll(filepath.Dir(lockPath), 0o755); err != nil {
+		return nil, fmt.Errorf("failed to create %s: %w", filepath.Dir(lockPath), err)
+	}
+	for {
+		f, err := os.OpenFile(lockPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o644)
+		if err == nil {
+			_ = f.Close()
+			return func() { _ = os.Remove(lockPath) }, nil
+		}
+		if !errors.Is(err, os.ErrExist) {
+			return nil, fmt.Errorf("failed to acquire setup lock: %w", err)
+		}
+		// Held by another process: reclaim it if stale, otherwise wait and retry.
+		if info, statErr := os.Stat(lockPath); statErr == nil && time.Since(info.ModTime()) > setupLockStaleAfter {
+			log.Warnf(ctx, "reclaiming stale agent-shim setup lock at %s", lockPath)
+			_ = os.Remove(lockPath)
+			continue
+		}
+		log.Infof(ctx, "waiting for a concurrent setup to finish")
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-time.After(time.Second):
+		}
+	}
+}
+
+func launchAgent(ctx context.Context, home string, agent agentSpec, workspace string, agentArgs []string) error {
+	ugPath, err := exec.LookPath("ucode")
+	if err != nil {
+		return fmt.Errorf("the Unity Gateway CLI was not found on PATH after setup: %w", err)
+	}
+	contextArgs, err := injectAgentContext(ctx, home, agent)
+	if err != nil {
+		return err
+	}
+	argv := []string{"ucode", agent.name}
+	if workspace != "" {
+		argv = append(argv, "--workspace", workspace)
+	}
+	argv = append(argv, "--")
+	argv = append(argv, contextArgs...)
+	argv = append(argv, agentArgs...)
+	// Pass the session token as DATABRICKS_BEARER so Unity Gateway CLI authenticates headlessly.
+	if env.Get(ctx, "DATABRICKS_BEARER") == "" {
+		if token := env.Get(ctx, "DATABRICKS_TOKEN"); token != "" {
+			_ = os.Setenv("DATABRICKS_BEARER", token)
+		}
+	}
+	return execProcess(ugPath, argv, os.Environ())
+}
+
+// put the Databricks context where the agent reads it, returning any extra argv.
+func injectAgentContext(ctx context.Context, home string, agent agentSpec) ([]string, error) {
+	systemContext := agentSystemContext(env.Get(ctx, workspaceHomeEnv))
+	switch {
+	case agent.contextFlag != "":
+		// Keep the scratch context file under the shim's own directory so it never
+		// clobbers an unrelated file the user happens to have in their home.
+		dir := filepath.Join(home, agentDir)
+		if err := os.MkdirAll(dir, 0o755); err != nil {
+			return nil, fmt.Errorf("failed to create %s: %w", dir, err)
+		}
+		f := filepath.Join(dir, contextFile)
+		if err := os.WriteFile(f, []byte(systemContext), 0o644); err != nil {
+			return nil, fmt.Errorf("failed to write agent context file: %w", err)
+		}
+		return []string{agent.contextFlag, f}, nil
+	case agent.contextHomeFile != "":
+		f := filepath.Join(home, agent.contextHomeFile)
+		if _, err := os.Stat(f); err == nil {
+			return nil, nil // already present — don't clobber
+		} else if !errors.Is(err, os.ErrNotExist) {
+			return nil, fmt.Errorf("failed to stat %s: %w", f, err)
+		}
+		if err := os.MkdirAll(filepath.Dir(f), 0o755); err != nil {
+			return nil, fmt.Errorf("failed to create %s: %w", filepath.Dir(f), err)
+		}
+		if err := os.WriteFile(f, []byte(systemContext), 0o644); err != nil {
+			return nil, fmt.Errorf("failed to write %s: %w", f, err)
+		}
+		return nil, nil
+	default:
+		return nil, nil
+	}
+}
+
+// download the latest Krypton LTS Node into deps/node once, returning its bin.
+// Linux-only by design: the shim runs on the serverless driver, so the tarball
+// name is hardcoded to linux while nodeDownloadArch guards the arch.
+func ensureNode(ctx context.Context, home string) (string, error) {
+	depsRoot := filepath.Join(home, depsDir)
+	nodeDir := filepath.Join(depsRoot, "node")
+	nodeBin := filepath.Join(nodeDir, "bin")
+	if _, err := os.Stat(filepath.Join(nodeBin, "npm")); err == nil {
+		return nodeBin, nil
+	}
+
+	arch := nodeDownloadArch(runtime.GOARCH)
+	if arch == "" {
+		return "", fmt.Errorf("unsupported architecture for Node download: %s", runtime.GOARCH)
+	}
+	const base = "https://nodejs.org/dist/latest-krypton"
+	tarName, wantSum, err := latestNodeTarball(ctx, base+"/SHASUMS256.txt", arch)
+	if err != nil {
+		return "", err
+	}
+	if err := os.MkdirAll(depsRoot, 0o755); err != nil {
+		return "", fmt.Errorf("failed to create %s: %w", depsRoot, err)
+	}
+	// Download to a temp file, verifying its SHA256 from SHASUMS256.txt before use.
+	tarball, err := downloadVerified(ctx, base+"/"+tarName, wantSum)
+	if err != nil {
+		return "", err
+	}
+	defer os.Remove(tarball)
+	// Extract into a sibling temp dir and atomically rename it into place, so an
+	// interrupted extraction never leaves a half-populated node dir that the
+	// os.Stat(npm) check above would then wrongly accept as a finished install.
+	tmpDir, err := os.MkdirTemp(depsRoot, "node-*")
+	if err != nil {
+		return "", fmt.Errorf("failed to create temp dir: %w", err)
+	}
+	defer os.RemoveAll(tmpDir) // no-op once renamed; cleans up a failed extraction
+	// Node's .tar.xz is the smallest download; extract it with the system tar.
+	if err := runCommand(ctx, "tar", "-xJf", tarball, "--strip-components=1", "-C", tmpDir); err != nil {
+		return "", fmt.Errorf("failed to extract Node.js: %w", err)
+	}
+	// Clear any partial leftover from a previously-interrupted run, then publish
+	// atomically (same-filesystem rename, since tmpDir is a sibling of nodeDir).
+	if err := os.RemoveAll(nodeDir); err != nil {
+		return "", fmt.Errorf("failed to remove %s: %w", nodeDir, err)
+	}
+	if err := os.Rename(tmpDir, nodeDir); err != nil {
+		return "", fmt.Errorf("failed to install Node.js: %w", err)
+	}
+	return nodeBin, nil
+}
+
+// fetch url to a temp file, failing unless its SHA256 matches wantSum. The caller
+// is responsible for removing the returned file.
+func downloadVerified(ctx context.Context, url, wantSum string) (string, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return "", err
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("failed to download %s: %w", url, err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("failed to download %s: HTTP %d", url, resp.StatusCode)
+	}
+
+	f, err := os.CreateTemp("", "node-*.tar.xz")
+	if err != nil {
+		return "", err
+	}
+	sum := sha256.New()
+	if _, err := io.Copy(io.MultiWriter(f, sum), resp.Body); err != nil {
+		f.Close()
+		os.Remove(f.Name())
+		return "", fmt.Errorf("failed to download %s: %w", url, err)
+	}
+	if err := f.Close(); err != nil {
+		os.Remove(f.Name())
+		return "", err
+	}
+	if got := hex.EncodeToString(sum.Sum(nil)); !strings.EqualFold(got, wantSum) {
+		os.Remove(f.Name())
+		return "", fmt.Errorf("checksum mismatch for %s: got %s, want %s", url, got, wantSum)
+	}
+	return f.Name(), nil
+}
+
+func nodeDownloadArch(goarch string) string {
+	switch goarch {
+	case "amd64":
+		return "x64"
+	case "arm64":
+		return "arm64"
+	default:
+		return ""
+	}
+}
+
+func latestNodeTarball(ctx context.Context, shasumsURL, arch string) (name, sum string, err error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, shasumsURL, nil)
+	if err != nil {
+		return "", "", err
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return "", "", fmt.Errorf("failed to fetch Node checksums: %w", err)
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+	if err != nil {
+		return "", "", fmt.Errorf("failed to read Node checksums: %w", err)
+	}
+	re := regexp.MustCompile(`^([0-9a-f]{64})\s+(node-v[0-9.]+-linux-` + regexp.QuoteMeta(arch) + `\.tar\.xz)$`)
+	for line := range strings.SplitSeq(string(body), "\n") {
+		if m := re.FindStringSubmatch(strings.TrimSpace(line)); m != nil {
+			return m[2], m[1], nil
+		}
+	}
+	return "", "", fmt.Errorf("no linux-%s Node tarball found in %s", arch, shasumsURL)
+}
+
+// turn off npm's "new version available" box so it doesn't clutter the installation
+// progress. Best-effort: it's cosmetic, so a failure (e.g. npm not runnable) is
+// logged and ignored rather than blocking the launch.
+func disableNpmUpdateNotifier(ctx context.Context) {
+	if err := exec.CommandContext(ctx, "npm", "config", "set", "update-notifier", "false").Run(); err != nil {
+		log.Debugf(ctx, "failed to disable npm update-notifier: %v", err)
+	}
+}
+
+func npmGlobalPrefix(ctx context.Context) string {
+	out, err := exec.CommandContext(ctx, "npm", "prefix", "-g").Output()
+	if err != nil {
+		return ""
+	}
+	return strings.TrimSpace(string(out))
+}
+
+func runCommand(ctx context.Context, name string, args ...string) error {
+	cmd := exec.CommandContext(ctx, name, args...)
+	cmd.Stdin = os.Stdin
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	return cmd.Run()
+}
+
+func runShell(ctx context.Context, script string) error {
+	cmd := exec.CommandContext(ctx, "sh", "-c", script)
+	cmd.Stdin = os.Stdin
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	return cmd.Run()
+}
+
+func prependPath(ctx context.Context, dir string) {
+	dirs := []string{dir}
+	for _, d := range filepath.SplitList(env.Get(ctx, "PATH")) {
+		if d != dir {
+			dirs = append(dirs, d)
+		}
+	}
+	_ = os.Setenv("PATH", strings.Join(dirs, string(os.PathListSeparator)))
+}
+
+func removePath(ctx context.Context, dir string) {
+	var dirs []string
+	for _, d := range filepath.SplitList(env.Get(ctx, "PATH")) {
+		if d != dir {
+			dirs = append(dirs, d)
+		}
+	}
+	_ = os.Setenv("PATH", strings.Join(dirs, string(os.PathListSeparator)))
+}
+
+func agentSystemContext(wsHome string) string {
+	cwd := "the user's Databricks workspace home directory"
+	if wsHome != "" {
+		cwd = wsHome
+	}
+	return fmt.Sprintf(`You are running inside a "databricks ssh connect" session on the driver node of a
+Databricks serverless cluster.
+- The "databricks" CLI is installed and already authenticated: DATABRICKS_HOST and DATABRICKS_TOKEN are set in the environment, so "databricks ..." commands work with no "databricks auth login". The same token governs Unity Catalog and serving-endpoint access.
+- This container is ephemeral; only paths under /Workspace and /Volumes persist across sessions. Your working directory is %s.
+- You can run shell commands and use the "databricks" CLI to explore the workspace.`, cwd)
+}
diff --git a/experimental/ssh/internal/client/agentshim_test.go b/experimental/ssh/internal/client/agentshim_test.go
new file mode 100644
index 00000000000..3cae09d978c
--- /dev/null
+++ b/experimental/ssh/internal/client/agentshim_test.go
@@ -0,0 +1,354 @@
+package client
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"io/fs"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/config"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPrependPath(t *testing.T) {
+	t.Setenv("PATH", strings.Join([]string{"/a", "/b"}, string(os.PathListSeparator)))
+	prependPath(t.Context(), "/b") // existing entry moves to the front (deduped)
+	assert.Equal(t, []string{"/b", "/a"}, filepath.SplitList(os.Getenv("PATH")))
+	prependPath(t.Context(), "/new")
+	assert.Equal(t, []string{"/new", "/b", "/a"}, filepath.SplitList(os.Getenv("PATH")))
+}
+
+func TestRemovePath(t *testing.T) {
+	t.Setenv("PATH", strings.Join([]string{"/a", "/b", "/a"}, string(os.PathListSeparator)))
+	removePath(t.Context(), "/a")
+	assert.Equal(t, []string{"/b"}, filepath.SplitList(os.Getenv("PATH")))
+}
+
+func TestAcquireSetupLock(t *testing.T) {
+	home := t.TempDir()
+	lockPath := filepath.Join(home, agentDir, setupLockName)
+
+	// Acquire creates the sentinel; release removes it.
+	unlock, err := acquireSetupLock(t.Context(), home)
+	require.NoError(t, err)
+	_, statErr := os.Stat(lockPath)
+	require.NoError(t, statErr, "lock sentinel should exist while held")
+	unlock()
+	_, statErr = os.Stat(lockPath)
+	assert.ErrorIs(t, statErr, fs.ErrNotExist, "lock sentinel should be gone after release")
+
+	// A lock left behind by a dead process is reclaimed once stale, not waited on
+	// forever.
+	require.NoError(t, os.WriteFile(lockPath, nil, 0o644))
+	stale := time.Now().Add(-2 * setupLockStaleAfter)
+	require.NoError(t, os.Chtimes(lockPath, stale, stale))
+	unlock2, err := acquireSetupLock(t.Context(), home)
+	require.NoError(t, err)
+	unlock2()
+}
+
+func TestNodeDownloadArch(t *testing.T) {
+	assert.Equal(t, "x64", nodeDownloadArch("amd64"))
+	assert.Equal(t, "arm64", nodeDownloadArch("arm64"))
+	assert.Empty(t, nodeDownloadArch("mips"))
+}
+
+func TestLatestNodeTarball(t *testing.T) {
+	gzSum, x64Sum, armSum := strings.Repeat("0", 64), strings.Repeat("a", 64), strings.Repeat("b", 64)
+	body := gzSum + "  node-v24.1.0-linux-x64.tar.gz\n" + // .gz is skipped (want .xz)
+		x64Sum + "  node-v24.1.0-linux-x64.tar.xz\n" +
+		armSum + "  node-v24.1.0-linux-arm64.tar.xz\n"
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(body))
+	}))
+	defer srv.Close()
+
+	name, sum, err := latestNodeTarball(t.Context(), srv.URL, "x64")
+	require.NoError(t, err)
+	assert.Equal(t, "node-v24.1.0-linux-x64.tar.xz", name)
+	assert.Equal(t, x64Sum, sum)
+
+	_, _, err = latestNodeTarball(t.Context(), srv.URL, "ppc64le")
+	assert.Error(t, err)
+}
+
+func TestDownloadVerified(t *testing.T) {
+	payload := []byte("fake node tarball")
+	sum := sha256.Sum256(payload)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write(payload)
+	}))
+	defer srv.Close()
+
+	t.Run("matching checksum writes the file", func(t *testing.T) {
+		path, err := downloadVerified(t.Context(), srv.URL, hex.EncodeToString(sum[:]))
+		require.NoError(t, err)
+		defer os.Remove(path)
+		got, err := os.ReadFile(path)
+		require.NoError(t, err)
+		assert.Equal(t, payload, got)
+	})
+
+	t.Run("mismatched checksum errors and leaves no file", func(t *testing.T) {
+		_, err := downloadVerified(t.Context(), srv.URL, strings.Repeat("0", 64))
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "checksum mismatch")
+	})
+}
+
+func TestSupportedAgents(t *testing.T) {
+	names := SupportedAgentNames()
+	assert.Equal(t, []string{"claude", "codex"}, names)
+
+	// Each agent injects context by exactly one mechanism (flag OR home file);
+	// the two are never both set.
+	expect := map[string]struct{ flag, home string }{
+		"claude": {flag: "--append-system-prompt-file"},
+		"codex":  {home: ".codex/AGENTS.md"},
+	}
+	for name, want := range expect {
+		a, ok := agentByName(name)
+		require.True(t, ok)
+		assert.Equal(t, want.flag, a.contextFlag, "%s contextFlag", name)
+		assert.Equal(t, want.home, a.contextHomeFile, "%s contextHomeFile", name)
+	}
+
+	_, ok := agentByName("not-an-agent")
+	assert.False(t, ok)
+}
+
+func TestInjectAgentContext(t *testing.T) {
+	claude, _ := agentByName("claude")
+	codex, _ := agentByName("codex")
+	noContext := agentSpec{name: "none"} // an agent with neither mechanism
+
+	t.Run("flag agent writes a scratch file and returns the flag", func(t *testing.T) {
+		home := t.TempDir()
+		args, err := injectAgentContext(t.Context(), home, claude)
+		require.NoError(t, err)
+		require.Len(t, args, 2)
+		assert.Equal(t, "--append-system-prompt-file", args[0])
+		data, err := os.ReadFile(args[1])
+		require.NoError(t, err)
+		assert.Contains(t, string(data), "databricks ssh connect")
+	})
+
+	t.Run("home-file agent writes the global instructions file, no args", func(t *testing.T) {
+		home := t.TempDir()
+		args, err := injectAgentContext(t.Context(), home, codex)
+		require.NoError(t, err)
+		assert.Nil(t, args)
+		data, err := os.ReadFile(filepath.Join(home, ".codex", "AGENTS.md"))
+		require.NoError(t, err)
+		assert.Contains(t, string(data), "databricks ssh connect")
+	})
+
+	t.Run("home-file agent does not clobber an existing file", func(t *testing.T) {
+		home := t.TempDir()
+		target := filepath.Join(home, ".codex", "AGENTS.md")
+		require.NoError(t, os.MkdirAll(filepath.Dir(target), 0o755))
+		require.NoError(t, os.WriteFile(target, []byte("user's own instructions"), 0o644))
+		args, err := injectAgentContext(t.Context(), home, codex)
+		require.NoError(t, err)
+		assert.Nil(t, args)
+		data, err := os.ReadFile(target)
+		require.NoError(t, err)
+		assert.Equal(t, "user's own instructions", string(data))
+	})
+
+	t.Run("agent without a mechanism writes nothing", func(t *testing.T) {
+		home := t.TempDir()
+		args, err := injectAgentContext(t.Context(), home, noContext)
+		require.NoError(t, err)
+		assert.Nil(t, args)
+		entries, err := os.ReadDir(home)
+		require.NoError(t, err)
+		assert.Empty(t, entries)
+	})
+}
+
+func TestAgentSystemContext(t *testing.T) {
+	assert.Contains(t, agentSystemContext("/Workspace/Users/me@example.com"), "working directory is /Workspace/Users/me@example.com")
+	// Falls back to a generic phrase when the workspace home is unknown.
+	assert.Contains(t, agentSystemContext(""), "working directory is the user's Databricks workspace home directory")
+}
+
+func newProbeClient(t *testing.T, host string) *databricks.WorkspaceClient {
+	t.Helper()
+	w, err := databricks.NewWorkspaceClient((*databricks.Config)(&config.Config{
+		Host:  host,
+		Token: "test-token",
+		// The fake gateway server has no /.well-known/databricks-config, so skip
+		// the metadata fetch that would otherwise log a noisy resolve warning.
+		HostMetadataResolver: func(context.Context, string) (*config.HostMetadata, error) {
+			return nil, nil
+		},
+	}))
+	require.NoError(t, err)
+	return w
+}
+
+// newGatewayServer routes the model-services (v3) and legacy endpoints (v2) probe
+// requests to per-API handlers, 404ing anything else (e.g. the SDK's config
+// discovery). A nil handler means that API is not served (404).
+func newGatewayServer(t *testing.T, modelServices, endpoints http.HandlerFunc) *httptest.Server {
+	t.Helper()
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case strings.HasPrefix(r.URL.Path, "/api/2.1/unity-catalog/model-services") && modelServices != nil:
+			assert.Equal(t, "Bearer test-token", r.Header.Get("Authorization"))
+			modelServices(w, r)
+		case strings.HasPrefix(r.URL.Path, "/api/ai-gateway/v2/endpoints") && endpoints != nil:
+			assert.Equal(t, "Bearer test-token", r.Header.Get("Authorization"))
+			endpoints(w, r)
+		default:
+			w.WriteHeader(http.StatusNotFound)
+			_, _ = w.Write([]byte("{}"))
+		}
+	}))
+}
+
+// jsonHandler replies with status and body for every request it receives.
+func jsonHandler(status int, body string) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(status)
+		_, _ = w.Write([]byte(body))
+	}
+}
+
+func TestProbeAIGateway(t *testing.T) {
+	const modelServicesBody = `{"model_services":[{"name":"model-services/system.ai.gpt-5"}]}`
+	const endpointsBody = `{"endpoints":[{"name":"databricks-gpt-5"}]}`
+	const emptyBody = `{}`
+
+	t.Run("model service available: connected, legacy not probed", func(t *testing.T) {
+		var legacyCalls int
+		srv := newGatewayServer(t,
+			jsonHandler(http.StatusOK, modelServicesBody),
+			func(w http.ResponseWriter, r *http.Request) {
+				legacyCalls++
+				w.WriteHeader(http.StatusOK)
+				_, _ = w.Write([]byte(emptyBody))
+			},
+		)
+		defer srv.Close()
+		require.NoError(t, probeAIGateway(t.Context(), newProbeClient(t, srv.URL)))
+		assert.Zero(t, legacyCalls, "legacy endpoints must not be probed once a model service is found")
+	})
+
+	t.Run("model service found across pages", func(t *testing.T) {
+		srv := newGatewayServer(t, func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusOK)
+			if r.URL.Query().Get("page_token") == "" {
+				_, _ = w.Write([]byte(`{"next_page_token":"cursor-1"}`))
+				return
+			}
+			_, _ = w.Write([]byte(modelServicesBody))
+		}, nil)
+		defer srv.Close()
+		require.NoError(t, probeAIGateway(t.Context(), newProbeClient(t, srv.URL)))
+	})
+
+	t.Run("empty model service but legacy reachable: proceeds", func(t *testing.T) {
+		// v3 200-empty and v2 200-empty: both reachable, no resources → proceed (warn).
+		srv := newGatewayServer(t, jsonHandler(http.StatusOK, emptyBody), jsonHandler(http.StatusOK, emptyBody))
+		defer srv.Close()
+		assert.NoError(t, probeAIGateway(t.Context(), newProbeClient(t, srv.URL)))
+	})
+
+	t.Run("legacy-only workspace (v3 404, v2 reachable): proceeds", func(t *testing.T) {
+		srv := newGatewayServer(t,
+			jsonHandler(http.StatusNotFound, `{"message":"not found"}`),
+			jsonHandler(http.StatusOK, endpointsBody),
+		)
+		defer srv.Close()
+		assert.NoError(t, probeAIGateway(t.Context(), newProbeClient(t, srv.URL)))
+	})
+
+	t.Run("auth failure (401) fails fast without probing legacy", func(t *testing.T) {
+		var legacyCalls int
+		srv := newGatewayServer(t,
+			jsonHandler(http.StatusUnauthorized, `{"message":"unauthorized"}`),
+			func(w http.ResponseWriter, r *http.Request) {
+				legacyCalls++
+				w.WriteHeader(http.StatusOK)
+				_, _ = w.Write([]byte(endpointsBody))
+			},
+		)
+		defer srv.Close()
+		err := probeAIGateway(t.Context(), newProbeClient(t, srv.URL))
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "rejected the access token")
+		assert.Zero(t, legacyCalls, "a definitive auth failure must not fall back to the legacy probe")
+	})
+
+	t.Run("missing OAuth scope on both paths routes to re-auth", func(t *testing.T) {
+		scope := jsonHandler(http.StatusForbidden, `{"error_code":"PERMISSION_DENIED","message":"Provided OAuth token does not have required scopes: unity-catalog"}`)
+		srv := newGatewayServer(t, scope, scope)
+		defer srv.Close()
+		err := probeAIGateway(t.Context(), newProbeClient(t, srv.URL))
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "missing an OAuth scope")
+		assert.Contains(t, err.Error(), "databricks auth login")
+	})
+
+	t.Run("plain 403 on both paths is a permission failure", func(t *testing.T) {
+		forbidden := jsonHandler(http.StatusForbidden, `{"message":"forbidden"}`)
+		srv := newGatewayServer(t, forbidden, forbidden)
+		defer srv.Close()
+		err := probeAIGateway(t.Context(), newProbeClient(t, srv.URL))
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "model service access could not be verified")
+	})
+
+	t.Run("neither gateway available: not enabled", func(t *testing.T) {
+		notFound := jsonHandler(http.StatusNotFound, `{"message":"not found"}`)
+		srv := newGatewayServer(t, notFound, notFound)
+		defer srv.Close()
+		err := probeAIGateway(t.Context(), newProbeClient(t, srv.URL))
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "not enabled on this workspace")
+	})
+
+	t.Run("transient failure (5xx both) is retryable, not disabled", func(t *testing.T) {
+		unavailable := jsonHandler(http.StatusServiceUnavailable, `{"message":"try later"}`)
+		srv := newGatewayServer(t, unavailable, unavailable)
+		defer srv.Close()
+		err := probeAIGateway(t.Context(), newProbeClient(t, srv.URL))
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "transient error")
+		assert.NotContains(t, err.Error(), "not enabled")
+	})
+}
+
+func TestProbeModelServicesPageError(t *testing.T) {
+	// A failure while paging surfaces as a probe error: the iterator can't tell a
+	// first-page failure from a later-page one, so any page error is not treated
+	// as a reachable "empty".
+	srv := newGatewayServer(t, func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Query().Get("page_token") == "" {
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte(`{"next_page_token":"cursor-1"}`))
+			return
+		}
+		w.WriteHeader(http.StatusInternalServerError)
+		_, _ = w.Write([]byte("boom"))
+	}, nil)
+	defer srv.Close()
+
+	apiClient, err := newProbeAPIClient(newProbeClient(t, srv.URL).Config)
+	require.NoError(t, err)
+	probe := probeModelServices(t.Context(), apiClient)
+	assert.False(t, probe.reachable)
+	assert.False(t, probe.resourceAvailable)
+	assert.Error(t, probe.err)
+}
diff --git a/experimental/ssh/internal/client/exec_unix.go b/experimental/ssh/internal/client/exec_unix.go
new file mode 100644
index 00000000000..8c024b980bf
--- /dev/null
+++ b/experimental/ssh/internal/client/exec_unix.go
@@ -0,0 +1,11 @@
+//go:build !windows
+
+package client
+
+import "syscall"
+
+// execProcess replaces the current process (execve(2)) so the ssh PTY connects
+// straight to the agent and its exit code propagates.
+func execProcess(argv0 string, argv, env []string) error {
+	return syscall.Exec(argv0, argv, env)
+}
diff --git a/experimental/ssh/internal/client/exec_windows.go b/experimental/ssh/internal/client/exec_windows.go
new file mode 100644
index 00000000000..3cfc5b2e1ed
--- /dev/null
+++ b/experimental/ssh/internal/client/exec_windows.go
@@ -0,0 +1,10 @@
+//go:build windows
+
+package client
+
+import "errors"
+
+// execProcess is unsupported on Windows (no syscall.Exec); the shim only runs on the Linux driver.
+func execProcess(argv0 string, argv, env []string) error {
+	return errors.New("process exec is not supported on Windows")
+}