Align internal publish terminology across stage, app, and artifacts

2026-05-23 13:39:15 +00:00
parent 7d584ee6cd
commit 96b886e711
25 changed files with 306 additions and 306 deletions
--- a/internal/stage/publish.go
+++ b/internal/stage/publish.go
@@ -0,0 +1,808 @@
+package stage
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"time"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+)
+
+type publishStage struct{}
+type publishUploadFile struct {
+	RelativePath string
+	LocalPath    string
+}
+
+var publishPrerequisiteStages = []string{
+	"prepare",
+	"transcribe",
+	"merge",
+	"polish",
+	"normalize",
+	"trim",
+	"analyze",
+}
+
+func (publishStage) Name() string { return "publish" }
+
+func (publishStage) Declares() IODecl {
+	return IODecl{
+		Inputs: []artifacts.Ref{
+			{Kind: "manifest", Category: "input", RelativePath: "manifest.json"},
+		},
+	}
+}
+
+func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
+	if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil {
+		return nil, fmt.Errorf("publish: resolved config must include pipeline and session")
+	}
+
+	if publishDisabled(env) {
+		return &StageResult{
+			Metadata: map[string]any{
+				"stage":                   "publish",
+				"skipped":                 true,
+				"publish_enabled":         false,
+				"audio_upload_skipped":    true,
+				"current_pointer_written": false,
+			},
+		}, nil
+	}
+	if publishRunUploadDisabled(env) {
+		return &StageResult{
+			Metadata: map[string]any{
+				"stage":                   "publish",
+				"skipped":                 true,
+				"upload_run_enabled":      false,
+				"audio_upload_skipped":    true,
+				"current_pointer_written": false,
+			},
+		}, nil
+	}
+
+	if err := validatePublishPrerequisites(m); err != nil {
+		return nil, fmt.Errorf("publish: %w", err)
+	}
+	if env.ObjectStore == nil {
+		return nil, fmt.Errorf("publish: remote object store backend is required when publish run upload is enabled")
+	}
+
+	runRoot, err := resolvePublishRunRoot(env, m)
+	if err != nil {
+		return nil, fmt.Errorf("publish: resolve run root: %w", err)
+	}
+	runRootInfo, err := os.Stat(runRoot)
+	if err != nil {
+		return nil, fmt.Errorf("publish: run root %q: %w", runRoot, err)
+	}
+	if !runRootInfo.IsDir() {
+		return nil, fmt.Errorf("publish: run root %q is not a directory", runRoot)
+	}
+
+	runPrefix, err := artifacts.ResolvePublishRunPrefix(env.Config, m)
+	if err != nil {
+		return nil, fmt.Errorf("publish: resolve s3 run prefix: %w", err)
+	}
+	sessionPrefix, err := artifacts.ResolvePublishSessionPrefix(env.Config, m)
+	if err != nil {
+		return nil, fmt.Errorf("publish: resolve s3 session prefix: %w", err)
+	}
+	bucket := artifacts.ResolvePublishBucket(env.Config, m)
+	if bucket == "" {
+		return nil, fmt.Errorf("publish: resolve s3 bucket: bucket is required")
+	}
+	runID := strings.TrimSpace(m.RunID)
+	if runID == "" {
+		return nil, fmt.Errorf("publish: run id is required")
+	}
+
+	manifestSource, err := resolvePublishRunManifestSource(runRoot)
+	if err != nil {
+		return nil, fmt.Errorf("publish: resolve run manifest source: %w", err)
+	}
+
+	runFiles, err := collectPublishRunFiles(runRoot, manifestSource)
+	if err != nil {
+		return nil, fmt.Errorf("publish: collect run files: %w", err)
+	}
+	sessionPaths := publishSessionPaths(env, m)
+	previousFiles, err := collectPublishPreviousFiles(sessionPaths.PreviousDir)
+	if err != nil {
+		return nil, fmt.Errorf("publish: collect previous files: %w", err)
+	}
+	runtimeCatalog, err := buildPublishRuntimeArtifactCatalog(sessionPaths, env.Config.Pipeline.Scriptorium)
+	if err != nil {
+		return nil, fmt.Errorf("publish: build runtime artifact catalog: %w", err)
+	}
+	publishOutputs, skippedOptionalOutputs, skippedUnselectedOutputs, lockedOutputs, err := resolvePublishOutputs(
+		sessionPaths,
+		m,
+		runtimeCatalog,
+		env.Config.Pipeline.Publish.Outputs,
+		env.Config.Pipeline.Publish.Locks,
+		env.SelectedArtifactKeys,
+		sessionPrefix,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("publish: resolve publish output rules: %w", err)
+	}
+	runUploaded := make([]string, 0, len(runFiles))
+	for _, file := range runFiles {
+		key := artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath)
+		if _, err := env.ObjectStore.Upload(ctx, file.LocalPath, key, storage.UploadOptions{}); err != nil {
+			return nil, fmt.Errorf("publish: upload run file %q to %q: %w", file.RelativePath, key, err)
+		}
+		runUploaded = append(runUploaded, file.RelativePath)
+	}
+
+	publishedUploaded := make([]string, 0, len(publishOutputs))
+	for _, publishedOutput := range publishOutputs {
+		key := artifacts.S3PublishedOutputKey(sessionPrefix, publishedOutput.Dest)
+		if _, err := env.ObjectStore.Upload(ctx, publishedOutput.LocalPath, key, storage.UploadOptions{}); err != nil {
+			return nil, fmt.Errorf("publish: upload published output source %q to %q: %w", publishedOutput.Source, key, err)
+		}
+		publishedUploaded = append(publishedUploaded, publishedOutput.Dest)
+	}
+
+	previousUploaded := make([]string, 0, len(previousFiles))
+	for _, file := range previousFiles {
+		key := artifacts.S3PublishedOutputKey(sessionPrefix, file.RelativePath)
+		if _, err := env.ObjectStore.Upload(ctx, file.LocalPath, key, storage.UploadOptions{}); err != nil {
+			return nil, fmt.Errorf("publish: upload previous file %q to %q: %w", file.RelativePath, key, err)
+		}
+		previousUploaded = append(previousUploaded, file.RelativePath)
+	}
+
+	currentManifestKey, currentRunPointerKey := artifacts.ResolveCurrentStateKeys(sessionPrefix)
+	manifestTempPath, err := writeCurrentManifestSnapshot(m, publishMetadataPreview(
+		bucket,
+		runPrefix,
+		sessionPrefix,
+		runUploaded,
+		publishedUploaded,
+		previousUploaded,
+		skippedOptionalOutputs,
+		skippedUnselectedOutputs,
+		lockedOutputs,
+		currentManifestKey,
+	))
+	if err != nil {
+		return nil, fmt.Errorf("publish: build current manifest snapshot: %w", err)
+	}
+	defer func() { _ = os.Remove(manifestTempPath) }()
+
+	if _, err := env.ObjectStore.Upload(ctx, manifestTempPath, currentManifestKey, storage.UploadOptions{
+		ContentType: "application/json",
+	}); err != nil {
+		return nil, fmt.Errorf("publish: upload current manifest to %q: %w", currentManifestKey, err)
+	}
+
+	runIDTempPath, err := writeCurrentRunIDPointer(runID)
+	if err != nil {
+		return nil, fmt.Errorf("publish: build current run id pointer: %w", err)
+	}
+	defer func() { _ = os.Remove(runIDTempPath) }()
+
+	if _, err := env.ObjectStore.Upload(ctx, runIDTempPath, currentRunPointerKey, storage.UploadOptions{
+		ContentType: "text/plain; charset=utf-8",
+	}); err != nil {
+		return nil, fmt.Errorf("publish: upload current run pointer to %q: %w", currentRunPointerKey, err)
+	}
+
+	return &StageResult{
+		Metadata: map[string]any{
+			"stage":                      "publish",
+			"uploaded":                   true,
+			"s3_bucket":                  bucket,
+			"s3_run_prefix":              runPrefix,
+			"run_files_uploaded":         len(runUploaded),
+			"run_uploaded_paths":         runUploaded,
+			"published_files_uploaded":   len(publishedUploaded),
+			"published_paths":            publishedUploaded,
+			"previous_files_uploaded":    len(previousUploaded),
+			"previous_uploaded_paths":    previousUploaded,
+			"skipped_optional_outputs":   skippedOptionalOutputs,
+			"skipped_unselected_outputs": skippedUnselectedOutputMetadata(skippedUnselectedOutputs),
+			"locked_output_count":        len(lockedOutputs),
+			"locked_outputs":             lockedOutputMetadata(lockedOutputs),
+			"current_manifest_key":       currentManifestKey,
+			"current_run_id_key":         currentRunPointerKey,
+			"current_pointer_written":    true,
+			"audio_upload_skipped":       true,
+		},
+	}, nil
+}
+
+type publishOutput struct {
+	Source     string
+	Dest       string
+	Required   bool
+	LocalPath  string
+	Provenance string
+}
+
+type publishLockedOutput struct {
+	Source     string
+	Dest       string
+	RemoteKey  string
+	Reason     string
+	Required   bool
+	LocalPath  string
+	Provenance string
+}
+
+type publishSkippedUnselectedOutput struct {
+	Source   string
+	Dest     string
+	Required bool
+}
+
+func publishDisabled(env *Env) bool {
+	cfg := env.Config.Pipeline.Publish
+	if cfg == nil {
+		return true
+	}
+	return cfg.Enabled != nil && !*cfg.Enabled
+}
+
+func publishRunUploadDisabled(env *Env) bool {
+	cfg := env.Config.Pipeline.Publish
+	if cfg == nil {
+		return true
+	}
+	return cfg.UploadRun != nil && !*cfg.UploadRun
+}
+
+func validatePublishPrerequisites(m *manifest.Manifest) error {
+	if m == nil {
+		return fmt.Errorf("manifest is required")
+	}
+	for _, stageName := range publishPrerequisiteStages {
+		sr := m.Stages[stageName]
+		if sr == nil {
+			return fmt.Errorf("prerequisite stage %q has not succeeded", stageName)
+		}
+		if sr.Status != manifest.StatusSucceeded {
+			return fmt.Errorf("prerequisite stage %q status is %q (want %q)", stageName, sr.Status, manifest.StatusSucceeded)
+		}
+	}
+	return nil
+}
+
+func resolvePublishRunRoot(env *Env, m *manifest.Manifest) (string, error) {
+	sessionID := strings.TrimSpace(env.Config.Session.SessionID)
+	if sessionID == "" && m != nil {
+		sessionID = strings.TrimSpace(m.SessionID)
+	}
+	campaign := strings.TrimSpace(env.Config.Session.Campaign)
+	if campaign == "" && m != nil {
+		campaign = strings.TrimSpace(m.Campaign)
+	}
+	runID := ""
+	if m != nil {
+		runID = strings.TrimSpace(m.RunID)
+	}
+	if sessionID == "" || campaign == "" {
+		return "", fmt.Errorf("campaign and session id are required")
+	}
+	if runID == "" {
+		return "", fmt.Errorf("run id is required")
+	}
+
+	canonical := filepath.Clean(artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID))
+	canonicalExists, err := directoryExists(canonical)
+	if err != nil {
+		return "", fmt.Errorf("check canonical run root %q: %w", canonical, err)
+	}
+	if !canonicalExists {
+		return "", fmt.Errorf("run root not found for campaign %q session %q run %q at canonical path %q", campaign, sessionID, runID, canonical)
+	}
+	return canonical, nil
+}
+
+func resolvePublishSessionRoot(env *Env, m *manifest.Manifest) (string, error) {
+	sessionID := strings.TrimSpace(env.Config.Session.SessionID)
+	if sessionID == "" && m != nil {
+		sessionID = strings.TrimSpace(m.SessionID)
+	}
+	campaign := strings.TrimSpace(env.Config.Session.Campaign)
+	if campaign == "" && m != nil {
+		campaign = strings.TrimSpace(m.Campaign)
+	}
+	if sessionID == "" {
+		return "", fmt.Errorf("session id is required")
+	}
+	if campaign == "" {
+		return "", fmt.Errorf("campaign is required")
+	}
+	return filepath.Clean(artifacts.SessionWorkDirForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID)), nil
+}
+
+func publishSessionPaths(env *Env, m *manifest.Manifest) artifacts.SessionPaths {
+	sessionID := strings.TrimSpace(env.Config.Session.SessionID)
+	if sessionID == "" && m != nil {
+		sessionID = strings.TrimSpace(m.SessionID)
+	}
+	campaign := strings.TrimSpace(env.Config.Session.Campaign)
+	if campaign == "" && m != nil {
+		campaign = strings.TrimSpace(m.Campaign)
+	}
+	store := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root)
+	return store.SessionPathsFor(campaign, sessionID)
+}
+
+func resolvePublishOutputs(
+	paths artifacts.SessionPaths,
+	m *manifest.Manifest,
+	catalog *artifacts.ArtifactCatalog,
+	rules []config.PublishOutputRule,
+	locks []config.PublishLockRule,
+	selectedArtifactKeys []string,
+	sessionPrefix string,
+) ([]publishOutput, []string, []publishSkippedUnselectedOutput, []publishLockedOutput, error) {
+	out := make([]publishOutput, 0, len(rules))
+	skippedOptionalOutputs := make([]string, 0)
+	skippedUnselectedOutputs := make([]publishSkippedUnselectedOutput, 0)
+	lockedOutputs := make([]publishLockedOutput, 0)
+	lockSet := publishLockSet(locks)
+	selectedSet := publishSelectedArtifactSet(selectedArtifactKeys)
+	configuredOutputs := configuredOutputPathMapFromCatalog(catalog)
+	for _, rule := range rules {
+		source := strings.TrimSpace(rule.Source)
+		required := rule.Required == nil || *rule.Required
+		dest, err := resolvePublishOutputDest(rule, configuredOutputs)
+		if err != nil {
+			return nil, nil, nil, nil, fmt.Errorf("source %q: %w", source, err)
+		}
+		if len(selectedSet) > 0 {
+			if key, ok := artifactpolicy.ParseConfiguredSource(source); ok {
+				if _, selected := selectedSet[key]; !selected {
+					skippedUnselectedOutputs = append(skippedUnselectedOutputs, publishSkippedUnselectedOutput{
+						Source:   source,
+						Dest:     dest,
+						Required: required,
+					})
+					continue
+				}
+			}
+		}
+		lock, locked := lockSet[source]
+		resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, catalog)
+		if err != nil {
+			if locked {
+				lockedOutputs = append(lockedOutputs, publishLockedOutput{
+					Source:    source,
+					Dest:      dest,
+					RemoteKey: artifacts.S3PublishedOutputKey(sessionPrefix, dest),
+					Reason:    strings.TrimSpace(lock.Reason),
+					Required:  required,
+				})
+				continue
+			}
+			if errors.Is(err, artifacts.ErrSessionArtifactNotFound) && !required {
+				skippedOptionalOutputs = append(skippedOptionalOutputs, dest)
+				continue
+			}
+			if errors.Is(err, artifacts.ErrSessionArtifactNotFound) {
+				return nil, nil, nil, nil, fmt.Errorf("required output source unavailable: %q", source)
+			}
+			return nil, nil, nil, nil, fmt.Errorf("resolve source %q: %w", source, err)
+		}
+		if locked {
+			lockedOutputs = append(lockedOutputs, publishLockedOutput{
+				Source:     source,
+				Dest:       dest,
+				RemoteKey:  artifacts.S3PublishedOutputKey(sessionPrefix, dest),
+				Reason:     strings.TrimSpace(lock.Reason),
+				Required:   required,
+				LocalPath:  resolved.Path,
+				Provenance: resolved.Provenance,
+			})
+			continue
+		}
+		out = append(out, publishOutput{
+			Source:     source,
+			Dest:       dest,
+			Required:   required,
+			LocalPath:  resolved.Path,
+			Provenance: resolved.Provenance,
+		})
+	}
+	return out, skippedOptionalOutputs, skippedUnselectedOutputs, lockedOutputs, nil
+}
+
+func publishSelectedArtifactSet(selected []string) map[string]struct{} {
+	if len(selected) == 0 {
+		return nil
+	}
+	out := make(map[string]struct{}, len(selected))
+	for _, key := range selected {
+		trimmed := strings.TrimSpace(key)
+		if trimmed == "" {
+			continue
+		}
+		out[trimmed] = struct{}{}
+	}
+	return out
+}
+
+func publishLockSet(locks []config.PublishLockRule) map[string]config.PublishLockRule {
+	out := make(map[string]config.PublishLockRule, len(locks))
+	for _, lock := range locks {
+		source := strings.TrimSpace(lock.Source)
+		if source == "" {
+			continue
+		}
+		lock.Source = source
+		lock.Reason = strings.TrimSpace(lock.Reason)
+		out[source] = lock
+	}
+	return out
+}
+
+func resolvePublishOutputDest(rule config.PublishOutputRule, configured map[string]string) (string, error) {
+	return artifactpolicy.ResolvePublishedDestination(rule.Source, rule.Dest, configured)
+}
+
+func configuredOutputPathMapFromCatalog(catalog *artifacts.ArtifactCatalog) map[string]string {
+	out := map[string]string{}
+	if catalog == nil {
+		return out
+	}
+	for _, entry := range catalog.ListConfigured() {
+		if strings.TrimSpace(entry.ConfiguredKey) == "" {
+			continue
+		}
+		out[entry.ConfiguredKey] = strings.TrimSpace(entry.CanonicalRelPath)
+	}
+	return out
+}
+
+func buildPublishRuntimeArtifactCatalog(
+	paths artifacts.SessionPaths,
+	scriptoriumCfg *config.ScriptoriumConfig,
+) (*artifacts.ArtifactCatalog, error) {
+	catalog := artifacts.NewArtifactCatalog()
+	if err := catalog.RegisterBuiltIns(); err != nil {
+		return nil, err
+	}
+	if scriptoriumCfg == nil {
+		return catalog, nil
+	}
+
+	configured := map[string]artifacts.ConfiguredArtifactDefinition{}
+	for key, artifactCfg := range scriptoriumCfg.Artifacts {
+		configured[key] = artifacts.ConfiguredArtifactDefinition{
+			Enabled:    artifactCfg.Enabled,
+			OutputPath: artifactCfg.OutputPath,
+		}
+	}
+	if err := catalog.RegisterConfiguredArtifacts(configured, nil); err != nil {
+		return nil, err
+	}
+
+	for _, entry := range catalog.ListConfigured() {
+		if strings.TrimSpace(entry.CanonicalRelPath) == "" {
+			continue
+		}
+		localPath, err := resolveConfiguredArtifactLocalPath(paths, entry.CanonicalRelPath)
+		if err != nil {
+			continue
+		}
+		info, statErr := os.Stat(localPath)
+		if statErr != nil {
+			if os.IsNotExist(statErr) {
+				continue
+			}
+			return nil, fmt.Errorf("stat configured artifact %q: %w", entry.SourceID, statErr)
+		}
+		if info.IsDir() {
+			continue
+		}
+		if err := catalog.MarkAvailableFromDisk(entry.SourceID, localPath); err != nil {
+			return nil, err
+		}
+	}
+
+	return catalog, nil
+}
+
+func resolveConfiguredArtifactLocalPath(paths artifacts.SessionPaths, configured string) (string, error) {
+	outputPath := strings.TrimSpace(configured)
+	if outputPath == "" {
+		return "", fmt.Errorf("configured artifact output path is required")
+	}
+	if filepath.IsAbs(outputPath) {
+		return filepath.Clean(outputPath), nil
+	}
+	rel := filepath.Clean(outputPath)
+	if rel == "." || rel == "" {
+		return "", fmt.Errorf("relative output path is required")
+	}
+	if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
+		return "", fmt.Errorf("relative output path escapes session root: %q", configured)
+	}
+	return filepath.Join(paths.Root, rel), nil
+}
+
+func collectPublishRunFiles(runRoot, manifestPath string) ([]publishUploadFile, error) {
+	files := make([]publishUploadFile, 0, 64)
+	err := filepath.WalkDir(runRoot, func(path string, d fs.DirEntry, walkErr error) error {
+		if walkErr != nil {
+			return walkErr
+		}
+		if d.IsDir() {
+			if path == runRoot {
+				return nil
+			}
+			relDir, err := filepath.Rel(runRoot, path)
+			if err != nil {
+				return fmt.Errorf("relative dir from %q to %q: %w", runRoot, path, err)
+			}
+			relDir = filepath.ToSlash(relDir)
+			// Preserve existing behavior: audio is not uploaded in publish run record.
+			if relDir == "audio" || strings.HasPrefix(relDir, "audio/") {
+				return filepath.SkipDir
+			}
+			return nil
+		}
+		rel, err := filepath.Rel(runRoot, path)
+		if err != nil {
+			return fmt.Errorf("relative path from %q to %q: %w", runRoot, path, err)
+		}
+		rel = filepath.ToSlash(rel)
+		files = append(files, publishUploadFile{
+			RelativePath: rel,
+			LocalPath:    path,
+		})
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("walk %q: %w", runRoot, err)
+	}
+
+	manifestInfo, err := os.Stat(manifestPath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, fmt.Errorf("manifest.json not found (checked path %q)", manifestPath)
+		}
+		return nil, fmt.Errorf("stat %q: %w", manifestPath, err)
+	}
+	if manifestInfo.IsDir() {
+		return nil, fmt.Errorf("manifest path %q is a directory", manifestPath)
+	}
+	files = append(files, publishUploadFile{
+		RelativePath: "manifest.json",
+		LocalPath:    manifestPath,
+	})
+	seen := map[string]publishUploadFile{}
+	for _, file := range files {
+		seen[file.RelativePath] = file
+	}
+	files = files[:0]
+	for _, file := range seen {
+		files = append(files, file)
+	}
+
+	sort.Slice(files, func(i, j int) bool {
+		return files[i].RelativePath < files[j].RelativePath
+	})
+	return files, nil
+}
+
+func collectPublishPreviousFiles(previousDir string) ([]publishUploadFile, error) {
+	previousDir = filepath.Clean(strings.TrimSpace(previousDir))
+	if previousDir == "" {
+		return nil, fmt.Errorf("previous directory is required")
+	}
+	info, err := os.Stat(previousDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("stat %q: %w", previousDir, err)
+	}
+	if !info.IsDir() {
+		return nil, fmt.Errorf("previous path %q is not a directory", previousDir)
+	}
+
+	files := make([]publishUploadFile, 0, 16)
+	err = filepath.WalkDir(previousDir, func(path string, d fs.DirEntry, walkErr error) error {
+		if walkErr != nil {
+			return walkErr
+		}
+		if d.IsDir() {
+			return nil
+		}
+		rel, err := filepath.Rel(previousDir, path)
+		if err != nil {
+			return fmt.Errorf("relative path from %q to %q: %w", previousDir, path, err)
+		}
+		rel = filepath.ToSlash(rel)
+		files = append(files, publishUploadFile{
+			RelativePath: filepath.ToSlash(filepath.Join(config.PathPreviousDirSegment, rel)),
+			LocalPath:    path,
+		})
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("walk %q: %w", previousDir, err)
+	}
+
+	sort.Slice(files, func(i, j int) bool {
+		return files[i].RelativePath < files[j].RelativePath
+	})
+	return files, nil
+}
+
+func resolvePublishRunManifestSource(runRoot string) (string, error) {
+	path := filepath.Join(filepath.Clean(runRoot), "manifest.json")
+	info, err := os.Stat(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return "", fmt.Errorf("manifest.json not found in run root %q", runRoot)
+		}
+		return "", fmt.Errorf("stat %q: %w", path, err)
+	}
+	if info.IsDir() {
+		return "", fmt.Errorf("manifest path %q is a directory", path)
+	}
+	return path, nil
+}
+
+func directoryExists(path string) (bool, error) {
+	info, err := os.Stat(path)
+	if err == nil {
+		return info.IsDir(), nil
+	}
+	if os.IsNotExist(err) {
+		return false, nil
+	}
+	return false, err
+}
+
+func writeCurrentManifestSnapshot(m *manifest.Manifest, publishMetadata map[string]any) (string, error) {
+	if m == nil {
+		return "", fmt.Errorf("manifest is required")
+	}
+	clone := *m
+	clone.Stages = make(map[string]*manifest.StageRecord, len(m.Stages))
+	for name, sr := range m.Stages {
+		if sr == nil {
+			continue
+		}
+		stageCopy := *sr
+		if sr.Outputs != nil {
+			stageCopy.Outputs = append([]manifest.ArtifactRecord(nil), sr.Outputs...)
+		}
+		if sr.Logs != nil {
+			stageCopy.Logs = append([]string(nil), sr.Logs...)
+		}
+		if sr.GeneratedConfigs != nil {
+			stageCopy.GeneratedConfigs = append([]string(nil), sr.GeneratedConfigs...)
+		}
+		if sr.Metadata != nil {
+			metaCopy := make(map[string]any, len(sr.Metadata))
+			for k, v := range sr.Metadata {
+				metaCopy[k] = v
+			}
+			stageCopy.Metadata = metaCopy
+		}
+		clone.Stages[name] = &stageCopy
+	}
+
+	now := time.Now().UTC()
+	clone.MarkStageSucceeded("publish", now, nil)
+	if sr := clone.Stages["publish"]; sr != nil {
+		sr.Metadata = publishMetadata
+	}
+
+	data, err := json.MarshalIndent(&clone, "", "  ")
+	if err != nil {
+		return "", fmt.Errorf("marshal manifest: %w", err)
+	}
+	data = append(data, '\n')
+
+	tmp, err := os.CreateTemp("", "narratio-current-manifest-*.json")
+	if err != nil {
+		return "", fmt.Errorf("create temp file: %w", err)
+	}
+	path := tmp.Name()
+	if _, err := tmp.Write(data); err != nil {
+		_ = tmp.Close()
+		return "", fmt.Errorf("write temp manifest: %w", err)
+	}
+	if err := tmp.Close(); err != nil {
+		return "", fmt.Errorf("close temp manifest: %w", err)
+	}
+	return path, nil
+}
+
+func writeCurrentRunIDPointer(runID string) (string, error) {
+	tmp, err := os.CreateTemp("", "narratio-current-run-id-*.txt")
+	if err != nil {
+		return "", fmt.Errorf("create temp file: %w", err)
+	}
+	path := tmp.Name()
+	if _, err := tmp.WriteString(runID + "\n"); err != nil {
+		_ = tmp.Close()
+		return "", fmt.Errorf("write temp run id pointer: %w", err)
+	}
+	if err := tmp.Close(); err != nil {
+		return "", fmt.Errorf("close temp run id pointer: %w", err)
+	}
+	return path, nil
+}
+
+func publishMetadataPreview(
+	bucket, runPrefix, sessionPrefix string,
+	runUploaded []string,
+	publishedUploaded []string,
+	previousUploaded []string,
+	skippedOptionalOutputs []string,
+	skippedUnselectedOutputs []publishSkippedUnselectedOutput,
+	lockedOutputs []publishLockedOutput,
+	currentManifestKey string,
+) map[string]any {
+	return map[string]any{
+		"stage":                      "publish",
+		"uploaded":                   true,
+		"s3_bucket":                  bucket,
+		"s3_run_prefix":              runPrefix,
+		"run_files_uploaded":         len(runUploaded),
+		"run_uploaded_paths":         append([]string(nil), runUploaded...),
+		"published_files_uploaded":   len(publishedUploaded),
+		"published_paths":            append([]string(nil), publishedUploaded...),
+		"previous_files_uploaded":    len(previousUploaded),
+		"previous_uploaded_paths":    append([]string(nil), previousUploaded...),
+		"skipped_optional_outputs":   append([]string(nil), skippedOptionalOutputs...),
+		"skipped_unselected_outputs": skippedUnselectedOutputMetadata(skippedUnselectedOutputs),
+		"locked_output_count":        len(lockedOutputs),
+		"locked_outputs":             lockedOutputMetadata(lockedOutputs),
+		"current_manifest_key":       currentManifestKey,
+		"current_run_id_key":         artifacts.S3CurrentRunPointerKey(sessionPrefix),
+		"current_pointer_written":    false,
+		"audio_upload_skipped":       true,
+	}
+}
+
+func skippedUnselectedOutputMetadata(skipped []publishSkippedUnselectedOutput) []map[string]any {
+	out := make([]map[string]any, 0, len(skipped))
+	for _, item := range skipped {
+		out = append(out, map[string]any{
+			"source":   item.Source,
+			"dest":     item.Dest,
+			"required": item.Required,
+		})
+	}
+	return out
+}
+
+func lockedOutputMetadata(locked []publishLockedOutput) []map[string]any {
+	out := make([]map[string]any, 0, len(locked))
+	for _, item := range locked {
+		out = append(out, map[string]any{
+			"source":     item.Source,
+			"dest":       item.Dest,
+			"remote_key": item.RemoteKey,
+			"reason":     item.Reason,
+			"required":   item.Required,
+			"local_path": item.LocalPath,
+			"provenance": item.Provenance,
+		})
+	}
+	return out
+}