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, lockedPromotions, err := resolveArchivePromotions( sessionPaths, m, runtimeCatalog, env.Config.Pipeline.Archive.PromoteArtifacts, env.Config.Pipeline.Archive.Locks, sessionPrefix, ) 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, lockedPromotions, 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, "locked_promotion_count": len(lockedPromotions), "locked_promotions": lockedPromotionMetadata(lockedPromotions), "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 } type archiveLockedPromotion struct { Source string Dest string RemoteKey string Reason 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, locks []config.ArchiveLockRule, sessionPrefix string, ) ([]archivePromotion, []string, []archiveLockedPromotion, error) { out := make([]archivePromotion, 0, len(rules)) skippedOptional := make([]string, 0) lockedPromotions := make([]archiveLockedPromotion, 0) lockSet := archiveLockSet(locks) 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, nil, fmt.Errorf("source %q: %w", source, err) } lock, locked := lockSet[source] resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, catalog) if err != nil { if locked { lockedPromotions = append(lockedPromotions, archiveLockedPromotion{ Source: source, Dest: dest, RemoteKey: artifacts.S3PromotedArtifactKey(sessionPrefix, dest), Reason: strings.TrimSpace(lock.Reason), Required: required, }) continue } if errors.Is(err, artifacts.ErrSessionArtifactNotFound) && !required { skippedOptional = append(skippedOptional, dest) continue } if errors.Is(err, artifacts.ErrSessionArtifactNotFound) { return nil, nil, nil, fmt.Errorf("required promotion source unavailable: %q", source) } return nil, nil, nil, fmt.Errorf("resolve source %q: %w", source, err) } if locked { lockedPromotions = append(lockedPromotions, archiveLockedPromotion{ Source: source, Dest: dest, RemoteKey: artifacts.S3PromotedArtifactKey(sessionPrefix, dest), Reason: strings.TrimSpace(lock.Reason), Required: required, LocalPath: resolved.Path, Provenance: resolved.Provenance, }) continue } out = append(out, archivePromotion{ Source: source, Dest: dest, Required: required, LocalPath: resolved.Path, Provenance: resolved.Provenance, }) } return out, skippedOptional, lockedPromotions, nil } func archiveLockSet(locks []config.ArchiveLockRule) map[string]config.ArchiveLockRule { out := make(map[string]config.ArchiveLockRule, 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 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, lockedPromotions []archiveLockedPromotion, 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...), "locked_promotion_count": len(lockedPromotions), "locked_promotions": lockedPromotionMetadata(lockedPromotions), "current_manifest_key": currentManifestKey, "current_run_id_key": artifacts.S3CurrentRunPointerKey(sessionPrefix), "current_pointer_written": false, "audio_upload_skipped": true, } } func lockedPromotionMetadata(locked []archiveLockedPromotion) []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 }