Add local source publication

2026-05-31 02:21:36 +00:00
parent aeee91940d
commit e296361042
17 changed files with 1019 additions and 84 deletions
--- a/internal/app/run.go
+++ b/internal/app/run.go
@@ -5,7 +5,10 @@ import (
 	"fmt"
 	"io"

+	"gitea.maximumdirect.net/eric/distributor/internal/adapters/local"
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
 	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/publish"
 )

 type RunOptions struct {
@@ -18,9 +21,6 @@ func Run(ctx context.Context, options RunOptions) error {
 	if err := ctx.Err(); err != nil {
 		return err
 	}
-	if !options.DryRun {
-		return fmt.Errorf("run command: %w", ErrNotImplemented)
-	}

 	configPath := options.ConfigPath
 	if configPath == "" {
@@ -30,25 +30,77 @@ func Run(ctx context.Context, options RunOptions) error {
 	if err != nil {
 		return err
 	}
-	return writeRunSummary(options.Stdout, cfg)
+	return runConfig(ctx, cfg, options)
 }

-func writeRunSummary(w io.Writer, cfg config.Config) error {
-	if w == nil {
-		return nil
-	}
-	if _, err := fmt.Fprintf(w, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil {
-		return err
-	}
-	for _, pipeline := range cfg.Pipelines {
-		if _, err := fmt.Fprintf(w, "- %s: source=%s destinations=%d\n", pipeline.ID, pipeline.Source.Backend, len(pipeline.Destinations)); err != nil {
+func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error {
+	if options.Stdout != nil {
+		if _, err := fmt.Fprintf(options.Stdout, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil {
 			return err
 		}
-		for _, destination := range pipeline.Destinations {
-			if _, err := fmt.Fprintf(w, "  - %s: backend=%s publish_source=%t publish_html=%t\n", destination.ID, destination.Backend, destination.Publish.Source, destination.Publish.HTML); err != nil {
+	}
+	for _, pipeline := range cfg.Pipelines {
+		if pipeline.Source.Backend != config.BackendLocal {
+			return fmt.Errorf("pipeline %s source backend %s is not implemented for execution", pipeline.ID, pipeline.Source.Backend)
+		}
+		sourceBackend, err := local.New(pipeline.Source.Path)
+		if err != nil {
+			return err
+		}
+		bundles, err := bundle.Discover(ctx, sourceBackend, "")
+		if err != nil {
+			return fmt.Errorf("pipeline %s discover source bundles: %w", pipeline.ID, err)
+		}
+		if options.Stdout != nil {
+			if _, err := fmt.Fprintf(options.Stdout, "- %s: source=local bundles=%d destinations=%d\n", pipeline.ID, len(bundles), len(pipeline.Destinations)); err != nil {
 				return err
 			}
 		}
+		for _, sourceBundle := range bundles {
+			for _, destination := range pipeline.Destinations {
+				if destination.Backend != config.BackendLocal {
+					return fmt.Errorf("pipeline %s destination %s backend %s is not implemented for execution", pipeline.ID, destination.ID, destination.Backend)
+				}
+				destinationBackend, err := local.New(destination.Path)
+				if err != nil {
+					return err
+				}
+				req := publish.Request{
+					PipelineID:            pipeline.ID,
+					DestinationID:         destination.ID,
+					SourceBundle:          sourceBundle,
+					SourceBackend:         sourceBackend,
+					DestinationBackend:    destinationBackend,
+					DestinationBundlePath: sourceBundle.RootRelativePath,
+					Publish:               *destination.Publish,
+					Transfer:              destination.Transfer,
+					DistributorVersion:    Version,
+				}
+				plan, err := publish.Build(ctx, req)
+				if options.Stdout != nil {
+					writePlanLine(options.Stdout, plan, err)
+				}
+				if err != nil {
+					return fmt.Errorf("pipeline %s destination %s bundle %s: %w", pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err)
+				}
+				if !options.DryRun {
+					if err := publish.Execute(ctx, req, plan); err != nil {
+						return fmt.Errorf("pipeline %s destination %s bundle %s: %w", pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err)
+					}
+				}
+			}
+		}
 	}
 	return nil
 }
+
+func writePlanLine(w io.Writer, plan publish.Plan, planErr error) {
+	if w == nil {
+		return
+	}
+	if planErr != nil {
+		fmt.Fprintf(w, "  - bundle=%s destination=%s action=error reason=%q\n", displayBundlePath(plan.BundlePath), plan.DestinationID, planErr.Error())
+		return
+	}
+	fmt.Fprintf(w, "  - bundle=%s destination=%s action=%s outputs=%d reason=%q\n", displayBundlePath(plan.BundlePath), plan.DestinationID, plan.Action, len(plan.Outputs), plan.Reason)
+}
--- a/internal/app/run_test.go
+++ b/internal/app/run_test.go
@@ -3,31 +3,25 @@ package app
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"os"
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/state"
 )

 func TestRunDryRunPrintsConfigSummary(t *testing.T) {
-	configPath := filepath.Join(t.TempDir(), "config.yml")
-	err := os.WriteFile(configPath, []byte(`
-pipelines:
-  - id: reports
-    source:
-      backend: local
-      path: /source
-    destinations:
-      - id: archive
-        backend: local
-        path: /archive
-`), 0o600)
-	if err != nil {
-		t.Fatalf("write config: %v", err)
-	}
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+	configPath := writeLocalConfig(t, sourceRoot, destinationRoot)

 	var stdout bytes.Buffer
-	err = Run(context.Background(), RunOptions{
+	err := Run(context.Background(), RunOptions{
 		ConfigPath: configPath,
 		DryRun:     true,
 		Stdout:     &stdout,
@@ -39,8 +33,8 @@ pipelines:
 	output := stdout.String()
 	for _, want := range []string{
 		"Configured pipelines: 1",
-		"- reports: source=local destinations=1",
-		"archive: backend=local publish_source=true publish_html=false",
+		"- reports: source=local bundles=1 destinations=1",
+		"bundle=. destination=archive action=publish_new",
 	} {
 		if !strings.Contains(output, want) {
 			t.Fatalf("Run() output = %q, want substring %q", output, want)
@@ -48,9 +42,302 @@ pipelines:
 	}
 }

-func TestRunWithoutDryRunIsNotImplemented(t *testing.T) {
-	err := Run(context.Background(), RunOptions{})
-	if err == nil || !strings.Contains(err.Error(), "not implemented") {
-		t.Fatalf("Run() error = %v, want not implemented", err)
+func TestRunPublishesNewLocalBundle(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+
+	var stdout bytes.Buffer
+	err := Run(context.Background(), RunOptions{
+		ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot),
+		Stdout:     &stdout,
+	})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
+	assertFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n")
+	if _, err := os.Stat(filepath.Join(destinationRoot, "manifest.json")); !os.IsNotExist(err) {
+		t.Fatalf("destination manifest stat error = %v, want not exist", err)
+	}
+	destinationState := readStateFile(t, filepath.Join(destinationRoot, ".distributor.json"))
+	if destinationState.PipelineID != "reports" || destinationState.DestinationID != "archive" {
+		t.Fatalf("state identity = %s/%s", destinationState.PipelineID, destinationState.DestinationID)
+	}
+	if destinationState.Source.Manifest.ID != manifest.ID {
+		t.Fatalf("state source id = %q, want %q", destinationState.Source.Manifest.ID, manifest.ID)
+	}
+	if got, want := len(destinationState.Outputs), 2; got != want {
+		t.Fatalf("state output count = %d, want %d", got, want)
+	}
+}
+
+func TestRunSkipsWhenDestinationStateMatches(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+	configPath := writeLocalConfig(t, sourceRoot, destinationRoot)
+	if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
+		t.Fatalf("first Run() error = %v", err)
+	}
+
+	var stdout bytes.Buffer
+	err := Run(context.Background(), RunOptions{ConfigPath: configPath, Stdout: &stdout})
+	if err != nil {
+		t.Fatalf("second Run() error = %v", err)
+	}
+	if !strings.Contains(stdout.String(), "action=skip_same") {
+		t.Fatalf("stdout = %q, want skip_same", stdout.String())
+	}
+}
+
+func TestRunReplacesOlderDestination(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+	older := manifest
+	older.Created = older.Created.Add(-time.Hour)
+	writeDestinationState(t, destinationRoot, "", older)
+	if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("old\n"), 0o600); err != nil {
+		t.Fatalf("write old output: %v", err)
+	}
+
+	var stdout bytes.Buffer
+	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), Stdout: &stdout})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if !strings.Contains(stdout.String(), "action=replace_older") {
+		t.Fatalf("stdout = %q, want replace_older", stdout.String())
+	}
+	assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
+}
+
+func TestRunSkipsNewerDestination(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+	newer := manifest
+	newer.Created = newer.Created.Add(time.Hour)
+	writeDestinationState(t, destinationRoot, "", newer)
+	if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("newer\n"), 0o600); err != nil {
+		t.Fatalf("write newer output: %v", err)
+	}
+
+	var stdout bytes.Buffer
+	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), Stdout: &stdout})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if !strings.Contains(stdout.String(), "action=skip_destination_newer") {
+		t.Fatalf("stdout = %q, want skip_destination_newer", stdout.String())
+	}
+	assertFile(t, filepath.Join(destinationRoot, "report.md"), "newer\n")
+}
+
+func TestRunFailsOnConflict(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+	manifest.ID = "other.source"
+	writeDestinationState(t, destinationRoot, "", manifest)
+
+	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot)})
+	if err == nil || !strings.Contains(err.Error(), "fail_conflict") {
+		t.Fatalf("Run() error = %v, want fail_conflict", err)
+	}
+}
+
+func TestRunFailsOnUnmanagedDestination(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+	if err := os.WriteFile(filepath.Join(destinationRoot, "unmanaged.txt"), []byte("data"), 0o600); err != nil {
+		t.Fatalf("write unmanaged file: %v", err)
+	}
+
+	err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot)})
+	if err == nil || !strings.Contains(err.Error(), "fail_unmanaged") {
+		t.Fatalf("Run() error = %v, want fail_unmanaged", err)
+	}
+}
+
+func TestRunFansOutToLocalDestinations(t *testing.T) {
+	sourceRoot := t.TempDir()
+	firstDestination := t.TempDir()
+	secondDestination := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "daily", testBundleOptions{})
+
+	err := Run(context.Background(), RunOptions{ConfigPath: writeFanoutConfig(t, sourceRoot, firstDestination, secondDestination)})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	assertFile(t, filepath.Join(firstDestination, "daily", "report.md"), "# Report\nSunny.\n")
+	assertFile(t, filepath.Join(secondDestination, "daily", "summary.txt"), "Summary\n")
+}
+
+func TestRunDryRunDoesNotWrite(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
+
+	err := Run(context.Background(), RunOptions{
+		ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot),
+		DryRun:     true,
+	})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if entries, err := os.ReadDir(destinationRoot); err != nil || len(entries) != 0 {
+		t.Fatalf("destination entries = %v err=%v, want empty", entries, err)
+	}
+}
+
+type testBundleOptions struct {
+	ID      string
+	Created time.Time
+}
+
+func writeSourceBundle(t *testing.T, root, relative string, opts testBundleOptions) bundle.Manifest {
+	t.Helper()
+	if opts.ID == "" {
+		opts.ID = "weather.daily.brentwood.2026-05-30"
+	}
+	if opts.Created.IsZero() {
+		opts.Created = time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC)
+	}
+	bundleRoot := filepath.Join(root, filepath.FromSlash(relative))
+	if err := os.MkdirAll(bundleRoot, 0o755); err != nil {
+		t.Fatalf("mkdir bundle: %v", err)
+	}
+	files := []struct {
+		path string
+		data string
+	}{
+		{path: "report.md", data: "# Report\nSunny.\n"},
+		{path: "summary.txt", data: "Summary\n"},
+	}
+	manifestFiles := make([]bundle.ManifestFile, 0, len(files))
+	for _, file := range files {
+		if err := os.WriteFile(filepath.Join(bundleRoot, filepath.FromSlash(file.path)), []byte(file.data), 0o600); err != nil {
+			t.Fatalf("write source file: %v", err)
+		}
+		manifestFiles = append(manifestFiles, bundle.ManifestFile{
+			Path:   file.path,
+			SHA256: bundle.FileDigest([]byte(file.data)),
+			Size:   int64(len(file.data)),
+		})
+	}
+	manifest := bundle.Manifest{
+		SchemaVersion: 1,
+		ID:            opts.ID,
+		Created:       opts.Created,
+		Files:         manifestFiles,
+	}
+	manifest.Digest = bundle.BundleDigest(manifest.Files)
+	data, err := json.MarshalIndent(manifest, "", "  ")
+	if err != nil {
+		t.Fatalf("marshal manifest: %v", err)
+	}
+	data = append(data, '\n')
+	if err := os.WriteFile(filepath.Join(bundleRoot, "manifest.json"), data, 0o600); err != nil {
+		t.Fatalf("write manifest: %v", err)
+	}
+	return manifest
+}
+
+func writeLocalConfig(t *testing.T, sourceRoot, destinationRoot string) string {
+	t.Helper()
+	return writeConfigFile(t, `
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: `+sourceRoot+`
+    destinations:
+      - id: archive
+        backend: local
+        path: `+destinationRoot+`
+        publish:
+          source: true
+          html: false
+`)
+}
+
+func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string {
+	t.Helper()
+	return writeConfigFile(t, `
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: `+sourceRoot+`
+    destinations:
+      - id: archive-one
+        backend: local
+        path: `+firstDestination+`
+      - id: archive-two
+        backend: local
+        path: `+secondDestination+`
+`)
+}
+
+func writeConfigFile(t *testing.T, body string) string {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "config.yml")
+	if err := os.WriteFile(path, []byte(strings.TrimSpace(body)+"\n"), 0o600); err != nil {
+		t.Fatalf("write config: %v", err)
+	}
+	return path
+}
+
+func writeDestinationState(t *testing.T, root, relative string, manifest bundle.Manifest) {
+	t.Helper()
+	bundleRoot := filepath.Join(root, filepath.FromSlash(relative))
+	if err := os.MkdirAll(bundleRoot, 0o755); err != nil {
+		t.Fatalf("mkdir destination: %v", err)
+	}
+	destinationState := state.DistributorState{
+		SchemaVersion: state.SchemaVersion,
+		PipelineID:    "reports",
+		DestinationID: "archive",
+		PublishedAt:   time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC),
+		Source:        state.SourceState{Manifest: manifest},
+		Outputs: []state.OutputFile{
+			{Path: "report.md", Kind: state.OutputKindSource, SourcePath: "report.md", SHA256: manifest.Files[0].SHA256, Size: manifest.Files[0].Size},
+			{Path: "summary.txt", Kind: state.OutputKindSource, SourcePath: "summary.txt", SHA256: manifest.Files[1].SHA256, Size: manifest.Files[1].Size},
+		},
+	}
+	data, err := json.MarshalIndent(destinationState, "", "  ")
+	if err != nil {
+		t.Fatalf("marshal state: %v", err)
+	}
+	data = append(data, '\n')
+	if err := os.WriteFile(filepath.Join(bundleRoot, ".distributor.json"), data, 0o600); err != nil {
+		t.Fatalf("write state: %v", err)
+	}
+}
+
+func readStateFile(t *testing.T, path string) state.DistributorState {
+	t.Helper()
+	data, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read state: %v", err)
+	}
+	destinationState, err := state.Parse(data)
+	if err != nil {
+		t.Fatalf("parse state: %v", err)
+	}
+	return destinationState
+}
+
+func assertFile(t *testing.T, path, want string) {
+	t.Helper()
+	data, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read file %s: %v", path, err)
+	}
+	if got := string(data); got != want {
+		t.Fatalf("%s = %q, want %q", path, got, want)
 	}
 }
--- a/internal/cli/root_test.go
+++ b/internal/cli/root_test.go
@@ -41,28 +41,6 @@ func TestExecuteVersion(t *testing.T) {
 	}
 }

-func TestRunWithoutDryRunFailsClearly(t *testing.T) {
-	tests := []string{"run"}
-
-	for _, command := range tests {
-		t.Run(command, func(t *testing.T) {
-			var stdout, stderr bytes.Buffer
-
-			code := Execute(context.Background(), []string{command}, &stdout, &stderr)
-
-			if code != exitError {
-				t.Fatalf("exit code = %d, want %d", code, exitError)
-			}
-			if stdout.Len() != 0 {
-				t.Fatalf("stdout = %q, want empty", stdout.String())
-			}
-			if !strings.Contains(stderr.String(), "not implemented") {
-				t.Fatalf("stderr = %q, want not implemented error", stderr.String())
-			}
-		})
-	}
-}
-
 func TestExecuteValidate(t *testing.T) {
 	var stdout, stderr bytes.Buffer

@@ -90,17 +68,19 @@ func TestExecuteInspect(t *testing.T) {
 }

 func TestExecuteRunDryRun(t *testing.T) {
+	sourceRoot := t.TempDir()
+	writeCLIBundle(t, sourceRoot)
 	configPath := filepath.Join(t.TempDir(), "config.yml")
 	err := os.WriteFile(configPath, []byte(`
 pipelines:
  - id: reports
    source:
      backend: local
-      path: /source
+      path: `+sourceRoot+`
    destinations:
      - id: archive
        backend: local
-        path: /archive
+        path: `+t.TempDir()+`
 `), 0o600)
 	if err != nil {
 		t.Fatalf("write config: %v", err)
@@ -113,7 +93,7 @@ pipelines:
 	if code != exitOK {
 		t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
 	}
-	if !strings.Contains(stdout.String(), "Configured pipelines: 1") {
+	if !strings.Contains(stdout.String(), "action=publish_new") {
 		t.Fatalf("stdout = %q, want config summary", stdout.String())
 	}
 	if stderr.Len() != 0 {
@@ -121,6 +101,38 @@ pipelines:
 	}
 }

+func TestExecuteRunPublishes(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	writeCLIBundle(t, sourceRoot)
+	configPath := filepath.Join(t.TempDir(), "config.yml")
+	err := os.WriteFile(configPath, []byte(`
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: `+sourceRoot+`
+    destinations:
+      - id: archive
+        backend: local
+        path: `+destinationRoot+`
+`), 0o600)
+	if err != nil {
+		t.Fatalf("write config: %v", err)
+	}
+
+	var stdout, stderr bytes.Buffer
+
+	code := Execute(context.Background(), []string{"run", "--config", configPath}, &stdout, &stderr)
+
+	if code != exitOK {
+		t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
+	}
+	if _, err := os.Stat(filepath.Join(destinationRoot, ".distributor.json")); err != nil {
+		t.Fatalf("state stat error = %v", err)
+	}
+}
+
 func TestUnknownCommandIsUsageError(t *testing.T) {
 	var stdout, stderr bytes.Buffer

@@ -133,3 +145,37 @@ func TestUnknownCommandIsUsageError(t *testing.T) {
 		t.Fatalf("stderr = %q, want unknown command error", stderr.String())
 	}
 }
+
+func writeCLIBundle(t *testing.T, root string) {
+	t.Helper()
+	for _, file := range []struct {
+		path string
+		data string
+	}{
+		{"manifest.json", `{
+  "schema_version": 1,
+  "id": "weather.daily.brentwood.2026-05-30",
+  "digest": "sha256:099b205780d2b050024868399961b05731729a548d5d6329c7b06a6740dd75fe",
+  "created": "2026-05-30T11:10:00Z",
+  "files": [
+    {
+      "path": "report.md",
+      "sha256": "sha256:3640fd37140ee4d2e0e93e78834f232ea67a50e7bc6279203690cc7de1975fa6",
+      "size": 16
+    },
+    {
+      "path": "summary.txt",
+      "sha256": "sha256:3cbb36aca330b3bd113955dfbada0adb7a5f95ad9f678bd61f175406c6a37e95",
+      "size": 8
+    }
+  ]
+}
+`},
+		{"report.md", "# Report\nSunny.\n"},
+		{"summary.txt", "Summary\n"},
+	} {
+		if err := os.WriteFile(filepath.Join(root, file.path), []byte(file.data), 0o600); err != nil {
+			t.Fatalf("write bundle file: %v", err)
+		}
+	}
+}
--- a/internal/config/load_test.go
+++ b/internal/config/load_test.go
@@ -264,6 +264,7 @@ pipelines:
 func TestExampleConfigsLoad(t *testing.T) {
 	for _, path := range []string{
 		"../../examples/local-to-local.yml",
+		"../../examples/local-publish.yml",
 		"../../examples/fan-out.yml",
 	} {
 		t.Run(path, func(t *testing.T) {
--- a/internal/publish/execute.go
+++ b/internal/publish/execute.go
@@ -0,0 +1,98 @@
+package publish
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/state"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+)
+
+func Execute(ctx context.Context, req Request, plan Plan) error {
+	switch plan.Action {
+	case ActionSkipSame, ActionSkipDestinationNewer:
+		return nil
+	case ActionPublishNew, ActionReplaceOlder:
+	default:
+		return fmt.Errorf("cannot execute action %s: %s", plan.Action, plan.Reason)
+	}
+
+	if plan.Action == ActionReplaceOlder {
+		if plan.ExistingState == nil {
+			return fmt.Errorf("replace requires existing destination state")
+		}
+		if err := req.DestinationBackend.DeleteManagedBundle(ctx, req.DestinationBundlePath, existingManagedOutputPaths(*plan.ExistingState), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
+			return err
+		}
+		if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil {
+			return err
+		}
+	}
+
+	writtenOutputs := make([]Output, 0, len(plan.Outputs))
+	cleanup := func() {
+		_ = req.DestinationBackend.DeleteManagedBundle(ctx, req.DestinationBundlePath, managedOutputPaths(writtenOutputs), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true})
+	}
+	for _, output := range plan.Outputs {
+		sourcePath, err := storage.Join(req.SourceBundle.RootRelativePath, output.SourcePath)
+		if err != nil {
+			cleanup()
+			return err
+		}
+		destinationPath, err := storage.Join(req.DestinationBundlePath, output.DestinationPath)
+		if err != nil {
+			cleanup()
+			return err
+		}
+		data, err := req.SourceBackend.ReadFile(ctx, sourcePath)
+		if err != nil {
+			cleanup()
+			return err
+		}
+		if _, err := req.DestinationBackend.WriteFile(ctx, destinationPath, data, storage.WriteOptions{Overwrite: false, PreferAtomic: true}); err != nil {
+			cleanup()
+			return err
+		}
+		writtenOutputs = append(writtenOutputs, output)
+	}
+
+	destinationState := state.DistributorState{
+		SchemaVersion:      state.SchemaVersion,
+		DistributorVersion: req.DistributorVersion,
+		PipelineID:         req.PipelineID,
+		DestinationID:      req.DestinationID,
+		PublishedAt:        time.Now().UTC(),
+		Source:             state.SourceState{Manifest: req.SourceBundle.Manifest},
+		Outputs:            stateOutputs(plan.Outputs),
+	}
+	if err := state.Validate(destinationState); err != nil {
+		cleanup()
+		return err
+	}
+	data, err := json.MarshalIndent(destinationState, "", "  ")
+	if err != nil {
+		cleanup()
+		return err
+	}
+	data = append(data, '\n')
+	statePath, err := storage.StatePath(req.DestinationBundlePath)
+	if err != nil {
+		cleanup()
+		return err
+	}
+	if _, err := req.DestinationBackend.WriteFile(ctx, statePath, data, storage.WriteOptions{Overwrite: false, PreferAtomic: true}); err != nil {
+		cleanup()
+		return err
+	}
+	return nil
+}
+
+func existingManagedOutputPaths(destinationState state.DistributorState) []string {
+	paths := make([]string, 0, len(destinationState.Outputs))
+	for _, output := range destinationState.Outputs {
+		paths = append(paths, output.Path)
+	}
+	return paths
+}
--- a/internal/publish/execute_test.go
+++ b/internal/publish/execute_test.go
@@ -0,0 +1,91 @@
+package publish
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
+)
+
+func TestExecuteCleansUpAfterWriteFailure(t *testing.T) {
+	sourceBackend := fake.New()
+	destinationBackend := &failingBackend{Backend: fake.New(), failPath: "summary.txt"}
+	sourceBundle := writeFakeSourceBundle(t, sourceBackend)
+	req := Request{
+		PipelineID:            "reports",
+		DestinationID:         "archive",
+		SourceBundle:          sourceBundle,
+		SourceBackend:         sourceBackend,
+		DestinationBackend:    destinationBackend,
+		DestinationBundlePath: "",
+		Publish:               config.PublishPolicy{Source: true},
+		Transfer:              config.TransferPolicy{OnDestinationSame: config.TransferActionSkip, OnDestinationOlder: config.TransferActionReplace, OnDestinationNewer: config.TransferActionSkip, OnConflict: config.TransferActionFail},
+		DistributorVersion:    "test",
+	}
+	plan, err := Build(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Build() error = %v", err)
+	}
+	err = Execute(context.Background(), req, plan)
+	if err == nil {
+		t.Fatal("Execute() error = nil, want error")
+	}
+	found, err := destinationBackend.HasAny(context.Background(), "")
+	if err != nil {
+		t.Fatalf("HasAny() error = %v", err)
+	}
+	if found {
+		t.Fatal("destination has content after failed execution")
+	}
+}
+
+type failingBackend struct {
+	*fake.Backend
+	failPath string
+}
+
+func (b *failingBackend) WriteFile(ctx context.Context, path string, data []byte, opts storage.WriteOptions) (storage.Entry, error) {
+	if path == b.failPath {
+		return storage.Entry{}, fmt.Errorf("injected write failure")
+	}
+	return b.Backend.WriteFile(ctx, path, data, opts)
+}
+
+func (b *failingBackend) WriteFrom(ctx context.Context, path string, r io.Reader, opts storage.WriteOptions) (storage.Entry, error) {
+	if path == b.failPath {
+		return storage.Entry{}, fmt.Errorf("injected write failure")
+	}
+	return b.Backend.WriteFrom(ctx, path, r, opts)
+}
+
+func writeFakeSourceBundle(t *testing.T, backend *fake.Backend) bundle.Bundle {
+	t.Helper()
+	files := []struct {
+		path string
+		data string
+	}{
+		{path: "report.md", data: "# Report\nSunny.\n"},
+		{path: "summary.txt", data: "Summary\n"},
+	}
+	manifestFiles := make([]bundle.ManifestFile, 0, len(files))
+	for _, file := range files {
+		if _, err := backend.WriteFile(context.Background(), file.path, []byte(file.data), storage.WriteOptions{}); err != nil {
+			t.Fatalf("WriteFile() error = %v", err)
+		}
+		manifestFiles = append(manifestFiles, bundle.ManifestFile{Path: file.path, SHA256: bundle.FileDigest([]byte(file.data)), Size: int64(len(file.data))})
+	}
+	manifest := bundle.Manifest{
+		SchemaVersion: 1,
+		ID:            "weather.daily.brentwood.2026-05-30",
+		Created:       time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC),
+		Files:         manifestFiles,
+	}
+	manifest.Digest = bundle.BundleDigest(manifest.Files)
+	return bundle.Bundle{Manifest: manifest}
+}
--- a/internal/publish/output.go
+++ b/internal/publish/output.go
@@ -0,0 +1,54 @@
+package publish
+
+import (
+	"fmt"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/state"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+)
+
+func PlanSourceOutputs(req Request) ([]Output, error) {
+	if !req.Publish.Source {
+		return nil, nil
+	}
+	outputs := make([]Output, 0, len(req.SourceBundle.Manifest.Files))
+	seen := make(map[string]struct{}, len(req.SourceBundle.Manifest.Files))
+	for _, file := range req.SourceBundle.Manifest.Files {
+		if _, exists := seen[file.Path]; exists {
+			return nil, fmt.Errorf("destination output path collision: %s", file.Path)
+		}
+		seen[file.Path] = struct{}{}
+		if err := storage.ValidatePath(file.Path); err != nil {
+			return nil, fmt.Errorf("destination output path %q: %w", file.Path, err)
+		}
+		outputs = append(outputs, Output{
+			SourcePath:      file.Path,
+			DestinationPath: file.Path,
+			SHA256:          file.SHA256,
+			Size:            file.Size,
+		})
+	}
+	return outputs, nil
+}
+
+func stateOutputs(outputs []Output) []state.OutputFile {
+	files := make([]state.OutputFile, 0, len(outputs))
+	for _, output := range outputs {
+		files = append(files, state.OutputFile{
+			Path:       output.DestinationPath,
+			Kind:       state.OutputKindSource,
+			SourcePath: output.SourcePath,
+			SHA256:     output.SHA256,
+			Size:       output.Size,
+		})
+	}
+	return files
+}
+
+func managedOutputPaths(outputs []Output) []string {
+	paths := make([]string, 0, len(outputs))
+	for _, output := range outputs {
+		paths = append(paths, output.DestinationPath)
+	}
+	return paths
+}
--- a/internal/publish/output_test.go
+++ b/internal/publish/output_test.go
@@ -0,0 +1,29 @@
+package publish
+
+import (
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+)
+
+func TestPlanSourceOutputsRejectsCollision(t *testing.T) {
+	_, err := PlanSourceOutputs(Request{
+		SourceBundle: bundle.Bundle{
+			Manifest: bundle.Manifest{
+				SchemaVersion: 1,
+				ID:            "bundle",
+				Created:       time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC),
+				Files: []bundle.ManifestFile{
+					{Path: "report.md", SHA256: "sha256:1111111111111111111111111111111111111111111111111111111111111111", Size: 1},
+					{Path: "report.md", SHA256: "sha256:2222222222222222222222222222222222222222222222222222222222222222", Size: 1},
+				},
+			},
+		},
+		Publish: config.PublishPolicy{Source: true},
+	})
+	if err == nil {
+		t.Fatal("PlanSourceOutputs() error = nil, want collision")
+	}
+}
--- a/internal/publish/plan.go
+++ b/internal/publish/plan.go
@@ -0,0 +1,141 @@
+package publish
+
+import (
+	"context"
+	"fmt"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/state"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+)
+
+type Action string
+
+const (
+	ActionPublishNew           Action = "publish_new"
+	ActionReplaceOlder         Action = "replace_older"
+	ActionSkipSame             Action = "skip_same"
+	ActionSkipDestinationNewer Action = "skip_destination_newer"
+	ActionFailConflict         Action = "fail_conflict"
+	ActionFailUnmanaged        Action = "fail_unmanaged"
+)
+
+type Request struct {
+	PipelineID            string
+	DestinationID         string
+	SourceBundle          bundle.Bundle
+	SourceBackend         storage.Backend
+	DestinationBackend    storage.Backend
+	DestinationBundlePath string
+	Publish               config.PublishPolicy
+	Transfer              config.TransferPolicy
+	DistributorVersion    string
+}
+
+type Plan struct {
+	PipelineID            string
+	DestinationID         string
+	BundleID              string
+	BundlePath            string
+	DestinationBundlePath string
+	Action                Action
+	Reason                string
+	Outputs               []Output
+	ExistingState         *state.DistributorState
+}
+
+type Output struct {
+	SourcePath      string
+	DestinationPath string
+	SHA256          string
+	Size            int64
+}
+
+func Build(ctx context.Context, req Request) (Plan, error) {
+	if err := validateRequest(req); err != nil {
+		return Plan{}, err
+	}
+	outputs, err := PlanSourceOutputs(req)
+	if err != nil {
+		return Plan{}, err
+	}
+	status, err := inspectDestination(ctx, req.DestinationBackend, req.DestinationBundlePath)
+	if err != nil {
+		return Plan{}, err
+	}
+	comparison := state.Compare(req.SourceBundle.Manifest, req.PipelineID, req.DestinationID, status)
+	action, reason := actionForComparison(comparison, req.Transfer)
+	plan := Plan{
+		PipelineID:            req.PipelineID,
+		DestinationID:         req.DestinationID,
+		BundleID:              req.SourceBundle.Manifest.ID,
+		BundlePath:            req.SourceBundle.RootRelativePath,
+		DestinationBundlePath: req.DestinationBundlePath,
+		Action:                action,
+		Reason:                reason,
+		Outputs:               outputs,
+		ExistingState:         status.State,
+	}
+	if action == ActionFailConflict || action == ActionFailUnmanaged {
+		return plan, fmt.Errorf("%s: %s", action, reason)
+	}
+	return plan, nil
+}
+
+func validateRequest(req Request) error {
+	if req.PipelineID == "" {
+		return fmt.Errorf("pipeline id is required")
+	}
+	if req.DestinationID == "" {
+		return fmt.Errorf("destination id is required")
+	}
+	if req.SourceBackend == nil {
+		return fmt.Errorf("source backend is required")
+	}
+	if req.DestinationBackend == nil {
+		return fmt.Errorf("destination backend is required")
+	}
+	if req.Publish.HTML {
+		return fmt.Errorf("publish html is not implemented")
+	}
+	if !req.Publish.Source {
+		return fmt.Errorf("publish source must be enabled")
+	}
+	return nil
+}
+
+func actionForComparison(comparison state.Comparison, transfer config.TransferPolicy) (Action, string) {
+	switch comparison.Outcome {
+	case state.OutcomeDestinationAbsent:
+		return ActionPublishNew, comparison.Reason
+	case state.OutcomeDestinationUnmanaged:
+		return ActionFailUnmanaged, comparison.Reason
+	case state.OutcomeInvalidState, state.OutcomeIdentityMismatch, state.OutcomeSameCreatedConflict, state.OutcomeDifferentSourceConflict:
+		return ActionFailConflict, comparison.Reason
+	case state.OutcomeSameSource:
+		if transfer.OnDestinationSame == config.TransferActionFail {
+			return ActionFailConflict, "destination matches source and transfer policy requires failure"
+		}
+		return ActionSkipSame, comparison.Reason
+	case state.OutcomeDestinationOlder:
+		if transfer.OnDestinationOlder == config.TransferActionFail {
+			return ActionFailConflict, "destination is older and transfer policy requires failure"
+		}
+		return ActionReplaceOlder, comparison.Reason
+	case state.OutcomeDestinationNewer:
+		if transfer.OnDestinationNewer == config.TransferActionFail {
+			return ActionFailConflict, "destination is newer and transfer policy requires failure"
+		}
+		return ActionSkipDestinationNewer, comparison.Reason
+	default:
+		return ActionFailConflict, "unsupported comparison outcome"
+	}
+}
+
+func displayPath(path string) string {
+	if path == "" {
+		return "."
+	}
+	return path
+}
--- a/internal/publish/reconcile.go
+++ b/internal/publish/reconcile.go
@@ -0,0 +1,31 @@
+package publish
+
+import (
+	"context"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/state"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+)
+
+func inspectDestination(ctx context.Context, backend storage.Backend, bundlePath string) (state.DestinationStatus, error) {
+	statePath, err := storage.StatePath(bundlePath)
+	if err != nil {
+		return state.DestinationStatus{}, err
+	}
+	data, err := backend.ReadFile(ctx, statePath)
+	if err == nil {
+		destinationState, parseErr := state.Parse(data)
+		if parseErr != nil {
+			return state.DestinationStatus{StateErr: parseErr}, nil
+		}
+		return state.DestinationStatus{State: &destinationState, HasContents: true}, nil
+	}
+	if !storage.IsNotFound(err) {
+		return state.DestinationStatus{}, err
+	}
+	hasContents, err := backend.HasAny(ctx, bundlePath)
+	if err != nil {
+		return state.DestinationStatus{}, err
+	}
+	return state.DestinationStatus{HasContents: hasContents}, nil
+}
--- a/internal/publish/safety.go
+++ b/internal/publish/safety.go
@@ -0,0 +1,19 @@
+package publish
+
+import (
+	"context"
+	"fmt"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+)
+
+func ensureDestinationEmpty(ctx context.Context, backend storage.Backend, bundlePath string) error {
+	hasAny, err := backend.HasAny(ctx, bundlePath)
+	if err != nil {
+		return err
+	}
+	if hasAny {
+		return fmt.Errorf("destination bundle path %q is not empty after managed cleanup", displayPath(bundlePath))
+	}
+	return nil
+}