diff --git a/README.md b/README.md
index 69252708..749711ae 100644
--- a/README.md
+++ b/README.md
@@ -11,6 +11,32 @@ SchemaBot provides a fully interactive CLI for planning, applying, and monitorin
 
 ![SchemaBot CLI Demo](./docs/assets/cli-demo.gif)
 
+### Quick start
+
+Use the CLI to manage schema changes on any MySQL database you can connect to. The `--dsn` flag accepts any MySQL connection string — local, remote, RDS, etc. Schema changes are applied online using [Spirit](https://github.com/block/spirit) — reads and writes continue uninterrupted during the change.
+
+```bash
+# Install
+go install github.com/block/schemabot/pkg/cmd@latest
+
+# Start local mysql on port 3306 if not running
+brew install mysql && mysql.server start
+
+# Ensure $GOPATH/bin is on your PATH
+export PATH="$PATH:$(go env GOPATH)/bin"
+
+# Pull your existing schema into a declarative schema directory
+schemabot pull --dsn "root@tcp(localhost:3306)/mydb" -e staging -o ./schema
+
+# Edit a .sql file, then plan and apply
+schemabot plan -s ./schema -e staging
+schemabot apply -s ./schema -e staging
+```
+
+A background server auto-starts on first use and stores state in `_schemabot` on your local MySQL.
+
+This quickstart runs entirely on your machine. For GitHub PR integration, long-running schema changes, or managing lots of databases at scale, deploy SchemaBot as a server — see [deploy/](./deploy/) for sample code and [GitHub App setup](./docs/github-app-setup.md) (docs are still a work in progress).
+
 ## PR Demo
 
 _Coming soon_
@@ -53,7 +79,9 @@ SchemaBot handles the full lifecycle:
 
 Simple changes (e.g., adding a column) use instant DDL and complete in milliseconds. Operations that require a row copy (e.g., adding an index) run online without blocking reads or writes.
 
-## Quick Start
+## Docker demo
+
+Create a Docker environment that runs schemabot server and test databases
 
 ```bash
 make demo    # Start services, apply schema, seed data
diff --git a/docs/configuration.md b/docs/configuration.md
index 574468f3..1323581e 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -1,10 +1,113 @@
 # Configuration
 
-SchemaBot loads config from the `SCHEMABOT_CONFIG_FILE` environment variable.
+- [Config Files](#config-files)
+- [Local](#local)
+- [Server Deployment](#server-deployment)
+- [Profiles](#profiles)
+- [Secret Resolution](#secret-resolution)
 
-## Local Mode
+## Config Files
 
-The SchemaBot process runs the engine directly. This is best for most users and recommended to start with.
+SchemaBot uses different config files depending on how you run it:
+
+| File | Location | Purpose | Used by |
+|------|----------|---------|---------|
+| `schemabot.yaml` | In your schema directory | Declares the database name and type | CLI (plan, apply, pull) |
+| `~/.schemabot/config.yaml` | Home directory | CLI profiles and local database connections | CLI (all commands) |
+| Server config (`SCHEMABOT_CONFIG_FILE`) | Anywhere | Storage DSN, database configs, Tern endpoints | `schemabot serve` |
+
+`schemabot.yaml` — declares which database your schema files belong to:
+
+```yaml
+database: mydb
+type: mysql
+```
+
+`~/.schemabot/config.yaml` — tells the CLI how to connect:
+
+```yaml
+default_profile: local
+
+profiles:
+  staging:
+    endpoint: http://localhost:8080
+
+local:
+  mydb:
+    type: mysql
+    environments:
+      staging:
+        dsn: "root@tcp(localhost:3306)/mydb"
+```
+
+Server config (`SCHEMABOT_CONFIG_FILE`) — only used when deploying SchemaBot as a server:
+
+```yaml
+storage:
+  dsn: "env:SCHEMABOT_DSN"
+databases:
+  mydb:
+    type: mysql
+    environments:
+      staging:
+        dsn: "env:STAGING_DSN"
+```
+
+The first two are for CLI usage. The server config is separate — see [Server Deployment](#server-deployment) for details.
+
+## Local
+
+When using SchemaBot without a server deployment, the CLI auto-starts a lightweight background server on your machine. Database connection details are stored in the `local` section of the config. The `pull` command sets this up automatically.
+
+```yaml
+# ~/.schemabot/config.yaml
+local:
+  mydb:
+    type: mysql
+    environments:
+      staging:
+        dsn: "root@tcp(localhost:3306)/mydb"
+      production:
+        dsn: "env:PRODUCTION_DSN"
+  mypsdb:
+    type: vitess
+    environments:
+      production:
+        organization: myorg
+        token: "env:PLANETSCALE_SERVICE_TOKEN"
+```
+
+### Adding a database
+
+```bash
+# Pull creates the local config entry automatically
+schemabot pull --dsn "root@tcp(localhost:3306)/mydb" -e staging -o ./schema
+```
+
+This writes the schema files and adds `mydb/staging` to `~/.schemabot/config.yaml`.
+
+### Managing the background server
+
+```bash
+schemabot local status   # show server state, PID, port
+schemabot local stop     # stop the background server
+schemabot local reset    # stop server and drop _schemabot database
+```
+
+## Server Deployment
+
+For GitHub PR integration, team coordination, or managing many databases, deploy SchemaBot as a server.
+
+The server loads config from a YAML file. Set `SCHEMABOT_CONFIG_FILE` to the path:
+
+```bash
+export SCHEMABOT_CONFIG_FILE=/etc/schemabot/config.yaml
+schemabot serve
+```
+
+### Single-process (recommended for most deployments)
+
+The server runs the schema change engine directly:
 
 ```yaml
 storage:
@@ -20,9 +123,9 @@ databases:
         dsn: "file:/run/secrets/prod-dsn"
 ```
 
-## gRPC Mode
+### Distributed (gRPC)
 
-SchemaBot delegates to remote services that implement the Tern proto. This is useful for distributed deployments where schema changes need to run in separate isolated environments.
+For deployments where schema changes need to run in separate isolated environments, SchemaBot delegates to remote services that implement the Tern proto:
 
 ```yaml
 storage:
@@ -36,9 +139,36 @@ tern_deployments:
     production: "tern1-production:9090"
 ```
 
+## Profiles
+
+CLI profiles let you switch between local mode and server deployments. Use `--profile` to select one, or set a default.
+
+```yaml
+# ~/.schemabot/config.yaml
+default_profile: local
+
+profiles:
+  staging:
+    endpoint: http://localhost:8080
+  production:
+    endpoint: https://schemabot.example.com
+```
+
+```bash
+schemabot plan -s ./schema -e staging                    # uses default profile
+schemabot plan -s ./schema -e staging --profile staging   # explicit server profile
+schemabot plan -s ./schema -e staging --profile local     # force local mode
+```
+
+`local` is a reserved profile name that triggers local mode. Set `default_profile: local` to make it the default.
+
+```bash
+schemabot configure   # interactive profile setup
+```
+
 ## Secret Resolution
 
-DSN values support secret resolution prefixes:
+DSN and credential values support secret resolution prefixes:
 
 | Prefix | Example | Description |
 |---|---|---|
diff --git a/integration/local_mode_test.go b/integration/local_mode_test.go
new file mode 100644
index 00000000..a7562c46
--- /dev/null
+++ b/integration/local_mode_test.go
@@ -0,0 +1,178 @@
+//go:build integration
+
+package integration
+
+import (
+	"bytes"
+	"database/sql"
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/schemabot/pkg/e2eutil"
+)
+
+var cliFinder e2eutil.CLIFinder
+
+// TestLocalMode_PullPlanApply exercises the full local mode UX:
+// pull schema from a live database, edit a file, plan, and apply —
+// with the background daemon auto-starting just like a real user.
+func TestLocalMode_PullPlanApply(t *testing.T) {
+	ctx := t.Context()
+
+	// Build a storage DSN from the target MySQL container.
+	// The testcontainer uses root:testpassword, so we need the full DSN.
+	storageDSN := strings.Replace(targetDSN, "/target_test", "/_schemabot", 1)
+
+	// Build the CLI binary (cached across tests in this package)
+	binPath := cliFinder.FindOrBuild(t, "..", "./pkg/cmd", "../bin/schemabot")
+
+	// Create a test database with a table
+	db, err := sql.Open("mysql", targetDSN+"&multiStatements=true")
+	require.NoError(t, err)
+	defer func() { _ = db.Close() }()
+
+	dbName := fmt.Sprintf("local_mode_%d", time.Now().UnixNano()%100000)
+	_, err = db.ExecContext(ctx, "CREATE DATABASE IF NOT EXISTS `"+dbName+"`")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_, _ = db.ExecContext(ctx, "DROP DATABASE IF EXISTS `"+dbName+"`")
+		// Clean up _schemabot storage database created by the daemon
+		_, _ = db.ExecContext(ctx, "DROP DATABASE IF EXISTS `_schemabot`")
+	})
+
+	appDSN := strings.Replace(targetDSN, "/target_test", "/"+dbName, 1)
+	appDB, err := sql.Open("mysql", appDSN+"&multiStatements=true")
+	require.NoError(t, err)
+	defer func() { _ = appDB.Close() }()
+
+	_, err = appDB.ExecContext(ctx, `CREATE TABLE users (
+  id bigint unsigned NOT NULL AUTO_INCREMENT,
+  name varchar(255) NOT NULL,
+  email varchar(255) NOT NULL,
+  PRIMARY KEY (id),
+  UNIQUE KEY email (email)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`)
+	require.NoError(t, err)
+
+	// Set up isolated HOME and schema directory
+	tmpHome := t.TempDir()
+	schemaDir := t.TempDir()
+
+	// Find a free port for the daemon
+	daemonPort := freePort(t)
+
+	// Environment for all CLI commands — isolated HOME, custom storage DSN/port
+	env := append(os.Environ(),
+		"HOME="+tmpHome,
+		"SCHEMABOT_LOCAL_STORAGE_DSN="+storageDSN,
+		"SCHEMABOT_LOCAL_PORT="+fmt.Sprintf("%d", daemonPort),
+	)
+
+	// Helper to run CLI commands with the test environment
+	run := func(args ...string) string {
+		t.Helper()
+		cmd := exec.CommandContext(ctx, binPath, args...)
+		cmd.Env = env
+		var stdout, stderr bytes.Buffer
+		cmd.Stdout = &stdout
+		cmd.Stderr = &stderr
+		err := cmd.Run()
+		output := stdout.String() + stderr.String()
+		require.NoErrorf(t, err, "CLI %v failed:\n%s", args, output)
+		return output
+	}
+
+	runInDir := func(dir string, args ...string) string {
+		t.Helper()
+		cmd := exec.CommandContext(ctx, binPath, args...)
+		cmd.Dir = dir
+		cmd.Env = env
+		var stdout, stderr bytes.Buffer
+		cmd.Stdout = &stdout
+		cmd.Stderr = &stderr
+		err := cmd.Run()
+		output := stdout.String() + stderr.String()
+		require.NoErrorf(t, err, "CLI %v failed:\n%s", args, output)
+		return output
+	}
+
+	// Step 1: Pull schema from the live database
+	pullOutput := run("pull", "--dsn", appDSN, "-o", schemaDir, "-e", "staging", "-d", dbName)
+	t.Logf("Pull output:\n%s", pullOutput)
+	assert.Contains(t, pullOutput, "Pulled 1 table")
+	assert.Contains(t, pullOutput, "users.sql")
+
+	// Verify schema files
+	usersSQL, err := os.ReadFile(filepath.Join(schemaDir, "users.sql"))
+	require.NoError(t, err)
+	assert.Contains(t, string(usersSQL), "CREATE TABLE `users`")
+
+	configData, err := os.ReadFile(filepath.Join(schemaDir, "schemabot.yaml"))
+	require.NoError(t, err)
+	assert.Contains(t, string(configData), "database: "+dbName)
+
+	// Verify local config was created
+	localConfig, err := os.ReadFile(filepath.Join(tmpHome, ".schemabot", "config.yaml"))
+	require.NoError(t, err)
+	assert.Contains(t, string(localConfig), dbName)
+	assert.Contains(t, string(localConfig), "staging")
+
+	// Step 2: Edit a .sql file — add a column
+	modifiedSQL := strings.Replace(string(usersSQL),
+		"PRIMARY KEY (`id`)",
+		"`phone` varchar(20) DEFAULT NULL,\n  PRIMARY KEY (`id`)",
+		1)
+	require.NoError(t, os.WriteFile(filepath.Join(schemaDir, "users.sql"), []byte(modifiedSQL), 0644))
+
+	// Step 3: Plan — this triggers daemon auto-start
+	planOutput := runInDir(schemaDir, "plan", "-s", schemaDir, "-e", "staging", "--json")
+	t.Logf("Plan output:\n%s", planOutput)
+	assert.Contains(t, planOutput, "phone")
+	assert.Contains(t, planOutput, "plan_id")
+
+	// Verify daemon is running
+	t.Cleanup(func() {
+		// Stop the daemon after the test
+		cmd := exec.CommandContext(ctx, binPath, "local", "stop")
+		cmd.Env = env
+		_ = cmd.Run()
+	})
+
+	// Step 4: Apply — the daemon is already running
+	applyOutput := runInDir(schemaDir, "apply", "-s", schemaDir, "-e", "staging", "-y", "-w", "-o", "log")
+	t.Logf("Apply output:\n%s", applyOutput)
+
+	// Step 5: Verify the column was added to the target database
+	var columnExists bool
+	err = appDB.QueryRowContext(ctx,
+		"SELECT COUNT(*) > 0 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = ? AND TABLE_NAME = 'users' AND COLUMN_NAME = 'phone'",
+		dbName,
+	).Scan(&columnExists)
+	require.NoError(t, err)
+	assert.True(t, columnExists, "phone column should exist after apply")
+	t.Log("Verified: phone column exists in target database")
+
+	// Step 6: Verify local server status
+	statusOutput := run("local", "status")
+	assert.Contains(t, statusOutput, "running")
+}
+
+// freePort finds a free TCP port by binding to :0 and returning the assigned port.
+func freePort(t *testing.T) int {
+	t.Helper()
+	l, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	port := l.Addr().(*net.TCPAddr).Port
+	require.NoError(t, l.Close())
+	return port
+}
diff --git a/integration/pull_plan_integration_test.go b/integration/pull_plan_integration_test.go
new file mode 100644
index 00000000..1d27fb6c
--- /dev/null
+++ b/integration/pull_plan_integration_test.go
@@ -0,0 +1,210 @@
+//go:build integration
+
+package integration
+
+import (
+	"database/sql"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/schemabot/pkg/cmd/commands"
+	"github.com/block/schemabot/pkg/local"
+)
+
+// TestPull verifies the pull command pulls schema from a live database
+// and writes canonical .sql files plus schemabot.yaml.
+func TestPull(t *testing.T) {
+	ctx := t.Context()
+
+	db, err := sql.Open("mysql", targetDSN+"&multiStatements=true")
+	require.NoError(t, err)
+	defer func() { _ = db.Close() }()
+
+	dbName := "pull_test"
+	_, err = db.ExecContext(ctx, "CREATE DATABASE IF NOT EXISTS `"+dbName+"`")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_, _ = db.ExecContext(ctx, "DROP DATABASE IF EXISTS `"+dbName+"`")
+	})
+
+	appDSN := strings.Replace(targetDSN, "/target_test", "/"+dbName, 1)
+	appDB, err := sql.Open("mysql", appDSN+"&multiStatements=true")
+	require.NoError(t, err)
+	defer func() { _ = appDB.Close() }()
+
+	_, err = appDB.ExecContext(ctx, "CREATE TABLE `users` (\n  `id` bigint unsigned NOT NULL AUTO_INCREMENT,\n  `name` varchar(255) NOT NULL,\n  `email` varchar(255) NOT NULL,\n  PRIMARY KEY (`id`),\n  UNIQUE KEY `email` (`email`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci")
+	require.NoError(t, err)
+
+	_, err = appDB.ExecContext(ctx, "CREATE TABLE `orders` (\n  `id` bigint unsigned NOT NULL AUTO_INCREMENT,\n  `user_id` bigint unsigned NOT NULL,\n  `total` decimal(10,2) NOT NULL,\n  PRIMARY KEY (`id`),\n  KEY `idx_user_id` (`user_id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci")
+	require.NoError(t, err)
+
+	outputDir := t.TempDir()
+	pullCmd := commands.PullCmd{
+		DSN:         appDSN,
+		OutputDir:   outputDir,
+		Environment: "staging",
+		Type:        "mysql",
+	}
+	err = pullCmd.Run(&commands.Globals{})
+	require.NoError(t, err)
+
+	// Verify .sql files
+	usersSQL, err := os.ReadFile(filepath.Join(outputDir, "users.sql"))
+	require.NoError(t, err)
+	assert.Contains(t, string(usersSQL), "CREATE TABLE `users`")
+	assert.Contains(t, string(usersSQL), "`email`")
+	assert.True(t, strings.HasSuffix(string(usersSQL), ";\n"), "should end with semicolon + newline")
+	assert.NotContains(t, string(usersSQL), "AUTO_INCREMENT=", "AUTO_INCREMENT should be stripped")
+
+	ordersSQL, err := os.ReadFile(filepath.Join(outputDir, "orders.sql"))
+	require.NoError(t, err)
+	assert.Contains(t, string(ordersSQL), "CREATE TABLE `orders`")
+
+	// Verify schemabot.yaml (no DSN — just database + type)
+	configData, err := os.ReadFile(filepath.Join(outputDir, "schemabot.yaml"))
+	require.NoError(t, err)
+	assert.Contains(t, string(configData), "database: "+dbName)
+	assert.Contains(t, string(configData), "type: mysql")
+	assert.NotContains(t, string(configData), "dsn:", "DSN should not be in schemabot.yaml")
+
+	// Verify local config was updated
+	cfg, err := local.LoadCLIConfig()
+	require.NoError(t, err)
+	require.NotNil(t, cfg.Local)
+	localDB, ok := cfg.Local[dbName]
+	require.True(t, ok, "database should be in local config")
+	assert.Equal(t, "mysql", localDB.Type)
+	require.Contains(t, localDB.Environments, "staging")
+	assert.Equal(t, appDSN, localDB.Environments["staging"].DSN)
+}
+
+// TestPullSkipsExistingConfig verifies that pull doesn't overwrite
+// an existing schemabot.yaml.
+func TestPullSkipsExistingConfig(t *testing.T) {
+	ctx := t.Context()
+
+	db, err := sql.Open("mysql", targetDSN+"&multiStatements=true")
+	require.NoError(t, err)
+	defer func() { _ = db.Close() }()
+
+	dbName := "pull_skip_config_test"
+	_, err = db.ExecContext(ctx, "CREATE DATABASE IF NOT EXISTS `"+dbName+"`")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_, _ = db.ExecContext(ctx, "DROP DATABASE IF EXISTS `"+dbName+"`")
+	})
+
+	appDSN := strings.Replace(targetDSN, "/target_test", "/"+dbName, 1)
+	appDB, err := sql.Open("mysql", appDSN)
+	require.NoError(t, err)
+	defer func() { _ = appDB.Close() }()
+
+	_, err = appDB.ExecContext(ctx, "CREATE TABLE `t1` (\n  `id` bigint unsigned NOT NULL AUTO_INCREMENT,\n  PRIMARY KEY (`id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci")
+	require.NoError(t, err)
+
+	outputDir := t.TempDir()
+
+	// Pre-create schemabot.yaml with custom content
+	existing := "database: custom\ntype: mysql\n"
+	err = os.WriteFile(filepath.Join(outputDir, "schemabot.yaml"), []byte(existing), 0644)
+	require.NoError(t, err)
+
+	pullCmd := commands.PullCmd{
+		DSN:         appDSN,
+		OutputDir:   outputDir,
+		Environment: "staging",
+		Type:        "mysql",
+	}
+	err = pullCmd.Run(&commands.Globals{})
+	require.NoError(t, err)
+
+	// Config should not be overwritten
+	configData, err := os.ReadFile(filepath.Join(outputDir, "schemabot.yaml"))
+	require.NoError(t, err)
+	assert.Equal(t, existing, string(configData))
+
+	// But .sql file should still be written
+	_, err = os.ReadFile(filepath.Join(outputDir, "t1.sql"))
+	require.NoError(t, err)
+}
+
+// TestPullMultiEnvironment verifies that pulling a second environment
+// upserts into the existing local config without overwriting the first.
+func TestPullMultiEnvironment(t *testing.T) {
+	ctx := t.Context()
+
+	db, err := sql.Open("mysql", targetDSN+"&multiStatements=true")
+	require.NoError(t, err)
+	defer func() { _ = db.Close() }()
+
+	dbName := "pull_multi_env_test"
+	_, err = db.ExecContext(ctx, "CREATE DATABASE IF NOT EXISTS `"+dbName+"`")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_, _ = db.ExecContext(ctx, "DROP DATABASE IF EXISTS `"+dbName+"`")
+	})
+
+	appDSN := strings.Replace(targetDSN, "/target_test", "/"+dbName, 1)
+	appDB, err := sql.Open("mysql", appDSN)
+	require.NoError(t, err)
+	defer func() { _ = appDB.Close() }()
+
+	_, err = appDB.ExecContext(ctx, "CREATE TABLE `t1` (\n  `id` bigint unsigned NOT NULL AUTO_INCREMENT,\n  PRIMARY KEY (`id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci")
+	require.NoError(t, err)
+
+	outputDir := t.TempDir()
+
+	// Pull staging
+	err = (&commands.PullCmd{
+		DSN:         appDSN,
+		OutputDir:   outputDir,
+		Environment: "staging",
+		Type:        "mysql",
+	}).Run(&commands.Globals{})
+	require.NoError(t, err)
+
+	// Pull production (same DB, different env name, same DSN)
+	err = (&commands.PullCmd{
+		DSN:         appDSN,
+		OutputDir:   outputDir,
+		Environment: "production",
+		Type:        "mysql",
+	}).Run(&commands.Globals{})
+	require.NoError(t, err)
+
+	// Verify both environments exist in local config
+	cfg, err := local.LoadCLIConfig()
+	require.NoError(t, err)
+	localDB := cfg.Local[dbName]
+	require.Contains(t, localDB.Environments, "staging")
+	require.Contains(t, localDB.Environments, "production")
+	assert.Equal(t, appDSN, localDB.Environments["staging"].DSN)
+	assert.Equal(t, appDSN, localDB.Environments["production"].DSN)
+}
+
+// TestExtractDatabaseFromDSN verifies database name extraction from various DSN formats.
+func TestExtractDatabaseFromDSN(t *testing.T) {
+	tests := []struct {
+		dsn      string
+		expected string
+	}{
+		{"root:pass@tcp(localhost:3306)/mydb", "mydb"},
+		{"root:pass@tcp(localhost:3306)/mydb?parseTime=true", "mydb"},
+		{"user@tcp(host)/db_name?charset=utf8", "db_name"},
+		{"root@/testdb", "testdb"},
+		{"root:pass@tcp(localhost:3306)/", ""},
+		{"no-slash-at-all", ""},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.dsn, func(t *testing.T) {
+			assert.Equal(t, tt.expected, commands.ExtractDatabaseFromDSN(tt.dsn))
+		})
+	}
+}
diff --git a/pkg/cmd/commands/apply.go b/pkg/cmd/commands/apply.go
index 9c53813a..04eeb2e5 100644
--- a/pkg/cmd/commands/apply.go
+++ b/pkg/cmd/commands/apply.go
@@ -47,7 +47,7 @@ func (cmd *ApplyCmd) Run(g *Globals) error {
 		return err
 	}
 
-	ep, err := resolveEndpoint(g.Endpoint, g.Profile)
+	ep, err := resolveEndpoint(g.Endpoint, g.Profile, cfg.Database)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/cmd/commands/common.go b/pkg/cmd/commands/common.go
index 324acb3d..031132bd 100644
--- a/pkg/cmd/commands/common.go
+++ b/pkg/cmd/commands/common.go
@@ -3,6 +3,7 @@ package commands
 
 import (
 	"bufio"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -16,6 +17,8 @@ import (
 	"github.com/block/schemabot/pkg/apitypes"
 	"github.com/block/schemabot/pkg/cmd/client"
 	ghclient "github.com/block/schemabot/pkg/github"
+	"github.com/block/schemabot/pkg/local"
+	"github.com/block/schemabot/pkg/secrets"
 )
 
 // Globals holds flags shared by all commands.
@@ -30,8 +33,9 @@ type Globals struct {
 }
 
 // Resolve resolves the API endpoint from the global flags.
-func (g *Globals) Resolve() (string, error) {
-	return resolveEndpoint(g.Endpoint, g.Profile)
+// Pass an optional database name to enable local mode detection.
+func (g *Globals) Resolve(database ...string) (string, error) {
+	return resolveEndpoint(g.Endpoint, g.Profile, database...)
 }
 
 // ControlFlags holds flags for commands that target a specific schema change.
@@ -112,16 +116,56 @@ func LoadCLIConfig(dir string) (*CLIConfig, error) {
 	return &cfg, nil
 }
 
-// resolveEndpoint resolves the API endpoint from explicit flag or profile config.
-func resolveEndpoint(endpoint, profile string) (string, error) {
+// LocalProfile is the reserved profile name that triggers local mode.
+// Use `--profile local` or set `default_profile: local` in config.
+const LocalProfile = "local"
+
+// resolveEndpoint resolves the SchemaBot API endpoint.
+// Local mode activates when:
+//   - --profile local (or default_profile: local)
+//   - No explicit endpoint/profile and the database is in the local config section
+func resolveEndpoint(endpoint, profile string, database ...string) (string, error) {
+	// Resolve which profile is active (explicit flag, env var, or default).
+	// Skip if an explicit endpoint is provided — that always wins.
+	activeProfile := profile
+	if activeProfile == "" && endpoint == "" {
+		if envProfile := os.Getenv("SCHEMABOT_PROFILE"); envProfile != "" {
+			activeProfile = envProfile
+		} else {
+			cfg, err := client.LoadConfig()
+			if err == nil && cfg.DefaultProfile != "" {
+				activeProfile = cfg.DefaultProfile
+			}
+		}
+	}
+
+	// Local mode: explicit --profile local, or database found in local config
+	if activeProfile == LocalProfile || (endpoint == "" && profile == "" && len(database) > 0 && database[0] != "") {
+		db := ""
+		if len(database) > 0 {
+			db = database[0]
+		}
+		localEP, err := tryLocalMode(db)
+		if err != nil {
+			return "", err
+		}
+		if localEP != "" {
+			return localEP, nil
+		}
+		if activeProfile == LocalProfile {
+			return "", fmt.Errorf("local mode: no databases configured — run 'schemabot pull' first")
+		}
+	}
+
 	ep, err := client.ResolveEndpointWithProfile(endpoint, profile)
 	if err != nil {
 		return "", fmt.Errorf("resolve endpoint: %w", err)
 	}
-	if ep == "" {
-		return "", fmt.Errorf("no endpoint configured (run 'schemabot configure' to set up a profile)")
+	if ep != "" {
+		return ep, nil
 	}
-	return ep, nil
+
+	return "", fmt.Errorf("no endpoint configured — run 'schemabot pull' to set up local mode, or 'schemabot configure' for a remote server")
 }
 
 // resolveApplyID resolves an apply ID to database and environment by calling the progress API.
@@ -244,7 +288,7 @@ func autoResolveApplyID(applyID *string, progressResult *apitypes.ProgressRespon
 // common control command flags (endpoint, profile, applyID, database, environment).
 // Returns the resolved endpoint.
 func resolveControlFlags(endpoint, profile string, applyID, database, environment *string) (string, error) {
-	ep, err := resolveEndpoint(endpoint, profile)
+	ep, err := resolveEndpoint(endpoint, profile, *database)
 	if err != nil {
 		return "", err
 	}
@@ -322,3 +366,51 @@ func printWatchInstructions(applyID, database, environment string) {
 		fmt.Printf("To watch and manage: schemabot progress -d %s -e %s\n", database, environment)
 	}
 }
+
+// tryLocalMode checks if the given database (or any database if empty) is
+// configured in the local section of ~/.schemabot/config.yaml. If so, ensures
+// the background server is running and returns the endpoint.
+func tryLocalMode(database string) (string, error) {
+	cfg, err := local.LoadCLIConfig()
+	if err != nil {
+		return "", fmt.Errorf("load local config: %w", err)
+	}
+	if len(cfg.Local) == 0 {
+		return "", nil
+	}
+
+	// If a specific database is given, check it exists in local config.
+	// If no database given, use any local database (for commands like status, locks).
+	var db local.LocalDatabase
+	selectedDB := database
+	if selectedDB != "" {
+		found, ok := cfg.Local[selectedDB]
+		if !ok {
+			return "", nil
+		}
+		db = found
+	} else {
+		for name, found := range cfg.Local {
+			selectedDB = name
+			db = found
+			break
+		}
+	}
+
+	// Resolve secret refs in all environment DSNs
+	resolvedDB := db
+	resolvedDB.Environments = make(map[string]local.LocalEnvironment, len(db.Environments))
+	for envName, env := range db.Environments {
+		if env.DSN != "" {
+			resolved, resolveErr := secrets.Resolve(env.DSN, "")
+			if resolveErr != nil {
+				return "", fmt.Errorf("resolve DSN for %s/%s: %w", selectedDB, envName, resolveErr)
+			}
+			env.DSN = resolved
+		}
+		resolvedDB.Environments[envName] = env
+	}
+
+	ctx := context.Background()
+	return local.EnsureRunning(ctx, selectedDB, resolvedDB)
+}
diff --git a/pkg/cmd/commands/common_test.go b/pkg/cmd/commands/common_test.go
index 674bec47..4805c54e 100644
--- a/pkg/cmd/commands/common_test.go
+++ b/pkg/cmd/commands/common_test.go
@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/block/schemabot/pkg/cmd/client"
 	"github.com/block/schemabot/pkg/e2eutil"
 )
 
@@ -56,3 +57,102 @@ func TestLoadCLIConfig_WithDeployment(t *testing.T) {
 	assert.Equal(t, "mydb", cfg.Database)
 	assert.Equal(t, "us-west", cfg.Deployment)
 }
+
+func TestResolveEndpoint_ExplicitEndpoint(t *testing.T) {
+	ep, err := resolveEndpoint("http://myserver:8080", "", "mydb")
+	require.NoError(t, err)
+	assert.Equal(t, "http://myserver:8080", ep)
+}
+
+func TestResolveEndpoint_ExplicitProfile(t *testing.T) {
+	// Set up a temp config with a profile
+	tmpDir := t.TempDir()
+	t.Setenv("HOME", tmpDir)
+	configDir := filepath.Join(tmpDir, ".schemabot")
+	require.NoError(t, os.MkdirAll(configDir, 0700))
+
+	cfg := &client.Config{
+		Profiles: map[string]client.Profile{
+			"staging": {Endpoint: "http://staging:8080"},
+		},
+	}
+	require.NoError(t, client.SaveConfig(cfg))
+
+	ep, err := resolveEndpoint("", "staging", "mydb")
+	require.NoError(t, err)
+	assert.Equal(t, "http://staging:8080", ep)
+}
+
+func TestResolveEndpoint_DefaultProfileLocal(t *testing.T) {
+	// Set up config with default_profile: local and a local database
+	tmpDir := t.TempDir()
+	t.Setenv("HOME", tmpDir)
+	configDir := filepath.Join(tmpDir, ".schemabot")
+	require.NoError(t, os.MkdirAll(configDir, 0700))
+
+	// Write config with default_profile: local
+	configContent := "default_profile: local\nprofiles:\n  staging:\n    endpoint: http://staging:8080\n"
+	require.NoError(t, os.WriteFile(filepath.Join(configDir, "config.yaml"), []byte(configContent), 0600))
+
+	// resolveEndpoint with no explicit flags should detect local profile.
+	// No local databases configured, so it errors with a local-mode-specific message.
+	_, err := resolveEndpoint("", "", "testdb")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "local mode")
+}
+
+func TestResolveEndpoint_ProfileLocalExplicit(t *testing.T) {
+	// --profile local should trigger local mode even if default profile is different
+	tmpDir := t.TempDir()
+	t.Setenv("HOME", tmpDir)
+	configDir := filepath.Join(tmpDir, ".schemabot")
+	require.NoError(t, os.MkdirAll(configDir, 0700))
+
+	cfg := &client.Config{
+		DefaultProfile: "staging",
+		Profiles: map[string]client.Profile{
+			"staging": {Endpoint: "http://staging:8080"},
+		},
+	}
+	require.NoError(t, client.SaveConfig(cfg))
+
+	// --profile local, even though default is staging
+	_, err := resolveEndpoint("", "local", "testdb")
+	// Should attempt local mode, not use the staging endpoint
+	assert.NotContains(t, err.Error(), "staging")
+}
+
+func TestResolveEndpoint_EndpointOverridesLocalProfile(t *testing.T) {
+	// Explicit --endpoint should override everything, even if default is local
+	tmpDir := t.TempDir()
+	t.Setenv("HOME", tmpDir)
+	configDir := filepath.Join(tmpDir, ".schemabot")
+	require.NoError(t, os.MkdirAll(configDir, 0700))
+
+	configContent := "default_profile: local\n"
+	require.NoError(t, os.WriteFile(filepath.Join(configDir, "config.yaml"), []byte(configContent), 0600))
+
+	ep, err := resolveEndpoint("http://explicit:9090", "", "mydb")
+	require.NoError(t, err)
+	assert.Equal(t, "http://explicit:9090", ep)
+}
+
+func TestResolveEndpoint_ProfileOverridesDefaultLocal(t *testing.T) {
+	// --profile staging overrides default_profile: local
+	tmpDir := t.TempDir()
+	t.Setenv("HOME", tmpDir)
+	configDir := filepath.Join(tmpDir, ".schemabot")
+	require.NoError(t, os.MkdirAll(configDir, 0700))
+
+	cfg := &client.Config{
+		DefaultProfile: "local",
+		Profiles: map[string]client.Profile{
+			"staging": {Endpoint: "http://staging:8080"},
+		},
+	}
+	require.NoError(t, client.SaveConfig(cfg))
+
+	ep, err := resolveEndpoint("", "staging", "mydb")
+	require.NoError(t, err)
+	assert.Equal(t, "http://staging:8080", ep)
+}
diff --git a/pkg/cmd/commands/local.go b/pkg/cmd/commands/local.go
new file mode 100644
index 00000000..456e1c7c
--- /dev/null
+++ b/pkg/cmd/commands/local.go
@@ -0,0 +1,58 @@
+package commands
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/block/schemabot/pkg/local"
+)
+
+// LocalCmd manages the local background server.
+type LocalCmd struct {
+	Status LocalStatusCmd `cmd:"" default:"withargs" help:"Show local server status."`
+	Stop   LocalStopCmd   `cmd:"" help:"Stop the local server."`
+	Reset  LocalResetCmd  `cmd:"" help:"Stop the server and drop the storage database."`
+}
+
+// LocalStatusCmd shows local server status.
+type LocalStatusCmd struct{}
+
+func (cmd *LocalStatusCmd) Run(g *Globals) error {
+	ctx := context.Background()
+	if local.IsRunning(ctx) {
+		pid := local.ReadPID()
+		fmt.Printf("Local server: running (PID %d, port %s)\n", pid, local.GetServerPort())
+		fmt.Printf("Storage: %s\n", local.RedactedStorageDSN())
+		fmt.Printf("Logs: ~/.schemabot/server.log\n")
+	} else {
+		fmt.Println("Local server: not running")
+		fmt.Println("Run any command (e.g., schemabot plan) to auto-start.")
+	}
+	return nil
+}
+
+// LocalStopCmd stops the local server.
+type LocalStopCmd struct{}
+
+func (cmd *LocalStopCmd) Run(g *Globals) error {
+	if err := local.Stop(); err != nil {
+		return err
+	}
+	fmt.Println("Local server stopped.")
+	return nil
+}
+
+// LocalResetCmd stops the server and drops the storage database.
+type LocalResetCmd struct{}
+
+func (cmd *LocalResetCmd) Run(g *Globals) error {
+	// Stop server first
+	_ = local.Stop()
+
+	// Drop the storage database
+	if err := local.DropStorage(); err != nil {
+		return fmt.Errorf("drop storage: %w", err)
+	}
+	fmt.Printf("Local server stopped and %s database dropped.\n", local.StorageDatabase)
+	return nil
+}
diff --git a/pkg/cmd/commands/plan.go b/pkg/cmd/commands/plan.go
index 82032241..464d0956 100644
--- a/pkg/cmd/commands/plan.go
+++ b/pkg/cmd/commands/plan.go
@@ -33,19 +33,12 @@ func (cmd *PlanCmd) Run(g *Globals) error {
 		return err
 	}
 
-	ep, err := client.ResolveEndpointWithProfile(g.Endpoint, g.Profile)
+	ep, err := resolveEndpoint(g.Endpoint, g.Profile, cfg.Database)
 	if err != nil {
 		if cmd.JSON {
 			return client.ExitWithJSON("config_error", err.Error())
 		}
-		return fmt.Errorf("resolve endpoint: %w", err)
-	}
-	if ep == "" {
-		errMsg := "no endpoint configured (run 'schemabot configure' to set up a profile)"
-		if cmd.JSON {
-			return client.ExitWithJSON("invalid_request", errMsg)
-		}
-		return fmt.Errorf("%s", errMsg)
+		return err
 	}
 
 	// If environment is not specified, get all environments and plan for each
diff --git a/pkg/cmd/commands/pull.go b/pkg/cmd/commands/pull.go
new file mode 100644
index 00000000..b986ae35
--- /dev/null
+++ b/pkg/cmd/commands/pull.go
@@ -0,0 +1,151 @@
+package commands
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"time"
+
+	_ "github.com/go-sql-driver/mysql"
+
+	"github.com/block/schemabot/pkg/local"
+	"github.com/block/schemabot/pkg/secrets"
+	"github.com/block/spirit/pkg/table"
+	"github.com/block/spirit/pkg/utils"
+
+	"gopkg.in/yaml.v3"
+)
+
+// PullCmd pulls the current schema from a live MySQL database and writes
+// one .sql file per table plus a schemabot.yaml config file.
+type PullCmd struct {
+	DSN         string `required:"" help:"MySQL DSN (e.g., root:pass@tcp(localhost:3306)/mydb). Supports env: and file: prefixes."`
+	OutputDir   string `short:"o" help:"Output directory for schema files." default:"."`
+	Database    string `short:"d" help:"Override database name (default: extracted from DSN)."`
+	Environment string `short:"e" help:"Environment name for this connection." default:"staging"`
+	Type        string `short:"t" help:"Database type." default:"mysql" enum:"mysql"`
+}
+
+// Run executes the pull command.
+func (cmd *PullCmd) Run(g *Globals) error {
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+
+	dsn, err := secrets.Resolve(cmd.DSN, "")
+	if err != nil {
+		return fmt.Errorf("resolve DSN: %w", err)
+	}
+	if dsn == "" {
+		return fmt.Errorf("DSN is empty after resolution")
+	}
+
+	dbName := cmd.Database
+	if dbName == "" {
+		dbName = ExtractDatabaseFromDSN(dsn)
+		if dbName == "" {
+			return fmt.Errorf("could not extract database name from DSN — use --database to specify it")
+		}
+	}
+
+	db, err := sql.Open("mysql", dsn)
+	if err != nil {
+		return fmt.Errorf("open database: %w", err)
+	}
+	defer utils.CloseAndLog(db)
+
+	if err := db.PingContext(ctx); err != nil {
+		return fmt.Errorf("connect to %s: %w", dbName, err)
+	}
+
+	tables, err := table.LoadSchemaFromDB(ctx, db,
+		table.WithoutUnderscoreTables,
+		table.WithStrippedAutoIncrement,
+	)
+	if err != nil {
+		return fmt.Errorf("load schema from %s: %w", dbName, err)
+	}
+
+	// Create output directory
+	if err := os.MkdirAll(cmd.OutputDir, 0755); err != nil {
+		return fmt.Errorf("create output directory %s: %w", cmd.OutputDir, err)
+	}
+
+	// Write .sql files
+	var written []string
+	for _, t := range tables {
+		filename := t.Name + ".sql"
+		path := filepath.Join(cmd.OutputDir, filename)
+
+		// Write the raw SHOW CREATE TABLE output — this is already canonical MySQL format.
+		// AUTO_INCREMENT is stripped by the WithStrippedAutoIncrement filter above.
+		content := strings.TrimRight(t.Schema, "\n") + ";\n"
+		if err := os.WriteFile(path, []byte(content), 0644); err != nil {
+			return fmt.Errorf("write %s: %w", path, err)
+		}
+		written = append(written, filename)
+	}
+
+	sort.Strings(written)
+
+	// Generate schemabot.yaml (database + type only, no connection details)
+	configPath := filepath.Join(cmd.OutputDir, "schemabot.yaml")
+	if _, err := os.Stat(configPath); os.IsNotExist(err) {
+		cfg := CLIConfig{
+			Database: dbName,
+			Type:     cmd.Type,
+		}
+		data, err := yaml.Marshal(cfg)
+		if err != nil {
+			return fmt.Errorf("marshal schemabot.yaml: %w", err)
+		}
+		if err := os.WriteFile(configPath, data, 0644); err != nil {
+			return fmt.Errorf("write schemabot.yaml: %w", err)
+		}
+		fmt.Printf("Created %s\n", configPath)
+	} else {
+		slog.Debug("schemabot.yaml already exists, skipping generation", "path", configPath)
+	}
+
+	// Upsert database environment into ~/.schemabot/config.yaml local section
+	if err := local.UpsertLocalEnvironment(dbName, cmd.Type, cmd.Environment, local.LocalEnvironment{
+		DSN: cmd.DSN,
+	}); err != nil {
+		slog.Warn("failed to update local config", "error", err)
+	} else {
+		fmt.Printf("Updated ~/.schemabot/config.yaml: %s/%s\n", dbName, cmd.Environment)
+	}
+
+	absDir, _ := filepath.Abs(cmd.OutputDir)
+	fmt.Printf("Pulled %d tables from %s to %s\n", len(written), dbName, absDir)
+	for _, f := range written {
+		fmt.Printf("  %s\n", f)
+	}
+
+	fmt.Printf("\nNext steps:\n")
+	fmt.Printf("  Edit a .sql file in %s\n", cmd.OutputDir)
+	fmt.Printf("  schemabot plan -s %s  # see the diff\n", cmd.OutputDir)
+	fmt.Printf("  schemabot apply -s %s # apply changes\n", cmd.OutputDir)
+
+	return nil
+}
+
+// ExtractDatabaseFromDSN extracts the database name from a MySQL DSN.
+// DSN format: [user[:password]@][net[(addr)]]/dbname[?param=value]
+func ExtractDatabaseFromDSN(dsn string) string {
+	// Find the last '/' that separates address from database
+	idx := strings.LastIndex(dsn, "/")
+	if idx < 0 {
+		return ""
+	}
+	rest := dsn[idx+1:]
+	// Strip query parameters
+	if qIdx := strings.Index(rest, "?"); qIdx >= 0 {
+		rest = rest[:qIdx]
+	}
+	return rest
+}
diff --git a/pkg/cmd/commands/status.go b/pkg/cmd/commands/status.go
index b6457b05..84245486 100644
--- a/pkg/cmd/commands/status.go
+++ b/pkg/cmd/commands/status.go
@@ -18,7 +18,7 @@ type StatusCmd struct {
 
 // Run executes the status command.
 func (cmd *StatusCmd) Run(g *Globals) error {
-	ep, err := resolveEndpoint(g.Endpoint, g.Profile)
+	ep, err := resolveEndpoint(g.Endpoint, g.Profile, cmd.Database)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/cmd/main.go b/pkg/cmd/main.go
index 41455110..aea84dab 100644
--- a/pkg/cmd/main.go
+++ b/pkg/cmd/main.go
@@ -38,6 +38,8 @@ type CLI struct {
 	Status    commands.StatusCmd    `cmd:"" help:"Show schema change status"`
 	Preview   commands.PreviewCmd   `cmd:"" help:"Preview CLI output templates (for development)"`
 	FixLint   commands.FixLintCmd   `cmd:"" name:"fix-lint" help:"Auto-fix lint issues in schema files"`
+	Pull      commands.PullCmd      `cmd:"" help:"Pull schema from a live database into a declarative schema directory"`
+	Local     commands.LocalCmd     `cmd:"" help:"Manage the local background server"`
 	Configure commands.ConfigureCmd `cmd:"" help:"Configure CLI settings (endpoint, profiles)"`
 	Settings  commands.SettingsCmd  `cmd:"" help:"View or update schema change settings"`
 	Serve     commands.ServeCmd     `cmd:"" help:"Start the SchemaBot HTTP API server"`
diff --git a/pkg/local/config.go b/pkg/local/config.go
new file mode 100644
index 00000000..db73426d
--- /dev/null
+++ b/pkg/local/config.go
@@ -0,0 +1,117 @@
+package local
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"gopkg.in/yaml.v3"
+)
+
+// LocalDatabase holds config for a locally-managed database.
+type LocalDatabase struct {
+	Type         string                      `yaml:"type"`
+	Environments map[string]LocalEnvironment `yaml:"environments"`
+}
+
+// LocalEnvironment holds connection details for one environment of a database.
+type LocalEnvironment struct {
+	DSN          string `yaml:"dsn,omitempty"`
+	Organization string `yaml:"organization,omitempty"`
+	Token        string `yaml:"token,omitempty"`
+}
+
+// CLIConfig is the structure of ~/.schemabot/config.yaml.
+// It extends the existing profile-based config with a top-level local section.
+type CLIConfig struct {
+	DefaultProfile string                   `yaml:"default_profile,omitempty"`
+	Profiles       map[string]CLIProfile    `yaml:"profiles,omitempty"`
+	Local          map[string]LocalDatabase `yaml:"local,omitempty"`
+}
+
+// CLIProfile is a named endpoint configuration.
+type CLIProfile struct {
+	Endpoint string `yaml:"endpoint,omitempty"`
+}
+
+// LoadCLIConfig loads ~/.schemabot/config.yaml.
+func LoadCLIConfig() (*CLIConfig, error) {
+	path, err := configPath()
+	if err != nil {
+		return nil, err
+	}
+
+	data, err := os.ReadFile(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return &CLIConfig{}, nil
+		}
+		return nil, fmt.Errorf("read config: %w", err)
+	}
+
+	var cfg CLIConfig
+	if err := yaml.Unmarshal(data, &cfg); err != nil {
+		return nil, fmt.Errorf("parse config: %w", err)
+	}
+	return &cfg, nil
+}
+
+// UpsertLocalEnvironment adds or updates an environment for a database
+// in the local section and saves.
+func UpsertLocalEnvironment(database, dbType, environment string, env LocalEnvironment) error {
+	cfg, err := LoadCLIConfig()
+	if err != nil {
+		return err
+	}
+
+	if cfg.Local == nil {
+		cfg.Local = make(map[string]LocalDatabase)
+	}
+	db, ok := cfg.Local[database]
+	if !ok {
+		db = LocalDatabase{Type: dbType}
+	}
+	db.Type = dbType
+	if db.Environments == nil {
+		db.Environments = make(map[string]LocalEnvironment)
+	}
+	db.Environments[environment] = env
+	cfg.Local[database] = db
+
+	return saveCLIConfig(cfg)
+}
+
+// saveCLIConfig writes the config back to ~/.schemabot/config.yaml.
+func saveCLIConfig(cfg *CLIConfig) error {
+	path, err := configPath()
+	if err != nil {
+		return err
+	}
+
+	if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
+		return fmt.Errorf("create config directory: %w", err)
+	}
+
+	data, err := yaml.Marshal(cfg)
+	if err != nil {
+		return fmt.Errorf("marshal config: %w", err)
+	}
+
+	return os.WriteFile(path, data, 0600)
+}
+
+func configPath() (string, error) {
+	dir, err := schemabotDir()
+	if err != nil {
+		return "", err
+	}
+	return filepath.Join(dir, "config.yaml"), nil
+}
+
+func schemabotDir() (string, error) {
+	home, err := os.UserHomeDir()
+	if err != nil {
+		return "", fmt.Errorf("get home directory: %w", err)
+	}
+	return filepath.Join(home, ".schemabot"), nil
+}
diff --git a/pkg/local/config_test.go b/pkg/local/config_test.go
new file mode 100644
index 00000000..da4c938f
--- /dev/null
+++ b/pkg/local/config_test.go
@@ -0,0 +1,91 @@
+package local
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUpsertLocalEnvironment(t *testing.T) {
+	// Use a temp dir for config
+	tmpDir := t.TempDir()
+	t.Setenv("HOME", tmpDir)
+
+	// First upsert creates the file and database entry
+	err := UpsertLocalEnvironment("mydb", "mysql", "staging", LocalEnvironment{
+		DSN: "root@tcp(localhost:3306)/mydb",
+	})
+	require.NoError(t, err)
+
+	cfg, err := LoadCLIConfig()
+	require.NoError(t, err)
+	require.Contains(t, cfg.Local, "mydb")
+	assert.Equal(t, "mysql", cfg.Local["mydb"].Type)
+	require.Contains(t, cfg.Local["mydb"].Environments, "staging")
+	assert.Equal(t, "root@tcp(localhost:3306)/mydb", cfg.Local["mydb"].Environments["staging"].DSN)
+
+	// Second upsert adds another environment without overwriting the first
+	err = UpsertLocalEnvironment("mydb", "mysql", "production", LocalEnvironment{
+		DSN: "root@tcp(prod-host:3306)/mydb",
+	})
+	require.NoError(t, err)
+
+	cfg, err = LoadCLIConfig()
+	require.NoError(t, err)
+	require.Contains(t, cfg.Local["mydb"].Environments, "staging")
+	require.Contains(t, cfg.Local["mydb"].Environments, "production")
+	assert.Equal(t, "root@tcp(localhost:3306)/mydb", cfg.Local["mydb"].Environments["staging"].DSN)
+	assert.Equal(t, "root@tcp(prod-host:3306)/mydb", cfg.Local["mydb"].Environments["production"].DSN)
+
+	// Third upsert adds a different database
+	err = UpsertLocalEnvironment("otherdb", "vitess", "production", LocalEnvironment{
+		Organization: "myorg",
+		Token:        "env:PS_TOKEN",
+	})
+	require.NoError(t, err)
+
+	cfg, err = LoadCLIConfig()
+	require.NoError(t, err)
+	require.Contains(t, cfg.Local, "mydb")
+	require.Contains(t, cfg.Local, "otherdb")
+	assert.Equal(t, "vitess", cfg.Local["otherdb"].Type)
+	assert.Equal(t, "myorg", cfg.Local["otherdb"].Environments["production"].Organization)
+}
+
+func TestLoadCLIConfig_NoFile(t *testing.T) {
+	tmpDir := t.TempDir()
+	t.Setenv("HOME", tmpDir)
+
+	cfg, err := LoadCLIConfig()
+	require.NoError(t, err)
+	assert.Empty(t, cfg.Local)
+	assert.Empty(t, cfg.Profiles)
+}
+
+func TestLoadCLIConfig_PreservesProfiles(t *testing.T) {
+	tmpDir := t.TempDir()
+	t.Setenv("HOME", tmpDir)
+
+	// Write a config with profiles
+	configDir := filepath.Join(tmpDir, ".schemabot")
+	require.NoError(t, os.MkdirAll(configDir, 0700))
+	configContent := "default_profile: staging\nprofiles:\n  staging:\n    endpoint: http://localhost:8080\n"
+	require.NoError(t, os.WriteFile(filepath.Join(configDir, "config.yaml"), []byte(configContent), 0600))
+
+	// Upsert a local database
+	err := UpsertLocalEnvironment("mydb", "mysql", "local", LocalEnvironment{
+		DSN: "root@tcp(localhost:3306)/mydb",
+	})
+	require.NoError(t, err)
+
+	// Verify profiles are preserved
+	cfg, err := LoadCLIConfig()
+	require.NoError(t, err)
+	assert.Equal(t, "staging", cfg.DefaultProfile)
+	require.Contains(t, cfg.Profiles, "staging")
+	assert.Equal(t, "http://localhost:8080", cfg.Profiles["staging"].Endpoint)
+	require.Contains(t, cfg.Local, "mydb")
+}
diff --git a/pkg/local/server.go b/pkg/local/server.go
new file mode 100644
index 00000000..cdfa464e
--- /dev/null
+++ b/pkg/local/server.go
@@ -0,0 +1,470 @@
+package local
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+
+	"database/sql"
+
+	"github.com/go-sql-driver/mysql"
+
+	"github.com/block/schemabot/pkg/api"
+	"github.com/block/spirit/pkg/utils"
+
+	"gopkg.in/yaml.v3"
+)
+
+const (
+	// StorageDatabase is the database created on the local MySQL server
+	// for SchemaBot's internal state (plans, applies, locks, tasks).
+	StorageDatabase = "_schemabot"
+
+	defaultServerPort = "18080"
+	defaultMySQLAddr  = "127.0.0.1:3306"
+)
+
+// GetServerPort returns the local HTTP server port.
+// Override with SCHEMABOT_LOCAL_PORT for testing or non-standard setups.
+func GetServerPort() string {
+	if p := os.Getenv("SCHEMABOT_LOCAL_PORT"); p != "" {
+		return p
+	}
+	return defaultServerPort
+}
+
+// GetStorageDSN returns the DSN for the local storage database.
+// Override with SCHEMABOT_LOCAL_STORAGE_DSN for non-standard MySQL setups
+// (e.g., different port, password, or auth).
+func GetStorageDSN() string {
+	if dsn := os.Getenv("SCHEMABOT_LOCAL_STORAGE_DSN"); dsn != "" {
+		return dsn
+	}
+	return "root@tcp(" + defaultMySQLAddr + ")/" + StorageDatabase + "?parseTime=true"
+}
+
+// RedactedStorageDSN returns a display-safe version of the storage DSN
+// with the password masked. Used for status output.
+func RedactedStorageDSN() string {
+	dsn := GetStorageDSN()
+	cfg, err := mysql.ParseDSN(dsn)
+	if err != nil {
+		return "(configured via SCHEMABOT_LOCAL_STORAGE_DSN)"
+	}
+	if cfg.Passwd != "" {
+		cfg.Passwd = "***"
+	}
+	return cfg.FormatDSN()
+}
+
+// getServerDSN returns the DSN for bootstrapping (no database selected).
+func getServerDSN() string {
+	if dsn := os.Getenv("SCHEMABOT_LOCAL_STORAGE_DSN"); dsn != "" {
+		// Strip database name from the override DSN for bootstrap
+		return stripDatabaseFromDSN(dsn)
+	}
+	return "root@tcp(" + defaultMySQLAddr + ")/"
+}
+
+// stripDatabaseFromDSN removes the database name from a DSN, keeping auth and params.
+// Uses mysql.ParseDSN to handle all DSN formats correctly (including unix socket paths).
+func stripDatabaseFromDSN(dsn string) string {
+	cfg, err := mysql.ParseDSN(dsn)
+	if err != nil {
+		slog.Warn("failed to parse DSN, returning as-is", "error", err)
+		return dsn
+	}
+	cfg.DBName = ""
+	return cfg.FormatDSN()
+}
+
+// EnsureRunning checks if the local server is running and starts it if not.
+// Returns the endpoint URL.
+//
+// Server lifecycle:
+//   - Auto-start: first CLI command that needs an endpoint forks `schemabot serve`
+//     as a background daemon (detached, PID file, log file).
+//   - Reuse: subsequent commands check the health endpoint and reuse the running server.
+//   - Stale recovery: if the port is bound but health check fails, the stale
+//     process is killed before starting a new one.
+//   - Explicit stop: `schemabot local stop` sends SIGTERM and waits for shutdown.
+//   - Binary upgrade: stop the old server, next command auto-starts with new binary.
+//   - Machine reboot: server dies, next command auto-starts fresh.
+//
+// Storage lives on the developer's local MySQL (_schemabot database),
+// fully decoupled from the target database.
+func EnsureRunning(ctx context.Context, database string, dbConfig LocalDatabase) (string, error) {
+	port := GetServerPort()
+	endpoint := "http://127.0.0.1:" + port
+
+	if isHealthy(ctx, endpoint) {
+		slog.Debug("local server already running")
+		return endpoint, nil
+	}
+
+	// Port bound but not healthy — kill the stale process
+	if findProcessOnPort(port) != 0 {
+		killStaleServer()
+	}
+
+	// Verify local MySQL is reachable before attempting anything.
+	if err := checkLocalMySQL(ctx); err != nil {
+		return "", err
+	}
+
+	// Bootstrap _schemabot database on the local MySQL server.
+	if err := bootstrapStorage(ctx); err != nil {
+		return "", err
+	}
+
+	// Generate server config file
+	configPath, err := writeLocalServerConfig(database, dbConfig)
+	if err != nil {
+		return "", fmt.Errorf("generate server config: %w", err)
+	}
+
+	// Fork schemabot serve as a background process
+	if err := startDaemon(configPath); err != nil {
+		return "", fmt.Errorf("start local server: %w", err)
+	}
+
+	// Wait for server to be healthy
+	if err := waitForHealthy(ctx, endpoint, 15*time.Second); err != nil {
+		return "", fmt.Errorf("local server not ready: %w", err)
+	}
+
+	slog.Debug("local server started", "endpoint", endpoint, "pid", readPID())
+	return endpoint, nil
+}
+
+// Stop stops the local background server. It tries the PID file first,
+// then falls back to finding the process by port if the PID file is missing.
+func Stop() error {
+	pid := readPID()
+
+	// If no PID file, try to find the process by port
+	if pid == 0 {
+		pid = findProcessOnPort(GetServerPort())
+		if pid == 0 {
+			return fmt.Errorf("no local server running")
+		}
+		slog.Debug("found orphan server process by port", "pid", pid)
+	}
+
+	process, err := os.FindProcess(pid)
+	if err != nil {
+		removePIDFile()
+		return fmt.Errorf("find process %d: %w", pid, err)
+	}
+
+	if err := process.Signal(syscall.SIGTERM); err != nil {
+		removePIDFile()
+		return fmt.Errorf("stop process %d: %w", pid, err)
+	}
+
+	// Wait for graceful shutdown
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if !isProcessRunning(pid) {
+			break
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+
+	removePIDFile()
+	return nil
+}
+
+// IsRunning returns true if the local server is running.
+func IsRunning(ctx context.Context) bool {
+	return isHealthy(ctx, "http://127.0.0.1:"+GetServerPort())
+}
+
+// killStaleServer kills any process holding the server port that isn't
+// responding to health checks. Called before starting a new server.
+func killStaleServer() {
+	pid := findProcessOnPort(GetServerPort())
+	if pid == 0 {
+		return
+	}
+	slog.Debug("killing stale server process", "pid", pid)
+	if process, err := os.FindProcess(pid); err == nil {
+		if err := process.Signal(syscall.SIGTERM); err != nil {
+			slog.Debug("failed to kill stale process", "pid", pid, "error", err)
+		}
+	}
+	time.Sleep(500 * time.Millisecond)
+	removePIDFile()
+}
+
+// findProcessOnPort returns the PID of the process listening on the given port,
+// or 0 if no process is found. Uses lsof (macOS/Linux only).
+// Restricts to LISTEN state to avoid matching client connections.
+func findProcessOnPort(port string) int {
+	out, err := exec.CommandContext(context.Background(), "lsof", "-ti", "tcp:"+port, "-sTCP:LISTEN").Output()
+	if err != nil {
+		return 0
+	}
+	// lsof may return multiple lines; take only the first PID.
+	line := strings.SplitN(strings.TrimSpace(string(out)), "\n", 2)[0]
+	if line == "" {
+		return 0
+	}
+	pid, err := strconv.Atoi(line)
+	if err != nil {
+		slog.Debug("failed to parse PID from lsof output", "output", line, "error", err)
+		return 0
+	}
+	return pid
+}
+
+// isProcessRunning checks if a process with the given PID exists.
+func isProcessRunning(pid int) bool {
+	process, err := os.FindProcess(pid)
+	if err != nil {
+		return false
+	}
+	err = process.Signal(syscall.Signal(0))
+	return err == nil
+}
+
+// checkLocalMySQL verifies that MySQL is running on the configured address
+// and accessible with the storage DSN.
+func checkLocalMySQL(ctx context.Context) error {
+	serverDSN := getServerDSN()
+	addr := defaultMySQLAddr
+	// Parse the DSN to get the actual address for error messages.
+	if cfg, err := mysql.ParseDSN(serverDSN); err == nil && cfg.Addr != "" {
+		addr = cfg.Addr
+	}
+
+	db, err := sql.Open("mysql", serverDSN)
+	if err != nil {
+		return fmt.Errorf("local mode requires MySQL on %s — is MySQL running?\n\n  Install: https://dev.mysql.com/downloads/ or `brew install mysql`\n  Start:   mysql.server start", addr)
+	}
+	defer utils.CloseAndLog(db)
+	if err := db.PingContext(ctx); err != nil {
+		if strings.Contains(err.Error(), "Access denied") {
+			return fmt.Errorf("local mode cannot connect to MySQL at %s: %w\n\nCheck your MySQL user configuration or override with SCHEMABOT_LOCAL_STORAGE_DSN", addr, err)
+		}
+		return fmt.Errorf("local mode requires MySQL on %s — is MySQL running?\n\n  Install: https://dev.mysql.com/downloads/ or `brew install mysql`\n  Start:   mysql.server start", addr)
+	}
+	return nil
+}
+
+// bootstrapStorage creates the _schemabot database on the local MySQL server
+// and runs EnsureSchema to bootstrap the storage tables.
+func bootstrapStorage(ctx context.Context) error {
+	db, err := sql.Open("mysql", getServerDSN())
+	if err != nil {
+		return fmt.Errorf("open MySQL server: %w", err)
+	}
+	defer utils.CloseAndLog(db)
+
+	if err := db.PingContext(ctx); err != nil {
+		return fmt.Errorf("connect to MySQL server: %w", err)
+	}
+
+	if _, err := db.ExecContext(ctx, "CREATE DATABASE IF NOT EXISTS `"+StorageDatabase+"`"); err != nil {
+		return fmt.Errorf("create storage database: %w", err)
+	}
+
+	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelWarn}))
+	if err := api.EnsureSchema(GetStorageDSN(), logger); err != nil {
+		return fmt.Errorf("bootstrap storage schema: %w", err)
+	}
+
+	return nil
+}
+
+// startDaemon forks the current binary as `schemabot serve` in the background.
+func startDaemon(configPath string) error {
+	dir, err := schemabotDir()
+	if err != nil {
+		return err
+	}
+	if err := os.MkdirAll(dir, 0700); err != nil {
+		return fmt.Errorf("create config directory: %w", err)
+	}
+
+	logPath := filepath.Join(dir, "server.log")
+	logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600)
+	if err != nil {
+		return fmt.Errorf("open log file: %w", err)
+	}
+
+	// Use context.Background because the child process must outlive the parent CLI command.
+	cmd := exec.CommandContext(context.Background(), os.Args[0], "serve")
+	cmd.Env = append(os.Environ(),
+		"SCHEMABOT_CONFIG_FILE="+configPath,
+		"PORT="+GetServerPort(),
+		"LOG_LEVEL=warn",
+	)
+	cmd.Stdout = logFile
+	cmd.Stderr = logFile
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
+
+	if err := cmd.Start(); err != nil {
+		utils.CloseAndLog(logFile)
+		return fmt.Errorf("start server process: %w", err)
+	}
+
+	// Write PID file
+	pidPath := filepath.Join(dir, "server.pid")
+	if err := os.WriteFile(pidPath, []byte(strconv.Itoa(cmd.Process.Pid)), 0600); err != nil {
+		slog.Warn("failed to write PID file", "error", err)
+	}
+
+	// Detach — don't wait for the child process
+	if err := cmd.Process.Release(); err != nil {
+		slog.Debug("release process", "error", err)
+	}
+	utils.CloseAndLog(logFile)
+
+	return nil
+}
+
+// writeLocalServerConfig generates a server config YAML and writes it to
+// ~/.schemabot/local-server.yaml.
+func writeLocalServerConfig(database string, dbConfig LocalDatabase) (string, error) {
+	envConfigs := make(map[string]serverEnvConfig, len(dbConfig.Environments))
+	for envName, env := range dbConfig.Environments {
+		envConfigs[envName] = serverEnvConfig{
+			DSN:            env.DSN,
+			Organization:   env.Organization,
+			TokenSecretRef: env.Token,
+		}
+	}
+
+	cfg := serverConfig{
+		Storage: serverStorageConfig{
+			DSN: GetStorageDSN(),
+		},
+		Databases: map[string]serverDatabaseConfig{
+			database: {
+				Type:         dbConfig.Type,
+				Environments: envConfigs,
+			},
+		},
+	}
+
+	data, err := yaml.Marshal(cfg)
+	if err != nil {
+		return "", fmt.Errorf("marshal config: %w", err)
+	}
+
+	dir, err := schemabotDir()
+	if err != nil {
+		return "", err
+	}
+	path := filepath.Join(dir, "local-server.yaml")
+	if err := os.WriteFile(path, data, 0600); err != nil {
+		return "", fmt.Errorf("write config: %w", err)
+	}
+	return path, nil
+}
+
+// Server config types for YAML generation.
+type serverConfig struct {
+	Storage   serverStorageConfig             `yaml:"storage"`
+	Databases map[string]serverDatabaseConfig `yaml:"databases"`
+}
+
+type serverStorageConfig struct {
+	DSN string `yaml:"dsn"`
+}
+
+type serverDatabaseConfig struct {
+	Type         string                     `yaml:"type"`
+	Environments map[string]serverEnvConfig `yaml:"environments"`
+}
+
+type serverEnvConfig struct {
+	DSN            string `yaml:"dsn,omitempty"`
+	Organization   string `yaml:"organization,omitempty"`
+	TokenSecretRef string `yaml:"token_secret_ref,omitempty"`
+}
+
+// Health check
+
+func isHealthy(ctx context.Context, endpoint string) bool {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint+"/health", nil)
+	if err != nil {
+		return false
+	}
+	client := &http.Client{Timeout: 1 * time.Second}
+	resp, err := client.Do(req)
+	if err != nil {
+		return false
+	}
+	resp.Body.Close()
+	return resp.StatusCode == http.StatusOK
+}
+
+func waitForHealthy(ctx context.Context, endpoint string, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		if isHealthy(ctx, endpoint) {
+			return nil
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	return fmt.Errorf("timeout waiting for %s", endpoint)
+}
+
+// DropStorage drops the _schemabot database from the local MySQL server.
+func DropStorage() error {
+	db, err := sql.Open("mysql", getServerDSN())
+	if err != nil {
+		return fmt.Errorf("connect to local MySQL: %w", err)
+	}
+	defer utils.CloseAndLog(db)
+	if _, err := db.ExecContext(context.Background(), "DROP DATABASE IF EXISTS `"+StorageDatabase+"`"); err != nil {
+		return fmt.Errorf("drop database: %w", err)
+	}
+	return nil
+}
+
+// ReadPID returns the PID from the PID file, or 0 if not found.
+func ReadPID() int {
+	return readPID()
+}
+
+// PID file management
+
+func readPID() int {
+	dir, err := schemabotDir()
+	if err != nil {
+		return 0
+	}
+	data, err := os.ReadFile(filepath.Join(dir, "server.pid"))
+	if err != nil {
+		return 0
+	}
+	pid, err := strconv.Atoi(strings.TrimSpace(string(data)))
+	if err != nil {
+		return 0
+	}
+	return pid
+}
+
+func removePIDFile() {
+	dir, err := schemabotDir()
+	if err != nil {
+		return
+	}
+	if err := os.Remove(filepath.Join(dir, "server.pid")); err != nil && !os.IsNotExist(err) {
+		slog.Debug("remove PID file", "error", err)
+	}
+}