From d001baa660dc03380a967a602d11b5703480b329 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 19 May 2026 20:05:24 -0500 Subject: [PATCH] Use artifact source IDs for archive promotion --- docs/config.md | 35 +++- docs/internal/stage-archive.md | 2 +- docs/operations.md | 4 +- docs/troubleshooting.md | 2 +- examples/pipeline.full.annotated.yml | 14 +- examples/pipeline.production.yml | 12 +- internal/app/post_archive_cleanup_test.go | 19 ++- internal/app/runner_test.go | 13 +- internal/config/config.go | 4 +- internal/config/defaults.go | 3 +- internal/config/storage_archive_test.go | 138 +++++++++++++-- internal/config/validate.go | 81 ++++++++- internal/stage/archive.go | 196 ++++++++++++++++------ internal/stage/archive_test.go | 33 ++-- 14 files changed, 432 insertions(+), 124 deletions(-) diff --git a/docs/config.md b/docs/config.md index 197141b..c3095b6 100644 --- a/docs/config.md +++ b/docs/config.md @@ -106,11 +106,11 @@ archive: enabled: true upload_run: true promote_artifacts: - - from: transcripts/trimmed.json - to: transcripts/trimmed.json + - source: narratio.transcript.trimmed + dest: transcripts/trimmed.json required: true - - from: artifacts/session_recap.md - to: artifacts/session_recap.md + - source: narratio.artifact.session_recap + dest: artifacts/session_recap.md required: true whisperx: @@ -130,7 +130,8 @@ scriptorium: Operational notes: -- archive promotion is explicit and path-based via `archive.promote_artifacts`. +- archive promotion is explicit and source-based via `archive.promote_artifacts`. +- `source` is required; `dest` is optional and derived when omitted. - Narratio does not auto-promote all generated analyze artifacts. ## 7. Full pipeline reference @@ -154,9 +155,9 @@ Operational notes: | `pipeline.spool.delete_audio_after_archive` | bool | No | `false` | | `pipeline.archive.enabled` | bool | No | `true` | | `pipeline.archive.upload_run` | bool | No | `true` | -| `pipeline.archive.promote_artifacts[]` | list | No | trimmed + session_recap rules | -| `pipeline.archive.promote_artifacts[].from` | string | Yes (per rule) | none | -| `pipeline.archive.promote_artifacts[].to` | string | Yes (per rule) | none | +| `pipeline.archive.promote_artifacts[]` | list | No | trimmed transcript rule | +| `pipeline.archive.promote_artifacts[].source` | string | Yes (per rule) | none | +| `pipeline.archive.promote_artifacts[].dest` | string | No | derived from source | | `pipeline.archive.promote_artifacts[].required` | bool | No | `true` | | `pipeline.whisperx.transcribe_url` | string | Yes | none | | `pipeline.whisperx.language` | string | No | `en` | @@ -248,6 +249,24 @@ Allowed `pipeline.scriptorium.artifacts..inputs..source` values: - `narratio.bounds.session` - `narratio.artifact.` +`pipeline.archive.promote_artifacts[].source` values: + +- `narratio.transcript.merged` +- `narratio.transcript.polished` +- `narratio.transcript.full` +- `narratio.transcript.trimmed` +- `narratio.bounds.session` +- `narratio.artifact.` + +Archive promotion destination rules: + +- `dest` must be a clean relative path (not absolute, no traversal). +- duplicate `dest` values are rejected. +- if `dest` is omitted: + - built-in sources derive their canonical destination path; + - configured sources derive from `pipeline.scriptorium.artifacts..output_path`; + - derivation failure is a config validation error. + ## 8. Full session reference | Path | Type | Required | Default | diff --git a/docs/internal/stage-archive.md b/docs/internal/stage-archive.md index e46853d..c19471f 100644 --- a/docs/internal/stage-archive.md +++ b/docs/internal/stage-archive.md @@ -7,7 +7,7 @@ Publish run records and promoted session artifacts to object storage, then atomi Inputs: - session manifest and prerequisite stage records - run root contents under `runs/{run_id}/` -- promotion sources from session root (`archive.promote_artifacts`) +- promotion rules with artifact `source` IDs and archive `dest` paths (`archive.promote_artifacts`) Outputs: - uploaded run files under `{session_prefix}/runs/{run_id}/...` diff --git a/docs/operations.md b/docs/operations.md index e0c75ca..84ce816 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -84,10 +84,12 @@ Publish order: `current/run_id.txt` is the remote commit marker. -Archive promotion is explicit and path-based: +Archive promotion is explicit and source-based: - Narratio does not auto-promote all generated analyze artifacts. +- each rule resolves `source` through the artifact resolver/catalog model, then uploads to `dest`. - missing required promotion sources fail archive stage. - missing optional promotion sources are skipped. +- invalid resolved artifacts fail archive stage. ## Resume, retry, and safe rerun behavior diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 1c54076..7e69715 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -280,7 +280,7 @@ narratio run-stage --config /path/to/pipeline.yml --session /path/to/session.yml Safe Fix: - rerun or resume upstream stages to generate required files. -- adjust promotion rules to match files that must exist. +- adjust promotion `source`/`dest` rules to match artifacts that must exist. - retry after storage issue is resolved. Links: diff --git a/examples/pipeline.full.annotated.yml b/examples/pipeline.full.annotated.yml index e19d1bf..5d28291 100644 --- a/examples/pipeline.full.annotated.yml +++ b/examples/pipeline.full.annotated.yml @@ -40,16 +40,16 @@ archive: # Optional booleans; defaults are true. enabled: true upload_run: true - # Optional promotion rules; required files fail archive if missing. + # Optional promotion rules; sources use Narratio artifact source IDs. promote_artifacts: - - from: transcripts/trimmed.json - to: transcripts/trimmed.json + - source: narratio.transcript.trimmed + dest: transcripts/trimmed.json required: true - - from: artifacts/session_recap.md - to: artifacts/session_recap.md + - source: narratio.artifact.session_recap + dest: artifacts/session_recap.md required: true - - from: artifacts/player_handout.md - to: artifacts/player_handout.md + - source: narratio.artifact.player_handout + dest: artifacts/player_handout.md required: false whisperx: diff --git a/examples/pipeline.production.yml b/examples/pipeline.production.yml index f48b82a..bc928e5 100644 --- a/examples/pipeline.production.yml +++ b/examples/pipeline.production.yml @@ -19,14 +19,14 @@ archive: enabled: true upload_run: true promote_artifacts: - - from: transcripts/trimmed.json - to: transcripts/trimmed.json + - source: narratio.transcript.trimmed + dest: transcripts/trimmed.json required: true - - from: artifacts/session_recap.md - to: artifacts/session_recap.md + - source: narratio.artifact.session_recap + dest: artifacts/session_recap.md required: true - - from: artifacts/player_handout.md - to: artifacts/player_handout.md + - source: narratio.artifact.player_handout + dest: artifacts/player_handout.md required: false whisperx: diff --git a/internal/app/post_archive_cleanup_test.go b/internal/app/post_archive_cleanup_test.go index ca9c246..1f363e4 100644 --- a/internal/app/post_archive_cleanup_test.go +++ b/internal/app/post_archive_cleanup_test.go @@ -195,7 +195,7 @@ func TestPostArchiveCleanupNotRunWhenPromotionIsMissing(t *testing.T) { cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ - {From: "artifacts/missing.md", To: "artifacts/missing.md", Required: boolPtr(true)}, + {Source: "narratio.transcript.merged", Dest: "transcripts/merged.json", Required: boolPtr(true)}, } archiveStageImpl, err := stage.Select("archive") @@ -203,7 +203,7 @@ func TestPostArchiveCleanupNotRunWhenPromotionIsMissing(t *testing.T) { t.Fatalf("Select(archive) error = %v", err) } _, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) - if err == nil || !strings.Contains(err.Error(), "required promotion source missing") { + if err == nil || !strings.Contains(err.Error(), "required promotion source unavailable") { t.Fatalf("executeStages() error = %v, want promotion-missing failure", err) } @@ -322,8 +322,15 @@ func archiveStageCleanupFixture(t *testing.T) (*config.Config, cleanupSeed, stri 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)}, + {Source: "narratio.transcript.trimmed", Dest: "transcripts/trimmed.json", Required: boolPtr(true)}, + {Source: "narratio.artifact.session_recap", Dest: "artifacts/session_recap.md", Required: boolPtr(true)}, + }, + } + cfg.Pipeline.Scriptorium = &config.ScriptoriumConfig{ + Artifacts: map[string]config.ScriptoriumArtifactConfig{ + "session_recap": { + OutputPath: "artifacts/session_recap.md", + }, }, } writeArchiveFixtureRunFiles( @@ -353,14 +360,14 @@ func writeArchiveFixtureRunFiles(t *testing.T, runWorkDir, sessionRoot string) { t.Helper() mustWriteFile(t, filepath.Join(runWorkDir, "prepare", "inputs", "session.yml"), "session_id: 2026-05-03\n") mustWriteFile(t, filepath.Join(runWorkDir, "transcribe", "outputs", "transcripts", "raw", "speaker.json"), "{}\n") - mustWriteFile(t, filepath.Join(runWorkDir, "trim", "outputs", "transcripts", "trimmed.json"), "{}\n") + mustWriteFile(t, filepath.Join(runWorkDir, "trim", "outputs", "transcripts", "trimmed.json"), "{\"segments\":[]}\n") mustWriteFile(t, filepath.Join(runWorkDir, "analyze", "outputs", "artifacts", "session_recap.md"), "# recap\n") mustWriteFile(t, filepath.Join(runWorkDir, "polish", "reports", "audita.report.json"), "{}\n") mustWriteFile(t, filepath.Join(runWorkDir, "merge", "config", "seriatim.generated.yml"), "key: value\n") mustWriteFile(t, filepath.Join(runWorkDir, "logs", "audita.stderr.log"), "stderr\n") mustWriteFile(t, filepath.Join(runWorkDir, "manifest.json"), "{}\n") - mustWriteFile(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), "{}\n") + mustWriteFile(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), "{\"segments\":[]}\n") mustWriteFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n") } diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index c0ff1c8..6278909 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -209,7 +209,14 @@ func TestExecuteStagesArchiveFailsWhenRequiredRecapPromotionMissingForSelectedAr Enabled: boolPtr(true), UploadRun: boolPtr(true), PromoteArtifacts: []config.ArchivePromotionRule{ - {From: "artifacts/session_recap.md", To: "artifacts/session_recap.md", Required: boolPtr(true)}, + {Source: "narratio.artifact.session_recap", Dest: "artifacts/session_recap.md", Required: boolPtr(true)}, + }, + } + cfg.Pipeline.Scriptorium = &config.ScriptoriumConfig{ + Artifacts: map[string]config.ScriptoriumArtifactConfig{ + "session_recap": { + OutputPath: "artifacts/session_recap.md", + }, }, } @@ -247,8 +254,8 @@ func TestExecuteStagesArchiveFailsWhenRequiredRecapPromotionMissingForSelectedAr if err == nil { t.Fatal("expected archive promotion failure, got nil") } - if !strings.Contains(err.Error(), "required promotion source missing") { - t.Fatalf("error = %q, want required promotion source missing", err.Error()) + if !strings.Contains(err.Error(), "required promotion source unavailable") { + t.Fatalf("error = %q, want required promotion source unavailable", err.Error()) } } diff --git a/internal/config/config.go b/internal/config/config.go index 7400f5e..31f1cc9 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -79,8 +79,8 @@ type ArchiveConfig struct { // ArchivePromotionRule configures one artifact promotion mapping. type ArchivePromotionRule struct { - From string `yaml:"from"` - To string `yaml:"to"` + Source string `yaml:"source"` + Dest string `yaml:"dest"` Required *bool `yaml:"required"` } diff --git a/internal/config/defaults.go b/internal/config/defaults.go index ee05de2..ee9c8a6 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -74,8 +74,7 @@ const ( // DefaultArchivePromoteArtifacts defines the default archive promotion rules. // Callers should copy this slice before mutating. var DefaultArchivePromoteArtifacts = []ArchivePromotionRule{ - {From: PathTranscriptTrimmed, To: PathTranscriptTrimmed}, - {From: "artifacts/session_recap.md", To: "artifacts/session_recap.md"}, + {Source: "narratio.transcript.trimmed", Dest: PathTranscriptTrimmed}, } // DefaultPipelineConfigSearchPaths defines the default search order for diff --git a/internal/config/storage_archive_test.go b/internal/config/storage_archive_test.go index 04903fd..0e77768 100644 --- a/internal/config/storage_archive_test.go +++ b/internal/config/storage_archive_test.go @@ -136,40 +136,86 @@ func TestSpoolAndArchiveDefaults(t *testing.T) { if cfg.Pipeline.Archive.UploadRun == nil || !*cfg.Pipeline.Archive.UploadRun { t.Fatalf("archive.upload_run = %#v, want true", cfg.Pipeline.Archive.UploadRun) } - if len(cfg.Pipeline.Archive.PromoteArtifacts) != 2 { - t.Fatalf("archive.promote_artifacts len = %d, want 2 defaults", len(cfg.Pipeline.Archive.PromoteArtifacts)) + if len(cfg.Pipeline.Archive.PromoteArtifacts) != 1 { + t.Fatalf("archive.promote_artifacts len = %d, want 1 default", len(cfg.Pipeline.Archive.PromoteArtifacts)) } - for i, item := range cfg.Pipeline.Archive.PromoteArtifacts { - if item.Required == nil || !*item.Required { - t.Fatalf("archive.promote_artifacts[%d].required = %#v, want true", i, item.Required) - } + item := cfg.Pipeline.Archive.PromoteArtifacts[0] + if item.Required == nil || !*item.Required { + t.Fatalf("archive.promote_artifacts[0].required = %#v, want true", item.Required) + } + if item.Source != "narratio.transcript.trimmed" { + t.Fatalf("archive.promote_artifacts[0].source = %q, want narratio.transcript.trimmed", item.Source) + } + if item.Dest != "transcripts/trimmed.json" { + t.Fatalf("archive.promote_artifacts[0].dest = %q, want transcripts/trimmed.json", item.Dest) } } -func TestArchivePromotionPathValidation(t *testing.T) { +func TestArchivePromotionValidation(t *testing.T) { tests := []struct { name string ruleYML string wantErr string }{ { - name: "absolute from path rejected", + name: "absolute dest path rejected", ruleYML: `archive: promote_artifacts: - - from: "/transcripts/trimmed.json" - to: "transcripts/trimmed.json" + - source: "narratio.transcript.trimmed" + dest: "/transcripts/trimmed.json" `, wantErr: "must be a relative path", }, { - name: "traversal to path rejected", + name: "traversal dest path rejected", ruleYML: `archive: promote_artifacts: - - from: "transcripts/trimmed.json" - to: "../trimmed.json" + - source: "narratio.transcript.trimmed" + dest: "../trimmed.json" `, wantErr: "must not contain path traversal", }, + { + name: "invalid source rejected", + ruleYML: `archive: + promote_artifacts: + - source: "narratio.unknown" + dest: "transcripts/trimmed.json" +`, + wantErr: "source \"narratio.unknown\" is unsupported", + }, + { + name: "duplicate destination rejected", + ruleYML: `archive: + promote_artifacts: + - source: "narratio.transcript.trimmed" + dest: "artifacts/shared.md" + - source: "narratio.transcript.full" + dest: "artifacts/shared.md" +`, + wantErr: "duplicates another archive promotion destination", + }, + { + name: "configured source requires configured artifact key", + ruleYML: `archive: + promote_artifacts: + - source: "narratio.artifact.session_recap" + dest: "artifacts/session_recap.md" +`, + wantErr: "configured artifact \"session_recap\" is not defined in pipeline.scriptorium.artifacts", + }, + { + name: "configured source without output path fails when dest omitted", + ruleYML: `scriptorium: + artifacts: + session_recap: + enabled: false +archive: + promote_artifacts: + - source: "narratio.artifact.session_recap" +`, + wantErr: "destination cannot be derived", + }, } for _, tt := range tests { @@ -189,6 +235,72 @@ func TestArchivePromotionPathValidation(t *testing.T) { } } +func TestArchivePromotionDerivesDestinationWhenOmitted(t *testing.T) { + tests := []struct { + name string + pipelineYML string + wantDest string + }{ + { + name: "built in source derives canonical destination", + pipelineYML: testPipelineBaseYAML + ` +archive: + promote_artifacts: + - source: narratio.transcript.full +`, + wantDest: "transcripts/normalized.json", + }, + { + name: "configured source derives configured output path", + pipelineYML: testPipelineBaseYAML + ` +scriptorium: + artifacts: + session_recap: + enabled: true + prompt_id: dnd.session_recap + output_path: artifacts/session_recap.md +archive: + promote_artifacts: + - source: narratio.artifact.session_recap +`, + wantDest: "artifacts/session_recap.md", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + pipelinePath, sessionPath := writeConfigFiles(t, tt.pipelineYML, testSessionBaseYAML) + cfg, err := Load(pipelinePath, sessionPath) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if err := Validate(cfg); err != nil { + t.Fatalf("Validate() error = %v", err) + } + if len(cfg.Pipeline.Archive.PromoteArtifacts) != 1 { + t.Fatalf("archive.promote_artifacts len = %d, want 1", len(cfg.Pipeline.Archive.PromoteArtifacts)) + } + if cfg.Pipeline.Archive.PromoteArtifacts[0].Dest != tt.wantDest { + t.Fatalf("archive.promote_artifacts[0].dest = %q, want %q", cfg.Pipeline.Archive.PromoteArtifacts[0].Dest, tt.wantDest) + } + }) + } +} + +func TestArchiveLegacyFromToFailsStrictDecode(t *testing.T) { + pipelineYAML := testPipelineBaseYAML + ` +archive: + promote_artifacts: + - from: transcripts/trimmed.json + to: transcripts/trimmed.json +` + pipelinePath, sessionPath := writeConfigFiles(t, pipelineYAML, testSessionBaseYAML) + _, err := Load(pipelinePath, sessionPath) + if err == nil || !strings.Contains(err.Error(), "strict decode failed") { + t.Fatalf("Load() error = %v, want strict decode failed", err) + } +} + func TestSessionAudioS3Validation(t *testing.T) { tests := []struct { name string diff --git a/internal/config/validate.go b/internal/config/validate.go index 307f4d1..a81f94e 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -47,7 +47,7 @@ func validatePipeline(cfg *PipelineConfig) error { if err := validateSpool(cfg.Spool); err != nil { return err } - if err := validateArchive(cfg.Archive); err != nil { + if err := validateArchive(cfg.Archive, cfg.Scriptorium); err != nil { return err } if err := validateWhisperX(cfg.WhisperX); err != nil { @@ -104,28 +104,91 @@ func validateSpool(cfg SpoolConfig) error { return nil } -func validateArchive(cfg *ArchiveConfig) error { +func validateArchive(cfg *ArchiveConfig, scriptorium *ScriptoriumConfig) error { if cfg == nil { return nil } + seenDest := map[string]struct{}{} for i, item := range cfg.PromoteArtifacts { prefix := fmt.Sprintf("pipeline.archive.promote_artifacts[%d]", i) - if strings.TrimSpace(item.From) == "" { - return fmt.Errorf("%s.from is required", prefix) + source := strings.TrimSpace(item.Source) + if source == "" { + return fmt.Errorf("%s.source is required", prefix) } - if strings.TrimSpace(item.To) == "" { - return fmt.Errorf("%s.to is required", prefix) + if _, err := archiveSourceKnown(source, scriptorium); err != nil { + return fmt.Errorf("%s.source %q is unsupported: %w", prefix, item.Source, err) } - if err := validateRelativeSafePath(prefix+".from", item.From); err != nil { + dest := strings.TrimSpace(item.Dest) + if dest == "" { + derivedDest, err := deriveArchivePromotionDest(source, scriptorium) + if err != nil { + return fmt.Errorf("%s.dest is required when destination cannot be derived from %q: %w", prefix, source, err) + } + dest = derivedDest + cfg.PromoteArtifacts[i].Dest = derivedDest + } + if err := validateRelativeSafePath(prefix+".dest", dest); err != nil { return err } - if err := validateRelativeSafePath(prefix+".to", item.To); err != nil { - return err + normalizedDest := filepath.ToSlash(filepath.Clean(dest)) + if _, ok := seenDest[normalizedDest]; ok { + return fmt.Errorf("%s.dest %q duplicates another archive promotion destination", prefix, dest) } + seenDest[normalizedDest] = struct{}{} } return nil } +func archiveSourceKnown(source string, scriptorium *ScriptoriumConfig) (string, error) { + trimmed := strings.TrimSpace(source) + switch trimmed { + case "narratio.transcript.merged", + "narratio.transcript.polished", + "narratio.transcript.full", + "narratio.transcript.trimmed", + "narratio.bounds.session": + return "", nil + } + matches := narratioArtifactSourceRE.FindStringSubmatch(trimmed) + if len(matches) != 2 { + return "", fmt.Errorf("must be a built-in source id or narratio.artifact.") + } + artifactKey := matches[1] + if scriptorium == nil || len(scriptorium.Artifacts) == 0 { + return "", fmt.Errorf("configured artifact %q is not defined in pipeline.scriptorium.artifacts", artifactKey) + } + if _, ok := scriptorium.Artifacts[artifactKey]; !ok { + return "", fmt.Errorf("configured artifact %q is not defined in pipeline.scriptorium.artifacts", artifactKey) + } + return artifactKey, nil +} + +func deriveArchivePromotionDest(source string, scriptorium *ScriptoriumConfig) (string, error) { + trimmed := strings.TrimSpace(source) + switch trimmed { + case "narratio.transcript.merged": + return PathTranscriptMerged, nil + case "narratio.transcript.polished": + return PathTranscriptProcessed, nil + case "narratio.transcript.full": + return PathTranscriptNormalized, nil + case "narratio.transcript.trimmed": + return PathTranscriptTrimmed, nil + case "narratio.bounds.session": + return filepath.ToSlash(filepath.Join(PathArtifactsDirSegment, "session_bounds.json")), nil + } + artifactKey, err := archiveSourceKnown(trimmed, scriptorium) + if err != nil { + return "", err + } + artifactCfg := scriptorium.Artifacts[artifactKey] + outputPath := strings.TrimSpace(artifactCfg.OutputPath) + if outputPath == "" { + return "", fmt.Errorf("pipeline.scriptorium.artifacts.%s.output_path is empty", artifactKey) + } + return outputPath, nil +} + func validateSecrets(cfg *SecretsConfig) error { if cfg == nil { return nil diff --git a/internal/stage/archive.go b/internal/stage/archive.go index 5d027a0..c91561a 100644 --- a/internal/stage/archive.go +++ b/internal/stage/archive.go @@ -3,6 +3,7 @@ package stage import ( "context" "encoding/json" + "errors" "fmt" "io/fs" "os" @@ -116,11 +117,12 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S if err != nil { return nil, fmt.Errorf("archive: collect run files: %w", err) } - sessionRoot, err := resolveArchiveSessionRoot(env, m) + sessionPaths := archiveSessionPaths(env, m) + runtimeCatalog, err := buildArchiveRuntimeArtifactCatalog(sessionPaths, env.Config.Pipeline.Scriptorium) if err != nil { - return nil, fmt.Errorf("archive: resolve session root for promotions: %w", err) + return nil, fmt.Errorf("archive: build runtime artifact catalog: %w", err) } - promotions, err := resolveArchivePromotions(sessionRoot, env.Config.Pipeline.Archive.PromoteArtifacts) + 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) } @@ -134,20 +136,12 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S } 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) + 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 %q to %q: %w", promotion.From, key, err) + return nil, fmt.Errorf("archive: upload promoted output source %q to %q: %w", promotion.Source, key, err) } - promotedUploaded = append(promotedUploaded, promotion.To) + promotedUploaded = append(promotedUploaded, promotion.Dest) } currentManifestKey := artifacts.S3CurrentManifestKey(sessionPrefix) @@ -204,11 +198,11 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S } type archivePromotion struct { - From string - To string - Required bool - LocalPath string - Exists bool + Source string + Dest string + Required bool + LocalPath string + Provenance string } func archiveDisabled(env *Env) bool { @@ -292,6 +286,19 @@ func resolveArchiveSessionRoot(env *Env, m *manifest.Manifest) (string, error) { 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 archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) { runPrefix := strings.TrimSpace(m.S3RunPrefix) if runPrefix != "" { @@ -342,55 +349,138 @@ func archiveBucket(env *Env, m *manifest.Manifest) string { return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket) } -func resolveArchivePromotions(sessionRoot string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) { - sessionRoot = filepath.Clean(strings.TrimSpace(sessionRoot)) - if sessionRoot == "" { - return nil, fmt.Errorf("session root is required") - } +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 { - from := strings.TrimSpace(rule.From) - to := strings.TrimSpace(rule.To) + source := strings.TrimSpace(rule.Source) required := rule.Required == nil || *rule.Required - - resolvedPath, err := resolveWorkDirRelativePath(sessionRoot, from) + dest, err := resolveArchivePromotionDest(rule, catalog) if err != nil { - return nil, fmt.Errorf("promotion from %q: %w", from, err) + return nil, nil, fmt.Errorf("source %q: %w", source, err) } - info, err := os.Stat(resolvedPath) - exists := err == nil && !info.IsDir() - if err != nil && !os.IsNotExist(err) { - return nil, fmt.Errorf("promotion source %q: %w", from, 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) } - localPath := resolvedPath - out = append(out, archivePromotion{ - From: from, - To: to, - Required: required, - LocalPath: localPath, - Exists: exists, + Source: source, + Dest: dest, + Required: required, + LocalPath: resolved.Path, + Provenance: resolved.Provenance, }) } - return out, nil + return out, skippedOptional, nil } -func resolveWorkDirRelativePath(workDir, rel string) (string, error) { - rel = filepath.Clean(filepath.FromSlash(strings.TrimSpace(rel))) - if rel == "." || rel == "" { +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") } - 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) + cleaned := filepath.ToSlash(filepath.Clean(filepath.FromSlash(trimmed))) + if cleaned == "." || cleaned == "" { + return "", fmt.Errorf("relative path is required") } - if relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) { - return "", fmt.Errorf("path escapes workdir") + if filepath.IsAbs(trimmed) || strings.HasPrefix(cleaned, "/") || cleaned == ".." || strings.HasPrefix(cleaned, "../") { + return "", fmt.Errorf("path must be a clean relative path") } - return cleanedFull, nil + 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) { diff --git a/internal/stage/archive_test.go b/internal/stage/archive_test.go index 68ab6d7..737a9f2 100644 --- a/internal/stage/archive_test.go +++ b/internal/stage/archive_test.go @@ -135,8 +135,8 @@ func TestArchiveUploadsRunRecordPromotionsAndCurrentPointer(t *testing.T) { func TestArchiveUsesCustomPromotionRules(t *testing.T) { env, m, _ := 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)}, + {Source: "narratio.transcript.trimmed", Dest: "published/trimmed.json", Required: boolPtr(true)}, + {Source: "narratio.artifact.session_recap", Dest: "published/recap.md", Required: boolPtr(true)}, } _, err := archiveStage{}.Run(context.Background(), env, m) @@ -156,8 +156,8 @@ func TestArchiveUsesCustomPromotionRules(t *testing.T) { 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)}, + {Source: "narratio.transcript.trimmed", Dest: "transcripts/trimmed.json", Required: boolPtr(true)}, + {Source: "narratio.transcript.merged", Dest: "transcripts/merged.json", Required: boolPtr(false)}, } result, err := archiveStage{}.Run(context.Background(), env, m) @@ -165,7 +165,7 @@ func TestArchiveSkipsOptionalMissingPromotion(t *testing.T) { t.Fatalf("Run() error = %v", err) } got, _ := result.Metadata["skipped_optional_promotions"].([]string) - want := []string{"artifacts/optional.md"} + want := []string{"transcripts/merged.json"} if !reflect.DeepEqual(got, want) { t.Fatalf("skipped_optional_promotions = %#v, want %#v", got, want) } @@ -174,12 +174,12 @@ func TestArchiveSkipsOptionalMissingPromotion(t *testing.T) { 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)}, + {Source: "narratio.transcript.merged", Dest: "transcripts/merged.json", 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) + if err == nil || !strings.Contains(err.Error(), "required promotion source unavailable") { + t.Fatalf("Run() error = %v, want required promotion source unavailable failure", err) } } @@ -259,13 +259,13 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) { sessionRoot := artifacts.SessionWorkDirForCampaign(root, campaign, sessionID) runRoot := artifacts.SessionRunRootForCampaign(root, campaign, sessionID, runID) - writeStageTestFile(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), "{}\n") + writeStageTestFile(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), "{\"segments\":[]}\n") writeStageTestFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n") writeStageTestFile(t, filepath.Join(runRoot, "prepare", "inputs", "session.yml"), "session_id: 2026-04-19\n") writeStageTestFile(t, filepath.Join(runRoot, "prepare", "outputs", "audio", "speaker.flac"), "flac\n") writeStageTestFile(t, filepath.Join(runRoot, "transcribe", "outputs", "transcripts", "raw", "speaker.json"), "{}\n") - writeStageTestFile(t, filepath.Join(runRoot, "trim", "outputs", "transcripts", "trimmed.json"), "{}\n") + writeStageTestFile(t, filepath.Join(runRoot, "trim", "outputs", "transcripts", "trimmed.json"), "{\"segments\":[]}\n") writeStageTestFile(t, filepath.Join(runRoot, "analyze", "outputs", "artifacts", "session_recap.md"), "# recap\n") writeStageTestFile(t, filepath.Join(runRoot, "polish", "reports", "audita.report.json"), "{}\n") writeStageTestFile(t, filepath.Join(runRoot, "merge", "config", "seriatim.generated.yml"), "key: value\n") @@ -298,8 +298,17 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) { 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)}, + {Source: "narratio.transcript.trimmed", Dest: "transcripts/trimmed.json", Required: boolPtr(true)}, + {Source: "narratio.artifact.session_recap", Dest: "artifacts/session_recap.md", Required: boolPtr(true)}, + }, + }, + Scriptorium: &config.ScriptoriumConfig{ + Artifacts: map[string]config.ScriptoriumArtifactConfig{ + "session_recap": { + Enabled: true, + PromptID: "dnd.session_recap", + OutputPath: "artifacts/session_recap.md", + }, }, }, },