cmd/relspec/job.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bytes"
 	"fmt"
 	"io"
 	"os"
@@ -11,6 +12,8 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"git.warky.dev/wdevs/relspecgo/pkg/diff"
+	"git.warky.dev/wdevs/relspecgo/pkg/inspector"
 	"git.warky.dev/wdevs/relspecgo/pkg/jobs"
 	"git.warky.dev/wdevs/relspecgo/pkg/merge"
 	"git.warky.dev/wdevs/relspecgo/pkg/models"
@@ -18,6 +21,7 @@ import (
 	"git.warky.dev/wdevs/relspecgo/pkg/readers/sqldir"
 	"git.warky.dev/wdevs/relspecgo/pkg/writers"
 	wpgsql "git.warky.dev/wdevs/relspecgo/pkg/writers/pgsql"
+	"git.warky.dev/wdevs/relspecgo/pkg/writers/sqlexec"
 	wtemplate "git.warky.dev/wdevs/relspecgo/pkg/writers/template"
 )
 
@@ -122,6 +126,9 @@ func loadJobSet() (*jobs.Set, error) {
 	if err := set.Validate(); err != nil {
 		return nil, err
 	}
+	for _, w := range set.Warnings {
+		fmt.Fprintf(os.Stderr, "warning: %s\n", w)
+	}
 	return set, nil
 }
 
@@ -180,12 +187,14 @@ func executeJobPlan(set *jobs.Set, name string, dryRun, noDeps bool, out io.Writ
 	// Pre-flight: resolve and check paths, output policy and env vars for the
 	// whole plan before anything runs. A failure here means no job executes.
 	resolved := make([]*resolvedJob, len(plan))
+	byName := make(map[string]*resolvedJob, len(plan))
 	for i, j := range plan {
-		rj, perr := preflightJob(j)
+		rj, perr := preflightJob(j, byName)
 		if perr != nil {
 			return fmt.Errorf("job %q: %w", j.Name, perr)
 		}
 		resolved[i] = rj
+		byName[j.Name] = rj
 	}
 
 	if dryRun {
@@ -217,7 +226,12 @@ type resolvedJob struct {
 	outputConn    string // resolved connection string (secret)
 	outputConnEnv string
 	logPath       string
+	logPolicy     jobs.LogPolicy
 	templatePath  string
+	reportPath    string // "" for a diff summary written to the log
+	reportFormat  string
+	rulesPath     string // "" means inspector defaults
+	selection     *splitSelection
 	secrets       []string // resolved secret values to redact from logs
 }
 
@@ -226,11 +240,12 @@ type resolvedInput struct {
 	path    string // "" when the input is a database
 	conn    string // resolved connection string (secret)
 	connEnv string
+	fromJob string // producer job name when this input came from from_job
 }
 
-func preflightJob(j *jobs.Job) (*resolvedJob, error) {
+func preflightJob(j *jobs.Job, resolvedByName map[string]*resolvedJob) (*resolvedJob, error) {
 	root := j.Dir()
-	rj := &resolvedJob{job: j, root: root}
+	rj := &resolvedJob{job: j, root: root, logPolicy: j.ResolvedLogPolicy()}
 
 	if j.Logfile != "" {
 		p, err := jobs.SafeJoin(root, j.Logfile)
@@ -253,6 +268,20 @@ func preflightJob(j *jobs.Job) (*resolvedJob, error) {
 
 	for i, in := range j.Inputs {
 		ri := resolvedInput{format: strings.ToLower(in.Format)}
+		if in.FromJob != "" {
+			producer, ok := resolvedByName[in.FromJob]
+			if !ok {
+				return nil, fmt.Errorf("input[%d]: from_job %q is not in this plan (do not use --no-deps with from_job inputs)", i, in.FromJob)
+			}
+			if producer.outputPath == "" {
+				return nil, fmt.Errorf("input[%d]: from_job %q does not write a file output", i, in.FromJob)
+			}
+			ri.path = producer.outputPath
+			ri.format = strings.ToLower(producer.job.Output.Format)
+			ri.fromJob = in.FromJob
+			rj.inputs = append(rj.inputs, ri)
+			continue
+		}
 		if in.ConnEnv != "" {
 			v, ok := os.LookupEnv(in.ConnEnv)
 			if !ok || v == "" {
@@ -313,6 +342,43 @@ func preflightJob(j *jobs.Job) (*resolvedJob, error) {
 			rj.outputPath = p
 		}
 	}
+
+	if j.Rules != "" {
+		p, err := jobs.SafeJoin(root, j.Rules)
+		if err != nil {
+			return nil, fmt.Errorf("rules: %w", err)
+		}
+		info, err := os.Stat(p)
+		if err != nil || info.IsDir() {
+			return nil, fmt.Errorf("rules %q: not found or is a directory", j.Rules)
+		}
+		rj.rulesPath = p
+	}
+
+	if j.Report != nil {
+		rj.reportFormat = strings.ToLower(j.Report.Format)
+		if j.Report.Path != "" {
+			p, err := jobs.SafeJoin(root, j.Report.Path)
+			if err != nil {
+				return nil, fmt.Errorf("report: %w", err)
+			}
+			if _, err := os.Stat(p); err == nil && !j.Report.Overwrite {
+				return nil, fmt.Errorf("report %s already exists (set report.overwrite: true to replace it)", j.Report.Path)
+			}
+			rj.reportPath = p
+		}
+	}
+
+	if j.Select != nil {
+		rj.selection = &splitSelection{
+			Schemas:        j.Select.Schemas,
+			Tables:         j.Select.Tables,
+			ExcludeSchemas: j.Select.ExcludeSchemas,
+			ExcludeTables:  j.Select.ExcludeTables,
+			DatabaseName:   j.Select.DatabaseName,
+		}
+	}
+
 	return rj, nil
 }
 
@@ -325,9 +391,12 @@ func printResolvedJob(out io.Writer, n, total int, rj *resolvedJob) {
 	}
 	fmt.Fprintf(out, "      job file: %s\n", j.SourceFile)
 	for _, ri := range rj.inputs {
-		if ri.path != "" {
+		switch {
+		case ri.fromJob != "":
+			fmt.Fprintf(out, "      input: %s (%s) from job %q\n", ri.path, ri.format, ri.fromJob)
+		case ri.path != "":
 			fmt.Fprintf(out, "      input: %s (%s)\n", ri.path, ri.format)
-		} else {
+		default:
 			fmt.Fprintf(out, "      input: env:%s (%s)\n", ri.connEnv, ri.format)
 		}
 	}
@@ -339,15 +408,31 @@ func printResolvedJob(out io.Writer, n, total int, rj *resolvedJob) {
 	} else if rj.outputConnEnv != "" {
 		fmt.Fprintf(out, "      output: env:%s (%s)\n", rj.outputConnEnv, j.Output.Format)
 	}
+	if j.Report != nil {
+		format := valueOr(rj.reportFormat, "default")
+		if rj.reportPath != "" {
+			fmt.Fprintf(out, "      report: %s (%s)\n", rj.reportPath, format)
+		} else {
+			fmt.Fprintf(out, "      report: (log) (%s)\n", format)
+		}
+	}
+	if rj.rulesPath != "" {
+		fmt.Fprintf(out, "      rules: %s\n", rj.rulesPath)
+	} else if j.Command == jobs.CommandInspect {
+		fmt.Fprintf(out, "      rules: (built-in defaults)\n")
+	}
+	if rj.selection != nil {
+		fmt.Fprintf(out, "      select: %s\n", rj.selection.summary())
+	}
 	if rj.logPath != "" {
-		fmt.Fprintf(out, "      logfile: %s\n", rj.logPath)
+		fmt.Fprintf(out, "      logfile: %s (rotate >= %d bytes, keep %d)\n", rj.logPath, rj.logPolicy.MaxSizeBytes, rj.logPolicy.Keep)
 	}
 	fmt.Fprintln(out)
 }
 
 // executeResolvedJob runs a single already-validated job.
 func executeResolvedJob(rj *resolvedJob) (err error) {
-	lg, closeLog, lerr := newJobLogger(rj.logPath, rj.secrets)
+	lg, closeLog, lerr := newJobLogger(rj.logPath, rj.logPolicy, rj.secrets)
 	if lerr != nil {
 		return lerr
 	}
@@ -364,6 +449,14 @@ func executeResolvedJob(rj *resolvedJob) (err error) {
 		err = runScriptsListJob(rj, lg)
 	case jobs.CommandTempl:
 		err = runTemplJob(rj, lg)
+	case jobs.CommandSplit:
+		err = runSplitJob(rj, lg)
+	case jobs.CommandInspect:
+		err = runInspectJob(rj, lg)
+	case jobs.CommandDiff:
+		err = runDiffJob(rj, lg)
+	case jobs.CommandScriptsExec:
+		err = runScriptsExecJob(rj, lg)
 	default:
 		err = fmt.Errorf("unsupported command %q", rj.job.Command)
 	}
@@ -505,6 +598,168 @@ func runScriptsListJob(rj *resolvedJob, lg *jobLogger) error {
 	return nil
 }
 
+func runSplitJob(rj *resolvedJob, lg *jobLogger) error {
+	db, err := readJobInputs(rj, lg)
+	if err != nil {
+		return err
+	}
+	sel := splitSelection{}
+	if rj.selection != nil {
+		sel = *rj.selection
+	}
+	filtered, err := filterDatabaseSelection(db, sel)
+	if err != nil {
+		return fmt.Errorf("split selection: %w", err)
+	}
+	if sel.DatabaseName != "" {
+		filtered.Name = sel.DatabaseName
+	}
+	tables := 0
+	for _, s := range filtered.Schemas {
+		tables += len(s.Tables)
+	}
+	lg.logf("split: selected %d schema(s), %d table(s)", len(filtered.Schemas), tables)
+	return writeJobOutput(rj, filtered, lg)
+}
+
+func runInspectJob(rj *resolvedJob, lg *jobLogger) error {
+	db, err := readJobInputs(rj, lg)
+	if err != nil {
+		return err
+	}
+	config, err := inspector.LoadConfig(rj.rulesPath) // "" -> built-in defaults
+	if err != nil {
+		return fmt.Errorf("load rules: %w", err)
+	}
+	report, err := inspector.NewInspector(db, config).Inspect()
+	if err != nil {
+		return fmt.Errorf("inspection failed: %w", err)
+	}
+
+	var formatted string
+	switch valueOr(rj.reportFormat, "markdown") {
+	case "json":
+		formatted, err = inspector.NewJSONFormatter().Format(report)
+	default:
+		formatted, err = inspector.NewMarkdownFormatter(io.Discard).Format(report)
+	}
+	if err != nil {
+		return fmt.Errorf("format report: %w", err)
+	}
+	if werr := atomicWrite(rj.reportPath, func(tmp string) error {
+		return os.WriteFile(tmp, []byte(formatted), 0o644)
+	}); werr != nil {
+		return werr
+	}
+	lg.logf("inspect: %d error(s), %d warning(s) -> %s",
+		report.Summary.ErrorCount, report.Summary.WarningCount, rj.reportPath)
+	if report.HasErrors() {
+		return fmt.Errorf("inspection found %d error(s)", report.Summary.ErrorCount)
+	}
+	return nil
+}
+
+func runDiffJob(rj *resolvedJob, lg *jobLogger) error {
+	if len(rj.inputs) != 2 {
+		return fmt.Errorf("diff requires exactly 2 inputs, got %d", len(rj.inputs))
+	}
+	source, err := readOneJobInput(rj.inputs[0])
+	if err != nil {
+		return fmt.Errorf("input[0]: %w", err)
+	}
+	lg.logf("diff source: %s", inputLabel(rj.inputs[0]))
+	target, err := readOneJobInput(rj.inputs[1])
+	if err != nil {
+		return fmt.Errorf("input[1]: %w", err)
+	}
+	lg.logf("diff target: %s", inputLabel(rj.inputs[1]))
+
+	result := diff.CompareDatabases(source, target)
+	s := diff.ComputeSummary(result)
+	lg.logf("diff: schemas %d/%d/%d, tables %d/%d/%d, columns %d/%d/%d (missing/extra/modified)",
+		s.Schemas.Missing, s.Schemas.Extra, s.Schemas.Modified,
+		s.Tables.Missing, s.Tables.Extra, s.Tables.Modified,
+		s.Columns.Missing, s.Columns.Extra, s.Columns.Modified)
+
+	format := diff.FormatSummary
+	switch rj.reportFormat {
+	case "json":
+		format = diff.FormatJSON
+	case "html":
+		format = diff.FormatHTML
+	}
+
+	if rj.reportPath == "" {
+		var buf bytes.Buffer
+		if err := diff.FormatDiff(result, format, &buf); err != nil {
+			return fmt.Errorf("format diff: %w", err)
+		}
+		for _, line := range strings.Split(strings.TrimRight(buf.String(), "\n"), "\n") {
+			lg.logf("%s", line)
+		}
+		return nil
+	}
+	if werr := atomicWrite(rj.reportPath, func(tmp string) error {
+		f, err := os.Create(tmp)
+		if err != nil {
+			return err
+		}
+		defer f.Close()
+		return diff.FormatDiff(result, format, f)
+	}); werr != nil {
+		return werr
+	}
+	lg.logf("diff report written: %s", rj.reportPath)
+	return nil
+}
+
+func runScriptsExecJob(rj *resolvedJob, lg *jobLogger) error {
+	schemaName := valueOr(rj.job.Options.Schema, "public")
+	combined := &models.Schema{Name: schemaName}
+	for _, dir := range rj.scriptDirs {
+		reader := sqldir.NewReader(&readers.ReaderOptions{
+			FilePath: dir,
+			Metadata: map[string]any{
+				"schema_name":   schemaName,
+				"database_name": "database",
+			},
+		})
+		db, err := reader.ReadDatabase()
+		if err != nil {
+			return fmt.Errorf("%s: %w", dir, err)
+		}
+		if len(db.Schemas) == 0 {
+			continue
+		}
+		combined.Scripts = append(combined.Scripts, db.Schemas[0].Scripts...)
+	}
+	if len(combined.Scripts) == 0 {
+		lg.logf("no scripts found; nothing to execute")
+		return nil
+	}
+	lg.logf("executing %d script(s) against database env:%s", len(combined.Scripts), rj.outputConnEnv)
+
+	writer := sqlexec.NewWriter(&writers.WriterOptions{
+		Metadata: map[string]any{
+			"connection_string": rj.outputConn,
+			"ignore_errors":     rj.job.Options.ContinueOnError,
+		},
+	})
+	if err := writer.WriteSchema(combined); err != nil {
+		return fmt.Errorf("script execution failed: %w", err)
+	}
+
+	opts := writer.Options()
+	total, _ := opts.Metadata["execution_total"].(int)
+	success, _ := opts.Metadata["execution_success"].(int)
+	failed, _ := opts.Metadata["execution_failed"].(int)
+	lg.logf("executed %d script(s): %d succeeded, %d failed", total, success, failed)
+	if failed > 0 && !rj.job.Options.ContinueOnError {
+		return fmt.Errorf("%d script(s) failed", failed)
+	}
+	return nil
+}
+
 // readJobInputs reads every input and additively merges them into one model.
 func readJobInputs(rj *resolvedJob, lg *jobLogger) (*models.Database, error) {
 	var base *models.Database
@@ -559,7 +814,37 @@ func writeJobOutput(rj *resolvedJob, db *models.Database, lg *jobLogger) error {
 		return fmt.Errorf("failed to create output directory: %w", err)
 	}
 	lg.logf("writing output: %s (%s)", rj.outputPath, format)
-	return writeDatabase(db, format, rj.outputPath, o.Package, o.Schema, o.FlattenSchema, "", "", o.ContinueOnError, "")
+
+	write := func(target string) error {
+		return writeDatabase(db, format, target, o.Package, o.Schema, o.FlattenSchema, "", "", o.ContinueOnError, "")
+	}
+	// Single-file formats are written to a temp file and renamed into place so
+	// a failure never leaves a partial or truncated output. Directory-emitting
+	// formats (gorm/bun/drizzle/typeorm/prisma) write in place.
+	if jobs.SingleFileOutputFormat(format) {
+		return atomicWrite(rj.outputPath, write)
+	}
+	return write(rj.outputPath)
+}
+
+// atomicWrite calls produce with a temp path in the same directory as
+// finalPath, then renames it over finalPath. The temp file is removed on any
+// error so the destination is only ever replaced by a complete file.
+func atomicWrite(finalPath string, produce func(tmpPath string) error) error {
+	dir := filepath.Dir(finalPath)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return fmt.Errorf("failed to create output directory: %w", err)
+	}
+	tmp := filepath.Join(dir, fmt.Sprintf(".%s.relspec-tmp-%d", filepath.Base(finalPath), os.Getpid()))
+	if err := produce(tmp); err != nil {
+		_ = os.Remove(tmp)
+		return err
+	}
+	if err := os.Rename(tmp, finalPath); err != nil {
+		_ = os.Remove(tmp)
+		return fmt.Errorf("failed to finalize %s: %w", finalPath, err)
+	}
+	return nil
 }
 
 // --- logging + redaction ---------------------------------------------------
@@ -572,7 +857,7 @@ type jobLogger struct {
 // newJobLogger returns a logger that mirrors to stderr and, when path is set,
 // to a job logfile. Connection strings and known secret values are redacted
 // from everything it writes.
-func newJobLogger(path string, secrets []string) (*jobLogger, func(err error), error) {
+func newJobLogger(path string, policy jobs.LogPolicy, secrets []string) (*jobLogger, func(err error), error) {
 	lg := &jobLogger{secrets: secrets}
 	if path == "" {
 		return lg, func(error) {}, nil
@@ -580,6 +865,7 @@ func newJobLogger(path string, secrets []string) (*jobLogger, func(err error), e
 	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
 		return nil, nil, fmt.Errorf("failed to create log directory: %w", err)
 	}
+	rotateLogIfNeeded(path, policy)
 	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to open logfile %q: %w", path, err)
@@ -593,6 +879,28 @@ func newJobLogger(path string, secrets []string) (*jobLogger, func(err error), e
 	}, nil
 }
 
+// rotateLogIfNeeded renames path -> path.1 -> path.2 ... up to policy.Keep
+// when path has grown to policy.MaxSizeBytes or more. The oldest file beyond
+// Keep is deleted. A zero/negative MaxSizeBytes disables rotation.
+func rotateLogIfNeeded(path string, policy jobs.LogPolicy) {
+	if policy.MaxSizeBytes <= 0 {
+		return
+	}
+	info, err := os.Stat(path)
+	if err != nil || info.Size() < policy.MaxSizeBytes {
+		return
+	}
+	if policy.Keep < 1 {
+		_ = os.Remove(path)
+		return
+	}
+	_ = os.Remove(fmt.Sprintf("%s.%d", path, policy.Keep))
+	for i := policy.Keep - 1; i >= 1; i-- {
+		_ = os.Rename(fmt.Sprintf("%s.%d", path, i), fmt.Sprintf("%s.%d", path, i+1))
+	}
+	_ = os.Rename(path, path+".1")
+}
+
 func (l *jobLogger) logf(format string, args ...interface{}) {
 	line := l.redact(fmt.Sprintf(format, args...))
 	fmt.Fprintf(os.Stderr, "      %s\n", line)

cmd/relspec/job_test.go

@@ -376,6 +376,286 @@ jobs:
 	}
 }
 
+func TestJobRun_SplitJob(t *testing.T) {
+	dir := jobFixture(t, `version: 1
+jobs:
+  extract:
+    command: split
+    inputs:
+      - path: schema/core.dbml
+        format: dbml
+      - path: schema/tenant.dbml
+        format: dbml
+    select:
+      tables: [users]
+    output:
+      format: json
+      path: build/subset.json
+      overwrite: true
+`)
+	set := mustLoadSet(t, filepath.Join(dir, "relspec.yml"))
+	if err := executeJobPlan(set, "extract", false, false, &bytes.Buffer{}); err != nil {
+		t.Fatalf("execute split job: %v", err)
+	}
+	out, err := os.ReadFile(filepath.Join(dir, "build", "subset.json"))
+	if err != nil {
+		t.Fatalf("read split output: %v", err)
+	}
+	s := string(out)
+	if !strings.Contains(s, "users") {
+		t.Fatalf("split output missing selected table:\n%s", s)
+	}
+	if strings.Contains(s, "posts") {
+		t.Fatalf("split output should have excluded posts:\n%s", s)
+	}
+}
+
+func TestJobRun_InspectJob(t *testing.T) {
+	dir := jobFixture(t, `version: 1
+jobs:
+  lint:
+    command: inspect
+    inputs:
+      - path: schema/core.dbml
+        format: dbml
+    report:
+      format: json
+      path: build/report.json
+      overwrite: true
+    logfile: .relspec/lint.log
+`)
+	set := mustLoadSet(t, filepath.Join(dir, "relspec.yml"))
+	// Default rules only warn, so the job succeeds.
+	if err := executeJobPlan(set, "lint", false, false, &bytes.Buffer{}); err != nil {
+		t.Fatalf("execute inspect job: %v", err)
+	}
+	if _, err := os.ReadFile(filepath.Join(dir, "build", "report.json")); err != nil {
+		t.Fatalf("expected report file: %v", err)
+	}
+	logData, _ := os.ReadFile(filepath.Join(dir, ".relspec", "lint.log"))
+	if !strings.Contains(string(logData), "inspect:") {
+		t.Fatalf("logfile missing inspect summary:\n%s", logData)
+	}
+}
+
+func TestJobRun_InspectJobFailsOnRuleError(t *testing.T) {
+	dir := jobFixture(t, `version: 1
+jobs:
+  lint:
+    command: inspect
+    inputs:
+      - path: schema/core.dbml
+        format: dbml
+    rules: rules.yaml
+    report:
+      format: json
+      path: build/report.json
+      overwrite: true
+    logfile: .relspec/lint.log
+`)
+	// A rule set to "error" level for a violation the fixture triggers.
+	writeFile(t, filepath.Join(dir, "rules.yaml"), `version: "1.0"
+rules:
+  primary_key_naming:
+    enabled: enforce
+    function: primary_key_naming
+    pattern: "^id_"
+    message: "Primary key columns should start with id_"
+`)
+	set := mustLoadSet(t, filepath.Join(dir, "relspec.yml"))
+	err := executeJobPlan(set, "lint", false, false, &bytes.Buffer{})
+	if err == nil || !strings.Contains(err.Error(), "error(s)") {
+		t.Fatalf("expected inspect job to fail on rule error, got %v", err)
+	}
+	logData, _ := os.ReadFile(filepath.Join(dir, ".relspec", "lint.log"))
+	if !strings.Contains(string(logData), "FAILED") {
+		t.Fatalf("failed inspect job should log FAILED:\n%s", logData)
+	}
+}
+
+func TestJobRun_DiffJob(t *testing.T) {
+	dir := jobFixture(t, `version: 1
+jobs:
+  compare:
+    command: diff
+    inputs:
+      - path: schema/core.dbml
+        format: dbml
+      - path: schema/tenant.dbml
+        format: dbml
+    report:
+      format: json
+      path: build/diff.json
+      overwrite: true
+`)
+	set := mustLoadSet(t, filepath.Join(dir, "relspec.yml"))
+	if err := executeJobPlan(set, "compare", false, false, &bytes.Buffer{}); err != nil {
+		t.Fatalf("execute diff job: %v", err)
+	}
+	out, err := os.ReadFile(filepath.Join(dir, "build", "diff.json"))
+	if err != nil {
+		t.Fatalf("read diff report: %v", err)
+	}
+	if len(out) == 0 {
+		t.Fatal("diff report is empty")
+	}
+}
+
+func TestJobRun_FromJobWiring(t *testing.T) {
+	dir := jobFixture(t, `version: 1
+jobs:
+  a:
+    command: convert
+    inputs:
+      - path: schema/core.dbml
+        format: dbml
+    output:
+      format: json
+      path: build/a.json
+      overwrite: true
+  b:
+    command: convert
+    inputs:
+      - from_job: a
+    output:
+      format: yaml
+      path: build/b.yaml
+      overwrite: true
+`)
+	set := mustLoadSet(t, filepath.Join(dir, "relspec.yml"))
+	if err := executeJobPlan(set, "b", false, false, &bytes.Buffer{}); err != nil {
+		t.Fatalf("execute from_job chain: %v", err)
+	}
+	if _, err := os.Stat(filepath.Join(dir, "build", "a.json")); err != nil {
+		t.Fatalf("producer output missing: %v", err)
+	}
+	out, err := os.ReadFile(filepath.Join(dir, "build", "b.yaml"))
+	if err != nil {
+		t.Fatalf("consumer output missing: %v", err)
+	}
+	if !strings.Contains(string(out), "users") {
+		t.Fatalf("consumer did not consume producer output:\n%s", out)
+	}
+}
+
+func TestJobRun_LogRotation(t *testing.T) {
+	dir := jobFixture(t, `version: 1
+jobs:
+  build:
+    command: scripts-list
+    script_dirs: [migrations]
+    log_max_size: "150B"
+    log_keep: 2
+    logfile: .relspec/build.log
+`)
+	writeFile(t, filepath.Join(dir, "migrations", "1_001_a.sql"), "CREATE TABLE a();\n")
+	logPath := filepath.Join(dir, ".relspec", "build.log")
+	writeFile(t, logPath, strings.Repeat("x", 300)+"\n")
+
+	set := mustLoadSet(t, filepath.Join(dir, "relspec.yml"))
+	if err := executeJobPlan(set, "build", false, false, &bytes.Buffer{}); err != nil {
+		t.Fatalf("execute job: %v", err)
+	}
+	rotated, err := os.ReadFile(logPath + ".1")
+	if err != nil {
+		t.Fatalf("expected rotated logfile build.log.1: %v", err)
+	}
+	if !strings.Contains(string(rotated), strings.Repeat("x", 300)) {
+		t.Fatalf("rotated logfile should hold the old content")
+	}
+	fresh, err := os.ReadFile(logPath)
+	if err != nil {
+		t.Fatalf("expected fresh logfile: %v", err)
+	}
+	if strings.Contains(string(fresh), strings.Repeat("x", 300)) {
+		t.Fatalf("fresh logfile should not contain the rotated-out content:\n%s", fresh)
+	}
+	if !strings.Contains(string(fresh), "OK") {
+		t.Fatalf("fresh logfile should hold the new run:\n%s", fresh)
+	}
+}
+
+func TestJobRun_AtomicOutputLeavesOriginalOnFailure(t *testing.T) {
+	dir := jobFixture(t, `version: 1
+jobs:
+  x:
+    command: convert
+    inputs:
+      - path: schema/core.dbml
+        format: dbml
+    output:
+      format: json
+      path: build/out.json
+      overwrite: true
+`)
+	// Seed the destination, then make its parent directory read-only so the
+	// rename step fails. The seeded file must survive intact.
+	seeded := filepath.Join(dir, "build", "out.json")
+	writeFile(t, seeded, `{"seeded":true}`)
+	if err := os.Chmod(filepath.Join(dir, "build"), 0o500); err != nil {
+		t.Skipf("cannot chmod: %v", err)
+	}
+	t.Cleanup(func() { _ = os.Chmod(filepath.Join(dir, "build"), 0o755) })
+
+	set := mustLoadSet(t, filepath.Join(dir, "relspec.yml"))
+	if err := executeJobPlan(set, "x", false, false, &bytes.Buffer{}); err == nil {
+		t.Skip("write unexpectedly succeeded (running as root?)")
+	}
+	if err := os.Chmod(filepath.Join(dir, "build"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	data, err := os.ReadFile(seeded)
+	if err != nil {
+		t.Fatalf("seeded file gone: %v", err)
+	}
+	if !strings.Contains(string(data), "seeded") {
+		t.Fatalf("seeded file was corrupted: %s", data)
+	}
+}
+
+func TestJobRun_ScriptsExecMissingConnEnv(t *testing.T) {
+	dir := jobFixture(t, `version: 1
+jobs:
+  migrate:
+    command: scripts-exec
+    script_dirs: [migrations]
+    output:
+      conn_env: RELSPEC_TEST_EXEC_MISSING
+    logfile: .relspec/migrate.log
+`)
+	writeFile(t, filepath.Join(dir, "migrations", "1_001_a.sql"), "CREATE TABLE a();\n")
+	os.Unsetenv("RELSPEC_TEST_EXEC_MISSING")
+	set := mustLoadSet(t, filepath.Join(dir, "relspec.yml"))
+	err := executeJobPlan(set, "migrate", false, false, &bytes.Buffer{})
+	if err == nil || !strings.Contains(err.Error(), "conn_env") {
+		t.Fatalf("expected missing conn_env error, got %v", err)
+	}
+}
+
+func TestJobRun_ScriptsExecDryRun(t *testing.T) {
+	dir := jobFixture(t, `version: 1
+jobs:
+  migrate:
+    command: scripts-exec
+    script_dirs: [migrations]
+    output:
+      conn_env: RELSPEC_TEST_EXEC_CONN
+`)
+	writeFile(t, filepath.Join(dir, "migrations", "1_001_a.sql"), "CREATE TABLE a();\n")
+	t.Setenv("RELSPEC_TEST_EXEC_CONN", "postgres://u:secretpw@h/db")
+	set := mustLoadSet(t, filepath.Join(dir, "relspec.yml"))
+	var buf bytes.Buffer
+	if err := executeJobPlan(set, "migrate", true, false, &buf); err != nil {
+		t.Fatalf("dry run: %v", err)
+	}
+	if strings.Contains(buf.String(), "secretpw") {
+		t.Fatalf("plan leaked secret:\n%s", buf.String())
+	}
+	if !strings.Contains(buf.String(), "env:RELSPEC_TEST_EXEC_CONN") {
+		t.Fatalf("plan should name the env var:\n%s", buf.String())
+	}
+}
+
 func TestJobRun_TemplDatabaseMode(t *testing.T) {
 	dir := jobFixture(t, `version: 1
 jobs:

cmd/relspec/split.go

@@ -205,8 +205,52 @@ func runSplit(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
-// filterDatabase filters the database based on provided criteria
+// splitSelection is the schema/table selection for a split, independent of the
+// CLI flag globals so the job runner can build one directly.
+type splitSelection struct {
+	Schemas        []string
+	Tables         []string
+	ExcludeSchemas []string
+	ExcludeTables  []string
+	DatabaseName   string
+}
+
+// summary renders a one-line human description of the selection.
+func (s splitSelection) summary() string {
+	var parts []string
+	if len(s.Schemas) > 0 {
+		parts = append(parts, "schemas="+strings.Join(s.Schemas, ","))
+	}
+	if len(s.Tables) > 0 {
+		parts = append(parts, "tables="+strings.Join(s.Tables, ","))
+	}
+	if len(s.ExcludeSchemas) > 0 {
+		parts = append(parts, "exclude_schemas="+strings.Join(s.ExcludeSchemas, ","))
+	}
+	if len(s.ExcludeTables) > 0 {
+		parts = append(parts, "exclude_tables="+strings.Join(s.ExcludeTables, ","))
+	}
+	if s.DatabaseName != "" {
+		parts = append(parts, "database_name="+s.DatabaseName)
+	}
+	if len(parts) == 0 {
+		return "(all schemas/tables)"
+	}
+	return strings.Join(parts, " ")
+}
+
+// filterDatabase filters the database based on the CLI split flags.
 func filterDatabase(db *models.Database) (*models.Database, error) {
+	return filterDatabaseSelection(db, splitSelection{
+		Schemas:        parseCommaSeparated(splitSchemas),
+		Tables:         parseCommaSeparated(splitTables),
+		ExcludeSchemas: parseCommaSeparated(splitExcludeSchema),
+		ExcludeTables:  parseCommaSeparated(splitExcludeTables),
+	})
+}
+
+// filterDatabaseSelection filters db down to the schemas/tables named by sel.
+func filterDatabaseSelection(db *models.Database, sel splitSelection) (*models.Database, error) {
 	filteredDB := &models.Database{
 		Name:            db.Name,
 		Description:     db.Description,
@@ -220,11 +264,11 @@ func filterDatabase(db *models.Database) (*models.Database, error) {
 		Domains:         db.Domains, // Keep domains for now
 	}
 
-	// Parse filter flags
-	includeSchemas := parseCommaSeparated(splitSchemas)
-	includeTables := parseCommaSeparated(splitTables)
-	excludeSchemas := parseCommaSeparated(splitExcludeSchema)
-	excludeTables := parseCommaSeparated(splitExcludeTables)
+	// Selection criteria
+	includeSchemas := sel.Schemas
+	includeTables := sel.Tables
+	excludeSchemas := sel.ExcludeSchemas
+	excludeTables := sel.ExcludeTables
 
 	// Convert table names to lowercase for case-insensitive matching
 	includeTablesLower := make(map[string]bool)

docs/JOB_FILES.md

@@ -9,10 +9,11 @@ relspec job run build-schema --plan  # validate + print plan, execute nothing
 relspec job run build-schema         # run the job (and its dependencies)
 ```
 
-## Design contract (first release)
+## Design contract
 
-This is the smallest coherent contract that is safe and useful end to end.
-Anything not listed under "Supported" is intentionally deferred.
+This is a deliberately small, safe contract. Every capability is offline-testable
+except live database execution (`scripts-exec`), which is validated and planned
+offline and only connects at run time.
 
 ### Not a shell
 
@@ -24,14 +25,15 @@ means adding a vetted adapter in the RelSpec source.
 |----------------|--------------------------------------------------------------------|
 | `convert`      | read one or more input schemas, additively merge them, write one output |
 | `merge`        | like `convert` but requires ≥2 inputs and exposes `skip_*` merge options |
+| `split`        | read one or more schemas, keep the selected schemas/tables, write one output |
 | `scripts-list` | deterministically list SQL scripts across one or more directories   |
+| `scripts-exec` | execute SQL scripts across one or more directories against a live PostgreSQL database |
 | `templ`        | apply a custom Go text template to one or more input schemas          |
+| `inspect`      | validate one or more schemas against rules and write a report       |
+| `diff`         | compare exactly two schemas and write a differences report          |
 
-Deferred (documented, not implemented here): `scripts` execution against a live
-database, `split`, `inspect`, `diff`, job-to-job output wiring,
-log rotation/retention. Live SQL execution already exists as
-`relspec scripts execute`; wiring it into the job runner is a follow-up because
-it needs live database credentials and cannot be covered by offline tests.
+`convert`, `merge` and `split` are **producers**: their file output can be fed
+directly into another job with `from_job` (see below).
 
 ### Discovery and precedence
 
@@ -49,12 +51,16 @@ already forbid duplicate keys within a single file.
 
 ### Paths
 
-* Every path (`inputs[].path`, `output.path`, `script_dirs[]`, `logfile`) is
-  **relative to the directory containing the job file that declared the job**,
-  not the process working directory.
+* Every path (`inputs[].path`, `output.path`, `report.path`, `rules`,
+  `script_dirs[]`, `template`, `logfile`) is **relative to the directory
+  containing the job file that declared the job**, not the process working
+  directory.
 * Absolute paths, `~`-relative paths and any path that resolves outside the job
   file directory (`../`, `a/../../b`, …) are **rejected during validation** —
   before anything runs.
+* At run time each path is additionally resolved through its symlinks: a symlink
+  inside the job-file directory that points outside it is rejected before the
+  path is opened.
 
 ### Credentials
 
@@ -76,58 +82,104 @@ already forbid duplicate keys within a single file.
 first. Nothing is read, written, connected to, or executed if validation fails.
 Checks include:
 
-* schema `version` (must be `1`), unknown YAML fields rejected
+* schema `version` — **forward-permissive**: any version `>= 1` is accepted.
+  An omitted `version` is treated as the current one. A version newer than this
+  build understands loads best-effort (unknown YAML fields are ignored and a
+  warning is printed); at the current version unknown YAML fields are still
+  rejected.
 * duplicate job names across files
 * unknown / missing `command`
-* per-command input/output shape (`convert`/`merge` need inputs + output;
-  `scripts-list` needs `script_dirs` and forbids inputs/output)
+* per-command input/output shape:
+  * `convert` needs ≥1 input + output; `merge` needs ≥2 inputs + output
+  * `split` needs ≥1 input + a file output, plus an optional `select:` block
+  * `scripts-list` needs `script_dirs` and forbids inputs/output
+  * `scripts-exec` needs `script_dirs` and `output.conn_env` (pgsql only)
+  * `inspect` needs ≥1 input + `report:` (format `markdown`|`json`)
+  * `diff` needs **exactly 2** inputs + `report:` (format `summary`|`json`|`html`)
 * unknown input/output `format`
+* `from_job` targets exist, are producers (`convert`/`merge`/`split`) and write a
+  single-file output
 * path traversal / absolute / home-relative paths
-* `depends_on` targets exist
-* dependency cycles (reported as `a -> b -> c -> a`)
+* `depends_on` and `from_job` targets exist
+* dependency cycles over the combined `depends_on` + `from_job` graph
+  (reported as `a -> b -> c -> a`)
 
 Then, immediately before running, per-job pre-flight resolves paths and checks:
 
-* every input file exists and is a file
+* every input file exists and is a file (a `from_job` input is exempt — its
+  producer runs earlier in the same plan)
 * every `script_dir` exists and is a directory
 * every `conn_env` variable is set
-* `output.path` does not already exist unless `output.overwrite: true`
+* `output.path` / `report.path` does not already exist unless the matching
+  `overwrite: true` is set
+* `rules` (inspect), when given, exists and is a file
+* symlinks in every resolved path stay inside the job-file directory
 
 If any pre-flight check fails for **any** job in the plan, **no** job runs.
 
 ### Execution and exit codes
 
-* `relspec job run <name>` runs the job's `depends_on` closure first, in
-  topological order (deterministic), then the job. `--no-deps` runs only the
-  named job.
+* `relspec job run <name>` runs the job's dependency closure first
+  (`depends_on` plus any `from_job` producers), in topological order
+  (deterministic), then the job. `--no-deps` runs only the named job and is
+  incompatible with `from_job` inputs.
 * `--dry-run` (alias `--plan`) prints the resolved plan and exits 0 without
   touching inputs, outputs or databases.
 * A failing job returns the underlying non-zero status (the process exits 1)
   and the error names the job. The logfile records `FAILED: <error>`; a
   successful job records `OK`. No separate success-marker file is written, so a
   failure can never leave a stale "success".
+* `inspect` fails the job when the report contains rule **errors** (enforced
+  rules); warnings do not fail it. `diff` never fails on differences.
+* Single-file outputs and reports are written to a temporary file in the target
+  directory and atomically renamed into place, so an interrupted run never
+  leaves a partial file. Directory-emitting formats (`gorm`, `bun`, `drizzle`,
+  `typeorm`, `prisma`) are written in place.
+
+### Logfile rotation
+
+When a job has a `logfile`, it is size-rotated before each run. Defaults are
+**5 MB** with **3** rotated files kept (`build.log` → `build.log.1` → …). Override
+per job with `log_max_size` / `log_keep`, or for a whole file with a top-level
+`defaults:` block. `log_max_size` accepts `B`/`KB`/`MB`/`GB` suffixes (e.g.
+`"512KB"`, `"5MB"`).
 
 ## Schema reference
 
 ```yaml
-version: 1                       # required, must be 1
+version: 1                       # optional; any value >= 1 is accepted
+defaults:                        # optional, file-wide
+  log_max_size: 5MB              # B / KB / MB / GB
+  log_keep: 3
 jobs:
   <job-name>:
-    command: convert | merge | scripts-list   # required
+    command: convert | merge | split | scripts-list | scripts-exec | templ | inspect | diff
     description: "free text"                   # optional, shown by `job list`
     depends_on: [other-job, ...]               # optional
-    inputs:                                    # convert (≥1) / merge (≥2)
+    inputs:                                    # convert (≥1) / merge (≥2) / split (≥1) / inspect (≥1) / diff (exactly 2)
       - path: relative/file.dbml               # file inputs
         format: dbml
       - format: pgsql                          # live-connection inputs
         conn_env: SOURCE_DB_URL                # env var NAME
-    script_dirs:                               # scripts-list (≥1)
+      - from_job: build-schema                 # consume another job's file output
+    script_dirs:                               # scripts-list / scripts-exec (≥1)
       - migrations/core
       - migrations/tenant
     template: templates/schema.tmpl             # templ (required)
     mode: table                                 # templ: database/schema/script/table
     filename_pattern: "{{.Name}}.go"           # templ multi-output modes
-    output:                                    # convert / merge (required)
+    select:                                    # split (optional; default = keep everything)
+      schemas: [public]
+      tables: [users, orders]
+      exclude_schemas: []
+      exclude_tables: []
+      database_name: SubsetDB                  # optional rename of the output database
+    rules: .relspec-rules.yaml                 # inspect (optional; built-in defaults if omitted)
+    report:                                    # inspect (required) / diff (required)
+      format: json                             # inspect: markdown|json ; diff: summary|json|html
+      path: build/report.json                  # required, except a diff "summary" (goes to the log)
+      overwrite: false
+    output:                                    # convert / merge / split (required); scripts-exec (required, conn_env)
       format: pgsql
       path: build/schema.sql                   # file output, OR:
       conn_env: TARGET_DB_URL                  # execute against DB (pgsql only)
@@ -136,13 +188,15 @@ jobs:
       flatten_schema: false
       schema: public
       package: models                          # for gorm/bun output
-      continue_on_error: false                 # pgsql output
+      continue_on_error: false                 # pgsql / scripts-exec output
       skip_relations: false                    # merge only
       skip_enums: false
       skip_views: false
       skip_domains: false
       skip_sequences: false
-    logfile: .relspec/log/<job-name>.log       # optional; appended to
+    logfile: .relspec/log/<job-name>.log       # optional; appended to, size-rotated
+    log_max_size: 5MB                           # optional per-job override
+    log_keep: 3                                 # optional per-job override
 ```
 
 For `templ`, `inputs` use the same file or `pgsql`/`conn_env` source forms as
@@ -150,6 +204,11 @@ schema conversion. `output` is optional (empty means stdout); when present it
 contains only `path` and `overwrite`, because templates do not select a schema
 writer format.
 
+A `from_job` input takes no `path`, `format` or `conn_env`: it resolves to the
+named job's `output.path` and inherits its format, and implies a dependency on
+that job. The producer must be a `convert`, `merge` or `split` job writing a
+single-file output.
+
 ### Supported input formats
 
 `dbml`, `dctx`, `drawdb`, `graphql`, `json`, `yaml`, `gorm`, `bun`, `drizzle`,
@@ -230,3 +289,77 @@ jobs:
       path: snapshots/prod.dbml
       overwrite: true
 ```
+
+### Chain jobs with `from_job`, then lint the result
+
+```yaml
+version: 1
+jobs:
+  build-json:
+    command: convert
+    inputs:
+      - { path: schema/core.dbml, format: dbml }
+      - { path: schema/tenant.dbml, format: dbml }
+    output: { format: json, path: build/schema.json, overwrite: true }
+  lint-schema:
+    command: inspect
+    inputs:
+      - from_job: build-json      # implies depends_on: [build-json]
+    rules: .relspec-rules.yaml    # optional; built-in rules if omitted
+    report:
+      format: markdown
+      path: build/lint-report.md
+      overwrite: true
+```
+
+`relspec job run lint-schema` runs `build-json` first, then inspects its output.
+The job fails (exit 1) if any enforced rule is violated.
+
+### Split a subset out of a larger schema
+
+```yaml
+version: 1
+jobs:
+  posts-only:
+    command: split
+    inputs:
+      - { path: schema/core.dbml, format: dbml }
+      - { path: schema/tenant.dbml, format: dbml }
+    select:
+      tables: [posts]
+    output: { format: dbml, path: build/posts.dbml, overwrite: true }
+```
+
+### Diff two schemas
+
+```yaml
+version: 1
+jobs:
+  drift:
+    command: diff
+    inputs:                       # exactly two
+      - { path: build/schema.json, format: json }
+      - format: pgsql
+        conn_env: PROD_DB_URL
+    report:
+      format: summary             # summary → logfile; json/html need a path
+```
+
+`diff` reports differences and always exits 0.
+
+### Execute migration scripts against a live database
+
+```yaml
+version: 1
+jobs:
+  apply-migrations:
+    command: scripts-exec
+    script_dirs:
+      - migrations/core
+      - migrations/tenant
+    output:
+      conn_env: TARGET_DB_URL     # pgsql only; no path
+    options:
+      continue_on_error: false
+    logfile: .relspec/log/apply-migrations.log
+```

examples/jobs/relspec.yml

@@ -4,7 +4,14 @@
 #   relspec job list
 #   relspec job run build-schema --plan
 #   relspec job run build-schema
+#   relspec job run lint-schema        # inspect, consuming build-json's output
 version: 1
+
+# File-wide defaults. Individual jobs may override log_max_size / log_keep.
+defaults:
+  log_max_size: 2MB
+  log_keep: 5
+
 jobs:
   build-schema:
     command: convert
@@ -43,3 +50,30 @@ jobs:
       - migrations/core
       - migrations/tenant
     logfile: .relspec/log/migration-order.log
+
+  lint-schema:
+    command: inspect
+    description: Validate build-json's output against the built-in rules
+    # No depends_on needed: the from_job input implies a dependency on build-json.
+    inputs:
+      - from_job: build-json
+    report:
+      format: markdown
+      path: build/lint-report.md
+      overwrite: true
+    logfile: .relspec/log/lint-schema.log
+
+  posts-only:
+    command: split
+    description: Extract just the posts table into its own DBML file
+    inputs:
+      - path: schema/core.dbml
+        format: dbml
+      - path: schema/tenant.dbml
+        format: dbml
+    select:
+      tables: [posts]
+    output:
+      format: dbml
+      path: build/posts.dbml
+      overwrite: true

pkg/jobs/jobs.go

@@ -20,24 +20,52 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
+	"strconv"
 	"strings"
 
 	"gopkg.in/yaml.v3"
 )
 
-// SchemaVersion is the only job-file schema version this build understands.
-const SchemaVersion = 1
+// CurrentSchemaVersion is the highest job-file schema version this build was
+// written for. MinSchemaVersion is the oldest it still accepts. A file that
+// declares a version in between loads normally; a newer version loads
+// best-effort with a warning (see Load); an older-than-minimum version is a
+// hard error.
+const (
+	CurrentSchemaVersion = 1
+	MinSchemaVersion     = 1
+)
+
+// Built-in logfile rotation policy, used when neither the job nor its file's
+// defaults block sets one.
+const (
+	defaultLogMaxSizeBytes int64 = 5 << 20 // 5 MiB
+	defaultLogKeep               = 3
+)
 
 // Command names are a closed allow-list. Arbitrary strings are rejected.
 const (
 	CommandConvert     = "convert"      // read one or more schema files, optionally merge, write one output
 	CommandMerge       = "merge"        // additive merge of two or more schema files into one output
 	CommandScriptsList = "scripts-list" // deterministically list SQL scripts across one or more directories
+	CommandScriptsExec = "scripts-exec" // execute SQL scripts across one or more directories against a live database
 	CommandTempl       = "templ"        // apply a custom Go text template to one or more schemas
+	CommandSplit       = "split"        // extract selected schemas/tables into a separate output
+	CommandInspect     = "inspect"      // validate one or more schemas against rules and write a report
+	CommandDiff        = "diff"         // compare exactly two schemas and write a differences report
 )
 
 // SupportedCommands lists every accepted command, in help order.
-var SupportedCommands = []string{CommandConvert, CommandMerge, CommandScriptsList, CommandTempl}
+var SupportedCommands = []string{
+	CommandConvert, CommandMerge, CommandScriptsList, CommandScriptsExec,
+	CommandTempl, CommandSplit, CommandInspect, CommandDiff,
+}
+
+// producerCommands are commands whose output is a schema file that another job
+// may consume via from_job.
+var producerCommands = map[string]bool{
+	CommandConvert: true, CommandMerge: true, CommandSplit: true,
+}
 
 // readerFormats are the file-based input formats a job may declare (path).
 var readerFormats = map[string]bool{
@@ -61,17 +89,51 @@ var writerFormats = map[string]bool{
 // live database) is supported instead of writing a file.
 var execOutputFormats = map[string]bool{"pgsql": true}
 
+// singleFileFormats are output formats that emit exactly one file (as opposed
+// to a directory of files). Only these are eligible for atomic temp+rename
+// writes and for being consumed by another job via from_job.
+var singleFileFormats = map[string]bool{
+	"json": true, "yaml": true, "dbml": true, "dctx": true, "drawdb": true,
+	"graphql": true, "pgsql": true, "mssql": true, "sqlite": true,
+}
+
+// SingleFileOutputFormat reports whether format writes exactly one file.
+func SingleFileOutputFormat(format string) bool {
+	return singleFileFormats[strings.ToLower(format)]
+}
+
+// diffReportFormats and inspectReportFormats are the report.format values
+// accepted by the diff and inspect commands respectively.
+var (
+	diffReportFormats    = map[string]bool{"summary": true, "json": true, "html": true}
+	inspectReportFormats = map[string]bool{"markdown": true, "json": true}
+)
+
 // File is the on-disk shape of a single job file.
 type File struct {
 	Version  int             `yaml:"version"`
+	Defaults *Defaults       `yaml:"defaults"`
 	Jobs     map[string]*Job `yaml:"jobs"`
 }
 
+// Defaults carries file-wide settings that individual jobs may override.
+type Defaults struct {
+	// LogMaxSize is a human-readable size ("5MB", "512KB", "1GB"). Empty
+	// means "use the built-in default".
+	LogMaxSize string `yaml:"log_max_size"`
+	// LogKeep is how many rotated logfiles to retain. Zero means "use the
+	// built-in default".
+	LogKeep int `yaml:"log_keep"`
+}
+
 // Job is one named job within a job file.
 type Job struct {
 	// Name and SourceFile are populated by Load, not parsed from YAML.
 	Name       string `yaml:"-"`
 	SourceFile string `yaml:"-"`
+	// fileDefaults is the Defaults block of the file that declared this job,
+	// captured by Load. nil when the file had none.
+	fileDefaults *Defaults `yaml:"-"`
 
 	Command         string   `yaml:"command"`
 	Description     string   `yaml:"description"`
@@ -82,8 +144,13 @@ type Job struct {
 	Mode            string   `yaml:"mode"`
 	FilenamePattern string   `yaml:"filename_pattern"`
 	Output          *Output  `yaml:"output"`
+	Rules           string   `yaml:"rules"`
+	Report          *Report  `yaml:"report"`
+	Select          *Select  `yaml:"select"`
 	Options         Options  `yaml:"options"`
 	Logfile         string   `yaml:"logfile"`
+	LogMaxSize      string   `yaml:"log_max_size"`
+	LogKeep         *int     `yaml:"log_keep"`
 }
 
 // Input is one declared input schema.
@@ -94,6 +161,10 @@ type Input struct {
 	// ConnEnv is the NAME of an environment variable holding a connection
 	// string, used with database formats. The value is never stored here.
 	ConnEnv string `yaml:"conn_env"`
+	// FromJob names another job in the set whose file output is used as this
+	// input. It implies a dependency on that job. Path/Format/ConnEnv must be
+	// empty when FromJob is set; the format is inherited from the producer.
+	FromJob string `yaml:"from_job"`
 }
 
 // Output is the declared output target.
@@ -104,6 +175,105 @@ type Output struct {
 	Overwrite bool   `yaml:"overwrite"`
 }
 
+// Report is the output target for the inspect and diff commands.
+type Report struct {
+	// Format is the report format: diff accepts summary|json|html, inspect
+	// accepts markdown|json. Empty means the command's default.
+	Format    string `yaml:"format"`
+	Path      string `yaml:"path"`
+	Overwrite bool   `yaml:"overwrite"`
+}
+
+// Select carries the schema/table selection for the split command.
+type Select struct {
+	Schemas        []string `yaml:"schemas"`
+	Tables         []string `yaml:"tables"`
+	ExcludeSchemas []string `yaml:"exclude_schemas"`
+	ExcludeTables  []string `yaml:"exclude_tables"`
+	DatabaseName   string   `yaml:"database_name"`
+}
+
+// LogPolicy is the resolved logfile rotation policy for a job.
+type LogPolicy struct {
+	MaxSizeBytes int64
+	Keep         int
+}
+
+// ResolvedLogPolicy returns the effective rotation policy: the job's own
+// overrides win, then its file's defaults block, then the built-in default.
+func (j *Job) ResolvedLogPolicy() LogPolicy {
+	p := LogPolicy{MaxSizeBytes: defaultLogMaxSizeBytes, Keep: defaultLogKeep}
+
+	if j.fileDefaults != nil {
+		if n, err := parseHumanSize(j.fileDefaults.LogMaxSize); err == nil && n > 0 {
+			p.MaxSizeBytes = n
+		}
+		if j.fileDefaults.LogKeep > 0 {
+			p.Keep = j.fileDefaults.LogKeep
+		}
+	}
+	if n, err := parseHumanSize(j.LogMaxSize); err == nil && n > 0 {
+		p.MaxSizeBytes = n
+	}
+	if j.LogKeep != nil && *j.LogKeep >= 0 {
+		p.Keep = *j.LogKeep
+	}
+	return p
+}
+
+// effectiveDeps returns the union of explicit depends_on entries and the jobs
+// referenced by from_job inputs, deduplicated in stable order.
+func (j *Job) effectiveDeps() []string {
+	seen := map[string]bool{}
+	var deps []string
+	add := func(name string) {
+		if name == "" || name == j.Name || seen[name] {
+			return
+		}
+		seen[name] = true
+		deps = append(deps, name)
+	}
+	for _, d := range j.DependsOn {
+		add(d)
+	}
+	for _, in := range j.Inputs {
+		add(in.FromJob)
+	}
+	return deps
+}
+
+// parseHumanSize parses a byte size such as "5MB", "512 KB", "1gb" or a bare
+// byte count. An empty string returns (0, nil) so callers can fall back.
+func parseHumanSize(s string) (int64, error) {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return 0, nil
+	}
+	upper := strings.ToUpper(s)
+	mult := int64(1)
+	// Check multi-character suffixes before the bare "B".
+	for _, u := range []struct {
+		suffix string
+		m      int64
+	}{
+		{"KB", 1 << 10}, {"MB", 1 << 20}, {"GB", 1 << 30}, {"B", 1},
+	} {
+		if strings.HasSuffix(upper, u.suffix) {
+			mult = u.m
+			upper = strings.TrimSpace(strings.TrimSuffix(upper, u.suffix))
+			break
+		}
+	}
+	n, err := strconv.ParseFloat(upper, 64)
+	if err != nil {
+		return 0, fmt.Errorf("invalid size %q", s)
+	}
+	if n < 0 {
+		return 0, fmt.Errorf("negative size %q", s)
+	}
+	return int64(n * float64(mult)), nil
+}
+
 // Options carries the subset of command flags a job file may set.
 type Options struct {
 	FlattenSchema   bool   `yaml:"flatten_schema"`
@@ -127,6 +297,9 @@ type Set struct {
 	Files []string
 	// Jobs is keyed by job name.
 	Jobs map[string]*Job
+	// Warnings holds non-fatal load-time messages (e.g. a newer-than-known
+	// schema version). Callers should surface these to the user.
+	Warnings []string
 }
 
 // Names returns all job names in deterministic (sorted) order.
@@ -199,15 +372,35 @@ func Load(paths []string) (*Set, error) {
 		if err != nil {
 			return nil, fmt.Errorf("failed to read job file %q: %w", path, err)
 		}
+
+		// Peek at the version first so a newer file can be parsed leniently
+		// (unknown fields ignored) instead of failing outright.
+		var probe struct {
+			Version int `yaml:"version"`
+		}
+		if err := yaml.Unmarshal(data, &probe); err != nil {
+			return nil, fmt.Errorf("invalid job file %q: %w", path, err)
+		}
+		version := probe.Version
+		if version == 0 {
+			version = CurrentSchemaVersion
+		}
+		if version < MinSchemaVersion {
+			return nil, fmt.Errorf("job file %q: unsupported version %d (this build accepts %d or newer)", path, version, MinSchemaVersion)
+		}
+		strict := version <= CurrentSchemaVersion
+		if !strict {
+			set.Warnings = append(set.Warnings, fmt.Sprintf(
+				"job file %q declares version %d, newer than this build understands (%d); loading best-effort and ignoring unknown fields",
+				path, version, CurrentSchemaVersion))
+		}
+
 		dec := yaml.NewDecoder(strings.NewReader(string(data)))
-		dec.KnownFields(true)
+		dec.KnownFields(strict)
 		var f File
 		if err := dec.Decode(&f); err != nil {
 			return nil, fmt.Errorf("invalid job file %q: %w", path, err)
 		}
-		if f.Version != SchemaVersion {
-			return nil, fmt.Errorf("job file %q: unsupported version %d (expected %d)", path, f.Version, SchemaVersion)
-		}
 		if len(f.Jobs) == 0 {
 			return nil, fmt.Errorf("job file %q: no jobs defined", path)
 		}
@@ -220,6 +413,7 @@ func Load(paths []string) (*Set, error) {
 			}
 			job.Name = name
 			job.SourceFile = path
+			job.fileDefaults = f.Defaults
 			origin[name] = path
 			set.Jobs[name] = job
 		}
@@ -240,13 +434,30 @@ func (s *Set) Validate() error {
 			errs = append(errs, fmt.Sprintf("job %q: %s", name, msg))
 		}
 	}
-	// Dependency references + cycles.
+	// Dependency references + cycles + from_job wiring.
 	for _, name := range s.Names() {
-		for _, dep := range s.Jobs[name].DependsOn {
+		j := s.Jobs[name]
+		for _, dep := range j.DependsOn {
 			if _, ok := s.Jobs[dep]; !ok {
 				errs = append(errs, fmt.Sprintf("job %q: depends_on unknown job %q", name, dep))
 			}
 		}
+		for i, in := range j.Inputs {
+			if in.FromJob == "" {
+				continue
+			}
+			producer, ok := s.Jobs[in.FromJob]
+			if !ok {
+				errs = append(errs, fmt.Sprintf("job %q: input[%d] from_job references unknown job %q", name, i, in.FromJob))
+				continue
+			}
+			if !producerCommands[producer.Command] || producer.Output == nil ||
+				producer.Output.Path == "" || !SingleFileOutputFormat(producer.Output.Format) {
+				errs = append(errs, fmt.Sprintf(
+					"job %q: input[%d] from_job %q must name a convert/merge/split job that writes a single-file output",
+					name, i, in.FromJob))
+			}
+		}
 	}
 	if cycle := s.findCycle(); cycle != "" {
 		errs = append(errs, fmt.Sprintf("dependency cycle detected: %s", cycle))
@@ -262,7 +473,8 @@ func (j *Job) validate() []string {
 	var e []string
 
 	switch j.Command {
-	case CommandConvert, CommandMerge, CommandScriptsList, CommandTempl:
+	case CommandConvert, CommandMerge, CommandScriptsList, CommandScriptsExec,
+		CommandTempl, CommandSplit, CommandInspect, CommandDiff:
 	case "":
 		e = append(e, "missing command")
 		return e
@@ -282,6 +494,7 @@ func (j *Job) validate() []string {
 	}
 	checkPath("logfile", j.Logfile)
 	checkPath("template", j.Template)
+	checkPath("rules", j.Rules)
 	for _, in := range j.Inputs {
 		checkPath("input path", in.Path)
 	}
@@ -291,6 +504,13 @@ func (j *Job) validate() []string {
 	if j.Output != nil {
 		checkPath("output path", j.Output.Path)
 	}
+	if j.Report != nil {
+		checkPath("report path", j.Report.Path)
+	}
+
+	if _, err := parseHumanSize(j.LogMaxSize); err != nil {
+		e = append(e, fmt.Sprintf("log_max_size: %v", err))
+	}
 
 	switch j.Command {
 	case CommandConvert, CommandMerge:
@@ -350,11 +570,113 @@ func (j *Job) validate() []string {
 		if j.Output != nil && j.Output.Format != "" {
 			e = append(e, "output.format is not valid for command \"templ\"")
 		}
+	case CommandSplit:
+		if len(j.Inputs) < 1 {
+			e = append(e, "command \"split\" requires at least 1 input")
+		}
+		for i, in := range j.Inputs {
+			e = append(e, validateInput(i, in)...)
+		}
+		if len(j.ScriptDirs) > 0 {
+			e = append(e, "script_dirs is not valid for command \"split\"")
+		}
+		if j.Report != nil {
+			e = append(e, "report is not valid for command \"split\" (use output)")
+		}
+		if j.Output == nil {
+			e = append(e, "missing output")
+		} else {
+			if j.Output.ConnEnv != "" {
+				e = append(e, "command \"split\" writes a file; output.conn_env is not supported")
+			}
+			e = append(e, validateOutput(*j.Output)...)
+		}
+	case CommandInspect:
+		if len(j.Inputs) < 1 {
+			e = append(e, "command \"inspect\" requires at least 1 input")
+		}
+		for i, in := range j.Inputs {
+			e = append(e, validateInput(i, in)...)
+		}
+		if len(j.ScriptDirs) > 0 {
+			e = append(e, "script_dirs is not valid for command \"inspect\"")
+		}
+		if j.Output != nil {
+			e = append(e, "output is not valid for command \"inspect\" (use report)")
+		}
+		e = append(e, validateReport(j.Report, "inspect", inspectReportFormats, "markdown")...)
+	case CommandDiff:
+		if len(j.Inputs) != 2 {
+			e = append(e, "command \"diff\" requires exactly 2 inputs (source, target)")
+		}
+		for i, in := range j.Inputs {
+			e = append(e, validateInput(i, in)...)
+		}
+		if len(j.ScriptDirs) > 0 {
+			e = append(e, "script_dirs is not valid for command \"diff\"")
+		}
+		if j.Output != nil {
+			e = append(e, "output is not valid for command \"diff\" (use report)")
+		}
+		e = append(e, validateReport(j.Report, "diff", diffReportFormats, "summary")...)
+	case CommandScriptsExec:
+		if len(j.ScriptDirs) == 0 {
+			e = append(e, "command \"scripts-exec\" requires at least one script_dir")
+		}
+		if len(j.Inputs) > 0 {
+			e = append(e, "inputs is not valid for command \"scripts-exec\"")
+		}
+		if j.Report != nil {
+			e = append(e, "report is not valid for command \"scripts-exec\"")
+		}
+		if j.Output == nil || j.Output.ConnEnv == "" {
+			e = append(e, "command \"scripts-exec\" requires output.conn_env (an environment variable name holding a connection string)")
+		} else {
+			if j.Output.Path != "" {
+				e = append(e, "command \"scripts-exec\" executes against a database; output.path is not supported")
+			}
+			f := strings.ToLower(j.Output.Format)
+			if f != "" && f != "pgsql" {
+				e = append(e, fmt.Sprintf("command \"scripts-exec\" only supports pgsql databases (got %q)", j.Output.Format))
+			}
+			if looksLikeSecret(j.Output.ConnEnv) {
+				e = append(e, "output: conn_env must be an environment variable name, not a connection string")
+			}
+		}
+	}
+	return e
+}
+
+// validateReport checks a Report block for the inspect/diff commands.
+func validateReport(r *Report, cmd string, allowed map[string]bool, defFmt string) []string {
+	if r == nil {
+		return []string{fmt.Sprintf("command %q requires a report block", cmd)}
+	}
+	var e []string
+	f := strings.ToLower(r.Format)
+	if f == "" {
+		f = defFmt
+	}
+	if !allowed[f] {
+		names := make([]string, 0, len(allowed))
+		for k := range allowed {
+			names = append(names, k)
+		}
+		sort.Strings(names)
+		e = append(e, fmt.Sprintf("command %q report.format %q is not supported (use: %s)", cmd, r.Format, strings.Join(names, ", ")))
+	}
+	// A diff summary may be written to the log; everything else needs a path.
+	summaryToLog := cmd == "diff" && f == "summary"
+	if r.Path == "" && !summaryToLog {
+		e = append(e, fmt.Sprintf("command %q requires report.path", cmd))
 	}
 	return e
 }
 
 func validateTemplInput(i int, in Input) []string {
+	if in.FromJob != "" {
+		return fromJobInputShape(i, in)
+	}
 	var e []string
 	if in.Format == "" {
 		return []string{fmt.Sprintf("input[%d]: missing format", i)}
@@ -383,7 +705,27 @@ func validateTemplInput(i int, in Input) []string {
 	return e
 }
 
+// fromJobInputShape checks the structural rules for an input that pulls its
+// schema from another job's output. The referenced job's existence and kind
+// are checked in Set.Validate, which can see the whole set.
+func fromJobInputShape(i int, in Input) []string {
+	var e []string
+	if in.Path != "" {
+		e = append(e, fmt.Sprintf("input[%d]: from_job takes no path", i))
+	}
+	if in.Format != "" {
+		e = append(e, fmt.Sprintf("input[%d]: from_job inherits the producer's format; drop format", i))
+	}
+	if in.ConnEnv != "" {
+		e = append(e, fmt.Sprintf("input[%d]: from_job takes no conn_env", i))
+	}
+	return e
+}
+
 func validateInput(i int, in Input) []string {
+	if in.FromJob != "" {
+		return fromJobInputShape(i, in)
+	}
 	var e []string
 	if in.Format == "" {
 		e = append(e, fmt.Sprintf("input[%d]: missing format", i))
@@ -487,9 +829,42 @@ func SafeJoin(root, rel string) (string, error) {
 	if rp == ".." || strings.HasPrefix(rp, ".."+string(filepath.Separator)) {
 		return "", fmt.Errorf("path %q escapes the job file directory", rel)
 	}
+	// Symlink hardening: resolve symlinks on the root and on the deepest
+	// existing ancestor of the target, and require the target to still live
+	// inside the resolved root. This catches a symlink inside the job-file
+	// directory that points outside it.
+	realRoot, err := filepath.EvalSymlinks(absRoot)
+	if err != nil {
+		return "", fmt.Errorf("cannot resolve job file directory: %w", err)
+	}
+	realAnc, err := filepath.EvalSymlinks(deepestExistingAncestor(joined))
+	if err != nil {
+		return "", fmt.Errorf("cannot resolve path %q: %w", rel, err)
+	}
+	if realAnc != realRoot {
+		if r, err := filepath.Rel(realRoot, realAnc); err != nil ||
+			r == ".." || strings.HasPrefix(r, ".."+string(filepath.Separator)) {
+			return "", fmt.Errorf("path %q resolves outside the job file directory via a symlink", rel)
+		}
+	}
 	return joined, nil
 }
 
+// deepestExistingAncestor returns p itself if it exists, otherwise the nearest
+// existing parent directory (falling back to the filesystem root).
+func deepestExistingAncestor(p string) string {
+	for {
+		if _, err := os.Lstat(p); err == nil {
+			return p
+		}
+		parent := filepath.Dir(p)
+		if parent == p {
+			return p
+		}
+		p = parent
+	}
+}
+
 // Plan returns the jobs to execute for name in dependency order. When
 // includeDeps is false only the named job is returned (its declared
 // dependencies are still validated to exist and be acyclic by Validate).
@@ -514,7 +889,7 @@ func (s *Set) Plan(name string, includeDeps bool) ([]*Job, error) {
 		}
 		inProgress[n] = true
 		j := s.Jobs[n]
-		deps := append([]string(nil), j.DependsOn...)
+		deps := j.effectiveDeps()
 		sort.Strings(deps)
 		for _, d := range deps {
 			if _, ok := s.Jobs[d]; !ok {
@@ -543,7 +918,7 @@ func (s *Set) findCycle() string {
 	dfs = func(n string) []string {
 		color[n] = 1
 		stack = append(stack, n)
-		deps := append([]string(nil), s.Jobs[n].DependsOn...)
+		deps := s.Jobs[n].effectiveDeps()
 		sort.Strings(deps)
 		for _, d := range deps {
 			if _, ok := s.Jobs[d]; !ok {

pkg/jobs/jobs_test.go

@@ -56,13 +56,82 @@ func TestLoadRejectsUnknownFields(t *testing.T) {
 	}
 }
 
-func TestLoadRejectsBadVersion(t *testing.T) {
+func TestLoadWarnsOnNewerVersion(t *testing.T) {
 	dir := t.TempDir()
 	p := filepath.Join(dir, "relspec.yml")
-	write(t, p, "version: 2\njobs:\n  a:\n    command: convert\n")
-	_, err := Load([]string{p})
-	if err == nil || !strings.Contains(err.Error(), "unsupported version") {
-		t.Fatalf("expected unsupported version error, got %v", err)
+	// A newer version loads best-effort with a warning, and unknown fields
+	// from the newer schema are ignored rather than rejected.
+	write(t, p, "version: 99\njobs:\n  a:\n    command: convert\n"+
+		"    inputs:\n      - path: a.dbml\n        format: dbml\n"+
+		"    output:\n      format: json\n      path: out.json\n"+
+		"    future_field: whatever\n")
+	set, err := Load([]string{p})
+	if err != nil {
+		t.Fatalf("newer version should load, got %v", err)
+	}
+	if len(set.Warnings) == 0 {
+		t.Fatal("expected a warning about the newer version")
+	}
+	if err := set.Validate(); err != nil {
+		t.Fatalf("validate: %v", err)
+	}
+}
+
+func TestLoadAcceptsOmittedVersion(t *testing.T) {
+	dir := t.TempDir()
+	p := filepath.Join(dir, "relspec.yml")
+	write(t, p, "jobs:\n  a:\n    command: convert\n"+
+		"    inputs:\n      - path: a.dbml\n        format: dbml\n"+
+		"    output:\n      format: json\n      path: out.json\n")
+	set, err := Load([]string{p})
+	if err != nil {
+		t.Fatalf("omitted version should load, got %v", err)
+	}
+	if len(set.Warnings) != 0 {
+		t.Fatalf("omitted version should not warn, got %v", set.Warnings)
+	}
+}
+
+func TestLoadStillRejectsUnknownFieldsAtCurrentVersion(t *testing.T) {
+	dir := t.TempDir()
+	p := filepath.Join(dir, "relspec.yml")
+	write(t, p, "version: 1\njobs:\n  a:\n    command: convert\n    bogus: true\n")
+	if _, err := Load([]string{p}); err == nil {
+		t.Fatal("expected unknown-field rejection at the current version")
+	}
+}
+
+func TestParseHumanSize(t *testing.T) {
+	cases := []struct {
+		in   string
+		want int64
+		bad  bool
+	}{
+		{"", 0, false},
+		{"512", 512, false},
+		{"512B", 512, false},
+		{"1KB", 1 << 10, false},
+		{"5MB", 5 << 20, false},
+		{"1gb", 1 << 30, false},
+		{" 2 MB ", 2 << 20, false},
+		{"nonsense", 0, true},
+		{"-1MB", 0, true},
+	}
+	for _, c := range cases {
+		got, err := parseHumanSize(c.in)
+		if c.bad {
+			if err == nil {
+				t.Errorf("parseHumanSize(%q): expected error", c.in)
+			}
+			continue
+		}
+		if err != nil {
+			t.Errorf("parseHumanSize(%q): %v", c.in, err)
+			continue
+		}
+		if got != c.want {
+			t.Errorf("parseHumanSize(%q) = %d, want %d", c.in, got, c.want)
+		}
 	}
 }
 
@@ -239,6 +308,135 @@ func TestShippedExampleIsValid(t *testing.T) {
 	}
 }
 
+func TestFromJobWiring(t *testing.T) {
+	content := "version: 1\njobs:\n" +
+		"  producer:\n    command: convert\n" +
+		"    inputs:\n      - path: a.dbml\n        format: dbml\n" +
+		"    output:\n      format: json\n      path: build/schema.json\n" +
+		"  consumer:\n    command: convert\n" +
+		"    inputs:\n      - from_job: producer\n" +
+		"    output:\n      format: yaml\n      path: build/schema.yaml\n"
+	set := loadOne(t, content)
+	if err := set.Validate(); err != nil {
+		t.Fatalf("validate: %v", err)
+	}
+	plan, err := set.Plan("consumer", true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(plan) != 2 || plan[0].Name != "producer" || plan[1].Name != "consumer" {
+		t.Fatalf("plan = %v, want [producer consumer]", plan)
+	}
+}
+
+func TestFromJobRejectsNonProducer(t *testing.T) {
+	content := "version: 1\njobs:\n" +
+		"  lister:\n    command: scripts-list\n    script_dirs: [migrations]\n" +
+		"  consumer:\n    command: convert\n" +
+		"    inputs:\n      - from_job: lister\n" +
+		"    output:\n      format: yaml\n      path: out.yaml\n"
+	set := loadOne(t, content)
+	if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "from_job") {
+		t.Fatalf("want from_job producer error, got %v", err)
+	}
+}
+
+func TestFromJobRejectsUnknownJob(t *testing.T) {
+	content := "version: 1\njobs:\n" +
+		"  consumer:\n    command: convert\n" +
+		"    inputs:\n      - from_job: ghost\n" +
+		"    output:\n      format: yaml\n      path: out.yaml\n"
+	set := loadOne(t, content)
+	if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "unknown job") {
+		t.Fatalf("want unknown job error, got %v", err)
+	}
+}
+
+func TestFromJobCycleDetected(t *testing.T) {
+	content := "version: 1\njobs:\n" +
+		"  a:\n    command: convert\n" +
+		"    inputs:\n      - from_job: b\n" +
+		"    output:\n      format: json\n      path: a.json\n" +
+		"  b:\n    command: convert\n" +
+		"    inputs:\n      - from_job: a\n" +
+		"    output:\n      format: json\n      path: b.json\n"
+	set := loadOne(t, content)
+	if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "cycle") {
+		t.Fatalf("want cycle error, got %v", err)
+	}
+}
+
+func TestSplitJobValidation(t *testing.T) {
+	set := loadOne(t, "version: 1\njobs:\n  s:\n    command: split\n"+
+		"    inputs:\n      - path: a.dbml\n        format: dbml\n")
+	if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "missing output") {
+		t.Fatalf("want missing output, got %v", err)
+	}
+	set = loadOne(t, "version: 1\njobs:\n  s:\n    command: split\n"+
+		"    inputs:\n      - path: a.dbml\n        format: dbml\n"+
+		"    select:\n      tables: [users]\n"+
+		"    output:\n      format: json\n      path: out.json\n")
+	if err := set.Validate(); err != nil {
+		t.Fatalf("expected valid split job, got %v", err)
+	}
+}
+
+func TestInspectJobValidation(t *testing.T) {
+	set := loadOne(t, "version: 1\njobs:\n  i:\n    command: inspect\n"+
+		"    inputs:\n      - path: a.dbml\n        format: dbml\n")
+	if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "report") {
+		t.Fatalf("want report required, got %v", err)
+	}
+	set = loadOne(t, "version: 1\njobs:\n  i:\n    command: inspect\n"+
+		"    inputs:\n      - path: a.dbml\n        format: dbml\n"+
+		"    report:\n      format: json\n      path: build/report.json\n")
+	if err := set.Validate(); err != nil {
+		t.Fatalf("expected valid inspect job, got %v", err)
+	}
+}
+
+func TestDiffJobValidation(t *testing.T) {
+	set := loadOne(t, "version: 1\njobs:\n  d:\n    command: diff\n"+
+		"    inputs:\n      - path: a.dbml\n        format: dbml\n"+
+		"    report:\n      format: summary\n")
+	if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "exactly 2 inputs") {
+		t.Fatalf("want exactly 2 inputs, got %v", err)
+	}
+	set = loadOne(t, "version: 1\njobs:\n  d:\n    command: diff\n"+
+		"    inputs:\n      - path: a.dbml\n        format: dbml\n"+
+		"      - path: b.dbml\n        format: dbml\n"+
+		"    report:\n      format: summary\n")
+	if err := set.Validate(); err != nil {
+		t.Fatalf("expected valid diff job, got %v", err)
+	}
+}
+
+func TestScriptsExecValidation(t *testing.T) {
+	set := loadOne(t, "version: 1\njobs:\n  x:\n    command: scripts-exec\n"+
+		"    script_dirs: [migrations]\n")
+	if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "conn_env") {
+		t.Fatalf("want output.conn_env required, got %v", err)
+	}
+	set = loadOne(t, "version: 1\njobs:\n  x:\n    command: scripts-exec\n"+
+		"    script_dirs: [migrations]\n"+
+		"    output:\n      conn_env: TARGET_DB_URL\n")
+	if err := set.Validate(); err != nil {
+		t.Fatalf("expected valid scripts-exec job, got %v", err)
+	}
+}
+
+func TestSafeJoinRejectsSymlinkEscape(t *testing.T) {
+	root := t.TempDir()
+	outside := t.TempDir()
+	link := filepath.Join(root, "link")
+	if err := os.Symlink(outside, link); err != nil {
+		t.Skipf("symlink not supported: %v", err)
+	}
+	if _, err := SafeJoin(root, "link/x.sql"); err == nil {
+		t.Fatal("expected rejection of a path escaping via a symlink")
+	}
+}
+
 func TestScriptsListValidation(t *testing.T) {
 	set := loadOne(t, "version: 1\njobs:\n  s:\n    command: scripts-list\n")
 	if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "script_dir") {