feat(cli): conduit run --dev hot-reload watcher (PR2 of §4)

cmd/conduit/root/pipelines/apply.go

@@ -111,7 +111,7 @@ func (c *ApplyCommand) Args(args []string) error {
 func (c *ApplyCommand) Config() ecdysis.Config {
 	path := filepath.Dir(c.flags.ConduitCfg.Path)
 	return ecdysis.Config{
-		EnvPrefix:     "CONDUIT",
+		EnvPrefix:     envPrefix,
 		Parsed:        &c.flags.Config,
 		Path:          c.flags.ConduitCfg.Path,
 		DefaultValues: conduit.DefaultConfigWithBasePath(path),

cmd/conduit/root/pipelines/deploy.go

@@ -124,7 +124,7 @@ func (c *DeployCommand) Args(args []string) error {
 func (c *DeployCommand) Config() ecdysis.Config {
 	path := filepath.Dir(c.flags.ConduitCfg.Path)
 	return ecdysis.Config{
-		EnvPrefix:     "CONDUIT",
+		EnvPrefix:     envPrefix,
 		Parsed:        &c.flags.Config,
 		Path:          c.flags.ConduitCfg.Path,
 		DefaultValues: conduit.DefaultConfigWithBasePath(path),

cmd/conduit/root/pipelines/dev.go

@@ -0,0 +1,168 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package pipelines
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/conduitio/conduit/pkg/conduit"
+	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/ecdysis"
+)
+
+var (
+	_ ecdysis.CommandWithFlags   = (*DevCommand)(nil)
+	_ ecdysis.CommandWithExecute = (*DevCommand)(nil)
+	_ ecdysis.CommandWithDocs    = (*DevCommand)(nil)
+	_ ecdysis.CommandWithConfig  = (*DevCommand)(nil)
+	_ ecdysis.CommandWithArgs    = (*DevCommand)(nil)
+)
+
+// DevArgs holds DevCommand's optional positional argument.
+type DevArgs struct {
+	// Dir is a GitOps-friendly alias for --pipelines.path, matching `run`'s
+	// --pipelines flag but as a positional argument (`conduit pipelines dev
+	// [dir]` reads more naturally than `conduit pipelines dev
+	// --pipelines.path dir`).
+	Dir string
+}
+
+// DevFlags holds DevCommand's flags. conduit.Config is embedded for the same
+// reason RunFlags embeds it: `pipelines dev` runs a full `conduit run --dev`
+// server, so it accepts the same db.*/api.*/pipelines.*/... configuration.
+type DevFlags struct {
+	conduit.Config
+}
+
+// DevCommand implements `conduit pipelines dev [dir]`: the design doc's §4
+// alias, "thin sugar over run --dev ... carrying no logic of its own". It
+// does not reimplement any watch/apply behavior (that all lives in
+// pkg/conduit/dev, driven from pkg/conduit.Runtime) — it only sets
+// Cfg.Dev.Enabled, points Cfg.Pipelines.Path at the optional dir argument,
+// applies the "dev defaults" (exit-on-error off, so one bad pipeline while
+// iterating doesn't take the whole dev server down), and calls the exact
+// same conduit.Entrypoint.Serve RunCommand does.
+type DevCommand struct {
+	args  DevArgs
+	flags DevFlags
+	Cfg   conduit.Config
+}
+
+func (c *DevCommand) Usage() string { return "dev [dir]" }
+
+func (c *DevCommand) Docs() ecdysis.Docs {
+	return ecdysis.Docs{
+		Short: "Run Conduit with the hot-reload dev watcher (alias for 'conduit run --dev')",
+		Long: `Starts the Conduit server exactly like 'conduit run --dev', watching pipelines.path (or
+[dir], if given) for changes and applying them into the running engine as you save — a processor
+edit applies in place (no restart), a source/destination edit applies via a labeled graceful
+restart, and a bad edit is reported without ever touching the running pipeline. See
+docs/design-documents/20260712-pipeline-dev-hot-reload.md §4 and docs/operations/dev-hot-reload.md.
+
+This is sugar over 'conduit run --dev [dir]', with exit-on-degraded forced off (dev iterates on
+pipelines that may be intentionally broken mid-edit; one degraded pipeline should not take the
+whole dev server down) — it carries no watcher logic of its own.`,
+		Example: "conduit pipelines dev\n" +
+			"conduit pipelines dev ./pipelines\n" +
+			"conduit pipelines dev --dev.json",
+	}
+}
+
+func (c *DevCommand) Args(args []string) error {
+	if len(args) > 1 {
+		return cerrors.Errorf("too many arguments")
+	}
+	if len(args) == 1 {
+		c.args.Dir = args[0]
+	}
+	return nil
+}
+
+func (c *DevCommand) Execute(ctx context.Context) error {
+	if c.args.Dir != "" {
+		c.Cfg.Pipelines.Path = c.args.Dir
+	}
+	// `pipelines dev` always runs in dev mode — that is the entire point of
+	// the alias — and defaults exit-on-degraded off (see the type doc).
+	c.Cfg.Dev.Enabled = true
+	c.Cfg.Pipelines.ExitOnDegraded = false
+
+	if !c.Cfg.API.Enabled {
+		fmt.Print("Warning: API is currently disabled. Most Conduit CLI commands won't work without the API enabled." +
+			"To enable it, run conduit with `--api.enabled=true` or set `CONDUIT_API_ENABLED=true` in your environment.")
+	}
+
+	(&conduit.Entrypoint{}).Serve(c.Cfg)
+	return nil
+}
+
+func (c *DevCommand) Config() ecdysis.Config {
+	path := filepath.Dir(c.flags.ConduitCfg.Path)
+	return ecdysis.Config{
+		EnvPrefix:     envPrefix,
+		Parsed:        &c.Cfg,
+		Path:          c.flags.ConduitCfg.Path,
+		DefaultValues: conduit.DefaultConfigWithBasePath(path),
+	}
+}
+
+// Flags mirrors run.RunCommand.Flags's defaulting (storeConfigDefaults
+// covers only the offline deploy/apply subset; `dev` runs a full server, so
+// it needs the same defaults `run` sets, including the dev.* group so
+// --dev.json works without also passing --dev).
+func (c *DevCommand) Flags() []ecdysis.Flag {
+	flags := ecdysis.BuildFlags(&c.flags)
+
+	currentPath, err := os.Getwd()
+	if err != nil {
+		panic(cerrors.Errorf("failed to get current working directory: %w", err))
+	}
+
+	c.Cfg = conduit.DefaultConfigWithBasePath(currentPath)
+	flags.SetDefault("config.path", c.Cfg.ConduitCfg.Path)
+	flags.SetDefault("db.type", c.Cfg.DB.Type)
+	flags.SetDefault("db.badger.path", c.Cfg.DB.Badger.Path)
+	flags.SetDefault("db.postgres.connection-string", c.Cfg.DB.Postgres.ConnectionString)
+	flags.SetDefault("db.postgres.table", c.Cfg.DB.Postgres.Table)
+	flags.SetDefault("db.sqlite.path", c.Cfg.DB.SQLite.Path)
+	flags.SetDefault("db.sqlite.table", c.Cfg.DB.SQLite.Table)
+	flags.SetDefault("api.enabled", c.Cfg.API.Enabled)
+	flags.SetDefault("api.http.address", c.Cfg.API.HTTP.Address)
+	flags.SetDefault("api.grpc.address", c.Cfg.API.GRPC.Address)
+	flags.SetDefault("api.allow-live-restart-apply", c.Cfg.API.AllowLiveRestartApply)
+	flags.SetDefault("log.level", c.Cfg.Log.Level)
+	flags.SetDefault("log.format", c.Cfg.Log.Format)
+	flags.SetDefault("connectors.path", c.Cfg.Connectors.Path)
+	flags.SetDefault("connectors.max-receive-record-size", c.Cfg.Connectors.MaxReceiveRecordSize)
+	flags.SetDefault("processors.path", c.Cfg.Processors.Path)
+	flags.SetDefault("pipelines.path", c.Cfg.Pipelines.Path)
+	flags.SetDefault("pipelines.error-recovery.min-delay", c.Cfg.Pipelines.ErrorRecovery.MinDelay)
+	flags.SetDefault("pipelines.error-recovery.max-delay", c.Cfg.Pipelines.ErrorRecovery.MaxDelay)
+	flags.SetDefault("pipelines.error-recovery.backoff-factor", c.Cfg.Pipelines.ErrorRecovery.BackoffFactor)
+	flags.SetDefault("pipelines.error-recovery.max-retries", c.Cfg.Pipelines.ErrorRecovery.MaxRetries)
+	flags.SetDefault("pipelines.error-recovery.max-retries-window", c.Cfg.Pipelines.ErrorRecovery.MaxRetriesWindow)
+	flags.SetDefault("schema-registry.type", c.Cfg.SchemaRegistry.Type)
+	flags.SetDefault("schema-registry.confluent.connection-string", c.Cfg.SchemaRegistry.Confluent.ConnectionString)
+	flags.SetDefault("preview.pipeline-arch-v2", c.Cfg.Preview.PipelineArchV2)
+	flags.SetDefault("preview.pipeline-arch-v2-disable-metrics", c.Cfg.Preview.PipelineArchV2DisableMetrics)
+	// pipelines.exit-on-degraded is deliberately NOT defaulted from
+	// DefaultConfigWithBasePath here: Execute always forces it false for
+	// `pipelines dev` regardless of flags/env/config file (see the type doc).
+
+	return flags
+}

cmd/conduit/root/pipelines/pipelines.go

@@ -18,6 +18,11 @@ import (
 	"github.com/conduitio/ecdysis"
 )
 
+// envPrefix is the environment variable prefix every subcommand's
+// ecdysis.Config uses (CONDUIT_DB_TYPE, CONDUIT_PIPELINES_PATH, ...) — a
+// shared constant so deploy/apply/dev can't drift on it independently.
+const envPrefix = "CONDUIT"
+
 var (
 	_ ecdysis.CommandWithDocs        = (*PipelinesCommand)(nil)
 	_ ecdysis.CommandWithSubCommands = (*PipelinesCommand)(nil)
@@ -42,6 +47,7 @@ func (c *PipelinesCommand) SubCommands() []ecdysis.Command {
 		&RepairCommand{},
 		&StartCommand{},
 		&StopCommand{},
+		&DevCommand{},
 	}
 }
 

cmd/conduit/root/run/run.go

@@ -37,6 +37,14 @@ var (
 // applied in Execute.
 const pipelinesFlagName = "pipelines"
 
+// devFlagName is the short --dev alias for --dev.enabled (see
+// docs/design-documents/20260712-pipeline-dev-hot-reload.md §4: "conduit run
+// --dev [--pipelines.path <dir>]"). Like --pipelines/--pipelines.path, it
+// can't be bound directly onto conduit.Config (the name "dev" is already a
+// struct — conduit.Config.Dev — not a bool), so it is excluded from the
+// viper config binding and applied in Execute.
+const devFlagName = "dev"
+
 type RunFlags struct {
 	conduit.Config
 }
@@ -49,13 +57,22 @@ type RunCommand struct {
 	// collides with the Pipelines struct key), so it's excluded from the config
 	// binding (see Config) and applied in Execute.
 	pipelinesAlias string
+	// devAlias backs the --dev flag — see devFlagName's doc.
+	devAlias bool
 }
 
 func (c *RunCommand) Execute(ctx context.Context) error {
+	cmd := ecdysis.CobraCmdFromContext(ctx)
+
 	// --pipelines is an alias for --pipelines.path; apply it when explicitly set.
-	if cmd := ecdysis.CobraCmdFromContext(ctx); cmd != nil && cmd.Flags().Changed(pipelinesFlagName) {
+	if cmd != nil && cmd.Flags().Changed(pipelinesFlagName) {
 		c.Cfg.Pipelines.Path = c.pipelinesAlias
 	}
+	// --dev is an alias for --dev.enabled; apply it when explicitly set (never
+	// clobber an explicit --dev.enabled=false with a stray --dev.enabled default).
+	if cmd != nil && cmd.Flags().Changed(devFlagName) {
+		c.Cfg.Dev.Enabled = c.devAlias
+	}
 
 	e := &conduit.Entrypoint{}
 
@@ -75,9 +92,10 @@ func (c *RunCommand) Config() ecdysis.Config {
 		EnvPrefix: "CONDUIT",
 		Parsed:    &c.Cfg,
 		Path:      c.flags.ConduitCfg.Path,
-		// --pipelines collides with the Pipelines config struct key; exclude it from
-		// the viper binding and apply it in Execute (see pipelinesAlias).
-		ExcludedFlags: []string{pipelinesFlagName},
+		// --pipelines and --dev collide with existing conduit.Config struct
+		// keys (Pipelines, Dev); exclude them from the viper binding and
+		// apply them in Execute (see pipelinesAlias/devAlias).
+		ExcludedFlags: []string{pipelinesFlagName, devFlagName},
 		DefaultValues: conduit.DefaultConfigWithBasePath(path),
 	}
 }
@@ -131,12 +149,34 @@ func (c *RunCommand) Flags() []ecdysis.Flag {
 		Ptr:     &c.pipelinesAlias,
 		Default: c.Cfg.Pipelines.Path,
 	})
+	// --dev is the short alias for --dev.enabled: watch --pipelines.path and
+	// hot-reload changes into the running engine (see
+	// docs/design-documents/20260712-pipeline-dev-hot-reload.md §4). Excluded
+	// from the viper config binding (see Config) because its name collides
+	// with the Dev struct; Execute copies it into Cfg.Dev.Enabled when set.
+	flags = append(flags, ecdysis.Flag{
+		Long:    devFlagName,
+		Usage:   "alias for --dev.enabled: watch pipelines.path and hot-reload changes into the running engine",
+		Ptr:     &c.devAlias,
+		Default: c.Cfg.Dev.Enabled,
+	})
 	return flags
 }
 
 func (c *RunCommand) Docs() ecdysis.Docs {
 	return ecdysis.Docs{
 		Short: "Run Conduit",
-		Long:  `Starts the Conduit server and runs the configured pipelines.`,
+		Long: `Starts the Conduit server and runs the configured pipelines.
+
+With --dev, additionally watches pipelines.path and hot-reloads changes into the running engine as
+you save: a processor-only edit applies in place (no restart); a source/destination/topology edit
+applies via a labeled graceful restart; a bad edit (parse/validation failure) is reported and never
+touches the running pipeline. Every apply --dev drives is authorized by the fact that you ran
+--dev and are watching it — it does not set --api.allow-live-restart-apply, which stays
+independently gated for the gRPC/HTTP/MCP surface. See
+docs/design-documents/20260712-pipeline-dev-hot-reload.md and docs/operations/dev-hot-reload.md.`,
+		Example: "conduit run\n" +
+			"conduit run --dev\n" +
+			"conduit run --dev --pipelines.path ./pipelines --dev.json",
 	}
 }

cmd/conduit/root/run/run_test.go

@@ -60,6 +60,9 @@ func TestRunCommandFlags(t *testing.T) {
 		{longName: "dev.cpuprofile", usage: "write CPU profile to file"},
 		{longName: "dev.memprofile", usage: "write memory profile to file"},
 		{longName: "dev.blockprofile", usage: "write block profile to file"},
+		{longName: "dev.enabled", usage: "watch pipelines.path and hot-reload changes into the running engine (see 'conduit run --dev')"},
+		{longName: "dev.json", usage: "emit dev-watcher apply events as JSON lines instead of human-readable text"},
+		{longName: "dev", usage: "alias for --dev.enabled: watch pipelines.path and hot-reload changes into the running engine"},
 	}
 
 	e := ecdysis.New()

docs/operations/dev-hot-reload.md

@@ -0,0 +1,67 @@
+# Dev-mode hot-reload: `conduit run --dev`
+
+`conduit run --dev` runs Conduit and watches the pipelines directory, applying config
+changes to running pipelines on save — so authoring a pipeline is an edit-see-edit-see loop
+instead of stop / edit / restart / wait. It is the developer-facing surface over the live
+in-place apply engine (see
+[`docs/design-documents/20260712-pipeline-dev-hot-reload.md`](../design-documents/20260712-pipeline-dev-hot-reload.md)).
+`conduit pipelines dev [dir]` is a thin alias that runs `conduit run --dev` with dev-tuned
+defaults.
+
+## What it does on each save
+
+1. Debounces rapid saves (a save-storm coalesces into one apply).
+2. Parses every pipeline in the changed file and runs the same enrich + validate the CLI uses.
+3. For each pipeline, computes the diff and applies it:
+   - A **processor config** or pipeline **name/description** change applies **in place** — the
+     pipeline keeps running, the source never restarts, positions are continuous, no record is
+     lost or reordered.
+   - A **connector**, **DLQ**, or **topology** change applies via a **graceful drain-and-restart**
+     (the same no-loss stop-drain-restart `apply` uses), and the output says so.
+4. Ensures the pipeline is running afterward (a brand-new pipeline file, or one left stopped by a
+   prior failed apply, is started — unless the config declares it `stopped`).
+
+Each apply prints whether it was in-place or a restart, the diff, the outcome, and timing;
+`--json` emits the same as structured events.
+
+## The authorization model
+
+`--dev` is the operator authorization. Applying any change to a running pipeline normally
+requires the process-level `--api.allow-live-restart-apply` gate (see
+[`live-restart-apply.md`](live-restart-apply.md)); `--dev` authorizes the applies **its own
+watcher drives from local file edits**, because the operator ran `--dev`, is watching the
+terminal, and sees every diff. It does **not** enable the API/MCP `apply` gate — a remote agent
+still can't apply to a running pipeline over the network unless that separate flag is set.
+
+This is the same trust boundary as `--pipelines.path` provisioning at startup: whoever can edit
+the watched files and run `--dev` on this box can already restart these pipelines.
+
+## Failure modes (by design)
+
+- **Syntax or validation error on save** → the error is printed with its code and path; the
+  running pipeline is **untouched** and keeps running the last-good config. Fix and save again.
+- **A new config fails at open/start** (e.g. an unreachable source) → the pipeline is left stopped
+  with the error; the next good save recovers it (ensure-running restarts it).
+- **A watched file is deleted** → the pipeline is **left running** and a message is logged;
+  `--dev` never deletes a running pipeline because a file vanished (e.g. a `git stash`). Stop it
+  explicitly if that was the intent.
+- **`Ctrl-C`** → the watcher is cancelled as part of graceful shutdown; in-flight records drain.
+
+## Operational notes
+
+- **Not a production deployment mechanism.** `--dev` is a foreground, single-author, local
+  convenience. For unattended or remote apply, use the API/MCP `apply` path with its own operator
+  gate.
+- **Startup is normal.** Existing pipelines are provisioned and started exactly as `conduit run`
+  does; the watcher only handles subsequent edits. An empty pipelines directory is fine — creating
+  the first file live works.
+- **In-place is scoped.** Only processor-config and name/description changes apply without a
+  restart today; connector/DLQ/topology changes restart. See the design doc for why (the engine
+  swaps a processor node live but not a source/destination's position/connection state).
+
+## Related
+
+- [`docs/design-documents/20260712-pipeline-dev-hot-reload.md`](../design-documents/20260712-pipeline-dev-hot-reload.md)
+  — design, failure-mode analysis, and the engine PR1 built.
+- [`live-restart-apply.md`](live-restart-apply.md) — the process-level gate for the API/MCP apply
+  path, which `--dev` deliberately does not enable.

go.mod

@@ -23,6 +23,7 @@ require (
 	github.com/conduitio/yaml/v3 v3.3.0
 	github.com/dop251/goja v0.0.0-20240806095544-3491d4a58fbe
 	github.com/dop251/goja_nodejs v0.0.0-20231122114759-e84d9a924c5c
+	github.com/fsnotify/fsnotify v1.9.0
 	github.com/gammazero/deque v1.2.1
 	github.com/goccy/go-json v0.10.6
 	github.com/google/go-cmp v0.7.0
@@ -134,7 +135,6 @@ require (
 	github.com/fatih/color v1.19.0 // indirect
 	github.com/fatih/structtag v1.2.0 // indirect
 	github.com/firefart/nonamedreturns v1.0.5 // indirect
-	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/fzipp/gocyclo v0.6.0 // indirect
 	github.com/ghostiam/protogetter v0.3.9 // indirect
 	github.com/go-critic/go-critic v0.12.0 // indirect