Repository: edgedb/edgedb-cli
Pull request #1475 (2025-02-14 18:42:55)

Watch parallel execution
Makes gel watch execute scripts in parallel instead of sequentially.




Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 70c44c8..258eed8 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -128,7 +128,7 @@ zip = "2.2.0"
 libflate = "2.1.0"
 open = "5.3.0"
 tokio = {version="1.23.0",features=[
-    "macros", "rt", "rt-multi-thread", "fs", "process", "io-std", "net",
+    "macros", "rt", "rt-multi-thread", "fs", "process", "io-std", "net", "signal"
 ]}
 dissimilar = "1.0.6"
 notify = "8.0"
diff --git a/src/branch/mod.rs b/src/branch/mod.rs
index 10137a8..921ee8e 100644
--- a/src/branch/mod.rs
+++ b/src/branch/mod.rs
@@ -11,28 +11,20 @@ pub mod switch;
 pub mod wipe;
 
 use crate::branding::BRANDING;
-use crate::commands::parser::BranchingCmd;
 use crate::commands::Options;
 use crate::connect::{Connection, Connector};
 use crate::options::ConnectionOptions;
-use crate::portable;
 
-#[tokio::main(flavor = "current_thread")]
-pub async fn run(options: &Options, cmd: &Command) -> anyhow::Result<CommandResult> {
-    do_run(&cmd.subcommand, options, None, cmd.conn.instance.as_ref()).await
-}
-
-pub async fn do_run(
+pub async fn run(
     cmd: &Subcommand,
     options: &Options,
-    connection: Option<&mut Connection>,
-    instance_arg: Option<&portable::options::InstanceName>,
+    conn: Option<&mut Connection>,
 ) -> anyhow::Result<CommandResult> {
-    let context = context::Context::new(instance_arg).await?;
+    let context = context::Context::new(options.instance_name.as_ref()).await?;
 
     let mut connector: Connector = options.conn_params.clone();
 
-    // commands that don't need existing connection
+    // commands that don't need connection
     match &cmd {
         Subcommand::Switch(switch) => return switch::run(switch, &context, &mut connector).await,
         Subcommand::Wipe(wipe) => {
@@ -42,25 +34,25 @@ pub async fn do_run(
         _ => {}
     }
 
-    // ensure connected
-    let mut conn;
-    let conn_ref = if let Some(c) = connection {
-        c
+    // connect
+    let mut conn_cell;
+    let conn = if let Some(conn) = conn {
+        conn
     } else {
-        conn = Some(connector.connect().await?);
-        conn.as_mut().unwrap()
+        conn_cell = options.conn_params.connect().await?;
+        &mut conn_cell
     };
 
-    verify_server_can_use_branches(conn_ref).await?;
+    verify_server_can_use_branches(conn).await?;
 
     match cmd {
-        Subcommand::Current(cmd) => current::run(cmd, &context, conn_ref).await?,
-        Subcommand::Create(cmd) => create::run(cmd, &context, conn_ref).await?,
-        Subcommand::Drop(cmd) => drop::main(cmd, &context, conn_ref).await?,
-        Subcommand::List(cmd) => list::main(cmd, &context, conn_ref).await?,
-        Subcommand::Rename(cmd) => return rename::run(cmd, &context, conn_ref, options).await,
-        Subcommand::Rebase(cmd) => rebase::main(cmd, &context, conn_ref, options).await?,
-        Subcommand::Merge(cmd) => merge::main(cmd, &context, conn_ref, options).await?,
+        Subcommand::Current(cmd) => current::run(cmd, &context, conn).await?,
+        Subcommand::Create(cmd) => create::run(cmd, &context, conn).await?,
+        Subcommand::Drop(cmd) => drop::main(cmd, &context, conn).await?,
+        Subcommand::List(cmd) => list::main(cmd, &context, conn).await?,
+        Subcommand::Rename(cmd) => return rename::run(cmd, &context, conn, options).await,
+        Subcommand::Rebase(cmd) => rebase::main(cmd, &context, conn, options).await?,
+        Subcommand::Merge(cmd) => merge::main(cmd, &context, conn, options).await?,
 
         // handled earlier
         Subcommand::Switch(_) | Subcommand::Wipe(_) => unreachable!(),
@@ -107,16 +99,3 @@ pub async fn verify_server_can_use_branches(connection: &mut Connection) -> anyh
 
     Ok(())
 }
-
-impl From<BranchingCmd> for Subcommand {
-    fn from(cmd: BranchingCmd) -> Self {
-        match cmd {
-            BranchingCmd::Create(args) => Subcommand::Create(args),
-            BranchingCmd::Drop(args) => Subcommand::Drop(args),
-            BranchingCmd::Wipe(args) => Subcommand::Wipe(args),
-            BranchingCmd::List(args) => Subcommand::List(args),
-            BranchingCmd::Switch(args) => Subcommand::Switch(args),
-            BranchingCmd::Rename(args) => Subcommand::Rename(args),
-        }
-    }
-}
diff --git a/src/cli/gen_completions.rs b/src/cli/gen_completions.rs
new file mode 100644
index 0000000..715a0be
--- /dev/null
+++ b/src/cli/gen_completions.rs
@@ -0,0 +1,118 @@
+use std::fs;
+use std::io::{stdout, BufWriter, Write};
+use std::path::{Path, PathBuf};
+use std::str::FromStr;
+
+use fn_error_context::context;
+
+use crate::branding::BRANDING_CLI_CMD;
+use crate::options::Options;
+use crate::platform::{current_exe, home_dir};
+
+#[derive(clap::Args, Clone, Debug)]
+pub struct Command {
+    /// Shell to print out completions for
+    #[arg(long)]
+    pub shell: Option<Shell>,
+
+    /// Install all completions into the prefix
+    #[arg(long, conflicts_with = "shell")]
+    pub prefix: Option<PathBuf>,
+
+    /// Install all completions into the prefix
+    #[arg(long, conflicts_with = "shell", conflicts_with = "prefix")]
+    pub home: bool,
+}
+
+pub fn run(cmd: &Command) -> anyhow::Result<()> {
+    if let Some(shell) = cmd.shell {
+        shell.generate(&mut stdout());
+    } else if let Some(prefix) = &cmd.prefix {
+        write_completion(
+            &prefix.join("share/bash-completion/completions/edgedb"),
+            Shell::Bash,
+        )?;
+        write_completion(
+            &prefix.join("share/fish/completions/edgedb.fish"),
+            Shell::Fish,
+        )?;
+        write_completion(&prefix.join("share/zsh/site-functions/_edgedb"), Shell::Zsh)?;
+    } else if cmd.home {
+        write_completions_home()?;
+    } else {
+        anyhow::bail!("either `--prefix` or `--shell=` is expected");
+    }
+    Ok(())
+}
+
+#[context("writing completion file {:?}", path)]
+fn write_completion(path: &Path, shell: Shell) -> anyhow::Result<()> {
+    if let Some(dir) = path.parent() {
+        fs::create_dir_all(dir)?;
+    }
+    shell.generate(&mut BufWriter::new(fs::File::create(path)?));
+    Ok(())
+}
+
+pub fn write_completions_home() -> anyhow::Result<()> {
+    let home = home_dir()?;
+    write_completion(
+        &home.join(".local/share/bash-completion/completions/edgedb"),
+        Shell::Bash,
+    )?;
+    write_completion(
+        &home.join(".config/fish/completions/edgedb.fish"),
+        Shell::Fish,
+    )?;
+    write_completion(&home.join(".zfunc/_edgedb"), Shell::Zsh)?;
+    Ok(())
+}
+#[derive(Debug, Clone, Copy, clap::ValueEnum)]
+#[value(rename_all = "kebab-case")]
+#[allow(clippy::enum_variant_names)]
+pub enum Shell {
+    Bash,
+    Elvish,
+    Fish,
+    PowerShell,
+    Zsh,
+}
+
+impl FromStr for Shell {
+    type Err = anyhow::Error;
+    fn from_str(v: &str) -> anyhow::Result<Shell> {
+        use Shell::*;
+        match v {
+            "bash" => Ok(Bash),
+            "elvish" => Ok(Elvish),
+            "fish" => Ok(Fish),
+            "powershell" => Ok(PowerShell),
+            "zsh" => Ok(Zsh),
+            _ => anyhow::bail!("unknown shell {:?}", v),
+        }
+    }
+}
+
+impl Shell {
+    fn generate(&self, buf: &mut dyn Write) {
+        use clap_complete::shells;
+
+        let mut app = Options::command();
+
+        let exe_path = current_exe().unwrap();
+        let bin_name = exe_path
+            .file_name()
+            .unwrap()
+            .to_str()
+            .unwrap_or(BRANDING_CLI_CMD);
+        match self {
+            Shell::Bash => clap_complete::generate(shells::Bash, &mut app, bin_name, buf),
+            Shell::Elvish => clap_complete::generate(shells::Elvish, &mut app, bin_name, buf),
+            Shell::Fish => clap_complete::generate(shells::Fish, &mut app, bin_name, buf),
+            Shell::Zsh => clap_complete::generate(shells::Zsh, &mut app, bin_name, buf),
+            Shell::PowerShell => {
+                clap_complete::generate(shells::PowerShell, &mut app, bin_name, buf)
+            }
+        }
+    }
+}
diff --git a/src/cli/install.rs b/src/cli/install.rs
index e19444c..ed843ae 100644
--- a/src/cli/install.rs
+++ b/src/cli/install.rs
@@ -4,15 +4,12 @@
 use std::env;
 use std::fs;
 use std::fs::OpenOptions;
-use std::io::{stdout, BufWriter, Seek, SeekFrom, Write};
+use std::io::{Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 use std::process::exit;
-use std::str::FromStr;
 
 use anyhow::Context;
-use clap_complete::{generate, shells};
 use const_format::concatcp;
-use fn_error_context::context;
 use gel_tokio::get_stash_path;
 use prettytable::{Cell, Row, Table};
 
@@ -22,7 +19,6 @@ use crate::branding::{BRANDING, BRANDING_CLI, BRANDING_CLI_CMD};
 use crate::cli::logo::print_logo;
 use crate::cli::{migrate, upgrade};
 use crate::commands::ExitCode;
-use crate::options::Options;
 use crate::platform::{binary_path, config_dir, current_exe, home_dir};
 use crate::portable::platform;
 use crate::portable::project;
@@ -32,8 +28,10 @@ use crate::process;
 use crate::question::{self, read_choice};
 use crate::table;
 
+use super::gen_completions;
+
 #[derive(clap::Parser, Clone, Debug)]
-pub struct CliInstall {
+pub struct Command {
     #[arg(long, hide = true)]
     pub nightly: bool,
     #[arg(long, hide = true)]
@@ -63,32 +61,6 @@ pub struct CliInstall {
     pub upgrade: bool,
 }
 
-#[derive(Debug, Clone, Copy, clap::ValueEnum)]
-#[value(rename_all = "kebab-case")]
-#[allow(clippy::enum_variant_names)]
-pub enum Shell {
-    Bash,
-    Elvish,
-    Fish,
-    PowerShell,
-    Zsh,
-}
-
-#[derive(clap::Args, Clone, Debug)]
-pub struct GenCompletions {
-    /// Shell to print out completions for
-    #[arg(long)]
-    pub shell: Option<Shell>,
-
-    /// Install all completions into the prefix
-    #[arg(long, conflicts_with = "shell")]
-    pub prefix: Option<PathBuf>,
-
-    /// Install all completions into the prefix
-    #[arg(long, conflicts_with = "shell", conflicts_with = "prefix")]
-    pub home: bool,
-}
-
 pub struct Settings {
     system: bool,
     installation_path: PathBuf,
@@ -97,6 +69,166 @@ pub struct Settings {
     rc_files: Vec<PathBuf>,
 }
 
+pub fn run(cmd: &Command) -> anyhow::Result<()> {
+    match _run(cmd) {
+        Ok(()) => {
+            if cfg!(windows) && !cmd.upgrade && !cmd.no_confirm && !cmd.no_wait_for_exit_prompt {
+                // This is needed so user can read the message if console
+                // was open just for this process
+                eprintln!("Press the Enter key to continue");
+                read_choice()?;
+            }
+            Ok(())
+        }
+        Err(e) => {
+            if cfg!(windows) && !cmd.upgrade && !cmd.no_confirm && !cmd.no_wait_for_exit_prompt {
+                // This is needed so user can read the message if console
+                // was open just for this process
+                eprintln!("edgedb error: {e:#}");
+                eprintln!("Press the Enter key to continue");
+                read_choice()?;
+                exit(1);
+            }
+            Err(e)
+        }
+    }
+}
+
+fn _run(cmd: &Command) -> anyhow::Result<()> {
+    #[cfg(unix)]
+    if !cmd.no_confirm {
+        match home_dir_from_passwd().zip(env::var_os("HOME")) {
+            Some((passwd, env)) if passwd != env => {
+                msg!(
+                    "$HOME differs from euid-obtained home directory: \
+                       you may be using sudo"
+                );
+                msg!("$HOME directory: {}", Path::new(&env).display());
+                msg!("euid-obtained home directory: {}", passwd.display());
+                msg!(
+                    "if this is what you want, \
+                       restart the installation with `-y'"
+                );
+                return Err(ExitCode::new(1).into());
+            }
+            _ => {}
+        }
+    }
+    let installation_path = binary_path()?.parent().unwrap().to_owned();
+    let mut settings = Settings {
+        rc_files: get_rc_files()?,
+        system: false,
+        modify_path: !cmd.no_modify_path && no_dir_in_path(&installation_path),
+        installation_path,
+        env_file: config_dir()?.join("env"),
+    };
+    if !cmd.quiet && !cmd.upgrade {
+        print_long_description(&settings);
+        settings.print();
+        if !cmd.no_confirm {
+            loop {
+                println!("1) Proceed with installation (default)");
+                println!("2) Customize installation");
+                println!("3) Cancel installation");
+                match read_choice()?.as_ref() {
+                    "" | "1" => break,
+                    "2" => {
+                        customize(&mut settings)?;
+                        settings.print();
+                    }
+                    _ => {
+                        print::error!("Aborting installation.");
+                        exit(7);
+                    }
+                }
+            }
+        }
+    }
+
+    if cfg!(all(target_os = "macos", target_arch = "x86_64")) && platform::is_arm64_hardware() {
+        msg!("{BRANDING} now supports native M1 build. Downloading binary...");
+        return upgrade::upgrade_to_arm64();
+    }
+
+    copy_to_installation_path(&settings.installation_path)?;
+    copy_to_alternative_executable(&settings.installation_path)?;
+
+    gen_completions::write_completions_home()?;
+
+    if settings.modify_path {
+        #[cfg(windows)]
+        {
+            use std::env::join_paths;
+
+            windows_augment_path(|orig_path| {
+                if orig_path.iter().any(|p| p == &settings.installation_path) {
+                    return None;
+                }
+                Some(
+                    join_paths(
+                        vec![&settings.installation_path]
+                            .into_iter()
+                            .chain(orig_path.iter()),
+                    )
+                    .expect("paths can be joined"),
+                )
+            })?;
+        }
+        if cfg!(unix) {
+            let line = format!(
+                "\nexport PATH=\"{}:$PATH\"",
+                settings.installation_path.display()
+            );
+            for path in &settings.rc_files {
+                ensure_line(path, &line)
+                    .with_context(|| format!("failed to update profile file {path:?}"))?;
+            }
+            if let Some(dir) = settings.env_file.parent() {
+                fs::create_dir_all(dir).with_context(|| format!("failed to create {dir:?}"))?;
+            }
+            fs::write(&settings.env_file, line + "\n")
+                .with_context(|| format!("failed to write env file {:?}", settings.env_file))?;
+        }
+    }
+
+    let base = home_dir()?.join(".edgedb");
+    let new_layout = if base.exists() {
+        eprintln!(
+            "\
+                {BRANDING_CLI} no longer uses '{}' to store data \
+                and now uses standard locations of your OS. \
+        ",
+            base.display()
+        );
+        let q = question::Confirm::new(format!(
+            "\
+            Do you want to run `{BRANDING_CLI_CMD} cli migrate` now to update \
+            the directory layout?\
+        "
+        ));
+        if q.ask()? {
+            migrate::migrate(&base, false)?;
+            true
+        } else {
+            false
+        }
+    } else {
+        true
+    };
+
+    if !cmd.upgrade {
+        let init_result = if cmd.no_confirm {
+            Ok(InitResult::NonInteractive)
+        } else {
+            try_project_init(new_layout)
+        };
+
+        print_post_install_message(&settings, init_result);
+    }
+
+    Ok(())
+}
+
 fn print_long_description(settings: &Settings) {
     println!();
 
@@ -371,39 +503,6 @@ fn print_post_install_message(settings: &Settings, init_result: anyhow::Result<I
     }
 }
 
-pub fn main(options: &CliInstall) -> anyhow::Result<()> {
-    match _main(options) {
-        Ok(()) => {
-            if cfg!(windows)
-                && !options.upgrade
-                && !options.no_confirm
-                && !options.no_wait_for_exit_prompt
-            {
-                // This is needed so user can read the message if console
-                // was open just for this process
-                eprintln!("Press the Enter key to continue");
-                read_choice()?;
-            }
-            Ok(())
-        }
-        Err(e) => {
-            if cfg!(windows)
-                && !options.upgrade
-                && !options.no_confirm
-                && !options.no_wait_for_exit_prompt
-            {
-                // This is needed so user can read the message if console
-                // was open just for this process
-                eprintln!("edgedb error: {e:#}");
-                eprintln!("Press the Enter key to continue");
-                read_choice()?;
-                exit(1);
-            }
-            Err(e)
-        }
-    }
-}
-
 fn customize(settings: &mut Settings) -> anyhow::Result<()> {
     if no_dir_in_path(&settings.installation_path) {
         let q = question::Confirm::new("Modify PATH variable?");
@@ -479,141 +578,6 @@ fn try_project_init(new_layout: bool) -> anyhow::Result<InitResult> {
     }
 }
 
-fn _main(options: &CliInstall) -> anyhow::Result<()> {
-    #[cfg(unix)]
-    if !options.no_confirm {
-        match home_dir_from_passwd().zip(env::var_os("HOME")) {
-            Some((passwd, env)) if passwd != env => {
-                msg!(
-                    "$HOME differs from euid-obtained home directory: \
-                       you may be using sudo"
-                );
-                msg!("$HOME directory: {}", Path::new(&env).display());
-                msg!("euid-obtained home directory: {}", passwd.display());
-                msg!(
-                    "if this is what you want, \
-                       restart the installation with `-y'"
-                );
-                return Err(ExitCode::new(1).into());
-            }
-            _ => {}
-        }
-    }
-    let installation_path = binary_path()?.parent().unwrap().to_owned();
-    let mut settings = Settings {
-        rc_files: get_rc_files()?,
-        system: false,
-        modify_path: !options.no_modify_path && no_dir_in_path(&installation_path),
-        installation_path,
-        env_file: config_dir()?.join("env"),
-    };
-    if !options.quiet && !options.upgrade {
-        print_long_description(&settings);
-        settings.print();
-        if !options.no_confirm {
-            loop {
-                println!("1) Proceed with installation (default)");
-                println!("2) Customize installation");
-                println!("3) Cancel installation");
-                match read_choice()?.as_ref() {
-                    "" | "1" => break,
-                    "2" => {
-                        customize(&mut settings)?;
-                        settings.print();
-                    }
-                    _ => {
-                        print::error!("Aborting installation.");
-                        exit(7);
-                    }
-                }
-            }
-        }
-    }
-
-    if cfg!(all(target_os = "macos", target_arch = "x86_64")) && platform::is_arm64_hardware() {
-        msg!("{BRANDING} now supports native M1 build. Downloading binary...");
-        return upgrade::upgrade_to_arm64();
-    }
-
-    copy_to_installation_path(&settings.installation_path)?;
-    copy_to_alternative_executable(&settings.installation_path)?;
-
-    write_completions_home()?;
-
-    if settings.modify_path {
-        #[cfg(windows)]
-        {
-            use std::env::join_paths;
-
-            windows_augment_path(|orig_path| {
-                if orig_path.iter().any(|p| p == &settings.installation_path) {
-                    return None;
-                }
-                Some(
-                    join_paths(
-                        vec![&settings.installation_path]
-                            .into_iter()
-                            .chain(orig_path.iter()),
-                    )
-                    .expect("paths can be joined"),
-                )
-            })?;
-        }
-        if cfg!(unix) {
-            let line = format!(
-                "\nexport PATH=\"{}:$PATH\"",
-                settings.installation_path.display()
-            );
-            for path in &settings.rc_files {
-                ensure_line(path, &line)
-                    .with_context(|| format!("failed to update profile file {path:?}"))?;
-            }
-            if let Some(dir) = settings.env_file.parent() {
-                fs::create_dir_all(dir).with_context(|| format!("failed to create {dir:?}"))?;
-            }
-            fs::write(&settings.env_file, line + "\n")
-                .with_context(|| format!("failed to write env file {:?}", settings.env_file))?;
-        }
-    }
-
-    let base = home_dir()?.join(".edgedb");
-    let new_layout = if base.exists() {
-        eprintln!(
-            "\
-                {BRANDING_CLI} no longer uses '{}' to store data \
-                and now uses standard locations of your OS. \
-        ",
-            base.display()
-        );
-        let q = question::Confirm::new(format!(
-            "\
-            Do you want to run `{BRANDING_CLI_CMD} cli migrate` now to update \
-            the directory layout?\
-        "
-        ));
-        if q.ask()? {
-            migrate::migrate(&base, false)?;
-            true
-        } else {
-            false
-        }
-    } else {
-        true
-    };
-
-    if !options.upgrade {
-        let init_result = if options.no_confirm {
-            Ok(InitResult::NonInteractive)
-        } else {
-            try_project_init(new_layout)
-        };
-
-        print_post_install_message(&settings, init_result);
-    }
-
-    Ok(())
-}
-
 fn copy_to_installation_path<P: AsRef<Path>>(installation_path: P) -> anyhow::Result<()> {
     let installation_path = installation_path.as_ref();
     let tmp_path = installation_path.join(concatcp!(BRANDING_CLI_CMD, ".tmp"));
@@ -853,50 +817,6 @@ pub fn windows_augment_path<F: FnOnce(&[PathBuf]) -> Option<std::ffi::OsString>>
     Ok(())
 }
 
-#[context("writing completion file {:?}", path)]
-fn write_completion(path: &Path, shell: Shell) -> anyhow::Result<()> {
-    if let Some(dir) = path.parent() {
-        fs::create_dir_all(dir)?;
-    }
-    shell.generate(&mut BufWriter::new(fs::File::create(path)?));
-    Ok(())
-}
-
-pub fn write_completions_home() -> anyhow::Result<()> {
-    let home = home_dir()?;
-    write_completion(
-        &home.join(".local/share/bash-completion/completions/edgedb"),
-        Shell::Bash,
-    )?;
-    write_completion(
-        &home.join(".config/fish/completions/edgedb.fish"),
-        Shell::Fish,
-    )?;
-    write_completion(&home.join(".zfunc/_edgedb"), Shell::Zsh)?;
-    Ok(())
-}
-
-pub fn gen_completions(options: &GenCompletions) -> anyhow::Result<()> {
-    if let Some(shell) = options.shell {
-        shell.generate(&mut stdout());
-    } else if let Some(prefix) = &options.prefix {
-        write_completion(
-            &prefix.join("share/bash-completion/completions/edgedb"),
-            Shell::Bash,
-        )?;
-        write_completion(
-            &prefix.join("share/fish/completions/edgedb.fish"),
-            Shell::Fish,
-        )?;
-        write_completion(&prefix.join("share/zsh/site-functions/_edgedb"), Shell::Zsh)?;
-    } else if options.home {
-        write_completions_home()?;
-    } else {
-        anyhow::bail!("either `--prefix` or `--shell=` is expected");
-    }
-    Ok(())
-}
-
 impl Settings {
     pub fn print(&self) {
         let mut table = Table::new();
@@ -926,37 +846,6 @@ impl Settings {
     }
 }
 
-impl FromStr for Shell {
-    type Err = anyhow::Error;
-    fn from_str(v: &str) -> anyhow::Result<Shell> {
-        use Shell::*;
-        match v {
-            "bash" => Ok(Bash),
-            "elvish" => Ok(Elvish),
-            "fish" => Ok(Fish),
-            "powershell" => Ok(PowerShell),
-            "zsh" => Ok(Zsh),
-            _ => anyhow::bail!("unknown shell {:?}", v),
-        }
-    }
-}
-
-impl Shell {
-    fn generate(&self, buf: &mut dyn Write) {
-        use Shell::*;
-
-        let mut app = Options::command();
-        let n = "edgedb";
-        match self {
-            Bash => generate(shells::Bash, &mut app, n, buf),
-            Elvish => generate(shells::Elvish, &mut app, n, buf),
-            Fish => generate(shells::Fish, &mut app, n, buf),
-            PowerShell => generate(shells::PowerShell, &mut app, n, buf),
-            Zsh => generate(shells::Zsh, &mut app, n, buf),
-        }
-    }
-}
-
 // search user database to get home dir of euid user
 #[cfg(unix)]
 pub(crate) fn home_dir_from_passwd() -> Option<PathBuf> {
diff --git a/src/cli/main.rs b/src/cli/main.rs
index 1df5c9e..e69de29 100644
--- a/src/cli/main.rs
+++ b/src/cli/main.rs
@@ -1,15 +0,0 @@
-use crate::cli::install;
-use crate::cli::migrate;
-use crate::cli::options::CliCommand;
-use crate::cli::options::Command;
-use crate::cli::upgrade;
-
-pub fn main(cmd: &CliCommand) -> anyhow::Result<()> {
-    use Command::*;
-
-    match &cmd.subcommand {
-        Upgrade(s) => upgrade::main(s),
-        Install(s) => install::main(s),
-        Migrate(s) => migrate::main(s),
-    }
-}
diff --git a/src/cli/migrate.rs b/src/cli/migrate.rs
index 7a813db..8bdee4c 100644
--- a/src/cli/migrate.rs
+++ b/src/cli/migrate.rs
@@ -18,7 +18,7 @@ use crate::print_markdown;
 use crate::question;
 
 #[derive(clap::Args, Clone, Debug)]
-pub struct CliMigrate {
+pub struct Command {
     /// Dry run: do not actually move anything
     #[arg(short = 'n', long)]
     pub dry_run: bool,
@@ -35,10 +35,10 @@ enum ConfirmOverwrite {
     Quit,
 }
 
-pub fn main(options: &CliMigrate) -> anyhow::Result<()> {
+pub fn run(cmd: &Command) -> anyhow::Result<()> {
     let base = home_dir()?.join(".edgedb");
     if base.exists() {
-        migrate(&base, options.dry_run)
+        migrate(&base, cmd.dry_run)
     } else {
         log::warn!(
             "Directory {:?} does not exist. No actions will be taken.",
diff --git a/src/cli/mod.rs b/src/cli/mod.rs
index 73563ee..3cf837c 100644
--- a/src/cli/mod.rs
+++ b/src/cli/mod.rs
@@ -1,13 +1,40 @@
 pub mod directory_check;
 pub mod env;
+pub mod gen_completions;
 pub mod install;
 pub mod logo;
-pub mod main;
 pub mod migrate;
-pub mod options;
 pub mod upgrade;
 
 #[macro_use]
 mod markdown;
 
-pub use main::main;
+#[derive(clap::Args, Clone, Debug)]
+#[command(version = "help_expand")]
+#[command(disable_version_flag = true)]
+pub struct Command {
+    #[command(subcommand)]
+    pub subcommand: Subcommand,
+}
+
+#[derive(clap::Subcommand, Clone, Debug)]
+pub enum Subcommand {
+    /// Upgrade the [`BRANDING_CLI_CMD`] command-line tool
+    Upgrade(upgrade::Command),
+    /// Install the [`BRANDING_CLI_CMD`] command-line tool
+    #[command(hide = true)]
+    Install(install::Command),
+    /// Migrate files from `~/.edgedb` to the new directory layout
+    #[command(hide = true)]
+    Migrate(migrate::Command),
+}
+
+pub fn run(cmd: &Command) -> anyhow::Result<()> {
+    use Subcommand::*;
+
+    match &cmd.subcommand {
+        Upgrade(s) => upgrade::run(s),
+        Install(s) => install::run(s),
+        Migrate(s) => migrate::run(s),
+    }
+}
diff --git a/src/cli/options.rs b/src/cli/options.rs
deleted file mode 100644
index f80f204..0000000
--- a/src/cli/options.rs
+++ /dev/null
@@ -1,25 +0,0 @@
-#[cfg(doc)]
-use crate::branding::BRANDING_CLI_CMD;
-use crate::cli::install;
-use crate::cli::migrate;
-use crate::cli::upgrade;
-
-#[derive(clap::Args, Clone, Debug)]
-#[command(version = "help_expand")]
-#[command(disable_version_flag = true)]
-pub struct CliCommand {
-    #[command(subcommand)]
-    pub subcommand: Command,
-}
-
-#[derive(clap::Subcommand, Clone, Debug)]
-pub enum Command {
-    /// Upgrade the [`BRANDING_CLI_CMD`] command-line tool
-    Upgrade(upgrade::CliUpgrade),
-    /// Install the [`BRANDING_CLI_CMD`] command-line tool
-    #[command(hide = true)]
-    Install(install::CliInstall),
-    /// Migrate files from `~/.edgedb` to the new directory layout
-    #[command(hide = true)]
-    Migrate(migrate::CliMigrate),
-}
diff --git a/src/cli/upgrade.rs b/src/cli/upgrade.rs
index c375f68..d428859 100644
--- a/src/cli/upgrade.rs
+++ b/src/cli/upgrade.rs
@@ -17,7 +17,7 @@ use crate::process;
 const INDEX_TIMEOUT: Duration = Duration::new(60, 0);
 
 #[derive(clap::Args, Clone, Debug)]
-pub struct CliUpgrade {
+pub struct Command {
     /// Enable verbose output
     #[arg(short = 'v', long)]
     pub verbose: bool,
@@ -45,6 +45,91 @@ pub struct CliUpgrade {
     pub to_channel: Option<Channel>,
 }
 
+pub fn run(cmd: &Command) -> anyhow::Result<()> {
+    let path = binary_path()?;
+    if !_can_upgrade(&path)? {
+        anyhow::bail!("Only binary installed at {:?} can be upgraded", path);
+    }
+    upgrade(cmd, path)
+}
+
+fn upgrade(cmd: &Command, path: PathBuf) -> anyhow::Result<()> {
+    let cur_channel = channel();
+    let channel = if let Some(channel) = cmd.to_channel {
+        channel
+    } else if cmd.to_stable {
+        Channel::Stable
+    } else if cmd.to_nightly {
+        Channel::Nightly
+    } else if cmd.to_testing {
+        Channel::Testing
+    } else {
+        cur_channel
+    };
+
+    #[allow(unused_mut)]
+    let mut target_plat = platform::get_cli()?;
+    // Always force upgrade when switching channel
+    #[allow(unused_mut)]
+    let mut force = cmd.force || cur_channel != channel;
+
+    if cfg!(all(target_os = "macos", target_arch = "x86_64")) && platform::is_arm64_hardware() {
+        target_plat = "aarch64-apple-darwin";
+        // Always force upgrade when need to switch platform
+        force = true;
+    }
+
+    let pkg = repository::get_platform_cli_packages(channel, target_plat, INDEX_TIMEOUT)?
+        .into_iter()
+        .max_by(|a, b| a.version.cmp(&b.version))
+        .context("cannot find new version")?;
+    if !force && pkg.version <= self_version()? {
+        log::info!("Version is identical; no update needed.");
+        if !cmd.quiet {
+            print::success!("Already up to date.");
+        }
+        return Ok(());
+    }
+
+    let down_dir = path
+        .parent()
+        .context("download path missing directory component")?;
+    fs::create_dir_all(down_dir).with_context(|| format!("failed to create {down_dir:?}"))?;
+
+    let down_path = path.with_extension("download");
+    let tmp_path = tmp_file_path(&path);
+
+    download(&down_path, &pkg.url, cmd.quiet)?;
+    unpack_file(&down_path, &tmp_path, pkg.compression)?;
+
+    let backup_path = path.with_extension("backup");
+    if cfg!(unix) {
+        fs::remove_file(&backup_path).ok();
+        fs::hard_link(&path, &backup_path)
+            .map_err(|e| log::warn!("Cannot keep a backup file: {:#}", e))
+            .ok();
+    } else if cfg!(windows) {
+        fs::remove_file(&backup_path).ok();
+        fs::rename(&path, &backup_path)?;
+    } else {
+        anyhow::bail!("unknown OS");
+    }
+    process::Native::new("upgrade", "cli", &tmp_path)
+        .arg("cli")
+        .arg("install")
+        .arg("--upgrade")
+        .no_proxy()
+        .run()?;
+    fs::remove_file(&tmp_path).ok();
+    if !cmd.quiet {
+        msg!(
+            "Upgraded to version {}",
+            pkg.version.to_string().emphasized()
+        );
+    }
+    Ok(())
+}
+
 pub fn can_upgrade() -> bool {
     binary_path()
         .and_then(|p| _can_upgrade(&p))
@@ -133,17 +218,9 @@ pub fn self_version() -> anyhow::Result<ver::Semver> {
         .context("cannot parse cli version")
 }
 
-pub fn main(options: &CliUpgrade) -> anyhow::Result<()> {
-    let path = binary_path()?;
-    if !_can_upgrade(&path)? {
-        anyhow::bail!("Only binary installed at {:?} can be upgraded", path);
-    }
-    _main(options, path)
-}
-
 pub fn upgrade_to_arm64() -> anyhow::Result<()> {
-    _main(
-        &CliUpgrade {
+    upgrade(
+        &Command {
             verbose: false,
             quiet: false,
             force: true,
@@ -155,80 +232,3 @@ pub fn upgrade_to_arm64() -> anyhow::Result<()> {
         binary_path()?,
     )
 }
-
-fn _main(options: &CliUpgrade, path: PathBuf) -> anyhow::Result<()> {
-    let cur_channel = channel();
-    let channel = if let Some(channel) = options.to_channel {
-        channel
-    } else if options.to_stable {
-        Channel::Stable
-    } else if options.to_nightly {
-        Channel::Nightly
-    } else if options.to_testing {
-        Channel::Testing
-    } else {
-        cur_channel
-    };
-
-    #[allow(unused_mut)]
-    let mut target_plat = platform::get_cli()?;
-    // Always force upgrade when switching channel
-    #[allow(unused_mut)]
-    let mut force = options.force || cur_channel != channel;
-
-    if cfg!(all(target_os = "macos", target_arch = "x86_64")) && platform::is_arm64_hardware() {
-        target_plat = "aarch64-apple-darwin";
-        // Always force upgrade when need to switch platform
-        force = true;
-    }
-
-    let pkg = repository::get_platform_cli_packages(channel, target_plat, INDEX_TIMEOUT)?
-        .into_iter()
-        .max_by(|a, b| a.version.cmp(&b.version))
-        .context("cannot find new version")?;
-    if !force && pkg.version <= self_version()? {
-        log::info!("Version is identical; no update needed.");
-        if !options.quiet {
-            print::success!("Already up to date.");
-        }
-        return Ok(());
-    }
-
-    let down_dir = path
-        .parent()
-        .context("download path missing directory component")?;
-    fs::create_dir_all(down_dir).with_context(|| format!("failed to create {down_dir:?}"))?;
-
-    let down_path = path.with_extension("download");
-    let tmp_path = tmp_file_path(&path);
-
-    download(&down_path, &pkg.url, options.quiet)?;
-    unpack_file(&down_path, &tmp_path, pkg.compression)?;
-
-    let backup_path = path.with_extension("backup");
-    if cfg!(unix) {
-        fs::remove_file(&backup_path).ok();
-        fs::hard_link(&path, &backup_path)
-            .map_err(|e| log::warn!("Cannot keep a backup file: {:#}", e))
-            .ok();
-    } else if cfg!(windows) {
-        fs::remove_file(&backup_path).ok();
-        fs::rename(&path, &backup_path)?;
-    } else {
-        anyhow::bail!("unknown OS");
-    }
-    process::Native::new("upgrade", "cli", &tmp_path)
-        .arg("cli")
-        .arg("install")
-        .arg("--upgrade")
-        .no_proxy()
-        .run()?;
-    fs::remove_file(&tmp_path).ok();
-    if !options.quiet {
-        msg!(
-            "Upgraded to version {}",
-            pkg.version.to_string().emphasized()
-        );
-    }
-    Ok(())
-}
diff --git a/src/commands/backslash.rs b/src/commands/backslash.rs
index 67bc862..5bee26f 100644
--- a/src/commands/backslash.rs
+++ b/src/commands/backslash.rs
@@ -67,6 +67,7 @@ Editing
                             Defaults to vi (Notepad in Windows).
 
 Connection
+  \b, \branch ...           Branch commands
   \c, \connect [DBNAME]     Connect to database/branch DBNAME
 
 Settings
@@ -342,8 +343,8 @@ impl CommandCache {
         aliases.insert("quit", &["exit"]);
         aliases.insert("?", &["help"]);
         aliases.insert("h", &["help"]);
-        aliases.insert("branch", &["branching"]);
-        aliases.insert("b", &["branching"]);
+        aliases.insert("branch", &["branch"]);
+        aliases.insert("b", &["branch"]);
         let mut setting_cmd = None;
         let commands: BTreeMap<_, _> = clap
             .get_subcommands_mut()
@@ -577,11 +578,13 @@ pub async fn execute(
     use ExecuteResult::*;
     use Setting::*;
 
-    let options = Options {
+    let mut options = Options {
         command_line: false,
         styler: Some(Styler::new()),
         conn_params: prompt.conn_params.clone(),
+        instance_name: None,
     };
+    options.infer_instance_name()?;
     match cmd {
         Help => {
             print!("{HELP}");
@@ -589,8 +592,8 @@ pub async fn execute(
         }
         Common(ref cmd) => {
             prompt.soft_reconnect().await?;
-            let cli = prompt.connection.as_mut().expect("connection established");
-            let result = execute::common(cli, cmd, &options).await?;
+            let conn = prompt.connection.as_mut().expect("connection established");
+            let result = execute::common(Some(conn), cmd, &options).await?;
 
             if let Some(branch) = result.new_branch {
                 prompt.try_connect(&branch).await?;
diff --git a/src/commands/cli.rs b/src/commands/cli.rs
index 3dafebb..248457b 100644
--- a/src/commands/cli.rs
+++ b/src/commands/cli.rs
@@ -1,5 +1,6 @@
 use is_terminal::IsTerminal;
 
+use crate::cli;
 use crate::cli::directory_check;
 use crate::cloud::main::cloud_main;
 use crate::commands;
@@ -11,7 +12,6 @@ use crate::options::{Command, Options};
 use crate::portable;
 use crate::print::style::Styler;
 use crate::watch;
-use crate::{branch, cli};
 
 #[tokio::main(flavor = "current_thread")]
 async fn common_cmd(
@@ -19,8 +19,7 @@ async fn common_cmd(
     cmdopt: commands::Options,
     cmd: &Common,
 ) -> Result<(), anyhow::Error> {
-    let mut conn = cmdopt.conn_params.connect().await?;
-    commands::execute::common(&mut conn, cmd, &cmdopt).await?;
+    commands::execute::common(None, cmd, &cmdopt).await?;
     Ok(())
 }
 
@@ -67,18 +66,13 @@ pub fn main(options: &Options) -> Result<(), anyhow::Error> {
             directory_check::check_and_warn();
             non_interactive::noninteractive_main(q, options)
         }
-        Command::_SelfInstall(s) => cli::install::main(s),
-        Command::_GenCompletions(s) => cli::install::gen_completions(s),
-        Command::Cli(c) => cli::main(c),
+        Command::_SelfInstall(s) => cli::install::run(s),
+        Command::_GenCompletions(s) => cli::gen_completions::run(s),
+        Command::Cli(c) => cli::run(c),
         Command::Info(info) => commands::info(options, info),
         Command::UI(c) => commands::show_ui(c, options),
         Command::Cloud(c) => cloud_main(c, &options.cloud_options),
         Command::Watch(c) => watch::run(options, c),
-        Command::Branch(c) => {
-            let opts = init_command_opts(options)?;
-            branch::run(&opts, c)?;
-            Ok(())
-        }
         Command::HashPassword(cmd) => {
             println!("{}", portable::password_hash(&cmd.password));
             Ok(())
@@ -94,6 +88,7 @@ fn init_command_opts(options: &Options) -> Result<commands::Options, anyhow::Err
         } else {
             None
         },
+        instance_name: options.conn_options.instance.clone(),
         conn_params: options.block_on_create_connector()?,
     })
 }
diff --git a/src/commands/execute.rs b/src/commands/execute.rs
index 4e5d2b1..441ade4 100644
--- a/src/commands/execute.rs
+++ b/src/commands/execute.rs
@@ -12,16 +12,35 @@ use crate::migrations::options::MigrationCmd;
 use crate::print;
 
 pub async fn common(
-    cli: &mut Connection,
+    conn: Option<&mut Connection>,
     cmd: &Common,
     options: &Options,
 ) -> Result<branch::CommandResult, anyhow::Error> {
     use Common::*;
+
+    // match commands that don't need connection
+    match cmd {
+        Branch(cmd) => {
+            return branch::run(&cmd.subcommand, options, conn).await;
+        }
+        _ => {}
+    }
+
+    // connect
+    let mut conn_cell;
+    let conn = if let Some(conn) = conn {
+        conn
+    } else {
+        conn_cell = options.conn_params.connect().await?;
+        &mut conn_cell
+    };
+
+    // match other
     match cmd {
         List(c) => match &c.subcommand {
             ListCmd::Aliases(c) => {
                 commands::list_aliases(
-                    cli,
+                    conn,
                     options,
                     &c.pattern,
                     c.system,
@@ -31,11 +50,11 @@ pub async fn common(
                 .await?;
             }
             ListCmd::Casts(c) => {
-                commands::list_casts(cli, options, &c.pattern, c.case_sensitive).await?;
+                commands::list_casts(conn, options, &c.pattern, c.case_sensitive).await?;
             }
             ListCmd::Indexes(c) => {
                 commands::list_indexes(
-                    cli,
+                    conn,
                     options,
                     &c.pattern,
                     c.system,
@@ -45,37 +64,37 @@ pub async fn common(
                 .await?;
             }
             ListCmd::Databases => {
-                commands::list_databases(cli, options).await?;
+                commands::list_databases(conn, options).await?;
             }
             ListCmd::Branches => {
-                commands::list_branches(cli, options).await?;
+                commands::list_branches(conn, options).await?;
             }
             ListCmd::Scalars(c) => {
-                commands::list_scalar_types(cli, options, &c.pattern, c.system, c.case_sensitive)
+                commands::list_scalar_types(conn, options, &c.pattern, c.system, c.case_sensitive)
                     .await?;
             }
             ListCmd::Types(c) => {
-                commands::list_object_types(cli, options, &c.pattern, c.system, c.case_sensitive)
+                commands::list_object_types(conn, options, &c.pattern, c.system, c.case_sensitive)
                     .await?;
             }
             ListCmd::Modules(c) => {
-                commands::list_modules(cli, options, &c.pattern, c.case_sensitive).await?;
+                commands::list_modules(conn, options, &c.pattern, c.case_sensitive).await?;
             }
             ListCmd::Roles(c) => {
-                commands::list_roles(cli, options, &c.pattern, c.case_sensitive).await?;
+                commands::list_roles(conn, options, &c.pattern, c.case_sensitive).await?;
             }
         },
         Analyze(c) => {
-            analyze::command(cli, c).await?;
+            analyze::command(conn, c).await?;
         }
-        Pgaddr => match cli.get_server_param::<PostgresAddress>() {
+        Pgaddr => match conn.get_server_param::<PostgresAddress>() {
             Some(addr) => {
                 // < 6.x
                 println!("{}", serde_json::to_string_pretty(addr)?);
             }
             None => {
                 // >= 6.x
-                match cli.get_server_param::<PostgresDsn>() {
+                match conn.get_server_param::<PostgresDsn>() {
                     Some(addr) => {
                         println!("{}", addr.0);
                     }
@@ -84,67 +103,64 @@ pub async fn common(
             }
         },
         Psql => {
-            commands::psql(cli, options).await?;
+            commands::psql(conn, options).await?;
         }
         Describe(c) => match &c.subcommand {
             DescribeCmd::Object(c) => {
-                commands::describe(cli, options, &c.name, c.verbose).await?;
+                commands::describe(conn, options, &c.name, c.verbose).await?;
             }
             DescribeCmd::Schema(_) => {
-                commands::describe_schema(cli, options).await?;
+                commands::describe_schema(conn, options).await?;
             }
         },
         Dump(c) => {
-            commands::dump(cli, options, c).await?;
+            commands::dump(conn, options, c).await?;
         }
         Restore(params) => {
-            commands::restore(cli, options, params).await?;
+            commands::restore(conn, options, params).await?;
         }
         Configure(c) => {
-            commands::configure(cli, options, c).await?;
+            commands::configure(conn, options, c).await?;
         }
         Database(c) => match &c.subcommand {
             DatabaseCmd::Create(c) => {
-                commands::database::create(cli, c, options).await?;
+                commands::database::create(conn, c, options).await?;
             }
             DatabaseCmd::Drop(d) => {
-                commands::database::drop(cli, d, options).await?;
+                commands::database::drop(conn, d, options).await?;
             }
             DatabaseCmd::Wipe(w) => {
-                commands::database::wipe(cli, w).await?;
+                commands::database::wipe(conn, w).await?;
             }
         },
-        Branching(cmd) => {
-            let cmd = branch::Subcommand::from(cmd.subcommand.clone());
-            return branch::do_run(&cmd, options, Some(cli), None).await;
-        }
+        Branch(_) => unreachable!(),
         Migrate(cmd) => {
-            migrations::apply::run(cmd, cli, options).await?;
+            migrations::apply::run(cmd, conn, options).await?;
         }
         Migration(m) => match &m.subcommand {
             MigrationCmd::Apply(cmd) => {
-                migrations::apply::run(cmd, cli, options).await?;
+                migrations::apply::run(cmd, conn, options).await?;
             }
             MigrationCmd::Create(cmd) => {
-                migrations::create::run(cmd, cli, options).await?;
+                migrations::create::run(cmd, conn, options).await?;
             }
             MigrationCmd::Status(params) => {
-                migrations::status(cli, options, params).await?;
+                migrations::status(conn, options, params).await?;
             }
             MigrationCmd::Log(params) => {
-                migrations::log(cli, options, params).await?;
+                migrations::log(conn, options, params).await?;
             }
             MigrationCmd::Edit(params) => {
-                migrations::edit(cli, options, params).await?;
+                migrations::edit(conn, options, params).await?;
             }
             MigrationCmd::UpgradeCheck(_) => {
                 anyhow::bail!("cannot be run in REPL mode");
             }
             MigrationCmd::Extract(params) => {
-                migrations::extract(cli, options, params).await?;
+                migrations::extract(conn, options, params).await?;
             }
             MigrationCmd::UpgradeFormat(params) => {
-                migrations::upgrade_format(cli, options, params).await?;
+                migrations::upgrade_format(conn, options, params).await?;
             }
         },
     }
diff --git a/src/commands/options.rs b/src/commands/options.rs
index 1123156..8663878 100644
--- a/src/commands/options.rs
+++ b/src/commands/options.rs
@@ -1,8 +1,24 @@
+use std::str::FromStr;
+
 use crate::connect::Connector;
+use crate::portable;
 use crate::print::style::Styler;
 
 pub struct Options {
     pub command_line: bool,
     pub styler: Option<Styler>,
     pub conn_params: Connector,
+    pub instance_name: Option<portable::options::InstanceName>,
+}
+
+impl Options {
+    pub fn infer_instance_name(&mut self) -> anyhow::Result<()> {
+        self.instance_name = self
+            .conn_params
+            .get()?
+            .instance_name()
+            .map(|x| portable::options::InstanceName::from_str(&x.to_string()))
+            .transpose()?;
+        Ok(())
+    }
 }
diff --git a/src/commands/parser.rs b/src/commands/parser.rs
index d89b57b..95bd6ad 100644
--- a/src/commands/parser.rs
+++ b/src/commands/parser.rs
@@ -1,11 +1,11 @@
 use std::path::PathBuf;
 
 use crate::branding::BRANDING_CLI_CMD;
-use crate::migrations;
 use crate::migrations::options::Migration;
 use crate::options::ConnectionOptions;
 use crate::portable::options::InstanceName;
 use crate::repl::{self, VectorLimit};
+use crate::{branch, migrations};
 
 use const_format::concatcp;
 
@@ -27,7 +27,8 @@ pub enum Common {
 
     /// Database commands
     Database(Database),
-    Branching(Branching),
+    /// Manage branches
+    Branch(branch::Command),
     /// Describe database schema or object
     Describe(Describe),
 
@@ -124,34 +125,6 @@ pub enum ListCmd {
     Types(ListTypes),
 }
 
-#[derive(clap::Args, Clone, Debug)]
-#[command(version = "help_expand", hide = true)]
-#[command(disable_version_flag = true)]
-pub struct Branching {
-    #[command(flatten)]
-    pub conn: ConnectionOptions,
-
-    #[command(subcommand)]
-    pub subcommand: BranchingCmd,
-}
-
-#[derive(clap::Subcommand, Clone, Debug)]
-pub enum BranchingCmd {
-    /// Create a new branch
-    Create(crate::branch::create::Command),
-    /// Delete a branch along with its data
-    Drop(crate::branch::drop::Command),
-    /// Delete a branch's data and reset its schema while preserving the branch
-    /// itself (its `cfg::DatabaseConfig`) and existing migration scripts
-    Wipe(crate::branch::wipe::Command),
-    /// List all branches.
-    List(crate::branch::list::Command),
-    /// Switches the current branch to a different one.
-    Switch(crate::branch::switch::Command),
-    /// Renames a branch.
-    Rename(crate::branch::rename::Command),
-}
-
 #[derive(clap::Args, Clone, Debug)]
 #[command(version = "help_expand")]
 #[command(disable_version_flag = true)]
diff --git a/src/hooks/mod.rs b/src/hooks/mod.rs
index bc051cb..5075f10 100644
--- a/src/hooks/mod.rs
+++ b/src/hooks/mod.rs
@@ -1,6 +1,4 @@
-use std::path;
-
-use crate::portable::{project, windows};
+use crate::portable::project;
 use crate::print::{self, Highlight};
 
 #[tokio::main(flavor = "current_thread")]
@@ -18,7 +16,7 @@ pub async fn on_action(action: &'static str, project: &project::Context) -> anyh
 
     print::msg!("{}", format!("hook {action}: {script}").muted());
 
-    let status = run_script(script, &project.location.root).await?;
+    let status = crate::watch::run_script(action, script, &project.location.root).await?;
 
     // abort on error
     if !status.success() {
@@ -29,23 +27,6 @@ pub async fn on_action(action: &'static str, project: &project::Context) -> anyh
     Ok(())
 }
 
-pub async fn run_script(
-    script: &str,
-    path: &path::Path,
-) -> Result<std::process::ExitStatus, anyhow::Error> {
-    let status = if !cfg!(windows) {
-        std::process::Command::new("/bin/sh")
-            .arg("-c")
-            .arg(script)
-            .current_dir(path)
-            .status()?
-    } else {
-        let wsl = windows::try_get_wsl()?;
-        wsl.sh(path).arg("-c").arg(script).run_for_status().await?
-    };
-    Ok(status)
-}
-
 fn get_hook<'m>(
     action: &'static str,
     manifest: &'m project::manifest::Manifest,
diff --git a/src/log_levels.rs b/src/log_levels.rs
index e523505..33e8000 100644
--- a/src/log_levels.rs
+++ b/src/log_levels.rs
@@ -1,4 +1,4 @@
-use crate::cli::options::Command as Cli;
+use crate::cli;
 use crate::commands::parser::Common;
 use crate::migrations::options::MigrationCmd;
 use crate::options::{Command, Options};
@@ -13,10 +13,10 @@ pub fn init(builder: &mut env_logger::Builder, opt: &Options) {
     }
     match &opt.subcommand {
         Some(Command::Cli(c)) => match &c.subcommand {
-            Cli::Upgrade(s) if s.verbose => {
+            cli::Subcommand::Upgrade(s) if s.verbose => {
                 builder.filter_module("edgedb::self_upgrade", log::LevelFilter::Info);
             }
-            Cli::Migrate(s) if s.verbose => {
+            cli::Subcommand::Migrate(s) if s.verbose => {
                 builder.filter_module("edgedb::self_migrate", log::LevelFilter::Info);
             }
             _ => {}
diff --git a/src/main.rs b/src/main.rs
index 46a5102..65af3c6 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -107,13 +107,11 @@ fn main() {
 }
 
 fn is_cli_upgrade(cmd: &Option<options::Command>) -> bool {
-    use cli::options::CliCommand;
-    use cli::options::Command::Upgrade;
     use options::Command::Cli;
     matches!(
         cmd,
-        Some(Cli(CliCommand {
-            subcommand: Upgrade(..)
+        Some(Cli(cli::Command {
+            subcommand: cli::Subcommand::Upgrade(..)
         }))
     )
 }
@@ -133,8 +131,8 @@ fn _main() -> anyhow::Result<()> {
     if let Some(arg0) = std::env::args_os().next() {
         if let Some(exe_name) = Path::new(&arg0).file_name() {
             if exe_name.to_string_lossy().contains("-init") {
-                let opt = cli::install::CliInstall::parse();
-                return cli::install::main(&opt);
+                let opt = cli::install::Command::parse();
+                return cli::install::run(&opt);
             }
         }
     }
diff --git a/src/options.rs b/src/options.rs
index a75473e..fbc40f8 100644
--- a/src/options.rs
+++ b/src/options.rs
@@ -14,10 +14,8 @@ use tokio::task::spawn_blocking as unblock;
 
 use edgedb_cli_derive::IntoArgs;
 
-use crate::cli::options::CliCommand;
 use crate::{cli, watch};
 
-use crate::branch;
 use crate::branding::{BRANDING, BRANDING_CLI_CMD, BRANDING_CLOUD, MANIFEST_FILE_DISPLAY_NAME};
 use crate::cloud::options::CloudCommand;
 use crate::commands::parser::Common;
@@ -364,26 +362,24 @@ pub enum Command {
     Instance(portable::instance::Command),
     /// Manage local [`BRANDING`] installations
     Server(portable::server::Command),
-    /// Manage local extensions
+    /// Manage extensions of local instances
     Extension(portable::extension::Command),
     /// Generate shell completions
     #[command(name = "_gen_completions")]
     #[command(hide = true)]
-    _GenCompletions(cli::install::GenCompletions),
+    _GenCompletions(cli::gen_completions::Command),
     /// Self-installation commands
     #[command(name = "cli")]
-    Cli(CliCommand),
+    Cli(cli::Command),
     /// Install [`BRANDING`]
     #[command(name = "_self_install")]
     #[command(hide = true)]
-    _SelfInstall(cli::install::CliInstall),
+    _SelfInstall(cli::install::Command),
     /// [`BRANDING_CLOUD`] authentication
     Cloud(CloudCommand),
     /// Start a long-running process that watches the project directory
     /// and runs scripts
     Watch(watch::Command),
-    /// Manage branches
-    Branch(branch::Command),
     /// Generate a `SCRAM-SHA-256` hash for a password.
     HashPassword(HashPasswordCommand),
 }
diff --git a/src/portable/extension.rs b/src/portable/extension.rs
index 18887fe..3ab6d52 100644
--- a/src/portable/extension.rs
+++ b/src/portable/extension.rs
@@ -3,7 +3,7 @@ use std::path::Path;
 
 use anyhow::Context;
 use edgedb_cli_derive::IntoArgs;
-use log::trace;
+use log::{debug, trace};
 use prettytable::{row, Table};
 
 use crate::branding::BRANDING_CLOUD;
@@ -42,13 +42,13 @@ pub struct Command {
 
 #[derive(clap::Subcommand, Clone, Debug)]
 pub enum Subcommands {
-    /// List installed extensions for a local instance.
+    /// List installed extensions
     List(ExtensionList),
-    /// List available extensions for a local instance.
+    /// List available extensions
     ListAvailable(ExtensionListAvailable),
-    /// Install an extension for a local instance.
+    /// Install an extension
     Install(ExtensionInstall),
-    /// Uninstall an extension from a local instance.
+    /// Uninstall an extension
     Uninstall(ExtensionUninstall),
 }
 
@@ -77,7 +77,6 @@ pub struct ExtensionInstall {
     pub instance: Option<InstanceName>,
 
     /// Name of the extension to install
-    #[arg(short = 'E', long)]
     pub extension: String,
     /// Specify the channel override (stable, testing, or nightly)
     #[arg(long, hide = true)]
@@ -89,14 +88,13 @@ pub struct ExtensionInstall {
     #[arg(long, hide = true)]
     pub reinstall: bool,
 }
-/// Represents the options for uninstalling an extension from a local EdgeDB instance.
+
 #[derive(clap::Args, IntoArgs, Debug, Clone)]
 pub struct ExtensionUninstall {
     #[arg(from_global)]
     pub instance: Option<InstanceName>,
 
-    /// The name of the extension to uninstall.
-    #[arg(short = 'E', long)]
+    /// Name of the extension to uninstall
     pub extension: String,
 }
 
@@ -182,9 +180,9 @@ fn install(cmd: &ExtensionInstall, _options: &Options) -> Result<(), anyhow::Err
     }
 
     let version = inst.get_version()?.specific();
-    let channel = cmd.channel.unwrap_or(Channel::from_version(&version)?);
-    let slot = cmd.slot.clone().unwrap_or(version.slot());
-    trace!("Instance: {version} {channel:?} {slot}");
+    let channel = cmd.channel.unwrap_or(Channel::Stable);
+    let slot = cmd.slot.clone().unwrap_or(version.extension_server_slot());
+    debug!("Instance: {version} {channel:?} {slot}");
     let packages = get_platform_extension_packages(channel, &slot, get_server()?)?;
 
     let package = packages
@@ -257,14 +255,14 @@ fn list_available(list: &ExtensionListAvailable, _options: &Options) -> Result<(
     let inst = get_local_instance(&list.instance)?;
 
     let version = inst.get_version()?.specific();
-    let channel = list.channel.unwrap_or(Channel::from_version(&version)?);
-    let slot = list.slot.clone().unwrap_or(version.slot());
-    trace!("Instance: {version} {channel:?} {slot}");
+    let channel = list.channel.unwrap_or(Channel::Stable);
+    let slot = list.slot.clone().unwrap_or(version.extension_server_slot());
+    debug!("Instance: {version} {channel:?} {slot}");
     let packages = get_platform_extension_packages(channel, &slot, get_server()?)?;
 
     let mut table = Table::new();
     table.set_format(*table::FORMAT);
-    table.add_row(row!["Name", "Version"]);
+    table.set_titles(row!["Name", "Version"]);
     for pkg in packages {
         let ext = pkg.tags.get("extension").cloned().unwrap_or_default();
         table.add_row(row![ext, pkg.version]);
diff --git a/src/portable/instance/upgrade.rs b/src/portable/instance/upgrade.rs
index 3395a3b..8c6fd3b 100644
--- a/src/portable/instance/upgrade.rs
+++ b/src/portable/instance/upgrade.rs
@@ -488,6 +488,7 @@ pub async fn dump_instance(inst: &InstanceInfo, destination: &Path) -> anyhow::R
         command_line: true,
         styler: None,
         conn_params: Connector::new(Ok(config)),
+        instance_name: Some(InstanceName::Local(inst.name.clone())),
     };
     commands::dump_all(
         &mut cli,
@@ -578,6 +579,7 @@ async fn restore_instance(inst: &InstanceInfo, path: &Path) -> anyhow::Result<()
         command_line: true,
         styler: None,
         conn_params: Connector::new(Ok(cfg)),
+        instance_name: Some(InstanceName::Local(inst.name.clone())),
     };
     commands::restore_all(
         &mut cli,
diff --git a/src/portable/project/info.rs b/src/portable/project/info.rs
index 1f42dc3..b5b3cf2 100644
--- a/src/portable/project/info.rs
+++ b/src/portable/project/info.rs
@@ -5,18 +5,18 @@ use clap::ValueHint;
 use const_format::concatcp;
 use gel_tokio::get_stash_path;
 
+use crate::branding::BRANDING_CLI_CMD;
 use crate::branding::BRANDING_CLOUD;
-use crate::branding::{BRANDING_CLI_CMD, MANIFEST_FILE_DISPLAY_NAME};
 use crate::commands::ExitCode;
 use crate::portable::project;
 use crate::print::{self, msg, Highlight};
 use crate::table;
 
 pub fn run(options: &Command) -> anyhow::Result<()> {
-    let Some(project) = project::find_project(options.project_dir.as_deref())? else {
-        anyhow::bail!("`{MANIFEST_FILE_DISPLAY_NAME}` not found, unable to get project info.");
-    };
-    let stash_dir = get_stash_path(&project.root)?;
+    let ctx = project::ensure_ctx(options.project_dir.as_deref())?;
+    let schema_dir = ctx.resolve_schema_dir()?;
+
+    let stash_dir = get_stash_path(&ctx.location.root)?;
     if !stash_dir.exists() {
         msg!(
             "{} {} Run `{BRANDING_CLI_CMD} project init`.",
@@ -52,6 +52,13 @@ pub fn run(options: &Command) -> anyhow::Result<()> {
                     println!("{profile}");
                 }
             }
+            "schema-dir" => {
+                if options.json {
+                    println!("{}", serde_json::to_string(&schema_dir)?);
+                } else {
+                    println!("{}", schema_dir.display());
+                }
+            }
             _ => unreachable!(),
         }
     } else if options.json {
@@ -60,13 +67,19 @@ pub fn run(options: &Command) -> anyhow::Result<()> {
             serde_json::to_string_pretty(&JsonInfo {
                 instance_name: &instance_name,
                 cloud_profile: cloud_profile.as_deref(),
-                root: &project.root,
+                root: &ctx.location.root,
+                schema_dir: &schema_dir,
             })?
         );
     } else {
-        let root = project.root.display().to_string();
-        let mut rows: Vec<(&str, String)> =
-            vec![("Instance name", instance_name), ("Project root", root)];
+        let root = ctx.location.root.display().to_string();
+        let schema_dir = schema_dir.display().to_string();
+
+        let mut rows: Vec<(&str, String)> = vec![
+            ("Instance name", instance_name),
+            ("Project root", root),
+            ("Schema dir", schema_dir),
+        ];
         if let Some(profile) = cloud_profile.as_deref() {
             rows.push((concatcp!(BRANDING_CLOUD, " profile"), profile.to_string()));
         }
@@ -92,6 +105,7 @@ pub struct Command {
     #[arg(long, value_parser=[
         "instance-name",
         "cloud-profile",
+        "schema-dir",
     ])]
     /// Get a specific value:
     ///
@@ -106,4 +120,5 @@ struct JsonInfo<'a> {
     #[serde(skip_serializing_if = "Option::is_none")]
     cloud_profile: Option<&'a str>,
     root: &'a Path,
+    schema_dir: &'a Path,
 }
diff --git a/src/portable/project/init.rs b/src/portable/project/init.rs
index f404fc0..09bec73 100644
--- a/src/portable/project/init.rs
+++ b/src/portable/project/init.rs
@@ -1261,6 +1261,7 @@ async fn migrate_async(inst: &project::Handle<'_>, ask_for_running: bool) -> any
             command_line: true,
             styler: None,
             conn_params: Connector::new(inst.get_builder()?.build_env().await.map_err(Into::into)),
+            instance_name: Some(InstanceName::Local(inst.name.clone())),
         },
     )
     .await?;
diff --git a/src/portable/ver.rs b/src/portable/ver.rs
index 1775c7c..9755036 100644
--- a/src/portable/ver.rs
+++ b/src/portable/ver.rs
@@ -218,13 +218,13 @@ impl Specific {
         matches!(self.minor, MinorVersion::Minor(_))
     }
 
-    pub fn slot(&self) -> String {
+    pub fn extension_server_slot(&self) -> String {
         match self.minor {
             MinorVersion::Minor(_) => self.major.to_string(),
             MinorVersion::Dev(v) => format!("{}-dev{}", self.major, v),
-            MinorVersion::Alpha(v) => format!("{}-alpha{}", self.major, v),
-            MinorVersion::Beta(v) => format!("{}-beta{}", self.major, v),
-            MinorVersion::Rc(v) => format!("{}-rc{}", self.major, v),
+            MinorVersion::Alpha(v) => format!("{}-alpha-{}", self.major, v),
+            MinorVersion::Beta(v) => format!("{}-beta-{}", self.major, v),
+            MinorVersion::Rc(v) => format!("{}-rc-{}", self.major, v),
         }
     }
 }
diff --git a/src/process.rs b/src/process.rs
index 3dee496..f5cced4 100644
--- a/src/process.rs
+++ b/src/process.rs
@@ -242,6 +242,11 @@ impl Native {
         self
     }
 
+    pub fn current_dir(&mut self, path: impl AsRef<Path>) -> &mut Self {
+        self.command.current_dir(path);
+        self
+    }
+
     pub fn env(&mut self, key: impl AsRef<OsStr>, val: impl AsRef<OsStr>) -> &mut Self {
         self.envs.insert(
             key.as_ref().to_os_string(),
@@ -361,7 +366,6 @@ impl Native {
     }
 
     async fn _run(&mut self, capture_out: bool, capture_err: bool) -> anyhow::Result<Output> {
-        let term = interrupt::Interrupt::term();
         let mut stdout = Vec::new();
         let mut stderr = Vec::new();
         log::info!("Running {}: {:?}", self.description, self.command);
@@ -391,11 +395,10 @@ impl Native {
                     stdout_loop(mark, err, capture_err.then_some(&mut stderr), self.quiet),
                 )
             } => child_result,
-            _ = self.signal_loop(pid, &term) => unreachable!(),
+            _ = self.signal_loop_tokio(pid) => unreachable!(),
         };
 
         remove_pid_file(&self.pid_file);
-        term.err_if_occurred()?;
 
         let status = child_result.with_context(|| {
             format!(
@@ -652,6 +655,35 @@ impl Native {
         wait_forever().await
     }
 
+    #[cfg(windows)]
+    async fn signal_loop_tokio(&self, _: u32) -> io::Result<()> {
+        // on windows Ctrl+C signals are propagated automatically and no other
+        // signals are supported, so there is nothing to do here
+        wait_forever().await;
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    async fn signal_loop_tokio(&self, pid: u32) -> io::Result<()> {
+        use signal_hook::consts::signal::{SIGINT, SIGKILL};
+        use std::time::Duration;
+        use tokio::time::timeout;
+
+        tokio::signal::ctrl_c().await?;
+        if self.try_stop_process().await.is_err() {
+            unsafe { libc::kill(pid as i32, SIGINT) };
+        }
+
+        timeout(Duration::from_secs(10), pending::<()>()).await.ok();
+        log::warn!(
+            "Process {} did not stop within 10 seconds, forcing...",
+            self.description
+        );
+        unsafe { libc::kill(pid as i32, SIGKILL) };
+        wait_forever().await;
+        Ok(())
+    }
+
     async fn try_stop_process(&self) -> Result<(), ()> {
         if let Some(stop_fn) = &self.stop_process {
             let mut stop_cmd = stop_fn();
diff --git a/src/watch/fs_watcher.rs b/src/watch/fs_watcher.rs
index 2b63581..d9f382d 100644
--- a/src/watch/fs_watcher.rs
+++ b/src/watch/fs_watcher.rs
@@ -5,8 +5,6 @@ use notify::{EventKind, RecursiveMode, Watcher};
 use tokio::sync::mpsc;
 use tokio::time::Duration;
 
-use crate::interrupt::Interrupt;
-
 const STABLE_TIME: Duration = Duration::from_millis(100);
 
 pub struct FsWatcher {
@@ -33,11 +31,10 @@ impl FsWatcher {
 
     /// Waits for either changes in fs, timeout or interrupt signal
     pub async fn wait(&mut self, timeout: Option<Duration>) -> Event {
-        let ctrl_c = Interrupt::ctrl_c();
         tokio::select! {
             changes = self.wait_for_changes() => Event::Changed(changes),
             _ = wait_for_timeout(timeout) => Event::Retry,
-            _ = ctrl_c.wait() => Event::Abort,
+            _ = tokio::signal::ctrl_c() => Event::Abort,
         }
     }
 
diff --git a/src/watch/migrate.rs b/src/watch/migrate.rs
index e15c051..fb947a4 100644
--- a/src/watch/migrate.rs
+++ b/src/watch/migrate.rs
@@ -1,3 +1,4 @@
+use std::sync::Arc;
 use std::time::Duration;
 
 use const_format::concatcp;
@@ -5,22 +6,62 @@ use const_format::concatcp;
 use edgeql_parser::helpers::quote_string;
 use gel_tokio::Error;
 use indicatif::ProgressBar;
+use tokio::sync::mpsc::UnboundedReceiver;
 
 use crate::branding::BRANDING_CLI_CMD;
-use crate::connect::Connection;
+use crate::connect::{Connection, Connector};
 use crate::migrations::{self, dev_mode};
+use crate::print;
 
-impl super::WatchContext {
-    pub async fn migration_apply_dev_mode(&mut self) -> anyhow::Result<()> {
-        let ctx = migrations::Context::for_project(self.project.clone())?;
+use super::{Context, ExecutionOrder, Matcher};
 
+pub struct Migrator {
+    ctx: Arc<Context>,
+    migration_ctx: migrations::Context,
+
+    connector: Connector,
+    is_force_database_error: bool,
+}
+
+impl Migrator {
+    pub async fn new(ctx: Arc<Context>) -> anyhow::Result<Self> {
+        let connector = ctx.options.create_connector().await?;
+        Ok(Migrator {
+            migration_ctx: migrations::Context::for_project(ctx.project.clone())?,
+            ctx,
+            connector,
+            is_force_database_error: false,
+        })
+    }
+
+    pub async fn run(
+        mut self,
+        mut input: UnboundedReceiver<ExecutionOrder>,
+        matcher: Arc<Matcher>,
+    ) {
+        while let Some(order) = ExecutionOrder::recv(&mut input).await {
+            order.print(&matcher, self.ctx.as_ref());
+
+            let res = self.migration_apply_dev_mode().await;
+
+            if let Err(e) = &res {
+                print::error!("{e}");
+                // TODO
+                // matcher.should_retry = true;
+            }
+        }
+
+        self.cleanup().await;
+    }
+
+    async fn migration_apply_dev_mode(&mut self) -> anyhow::Result<()> {
         let bar = ProgressBar::new_spinner();
         bar.enable_steady_tick(Duration::from_millis(100));
         bar.set_message("Connecting");
         let mut cli = self.connector.connect().await?;
 
         let old_state = cli.set_ignore_error_state();
-        let result = dev_mode::migrate(&mut cli, &ctx, &bar).await;
+        let result = dev_mode::migrate(&mut cli, &self.migration_ctx, &bar).await;
         cli.restore_state(old_state);
 
         bar.finish_and_clear();
@@ -41,7 +82,7 @@ impl super::WatchContext {
         Ok(())
     }
 
-    pub async fn cleanup(&mut self) {
+    async fn cleanup(&mut self) {
         if !self.is_force_database_error {
             return;
         }
diff --git a/src/watch/mod.rs b/src/watch/mod.rs
index 0511f98..be98c9d 100644
--- a/src/watch/mod.rs
+++ b/src/watch/mod.rs
@@ -1,11 +1,20 @@
 mod fs_watcher;
 mod migrate;
+mod scripts;
 
 pub use fs_watcher::{Event, FsWatcher};
+pub use scripts::run_script;
+
+use std::collections::HashSet;
+use std::iter::FromIterator;
+use std::sync::Arc;
+
+use tokio::sync::mpsc::error::TryRecvError;
+use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+use tokio::task::JoinSet;
 
 #[allow(unused_imports)]
 use crate::branding::{BRANDING_CLI_CMD, MANIFEST_FILE_DISPLAY_NAME};
-use crate::connect::Connector;
 use crate::options::Options;
 use crate::portable::project;
 use crate::print::{self, AsRelativeToCurrentDir, Highlight};
@@ -20,14 +29,6 @@ pub struct Command {
     #[arg(short = 'm', long)]
     pub migrate: bool,
 
-    /// On failure, retry after a given number of seconds
-    #[arg(short = 'r', long)]
-    pub retry_sec: Option<u16>,
-
-    /// Don't suspend watch during script execution. Might cause loops.
-    #[arg(short = 'c', long)]
-    pub continuous: bool,
-
     #[arg(short = 'v', long)]
     pub verbose: bool,
 }
@@ -35,14 +36,14 @@ pub struct Command {
 #[tokio::main(flavor = "current_thread")]
 pub async fn run(options: &Options, cmd: &Command) -> anyhow::Result<()> {
     let project = project::ensure_ctx_async(None).await?;
-    let mut ctx = WatchContext {
+    let ctx = Arc::new(Context {
         project,
-        connector: options.create_connector().await?,
-        is_force_database_error: false,
-    };
+        options: options.clone(),
+        cmd: cmd.clone(),
+    });
 
     // determine what we will be watching
-    let mut matchers = assemble_matchers(cmd, &ctx)?;
+    let matchers = assemble_matchers(cmd, &ctx.project)?;
 
     if cmd.migrate {
         print::msg!(
@@ -69,126 +70,41 @@ pub async fn run(options: &Options, cmd: &Command) -> anyhow::Result<()> {
     }
     print::msg!("");
 
-    let mut watcher = fs_watcher::FsWatcher::new()?;
-    // TODO: watch only directories that are needed, not the whole project
+    // spawn tasks that will execute the scripts
+    // these tasks wait for ExecutionOrders to be emitted into `tx`
+    let (tx, join_handle) = start_executors(&matchers, &ctx).await?;
 
-    watcher.watch(&ctx.project.location.root, notify::RecursiveMode::Recursive)?;
-    let schema_dir = ctx.project.manifest.project().get_schema_dir();
-    if cmd.migrate && !schema_dir.starts_with(&ctx.project.location.root) {
-        watcher.watch(
-            &ctx.project.manifest.project().get_schema_dir(),
-            notify::RecursiveMode::Recursive,
-        )?;
-    }
+    // watch file system, debounce and match to globs
+    // sends events to executors via tx channel
+    watch_and_match(&matchers, &tx, &ctx).await?;
 
-    loop {
-        let timeout = if let Some(retry_sec) = &cmd.retry_sec {
-            let any_failed = matchers.iter().any(|m| m.last_failed);
-            any_failed.then(|| tokio::time::Duration::from_secs(*retry_sec as u64))
-        } else {
-            None
-        };
-        if let Some(timeout) = timeout {
-            print::warn!("Retrying in {} sec...", timeout.as_secs());
-        }
-
-        if !cmd.continuous {
-            watcher.clear_queue();
-        }
-
-        // wait for changes
-        let event = watcher.wait(timeout).await;
-
-        let changed_paths = match event {
-            Event::Changed(paths) => paths,
-            Event::Retry => Default::default(),
-            Event::Abort => break,
-        };
-        // strip prefix
-        let changed_paths: Vec<_> = changed_paths
-            .iter()
-            .flat_map(|p| p.strip_prefix(&ctx.project.location.root).ok())
-            .map(|p| (p, globset::Candidate::new(p)))
-            .collect();
-
-        // run all matching scripts
-        for matcher in &mut matchers {
-            // does it match?
-            let (is_matched, reason) = if matcher.last_failed {
-                (true, "(retry)".to_string())
-            } else {
-                let matched_paths = changed_paths
-                    .iter()
-                    .filter(|x| matcher.matcher.is_match_candidate(&x.1))
-                    .map(|x| x.0.display().to_string())
-                    .collect::<Vec<_>>();
-                (!matched_paths.is_empty(), matched_paths.join(", "))
-            };
-
-            if !is_matched {
-                continue;
-            }
-
-            // print
-            print::msg!(
-                "{}",
-                format!(
-                    "--- {}: {} ---",
-                    matcher.glob,
-                    matcher.target.to_string().muted()
-                )
-            );
-            if cmd.verbose {
-                print::msg!("{}", format!("  triggered by: {reason}").muted());
-            }
-
-            // run
-            let success = match &matcher.target {
-                Target::Script(script) => {
-                    let status =
-                        crate::hooks::run_script(script, &ctx.project.location.root).await?;
-
-                    if !status.success() {
-                        print::error!("script exited with status {status}");
-                    }
-                    status.success()
-                }
-                Target::MigrateDevMode => {
-                    let res = ctx.migration_apply_dev_mode().await;
-
-                    if let Err(e) = &res {
-                        print::error!("{e}");
-                    }
-                    res.is_ok()
-                }
-            };
-
-            matcher.last_failed = !success;
-        }
-
-        print::msg!(""); // a bit of space between runs
-    }
-
-    if cmd.migrate {
-        ctx.cleanup().await;
+    // close all tx
+    for t in tx {
+        drop(t);
     }
+    // wait for executors to finish
+    join_handle.join_all().await;
 
     Ok(())
 }
 
-struct WatchContext {
+/// Information about the current watch process
+struct Context {
     project: project::Context,
-
-    // things needed for migrate
-    connector: Connector,
-    is_force_database_error: bool,
+    options: Options,
+    cmd: Command,
 }
 
 struct Matcher {
     glob: String,
     matcher: globset::GlobMatcher,
     target: Target,
-    last_failed: bool,
+}
+
+impl Matcher {
+    fn name(&self) -> &str {
+        self.glob.as_str()
+    }
 }
 
 enum Target {
@@ -208,8 +124,11 @@ impl std::fmt::Display for Target {
     }
 }
 
-fn assemble_matchers(cmd: &Command, ctx: &WatchContext) -> anyhow::Result<Vec<Matcher>> {
-    let watch = ctx.project.manifest.watch.as_ref();
+fn assemble_matchers(
+    cmd: &Command,
+    project: &project::Context,
+) -> anyhow::Result<Vec<Arc<Matcher>>> {
+    let watch = project.manifest.watch.as_ref();
     let files = match watch.and_then(|x| x.files.as_ref()) {
         Some(files) => files.clone(),
         None if cmd.migrate => Default::default(),
@@ -225,29 +144,144 @@ fn assemble_matchers(cmd: &Command, ctx: &WatchContext) -> anyhow::Result<Vec<Ma
     for (glob_str, script) in files {
         let glob = globset::Glob::new(&glob_str)?;
 
-        matchers.push(Matcher {
+        matchers.push(Arc::new(Matcher {
             glob: glob_str,
             matcher: glob.compile_matcher(),
             target: Target::Script(script),
-            last_failed: false,
-        });
+        }));
     }
     matchers.sort_by(|a, b| b.glob.cmp(&a.glob));
 
     if cmd.migrate {
-        let schema_dir = ctx.project.manifest.project().get_schema_dir();
+        let schema_dir = project.manifest.project().get_schema_dir();
         let schema_dir = schema_dir
             .to_str()
             .ok_or_else(|| anyhow::anyhow!("bad path: {}", schema_dir.display()))?;
         let glob_str = format!("{schema_dir}/**/*.{{gel,esdl}}");
         let glob = globset::Glob::new(&glob_str)?;
-        matchers.push(Matcher {
+        matchers.push(Arc::new(Matcher {
             glob: glob_str,
             matcher: glob.compile_matcher(),
             target: Target::MigrateDevMode,
-            last_failed: false,
-        });
+        }));
     }
 
     Ok(matchers)
 }
+
+async fn start_executors(
+    matchers: &[Arc<Matcher>],
+    ctx: &Arc<Context>,
+) -> anyhow::Result<(Vec<UnboundedSender<ExecutionOrder>>, JoinSet<()>)> {
+    let mut senders = Vec::with_capacity(matchers.len());
+    let mut join_set = JoinSet::new();
+    for matcher in matchers {
+        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+        senders.push(tx);
+
+        match &matcher.target {
+            Target::Script(_) => join_set.spawn(scripts::execute(rx, matcher.clone(), ctx.clone())),
+            Target::MigrateDevMode => {
+                let migrator = migrate::Migrator::new(ctx.clone()).await?;
+
+                join_set.spawn(migrator.run(rx, matcher.clone()))
+            }
+        };
+    }
+    Ok((senders, join_set))
+}
+
+async fn watch_and_match(
+    matchers: &[Arc<Matcher>],
+    tx: &[UnboundedSender<ExecutionOrder>],
+    ctx: &Arc<Context>,
+) -> anyhow::Result<()> {
+    let mut watcher = fs_watcher::FsWatcher::new()?;
+    // TODO: watch only directories that are needed, not the whole project
+
+    watcher.watch(&ctx.project.location.root, notify::RecursiveMode::Recursive)?;
+    let schema_dir = ctx.project.manifest.project().get_schema_dir();
+    if ctx.cmd.migrate && !schema_dir.starts_with(&ctx.project.location.root) {
+        watcher.watch(
+            &ctx.project.manifest.project().get_schema_dir(),
+            notify::RecursiveMode::Recursive,
+        )?;
+    }
+
+    loop {
+        // wait for changes
+        let event = watcher.wait(None).await;
+
+        let changed_paths = match event {
+            Event::Changed(paths) => paths,
+            Event::Retry => Default::default(),
+            Event::Abort => break,
+        };
+        // strip prefix
+        let changed_paths: Vec<_> = changed_paths
+            .iter()
+            .flat_map(|p| p.strip_prefix(&ctx.project.location.root).ok())
+            .map(|p| (p, globset::Candidate::new(p)))
+            .collect();
+
+        // run all matching scripts
+        for (matcher, tx) in std::iter::zip(matchers, tx) {
+            // does it match?
+            let matched_paths = changed_paths
+                .iter()
+                .filter(|x| matcher.matcher.is_match_candidate(&x.1))
+                .map(|x| x.0.display().to_string())
+                .collect::<Vec<_>>();
+            if matched_paths.is_empty() {
+                continue;
+            }
+
+            let order = ExecutionOrder {
+                matched_paths: HashSet::from_iter(matched_paths),
+            };
+            tx.send(order).unwrap();
+        }
+    }
+    Ok(())
+}
+
+struct ExecutionOrder {
+    matched_paths: HashSet<String>,
+}
+
+impl ExecutionOrder {
+    fn merge(&mut self, other: ExecutionOrder) {
+        self.matched_paths.extend(other.matched_paths);
+    }
+
+    async fn recv(input: &mut UnboundedReceiver<ExecutionOrder>) -> Option<ExecutionOrder> {
+        let mut order = input.recv().await?;
+        loop {
+            match input.try_recv() {
+                Ok(o) => order.merge(o),
+                Err(TryRecvError::Empty) => break,
+                Err(TryRecvError::Disconnected) => return None,
+            }
+        }
+        Some(order)
+    }
+
+    fn print(&self, matcher: &Matcher, ctx: &Context) {
+        // print
+        print::msg!(
+            "{}",
+            format!(
+                "--- {}: {} ---",
+                matcher.name(),
+                matcher.target.to_string().muted()
+            )
+        );
+        if ctx.cmd.verbose {
+            let mut matched_paths: Vec<_> = self.matched_paths.iter().map(|p| p.as_str()).collect();
+            matched_paths.sort();
+            let reason = matched_paths.join(", ");
+
+            print::msg!("{}", format!("  triggered by: {reason}").muted());
+        }
+    }
+}
diff --git a/src/watch/scripts.rs b/src/watch/scripts.rs
new file mode 100644
index 0000000..a973201
--- /dev/null
+++ b/src/watch/scripts.rs
@@ -0,0 +1,63 @@
+use std::path;
+use std::sync::Arc;
+use tokio::sync::mpsc::UnboundedReceiver;
+
+use crate::portable::windows;
+use crate::print;
+use crate::process;
+
+use super::Target;
+use super::{Context, ExecutionOrder, Matcher};
+
+pub async fn execute(
+    mut input: UnboundedReceiver<ExecutionOrder>,
+    matcher: Arc<Matcher>,
+    ctx: Arc<Context>,
+) {
+    let project_root = &ctx.project.location.root;
+
+    while let Some(order) = ExecutionOrder::recv(&mut input).await {
+        order.print(&matcher, ctx.as_ref());
+
+        let Target::Script(script) = &matcher.target else {
+            unreachable!()
+        };
+
+        let res = run_script(matcher.name(), script, project_root).await;
+
+        match res {
+            Ok(status) => {
+                if !status.success() {
+                    print::error!("script exited with status {status}");
+                }
+            }
+            Err(e) => {
+                print::error!("{e}")
+            }
+        }
+    }
+}
+
+pub async fn run_script(
+    marker: &str,
+    script: &str,
+    current_dir: &path::Path,
+) -> Result<std::process::ExitStatus, anyhow::Error> {
+    let status = if !cfg!(windows) {
+        let marker = marker.to_string();
+        process::Native::new("", marker, "/bin/sh")
+            .arg("-c")
+            .arg(script)
+            .current_dir(current_dir)
+            .run_for_status()
+            .await?
+    } else {
+        let wsl = windows::try_get_wsl()?;
+        wsl.sh(current_dir)
+            .arg("-c")
+            .arg(script)
+            .run_for_status()
+            .await?
+    };
+    Ok(status)
+}


Test patch:
diff --git a/tests/func/interactive.rs b/tests/func/interactive.rs
index 6250d14..a6fa4e7 100644
--- a/tests/func/interactive.rs
+++ b/tests/func/interactive.rs
@@ -48,7 +48,8 @@ fn create_report() {
 
     cmd.exp_string(&format!("{main}>")).unwrap();
     cmd.send_line("CREATE TYPE default::Type1;\n").unwrap();
-    cmd.exp_string("OK: CREATE").unwrap();
+    cmd.exp_string("OK: ").unwrap();
+    cmd.exp_string("CREATE").unwrap();
 }
 
 #[test]