Separate devtunnel lifecycle into create/host/forward phases

Tunnel creation no longer registers ports — hosting starts the relay
without forwarding, and ports are added individually via DevTunnelForward
which restarts the host CLI with -p flags. This fixes the SSH banner
timeout caused by the missing -p flag on `devtunnel host`.

Other changes:
- Remove dead code: conda stubs, FRP visitor stub, kernel conda helper
- Implement CloseDevTunnel API (was a no-op stub)
- Fix KillAll zombie leak by waiting on done channel after Kill
- Add SDKAddPort for registering ports on existing tunnels
- Reorder workflow: create → host → create_session → forward

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: yasithdev

examples/cs-bridge-workflow.yaml

@@ -1,27 +1,32 @@
 name: "cs-bridge-hpc-setup"
 
 steps:
-  - action: "vscode.create_session"
-    name: "Start SSH server"
-    outputs:
-      bind_port: "ssh_port"
-
   - action: "tunnel.devtunnel_create"
     name: "Create devtunnel"
     params:
       tunnel_name: "linkspan-tunnel"
       expiration: "1d"
       auth_token: "{{.TunnelAuthToken}}"
-      ports:
-        - "{{.ssh_port}}"
+    outputs:
+      tunnel_id: "tunnel_id"
 
   - action: "tunnel.devtunnel_host"
     name: "Host devtunnel"
     params:
       tunnel_name: "linkspan-tunnel"
       auth_token: "{{.TunnelAuthToken}}"
+    outputs:
+      connection_url: "tunnel_url"
+      token: "tunnel_token"
 
-  - action: "shell.exec"
-    name: "Clone repo"
+  - action: "vscode.create_session"
+    name: "Start SSH server"
+    outputs:
+      bind_port: "ssh_port"
+
+  - action: "tunnel.devtunnel_forward"
+    name: "Forward SSH port"
     params:
-      command: "git clone https://github.com/org/repo.git /workspace"
+      tunnel_name: "linkspan-tunnel"
+      auth_token: "{{.TunnelAuthToken}}"
+      port: "{{.ssh_port}}"

internal/process/process_manager.go

@@ -176,14 +176,33 @@ func (pm *ProcessManager) Wait(id string) error {
 
 
 
-// KillAll forcefully kills all managed processes.
+// KillAll forcefully kills all managed processes, waiting up to 2 seconds for
+// each to exit so the OS can reap them (avoiding zombies).
 func (pm *ProcessManager) KillAll() {
 	pm.mu.Lock()
-	defer pm.mu.Unlock()
+	// Snapshot the map so we can release the lock before waiting.
+	snapshot := make(map[string]*ManagedProcess, len(pm.procs))
 	for id, mp := range pm.procs {
+		snapshot[id] = mp
+	}
+	pm.mu.Unlock()
+
+	for id, mp := range snapshot {
 		if mp.Cmd.Process != nil {
 			_ = mp.Cmd.Process.Kill()
+
+			// Wait for the process to be reaped via the done channel (which
+			// is closed by the background goroutine in Start after cmd.Wait
+			// completes).  Use a short timeout so KillAll doesn't hang if a
+			// process refuses to exit.
+			select {
+			case <-mp.done:
+			case <-time.After(2 * time.Second):
+			}
 		}
+
+		pm.mu.Lock()
 		delete(pm.procs, id)
+		pm.mu.Unlock()
 	}
 }

internal/workflow/actions.go

@@ -16,6 +16,7 @@ func registerBuiltinActions(r *Registry) {
 	r.Register("vscode.create_session", actionVSCodeCreateSession)
 	r.Register("tunnel.devtunnel_create", actionDevTunnelCreate)
 	r.Register("tunnel.devtunnel_host", actionDevTunnelHost)
+	r.Register("tunnel.devtunnel_forward", actionDevTunnelForward)
 	r.Register("tunnel.devtunnel_delete", actionDevTunnelDelete)
 	r.Register("tunnel.devtunnel_connect", actionDevTunnelConnect)
 	r.Register("tunnel.frp_proxy_create", actionFrpProxyCreate)
@@ -55,9 +56,7 @@ func actionDevTunnelCreate(params map[string]any) (*ActionResult, error) {
 		return nil, fmt.Errorf("tunnel.devtunnel_create: auth_token is required")
 	}
 
-	ports := toIntSlice(params["ports"])
-
-	info, err := tunnel.DevTunnelCreate(tunnelName, expiration, ports, authToken)
+	info, err := tunnel.DevTunnelCreate(tunnelName, expiration, authToken)
 	if err != nil {
 		return nil, err
 	}
@@ -91,6 +90,29 @@ func actionDevTunnelHost(params map[string]any) (*ActionResult, error) {
 	return &result, nil
 }
 
+// --- tunnel.devtunnel_forward ---
+
+func actionDevTunnelForward(params map[string]any) (*ActionResult, error) {
+	tunnelName, _ := params["tunnel_name"].(string)
+	if tunnelName == "" {
+		return nil, fmt.Errorf("tunnel.devtunnel_forward: tunnel_name is required")
+	}
+	authToken, _ := params["auth_token"].(string)
+	if authToken == "" {
+		return nil, fmt.Errorf("tunnel.devtunnel_forward: auth_token is required")
+	}
+	port := toInt(params["port"])
+	if port == 0 {
+		return nil, fmt.Errorf("tunnel.devtunnel_forward: port is required")
+	}
+
+	if err := tunnel.DevTunnelForward(tunnelName, port, authToken); err != nil {
+		return nil, err
+	}
+
+	return &ActionResult{"port": port}, nil
+}
+
 // --- tunnel.devtunnel_delete ---
 
 func actionDevTunnelDelete(params map[string]any) (*ActionResult, error) {

main.go

@@ -178,12 +178,25 @@ func main() {
 		go func() {
 			tunnelName := fmt.Sprintf("aget-tunnel-%d", time.Now().UnixNano())
 
+			// cleanupAttempt kills any host CLI process and removes the tunnel
+			// from the manager so a timed-out or failed attempt doesn't leak.
+			cleanupAttempt := func() {
+				info, err := tunnel.GlobalDevTunnelManager.Find(tunnelName)
+				if err != nil {
+					return // not registered yet, nothing to clean up
+				}
+				if info.HostCmdID != "" {
+					_ = pm.GlobalProcessManager.Kill(info.HostCmdID)
+				}
+				tunnel.GlobalDevTunnelManager.Remove(tunnelName)
+			}
+
 			for attempt := 1; attempt <= *tunnelRetries; attempt++ {
 				log.Printf("devtunnel: attempt %d/%d to create tunnel %s", attempt, *tunnelRetries, tunnelName)
 
 				ch := make(chan error, 1)
 				go func() {
-					_, err := tunnel.DevTunnelCreate(tunnelName, "1d", []int{serverPort}, authToken)
+					_, err := tunnel.DevTunnelCreate(tunnelName, "1d", authToken)
 					if err != nil {
 						ch <- err
 						return
@@ -193,6 +206,10 @@ func main() {
 						ch <- err
 						return
 					}
+					if err := tunnel.DevTunnelForward(tunnelName, serverPort, authToken); err != nil {
+						ch <- err
+						return
+					}
 
 					log.Printf("Connect to agent using the URL: %s", tunnelConnection.ConnectionURL)
 					log.Printf("DevTunnel ID: %s", tunnelConnection.DevTunnelInfo.TunnelID)
@@ -209,10 +226,12 @@ func main() {
 						return
 					}
 					log.Printf("devtunnel: attempt %d failed: %v", attempt, err)
+					cleanupAttempt()
 				case <-attemptCtx.Done():
 					log.Printf("devtunnel: attempt %d timed out after %s", attempt, tunnelAttemptTimeout.String())
+					cancel()
+					cleanupAttempt()
 				}
-				cancel()
 
 				if attempt < *tunnelRetries {
 					time.Sleep(*tunnelRetryDelay)

subsystems/env/conda/conda.go

@@ -71,7 +71,10 @@ func ListKernels(w http.ResponseWriter, r *http.Request) {
 func ProvisionKernel(w http.ResponseWriter, r *http.Request) {
 	// placeholder: parse request body to create kernel
 	provisionReq := KernelProvisionRequest{}
-	_ = json.NewDecoder(r.Body).Decode(&provisionReq)
+	if err := json.NewDecoder(r.Body).Decode(&provisionReq); err != nil {
+		utils.RespondJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()})
+		return
+	}
 	_ = r.Body.Close()
 
 	if provisionReq.KernelName == "" {
@@ -114,7 +117,10 @@ func ProvisionKernel(w http.ResponseWriter, r *http.Request) {
 func ShutdownKernel(w http.ResponseWriter, r *http.Request) {
 	// placeholder: shutdown by id
 	shutdownReq := KernelShutdownRequest{}
-	_ = json.NewDecoder(r.Body).Decode(&shutdownReq)
+	if err := json.NewDecoder(r.Body).Decode(&shutdownReq); err != nil {
+		utils.RespondJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()})
+		return
+	}
 	_ = r.Body.Close()
 
 	err := stopKernel(shutdownReq.KernelID)
@@ -133,7 +139,10 @@ func ShutdownKernel(w http.ResponseWriter, r *http.Request) {
 func DeleteKernel(w http.ResponseWriter, r *http.Request) {
 	// placeholder: delete by id
 	deleteReq := KernelShutdownRequest{}
-	_ = json.NewDecoder(r.Body).Decode(&deleteReq)
+	if err := json.NewDecoder(r.Body).Decode(&deleteReq); err != nil {
+		utils.RespondJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()})
+		return
+	}
 	_ = r.Body.Close()
 
 	err := stopKernel(deleteReq.KernelID)

subsystems/env/conda/conda_test.go

@@ -88,10 +88,6 @@ func getKernelConnectionFile(kernelInternalID string) (string, error) {
 	return "", fmt.Errorf("connection file not found in kernel output: %v", err)
 }
 
-func startKernelWithCondaEnv(kernelName string) (string, int, error) {
-	return "", -1, fmt.Errorf("Not implemented")
-}
-
 func getKernelStatus(kernelInternalID string) (string, error) {
 	// placeholder: get the Jupyter kernel process status
 	info, err := pm.GlobalProcessManager.GetInfo(kernelInternalID)

subsystems/jupyter/api.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"net/http"
 
+	pm "github.com/cyber-shuttle/linkspan/internal/process"
 	utils "github.com/cyber-shuttle/linkspan/utils"
 	"github.com/gorilla/mux"
 )
@@ -20,7 +21,6 @@ type Tunnel struct {
 type DevTunnelCreateRequest struct {
 	TunnelName string `json:"tunnelName"`
 	Expiration string `json:"expiration"`
-	Ports      []int  `json:"ports"`
 	// AuthToken is the Microsoft Entra ID (Azure AD) bearer token used to
 	// authenticate against the Dev Tunnels service.  It is required for all
 	// devtunnel operations.
@@ -74,15 +74,18 @@ func ListDevTunnels(w http.ResponseWriter, r *http.Request) {
 // It creates the tunnel via the SDK and immediately starts hosting it via the CLI.
 func CreateDevTunnel(w http.ResponseWriter, r *http.Request) {
 	var req DevTunnelCreateRequest
-	_ = json.NewDecoder(r.Body).Decode(&req)
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		utils.RespondJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()})
+		return
+	}
 	_ = r.Body.Close()
 
 	if req.AuthToken == "" {
 		utils.RespondJSON(w, http.StatusBadRequest, map[string]string{"error": "authToken is required"})
 		return
 	}
 
-	info, err := DevTunnelCreate(req.TunnelName, req.Expiration, req.Ports, req.AuthToken)
+	info, err := DevTunnelCreate(req.TunnelName, req.Expiration, req.AuthToken)
 	if err != nil {
 		utils.RespondJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
 		return
@@ -101,15 +104,46 @@ func CreateDevTunnel(w http.ResponseWriter, r *http.Request) {
 	})
 }
 
-// CloseDevTunnel handles DELETE /tunnels/devtunnel/{id} (stub).
+// CloseDevTunnel handles DELETE /tunnels/devtunnel/{id}.
+// It kills the host CLI process, deletes the tunnel via the SDK, and removes
+// it from the in-memory manager.
 func CloseDevTunnel(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	tunnelName := vars["id"]
+	if tunnelName == "" {
+		utils.RespondJSON(w, http.StatusBadRequest, map[string]string{"error": "tunnel name required"})
+		return
+	}
+
+	info, err := GlobalDevTunnelManager.Find(tunnelName)
+	if err != nil {
+		utils.RespondJSON(w, http.StatusNotFound, map[string]string{"error": err.Error()})
+		return
+	}
+
+	// Kill the host CLI process if one is running.
+	if info.HostCmdID != "" {
+		_ = pm.GlobalProcessManager.Kill(info.HostCmdID)
+	}
+
+	// Delete the tunnel on the service.
+	if err := DevTunnelDelete(tunnelName, info.AuthToken); err != nil {
+		utils.RespondJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
+		return
+	}
+
+	GlobalDevTunnelManager.Remove(tunnelName)
+
 	utils.RespondJSON(w, http.StatusNoContent, nil)
 }
 
 // CreateFrpTunnelProxy handles POST /tunnels/frp.
 func CreateFrpTunnelProxy(w http.ResponseWriter, r *http.Request) {
 	var req FrpTunnelProxyCreateRequest
-	_ = json.NewDecoder(r.Body).Decode(&req)
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		utils.RespondJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()})
+		return
+	}
 	_ = r.Body.Close()
 
 	info, err := FrpTunnelProxyCreate(
@@ -136,11 +170,6 @@ func ListFrpTunnels(w http.ResponseWriter, r *http.Request) {
 	utils.RespondJSON(w, http.StatusOK, FrpTunnelListResponse{FrpTunnelInfos: ts})
 }
 
-// GetFrpTunnelStatus handles GET /tunnels/frp/{id}/status (stub).
-func GetFrpTunnelStatus(w http.ResponseWriter, r *http.Request) {
-	utils.RespondJSON(w, http.StatusOK, map[string]string{"status": "unknown"})
-}
-
 // TerminateFrpTunnel handles DELETE /tunnels/frp/{id}.
 func TerminateFrpTunnel(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)