686 lines
21 KiB
Go
686 lines
21 KiB
Go
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/artifacts"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
|
)
|
|
|
|
type archiveStage struct{}
|
|
type archiveUploadFile struct {
|
|
RelativePath string
|
|
LocalPath string
|
|
}
|
|
|
|
var archivePrerequisiteStages = []string{
|
|
"prepare",
|
|
"transcribe",
|
|
"merge",
|
|
"polish",
|
|
"normalize",
|
|
"trim",
|
|
"analyze",
|
|
}
|
|
|
|
func (archiveStage) Name() string { return "archive" }
|
|
|
|
func (archiveStage) Declares() IODecl {
|
|
return IODecl{
|
|
Inputs: []artifacts.Ref{
|
|
{Kind: "manifest", Category: "input", RelativePath: "manifest.json"},
|
|
},
|
|
}
|
|
}
|
|
|
|
func (archiveStage) 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("archive: resolved config must include pipeline and session")
|
|
}
|
|
|
|
if archiveDisabled(env) {
|
|
return &StageResult{
|
|
Metadata: map[string]any{
|
|
"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,
|
|
"current_pointer_written": false,
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
if err := validateArchivePrerequisites(m); err != nil {
|
|
return nil, fmt.Errorf("archive: %w", err)
|
|
}
|
|
if env.ObjectStore == nil {
|
|
return nil, fmt.Errorf("archive: remote object store backend is required when archive run upload is enabled")
|
|
}
|
|
|
|
runRoot, err := resolveArchiveRunRoot(env, m)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve run root: %w", err)
|
|
}
|
|
runRootInfo, err := os.Stat(runRoot)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: run root %q: %w", runRoot, err)
|
|
}
|
|
if !runRootInfo.IsDir() {
|
|
return nil, fmt.Errorf("archive: run root %q is not a directory", runRoot)
|
|
}
|
|
|
|
runPrefix, err := artifacts.ResolveArchiveRunPrefix(env.Config, m)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve s3 run prefix: %w", err)
|
|
}
|
|
sessionPrefix, err := artifacts.ResolveArchiveSessionPrefix(env.Config, m)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve s3 session prefix: %w", err)
|
|
}
|
|
bucket := artifacts.ResolveArchiveBucket(env.Config, 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")
|
|
}
|
|
|
|
manifestSource, err := resolveArchiveRunManifestSource(runRoot)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve run manifest source: %w", err)
|
|
}
|
|
|
|
runFiles, err := collectArchiveRunFiles(runRoot, manifestSource)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: collect run files: %w", err)
|
|
}
|
|
sessionPaths := archiveSessionPaths(env, m)
|
|
previousFiles, err := collectArchivePreviousFiles(sessionPaths.PreviousDir)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: collect previous files: %w", err)
|
|
}
|
|
runtimeCatalog, err := buildArchiveRuntimeArtifactCatalog(sessionPaths, env.Config.Pipeline.Scriptorium)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: build runtime artifact catalog: %w", err)
|
|
}
|
|
promotions, skippedOptional, err := resolveArchivePromotions(sessionPaths, m, runtimeCatalog, env.Config.Pipeline.Archive.PromoteArtifacts)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve promotion 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("archive: upload run file %q to %q: %w", file.RelativePath, key, err)
|
|
}
|
|
runUploaded = append(runUploaded, file.RelativePath)
|
|
}
|
|
|
|
promotedUploaded := make([]string, 0, len(promotions))
|
|
for _, promotion := range promotions {
|
|
key := artifacts.S3PromotedArtifactKey(sessionPrefix, promotion.Dest)
|
|
if _, err := env.ObjectStore.Upload(ctx, promotion.LocalPath, key, storage.UploadOptions{}); err != nil {
|
|
return nil, fmt.Errorf("archive: upload promoted output source %q to %q: %w", promotion.Source, key, err)
|
|
}
|
|
promotedUploaded = append(promotedUploaded, promotion.Dest)
|
|
}
|
|
|
|
previousUploaded := make([]string, 0, len(previousFiles))
|
|
for _, file := range previousFiles {
|
|
key := artifacts.S3PromotedArtifactKey(sessionPrefix, file.RelativePath)
|
|
if _, err := env.ObjectStore.Upload(ctx, file.LocalPath, key, storage.UploadOptions{}); err != nil {
|
|
return nil, fmt.Errorf("archive: upload previous file %q to %q: %w", file.RelativePath, key, err)
|
|
}
|
|
previousUploaded = append(previousUploaded, file.RelativePath)
|
|
}
|
|
|
|
currentManifestKey, currentRunPointerKey := artifacts.ResolveArchiveCurrentStateKeys(sessionPrefix)
|
|
manifestTempPath, err := writeCurrentManifestSnapshot(m, archiveMetadataPreview(
|
|
bucket,
|
|
runPrefix,
|
|
sessionPrefix,
|
|
runUploaded,
|
|
promotedUploaded,
|
|
previousUploaded,
|
|
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)
|
|
}
|
|
|
|
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,
|
|
"run_files_uploaded": len(runUploaded),
|
|
"run_uploaded_paths": runUploaded,
|
|
"promoted_files_uploaded": len(promotedUploaded),
|
|
"promoted_paths": promotedUploaded,
|
|
"previous_files_uploaded": len(previousUploaded),
|
|
"previous_uploaded_paths": previousUploaded,
|
|
"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 {
|
|
Source string
|
|
Dest string
|
|
Required bool
|
|
LocalPath string
|
|
Provenance string
|
|
}
|
|
|
|
func archiveDisabled(env *Env) bool {
|
|
cfg := env.Config.Pipeline.Archive
|
|
if cfg == nil {
|
|
return true
|
|
}
|
|
return cfg.Enabled != nil && !*cfg.Enabled
|
|
}
|
|
|
|
func archiveRunUploadDisabled(env *Env) bool {
|
|
cfg := env.Config.Pipeline.Archive
|
|
if cfg == nil {
|
|
return true
|
|
}
|
|
return cfg.UploadRun != nil && !*cfg.UploadRun
|
|
}
|
|
|
|
func validateArchivePrerequisites(m *manifest.Manifest) error {
|
|
if m == nil {
|
|
return fmt.Errorf("manifest is required")
|
|
}
|
|
for _, stageName := range archivePrerequisiteStages {
|
|
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 resolveArchiveRunRoot(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 resolveArchiveSessionRoot(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 archiveSessionPaths(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 resolveArchivePromotions(
|
|
paths artifacts.SessionPaths,
|
|
m *manifest.Manifest,
|
|
catalog *artifacts.ArtifactCatalog,
|
|
rules []config.ArchivePromotionRule,
|
|
) ([]archivePromotion, []string, error) {
|
|
out := make([]archivePromotion, 0, len(rules))
|
|
skippedOptional := make([]string, 0)
|
|
for _, rule := range rules {
|
|
source := strings.TrimSpace(rule.Source)
|
|
required := rule.Required == nil || *rule.Required
|
|
dest, err := resolveArchivePromotionDest(rule, catalog)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("source %q: %w", source, err)
|
|
}
|
|
resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, catalog)
|
|
if err != nil {
|
|
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) && !required {
|
|
skippedOptional = append(skippedOptional, dest)
|
|
continue
|
|
}
|
|
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) {
|
|
return nil, nil, fmt.Errorf("required promotion source unavailable: %q", source)
|
|
}
|
|
return nil, nil, fmt.Errorf("resolve source %q: %w", source, err)
|
|
}
|
|
out = append(out, archivePromotion{
|
|
Source: source,
|
|
Dest: dest,
|
|
Required: required,
|
|
LocalPath: resolved.Path,
|
|
Provenance: resolved.Provenance,
|
|
})
|
|
}
|
|
return out, skippedOptional, nil
|
|
}
|
|
|
|
func resolveArchivePromotionDest(rule config.ArchivePromotionRule, catalog *artifacts.ArtifactCatalog) (string, error) {
|
|
dest := strings.TrimSpace(rule.Dest)
|
|
if dest == "" {
|
|
entry, ok := catalog.Lookup(strings.TrimSpace(rule.Source))
|
|
if !ok {
|
|
return "", fmt.Errorf("destination omitted and source is unknown")
|
|
}
|
|
dest = strings.TrimSpace(entry.CanonicalRelPath)
|
|
if dest == "" {
|
|
return "", fmt.Errorf("destination omitted and no canonical destination is available")
|
|
}
|
|
}
|
|
return normalizeArchiveRelativePath(dest)
|
|
}
|
|
|
|
func normalizeArchiveRelativePath(rel string) (string, error) {
|
|
trimmed := strings.TrimSpace(rel)
|
|
if trimmed == "" {
|
|
return "", fmt.Errorf("relative path is required")
|
|
}
|
|
cleaned := filepath.ToSlash(filepath.Clean(filepath.FromSlash(trimmed)))
|
|
if cleaned == "." || cleaned == "" {
|
|
return "", fmt.Errorf("relative path is required")
|
|
}
|
|
if filepath.IsAbs(trimmed) || strings.HasPrefix(cleaned, "/") || cleaned == ".." || strings.HasPrefix(cleaned, "../") {
|
|
return "", fmt.Errorf("path must be a clean relative path")
|
|
}
|
|
return cleaned, nil
|
|
}
|
|
|
|
func buildArchiveRuntimeArtifactCatalog(
|
|
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 collectArchiveRunFiles(runRoot, manifestPath string) ([]archiveUploadFile, error) {
|
|
files := make([]archiveUploadFile, 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 archive 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, archiveUploadFile{
|
|
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, archiveUploadFile{
|
|
RelativePath: "manifest.json",
|
|
LocalPath: manifestPath,
|
|
})
|
|
seen := map[string]archiveUploadFile{}
|
|
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 collectArchivePreviousFiles(previousDir string) ([]archiveUploadFile, 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([]archiveUploadFile, 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, archiveUploadFile{
|
|
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 resolveArchiveRunManifestSource(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, 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,
|
|
previousUploaded []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...),
|
|
"previous_files_uploaded": len(previousUploaded),
|
|
"previous_uploaded_paths": append([]string(nil), previousUploaded...),
|
|
"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,
|
|
}
|
|
}
|