eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 15 Wiki Activity

Rename archive config and stage contract to publish

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-23 04:30:45 +00:00
parent f050b9dd54
commit df2c765b7f
24 changed files with 273 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
examples/pipeline.full.annotated.yml
View File

@@ -4,18 +4,18 @@
workspace: workspace:
# Optional: defaults to /var/lib/narratio. # Optional: defaults to /var/lib/narratio.
root: /var/lib/narratio/workspace root: /var/lib/narratio/workspace
# Optional: remove run-scoped workdir after successful archive commit. # Optional: remove run-scoped workdir after successful publish commit.
cleanup_after_archive: false cleanup_after_publish: false
# Optional: local secret file loader (directory of ENV_VAR_NAME files). # Optional: local secret file loader (directory of ENV_VAR_NAME files).
# secrets: # secrets:
# env_dir: ./secrets # env_dir: ./secrets
storage: storage:
# Optional storage backend selector; use "s3" for archive + S3 audio workflows. # Optional storage backend selector; use "s3" for publish + S3 audio workflows.
backend: s3 backend: s3
s3: s3:
# Required when using S3 audio or S3 archive uploads. # Required when using S3 audio or S3 publish uploads.
bucket: my-dnd-archive bucket: my-dnd-archive
# Optional; defaults to "dnd". # Optional; defaults to "dnd".
root_prefix: dnd root_prefix: dnd
@@ -36,15 +36,15 @@ campaigns:
spool: spool:
# Optional; defaults to /var/spool/narratio. # Optional; defaults to /var/spool/narratio.
root: /var/spool/narratio root: /var/spool/narratio
# Optional cleanup of run-scoped spool audio after successful archive commit. # Optional cleanup of run-scoped spool audio after successful publish commit.
delete_audio_after_archive: false delete_audio_after_publish: false
archive: publish:
# Optional booleans; defaults are true. # Optional booleans; defaults are true.
enabled: true enabled: true
upload_run: true upload_run: true
# Optional promotion rules; sources use Narratio artifact source IDs. # Optional promotion rules; sources use Narratio artifact source IDs.
promote_artifacts: outputs:
- source: narratio.transcript.final_trimmed - source: narratio.transcript.final_trimmed
dest: transcripts/final.trimmed.json dest: transcripts/final.trimmed.json
required: true required: true

8
examples/pipeline.production.yml
View File

@@ -1,6 +1,6 @@
workspace: workspace:
root: /var/lib/narratio/workspace root: /var/lib/narratio/workspace
cleanup_after_archive: true cleanup_after_publish: true
storage: storage:
backend: s3 backend: s3
@@ -17,12 +17,12 @@ campaigns:
spool: spool:
root: /var/spool/narratio root: /var/spool/narratio
delete_audio_after_archive: true delete_audio_after_publish: true
archive: publish:
enabled: true enabled: true
upload_run: true upload_run: true
promote_artifacts: outputs:
- source: narratio.transcript.final_trimmed - source: narratio.transcript.final_trimmed
dest: transcripts/final.trimmed.json dest: transcripts/final.trimmed.json
required: true required: true

18
internal/app/analyze_artifacts_commands_test.go
View File

@@ -28,7 +28,7 @@ func TestExecuteRunStageArtifactsUnsupportedStageFails(t *testing.T) {
if code == 0 { if code == 0 {
t.Fatal("exit code = 0, want non-zero") t.Fatal("exit code = 0, want non-zero")
} }
if !strings.Contains(stderr.String(), `run-stage: --artifacts is only supported for stages "analyze" and "archive"`) { if !strings.Contains(stderr.String(), `run-stage: --artifacts is only supported for stages "analyze" and "publish"`) {
t.Fatalf("stderr = %q, want stage-gating error", stderr.String()) t.Fatalf("stderr = %q, want stage-gating error", stderr.String())
} }
} }
@@ -48,14 +48,14 @@ func TestExecuteRunStageArchivePropagatesSelectedArtifacts(t *testing.T) {
capturedStages = append(capturedStages, s.Name()) capturedStages = append(capturedStages, s.Name())
} }
capturedArtifacts = append([]string(nil), opts.SelectedArtifacts...) capturedArtifacts = append([]string(nil), opts.SelectedArtifacts...)
return &RunSummary{ManifestPath: filepath.Join(workspaceRoot, "manifest.json"), Executed: []string{"archive"}}, nil return &RunSummary{ManifestPath: filepath.Join(workspaceRoot, "manifest.json"), Executed: []string{"publish"}}, nil
} }
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute( code := Execute(
[]string{ []string{
"run-stage", "archive", "2026-05-03", "run-stage", "publish", "2026-05-03",
"--config", pipelinePath, "--config", pipelinePath,
"--campaign-file", campaignPath, "--campaign-file", campaignPath,
"--session", sessionPath, "--session", sessionPath,
@@ -67,8 +67,8 @@ func TestExecuteRunStageArchivePropagatesSelectedArtifacts(t *testing.T) {
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
if len(capturedStages) != 1 || capturedStages[0] != "archive" { if len(capturedStages) != 1 || capturedStages[0] != "publish" {
t.Fatalf("captured stages = %#v, want [archive]", capturedStages) t.Fatalf("captured stages = %#v, want [publish]", capturedStages)
} }
if strings.Join(capturedArtifacts, ",") != "session_recap" { if strings.Join(capturedArtifacts, ",") != "session_recap" {
t.Fatalf("captured artifacts = %#v, want [session_recap]", capturedArtifacts) t.Fatalf("captured artifacts = %#v, want [session_recap]", capturedArtifacts)
@@ -127,7 +127,7 @@ func TestResumeArtifactsWithSucceededAnalyzeSkipsUnlessForced(t *testing.T) {
store := &manifest.LocalStore{} store := &manifest.LocalStore{}
seed := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) seed := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "archive", "notify"} { for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "publish", "notify"} {
seed.MarkStageSucceeded(stageName, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) seed.MarkStageSucceeded(stageName, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil)
} }
if err := store.Save(context.Background(), manifestPath, seed); err != nil { if err := store.Save(context.Background(), manifestPath, seed); err != nil {
@@ -300,7 +300,7 @@ func TestExecutePublishForceRunsArchive(t *testing.T) {
capturedArtifacts = append([]string(nil), opts.SelectedArtifacts...) capturedArtifacts = append([]string(nil), opts.SelectedArtifacts...)
return &RunSummary{ return &RunSummary{
ManifestPath: filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json"), ManifestPath: filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json"),
Executed: []string{"archive"}, Executed: []string{"publish"},
}, nil }, nil
} }
@@ -314,8 +314,8 @@ func TestExecutePublishForceRunsArchive(t *testing.T) {
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
if len(capturedStages) != 1 || capturedStages[0] != "archive" { if len(capturedStages) != 1 || capturedStages[0] != "publish" {
t.Fatalf("captured stages = %#v, want [archive]", capturedStages) t.Fatalf("captured stages = %#v, want [publish]", capturedStages)
} }
if !capturedForce { if !capturedForce {
t.Fatal("captured force = false, want true") t.Fatal("captured force = false, want true")

20
internal/app/commands_test.go
View File

@@ -32,7 +32,7 @@ func TestExecuteValidCommands(t *testing.T) {
wantOut string wantOut string
}{ }{
{name: "run", args: []string{"run", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run: session 2026-05-03; executed=9 skipped=0; manifest="}, {name: "run", args: []string{"run", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run: session 2026-05-03; executed=9 skipped=0; manifest="},
{name: "session plan", args: []string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "prepare: skip\ntranscribe: skip\nmerge: skip\npolish: skip\nnormalize: skip\ntrim: skip\nanalyze: skip\narchive: skip\nnotify: skip"}, {name: "session plan", args: []string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "prepare: skip\ntranscribe: skip\nmerge: skip\npolish: skip\nnormalize: skip\ntrim: skip\nanalyze: skip\npublish: skip\nnotify: skip"},
{name: "session status", args: []string{"session", "status", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "Session: 2026-05-03"}, {name: "session status", args: []string{"session", "status", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "Session: 2026-05-03"},
{name: "resume", args: []string{"resume", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio resume: session 2026-05-03 has no remaining stages"}, {name: "resume", args: []string{"resume", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio resume: session 2026-05-03 has no remaining stages"},
{name: "run-stage", args: []string{"run-stage", "polish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run-stage: stage=polish executed=0 skipped=1 force=false; manifest="}, {name: "run-stage", args: []string{"run-stage", "polish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run-stage: stage=polish executed=0 skipped=1 force=false; manifest="},
@@ -107,6 +107,22 @@ func TestExecuteRunStageUnknownFails(t *testing.T) {
} }
} }
func TestExecuteRunStageArchiveAliasFails(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot, "https://example.com/transcribe")
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"run-stage", "archive", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, &stdout, &stderr)
if code == 0 {
t.Fatal("exit code = 0, want non-zero")
}
if !strings.Contains(stderr.String(), `unknown stage "archive"`) {
t.Fatalf("stderr = %q, want unknown archive stage error", stderr.String())
}
}
func TestExecuteRunStageNormalizeIsAccepted(t *testing.T) { func TestExecuteRunStageNormalizeIsAccepted(t *testing.T) {
workspaceRoot := t.TempDir() workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot, "https://example.com/transcribe") pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot, "https://example.com/transcribe")
@@ -469,7 +485,7 @@ storage:
backend: s3 backend: s3
s3: s3:
bucket: test-bucket bucket: test-bucket
archive: publish:
enabled: true enabled: true
upload_run: false upload_run: false
whisperx: whisperx:

8
internal/app/operator_helpers.go
View File

@@ -1001,7 +1001,7 @@ func writeArtifactList(out io.Writer, cfg *config.Config, catalog *artifacts.Art
fmt.Fprintf(out, "- narratio.previous_session.artifact.%s required=%t\n", req.Name, req.Required) fmt.Fprintf(out, "- narratio.previous_session.artifact.%s required=%t\n", req.Name, req.Required)
} }
fmt.Fprintln(out, "Promoted:") fmt.Fprintln(out, "Promoted:")
for _, rule := range cfg.Pipeline.Archive.PromoteArtifacts { for _, rule := range cfg.Pipeline.Publish.PromoteArtifacts {
writePromotedArtifactLine(out, rule, catalog, lockSet, promotedRemoteState) writePromotedArtifactLine(out, rule, catalog, lockSet, promotedRemoteState)
} }
} }
@@ -1038,7 +1038,7 @@ func writePromotedArtifactLine(out io.Writer, rule config.ArchivePromotionRule,
func remotePromotionAvailability(ctx context.Context, cfg *config.Config, store storage.ObjectStore, catalog *artifacts.ArtifactCatalog) map[string]string { func remotePromotionAvailability(ctx context.Context, cfg *config.Config, store storage.ObjectStore, catalog *artifacts.ArtifactCatalog) map[string]string {
out := map[string]string{} out := map[string]string{}
sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID) sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID)
for _, rule := range cfg.Pipeline.Archive.PromoteArtifacts { for _, rule := range cfg.Pipeline.Publish.PromoteArtifacts {
source := strings.TrimSpace(rule.Source) source := strings.TrimSpace(rule.Source)
dest, _, err := helperPromotionDest(rule, catalog) dest, _, err := helperPromotionDest(rule, catalog)
if err != nil { if err != nil {
@@ -1105,8 +1105,8 @@ func writeLocks(out io.Writer, cfg *config.Config, locks *effectiveLocks) {
} }
fmt.Fprintln(out, "Archive locks:") fmt.Fprintln(out, "Archive locks:")
promoted := map[string]config.ArchivePromotionRule{} promoted := map[string]config.ArchivePromotionRule{}
if cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Archive != nil { if cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Publish != nil {
for _, rule := range cfg.Pipeline.Archive.PromoteArtifacts { for _, rule := range cfg.Pipeline.Publish.PromoteArtifacts {
promoted[strings.TrimSpace(rule.Source)] = rule promoted[strings.TrimSpace(rule.Source)] = rule
} }
} }

14
internal/app/operator_helpers_test.go
View File

@@ -687,7 +687,7 @@ func TestExecuteArtifactsListRemoteReportsPromotedAvailability(t *testing.T) {
workspaceRoot := t.TempDir() workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
addArchivePromotionsToPipeline(t, pipelinePath, ` addArchivePromotionsToPipeline(t, pipelinePath, `
promote_artifacts: outputs:
- source: narratio.transcript.final_trimmed - source: narratio.transcript.final_trimmed
dest: transcripts/final.trimmed.json dest: transcripts/final.trimmed.json
required: true required: true
@@ -722,7 +722,7 @@ func TestExecuteArtifactsListRemoteUsesPromotionDestinations(t *testing.T) {
workspaceRoot := t.TempDir() workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
addArchivePromotionsToPipeline(t, pipelinePath, ` addArchivePromotionsToPipeline(t, pipelinePath, `
promote_artifacts: outputs:
- source: narratio.transcript.final - source: narratio.transcript.final
dest: transcripts/full.json dest: transcripts/full.json
required: true required: true
@@ -772,7 +772,7 @@ func TestExecuteStatusReportsRemoteArtifactCatalog(t *testing.T) {
workspaceRoot := t.TempDir() workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
addArchivePromotionsToPipeline(t, pipelinePath, ` addArchivePromotionsToPipeline(t, pipelinePath, `
promote_artifacts: outputs:
- source: narratio.transcript.final_trimmed - source: narratio.transcript.final_trimmed
dest: transcripts/final.trimmed.json dest: transcripts/final.trimmed.json
required: true required: true
@@ -829,7 +829,7 @@ func TestExecuteStatusReportsRemoteArtifactCatalogErrorsWithoutFailing(t *testin
workspaceRoot := t.TempDir() workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
addArchivePromotionsToPipeline(t, pipelinePath, ` addArchivePromotionsToPipeline(t, pipelinePath, `
promote_artifacts: outputs:
- source: narratio.transcript.final_trimmed - source: narratio.transcript.final_trimmed
dest: transcripts/final.trimmed.json dest: transcripts/final.trimmed.json
required: true required: true
@@ -879,7 +879,7 @@ func TestExecuteArchiveLoadsRemoteLocks(t *testing.T) {
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"run-stage", "archive", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--force"}, &stdout, &stderr) code := Execute([]string{"run-stage", "publish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--force"}, &stdout, &stderr)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
@@ -949,8 +949,8 @@ func addStaticArchiveLockToPipelineConfig(t *testing.T, pipelinePath, source str
} }
updated := strings.Replace( updated := strings.Replace(
string(data), string(data),
"archive:\n enabled: true\n upload_run: false\n", "publish:\n enabled: true\n upload_run: false\n",
"archive:\n enabled: true\n upload_run: false\n locks:\n - source: "+source+"\n reason: static review\n", "publish:\n enabled: true\n upload_run: false\n locks:\n - source: "+source+"\n reason: static review\n",
1, 1,
) )
if updated == string(data) { if updated == string(data) {

2
internal/app/plan_test.go
View File

@@ -27,7 +27,7 @@ func TestPlanCreatesAndReusesWorkdir(t *testing.T) {
if !strings.Contains(got, "narratio session plan: workdir prepared at") { if !strings.Contains(got, "narratio session plan: workdir prepared at") {
t.Fatalf("first output = %q, want workdir prepared", got) t.Fatalf("first output = %q, want workdir prepared", got)
} }
for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "archive", "notify"} { for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "publish", "notify"} {
if !strings.Contains(got, name+": run") { if !strings.Contains(got, name+": run") {
t.Fatalf("first output = %q, missing stage %q", got, name) t.Fatalf("first output = %q, missing stage %q", got, name)
} }

2
internal/app/planner_test.go
View File

@@ -4,7 +4,7 @@ import "testing"
func TestBuildFullPlanOrder(t *testing.T) { func TestBuildFullPlanOrder(t *testing.T) {
got := BuildFullPlan() got := BuildFullPlan()
want := []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "archive", "notify"} want := []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "publish", "notify"}
if len(got) != len(want) { if len(got) != len(want) {
t.Fatalf("len(plan) = %d, want %d", len(got), len(want)) t.Fatalf("len(plan) = %d, want %d", len(got), len(want))
} }

48
internal/app/post_archive_cleanup.go
View File

@@ -17,8 +17,8 @@ func runPostArchiveCleanup(ctx context.Context, env *Env, manifestPath string, m
return nil return nil
} }
spoolRequested := env.Config.Pipeline.Spool.DeleteAudioAfterArchive spoolRequested := env.Config.Pipeline.Spool.DeleteAudioAfterPublish
workRequested := env.Config.Pipeline.Workspace.CleanupAfterArchive workRequested := env.Config.Pipeline.Workspace.CleanupAfterPublish
if !spoolRequested && !workRequested { if !spoolRequested && !workRequested {
return nil return nil
} }
@@ -63,9 +63,9 @@ func runPostArchiveCleanup(ctx context.Context, env *Env, manifestPath string, m
} }
if spoolRequested { if spoolRequested {
if err := removeRunScopedDir(strings.TrimSpace(env.Config.Pipeline.Spool.Root), spoolDir, "pipeline.spool.delete_audio_after_archive"); err != nil { if err := removeRunScopedDir(strings.TrimSpace(env.Config.Pipeline.Spool.Root), spoolDir, "pipeline.spool.delete_audio_after_publish"); err != nil {
sr.Metadata["cleanup_failed"] = true sr.Metadata["cleanup_failed"] = true
sr.Metadata["cleanup_failed_policy"] = "pipeline.spool.delete_audio_after_archive" sr.Metadata["cleanup_failed_policy"] = "pipeline.spool.delete_audio_after_publish"
sr.Metadata["cleanup_failed_path"] = spoolDir sr.Metadata["cleanup_failed_path"] = spoolDir
_ = env.ManifestStore.Save(ctx, manifestPath, m) _ = env.ManifestStore.Save(ctx, manifestPath, m)
return err return err
@@ -82,9 +82,9 @@ func runPostArchiveCleanup(ctx context.Context, env *Env, manifestPath string, m
return nil return nil
} }
if err := removeRunScopedDir(strings.TrimSpace(env.Config.Pipeline.Workspace.Root), workDir, "pipeline.workspace.cleanup_after_archive"); err != nil { if err := removeRunScopedDir(strings.TrimSpace(env.Config.Pipeline.Workspace.Root), workDir, "pipeline.workspace.cleanup_after_publish"); err != nil {
sr.Metadata["cleanup_failed"] = true sr.Metadata["cleanup_failed"] = true
sr.Metadata["cleanup_failed_policy"] = "pipeline.workspace.cleanup_after_archive" sr.Metadata["cleanup_failed_policy"] = "pipeline.workspace.cleanup_after_publish"
sr.Metadata["cleanup_failed_path"] = workDir sr.Metadata["cleanup_failed_path"] = workDir
_ = env.ManifestStore.Save(ctx, manifestPath, m) _ = env.ManifestStore.Save(ctx, manifestPath, m)
return err return err
@@ -100,17 +100,17 @@ func archiveStageRecordForCleanup(m *manifest.Manifest, executed []string) *mani
if m == nil { if m == nil {
return nil return nil
} }
archiveRan := false publishRan := false
for _, name := range executed { for _, name := range executed {
if name == "archive" { if name == "publish" {
archiveRan = true publishRan = true
break break
} }
} }
if !archiveRan { if !publishRan {
return nil return nil
} }
sr := m.Stages["archive"] sr := m.Stages["publish"]
if sr == nil || sr.Status != manifest.StatusSucceeded { if sr == nil || sr.Status != manifest.StatusSucceeded {
return nil return nil
} }
@@ -118,37 +118,37 @@ func archiveStageRecordForCleanup(m *manifest.Manifest, executed []string) *mani
} }
func archiveCleanupEligible(cfg *config.Config, sr *manifest.StageRecord) (bool, string) { func archiveCleanupEligible(cfg *config.Config, sr *manifest.StageRecord) (bool, string) {
if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Archive == nil { if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Publish == nil {
return false, "archive configuration is missing" return false, "publish configuration is missing"
} }
enabled := true enabled := true
if cfg.Pipeline.Archive.Enabled != nil { if cfg.Pipeline.Publish.Enabled != nil {
enabled = *cfg.Pipeline.Archive.Enabled enabled = *cfg.Pipeline.Publish.Enabled
} }
if !enabled { if !enabled {
return false, "archive.enabled is false" return false, "publish.enabled is false"
} }
uploadRun := true uploadRun := true
if cfg.Pipeline.Archive.UploadRun != nil { if cfg.Pipeline.Publish.UploadRun != nil {
uploadRun = *cfg.Pipeline.Archive.UploadRun uploadRun = *cfg.Pipeline.Publish.UploadRun
} }
if !uploadRun { if !uploadRun {
return false, "archive.upload_run is false" return false, "publish.upload_run is false"
} }
if sr == nil || sr.Metadata == nil { if sr == nil || sr.Metadata == nil {
return false, "archive metadata is missing" return false, "publish metadata is missing"
} }
if skipped, _ := sr.Metadata["skipped"].(bool); skipped { if skipped, _ := sr.Metadata["skipped"].(bool); skipped {
return false, "archive stage was skipped" return false, "publish stage was skipped"
} }
if uploaded, _ := sr.Metadata["uploaded"].(bool); !uploaded { if uploaded, _ := sr.Metadata["uploaded"].(bool); !uploaded {
return false, "archive did not upload run record" return false, "publish did not upload run record"
} }
if pointer, _ := sr.Metadata["current_pointer_written"].(bool); !pointer { if pointer, _ := sr.Metadata["current_pointer_written"].(bool); !pointer {
return false, "archive did not write current pointer" return false, "publish did not write current pointer"
} }
if strings.TrimSpace(asString(sr.Metadata["current_run_id_key"])) == "" { if strings.TrimSpace(asString(sr.Metadata["current_run_id_key"])) == "" {
return false, "archive current run pointer key is missing" return false, "publish current run pointer key is missing"
} }
return true, "" return true, ""
} }

82
internal/app/post_archive_cleanup_test.go
View File

@@ -20,11 +20,11 @@ type archiveSuccessStage struct {
metadata map[string]any metadata map[string]any
} }
func (archiveSuccessStage) Name() string { return "archive" } func (archiveSuccessStage) Name() string { return "publish" }
func (archiveSuccessStage) Declares() stage.IODecl { return stage.IODecl{} } func (archiveSuccessStage) Declares() stage.IODecl { return stage.IODecl{} }
func (s archiveSuccessStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { func (s archiveSuccessStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) {
md := map[string]any{ md := map[string]any{
"stage": "archive", "stage": "publish",
"uploaded": true, "uploaded": true,
"current_pointer_written": true, "current_pointer_written": true,
"current_run_id_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/current/run_id.txt", "current_run_id_key": "dnd/campaigns/sample-campaign/sessions/2026-05-03/current/run_id.txt",
@@ -45,8 +45,8 @@ func (notifyFailStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest
func TestPostArchiveCleanupDisabledKeepsLocalDirs(t *testing.T) { func TestPostArchiveCleanupDisabledKeepsLocalDirs(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t) cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = false cfg.Pipeline.Spool.DeleteAudioAfterPublish = false
cfg.Pipeline.Workspace.CleanupAfterArchive = false cfg.Pipeline.Workspace.CleanupAfterPublish = false
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err) t.Fatalf("executeStages() error = %v", err)
@@ -59,8 +59,8 @@ func TestPostArchiveCleanupDisabledKeepsLocalDirs(t *testing.T) {
func TestPostArchiveCleanupSpoolOnly(t *testing.T) { func TestPostArchiveCleanupSpoolOnly(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t) cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = false cfg.Pipeline.Workspace.CleanupAfterPublish = false
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err) t.Fatalf("executeStages() error = %v", err)
@@ -73,8 +73,8 @@ func TestPostArchiveCleanupSpoolOnly(t *testing.T) {
func TestPostArchiveCleanupWorkdirOnly(t *testing.T) { func TestPostArchiveCleanupWorkdirOnly(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t) cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = false cfg.Pipeline.Spool.DeleteAudioAfterPublish = false
cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterPublish = true
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err) t.Fatalf("executeStages() error = %v", err)
@@ -89,8 +89,8 @@ func TestPostArchiveCleanupWorkdirOnly(t *testing.T) {
func TestPostArchiveCleanupBothPolicies(t *testing.T) { func TestPostArchiveCleanupBothPolicies(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t) cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterPublish = true
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err) t.Fatalf("executeStages() error = %v", err)
@@ -104,12 +104,12 @@ func TestPostArchiveCleanupBothPolicies(t *testing.T) {
func TestPostArchiveCleanupNotRunWhenArchiveFails(t *testing.T) { func TestPostArchiveCleanupNotRunWhenArchiveFails(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t) cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterPublish = true
_, err := executeStages(context.Background(), cfg, []stage.Stage{failingStage{name: "archive", err: errors.New("archive failed")}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) _, err := executeStages(context.Background(), cfg, []stage.Stage{failingStage{name: "publish", err: errors.New("archive failed")}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}})
if err == nil || !strings.Contains(err.Error(), "stage \"archive\" failed") { if err == nil || !strings.Contains(err.Error(), "stage \"publish\" failed") {
t.Fatalf("executeStages() error = %v, want archive failure", err) t.Fatalf("executeStages() error = %v, want publish failure", err)
} }
assertExists(t, seed.spoolAudioDir) assertExists(t, seed.spoolAudioDir)
@@ -118,8 +118,8 @@ func TestPostArchiveCleanupNotRunWhenArchiveFails(t *testing.T) {
func TestPostArchiveCleanupNotRunWhenArchiveSkipped(t *testing.T) { func TestPostArchiveCleanupNotRunWhenArchiveSkipped(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t) cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterPublish = true
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{metadata: map[string]any{"skipped": true}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{metadata: map[string]any{"skipped": true}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err) t.Fatalf("executeStages() error = %v", err)
@@ -131,8 +131,8 @@ func TestPostArchiveCleanupNotRunWhenArchiveSkipped(t *testing.T) {
func TestPostArchiveCleanupNotRunWhenCurrentPointerMissing(t *testing.T) { func TestPostArchiveCleanupNotRunWhenCurrentPointerMissing(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t) cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterPublish = true
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{metadata: map[string]any{"current_pointer_written": false}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{metadata: map[string]any{"current_pointer_written": false}}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err) t.Fatalf("executeStages() error = %v", err)
@@ -144,9 +144,9 @@ func TestPostArchiveCleanupNotRunWhenCurrentPointerMissing(t *testing.T) {
func TestPostArchiveCleanupNotRunWhenArchiveUploadDisabled(t *testing.T) { func TestPostArchiveCleanupNotRunWhenArchiveUploadDisabled(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t) cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterPublish = true
cfg.Pipeline.Archive.UploadRun = boolPtr(false) cfg.Pipeline.Publish.UploadRun = boolPtr(false)
if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { if _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil {
t.Fatalf("executeStages() error = %v", err) t.Fatalf("executeStages() error = %v", err)
@@ -158,8 +158,8 @@ func TestPostArchiveCleanupNotRunWhenArchiveUploadDisabled(t *testing.T) {
func TestPostArchiveCleanupWaitsUntilAllStagesSucceed(t *testing.T) { func TestPostArchiveCleanupWaitsUntilAllStagesSucceed(t *testing.T) {
cfg, seed := cleanupFixtureConfig(t) cfg, seed := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterPublish = true
_, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}, notifyFailStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) _, err := executeStages(context.Background(), cfg, []stage.Stage{archiveSuccessStage{}, notifyFailStage{}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}})
if err == nil || !strings.Contains(err.Error(), "stage \"notify\" failed") { if err == nil || !strings.Contains(err.Error(), "stage \"notify\" failed") {
@@ -172,8 +172,8 @@ func TestPostArchiveCleanupWaitsUntilAllStagesSucceed(t *testing.T) {
func TestPostArchiveCleanupFailsOnUnsafePath(t *testing.T) { func TestPostArchiveCleanupFailsOnUnsafePath(t *testing.T) {
cfg, _ := cleanupFixtureConfig(t) cfg, _ := cleanupFixtureConfig(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = false cfg.Pipeline.Workspace.CleanupAfterPublish = false
manifestPath := manifestPathFor(cfg) manifestPath := manifestPathFor(cfg)
store := &manifest.LocalStore{} store := &manifest.LocalStore{}
@@ -194,15 +194,15 @@ func TestPostArchiveCleanupFailsOnUnsafePath(t *testing.T) {
func TestPostArchiveCleanupNotRunWhenPromotionIsMissing(t *testing.T) { func TestPostArchiveCleanupNotRunWhenPromotionIsMissing(t *testing.T) {
cfg, seed, runID := archiveStageCleanupFixture(t) cfg, seed, runID := archiveStageCleanupFixture(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterPublish = true
cfg.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ cfg.Pipeline.Publish.PromoteArtifacts = []config.ArchivePromotionRule{
{Source: "narratio.transcript.base", Dest: "transcripts/base.json", Required: boolPtr(true)}, {Source: "narratio.transcript.base", Dest: "transcripts/base.json", Required: boolPtr(true)},
} }
archiveStageImpl, err := stage.Select("archive") archiveStageImpl, err := stage.Select("publish")
if err != nil { if err != nil {
t.Fatalf("Select(archive) error = %v", err) t.Fatalf("Select(publish) error = %v", err)
} }
_, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) _, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}})
if err == nil || !strings.Contains(err.Error(), "required promotion source unavailable") { if err == nil || !strings.Contains(err.Error(), "required promotion source unavailable") {
@@ -217,13 +217,13 @@ func TestPostArchiveCleanupNotRunWhenPromotionIsMissing(t *testing.T) {
func TestPostArchiveCleanupNotRunWhenCurrentManifestUploadFails(t *testing.T) { func TestPostArchiveCleanupNotRunWhenCurrentManifestUploadFails(t *testing.T) {
cfg, seed, _ := archiveStageCleanupFixture(t) cfg, seed, _ := archiveStageCleanupFixture(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterPublish = true
failKey := seed.sessionPrefix + "current/manifest.json" failKey := seed.sessionPrefix + "current/manifest.json"
archiveStageImpl, err := stage.Select("archive") archiveStageImpl, err := stage.Select("publish")
if err != nil { if err != nil {
t.Fatalf("Select(archive) error = %v", err) t.Fatalf("Select(publish) error = %v", err)
} }
_, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{ _, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{
Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}}, Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}},
@@ -238,13 +238,13 @@ func TestPostArchiveCleanupNotRunWhenCurrentManifestUploadFails(t *testing.T) {
func TestPostArchiveCleanupNotRunWhenCurrentPointerUploadFails(t *testing.T) { func TestPostArchiveCleanupNotRunWhenCurrentPointerUploadFails(t *testing.T) {
cfg, seed, _ := archiveStageCleanupFixture(t) cfg, seed, _ := archiveStageCleanupFixture(t)
cfg.Pipeline.Spool.DeleteAudioAfterArchive = true cfg.Pipeline.Spool.DeleteAudioAfterPublish = true
cfg.Pipeline.Workspace.CleanupAfterArchive = true cfg.Pipeline.Workspace.CleanupAfterPublish = true
failKey := seed.sessionPrefix + "current/run_id.txt" failKey := seed.sessionPrefix + "current/run_id.txt"
archiveStageImpl, err := stage.Select("archive") archiveStageImpl, err := stage.Select("publish")
if err != nil { if err != nil {
t.Fatalf("Select(archive) error = %v", err) t.Fatalf("Select(publish) error = %v", err)
} }
_, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{ _, err = executeStages(context.Background(), cfg, []stage.Stage{archiveStageImpl}, RunOptions{
Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}}, Env: &Env{ObjectStore: &failKeyStore{delegate: &storage.FakeBackend{}, failKey: failKey}},
@@ -270,7 +270,7 @@ func cleanupFixtureConfig(t *testing.T) (*config.Config, cleanupSeed) {
t.Helper() t.Helper()
cfg := testConfig(t) cfg := testConfig(t)
cfg.Pipeline.Archive = &config.ArchiveConfig{Enabled: boolPtr(true), UploadRun: boolPtr(true)} cfg.Pipeline.Publish = &config.ArchiveConfig{Enabled: boolPtr(true), UploadRun: boolPtr(true)}
cfg.Pipeline.Spool.Root = filepath.Join(t.TempDir(), "spool") cfg.Pipeline.Spool.Root = filepath.Join(t.TempDir(), "spool")
runID := "20260516T010203Z-1a2b3c4d" runID := "20260516T010203Z-1a2b3c4d"
@@ -329,7 +329,7 @@ func archiveStageCleanupFixture(t *testing.T) (*config.Config, cleanupSeed, stri
Bucket: "my-dnd-archive", Bucket: "my-dnd-archive",
RootPrefix: "dnd", RootPrefix: "dnd",
} }
cfg.Pipeline.Archive = &config.ArchiveConfig{ cfg.Pipeline.Publish = &config.ArchiveConfig{
Enabled: boolPtr(true), Enabled: boolPtr(true),
UploadRun: boolPtr(true), UploadRun: boolPtr(true),
PromoteArtifacts: []config.ArchivePromotionRule{ PromoteArtifacts: []config.ArchivePromotionRule{

10
internal/app/remote_locks.go
View File

@@ -84,20 +84,20 @@ func loadEffectiveLocks(ctx context.Context, cfg *config.Config, store storage.O
} }
func staticArchiveLocks(cfg *config.Config) []config.ArchiveLockRule { func staticArchiveLocks(cfg *config.Config) []config.ArchiveLockRule {
if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Archive == nil { if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Publish == nil {
return nil return nil
} }
return append([]config.ArchiveLockRule(nil), cfg.Pipeline.Archive.Locks...) return append([]config.ArchiveLockRule(nil), cfg.Pipeline.Publish.Locks...)
} }
func applyEffectiveLocks(cfg *config.Config, locks []config.ArchiveLockRule) { func applyEffectiveLocks(cfg *config.Config, locks []config.ArchiveLockRule) {
if cfg == nil || cfg.Pipeline == nil { if cfg == nil || cfg.Pipeline == nil {
return return
} }
if cfg.Pipeline.Archive == nil { if cfg.Pipeline.Publish == nil {
cfg.Pipeline.Archive = &config.ArchiveConfig{} cfg.Pipeline.Publish = &config.ArchiveConfig{}
} }
cfg.Pipeline.Archive.Locks = append([]config.ArchiveLockRule(nil), locks...) cfg.Pipeline.Publish.Locks = append([]config.ArchiveLockRule(nil), locks...)
} }
func uploadRemoteLockStore(ctx context.Context, store storage.ObjectStore, key string, lockStore *config.ArchiveLockStore) error { func uploadRemoteLockStore(ctx context.Context, store storage.ObjectStore, key string, lockStore *config.ArchiveLockStore) error {

8
internal/app/resume_run_stage_test.go
View File

@@ -56,7 +56,7 @@ func TestResumeNoRemainingStages(t *testing.T) {
store := &manifest.LocalStore{} store := &manifest.LocalStore{}
m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "archive", "notify"} { for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "publish", "notify"} {
m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil)
} }
if err := store.Save(context.Background(), manifestPath, m); err != nil { if err := store.Save(context.Background(), manifestPath, m); err != nil {
@@ -85,7 +85,7 @@ func TestResumeForceRerunsSucceeded(t *testing.T) {
store := &manifest.LocalStore{} store := &manifest.LocalStore{}
m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "archive", "notify"} { for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "publish", "notify"} {
m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil)
} }
if err := store.Save(context.Background(), manifestPath, m); err != nil { if err := store.Save(context.Background(), manifestPath, m); err != nil {
@@ -176,7 +176,7 @@ func TestRunStageForceMarksDownstreamStaleAndResumeContinuesFromStale(t *testing
store := &manifest.LocalStore{} store := &manifest.LocalStore{}
seed := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) seed := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "archive", "notify"} { for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "publish", "notify"} {
seed.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) seed.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil)
} }
if err := store.Save(context.Background(), manifestPath, seed); err != nil { if err := store.Save(context.Background(), manifestPath, seed); err != nil {
@@ -196,7 +196,7 @@ func TestRunStageForceMarksDownstreamStaleAndResumeContinuesFromStale(t *testing
if err != nil { if err != nil {
t.Fatalf("load manifest after force: %v", err) t.Fatalf("load manifest after force: %v", err)
} }
for _, name := range []string{"normalize", "trim", "analyze", "archive", "notify"} { for _, name := range []string{"normalize", "trim", "analyze", "publish", "notify"} {
if afterForce.Stages[name] == nil || afterForce.Stages[name].Status != manifest.StatusStale { if afterForce.Stages[name] == nil || afterForce.Stages[name].Status != manifest.StatusStale {
t.Fatalf("stage %q = %#v, want stale", name, afterForce.Stages[name]) t.Fatalf("stage %q = %#v, want stale", name, afterForce.Stages[name])
} }

6
internal/app/run_control_test.go
View File

@@ -44,7 +44,7 @@ func TestDecideStageActions(t *testing.T) {
func TestDownstreamStageNames(t *testing.T) { func TestDownstreamStageNames(t *testing.T) {
got := downstreamStageNames("polish") got := downstreamStageNames("polish")
want := []string{"normalize", "trim", "analyze", "archive", "notify"} want := []string{"normalize", "trim", "analyze", "publish", "notify"}
if !reflect.DeepEqual(got, want) { if !reflect.DeepEqual(got, want) {
t.Fatalf("downstreamStageNames(polish) = %#v, want %#v", got, want) t.Fatalf("downstreamStageNames(polish) = %#v, want %#v", got, want)
} }
@@ -65,11 +65,11 @@ func TestInvalidateDownstreamSucceededStages(t *testing.T) {
m.MarkStageSucceeded("normalize", now, nil) m.MarkStageSucceeded("normalize", now, nil)
m.MarkStageSucceeded("trim", now, nil) m.MarkStageSucceeded("trim", now, nil)
m.MarkStageFailed("analyze", now, "analysis failed") m.MarkStageFailed("analyze", now, "analysis failed")
m.MarkStageSucceeded("archive", now, nil) m.MarkStageSucceeded("publish", now, nil)
m.MarkStageSucceeded("notify", now, nil) m.MarkStageSucceeded("notify", now, nil)
got := invalidateDownstreamSucceededStages(m, "polish", now.Add(1*time.Second)) got := invalidateDownstreamSucceededStages(m, "polish", now.Add(1*time.Second))
want := []string{"normalize", "trim", "archive", "notify"} want := []string{"normalize", "trim", "publish", "notify"}
if !reflect.DeepEqual(got, want) { if !reflect.DeepEqual(got, want) {
t.Fatalf("invalidateDownstreamSucceededStages() = %#v, want %#v", got, want) t.Fatalf("invalidateDownstreamSucceededStages() = %#v, want %#v", got, want)
} }

8
internal/app/run_stage.go
View File

@@ -63,8 +63,8 @@ func RunStage(ctx context.Context, args []string, out io.Writer) error {
if err != nil { if err != nil {
return fmt.Errorf("run-stage: invalid --artifacts: %w", err) return fmt.Errorf("run-stage: invalid --artifacts: %w", err)
} }
if len(normalizedArtifacts) > 0 && stageName != "analyze" && stageName != "archive" { if len(normalizedArtifacts) > 0 && stageName != "analyze" && stageName != "publish" {
return fmt.Errorf("run-stage: --artifacts is only supported for stages \"analyze\" and \"archive\"") return fmt.Errorf("run-stage: --artifacts is only supported for stages \"analyze\" and \"publish\"")
} }
summary, err := runSingleStageCommand(ctx, singleStageCommand{ summary, err := runSingleStageCommand(ctx, singleStageCommand{
@@ -164,7 +164,7 @@ func Analyze(ctx context.Context, args []string, out io.Writer) error {
return err return err
} }
// Publish force-runs the archive stage. // Publish force-runs the publish stage.
func Publish(ctx context.Context, args []string, out io.Writer) error { func Publish(ctx context.Context, args []string, out io.Writer) error {
positionalSessionID, args := pullLeadingSessionID(args) positionalSessionID, args := pullLeadingSessionID(args)
fs := flag.NewFlagSet("publish", flag.ContinueOnError) fs := flag.NewFlagSet("publish", flag.ContinueOnError)
@@ -209,7 +209,7 @@ func Publish(ctx context.Context, args []string, out io.Writer) error {
summary, err := runSingleStageCommand(ctx, singleStageCommand{ summary, err := runSingleStageCommand(ctx, singleStageCommand{
CommandName: "publish", CommandName: "publish",
StageName: "archive", StageName: "publish",
PipelinePath: pipelinePath, PipelinePath: pipelinePath,
CampaignPath: campaignPath, CampaignPath: campaignPath,
CampaignFilePath: campaignFilePath, CampaignFilePath: campaignFilePath,

22
internal/app/runner.go
View File

@@ -568,16 +568,16 @@ func needsObjectStoreForRun(cfg *config.Config, stages []stage.Stage) bool {
return true return true
} }
} }
if !stageRequested("archive") { if !stageRequested("publish") {
return false return false
} }
if cfg.Pipeline.Archive == nil { if cfg.Pipeline.Publish == nil {
return false return false
} }
if cfg.Pipeline.Archive.Enabled != nil && !*cfg.Pipeline.Archive.Enabled { if cfg.Pipeline.Publish.Enabled != nil && !*cfg.Pipeline.Publish.Enabled {
return false return false
} }
if cfg.Pipeline.Archive.UploadRun != nil && !*cfg.Pipeline.Archive.UploadRun { if cfg.Pipeline.Publish.UploadRun != nil && !*cfg.Pipeline.Publish.UploadRun {
return false return false
} }
return true return true
@@ -587,23 +587,23 @@ func needsRemoteLocksForRun(cfg *config.Config, stages []stage.Stage) bool {
if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil {
return false return false
} }
archiveRequested := false publishRequested := false
for _, s := range stages { for _, s := range stages {
if s != nil && s.Name() == "archive" { if s != nil && s.Name() == "publish" {
archiveRequested = true publishRequested = true
break break
} }
} }
if !archiveRequested { if !publishRequested {
return false return false
} }
if cfg.Pipeline.Archive == nil { if cfg.Pipeline.Publish == nil {
return false return false
} }
if cfg.Pipeline.Archive.Enabled != nil && !*cfg.Pipeline.Archive.Enabled { if cfg.Pipeline.Publish.Enabled != nil && !*cfg.Pipeline.Publish.Enabled {
return false return false
} }
if cfg.Pipeline.Archive.UploadRun != nil && !*cfg.Pipeline.Archive.UploadRun { if cfg.Pipeline.Publish.UploadRun != nil && !*cfg.Pipeline.Publish.UploadRun {
return false return false
} }
return cfg.Pipeline.Storage.S3 != nil return cfg.Pipeline.Storage.S3 != nil

28
internal/app/runner_test.go
View File

@@ -254,7 +254,7 @@ func TestExecuteStagesArchiveSkipsRequiredUnselectedConfiguredPromotion(t *testi
Bucket: "my-dnd-archive", Bucket: "my-dnd-archive",
RootPrefix: "dnd", RootPrefix: "dnd",
} }
cfg.Pipeline.Archive = &config.ArchiveConfig{ cfg.Pipeline.Publish = &config.ArchiveConfig{
Enabled: boolPtr(true), Enabled: boolPtr(true),
UploadRun: boolPtr(true), UploadRun: boolPtr(true),
PromoteArtifacts: []config.ArchivePromotionRule{ PromoteArtifacts: []config.ArchivePromotionRule{
@@ -283,9 +283,9 @@ func TestExecuteStagesArchiveSkipsRequiredUnselectedConfiguredPromotion(t *testi
t.Fatalf("Save manifest error = %v", err) t.Fatalf("Save manifest error = %v", err)
} }
archiveStageImpl, err := stage.Select("archive") archiveStageImpl, err := stage.Select("publish")
if err != nil { if err != nil {
t.Fatalf("Select(archive) error = %v", err) t.Fatalf("Select(publish) error = %v", err)
} }
summary, err := executeStages( summary, err := executeStages(
@@ -303,7 +303,7 @@ func TestExecuteStagesArchiveSkipsRequiredUnselectedConfiguredPromotion(t *testi
if err != nil { if err != nil {
t.Fatalf("executeStages() error = %v", err) t.Fatalf("executeStages() error = %v", err)
} }
if len(summary.Executed) != 2 || summary.Executed[0] != "analyze" || summary.Executed[1] != "archive" { if len(summary.Executed) != 2 || summary.Executed[0] != "analyze" || summary.Executed[1] != "publish" {
t.Fatalf("executed = %#v, want analyze and archive", summary.Executed) t.Fatalf("executed = %#v, want analyze and archive", summary.Executed)
} }
@@ -311,7 +311,7 @@ func TestExecuteStagesArchiveSkipsRequiredUnselectedConfiguredPromotion(t *testi
if err != nil { if err != nil {
t.Fatalf("Load manifest error = %v", err) t.Fatalf("Load manifest error = %v", err)
} }
meta := loadedManifest.Stages["archive"].Metadata meta := loadedManifest.Stages["publish"].Metadata
skipped, ok := meta["skipped_unselected_promotions"].([]any) skipped, ok := meta["skipped_unselected_promotions"].([]any)
if !ok || len(skipped) != 1 { if !ok || len(skipped) != 1 {
t.Fatalf("skipped_unselected_promotions = %#v, want one item", meta["skipped_unselected_promotions"]) t.Fatalf("skipped_unselected_promotions = %#v, want one item", meta["skipped_unselected_promotions"])
@@ -342,7 +342,7 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) {
t.Fatalf("Load manifest error = %v", err) t.Fatalf("Load manifest error = %v", err)
} }
for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "archive", "notify"} { for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "publish", "notify"} {
sr := m.Stages[name] sr := m.Stages[name]
if sr == nil { if sr == nil {
t.Fatalf("missing stage record %q", name) t.Fatalf("missing stage record %q", name)
@@ -425,9 +425,9 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) {
} }
continue continue
} }
if name == "archive" { if name == "publish" {
if sr.Metadata == nil || sr.Metadata["stage"] != "archive" { if sr.Metadata == nil || sr.Metadata["stage"] != "publish" {
t.Fatalf("archive metadata missing stage=archive: %#v", sr.Metadata) t.Fatalf("archive metadata missing stage=publish: %#v", sr.Metadata)
} }
if sr.Metadata["skipped"] != true { if sr.Metadata["skipped"] != true {
t.Fatalf("archive metadata missing skipped=true for test config without archive section: %#v", sr.Metadata) t.Fatalf("archive metadata missing skipped=true for test config without archive section: %#v", sr.Metadata)
@@ -522,7 +522,7 @@ func TestExecuteStagesForceSuccessInvalidatesDownstreamSucceededStages(t *testin
store := &manifest.LocalStore{} store := &manifest.LocalStore{}
existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC))
for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "archive", "notify"} { for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "publish", "notify"} {
existing.MarkStageSucceeded(stageName, time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil) existing.MarkStageSucceeded(stageName, time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil)
} }
existing.MarkStageFailed("analyze", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), "previous analyze failure") existing.MarkStageFailed("analyze", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), "previous analyze failure")
@@ -553,7 +553,7 @@ func TestExecuteStagesForceSuccessInvalidatesDownstreamSucceededStages(t *testin
if loaded.Stages["polish"] == nil || loaded.Stages["polish"].Status != manifest.StatusSucceeded { if loaded.Stages["polish"] == nil || loaded.Stages["polish"].Status != manifest.StatusSucceeded {
t.Fatalf("polish status = %#v, want succeeded", loaded.Stages["polish"]) t.Fatalf("polish status = %#v, want succeeded", loaded.Stages["polish"])
} }
for _, stageName := range []string{"normalize", "trim", "archive", "notify"} { for _, stageName := range []string{"normalize", "trim", "publish", "notify"} {
if loaded.Stages[stageName] == nil || loaded.Stages[stageName].Status != manifest.StatusStale { if loaded.Stages[stageName] == nil || loaded.Stages[stageName].Status != manifest.StatusStale {
t.Fatalf("%s status = %#v, want stale", stageName, loaded.Stages[stageName]) t.Fatalf("%s status = %#v, want stale", stageName, loaded.Stages[stageName])
} }
@@ -864,7 +864,7 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) {
{name: "merge", env: &Env{Seriatim: &seriatim.FakeRunner{Err: errors.New("merge fail")}}}, {name: "merge", env: &Env{Seriatim: &seriatim.FakeRunner{Err: errors.New("merge fail")}}},
{name: "polish", env: &Env{Audita: &audita.FakeRunner{Err: errors.New("polish fail")}}}, {name: "polish", env: &Env{Audita: &audita.FakeRunner{Err: errors.New("polish fail")}}},
{name: "analyze", env: &Env{Scriptorium: &scriptorium.FakeRunner{RunErr: errors.New("analyze fail")}}}, {name: "analyze", env: &Env{Scriptorium: &scriptorium.FakeRunner{RunErr: errors.New("analyze fail")}}},
{name: "archive", env: &Env{ObjectStore: &storage.FakeBackend{UploadErr: errors.New("archive fail")}}}, {name: "publish", env: &Env{ObjectStore: &storage.FakeBackend{UploadErr: errors.New("archive fail")}}},
{name: "notify", env: &Env{Notifier: &notify.FakeSender{Err: errors.New("notify fail")}}}, {name: "notify", env: &Env{Notifier: &notify.FakeSender{Err: errors.New("notify fail")}}},
} }
@@ -949,8 +949,8 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) {
}, },
} }
} }
if tc.name == "archive" { if tc.name == "publish" {
cfg.Pipeline.Archive = &config.ArchiveConfig{ cfg.Pipeline.Publish = &config.ArchiveConfig{
Enabled: boolPtr(true), Enabled: boolPtr(true),
UploadRun: boolPtr(true), UploadRun: boolPtr(true),
} }

6
internal/app/session_oriented_cli_test.go
View File

@@ -155,13 +155,13 @@ func TestExecuteWorkflowCommandsAcceptPositionalSessionID(t *testing.T) {
{ {
name: "publish", name: "publish",
args: []string{"publish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--artifacts", "session_recap"}, args: []string{"publish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--artifacts", "session_recap"},
wantStage: "archive", wantStage: "publish",
wantForce: true, wantForce: true,
}, },
{ {
name: "run-stage", name: "run-stage",
args: []string{"run-stage", "archive", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--artifacts", "session_recap"}, args: []string{"run-stage", "publish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--artifacts", "session_recap"},
wantStage: "archive", wantStage: "publish",
wantForce: false, wantForce: false,
}, },
} }

8
internal/config/config.go
View File

@@ -21,7 +21,7 @@ type PipelineConfig struct {
Storage StorageConfig `yaml:"storage"` Storage StorageConfig `yaml:"storage"`
Spool SpoolConfig `yaml:"spool"` Spool SpoolConfig `yaml:"spool"`
Cache CacheConfig `yaml:"cache"` Cache CacheConfig `yaml:"cache"`
Archive *ArchiveConfig `yaml:"archive"` Publish *ArchiveConfig `yaml:"publish"`
Secrets *SecretsConfig `yaml:"secrets"` Secrets *SecretsConfig `yaml:"secrets"`
WhisperX WhisperXConfig `yaml:"whisperx"` WhisperX WhisperXConfig `yaml:"whisperx"`
Seriatim SeriatimConfig `yaml:"seriatim"` Seriatim SeriatimConfig `yaml:"seriatim"`
@@ -65,7 +65,7 @@ type SessionConfig struct {
// WorkspaceConfig configures local workspace behavior. // WorkspaceConfig configures local workspace behavior.
type WorkspaceConfig struct { type WorkspaceConfig struct {
Root string `yaml:"root"` Root string `yaml:"root"`
CleanupAfterArchive bool `yaml:"cleanup_after_archive"` CleanupAfterPublish bool `yaml:"cleanup_after_publish"`
} }
// SecretsConfig configures optional local filesystem secret loading. // SecretsConfig configures optional local filesystem secret loading.
@@ -93,7 +93,7 @@ type StorageS3Config struct {
// SpoolConfig configures local spool storage for staged data. // SpoolConfig configures local spool storage for staged data.
type SpoolConfig struct { type SpoolConfig struct {
Root string `yaml:"root"` Root string `yaml:"root"`
DeleteAudioAfterArchive bool `yaml:"delete_audio_after_archive"` DeleteAudioAfterPublish bool `yaml:"delete_audio_after_publish"`
} }
// CacheConfig configures durable local caches for reusable remote inputs. // CacheConfig configures durable local caches for reusable remote inputs.
@@ -106,7 +106,7 @@ type CacheConfig struct {
type ArchiveConfig struct { type ArchiveConfig struct {
Enabled *bool `yaml:"enabled"` Enabled *bool `yaml:"enabled"`
UploadRun *bool `yaml:"upload_run"` UploadRun *bool `yaml:"upload_run"`
PromoteArtifacts []ArchivePromotionRule `yaml:"promote_artifacts"` PromoteArtifacts []ArchivePromotionRule `yaml:"outputs"`
Locks []ArchiveLockRule `yaml:"locks"` Locks []ArchiveLockRule `yaml:"locks"`
} }

2
internal/config/load.go
View File

@@ -328,7 +328,7 @@ func applyPipelineDefaults(cfg *PipelineConfig) {
applyStorageDefaults(&cfg.Storage) applyStorageDefaults(&cfg.Storage)
applySpoolDefaults(&cfg.Spool) applySpoolDefaults(&cfg.Spool)
applyCacheDefaults(&cfg.Cache) applyCacheDefaults(&cfg.Cache)
applyArchiveDefaults(&cfg.Archive) applyArchiveDefaults(&cfg.Publish)
applyWhisperXDefaults(&cfg.WhisperX) applyWhisperXDefaults(&cfg.WhisperX)
applySeriatimDefaults(&cfg.Seriatim) applySeriatimDefaults(&cfg.Seriatim)
applyAuditaDefaults(&cfg.Audita) applyAuditaDefaults(&cfg.Audita)

96
internal/config/storage_archive_test.go
View File

@@ -164,33 +164,33 @@ func TestSpoolAndArchiveDefaults(t *testing.T) {
if cfg.Pipeline.Spool.Root != "/var/spool/narratio" { if cfg.Pipeline.Spool.Root != "/var/spool/narratio" {
t.Fatalf("spool.root = %q, want /var/spool/narratio", cfg.Pipeline.Spool.Root) t.Fatalf("spool.root = %q, want /var/spool/narratio", cfg.Pipeline.Spool.Root)
} }
if cfg.Pipeline.Spool.DeleteAudioAfterArchive { if cfg.Pipeline.Spool.DeleteAudioAfterPublish {
t.Fatalf("spool.delete_audio_after_archive = true, want false") t.Fatalf("spool.delete_audio_after_publish = true, want false")
} }
if cfg.Pipeline.Workspace.CleanupAfterArchive { if cfg.Pipeline.Workspace.CleanupAfterPublish {
t.Fatalf("workspace.cleanup_after_archive = true, want false") t.Fatalf("workspace.cleanup_after_publish = true, want false")
} }
if cfg.Pipeline.Archive == nil { if cfg.Pipeline.Publish == nil {
t.Fatal("archive should be initialized by defaults") t.Fatal("archive should be initialized by defaults")
} }
if cfg.Pipeline.Archive.Enabled == nil || !*cfg.Pipeline.Archive.Enabled { if cfg.Pipeline.Publish.Enabled == nil || !*cfg.Pipeline.Publish.Enabled {
t.Fatalf("archive.enabled = %#v, want true", cfg.Pipeline.Archive.Enabled) t.Fatalf("archive.enabled = %#v, want true", cfg.Pipeline.Publish.Enabled)
} }
if cfg.Pipeline.Archive.UploadRun == nil || !*cfg.Pipeline.Archive.UploadRun { if cfg.Pipeline.Publish.UploadRun == nil || !*cfg.Pipeline.Publish.UploadRun {
t.Fatalf("archive.upload_run = %#v, want true", cfg.Pipeline.Archive.UploadRun) t.Fatalf("archive.upload_run = %#v, want true", cfg.Pipeline.Publish.UploadRun)
} }
if len(cfg.Pipeline.Archive.PromoteArtifacts) != 1 { if len(cfg.Pipeline.Publish.PromoteArtifacts) != 1 {
t.Fatalf("archive.promote_artifacts len = %d, want 1 default", len(cfg.Pipeline.Archive.PromoteArtifacts)) t.Fatalf("publish.outputs len = %d, want 1 default", len(cfg.Pipeline.Publish.PromoteArtifacts))
} }
item := cfg.Pipeline.Archive.PromoteArtifacts[0] item := cfg.Pipeline.Publish.PromoteArtifacts[0]
if item.Required == nil || !*item.Required { if item.Required == nil || !*item.Required {
t.Fatalf("archive.promote_artifacts[0].required = %#v, want true", item.Required) t.Fatalf("publish.outputs[0].required = %#v, want true", item.Required)
} }
if item.Source != "narratio.transcript.final_trimmed" { if item.Source != "narratio.transcript.final_trimmed" {
t.Fatalf("archive.promote_artifacts[0].source = %q, want narratio.transcript.final_trimmed", item.Source) t.Fatalf("publish.outputs[0].source = %q, want narratio.transcript.final_trimmed", item.Source)
} }
if item.Dest != "transcripts/final.trimmed.json" { if item.Dest != "transcripts/final.trimmed.json" {
t.Fatalf("archive.promote_artifacts[0].dest = %q, want transcripts/final.trimmed.json", item.Dest) t.Fatalf("publish.outputs[0].dest = %q, want transcripts/final.trimmed.json", item.Dest)
} }
} }
@@ -202,8 +202,8 @@ func TestArchivePromotionValidation(t *testing.T) {
}{ }{
{ {
name: "absolute dest path rejected", name: "absolute dest path rejected",
ruleYML: `archive: ruleYML: `publish:
promote_artifacts: outputs:
- source: "narratio.transcript.final_trimmed" - source: "narratio.transcript.final_trimmed"
dest: "/transcripts/final.trimmed.json" dest: "/transcripts/final.trimmed.json"
`, `,
@@ -211,8 +211,8 @@ func TestArchivePromotionValidation(t *testing.T) {
}, },
{ {
name: "traversal dest path rejected", name: "traversal dest path rejected",
ruleYML: `archive: ruleYML: `publish:
promote_artifacts: outputs:
- source: "narratio.transcript.final_trimmed" - source: "narratio.transcript.final_trimmed"
dest: "../trimmed.json" dest: "../trimmed.json"
`, `,
@@ -220,8 +220,8 @@ func TestArchivePromotionValidation(t *testing.T) {
}, },
{ {
name: "invalid source rejected", name: "invalid source rejected",
ruleYML: `archive: ruleYML: `publish:
promote_artifacts: outputs:
- source: "narratio.unknown" - source: "narratio.unknown"
dest: "transcripts/final.trimmed.json" dest: "transcripts/final.trimmed.json"
`, `,
@@ -229,8 +229,8 @@ func TestArchivePromotionValidation(t *testing.T) {
}, },
{ {
name: "duplicate destination rejected", name: "duplicate destination rejected",
ruleYML: `archive: ruleYML: `publish:
promote_artifacts: outputs:
- source: "narratio.transcript.final_trimmed" - source: "narratio.transcript.final_trimmed"
dest: "artifacts/shared.md" dest: "artifacts/shared.md"
- source: "narratio.transcript.final" - source: "narratio.transcript.final"
@@ -240,8 +240,8 @@ func TestArchivePromotionValidation(t *testing.T) {
}, },
{ {
name: "configured source requires configured artifact key", name: "configured source requires configured artifact key",
ruleYML: `archive: ruleYML: `publish:
promote_artifacts: outputs:
- source: "narratio.artifact.session_recap" - source: "narratio.artifact.session_recap"
dest: "artifacts/session_recap.md" dest: "artifacts/session_recap.md"
`, `,
@@ -253,8 +253,8 @@ func TestArchivePromotionValidation(t *testing.T) {
artifacts: artifacts:
session_recap: session_recap:
enabled: false enabled: false
archive: publish:
promote_artifacts: outputs:
- source: "narratio.artifact.session_recap" - source: "narratio.artifact.session_recap"
`, `,
wantErr: "destination cannot be derived", wantErr: "destination cannot be derived",
@@ -288,8 +288,8 @@ func TestArchivePromotionLegacyTranscriptSourcesRejected(t *testing.T) {
for _, source := range legacyTranscriptSources { for _, source := range legacyTranscriptSources {
t.Run(source, func(t *testing.T) { t.Run(source, func(t *testing.T) {
pipelineYAML := testPipelineBaseYAML + ` pipelineYAML := testPipelineBaseYAML + `
archive: publish:
promote_artifacts: outputs:
- source: ` + source + ` - source: ` + source + `
dest: transcripts/final.trimmed.json dest: transcripts/final.trimmed.json
` `
@@ -316,8 +316,8 @@ func TestArchivePromotionDerivesDestinationWhenOmitted(t *testing.T) {
{ {
name: "built in source derives canonical destination", name: "built in source derives canonical destination",
pipelineYML: testPipelineBaseYAML + ` pipelineYML: testPipelineBaseYAML + `
archive: publish:
promote_artifacts: outputs:
- source: narratio.transcript.final - source: narratio.transcript.final
`, `,
wantDest: "transcripts/final.json", wantDest: "transcripts/final.json",
@@ -331,8 +331,8 @@ scriptorium:
enabled: true enabled: true
prompt_id: dnd.session_recap prompt_id: dnd.session_recap
output_path: artifacts/session_recap.md output_path: artifacts/session_recap.md
archive: publish:
promote_artifacts: outputs:
- source: narratio.artifact.session_recap - source: narratio.artifact.session_recap
`, `,
wantDest: "artifacts/session_recap.md", wantDest: "artifacts/session_recap.md",
@@ -349,11 +349,11 @@ archive:
if err := Validate(cfg); err != nil { if err := Validate(cfg); err != nil {
t.Fatalf("Validate() error = %v", err) t.Fatalf("Validate() error = %v", err)
} }
if len(cfg.Pipeline.Archive.PromoteArtifacts) != 1 { if len(cfg.Pipeline.Publish.PromoteArtifacts) != 1 {
t.Fatalf("archive.promote_artifacts len = %d, want 1", len(cfg.Pipeline.Archive.PromoteArtifacts)) t.Fatalf("publish.outputs len = %d, want 1", len(cfg.Pipeline.Publish.PromoteArtifacts))
} }
if cfg.Pipeline.Archive.PromoteArtifacts[0].Dest != tt.wantDest { if cfg.Pipeline.Publish.PromoteArtifacts[0].Dest != tt.wantDest {
t.Fatalf("archive.promote_artifacts[0].dest = %q, want %q", cfg.Pipeline.Archive.PromoteArtifacts[0].Dest, tt.wantDest) t.Fatalf("publish.outputs[0].dest = %q, want %q", cfg.Pipeline.Publish.PromoteArtifacts[0].Dest, tt.wantDest)
} }
}) })
} }
@@ -368,7 +368,7 @@ func TestArchiveLockValidation(t *testing.T) {
{ {
name: "valid built in source", name: "valid built in source",
pipelineYML: testPipelineBaseYAML + ` pipelineYML: testPipelineBaseYAML + `
archive: publish:
locks: locks:
- source: narratio.transcript.final_trimmed - source: narratio.transcript.final_trimmed
reason: reviewed transcript reason: reviewed transcript
@@ -383,7 +383,7 @@ scriptorium:
enabled: true enabled: true
prompt_id: dnd.session_recap prompt_id: dnd.session_recap
output_path: artifacts/session_recap.md output_path: artifacts/session_recap.md
archive: publish:
locks: locks:
- source: narratio.artifact.session_recap - source: narratio.artifact.session_recap
`, `,
@@ -391,25 +391,25 @@ archive:
{ {
name: "missing source rejected", name: "missing source rejected",
pipelineYML: testPipelineBaseYAML + ` pipelineYML: testPipelineBaseYAML + `
archive: publish:
locks: locks:
- reason: no source - reason: no source
`, `,
wantErr: "pipeline.archive.locks[0].source is required", wantErr: "pipeline.publish.locks[0].source is required",
}, },
{ {
name: "invalid source rejected", name: "invalid source rejected",
pipelineYML: testPipelineBaseYAML + ` pipelineYML: testPipelineBaseYAML + `
archive: publish:
locks: locks:
- source: narratio.unknown - source: narratio.unknown
`, `,
wantErr: "pipeline.archive.locks[0].source \"narratio.unknown\" is unsupported", wantErr: "pipeline.publish.locks[0].source \"narratio.unknown\" is unsupported",
}, },
{ {
name: "duplicate source rejected", name: "duplicate source rejected",
pipelineYML: testPipelineBaseYAML + ` pipelineYML: testPipelineBaseYAML + `
archive: publish:
locks: locks:
- source: narratio.transcript.final_trimmed - source: narratio.transcript.final_trimmed
- source: " narratio.transcript.final_trimmed " - source: " narratio.transcript.final_trimmed "
@@ -449,7 +449,7 @@ func TestArchiveLockLegacyTranscriptSourcesRejected(t *testing.T) {
for _, source := range legacyTranscriptSources { for _, source := range legacyTranscriptSources {
t.Run(source, func(t *testing.T) { t.Run(source, func(t *testing.T) {
pipelineYAML := testPipelineBaseYAML + ` pipelineYAML := testPipelineBaseYAML + `
archive: publish:
locks: locks:
- source: ` + source + ` - source: ` + source + `
` `
@@ -469,7 +469,7 @@ archive:
func TestArchiveLockUnknownFieldFailsStrictDecode(t *testing.T) { func TestArchiveLockUnknownFieldFailsStrictDecode(t *testing.T) {
pipelineYAML := testPipelineBaseYAML + ` pipelineYAML := testPipelineBaseYAML + `
archive: publish:
locks: locks:
- source: narratio.transcript.final_trimmed - source: narratio.transcript.final_trimmed
dest: transcripts/final.trimmed.json dest: transcripts/final.trimmed.json
@@ -483,8 +483,8 @@ archive:
func TestArchiveLegacyFromToFailsStrictDecode(t *testing.T) { func TestArchiveLegacyFromToFailsStrictDecode(t *testing.T) {
pipelineYAML := testPipelineBaseYAML + ` pipelineYAML := testPipelineBaseYAML + `
archive: publish:
promote_artifacts: outputs:
- from: transcripts/final.trimmed.json - from: transcripts/final.trimmed.json
to: transcripts/final.trimmed.json to: transcripts/final.trimmed.json
` `

18
internal/config/validate.go
View File

@@ -68,7 +68,7 @@ func validatePipeline(cfg *PipelineConfig) error {
if err := validateCache(cfg.Cache); err != nil { if err := validateCache(cfg.Cache); err != nil {
return err return err
} }
if err := validateArchive(cfg.Archive, cfg.Scriptorium); err != nil { if err := validateArchive(cfg.Publish, cfg.Scriptorium); err != nil {
return err return err
} }
if err := validateWhisperX(cfg.WhisperX); err != nil { if err := validateWhisperX(cfg.WhisperX); err != nil {
@@ -135,7 +135,7 @@ func validateArchive(cfg *ArchiveConfig, scriptorium *ScriptoriumConfig) error {
} }
seenDest := map[string]struct{}{} seenDest := map[string]struct{}{}
for i, item := range cfg.PromoteArtifacts { for i, item := range cfg.PromoteArtifacts {
prefix := fmt.Sprintf("pipeline.archive.promote_artifacts[%d]", i) prefix := fmt.Sprintf("pipeline.publish.outputs[%d]", i)
source := strings.TrimSpace(item.Source) source := strings.TrimSpace(item.Source)
if source == "" { if source == "" {
return fmt.Errorf("%s.source is required", prefix) return fmt.Errorf("%s.source is required", prefix)
@@ -161,7 +161,7 @@ func validateArchive(cfg *ArchiveConfig, scriptorium *ScriptoriumConfig) error {
} }
seenDest[normalizedDest] = struct{}{} seenDest[normalizedDest] = struct{}{}
} }
locks, err := ValidateArchiveLockRules(cfg.Locks, scriptorium, "pipeline.archive.locks") locks, err := ValidateArchiveLockRules(cfg.Locks, scriptorium, "pipeline.publish.locks")
if err != nil { if err != nil {
return err return err
} }
@@ -174,7 +174,7 @@ func ValidateArchiveLockRules(locks []ArchiveLockRule, scriptorium *ScriptoriumC
seenLocks := map[string]struct{}{} seenLocks := map[string]struct{}{}
out := make([]ArchiveLockRule, 0, len(locks)) out := make([]ArchiveLockRule, 0, len(locks))
if strings.TrimSpace(label) == "" { if strings.TrimSpace(label) == "" {
label = "archive.locks" label = "publish.locks"
} }
for i, item := range locks { for i, item := range locks {
prefix := fmt.Sprintf("%s[%d]", label, i) prefix := fmt.Sprintf("%s[%d]", label, i)
@@ -637,19 +637,19 @@ func validateCrossConfig(pipeline *PipelineConfig, session *SessionConfig) error
} }
func archiveUploadConfiguredForS3(pipeline *PipelineConfig) bool { func archiveUploadConfiguredForS3(pipeline *PipelineConfig) bool {
if pipeline == nil || pipeline.Archive == nil { if pipeline == nil || pipeline.Publish == nil {
return false return false
} }
if !strings.EqualFold(strings.TrimSpace(pipeline.Storage.Backend), "s3") { if !strings.EqualFold(strings.TrimSpace(pipeline.Storage.Backend), "s3") {
return false return false
} }
enabled := true enabled := true
if pipeline.Archive.Enabled != nil { if pipeline.Publish.Enabled != nil {
enabled = *pipeline.Archive.Enabled enabled = *pipeline.Publish.Enabled
} }
upload := true upload := true
if pipeline.Archive.UploadRun != nil { if pipeline.Publish.UploadRun != nil {
upload = *pipeline.Archive.UploadRun upload = *pipeline.Publish.UploadRun
} }
return enabled && upload return enabled && upload
} }

66
internal/stage/archive.go
View File

@@ -34,7 +34,7 @@ var archivePrerequisiteStages = []string{
"analyze", "analyze",
} }
func (archiveStage) Name() string { return "archive" } func (archiveStage) Name() string { return "publish" }
func (archiveStage) Declares() IODecl { func (archiveStage) Declares() IODecl {
return IODecl{ return IODecl{
@@ -46,13 +46,13 @@ func (archiveStage) Declares() IODecl {
func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) { 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 { 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") return nil, fmt.Errorf("publish: resolved config must include pipeline and session")
} }
if archiveDisabled(env) { if archiveDisabled(env) {
return &StageResult{ return &StageResult{
Metadata: map[string]any{ Metadata: map[string]any{
"stage": "archive", "stage": "publish",
"skipped": true, "skipped": true,
"archive_enabled": false, "archive_enabled": false,
"audio_upload_skipped": true, "audio_upload_skipped": true,
@@ -63,7 +63,7 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if archiveRunUploadDisabled(env) { if archiveRunUploadDisabled(env) {
return &StageResult{ return &StageResult{
Metadata: map[string]any{ Metadata: map[string]any{
"stage": "archive", "stage": "publish",
"skipped": true, "skipped": true,
"upload_run_enabled": false, "upload_run_enabled": false,
"audio_upload_skipped": true, "audio_upload_skipped": true,
@@ -73,76 +73,76 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
} }
if err := validateArchivePrerequisites(m); err != nil { if err := validateArchivePrerequisites(m); err != nil {
return nil, fmt.Errorf("archive: %w", err) return nil, fmt.Errorf("publish: %w", err)
} }
if env.ObjectStore == nil { if env.ObjectStore == nil {
return nil, fmt.Errorf("archive: remote object store backend is required when archive run upload is enabled") return nil, fmt.Errorf("publish: remote object store backend is required when publish run upload is enabled")
} }
runRoot, err := resolveArchiveRunRoot(env, m) runRoot, err := resolveArchiveRunRoot(env, m)
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: resolve run root: %w", err) return nil, fmt.Errorf("publish: resolve run root: %w", err)
} }
runRootInfo, err := os.Stat(runRoot) runRootInfo, err := os.Stat(runRoot)
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: run root %q: %w", runRoot, err) return nil, fmt.Errorf("publish: run root %q: %w", runRoot, err)
} }
if !runRootInfo.IsDir() { if !runRootInfo.IsDir() {
return nil, fmt.Errorf("archive: run root %q is not a directory", runRoot) return nil, fmt.Errorf("publish: run root %q is not a directory", runRoot)
} }
runPrefix, err := artifacts.ResolveArchiveRunPrefix(env.Config, m) runPrefix, err := artifacts.ResolveArchiveRunPrefix(env.Config, m)
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: resolve s3 run prefix: %w", err) return nil, fmt.Errorf("publish: resolve s3 run prefix: %w", err)
} }
sessionPrefix, err := artifacts.ResolveArchiveSessionPrefix(env.Config, m) sessionPrefix, err := artifacts.ResolveArchiveSessionPrefix(env.Config, m)
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: resolve s3 session prefix: %w", err) return nil, fmt.Errorf("publish: resolve s3 session prefix: %w", err)
} }
bucket := artifacts.ResolveArchiveBucket(env.Config, m) bucket := artifacts.ResolveArchiveBucket(env.Config, m)
if bucket == "" { if bucket == "" {
return nil, fmt.Errorf("archive: resolve s3 bucket: bucket is required") return nil, fmt.Errorf("publish: resolve s3 bucket: bucket is required")
} }
runID := strings.TrimSpace(m.RunID) runID := strings.TrimSpace(m.RunID)
if runID == "" { if runID == "" {
return nil, fmt.Errorf("archive: run id is required") return nil, fmt.Errorf("publish: run id is required")
} }
manifestSource, err := resolveArchiveRunManifestSource(runRoot) manifestSource, err := resolveArchiveRunManifestSource(runRoot)
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: resolve run manifest source: %w", err) return nil, fmt.Errorf("publish: resolve run manifest source: %w", err)
} }
runFiles, err := collectArchiveRunFiles(runRoot, manifestSource) runFiles, err := collectArchiveRunFiles(runRoot, manifestSource)
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: collect run files: %w", err) return nil, fmt.Errorf("publish: collect run files: %w", err)
} }
sessionPaths := archiveSessionPaths(env, m) sessionPaths := archiveSessionPaths(env, m)
previousFiles, err := collectArchivePreviousFiles(sessionPaths.PreviousDir) previousFiles, err := collectArchivePreviousFiles(sessionPaths.PreviousDir)
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: collect previous files: %w", err) return nil, fmt.Errorf("publish: collect previous files: %w", err)
} }
runtimeCatalog, err := buildArchiveRuntimeArtifactCatalog(sessionPaths, env.Config.Pipeline.Scriptorium) runtimeCatalog, err := buildArchiveRuntimeArtifactCatalog(sessionPaths, env.Config.Pipeline.Scriptorium)
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: build runtime artifact catalog: %w", err) return nil, fmt.Errorf("publish: build runtime artifact catalog: %w", err)
} }
promotions, skippedOptional, skippedUnselected, lockedPromotions, err := resolveArchivePromotions( promotions, skippedOptional, skippedUnselected, lockedPromotions, err := resolveArchivePromotions(
sessionPaths, sessionPaths,
m, m,
runtimeCatalog, runtimeCatalog,
env.Config.Pipeline.Archive.PromoteArtifacts, env.Config.Pipeline.Publish.PromoteArtifacts,
env.Config.Pipeline.Archive.Locks, env.Config.Pipeline.Publish.Locks,
env.SelectedArtifactKeys, env.SelectedArtifactKeys,
sessionPrefix, sessionPrefix,
) )
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err) return nil, fmt.Errorf("publish: resolve promotion rules: %w", err)
} }
runUploaded := make([]string, 0, len(runFiles)) runUploaded := make([]string, 0, len(runFiles))
for _, file := range runFiles { for _, file := range runFiles {
key := artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath) key := artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath)
if _, err := env.ObjectStore.Upload(ctx, file.LocalPath, key, storage.UploadOptions{}); err != nil { 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) return nil, fmt.Errorf("publish: upload run file %q to %q: %w", file.RelativePath, key, err)
} }
runUploaded = append(runUploaded, file.RelativePath) runUploaded = append(runUploaded, file.RelativePath)
} }
@@ -151,7 +151,7 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
for _, promotion := range promotions { for _, promotion := range promotions {
key := artifacts.S3PromotedArtifactKey(sessionPrefix, promotion.Dest) key := artifacts.S3PromotedArtifactKey(sessionPrefix, promotion.Dest)
if _, err := env.ObjectStore.Upload(ctx, promotion.LocalPath, key, storage.UploadOptions{}); err != nil { 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) return nil, fmt.Errorf("publish: upload promoted output source %q to %q: %w", promotion.Source, key, err)
} }
promotedUploaded = append(promotedUploaded, promotion.Dest) promotedUploaded = append(promotedUploaded, promotion.Dest)
} }
@@ -160,7 +160,7 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
for _, file := range previousFiles { for _, file := range previousFiles {
key := artifacts.S3PromotedArtifactKey(sessionPrefix, file.RelativePath) key := artifacts.S3PromotedArtifactKey(sessionPrefix, file.RelativePath)
if _, err := env.ObjectStore.Upload(ctx, file.LocalPath, key, storage.UploadOptions{}); err != nil { 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) return nil, fmt.Errorf("publish: upload previous file %q to %q: %w", file.RelativePath, key, err)
} }
previousUploaded = append(previousUploaded, file.RelativePath) previousUploaded = append(previousUploaded, file.RelativePath)
} }
@@ -179,31 +179,31 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
currentManifestKey, currentManifestKey,
)) ))
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: build current manifest snapshot: %w", err) return nil, fmt.Errorf("publish: build current manifest snapshot: %w", err)
} }
defer func() { _ = os.Remove(manifestTempPath) }() defer func() { _ = os.Remove(manifestTempPath) }()
if _, err := env.ObjectStore.Upload(ctx, manifestTempPath, currentManifestKey, storage.UploadOptions{ if _, err := env.ObjectStore.Upload(ctx, manifestTempPath, currentManifestKey, storage.UploadOptions{
ContentType: "application/json", ContentType: "application/json",
}); err != nil { }); err != nil {
return nil, fmt.Errorf("archive: upload current manifest to %q: %w", currentManifestKey, err) return nil, fmt.Errorf("publish: upload current manifest to %q: %w", currentManifestKey, err)
} }
runIDTempPath, err := writeCurrentRunIDPointer(runID) runIDTempPath, err := writeCurrentRunIDPointer(runID)
if err != nil { if err != nil {
return nil, fmt.Errorf("archive: build current run id pointer: %w", err) return nil, fmt.Errorf("publish: build current run id pointer: %w", err)
} }
defer func() { _ = os.Remove(runIDTempPath) }() defer func() { _ = os.Remove(runIDTempPath) }()
if _, err := env.ObjectStore.Upload(ctx, runIDTempPath, currentRunPointerKey, storage.UploadOptions{ if _, err := env.ObjectStore.Upload(ctx, runIDTempPath, currentRunPointerKey, storage.UploadOptions{
ContentType: "text/plain; charset=utf-8", ContentType: "text/plain; charset=utf-8",
}); err != nil { }); err != nil {
return nil, fmt.Errorf("archive: upload current run pointer to %q: %w", currentRunPointerKey, err) return nil, fmt.Errorf("publish: upload current run pointer to %q: %w", currentRunPointerKey, err)
} }
return &StageResult{ return &StageResult{
Metadata: map[string]any{ Metadata: map[string]any{
"stage": "archive", "stage": "publish",
"uploaded": true, "uploaded": true,
"s3_bucket": bucket, "s3_bucket": bucket,
"s3_run_prefix": runPrefix, "s3_run_prefix": runPrefix,
@@ -250,7 +250,7 @@ type archiveSkippedUnselectedPromotion struct {
} }
func archiveDisabled(env *Env) bool { func archiveDisabled(env *Env) bool {
cfg := env.Config.Pipeline.Archive cfg := env.Config.Pipeline.Publish
if cfg == nil { if cfg == nil {
return true return true
} }
@@ -258,7 +258,7 @@ func archiveDisabled(env *Env) bool {
} }
func archiveRunUploadDisabled(env *Env) bool { func archiveRunUploadDisabled(env *Env) bool {
cfg := env.Config.Pipeline.Archive cfg := env.Config.Pipeline.Publish
if cfg == nil { if cfg == nil {
return true return true
} }
@@ -715,8 +715,8 @@ func writeCurrentManifestSnapshot(m *manifest.Manifest, archiveMetadata map[stri
} }
now := time.Now().UTC() now := time.Now().UTC()
clone.MarkStageSucceeded("archive", now, nil) clone.MarkStageSucceeded("publish", now, nil)
if sr := clone.Stages["archive"]; sr != nil { if sr := clone.Stages["publish"]; sr != nil {
sr.Metadata = archiveMetadata sr.Metadata = archiveMetadata
} }
@@ -768,7 +768,7 @@ func archiveMetadataPreview(
currentManifestKey string, currentManifestKey string,
) map[string]any { ) map[string]any {
return map[string]any{ return map[string]any{
"stage": "archive", "stage": "publish",
"uploaded": true, "uploaded": true,
"s3_bucket": bucket, "s3_bucket": bucket,
"s3_run_prefix": runPrefix, "s3_run_prefix": runPrefix,

26
internal/stage/archive_test.go
View File

@@ -19,7 +19,7 @@ import (
func TestArchiveSkipsWhenDisabled(t *testing.T) { func TestArchiveSkipsWhenDisabled(t *testing.T) {
env, m, _ := archiveFixture(t) env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.Enabled = boolPtr(false) env.Config.Pipeline.Publish.Enabled = boolPtr(false)
result, err := archiveStage{}.Run(context.Background(), env, m) result, err := archiveStage{}.Run(context.Background(), env, m)
if err != nil { if err != nil {
@@ -35,7 +35,7 @@ func TestArchiveSkipsWhenDisabled(t *testing.T) {
func TestArchiveSkipsRunUploadWhenDisabled(t *testing.T) { func TestArchiveSkipsRunUploadWhenDisabled(t *testing.T) {
env, m, _ := archiveFixture(t) env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.UploadRun = boolPtr(false) env.Config.Pipeline.Publish.UploadRun = boolPtr(false)
result, err := archiveStage{}.Run(context.Background(), env, m) result, err := archiveStage{}.Run(context.Background(), env, m)
if err != nil { if err != nil {
@@ -179,7 +179,7 @@ func TestArchiveToleratesMissingPreviousCache(t *testing.T) {
func TestArchiveUsesCustomPromotionRules(t *testing.T) { func TestArchiveUsesCustomPromotionRules(t *testing.T) {
env, m, _ := archiveFixture(t) env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ env.Config.Pipeline.Publish.PromoteArtifacts = []config.ArchivePromotionRule{
{Source: "narratio.transcript.final_trimmed", Dest: "published/trimmed.json", Required: boolPtr(true)}, {Source: "narratio.transcript.final_trimmed", Dest: "published/trimmed.json", Required: boolPtr(true)},
{Source: "narratio.artifact.session_recap", Dest: "published/recap.md", Required: boolPtr(true)}, {Source: "narratio.artifact.session_recap", Dest: "published/recap.md", Required: boolPtr(true)},
} }
@@ -200,7 +200,7 @@ func TestArchiveUsesCustomPromotionRules(t *testing.T) {
func TestArchiveSkipsOptionalMissingPromotion(t *testing.T) { func TestArchiveSkipsOptionalMissingPromotion(t *testing.T) {
env, m, _ := archiveFixture(t) env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ env.Config.Pipeline.Publish.PromoteArtifacts = []config.ArchivePromotionRule{
{Source: "narratio.transcript.final_trimmed", Dest: "transcripts/final.trimmed.json", Required: boolPtr(true)}, {Source: "narratio.transcript.final_trimmed", Dest: "transcripts/final.trimmed.json", Required: boolPtr(true)},
{Source: "narratio.transcript.base", Dest: "transcripts/base.json", Required: boolPtr(false)}, {Source: "narratio.transcript.base", Dest: "transcripts/base.json", Required: boolPtr(false)},
} }
@@ -272,7 +272,7 @@ func TestArchiveSelectedConfiguredPromotionStillFailsWhenMissing(t *testing.T) {
func TestArchiveLockedSelectedPromotionSkipsAsLocked(t *testing.T) { func TestArchiveLockedSelectedPromotionSkipsAsLocked(t *testing.T) {
env, m, _ := archiveFixture(t) env, m, _ := archiveFixture(t)
env.SelectedArtifactKeys = []string{"session_recap"} env.SelectedArtifactKeys = []string{"session_recap"}
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{ env.Config.Pipeline.Publish.Locks = []config.ArchiveLockRule{
{Source: "narratio.artifact.session_recap", Reason: "reviewed"}, {Source: "narratio.artifact.session_recap", Reason: "reviewed"},
} }
fake := env.ObjectStore.(*storage.FakeBackend) fake := env.ObjectStore.(*storage.FakeBackend)
@@ -301,7 +301,7 @@ func TestArchiveLockedUnselectedConfiguredPromotionSkipsAsUnselected(t *testing.
PromptID: "dnd.player_handout", PromptID: "dnd.player_handout",
OutputPath: "artifacts/player_handout.md", OutputPath: "artifacts/player_handout.md",
} }
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{ env.Config.Pipeline.Publish.Locks = []config.ArchiveLockRule{
{Source: "narratio.artifact.session_recap", Reason: "reviewed"}, {Source: "narratio.artifact.session_recap", Reason: "reviewed"},
} }
@@ -320,7 +320,7 @@ func TestArchiveLockedUnselectedConfiguredPromotionSkipsAsUnselected(t *testing.
func TestArchiveSkipsLockedRequiredPromotionAndCommits(t *testing.T) { func TestArchiveSkipsLockedRequiredPromotionAndCommits(t *testing.T) {
env, m, _ := archiveFixture(t) env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{ env.Config.Pipeline.Publish.Locks = []config.ArchiveLockRule{
{Source: "narratio.transcript.final_trimmed", Reason: "human reviewed"}, {Source: "narratio.transcript.final_trimmed", Reason: "human reviewed"},
} }
fake := env.ObjectStore.(*storage.FakeBackend) fake := env.ObjectStore.(*storage.FakeBackend)
@@ -374,7 +374,7 @@ func TestArchiveSkipsLockedRequiredPromotionAndCommits(t *testing.T) {
t.Fatalf("unmarshal current manifest: %v", err) t.Fatalf("unmarshal current manifest: %v", err)
} }
stages := current["stages"].(map[string]any) stages := current["stages"].(map[string]any)
archive := stages["archive"].(map[string]any) archive := stages["publish"].(map[string]any)
meta := archive["metadata"].(map[string]any) meta := archive["metadata"].(map[string]any)
if meta["locked_promotion_count"] != float64(1) { if meta["locked_promotion_count"] != float64(1) {
t.Fatalf("current manifest locked_promotion_count = %#v, want 1", meta["locked_promotion_count"]) t.Fatalf("current manifest locked_promotion_count = %#v, want 1", meta["locked_promotion_count"])
@@ -387,10 +387,10 @@ func TestArchiveSkipsLockedRequiredPromotionAndCommits(t *testing.T) {
func TestArchiveLockedRequiredMissingPromotionSucceeds(t *testing.T) { func TestArchiveLockedRequiredMissingPromotionSucceeds(t *testing.T) {
env, m, _ := archiveFixture(t) env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ env.Config.Pipeline.Publish.PromoteArtifacts = []config.ArchivePromotionRule{
{Source: "narratio.transcript.base", Dest: "transcripts/base.json", Required: boolPtr(true)}, {Source: "narratio.transcript.base", Dest: "transcripts/base.json", Required: boolPtr(true)},
} }
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{ env.Config.Pipeline.Publish.Locks = []config.ArchiveLockRule{
{Source: "narratio.transcript.base", Reason: "manual merge is locked"}, {Source: "narratio.transcript.base", Reason: "manual merge is locked"},
} }
@@ -418,7 +418,7 @@ func TestArchiveLockedRequiredMissingPromotionSucceeds(t *testing.T) {
func TestArchiveLockDoesNotOverwriteExistingPromotion(t *testing.T) { func TestArchiveLockDoesNotOverwriteExistingPromotion(t *testing.T) {
env, m, _ := archiveFixture(t) env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{ env.Config.Pipeline.Publish.Locks = []config.ArchiveLockRule{
{Source: "narratio.transcript.final_trimmed", Reason: "already published"}, {Source: "narratio.transcript.final_trimmed", Reason: "already published"},
} }
fake := env.ObjectStore.(*storage.FakeBackend) fake := env.ObjectStore.(*storage.FakeBackend)
@@ -442,7 +442,7 @@ func TestArchiveLockDoesNotOverwriteExistingPromotion(t *testing.T) {
func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) { func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) {
env, m, _ := archiveFixture(t) env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{ env.Config.Pipeline.Publish.PromoteArtifacts = []config.ArchivePromotionRule{
{Source: "narratio.transcript.base", Dest: "transcripts/base.json", Required: boolPtr(true)}, {Source: "narratio.transcript.base", Dest: "transcripts/base.json", Required: boolPtr(true)},
} }
@@ -563,7 +563,7 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) {
RootPrefix: "dnd", RootPrefix: "dnd",
}, },
}, },
Archive: &config.ArchiveConfig{ Publish: &config.ArchiveConfig{
Enabled: boolPtr(true), Enabled: boolPtr(true),
UploadRun: boolPtr(true), UploadRun: boolPtr(true),
PromoteArtifacts: []config.ArchivePromotionRule{ PromoteArtifacts: []config.ArchivePromotionRule{

8
internal/stage/placeholders_test.go
View File

@@ -61,7 +61,7 @@ func TestStagesReturnExpectedMetadata(t *testing.T) {
RootPrefix: "dnd", RootPrefix: "dnd",
}, },
}, },
Archive: &config.ArchiveConfig{ Publish: &config.ArchiveConfig{
Enabled: boolPtr(true), Enabled: boolPtr(true),
UploadRun: boolPtr(true), UploadRun: boolPtr(true),
}, },
@@ -195,9 +195,9 @@ func TestStagesReturnExpectedMetadata(t *testing.T) {
} }
continue continue
} }
if s.Name() == "archive" { if s.Name() == "publish" {
if result.Metadata["stage"] != "archive" { if result.Metadata["stage"] != "publish" {
t.Fatalf("archive metadata = %#v, want stage=archive", result.Metadata) t.Fatalf("archive metadata = %#v, want stage=publish", result.Metadata)
} }
if result.Metadata["uploaded"] != true { if result.Metadata["uploaded"] != true {
t.Fatalf("archive metadata = %#v, want uploaded=true", result.Metadata) t.Fatalf("archive metadata = %#v, want uploaded=true", result.Metadata)