diff --git a/README.md b/README.md index ff322b5..9f89f1c 100644 --- a/README.md +++ b/README.md @@ -80,8 +80,11 @@ Current boundaries: - storage backend tests use fake storage and do not require live S3 - archive uploads successful run records under `runs/{run_id}/` - archive does not upload local audio by default -- archive does not yet perform promotion writes -- no `current/manifest.json` or `current/run_id.txt` uploads yet +- archive uploads promoted outputs to session-level keys using `archive.promote_artifacts` +- archive uploads `current/manifest.json` +- archive uploads `current/run_id.txt` last as the effective commit marker +- required missing promotions fail archive +- optional missing promotions are skipped and recorded S3 input details and current boundaries are documented in [docs/s3-audio-input.md](docs/s3-audio-input.md). @@ -108,7 +111,9 @@ Current boundary: - `prepare` uses `List` + `Download` through the backend when `session.inputs.audio_s3` is configured - `archive` uses `Upload` through the backend for successful run-record uploads -- promotion uploads and current-pointer writes are still not implemented +- `archive` also uses `Upload` for promotion writes and current pointers +- no failed or incomplete runs are uploaded +- local audio is not re-uploaded by default Archive run-upload details and boundaries are documented in [docs/archive-storage.md](docs/archive-storage.md). diff --git a/architecture.md b/architecture.md index 7734a53..35d6319 100644 --- a/architecture.md +++ b/architecture.md @@ -40,7 +40,6 @@ Implemented: Still placeholder/future: -- archive-stage promotion behavior - `notify` stage behavior - additional Scriptorium artifact types beyond `session_recap` - artifact-to-artifact workflows beyond the initial single-artifact implementation @@ -168,7 +167,7 @@ Remote object-store backend scope: - backend methods do not prepend `root_prefix` or infer campaign/session/run paths - `prepare` now uses object-store `List` and `Download` for S3 audio input - `archive` now uses object-store `Upload` for successful run-record upload under the run prefix -- archive promotion behavior using top-level writes and current pointers remains future work +- `archive` now uses object-store `Upload` for promoted outputs and current pointers Prepare S3 audio behavior (implemented): @@ -183,7 +182,7 @@ Prepare S3 audio behavior (implemented): - record S3 provenance in manifest input records (bucket/key/metadata/local paths/checksum) - no AWS SDK types are used in stage code; storage implementation details stay in storage adapter packages -Archive run upload behavior (implemented): +Archive publishing behavior (implemented): - `archive` verifies prerequisite stage success before upload: - `prepare`, `transcribe`, `merge`, `polish`, `normalize`, `trim`, `analyze` @@ -196,11 +195,22 @@ Archive run upload behavior (implemented): - file upload order is deterministic (sorted relative paths) - `archive.enabled: false` and `archive.upload_run: false` skip upload cleanly - stage metadata records non-secret upload context: - - `s3_bucket`, `s3_run_prefix`, `files_uploaded`, `uploaded_paths`, `audio_upload_skipped` -- archive does not yet: - - write top-level promoted outputs - - write `current/manifest.json` - - write `current/run_id.txt` + - run upload details, promoted output details, current manifest key, current pointer key + - no secrets, transcript contents, prompt contents, or environment dumps +- promotion rules: + - `from` resolves from local workdir + - `to` resolves under session-level S3 root + - missing required source fails archive + - missing optional source is skipped and recorded +- default promoted outputs: + - `transcripts/trimmed.json` + - `artifacts/session_recap.md` +- current pointers: + - `current/manifest.json` uploaded after run upload and promotions + - `current/run_id.txt` uploaded last with `{run_id}\n` + - `current/run_id.txt` is the effective commit marker +- if promotion or current-manifest upload fails, archive returns failure and does not write `current/run_id.txt` +- failed/incomplete runs remain local and are not uploaded `pipeline.scriptorium` is optional. Existing pipelines without Scriptorium continue to work. diff --git a/docs/archive-storage.md b/docs/archive-storage.md index 7c26ed1..626803f 100644 --- a/docs/archive-storage.md +++ b/docs/archive-storage.md @@ -1,27 +1,38 @@ # Archive Storage -This document describes implemented archive-stage storage behavior. +This document describes implemented archive-stage publish behavior. + +## S3 Paths + +Session root: + +`{root_prefix}/campaigns/{campaign}/sessions/{session_id}/` + +Run prefix: + +`{root_prefix}/campaigns/{campaign}/sessions/{session_id}/runs/{run_id}/` ## Scope Implemented: - archive uploads successful run records to remote object storage through the storage backend abstraction. -- upload destination is the run prefix: - - `{root_prefix}/campaigns/{campaign}/sessions/{session_id}/runs/{run_id}/` +- archive uploads configured promoted outputs to session-level keys. +- archive uploads `current/manifest.json`. +- archive uploads `current/run_id.txt` last as the effective commit marker. - tests use fake storage and do not require live S3. -Not implemented: +Future work: -- promotion uploads to top-level `transcripts/` or `artifacts/` -- `current/manifest.json` write -- `current/run_id.txt` write -- audio re-upload from local workdir -- failed/incomplete run uploads +- spool audio cleanup/deletion behavior +- `notify` stage behavior +- stale detection +- optional future source-audio upload mode +- additional artifact generation beyond current implemented set ## Prerequisites -Archive verifies the following stages succeeded before upload: +Archive verifies these stages succeeded before upload: - `prepare` - `transcribe` @@ -32,8 +43,9 @@ Archive verifies the following stages succeeded before upload: - `analyze` If any prerequisite is missing or not succeeded, archive fails and does not upload. +Failed or incomplete runs remain local only. -## Uploaded Run Record +## Run Upload Archive uploads existing files from the run workdir when present: @@ -45,12 +57,38 @@ Archive uploads existing files from the run workdir when present: - `logs/` - `manifest.json` -Archive preserves relative file paths under `runs/{run_id}/`. +Relative paths are preserved under `runs/{run_id}/`. -Example: +## Promotion Rules -- local `transcripts/raw/speaker.json` -- remote `runs/{run_id}/transcripts/raw/speaker.json` +Archive applies `archive.promote_artifacts` in config order. + +Rule behavior: + +- `from`: local workdir-relative source path +- `to`: session-root-relative destination key +- `required: true`: missing source fails archive +- `required: false`: missing source is skipped and recorded + +Default promoted outputs: + +- `transcripts/trimmed.json` +- `artifacts/session_recap.md` + +## Current Pointers + +Archive writes: + +1. `current/manifest.json` (after run upload + promotions) +2. `current/run_id.txt` last + +`current/run_id.txt` contains exactly: + +- `{run_id}` plus trailing newline + +Writing `current/run_id.txt` last makes it the effective commit marker for published session state. + +If any required run upload, promotion upload, or current-manifest upload fails, archive returns failure and does not write `current/run_id.txt`. ## Audio Upload Policy @@ -64,10 +102,14 @@ Original audio is expected at the session-level audio prefix and is not duplicat ## Metadata -Archive stage metadata includes non-secret upload context: +Archive stage metadata includes non-secret upload context (for example): - `s3_bucket` - `s3_run_prefix` -- `files_uploaded` -- `uploaded_paths` +- run upload counts/paths +- promoted upload counts/paths +- skipped optional promotions +- `current_manifest_key` +- `current_run_id_key` +- `current_pointer_written` - `audio_upload_skipped` diff --git a/docs/roadmap/narratio-s3-archive-implementation-plan.md b/docs/roadmap/narratio-s3-archive-implementation-plan.md index dc7d307..922ebae 100644 --- a/docs/roadmap/narratio-s3-archive-implementation-plan.md +++ b/docs/roadmap/narratio-s3-archive-implementation-plan.md @@ -107,11 +107,22 @@ Implemented in repository: - local audio is not uploaded by default - upload uses storage backend abstraction and deterministic ordering - archive skips cleanly when `archive.enabled` or `archive.upload_run` is false +- archive promotion and current publish behavior: + - promotion rules upload configured outputs to session-level destinations + - required missing promotion sources fail archive + - optional missing promotion sources are skipped and recorded + - `current/manifest.json` is uploaded after run upload and promotions + - `current/run_id.txt` is uploaded last as the effective commit marker + - current pointer content is `{run_id}` plus trailing newline + - if promotion/current manifest upload fails, current pointer is not written Not implemented yet: -- promotion uploads -- writing `current/manifest.json` and `current/run_id.txt` to S3 +- spool audio cleanup / deletion behavior +- `notify` stage behavior +- generic stale detection based on input/config checksums +- optional future mode for uploading source audio from local workspace/spool +- additional artifact generation beyond current implemented set ## 3. S3 Layout diff --git a/internal/adapters/storage/fake.go b/internal/adapters/storage/fake.go index b25f58b..0039a17 100644 --- a/internal/adapters/storage/fake.go +++ b/internal/adapters/storage/fake.go @@ -28,6 +28,7 @@ type FakeBackend struct { Result ArchiveResult Objects map[string]FakeObject + Uploads []FakeUploadCall ListErr error DownloadErr error @@ -35,6 +36,13 @@ type FakeBackend struct { ExistsErr error } +// FakeUploadCall captures one upload invocation in call order. +type FakeUploadCall struct { + LocalPath string + Key string + Options UploadOptions +} + // Archive records request and returns configured response. func (f *FakeBackend) Archive(ctx context.Context, req ArchiveRequest) (ArchiveResult, error) { if err := ctx.Err(); err != nil { @@ -153,6 +161,14 @@ func (f *FakeBackend) Upload(ctx context.Context, localPath, key string, opts Up } normalizedKey := normalizeObjectKey(key) + f.Uploads = append(f.Uploads, FakeUploadCall{ + LocalPath: localPath, + Key: normalizedKey, + Options: UploadOptions{ + Metadata: copyMetadata(opts.Metadata), + ContentType: opts.ContentType, + }, + }) now := time.Now().UTC() obj := FakeObject{ Key: normalizedKey, diff --git a/internal/stage/archive.go b/internal/stage/archive.go index 86e4136..ffbedcb 100644 --- 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, + } +} diff --git a/internal/stage/archive_test.go b/internal/stage/archive_test.go index 17bb62c..8fd54c2 100644 --- a/internal/stage/archive_test.go +++ b/internal/stage/archive_test.go @@ -3,7 +3,6 @@ package stage import ( "context" "errors" - "os" "path/filepath" "reflect" "strings" @@ -11,6 +10,7 @@ import ( "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" ) @@ -26,7 +26,7 @@ func TestArchiveSkipsWhenDisabled(t *testing.T) { if result.Metadata["skipped"] != true { t.Fatalf("metadata = %#v, want skipped=true", result.Metadata) } - if len(env.ObjectStore.(*storage.FakeBackend).Objects) != 0 { + if len(env.ObjectStore.(*storage.FakeBackend).Uploads) != 0 { t.Fatalf("unexpected uploads when archive disabled") } } @@ -42,7 +42,7 @@ func TestArchiveSkipsRunUploadWhenDisabled(t *testing.T) { if result.Metadata["skipped"] != true { t.Fatalf("metadata = %#v, want skipped=true", result.Metadata) } - if len(env.ObjectStore.(*storage.FakeBackend).Objects) != 0 { + if len(env.ObjectStore.(*storage.FakeBackend).Uploads) != 0 { t.Fatalf("unexpected uploads when upload_run disabled") } } @@ -55,10 +55,13 @@ func TestArchiveFailsWhenPrerequisiteNotSucceeded(t *testing.T) { if err == nil || !strings.Contains(err.Error(), `prerequisite stage "trim"`) { t.Fatalf("Run() error = %v, want prerequisite failure", err) } + if len(env.ObjectStore.(*storage.FakeBackend).Uploads) != 0 { + t.Fatalf("unexpected uploads on prerequisite failure") + } } -func TestArchiveUploadsRunRecordWithoutAudio(t *testing.T) { - env, m, workDir := archiveFixture(t) +func TestArchiveUploadsRunRecordPromotionsAndCurrentPointer(t *testing.T) { + env, m, _ := archiveFixture(t) fake := env.ObjectStore.(*storage.FakeBackend) result, err := archiveStage{}.Run(context.Background(), env, m) @@ -66,7 +69,9 @@ func TestArchiveUploadsRunRecordWithoutAudio(t *testing.T) { t.Fatalf("Run() error = %v", err) } - wantUploaded := []string{ + runPrefix := m.S3RunPrefix + sessionPrefix := m.S3SessionPrefix + wantRunUploads := []string{ "artifacts/session_recap.md", "config/audita.generated.yml", "inputs/session.yml", @@ -76,37 +81,145 @@ func TestArchiveUploadsRunRecordWithoutAudio(t *testing.T) { "transcripts/raw/speaker.json", "transcripts/trimmed.json", } - if got := result.Metadata["uploaded_paths"]; !reflect.DeepEqual(got, wantUploaded) { - t.Fatalf("uploaded_paths = %#v, want %#v", got, wantUploaded) - } - if got := result.Metadata["files_uploaded"]; got != len(wantUploaded) { - t.Fatalf("files_uploaded = %#v, want %d", got, len(wantUploaded)) - } - - for _, rel := range wantUploaded { - key := m.S3RunPrefix + rel + for _, rel := range wantRunUploads { + key := runPrefix + rel if _, ok := fake.Objects[key]; !ok { - t.Fatalf("missing uploaded key %q", key) + t.Fatalf("missing run upload key %q", key) } } - audioKey := m.S3RunPrefix + "audio/speaker.flac" + + trimmedKey := sessionPrefix + "transcripts/trimmed.json" + recapKey := sessionPrefix + "artifacts/session_recap.md" + if _, ok := fake.Objects[trimmedKey]; !ok { + t.Fatalf("missing promoted key %q", trimmedKey) + } + if _, ok := fake.Objects[recapKey]; !ok { + t.Fatalf("missing promoted key %q", recapKey) + } + + currentManifestKey := sessionPrefix + "current/manifest.json" + currentRunIDKey := sessionPrefix + "current/run_id.txt" + if _, ok := fake.Objects[currentManifestKey]; !ok { + t.Fatalf("missing current manifest key %q", currentManifestKey) + } + if _, ok := fake.Objects[currentRunIDKey]; !ok { + t.Fatalf("missing current run pointer key %q", currentRunIDKey) + } + if got := string(fake.Objects[currentRunIDKey].Data); got != m.RunID+"\n" { + t.Fatalf("run pointer contents = %q, want %q", got, m.RunID+"\\n") + } + + audioKey := runPrefix + "audio/speaker.flac" if _, ok := fake.Objects[audioKey]; ok { t.Fatalf("audio key %q should not be uploaded", audioKey) } - // Ensure relative paths are preserved under runs/{run_id}/. - if _, err := os.Stat(filepath.Join(workDir, "transcripts", "raw", "speaker.json")); err != nil { - t.Fatalf("expected local fixture transcript file: %v", err) + uploads := fake.Uploads + if len(uploads) == 0 { + t.Fatal("expected uploads") + } + if uploads[len(uploads)-1].Key != currentRunIDKey { + t.Fatalf("last upload key = %q, want current run pointer key %q", uploads[len(uploads)-1].Key, currentRunIDKey) + } + + if result.Metadata["current_pointer_written"] != true { + t.Fatalf("metadata = %#v, want current_pointer_written=true", result.Metadata) + } + if result.Metadata["promoted_files_uploaded"] != 2 { + t.Fatalf("metadata promoted_files_uploaded = %#v, want 2", result.Metadata["promoted_files_uploaded"]) } } -func TestArchiveFailsWhenUploadFails(t *testing.T) { - env, m, _ := archiveFixture(t) - env.ObjectStore.(*storage.FakeBackend).UploadErr = errors.New("upload failed") +func TestArchiveUsesCustomPromotionRules(t *testing.T) { + env, m, workDir := archiveFixture(t) + env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ + {From: "transcripts/trimmed.json", To: "published/trimmed.json", Required: boolPtr(true)}, + {From: "artifacts/session_recap.md", To: "published/recap.md", Required: boolPtr(true)}, + } + writeStageTestFile(t, filepath.Join(workDir, "published", "ignored.txt"), "ignore\n") _, err := archiveStage{}.Run(context.Background(), env, m) - if err == nil || !strings.Contains(err.Error(), "upload") { - t.Fatalf("Run() error = %v, want upload failure", err) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + + fake := env.ObjectStore.(*storage.FakeBackend) + if _, ok := fake.Objects[m.S3SessionPrefix+"published/trimmed.json"]; !ok { + t.Fatalf("missing custom promoted trimmed key") + } + if _, ok := fake.Objects[m.S3SessionPrefix+"published/recap.md"]; !ok { + t.Fatalf("missing custom promoted recap key") + } +} + +func TestArchiveSkipsOptionalMissingPromotion(t *testing.T) { + env, m, _ := archiveFixture(t) + env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ + {From: "transcripts/trimmed.json", To: "transcripts/trimmed.json", Required: boolPtr(true)}, + {From: "artifacts/optional.md", To: "artifacts/optional.md", Required: boolPtr(false)}, + } + + result, err := archiveStage{}.Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + got, _ := result.Metadata["skipped_optional_promotions"].([]string) + want := []string{"artifacts/optional.md"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("skipped_optional_promotions = %#v, want %#v", got, want) + } +} + +func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) { + env, m, _ := archiveFixture(t) + env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ + {From: "artifacts/missing.md", To: "artifacts/missing.md", Required: boolPtr(true)}, + } + + _, err := archiveStage{}.Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "required promotion source missing") { + t.Fatalf("Run() error = %v, want required promotion missing failure", err) + } +} + +func TestArchiveDoesNotWriteCurrentPointerWhenPromotionUploadFails(t *testing.T) { + env, m, _ := archiveFixture(t) + fake := env.ObjectStore.(*storage.FakeBackend) + trimmedKey := m.S3SessionPrefix + "transcripts/trimmed.json" + + origUploadErr := fake.UploadErr + fake.UploadErr = nil + failingKey := trimmedKey + fake.Uploads = nil + + originalUpload := fake.Upload + _ = originalUpload + // Use UploadErr toggle by checking call sequence in postcondition. + // First failure point is promotion upload; simulate by setting error immediately before promotion key write. + // We cannot hook FakeBackend per-key without changing public behavior; use dedicated backend wrapper instead. + env.ObjectStore = &promotionFailingStore{delegate: fake, failKey: failingKey} + + _, err := archiveStage{}.Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "promoted output") { + t.Fatalf("Run() error = %v, want promotion upload failure", err) + } + if _, ok := fake.Objects[m.S3SessionPrefix+"current/run_id.txt"]; ok { + t.Fatalf("unexpected current pointer write on promotion failure") + } + fake.UploadErr = origUploadErr +} + +func TestArchiveDoesNotWriteCurrentPointerWhenCurrentManifestUploadFails(t *testing.T) { + env, m, _ := archiveFixture(t) + fake := env.ObjectStore.(*storage.FakeBackend) + env.ObjectStore = &promotionFailingStore{delegate: fake, failKey: m.S3SessionPrefix + "current/manifest.json"} + + _, err := archiveStage{}.Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "current manifest") { + t.Fatalf("Run() error = %v, want current manifest upload failure", err) + } + if _, ok := fake.Objects[m.S3SessionPrefix+"current/run_id.txt"]; ok { + t.Fatalf("unexpected current pointer write when current manifest upload fails") } } @@ -120,23 +233,6 @@ func TestArchiveFailsWithoutObjectStore(t *testing.T) { } } -func TestArchiveDoesNotWriteCurrentPointers(t *testing.T) { - env, m, _ := archiveFixture(t) - fake := env.ObjectStore.(*storage.FakeBackend) - - _, err := archiveStage{}.Run(context.Background(), env, m) - if err != nil { - t.Fatalf("Run() error = %v", err) - } - - if _, ok := fake.Objects[m.S3SessionPrefix+"current/run_id.txt"]; ok { - t.Fatalf("unexpected current/run_id.txt upload") - } - if _, ok := fake.Objects[m.S3SessionPrefix+"current/manifest.json"]; ok { - t.Fatalf("unexpected current/manifest.json upload") - } -} - func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) { t.Helper() @@ -156,15 +252,15 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) { writeStageTestFile(t, filepath.Join(workDir, "audio", "speaker.flac"), "flac") writeStageTestFile(t, filepath.Join(workDir, "manifest.json"), "{}\n") - m := manifest.New(sessionID, time.Now().UTC()) + m := manifest.New(sessionID, time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC)) m.Campaign = campaign m.RunID = runID m.LocalWorkDir = workDir m.S3Bucket = "my-dnd-archive" - m.S3SessionPrefix = "dnd/campaigns/forsaken/sessions/2026-04-19/" - m.S3RunPrefix = m.S3SessionPrefix + "runs/" + runID + "/" + m.S3SessionPrefix = artifacts.S3SessionPrefix("dnd", campaign, sessionID) + m.S3RunPrefix = artifacts.S3RunPrefix(m.S3SessionPrefix, runID) for _, name := range archivePrerequisiteStages { - m.MarkStageSucceeded(name, time.Now().UTC(), nil) + m.MarkStageSucceeded(name, time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC), nil) } env := &Env{ @@ -180,6 +276,10 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) { Archive: &config.ArchiveConfig{ Enabled: boolPtr(true), UploadRun: boolPtr(true), + PromoteArtifacts: []config.ArchivePromotionRule{ + {From: "transcripts/trimmed.json", To: "transcripts/trimmed.json", Required: boolPtr(true)}, + {From: "artifacts/session_recap.md", To: "artifacts/session_recap.md", Required: boolPtr(true)}, + }, }, }, Session: &config.SessionConfig{ @@ -192,6 +292,30 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) { return env, m, workDir } +type promotionFailingStore struct { + delegate *storage.FakeBackend + failKey string +} + +func (s *promotionFailingStore) List(ctx context.Context, prefix string) ([]storage.ObjectInfo, error) { + return s.delegate.List(ctx, prefix) +} + +func (s *promotionFailingStore) Download(ctx context.Context, key, localPath string) error { + return s.delegate.Download(ctx, key, localPath) +} + +func (s *promotionFailingStore) Upload(ctx context.Context, localPath, key string, opts storage.UploadOptions) (storage.ObjectInfo, error) { + if strings.TrimSpace(key) == strings.TrimSpace(s.failKey) { + return storage.ObjectInfo{}, errors.New("forced upload failure") + } + return s.delegate.Upload(ctx, localPath, key, opts) +} + +func (s *promotionFailingStore) Exists(ctx context.Context, key string) (bool, error) { + return s.delegate.Exists(ctx, key) +} + func boolPtr(v bool) *bool { p := v return &p