Promote current session artifacts to storage

2026-05-16 15:01:01 +00:00
parent d40c91acde
commit f7f8f1a949
7 changed files with 555 additions and 104 deletions
--- a/internal/stage/archive.go
+++ b/internal/stage/archive.go
@@ -2,15 +2,18 @@ package stage

 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"io/fs"
 	"os"
 	"path/filepath"
 	"sort"
 	"strings"
+	"time"

 	"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
 	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
 	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
 )

@@ -53,20 +56,22 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 	if archiveDisabled(env) {
 		return &StageResult{
 			Metadata: map[string]any{
-				"stage":                "archive",
-				"skipped":              true,
-				"archive_enabled":      false,
-				"audio_upload_skipped": true,
+				"stage":                   "archive",
+				"skipped":                 true,
+				"archive_enabled":         false,
+				"audio_upload_skipped":    true,
+				"current_pointer_written": false,
 			},
 		}, nil
 	}
 	if archiveRunUploadDisabled(env) {
 		return &StageResult{
 			Metadata: map[string]any{
-				"stage":                "archive",
-				"skipped":              true,
-				"upload_run_enabled":   false,
-				"audio_upload_skipped": true,
+				"stage":                   "archive",
+				"skipped":                 true,
+				"upload_run_enabled":      false,
+				"audio_upload_skipped":    true,
+				"current_pointer_written": false,
 			},
 		}, nil
 	}
@@ -82,11 +87,11 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 	if err != nil {
 		return nil, fmt.Errorf("archive: resolve local workdir: %w", err)
 	}
-	info, err := os.Stat(workDir)
+	workDirInfo, err := os.Stat(workDir)
 	if err != nil {
 		return nil, fmt.Errorf("archive: local workdir %q: %w", workDir, err)
 	}
-	if !info.IsDir() {
+	if !workDirInfo.IsDir() {
 		return nil, fmt.Errorf("archive: local workdir %q is not a directory", workDir)
 	}

@@ -94,39 +99,122 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 	if err != nil {
 		return nil, fmt.Errorf("archive: resolve s3 run prefix: %w", err)
 	}
+	sessionPrefix, err := archiveSessionPrefix(env, m)
+	if err != nil {
+		return nil, fmt.Errorf("archive: resolve s3 session prefix: %w", err)
+	}
 	bucket := archiveBucket(env, m)
 	if bucket == "" {
 		return nil, fmt.Errorf("archive: resolve s3 bucket: bucket is required")
 	}
+	runID := strings.TrimSpace(m.RunID)
+	if runID == "" {
+		return nil, fmt.Errorf("archive: run id is required")
+	}

-	relFiles, err := collectArchiveRunFiles(workDir)
+	runFiles, err := collectArchiveRunFiles(workDir)
 	if err != nil {
 		return nil, fmt.Errorf("archive: collect run files: %w", err)
 	}
+	promotions, err := resolveArchivePromotions(workDir, env.Config.Pipeline.Archive.PromoteArtifacts)
+	if err != nil {
+		return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
+	}
+	currentManifestSource := filepath.Join(workDir, "manifest.json")
+	if info, err := os.Stat(currentManifestSource); err != nil {
+		return nil, fmt.Errorf("archive: current manifest source %q: %w", currentManifestSource, err)
+	} else if info.IsDir() {
+		return nil, fmt.Errorf("archive: current manifest source %q is a directory", currentManifestSource)
+	}

-	uploaded := make([]string, 0, len(relFiles))
-	for _, rel := range relFiles {
+	runUploaded := make([]string, 0, len(runFiles))
+	for _, rel := range runFiles {
 		localPath := filepath.Join(workDir, filepath.FromSlash(rel))
 		key := artifacts.S3RunRelativeDestinationKey(runPrefix, rel)
 		if _, err := env.ObjectStore.Upload(ctx, localPath, key, storage.UploadOptions{}); err != nil {
-			return nil, fmt.Errorf("archive: upload %q to %q: %w", rel, key, err)
+			return nil, fmt.Errorf("archive: upload run file %q to %q: %w", rel, key, err)
 		}
-		uploaded = append(uploaded, rel)
+		runUploaded = append(runUploaded, rel)
+	}
+
+	promotedUploaded := make([]string, 0, len(promotions))
+	skippedOptional := make([]string, 0)
+	for _, promotion := range promotions {
+		if !promotion.Exists {
+			if promotion.Required {
+				return nil, fmt.Errorf("archive: required promotion source missing: %q", promotion.From)
+			}
+			skippedOptional = append(skippedOptional, promotion.To)
+			continue
+		}
+		key := artifacts.S3PromotedArtifactKey(sessionPrefix, promotion.To)
+		if _, err := env.ObjectStore.Upload(ctx, promotion.LocalPath, key, storage.UploadOptions{}); err != nil {
+			return nil, fmt.Errorf("archive: upload promoted output %q to %q: %w", promotion.From, key, err)
+		}
+		promotedUploaded = append(promotedUploaded, promotion.To)
+	}
+
+	currentManifestKey := artifacts.S3CurrentManifestKey(sessionPrefix)
+	manifestTempPath, err := writeCurrentManifestSnapshot(m, archiveMetadataPreview(
+		bucket,
+		runPrefix,
+		sessionPrefix,
+		runUploaded,
+		promotedUploaded,
+		skippedOptional,
+		currentManifestKey,
+	))
+	if err != nil {
+		return nil, fmt.Errorf("archive: 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("archive: upload current manifest to %q: %w", currentManifestKey, err)
+	}
+
+	currentRunPointerKey := artifacts.S3CurrentRunPointerKey(sessionPrefix)
+	runIDTempPath, err := writeCurrentRunIDPointer(runID)
+	if err != nil {
+		return nil, fmt.Errorf("archive: 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("archive: upload current run pointer to %q: %w", currentRunPointerKey, err)
 	}

 	return &StageResult{
 		Metadata: map[string]any{
-			"stage":                "archive",
-			"uploaded":             true,
-			"s3_bucket":            bucket,
-			"s3_run_prefix":        runPrefix,
-			"files_uploaded":       len(uploaded),
-			"uploaded_paths":       uploaded,
-			"audio_upload_skipped": true,
+			"stage":                       "archive",
+			"uploaded":                    true,
+			"s3_bucket":                   bucket,
+			"s3_run_prefix":               runPrefix,
+			"run_files_uploaded":          len(runUploaded),
+			"run_uploaded_paths":          runUploaded,
+			"promoted_files_uploaded":     len(promotedUploaded),
+			"promoted_paths":              promotedUploaded,
+			"skipped_optional_promotions": skippedOptional,
+			"current_manifest_key":        currentManifestKey,
+			"current_run_id_key":          currentRunPointerKey,
+			"current_pointer_written":     true,
+			"audio_upload_skipped":        true,
 		},
 	}, nil
 }

+type archivePromotion struct {
+	From      string
+	To        string
+	Required  bool
+	LocalPath string
+	Exists    bool
+}
+
 func archiveDisabled(env *Env) bool {
 	cfg := env.Config.Pipeline.Archive
 	if cfg == nil {
@@ -197,6 +285,22 @@ func archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) {
 		return runPrefix, nil
 	}

+	sessionPrefix, err := archiveSessionPrefix(env, m)
+	if err != nil {
+		return "", err
+	}
+	runID := strings.TrimSpace(m.RunID)
+	if runID == "" {
+		return "", fmt.Errorf("run id is required")
+	}
+	return artifacts.S3RunPrefix(sessionPrefix, runID), nil
+}
+
+func archiveSessionPrefix(env *Env, m *manifest.Manifest) (string, error) {
+	if m != nil && strings.TrimSpace(m.S3SessionPrefix) != "" {
+		return strings.TrimSpace(m.S3SessionPrefix), nil
+	}
+
 	sessionID := strings.TrimSpace(env.Config.Session.SessionID)
 	if sessionID == "" {
 		sessionID = strings.TrimSpace(m.SessionID)
@@ -205,10 +309,6 @@ func archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) {
 	if campaign == "" {
 		campaign = strings.TrimSpace(m.Campaign)
 	}
-	runID := strings.TrimSpace(m.RunID)
-	if runID == "" {
-		return "", fmt.Errorf("run id is required")
-	}
 	if env.Config.Pipeline.Storage.S3 == nil {
 		return "", fmt.Errorf("pipeline.storage.s3 configuration is required")
 	}
@@ -216,7 +316,7 @@ func archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) {
 	if strings.TrimSpace(sessionPrefix) == "" {
 		return "", fmt.Errorf("session prefix is required")
 	}
-	return artifacts.S3RunPrefix(sessionPrefix, runID), nil
+	return sessionPrefix, nil
 }

 func archiveBucket(env *Env, m *manifest.Manifest) string {
@@ -229,6 +329,52 @@ func archiveBucket(env *Env, m *manifest.Manifest) string {
 	return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket)
 }

+func resolveArchivePromotions(workDir string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) {
+	out := make([]archivePromotion, 0, len(rules))
+	for _, rule := range rules {
+		from := strings.TrimSpace(rule.From)
+		to := strings.TrimSpace(rule.To)
+		required := rule.Required == nil || *rule.Required
+
+		localPath, err := resolveWorkDirRelativePath(workDir, from)
+		if err != nil {
+			return nil, fmt.Errorf("promotion from %q: %w", from, err)
+		}
+		info, err := os.Stat(localPath)
+		exists := err == nil && !info.IsDir()
+		if err != nil && !os.IsNotExist(err) {
+			return nil, fmt.Errorf("promotion source %q: %w", from, err)
+		}
+
+		out = append(out, archivePromotion{
+			From:      from,
+			To:        to,
+			Required:  required,
+			LocalPath: localPath,
+			Exists:    exists,
+		})
+	}
+	return out, nil
+}
+
+func resolveWorkDirRelativePath(workDir, rel string) (string, error) {
+	rel = filepath.Clean(filepath.FromSlash(strings.TrimSpace(rel)))
+	if rel == "." || rel == "" {
+		return "", fmt.Errorf("relative path is required")
+	}
+	full := filepath.Join(workDir, rel)
+	cleanedWork := filepath.Clean(workDir)
+	cleanedFull := filepath.Clean(full)
+	relative, err := filepath.Rel(cleanedWork, cleanedFull)
+	if err != nil {
+		return "", fmt.Errorf("compute relative path: %w", err)
+	}
+	if relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
+		return "", fmt.Errorf("path escapes workdir")
+	}
+	return cleanedFull, nil
+}
+
 func collectArchiveRunFiles(workDir string) ([]string, error) {
 	files := make([]string, 0, 64)

@@ -280,3 +426,100 @@ func collectArchiveRunFiles(workDir string) ([]string, error) {
 	sort.Strings(files)
 	return files, nil
 }
+
+func writeCurrentManifestSnapshot(m *manifest.Manifest, archiveMetadata 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("archive", now, nil)
+	if sr := clone.Stages["archive"]; sr != nil {
+		sr.Metadata = archiveMetadata
+	}
+
+	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 archiveMetadataPreview(
+	bucket, runPrefix, sessionPrefix string,
+	runUploaded []string,
+	promotedUploaded []string,
+	skippedOptional []string,
+	currentManifestKey string,
+) map[string]any {
+	return map[string]any{
+		"stage":                       "archive",
+		"uploaded":                    true,
+		"s3_bucket":                   bucket,
+		"s3_run_prefix":               runPrefix,
+		"run_files_uploaded":          len(runUploaded),
+		"run_uploaded_paths":          append([]string(nil), runUploaded...),
+		"promoted_files_uploaded":     len(promotedUploaded),
+		"promoted_paths":              append([]string(nil), promotedUploaded...),
+		"skipped_optional_promotions": append([]string(nil), skippedOptional...),
+		"current_manifest_key":        currentManifestKey,
+		"current_run_id_key":          artifacts.S3CurrentRunPointerKey(sessionPrefix),
+		"current_pointer_written":     false,
+		"audio_upload_skipped":        true,
+	}
+}