narratio/internal/stage/publish.go

package stage

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strings"

	"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/fileops"
	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
	"gitea.maximumdirect.net/eric/narratio/internal/pathsafe"
)

type publishStage struct{}
type publishUploadFile struct {
	RelativePath       string
	SourceRelativePath string
	Checksum           string
}

var publishPrerequisiteStages = []string{
	"prepare",
	"transcribe",
	"merge",
	"polish",
	"normalize",
	"trim",
	"render",
	"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,
			},
		}, nil
	}
	if publishRunUploadDisabled(env) {
		return &StageResult{
			Metadata: map[string]any{
				"stage":                "publish",
				"skipped":              true,
				"upload_run_enabled":   false,
				"audio_upload_skipped": true,
			},
		}, 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")
	}

	runManifestPath, err := resolvePublishRunManifestPath(env, m)
	if err != nil {
		return nil, fmt.Errorf("publish: resolve run manifest path: %w", err)
	}
	runRoot := filepath.Dir(runManifestPath)
	runRootInfo, err := os.Lstat(runRoot)
	if err != nil {
		return nil, fmt.Errorf("publish: run root %q: %w", runRoot, err)
	}
	if runRootInfo.Mode()&os.ModeSymlink != 0 || !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")
	}

	runManifestFile, err := fileops.OpenConfinedRegularFile(runRoot, filepath.Base(runManifestPath))
	if err != nil {
		return nil, fmt.Errorf("publish: open run manifest: %w", err)
	}
	defer runManifestFile.Close()
	runManifest, err := (&manifest.LocalStore{}).LoadRunReader(ctx, runManifestFile)
	if err != nil {
		return nil, fmt.Errorf("publish: load run manifest: %w", err)
	}
	if _, err := runManifestFile.Seek(0, io.SeekStart); err != nil {
		return nil, fmt.Errorf("publish: rewind run manifest: %w", err)
	}

	runFiles, err := collectPublishRunFiles(runRoot, runManifest)
	if err != nil {
		return nil, fmt.Errorf("publish: collect run files: %w", err)
	}
	runSources, err := openPublishRunFiles(runRoot, runFiles, runManifestFile)
	if err != nil {
		return nil, fmt.Errorf("publish: verify run files: %w", err)
	}
	defer closePublishSources(runSources)
	uploader, ok := env.ObjectStore.(storage.ReaderUploader)
	if !ok {
		return nil, fmt.Errorf("publish: object store does not support verified source uploads")
	}
	sessionPaths := publishSessionPaths(env, m)
	previousFiles, previousManifestFile, err := collectPublishPreviousFiles(ctx, sessionPaths.PreviousDir)
	if err != nil {
		return nil, fmt.Errorf("publish: collect previous files: %w", err)
	}
	if previousManifestFile != nil {
		defer previousManifestFile.Close()
	}
	previousSources, err := openPublishRunFiles(sessionPaths.PreviousDir, previousFiles, previousManifestFile)
	if err != nil {
		return nil, fmt.Errorf("publish: verify previous files: %w", err)
	}
	defer closePublishSources(previousSources)
	runtimeCatalog, err := buildPublishRuntimeArtifactCatalog(
		sessionPaths,
		m,
		env.Config.Pipeline.Scriptorium,
		env.Config.Pipeline.Notarius,
	)
	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,
		runPrefix,
	)
	if err != nil {
		return nil, fmt.Errorf("publish: resolve publish output rules: %w", err)
	}

	runUploaded := publishUploadPaths(runFiles)
	publishedUploaded := publishOutputPaths(publishOutputs)
	previousUploaded := publishUploadPaths(previousFiles)
	commitKey := artifacts.S3RunCommitKey(sessionPrefix, runID)
	pointerKey := artifacts.S3CurrentCommitPointerKey(sessionPrefix)

	transfers, err := buildPublishTransferPlan(runPrefix, runFiles, runSources, previousFiles, previousSources, publishOutputs)
	if err != nil {
		return nil, fmt.Errorf("publish: build immutable upload mapping: %w", err)
	}
	defer func() { closePublishTransfers(transfers) }()

	manifestTempPath, err := writeCommittedSessionManifest(m, publishMetadata(
		bucket,
		runPrefix,
		runID,
		runUploaded,
		publishedUploaded,
		previousUploaded,
		skippedOptionalOutputs,
		skippedUnselectedOutputs,
		lockedOutputs,
		commitKey,
		pointerKey,
	))
	if err != nil {
		return nil, fmt.Errorf("publish: build committed session manifest: %w", err)
	}
	if err := appendCommittedSessionManifest(&transfers, manifestTempPath, artifacts.S3RunSessionManifestKey(sessionPrefix, runID)); err != nil {
		return nil, fmt.Errorf("publish: add committed session manifest: %w", err)
	}
	if err := validatePublishTransferPlan(transfers, commitKey); err != nil {
		return nil, fmt.Errorf("publish: validate immutable upload mapping: %w", err)
	}

	commit := artifacts.RemoteCommitManifest{
		FormatVersion: artifacts.RemoteCommitFormatVersion,
		Campaign:      strings.TrimSpace(m.Campaign),
		SessionID:     strings.TrimSpace(m.SessionID),
		RunID:         runID,
		Artifacts:     make([]artifacts.RemoteArtifact, 0, len(transfers)),
	}
	for index := range transfers {
		artifact, err := uploadVerifiedPublishTransfer(ctx, env.ObjectStore, uploader, &transfers[index])
		if err != nil {
			return nil, fmt.Errorf("publish: upload immutable object %q: %w", transfers[index].Artifact.DestinationKey, err)
		}
		commit.Artifacts = append(commit.Artifacts, artifact)
	}
	commitData, err := artifacts.EncodeRemoteCommitManifest(commit)
	if err != nil {
		return nil, fmt.Errorf("publish: encode immutable commit: %w", err)
	}
	commitInfo, err := uploadVerifiedPublishBytes(ctx, env.ObjectStore, uploader, commitKey, commitData, "application/json")
	if err != nil {
		return nil, fmt.Errorf("publish: upload immutable commit %q: %w", commitKey, err)
	}
	if err := revalidatePublishCommitLocks(ctx, env, publishOutputs); err != nil {
		return nil, fmt.Errorf("publish: revalidate locks before current commit selection: %w", err)
	}
	pointerData, err := artifacts.EncodeCurrentCommitPointer(artifacts.CurrentCommitPointer{
		FormatVersion:    artifacts.RemoteCommitFormatVersion,
		Campaign:         commit.Campaign,
		SessionID:        commit.SessionID,
		RunID:            commit.RunID,
		CommitKey:        commitKey,
		CommitSHA256:     publishSHA256(commitData),
		CommitSize:       int64(len(commitData)),
		CommitGeneration: commitInfo.ETag,
	})
	if err != nil {
		return nil, fmt.Errorf("publish: encode current commit pointer: %w", err)
	}
	if _, err := uploader.UploadReader(ctx, bytes.NewReader(pointerData), pointerKey, storage.UploadOptions{ContentType: "application/json"}); err != nil {
		return nil, fmt.Errorf("publish: update current commit pointer %q: %w", pointerKey, 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),
			"remote_commit_key":          commitKey,
			"current_commit_pointer_key": pointerKey,
			"audio_upload_skipped":       true,
		},
	}, nil
}

func revalidatePublishCommitLocks(ctx context.Context, env *Env, outputs []publishOutput) error {
	if env == nil || env.RevalidatePublishLocks == nil || len(outputs) == 0 {
		return nil
	}
	locks, err := env.RevalidatePublishLocks(ctx)
	if err != nil {
		return err
	}
	lockSet := publishLockSet(locks)
	for _, output := range outputs {
		if lock, locked := lockSet[output.Source]; locked {
			return fmt.Errorf("source %q is locked: %s", output.Source, strings.TrimSpace(lock.Reason))
		}
	}
	return nil
}

type publishOutput struct {
	Source     string
	Dest       string
	Required   bool
	LocalPath  string
	Provenance string
}

type publishTransfer struct {
	Artifact      artifacts.RemoteArtifact
	Source        *os.File
	TemporaryPath 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 buildPublishTransferPlan(
	runPrefix string,
	runFiles []publishUploadFile,
	runSources []*os.File,
	previousFiles []publishUploadFile,
	previousSources []*os.File,
	publishOutputs []publishOutput,
) ([]publishTransfer, error) {
	if len(runFiles) != len(runSources) || len(previousFiles) != len(previousSources) {
		return nil, fmt.Errorf("verified publish sources do not match their declared files")
	}
	transfers := make([]publishTransfer, 0, len(runFiles)+len(previousFiles)+len(publishOutputs))
	add := func(artifact artifacts.RemoteArtifact, source *os.File) error {
		if source == nil {
			return fmt.Errorf("source for %q is required", artifact.DestinationKey)
		}
		transfers = append(transfers, publishTransfer{Artifact: artifact, Source: source})
		return nil
	}
	for index, file := range runFiles {
		artifactType := artifacts.RemoteArtifactTypeRunFile
		source := "run:" + file.RelativePath
		if file.RelativePath == config.S3ManifestFile {
			artifactType = artifacts.RemoteArtifactTypeRunManifest
			source = "run.manifest"
		}
		if err := add(artifacts.RemoteArtifact{
			Type:           artifactType,
			Source:         source,
			DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath),
		}, runSources[index]); err != nil {
			closePublishTransfers(transfers)
			return nil, err
		}
	}
	for index, file := range previousFiles {
		if err := add(artifacts.RemoteArtifact{
			Type:           artifacts.RemoteArtifactTypePreviousArtifact,
			Source:         "previous:" + file.SourceRelativePath,
			DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath),
		}, previousSources[index]); err != nil {
			closePublishTransfers(transfers)
			return nil, err
		}
	}
	for _, output := range publishOutputs {
		source, err := openPublishOutputSource(output.LocalPath)
		if err != nil {
			closePublishTransfers(transfers)
			return nil, fmt.Errorf("open output source %q: %w", output.Source, err)
		}
		if err := add(artifacts.RemoteArtifact{
			Type:           artifacts.RemoteArtifactTypePublishedOutput,
			Source:         output.Source,
			DestinationKey: publishOutputRemoteKey(runPrefix, output.Dest),
		}, source); err != nil {
			_ = source.Close()
			closePublishTransfers(transfers)
			return nil, err
		}
	}
	sort.Slice(transfers, func(i, j int) bool {
		return transfers[i].Artifact.DestinationKey < transfers[j].Artifact.DestinationKey
	})
	return transfers, nil
}

func appendCommittedSessionManifest(transfers *[]publishTransfer, localPath, destinationKey string) error {
	if transfers == nil {
		return fmt.Errorf("transfer plan is required")
	}
	source, err := os.Open(localPath)
	if err != nil {
		return err
	}
	*transfers = append(*transfers, publishTransfer{
		Artifact: artifacts.RemoteArtifact{
			Type:           artifacts.RemoteArtifactTypeSessionManifest,
			Source:         "session.manifest",
			DestinationKey: destinationKey,
		},
		Source:        source,
		TemporaryPath: localPath,
	})
	sort.Slice(*transfers, func(i, j int) bool {
		return (*transfers)[i].Artifact.DestinationKey < (*transfers)[j].Artifact.DestinationKey
	})
	return nil
}

func validatePublishTransferPlan(transfers []publishTransfer, commitKey string) error {
	artifactsToUpload := make([]artifacts.RemoteArtifact, 0, len(transfers))
	for _, transfer := range transfers {
		if transfer.Artifact.DestinationKey == commitKey {
			return fmt.Errorf("artifact destination %q is reserved for the immutable commit", commitKey)
		}
		artifactsToUpload = append(artifactsToUpload, transfer.Artifact)
	}
	return artifacts.ValidateRemoteArtifactMapping(artifactsToUpload)
}

func openPublishOutputSource(localPath string) (*os.File, error) {
	path := filepath.Clean(strings.TrimSpace(localPath))
	if path == "" || path == "." {
		return nil, fmt.Errorf("local path is required")
	}
	return fileops.OpenConfinedRegularFile(filepath.Dir(path), filepath.Base(path))
}

func publishOutputRemoteKey(runPrefix, destination string) string {
	return artifacts.S3RunRelativeDestinationKey(runPrefix, filepath.ToSlash(filepath.Join("published", destination)))
}

func uploadVerifiedPublishTransfer(
	ctx context.Context,
	store storage.ObjectStore,
	uploader storage.ReaderUploader,
	transfer *publishTransfer,
) (artifacts.RemoteArtifact, error) {
	if transfer == nil || transfer.Source == nil {
		return artifacts.RemoteArtifact{}, fmt.Errorf("transfer source is required")
	}
	if err := preparePublishTransfer(transfer); err != nil {
		return artifacts.RemoteArtifact{}, err
	}
	if existing, found, err := verifyExistingImmutableObject(ctx, store, transfer.Artifact.DestinationKey, transfer.Artifact.SHA256, transfer.Artifact.Size); err != nil {
		return artifacts.RemoteArtifact{}, err
	} else if found {
		transfer.Artifact.Generation = existing.ETag
		return transfer.Artifact, nil
	}
	info, err := uploader.UploadReader(ctx, transfer.Source, transfer.Artifact.DestinationKey, storage.UploadOptions{})
	if err != nil {
		return artifacts.RemoteArtifact{}, err
	}
	verified, err := verifyPublishedObject(ctx, store, transfer.Artifact.DestinationKey, transfer.Artifact.Size, info.ETag)
	if err != nil {
		return artifacts.RemoteArtifact{}, err
	}
	transfer.Artifact.Generation = verified.ETag
	return transfer.Artifact, nil
}

func uploadVerifiedPublishBytes(
	ctx context.Context,
	store storage.ObjectStore,
	uploader storage.ReaderUploader,
	key string,
	data []byte,
	contentType string,
) (storage.ObjectInfo, error) {
	checksum := publishSHA256(data)
	if existing, found, err := verifyExistingImmutableObject(ctx, store, key, checksum, int64(len(data))); err != nil {
		return storage.ObjectInfo{}, err
	} else if found {
		return existing, nil
	}
	info, err := uploader.UploadReader(ctx, bytes.NewReader(data), key, storage.UploadOptions{ContentType: contentType})
	if err != nil {
		return storage.ObjectInfo{}, err
	}
	return verifyPublishedObject(ctx, store, key, int64(len(data)), info.ETag)
}

func verifyExistingImmutableObject(ctx context.Context, store storage.ObjectStore, key, wantSHA256 string, wantSize int64) (storage.ObjectInfo, bool, error) {
	exists, err := store.Exists(ctx, key)
	if err != nil {
		return storage.ObjectInfo{}, false, fmt.Errorf("check immutable object %q: %w", key, err)
	}
	if !exists {
		return storage.ObjectInfo{}, false, nil
	}
	path, err := storage.DownloadObjectToTemp(ctx, store, key, "narratio-verify-immutable-*.bin")
	if err != nil {
		return storage.ObjectInfo{}, false, fmt.Errorf("download existing immutable object %q: %w", key, err)
	}
	defer func() { _ = os.Remove(path) }()
	file, err := os.Open(path)
	if err != nil {
		return storage.ObjectInfo{}, false, fmt.Errorf("open existing immutable object %q: %w", key, err)
	}
	hash := sha256.New()
	size, copyErr := io.Copy(hash, file)
	closeErr := file.Close()
	if copyErr != nil {
		return storage.ObjectInfo{}, false, fmt.Errorf("checksum existing immutable object %q: %w", key, copyErr)
	}
	if closeErr != nil {
		return storage.ObjectInfo{}, false, fmt.Errorf("close existing immutable object %q: %w", key, closeErr)
	}
	if size != wantSize || hex.EncodeToString(hash.Sum(nil)) != wantSHA256 {
		return storage.ObjectInfo{}, false, fmt.Errorf("immutable object %q already exists with different content", key)
	}
	info, err := verifyPublishedObject(ctx, store, key, wantSize, "")
	if err != nil {
		return storage.ObjectInfo{}, false, err
	}
	return info, true, nil
}

func preparePublishTransfer(transfer *publishTransfer) error {
	if _, err := transfer.Source.Seek(0, io.SeekStart); err != nil {
		return fmt.Errorf("rewind source %q: %w", transfer.Artifact.Source, err)
	}
	info, err := transfer.Source.Stat()
	if err != nil {
		return fmt.Errorf("stat source %q: %w", transfer.Artifact.Source, err)
	}
	if !info.Mode().IsRegular() {
		return fmt.Errorf("source %q is not a regular file", transfer.Artifact.Source)
	}
	hash := sha256.New()
	if _, err := io.Copy(hash, transfer.Source); err != nil {
		return fmt.Errorf("checksum source %q: %w", transfer.Artifact.Source, err)
	}
	if _, err := transfer.Source.Seek(0, io.SeekStart); err != nil {
		return fmt.Errorf("rewind source %q after checksum: %w", transfer.Artifact.Source, err)
	}
	transfer.Artifact.Size = info.Size()
	transfer.Artifact.SHA256 = hex.EncodeToString(hash.Sum(nil))
	return nil
}

func verifyPublishedObject(ctx context.Context, store storage.ObjectStore, key string, wantSize int64, uploadedGeneration string) (storage.ObjectInfo, error) {
	objects, err := store.List(ctx, key)
	if err != nil {
		return storage.ObjectInfo{}, fmt.Errorf("list uploaded object %q: %w", key, err)
	}
	var found *storage.ObjectInfo
	for _, object := range objects {
		if object.Key != key {
			continue
		}
		if found != nil {
			return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q is ambiguous", key)
		}
		copy := object
		found = &copy
	}
	if found == nil {
		return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q is missing", key)
	}
	if found.Size != wantSize {
		return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q size is %d, want %d", key, found.Size, wantSize)
	}
	if strings.TrimSpace(found.ETag) == "" {
		return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q has no generation", key)
	}
	if strings.TrimSpace(uploadedGeneration) != "" && found.ETag != uploadedGeneration {
		return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q generation changed during verification", key)
	}
	return *found, nil
}

func closePublishTransfers(transfers []publishTransfer) {
	for _, transfer := range transfers {
		if transfer.Source != nil {
			_ = transfer.Source.Close()
		}
		if transfer.TemporaryPath != "" {
			_ = os.Remove(transfer.TemporaryPath)
		}
	}
}

func publishUploadPaths(files []publishUploadFile) []string {
	paths := make([]string, 0, len(files))
	for _, file := range files {
		paths = append(paths, file.RelativePath)
	}
	return paths
}

func publishOutputPaths(outputs []publishOutput) []string {
	paths := make([]string, 0, len(outputs))
	for _, output := range outputs {
		paths = append(paths, output.Dest)
	}
	return paths
}

func publishSHA256(data []byte) string {
	sum := sha256.Sum256(data)
	return hex.EncodeToString(sum[:])
}

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 resolvePublishRunManifestPath(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.SessionRunManifestPathForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID))
	canonicalExists, err := directoryExists(filepath.Dir(canonical))
	if err != nil {
		return "", fmt.Errorf("check canonical run manifest directory %q: %w", filepath.Dir(canonical), err)
	}
	if !canonicalExists {
		return "", fmt.Errorf("run manifest directory not found for campaign %q session %q run %q at canonical path %q", campaign, sessionID, runID, filepath.Dir(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,
	runPrefix 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)
	extractionOutputs := extractionOutputSetFromCatalog(catalog)
	for _, rule := range rules {
		source := strings.TrimSpace(rule.Source)
		required := rule.Required == nil || *rule.Required
		dest, err := resolvePublishOutputDest(rule, configuredOutputs, extractionOutputs)
		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: publishOutputRemoteKey(runPrefix, 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:  publishOutputRemoteKey(runPrefix, 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, extractions map[string]struct{}) (string, error) {
	return artifactpolicy.ResolvePublishedDestinationWithExtractions(rule.Source, rule.Dest, configured, extractions)
}

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 extractionOutputSetFromCatalog(catalog *artifacts.ArtifactCatalog) map[string]struct{} {
	out := map[string]struct{}{}
	if catalog == nil {
		return out
	}
	for _, entry := range catalog.ListExtraction() {
		if strings.TrimSpace(entry.ExtractionKey) != "" {
			out[entry.ExtractionKey] = struct{}{}
		}
	}
	return out
}

func buildPublishRuntimeArtifactCatalog(
	paths artifacts.SessionPaths,
	m *manifest.Manifest,
	scriptoriumCfg *config.ScriptoriumConfig,
	notariusCfg *config.NotariusConfig,
) (*artifacts.ArtifactCatalog, error) {
	catalog := artifacts.NewArtifactCatalog()
	if err := catalog.RegisterBuiltIns(); err != nil {
		return nil, err
	}
	extractionDefinitions := artifacts.ExtractionDefinitionsFromConfig(notariusCfg)
	if err := catalog.RegisterExtractionArtifacts(extractionDefinitions); err != nil {
		return nil, err
	}
	if notariusCfg != nil && notariusCfg.Enabled {
		catalog.HydrateExtractionArtifacts(paths, m, extractionDefinitions)
	}
	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 string, runManifest *manifest.RunManifest) ([]publishUploadFile, error) {
	if runManifest == nil {
		return nil, fmt.Errorf("run manifest is required")
	}
	files := map[string]publishUploadFile{
		"manifest.json": {RelativePath: "manifest.json", SourceRelativePath: "manifest.json"},
	}
	add := func(path, checksum string) error {
		if strings.TrimSpace(path) == "" {
			return nil
		}
		rel, err := pathsafe.SlashRelativeFromRoot(runRoot, path)
		if err != nil {
			return nil
		}
		candidate := publishUploadFile{RelativePath: rel, SourceRelativePath: rel, Checksum: strings.TrimSpace(checksum)}
		if prior, exists := files[rel]; exists && prior.Checksum != "" && candidate.Checksum != "" && prior.Checksum != candidate.Checksum {
			return fmt.Errorf("source %q has conflicting declared checksums", rel)
		}
		if prior, exists := files[rel]; exists && prior.Checksum != "" {
			candidate.Checksum = prior.Checksum
		}
		files[rel] = candidate
		return nil
	}
	for _, record := range runManifest.Stages {
		if record == nil {
			continue
		}
		for _, output := range record.Outputs {
			if err := add(output.ArchivePath, output.Checksum); err != nil {
				return nil, err
			}
			if output.ArchivePath == "" {
				if err := add(output.LocalPath, output.Checksum); err != nil {
					return nil, err
				}
			}
		}
		for _, path := range append(append([]string(nil), record.Logs...), record.GeneratedConfigs...) {
			if err := add(path, ""); err != nil {
				return nil, err
			}
		}
	}
	out := make([]publishUploadFile, 0, len(files))
	for _, file := range files {
		out = append(out, file)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].RelativePath < out[j].RelativePath })
	return out, nil
}

func collectPublishPreviousFiles(ctx context.Context, previousDir string) ([]publishUploadFile, *os.File, error) {
	previousDir = filepath.Clean(strings.TrimSpace(previousDir))
	if previousDir == "" || previousDir == "." {
		return nil, nil, fmt.Errorf("previous directory is required")
	}
	info, err := os.Lstat(previousDir)
	if errors.Is(err, os.ErrNotExist) {
		return nil, nil, nil
	}
	if err != nil {
		return nil, nil, fmt.Errorf("inspect previous directory: %w", err)
	}
	if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
		return nil, nil, fmt.Errorf("previous directory is not a regular directory")
	}

	manifestFile, err := fileops.OpenConfinedRegularFile(previousDir, "manifest.json")
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return nil, nil, nil
		}
		return nil, nil, fmt.Errorf("open previous manifest: %w", err)
	}
	previousManifest, err := (&manifest.LocalStore{}).LoadReader(ctx, manifestFile)
	if err != nil {
		_ = manifestFile.Close()
		return nil, nil, fmt.Errorf("load previous manifest: %w", err)
	}
	if _, err := manifestFile.Seek(0, io.SeekStart); err != nil {
		_ = manifestFile.Close()
		return nil, nil, fmt.Errorf("rewind previous manifest: %w", err)
	}

	files := map[string]publishUploadFile{
		"manifest.json": {RelativePath: filepath.ToSlash(filepath.Join(config.PathPreviousDirSegment, "manifest.json")), SourceRelativePath: "manifest.json"},
	}
	add := func(path, checksum string) error {
		if strings.TrimSpace(path) == "" {
			return nil
		}
		rel, err := pathsafe.SlashRelativeFromRoot(previousDir, path)
		if err != nil {
			return nil
		}
		archivePath := filepath.ToSlash(filepath.Join(config.PathPreviousDirSegment, rel))
		candidate := publishUploadFile{RelativePath: archivePath, SourceRelativePath: rel, Checksum: strings.TrimSpace(checksum)}
		if prior, exists := files[rel]; exists && prior.Checksum != "" && candidate.Checksum != "" && prior.Checksum != candidate.Checksum {
			return fmt.Errorf("source %q has conflicting declared checksums", rel)
		}
		if prior, exists := files[rel]; exists && prior.Checksum != "" {
			candidate.Checksum = prior.Checksum
		}
		files[rel] = candidate
		return nil
	}
	for _, artifact := range previousManifest.Artifacts {
		if err := add(artifact.ArchivePath, artifact.Checksum); err != nil {
			_ = manifestFile.Close()
			return nil, nil, err
		}
		if artifact.ArchivePath == "" {
			if err := add(artifact.LocalPath, artifact.Checksum); err != nil {
				_ = manifestFile.Close()
				return nil, nil, err
			}
		}
	}
	for _, record := range previousManifest.Stages {
		if record == nil {
			continue
		}
		for _, artifact := range record.Outputs {
			if err := add(artifact.ArchivePath, artifact.Checksum); err != nil {
				_ = manifestFile.Close()
				return nil, nil, err
			}
			if artifact.ArchivePath == "" {
				if err := add(artifact.LocalPath, artifact.Checksum); err != nil {
					_ = manifestFile.Close()
					return nil, nil, err
				}
			}
		}
		for _, path := range append(append([]string(nil), record.Logs...), record.GeneratedConfigs...) {
			if err := add(path, ""); err != nil {
				_ = manifestFile.Close()
				return nil, nil, err
			}
		}
	}
	out := make([]publishUploadFile, 0, len(files))
	for _, file := range files {
		out = append(out, file)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].RelativePath < out[j].RelativePath })
	return out, manifestFile, nil
}

func openPublishRunFiles(runRoot string, files []publishUploadFile, manifestFile *os.File) ([]*os.File, error) {
	sources := make([]*os.File, 0, len(files))
	for _, candidate := range files {
		if manifestFile == nil {
			closePublishSources(sources)
			return nil, fmt.Errorf("archive manifest is required")
		}
		file := manifestFile
		if candidate.SourceRelativePath != "manifest.json" {
			var err error
			file, err = fileops.OpenConfinedRegularFile(runRoot, candidate.SourceRelativePath)
			if err != nil {
				closePublishSources(sources)
				return nil, err
			}
		}
		if err := verifyPublishSourceChecksum(file, candidate.RelativePath, candidate.Checksum); err != nil {
			if file != manifestFile {
				_ = file.Close()
			}
			closePublishSources(sources)
			return nil, err
		}
		sources = append(sources, file)
	}
	return sources, nil
}

func verifyPublishSourceChecksum(file *os.File, relativePath, expected string) error {
	if strings.TrimSpace(expected) == "" {
		return nil
	}
	hash := sha256.New()
	if _, err := io.Copy(hash, file); err != nil {
		return fmt.Errorf("read source %q for checksum: %w", relativePath, err)
	}
	if _, err := file.Seek(0, io.SeekStart); err != nil {
		return fmt.Errorf("rewind source %q after checksum: %w", relativePath, err)
	}
	if actual := hex.EncodeToString(hash.Sum(nil)); !strings.EqualFold(actual, expected) {
		return fmt.Errorf("source %q does not match its declared checksum", relativePath)
	}
	return nil
}

func closePublishSources(sources []*os.File) {
	for _, source := range sources {
		_ = source.Close()
	}
}

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 writeCommittedSessionManifest(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
	}

	clone.MarkStageSucceeded("publish", clone.UpdatedAt, 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-committed-session-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 publishMetadata(
	bucket, runPrefix, runID string,
	runUploaded []string,
	publishedUploaded []string,
	previousUploaded []string,
	skippedOptionalOutputs []string,
	skippedUnselectedOutputs []publishSkippedUnselectedOutput,
	lockedOutputs []publishLockedOutput,
	commitKey string,
	pointerKey string,
) map[string]any {
	return map[string]any{
		"stage":                      "publish",
		"uploaded":                   true,
		"s3_bucket":                  bucket,
		"s3_run_prefix":              runPrefix,
		"published_run_id":           runID,
		"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),
		"remote_commit_key":          commitKey,
		"current_commit_pointer_key": pointerKey,
		"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
}