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

Use artifact source IDs for archive promotion

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-19 20:05:24 -05:00
parent c5c35cd3b4
commit d001baa660
14 changed files with 432 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
internal/stage/archive.go
View File

@@ -3,6 +3,7 @@ package stage
import (
"context"
"encoding/json"
"errors"
"fmt"
"io/fs"
"os"
@@ -116,11 +117,12 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if err != nil {
return nil, fmt.Errorf("archive: collect run files: %w", err)
}
sessionRoot, err := resolveArchiveSessionRoot(env, m)
sessionPaths := archiveSessionPaths(env, m)
runtimeCatalog, err := buildArchiveRuntimeArtifactCatalog(sessionPaths, env.Config.Pipeline.Scriptorium)
if err != nil {
return nil, fmt.Errorf("archive: resolve session root for promotions: %w", err)
return nil, fmt.Errorf("archive: build runtime artifact catalog: %w", err)
}
promotions, err := resolveArchivePromotions(sessionRoot, env.Config.Pipeline.Archive.PromoteArtifacts)
promotions, skippedOptional, err := resolveArchivePromotions(sessionPaths, m, runtimeCatalog, env.Config.Pipeline.Archive.PromoteArtifacts)
if err != nil {
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
}
@@ -134,20 +136,12 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
}
promotedUploaded := make([]string, 0, len(promotions))
skippedOptional := make([]string, 0)
for _, promotion := range promotions {
if !promotion.Exists {
if promotion.Required {
return nil, fmt.Errorf("archive: required promotion source missing: %q", promotion.From)
}
skippedOptional = append(skippedOptional, promotion.To)
continue
}
key := artifacts.S3PromotedArtifactKey(sessionPrefix, promotion.To)
key := artifacts.S3PromotedArtifactKey(sessionPrefix, promotion.Dest)
if _, err := env.ObjectStore.Upload(ctx, promotion.LocalPath, key, storage.UploadOptions{}); err != nil {
return nil, fmt.Errorf("archive: upload promoted output %q to %q: %w", promotion.From, key, err)
return nil, fmt.Errorf("archive: upload promoted output source %q to %q: %w", promotion.Source, key, err)
}
promotedUploaded = append(promotedUploaded, promotion.To)
promotedUploaded = append(promotedUploaded, promotion.Dest)
}
currentManifestKey := artifacts.S3CurrentManifestKey(sessionPrefix)
@@ -204,11 +198,11 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
}
type archivePromotion struct {
From string
To string
Required bool
LocalPath string
Exists bool
Source string
Dest string
Required bool
LocalPath string
Provenance string
}
func archiveDisabled(env *Env) bool {
@@ -292,6 +286,19 @@ func resolveArchiveSessionRoot(env *Env, m *manifest.Manifest) (string, error) {
return filepath.Clean(artifacts.SessionWorkDirForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID)), nil
}
func archiveSessionPaths(env *Env, m *manifest.Manifest) artifacts.SessionPaths {
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
if sessionID == "" && m != nil {
sessionID = strings.TrimSpace(m.SessionID)
}
campaign := strings.TrimSpace(env.Config.Session.Campaign)
if campaign == "" && m != nil {
campaign = strings.TrimSpace(m.Campaign)
}
store := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root)
return store.SessionPathsFor(campaign, sessionID)
}
func archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) {
runPrefix := strings.TrimSpace(m.S3RunPrefix)
if runPrefix != "" {
@@ -342,55 +349,138 @@ func archiveBucket(env *Env, m *manifest.Manifest) string {
return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket)
}
func resolveArchivePromotions(sessionRoot string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) {
sessionRoot = filepath.Clean(strings.TrimSpace(sessionRoot))
if sessionRoot == "" {
return nil, fmt.Errorf("session root is required")
}
func resolveArchivePromotions(
paths artifacts.SessionPaths,
m *manifest.Manifest,
catalog *artifacts.ArtifactCatalog,
rules []config.ArchivePromotionRule,
) ([]archivePromotion, []string, error) {
out := make([]archivePromotion, 0, len(rules))
skippedOptional := make([]string, 0)
for _, rule := range rules {
from := strings.TrimSpace(rule.From)
to := strings.TrimSpace(rule.To)
source := strings.TrimSpace(rule.Source)
required := rule.Required == nil || *rule.Required
resolvedPath, err := resolveWorkDirRelativePath(sessionRoot, from)
dest, err := resolveArchivePromotionDest(rule, catalog)
if err != nil {
return nil, fmt.Errorf("promotion from %q: %w", from, err)
return nil, nil, fmt.Errorf("source %q: %w", source, err)
}
info, err := os.Stat(resolvedPath)
exists := err == nil && !info.IsDir()
if err != nil && !os.IsNotExist(err) {
return nil, fmt.Errorf("promotion source %q: %w", from, err)
resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, catalog)
if err != nil {
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) && !required {
skippedOptional = append(skippedOptional, dest)
continue
}
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) {
return nil, nil, fmt.Errorf("required promotion source unavailable: %q", source)
}
return nil, nil, fmt.Errorf("resolve source %q: %w", source, err)
}
localPath := resolvedPath
out = append(out, archivePromotion{
From: from,
To: to,
Required: required,
LocalPath: localPath,
Exists: exists,
Source: source,
Dest: dest,
Required: required,
LocalPath: resolved.Path,
Provenance: resolved.Provenance,
})
}
return out, nil
return out, skippedOptional, nil
}
func resolveWorkDirRelativePath(workDir, rel string) (string, error) {
rel = filepath.Clean(filepath.FromSlash(strings.TrimSpace(rel)))
if rel == "." || rel == "" {
func resolveArchivePromotionDest(rule config.ArchivePromotionRule, catalog *artifacts.ArtifactCatalog) (string, error) {
dest := strings.TrimSpace(rule.Dest)
if dest == "" {
entry, ok := catalog.Lookup(strings.TrimSpace(rule.Source))
if !ok {
return "", fmt.Errorf("destination omitted and source is unknown")
}
dest = strings.TrimSpace(entry.CanonicalRelPath)
if dest == "" {
return "", fmt.Errorf("destination omitted and no canonical destination is available")
}
}
return normalizeArchiveRelativePath(dest)
}
func normalizeArchiveRelativePath(rel string) (string, error) {
trimmed := strings.TrimSpace(rel)
if trimmed == "" {
return "", fmt.Errorf("relative path is required")
}
full := filepath.Join(workDir, rel)
cleanedWork := filepath.Clean(workDir)
cleanedFull := filepath.Clean(full)
relative, err := filepath.Rel(cleanedWork, cleanedFull)
if err != nil {
return "", fmt.Errorf("compute relative path: %w", err)
cleaned := filepath.ToSlash(filepath.Clean(filepath.FromSlash(trimmed)))
if cleaned == "." || cleaned == "" {
return "", fmt.Errorf("relative path is required")
}
if relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
return "", fmt.Errorf("path escapes workdir")
if filepath.IsAbs(trimmed) || strings.HasPrefix(cleaned, "/") || cleaned == ".." || strings.HasPrefix(cleaned, "../") {
return "", fmt.Errorf("path must be a clean relative path")
}
return cleanedFull, nil
return cleaned, nil
}
func buildArchiveRuntimeArtifactCatalog(
paths artifacts.SessionPaths,
scriptoriumCfg *config.ScriptoriumConfig,
) (*artifacts.ArtifactCatalog, error) {
catalog := artifacts.NewArtifactCatalog()
if err := catalog.RegisterBuiltIns(); err != nil {
return nil, err
}
if scriptoriumCfg == nil {
return catalog, nil
}
configured := map[string]artifacts.ConfiguredArtifactDefinition{}
for key, artifactCfg := range scriptoriumCfg.Artifacts {
configured[key] = artifacts.ConfiguredArtifactDefinition{
Enabled: artifactCfg.Enabled,
OutputPath: artifactCfg.OutputPath,
}
}
if err := catalog.RegisterConfiguredArtifacts(configured, nil); err != nil {
return nil, err
}
for _, entry := range catalog.ListConfigured() {
if strings.TrimSpace(entry.CanonicalRelPath) == "" {
continue
}
localPath, err := resolveConfiguredArtifactLocalPath(paths, entry.CanonicalRelPath)
if err != nil {
continue
}
info, statErr := os.Stat(localPath)
if statErr != nil {
if os.IsNotExist(statErr) {
continue
}
return nil, fmt.Errorf("stat configured artifact %q: %w", entry.SourceID, statErr)
}
if info.IsDir() {
continue
}
if err := catalog.MarkAvailableFromDisk(entry.SourceID, localPath); err != nil {
return nil, err
}
}
return catalog, nil
}
func resolveConfiguredArtifactLocalPath(paths artifacts.SessionPaths, configured string) (string, error) {
outputPath := strings.TrimSpace(configured)
if outputPath == "" {
return "", fmt.Errorf("configured artifact output path is required")
}
if filepath.IsAbs(outputPath) {
return filepath.Clean(outputPath), nil
}
rel := filepath.Clean(outputPath)
if rel == "." || rel == "" {
return "", fmt.Errorf("relative output path is required")
}
if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
return "", fmt.Errorf("relative output path escapes session root: %q", configured)
}
return filepath.Join(paths.Root, rel), nil
}
func collectArchiveRunFiles(runRoot, manifestPath string) ([]archiveUploadFile, error) {

33
internal/stage/archive_test.go
View File

@@ -135,8 +135,8 @@ func TestArchiveUploadsRunRecordPromotionsAndCurrentPointer(t *testing.T) {
func TestArchiveUsesCustomPromotionRules(t *testing.T) {
env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{
{From: "transcripts/trimmed.json", To: "published/trimmed.json", Required: boolPtr(true)},
{From: "artifacts/session_recap.md", To: "published/recap.md", Required: boolPtr(true)},
{Source: "narratio.transcript.trimmed", Dest: "published/trimmed.json", Required: boolPtr(true)},
{Source: "narratio.artifact.session_recap", Dest: "published/recap.md", Required: boolPtr(true)},
}
_, err := archiveStage{}.Run(context.Background(), env, m)
@@ -156,8 +156,8 @@ func TestArchiveUsesCustomPromotionRules(t *testing.T) {
func TestArchiveSkipsOptionalMissingPromotion(t *testing.T) {
env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{
{From: "transcripts/trimmed.json", To: "transcripts/trimmed.json", Required: boolPtr(true)},
{From: "artifacts/optional.md", To: "artifacts/optional.md", Required: boolPtr(false)},
{Source: "narratio.transcript.trimmed", Dest: "transcripts/trimmed.json", Required: boolPtr(true)},
{Source: "narratio.transcript.merged", Dest: "transcripts/merged.json", Required: boolPtr(false)},
}
result, err := archiveStage{}.Run(context.Background(), env, m)
@@ -165,7 +165,7 @@ func TestArchiveSkipsOptionalMissingPromotion(t *testing.T) {
t.Fatalf("Run() error = %v", err)
}
got, _ := result.Metadata["skipped_optional_promotions"].([]string)
want := []string{"artifacts/optional.md"}
want := []string{"transcripts/merged.json"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("skipped_optional_promotions = %#v, want %#v", got, want)
}
@@ -174,12 +174,12 @@ func TestArchiveSkipsOptionalMissingPromotion(t *testing.T) {
func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) {
env, m, _ := archiveFixture(t)
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{
{From: "artifacts/missing.md", To: "artifacts/missing.md", Required: boolPtr(true)},
{Source: "narratio.transcript.merged", Dest: "transcripts/merged.json", Required: boolPtr(true)},
}
_, err := archiveStage{}.Run(context.Background(), env, m)
if err == nil || !strings.Contains(err.Error(), "required promotion source missing") {
t.Fatalf("Run() error = %v, want required promotion missing failure", err)
if err == nil || !strings.Contains(err.Error(), "required promotion source unavailable") {
t.Fatalf("Run() error = %v, want required promotion source unavailable failure", err)
}
}
@@ -259,13 +259,13 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) {
sessionRoot := artifacts.SessionWorkDirForCampaign(root, campaign, sessionID)
runRoot := artifacts.SessionRunRootForCampaign(root, campaign, sessionID, runID)
writeStageTestFile(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), "{}\n")
writeStageTestFile(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), "{\"segments\":[]}\n")
writeStageTestFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n")
writeStageTestFile(t, filepath.Join(runRoot, "prepare", "inputs", "session.yml"), "session_id: 2026-04-19\n")
writeStageTestFile(t, filepath.Join(runRoot, "prepare", "outputs", "audio", "speaker.flac"), "flac\n")
writeStageTestFile(t, filepath.Join(runRoot, "transcribe", "outputs", "transcripts", "raw", "speaker.json"), "{}\n")
writeStageTestFile(t, filepath.Join(runRoot, "trim", "outputs", "transcripts", "trimmed.json"), "{}\n")
writeStageTestFile(t, filepath.Join(runRoot, "trim", "outputs", "transcripts", "trimmed.json"), "{\"segments\":[]}\n")
writeStageTestFile(t, filepath.Join(runRoot, "analyze", "outputs", "artifacts", "session_recap.md"), "# recap\n")
writeStageTestFile(t, filepath.Join(runRoot, "polish", "reports", "audita.report.json"), "{}\n")
writeStageTestFile(t, filepath.Join(runRoot, "merge", "config", "seriatim.generated.yml"), "key: value\n")
@@ -298,8 +298,17 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) {
Enabled: boolPtr(true),
UploadRun: boolPtr(true),
PromoteArtifacts: []config.ArchivePromotionRule{
{From: "transcripts/trimmed.json", To: "transcripts/trimmed.json", Required: boolPtr(true)},
{From: "artifacts/session_recap.md", To: "artifacts/session_recap.md", Required: boolPtr(true)},
{Source: "narratio.transcript.trimmed", Dest: "transcripts/trimmed.json", Required: boolPtr(true)},
{Source: "narratio.artifact.session_recap", Dest: "artifacts/session_recap.md", Required: boolPtr(true)},
},
},
Scriptorium: &config.ScriptoriumConfig{
Artifacts: map[string]config.ScriptoriumArtifactConfig{
"session_recap": {
Enabled: true,
PromptID: "dnd.session_recap",
OutputPath: "artifacts/session_recap.md",
},
},
},
},