Updated the analyze stage to accept --artifacts as a CLI flag
This commit is contained in:
@@ -83,7 +83,7 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
||||
}}, nil
|
||||
}
|
||||
|
||||
runtimeCatalog, err := buildAnalyzeRuntimeArtifactCatalog(paths, env.Config.Pipeline.Scriptorium, env.SelectedAnalyzeArtifacts)
|
||||
runtimeCatalog, err := buildAnalyzeRuntimeArtifactCatalog(paths, env.Config.Pipeline.Scriptorium, env.SelectedArtifactKeys)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("analyze: build runtime artifact catalog: %w", err)
|
||||
}
|
||||
|
||||
@@ -613,7 +613,7 @@ func TestAnalyzeAppliesSelectedArtifactsFilter(t *testing.T) {
|
||||
},
|
||||
},
|
||||
}
|
||||
env.SelectedAnalyzeArtifacts = []string{"player_handout"}
|
||||
env.SelectedArtifactKeys = []string{"player_handout"}
|
||||
|
||||
result, err := (analyzeStage{}).Run(context.Background(), env, m)
|
||||
if err != nil {
|
||||
|
||||
@@ -126,12 +126,13 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("archive: build runtime artifact catalog: %w", err)
|
||||
}
|
||||
promotions, skippedOptional, lockedPromotions, err := resolveArchivePromotions(
|
||||
promotions, skippedOptional, skippedUnselected, lockedPromotions, err := resolveArchivePromotions(
|
||||
sessionPaths,
|
||||
m,
|
||||
runtimeCatalog,
|
||||
env.Config.Pipeline.Archive.PromoteArtifacts,
|
||||
env.Config.Pipeline.Archive.Locks,
|
||||
env.SelectedArtifactKeys,
|
||||
sessionPrefix,
|
||||
)
|
||||
if err != nil {
|
||||
@@ -173,6 +174,7 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
||||
promotedUploaded,
|
||||
previousUploaded,
|
||||
skippedOptional,
|
||||
skippedUnselected,
|
||||
lockedPromotions,
|
||||
currentManifestKey,
|
||||
))
|
||||
@@ -201,23 +203,24 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
||||
|
||||
return &StageResult{
|
||||
Metadata: map[string]any{
|
||||
"stage": "archive",
|
||||
"uploaded": true,
|
||||
"s3_bucket": bucket,
|
||||
"s3_run_prefix": runPrefix,
|
||||
"run_files_uploaded": len(runUploaded),
|
||||
"run_uploaded_paths": runUploaded,
|
||||
"promoted_files_uploaded": len(promotedUploaded),
|
||||
"promoted_paths": promotedUploaded,
|
||||
"previous_files_uploaded": len(previousUploaded),
|
||||
"previous_uploaded_paths": previousUploaded,
|
||||
"skipped_optional_promotions": skippedOptional,
|
||||
"locked_promotion_count": len(lockedPromotions),
|
||||
"locked_promotions": lockedPromotionMetadata(lockedPromotions),
|
||||
"current_manifest_key": currentManifestKey,
|
||||
"current_run_id_key": currentRunPointerKey,
|
||||
"current_pointer_written": true,
|
||||
"audio_upload_skipped": true,
|
||||
"stage": "archive",
|
||||
"uploaded": true,
|
||||
"s3_bucket": bucket,
|
||||
"s3_run_prefix": runPrefix,
|
||||
"run_files_uploaded": len(runUploaded),
|
||||
"run_uploaded_paths": runUploaded,
|
||||
"promoted_files_uploaded": len(promotedUploaded),
|
||||
"promoted_paths": promotedUploaded,
|
||||
"previous_files_uploaded": len(previousUploaded),
|
||||
"previous_uploaded_paths": previousUploaded,
|
||||
"skipped_optional_promotions": skippedOptional,
|
||||
"skipped_unselected_promotions": skippedUnselectedPromotionMetadata(skippedUnselected),
|
||||
"locked_promotion_count": len(lockedPromotions),
|
||||
"locked_promotions": lockedPromotionMetadata(lockedPromotions),
|
||||
"current_manifest_key": currentManifestKey,
|
||||
"current_run_id_key": currentRunPointerKey,
|
||||
"current_pointer_written": true,
|
||||
"audio_upload_skipped": true,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
@@ -240,6 +243,12 @@ type archiveLockedPromotion struct {
|
||||
Provenance string
|
||||
}
|
||||
|
||||
type archiveSkippedUnselectedPromotion struct {
|
||||
Source string
|
||||
Dest string
|
||||
Required bool
|
||||
}
|
||||
|
||||
func archiveDisabled(env *Env) bool {
|
||||
cfg := env.Config.Pipeline.Archive
|
||||
if cfg == nil {
|
||||
@@ -340,18 +349,33 @@ func resolveArchivePromotions(
|
||||
catalog *artifacts.ArtifactCatalog,
|
||||
rules []config.ArchivePromotionRule,
|
||||
locks []config.ArchiveLockRule,
|
||||
selectedArtifactKeys []string,
|
||||
sessionPrefix string,
|
||||
) ([]archivePromotion, []string, []archiveLockedPromotion, error) {
|
||||
) ([]archivePromotion, []string, []archiveSkippedUnselectedPromotion, []archiveLockedPromotion, error) {
|
||||
out := make([]archivePromotion, 0, len(rules))
|
||||
skippedOptional := make([]string, 0)
|
||||
skippedUnselected := make([]archiveSkippedUnselectedPromotion, 0)
|
||||
lockedPromotions := make([]archiveLockedPromotion, 0)
|
||||
lockSet := archiveLockSet(locks)
|
||||
selectedSet := archiveSelectedArtifactSet(selectedArtifactKeys)
|
||||
for _, rule := range rules {
|
||||
source := strings.TrimSpace(rule.Source)
|
||||
required := rule.Required == nil || *rule.Required
|
||||
dest, err := resolveArchivePromotionDest(rule, catalog)
|
||||
if err != nil {
|
||||
return nil, nil, nil, fmt.Errorf("source %q: %w", source, err)
|
||||
return nil, nil, nil, nil, fmt.Errorf("source %q: %w", source, err)
|
||||
}
|
||||
if len(selectedSet) > 0 {
|
||||
if key, ok := artifacts.ConfiguredArtifactName(source); ok {
|
||||
if _, selected := selectedSet[key]; !selected {
|
||||
skippedUnselected = append(skippedUnselected, archiveSkippedUnselectedPromotion{
|
||||
Source: source,
|
||||
Dest: dest,
|
||||
Required: required,
|
||||
})
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
lock, locked := lockSet[source]
|
||||
resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, catalog)
|
||||
@@ -371,9 +395,9 @@ func resolveArchivePromotions(
|
||||
continue
|
||||
}
|
||||
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) {
|
||||
return nil, nil, nil, fmt.Errorf("required promotion source unavailable: %q", source)
|
||||
return nil, nil, nil, nil, fmt.Errorf("required promotion source unavailable: %q", source)
|
||||
}
|
||||
return nil, nil, nil, fmt.Errorf("resolve source %q: %w", source, err)
|
||||
return nil, nil, nil, nil, fmt.Errorf("resolve source %q: %w", source, err)
|
||||
}
|
||||
if locked {
|
||||
lockedPromotions = append(lockedPromotions, archiveLockedPromotion{
|
||||
@@ -395,7 +419,22 @@ func resolveArchivePromotions(
|
||||
Provenance: resolved.Provenance,
|
||||
})
|
||||
}
|
||||
return out, skippedOptional, lockedPromotions, nil
|
||||
return out, skippedOptional, skippedUnselected, lockedPromotions, nil
|
||||
}
|
||||
|
||||
func archiveSelectedArtifactSet(selected []string) map[string]struct{} {
|
||||
if len(selected) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make(map[string]struct{}, len(selected))
|
||||
for _, key := range selected {
|
||||
trimmed := strings.TrimSpace(key)
|
||||
if trimmed == "" {
|
||||
continue
|
||||
}
|
||||
out[trimmed] = struct{}{}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func archiveLockSet(locks []config.ArchiveLockRule) map[string]config.ArchiveLockRule {
|
||||
@@ -724,30 +763,44 @@ func archiveMetadataPreview(
|
||||
promotedUploaded []string,
|
||||
previousUploaded []string,
|
||||
skippedOptional []string,
|
||||
skippedUnselected []archiveSkippedUnselectedPromotion,
|
||||
lockedPromotions []archiveLockedPromotion,
|
||||
currentManifestKey string,
|
||||
) map[string]any {
|
||||
return map[string]any{
|
||||
"stage": "archive",
|
||||
"uploaded": true,
|
||||
"s3_bucket": bucket,
|
||||
"s3_run_prefix": runPrefix,
|
||||
"run_files_uploaded": len(runUploaded),
|
||||
"run_uploaded_paths": append([]string(nil), runUploaded...),
|
||||
"promoted_files_uploaded": len(promotedUploaded),
|
||||
"promoted_paths": append([]string(nil), promotedUploaded...),
|
||||
"previous_files_uploaded": len(previousUploaded),
|
||||
"previous_uploaded_paths": append([]string(nil), previousUploaded...),
|
||||
"skipped_optional_promotions": append([]string(nil), skippedOptional...),
|
||||
"locked_promotion_count": len(lockedPromotions),
|
||||
"locked_promotions": lockedPromotionMetadata(lockedPromotions),
|
||||
"current_manifest_key": currentManifestKey,
|
||||
"current_run_id_key": artifacts.S3CurrentRunPointerKey(sessionPrefix),
|
||||
"current_pointer_written": false,
|
||||
"audio_upload_skipped": true,
|
||||
"stage": "archive",
|
||||
"uploaded": true,
|
||||
"s3_bucket": bucket,
|
||||
"s3_run_prefix": runPrefix,
|
||||
"run_files_uploaded": len(runUploaded),
|
||||
"run_uploaded_paths": append([]string(nil), runUploaded...),
|
||||
"promoted_files_uploaded": len(promotedUploaded),
|
||||
"promoted_paths": append([]string(nil), promotedUploaded...),
|
||||
"previous_files_uploaded": len(previousUploaded),
|
||||
"previous_uploaded_paths": append([]string(nil), previousUploaded...),
|
||||
"skipped_optional_promotions": append([]string(nil), skippedOptional...),
|
||||
"skipped_unselected_promotions": skippedUnselectedPromotionMetadata(skippedUnselected),
|
||||
"locked_promotion_count": len(lockedPromotions),
|
||||
"locked_promotions": lockedPromotionMetadata(lockedPromotions),
|
||||
"current_manifest_key": currentManifestKey,
|
||||
"current_run_id_key": artifacts.S3CurrentRunPointerKey(sessionPrefix),
|
||||
"current_pointer_written": false,
|
||||
"audio_upload_skipped": true,
|
||||
}
|
||||
}
|
||||
|
||||
func skippedUnselectedPromotionMetadata(skipped []archiveSkippedUnselectedPromotion) []map[string]any {
|
||||
out := make([]map[string]any, 0, len(skipped))
|
||||
for _, item := range skipped {
|
||||
out = append(out, map[string]any{
|
||||
"source": item.Source,
|
||||
"dest": item.Dest,
|
||||
"required": item.Required,
|
||||
})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func lockedPromotionMetadata(locked []archiveLockedPromotion) []map[string]any {
|
||||
out := make([]map[string]any, 0, len(locked))
|
||||
for _, item := range locked {
|
||||
|
||||
@@ -216,6 +216,108 @@ func TestArchiveSkipsOptionalMissingPromotion(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestArchiveSkipsRequiredUnselectedConfiguredPromotion(t *testing.T) {
|
||||
env, m, _ := archiveFixture(t)
|
||||
env.SelectedArtifactKeys = []string{"player_handout"}
|
||||
env.Config.Pipeline.Scriptorium.Artifacts["player_handout"] = config.ScriptoriumArtifactConfig{
|
||||
Enabled: true,
|
||||
PromptID: "dnd.player_handout",
|
||||
OutputPath: "artifacts/player_handout.md",
|
||||
}
|
||||
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||
|
||||
result, err := archiveStage{}.Run(context.Background(), env, m)
|
||||
if err != nil {
|
||||
t.Fatalf("Run() error = %v", err)
|
||||
}
|
||||
if _, ok := fake.Objects[m.S3SessionPrefix+"transcripts/trimmed.json"]; !ok {
|
||||
t.Fatalf("missing built-in promoted trimmed key")
|
||||
}
|
||||
if _, ok := fake.Objects[m.S3SessionPrefix+"artifacts/session_recap.md"]; ok {
|
||||
t.Fatalf("unexpected unselected recap promotion upload")
|
||||
}
|
||||
skipped := result.Metadata["skipped_unselected_promotions"].([]map[string]any)
|
||||
if len(skipped) != 1 {
|
||||
t.Fatalf("skipped_unselected_promotions = %#v, want one item", skipped)
|
||||
}
|
||||
if skipped[0]["source"] != "narratio.artifact.session_recap" || skipped[0]["dest"] != "artifacts/session_recap.md" || skipped[0]["required"] != true {
|
||||
t.Fatalf("skipped_unselected_promotions[0] = %#v, want session recap", skipped[0])
|
||||
}
|
||||
if result.Metadata["locked_promotion_count"] != 0 {
|
||||
t.Fatalf("locked_promotion_count = %#v, want 0", result.Metadata["locked_promotion_count"])
|
||||
}
|
||||
if _, ok := fake.Objects[m.S3SessionPrefix+"current/run_id.txt"]; !ok {
|
||||
t.Fatalf("missing current pointer")
|
||||
}
|
||||
}
|
||||
|
||||
func TestArchiveSelectedConfiguredPromotionStillFailsWhenMissing(t *testing.T) {
|
||||
env, m, _ := archiveFixture(t)
|
||||
env.SelectedArtifactKeys = []string{"session_recap"}
|
||||
sessionRoot := artifacts.SessionWorkDirForCampaign(
|
||||
env.Config.Pipeline.Workspace.Root,
|
||||
env.Config.Session.Campaign,
|
||||
env.Config.Session.SessionID,
|
||||
)
|
||||
if err := os.Remove(filepath.Join(sessionRoot, "artifacts", "session_recap.md")); err != nil {
|
||||
t.Fatalf("remove recap: %v", err)
|
||||
}
|
||||
|
||||
_, err := archiveStage{}.Run(context.Background(), env, m)
|
||||
if err == nil || !strings.Contains(err.Error(), `required promotion source unavailable: "narratio.artifact.session_recap"`) {
|
||||
t.Fatalf("Run() error = %v, want required selected promotion failure", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestArchiveLockedSelectedPromotionSkipsAsLocked(t *testing.T) {
|
||||
env, m, _ := archiveFixture(t)
|
||||
env.SelectedArtifactKeys = []string{"session_recap"}
|
||||
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{
|
||||
{Source: "narratio.artifact.session_recap", Reason: "reviewed"},
|
||||
}
|
||||
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||
|
||||
result, err := archiveStage{}.Run(context.Background(), env, m)
|
||||
if err != nil {
|
||||
t.Fatalf("Run() error = %v", err)
|
||||
}
|
||||
if _, ok := fake.Objects[m.S3SessionPrefix+"artifacts/session_recap.md"]; ok {
|
||||
t.Fatalf("unexpected locked recap promotion upload")
|
||||
}
|
||||
if result.Metadata["locked_promotion_count"] != 1 {
|
||||
t.Fatalf("locked_promotion_count = %#v, want 1", result.Metadata["locked_promotion_count"])
|
||||
}
|
||||
skipped := result.Metadata["skipped_unselected_promotions"].([]map[string]any)
|
||||
if len(skipped) != 0 {
|
||||
t.Fatalf("skipped_unselected_promotions = %#v, want empty", skipped)
|
||||
}
|
||||
}
|
||||
|
||||
func TestArchiveLockedUnselectedConfiguredPromotionSkipsAsUnselected(t *testing.T) {
|
||||
env, m, _ := archiveFixture(t)
|
||||
env.SelectedArtifactKeys = []string{"player_handout"}
|
||||
env.Config.Pipeline.Scriptorium.Artifacts["player_handout"] = config.ScriptoriumArtifactConfig{
|
||||
Enabled: true,
|
||||
PromptID: "dnd.player_handout",
|
||||
OutputPath: "artifacts/player_handout.md",
|
||||
}
|
||||
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{
|
||||
{Source: "narratio.artifact.session_recap", Reason: "reviewed"},
|
||||
}
|
||||
|
||||
result, err := archiveStage{}.Run(context.Background(), env, m)
|
||||
if err != nil {
|
||||
t.Fatalf("Run() error = %v", err)
|
||||
}
|
||||
if result.Metadata["locked_promotion_count"] != 0 {
|
||||
t.Fatalf("locked_promotion_count = %#v, want 0", result.Metadata["locked_promotion_count"])
|
||||
}
|
||||
skipped := result.Metadata["skipped_unselected_promotions"].([]map[string]any)
|
||||
if len(skipped) != 1 || skipped[0]["source"] != "narratio.artifact.session_recap" {
|
||||
t.Fatalf("skipped_unselected_promotions = %#v, want unselected recap", skipped)
|
||||
}
|
||||
}
|
||||
|
||||
func TestArchiveSkipsLockedRequiredPromotionAndCommits(t *testing.T) {
|
||||
env, m, _ := archiveFixture(t)
|
||||
env.Config.Pipeline.Archive.Locks = []config.ArchiveLockRule{
|
||||
|
||||
@@ -18,11 +18,11 @@ import (
|
||||
|
||||
// Env is the shared dependency container visible to stages.
|
||||
type Env struct {
|
||||
Config *config.Config
|
||||
SelectedAnalyzeArtifacts []string
|
||||
ArtifactStore artifacts.Store
|
||||
ManifestStore manifest.Store
|
||||
Logger *slog.Logger
|
||||
Config *config.Config
|
||||
SelectedArtifactKeys []string
|
||||
ArtifactStore artifacts.Store
|
||||
ManifestStore manifest.Store
|
||||
Logger *slog.Logger
|
||||
|
||||
WhisperX whisperx.Client
|
||||
Seriatim seriatim.Runner
|
||||
|
||||
Reference in New Issue
Block a user