Aligned the archive stage with the new work directory layout
This commit is contained in:
@@ -54,6 +54,13 @@ func runPostArchiveCleanup(ctx context.Context, env *Env, manifestPath string, m
|
|||||||
}
|
}
|
||||||
workDir := strings.TrimSpace(m.LocalWorkDir)
|
workDir := strings.TrimSpace(m.LocalWorkDir)
|
||||||
if workDir == "" {
|
if workDir == "" {
|
||||||
|
workDir = artifacts.SessionRunRootForCampaign(
|
||||||
|
env.Config.Pipeline.Workspace.Root,
|
||||||
|
strings.TrimSpace(env.Config.Session.Campaign),
|
||||||
|
strings.TrimSpace(env.Config.Session.SessionID),
|
||||||
|
strings.TrimSpace(m.RunID),
|
||||||
|
)
|
||||||
|
if info, err := os.Stat(workDir); err != nil || !info.IsDir() {
|
||||||
workDir = artifacts.SessionRunWorkDir(
|
workDir = artifacts.SessionRunWorkDir(
|
||||||
env.Config.Pipeline.Workspace.Root,
|
env.Config.Pipeline.Workspace.Root,
|
||||||
strings.TrimSpace(env.Config.Session.Campaign),
|
strings.TrimSpace(env.Config.Session.Campaign),
|
||||||
@@ -61,6 +68,7 @@ func runPostArchiveCleanup(ctx context.Context, env *Env, manifestPath string, m
|
|||||||
strings.TrimSpace(m.RunID),
|
strings.TrimSpace(m.RunID),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
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_archive"); err != nil {
|
||||||
|
|||||||
@@ -326,7 +326,11 @@ func archiveStageCleanupFixture(t *testing.T) (*config.Config, cleanupSeed, stri
|
|||||||
{From: "artifacts/session_recap.md", To: "artifacts/session_recap.md", Required: boolPtr(true)},
|
{From: "artifacts/session_recap.md", To: "artifacts/session_recap.md", Required: boolPtr(true)},
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
writeArchiveFixtureRunFiles(t, seed.runWorkDir)
|
writeArchiveFixtureRunFiles(
|
||||||
|
t,
|
||||||
|
seed.runWorkDir,
|
||||||
|
artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID),
|
||||||
|
)
|
||||||
|
|
||||||
store := &manifest.LocalStore{}
|
store := &manifest.LocalStore{}
|
||||||
seedManifest, err := store.Load(context.Background(), manifestPathFor(cfg))
|
seedManifest, err := store.Load(context.Background(), manifestPathFor(cfg))
|
||||||
@@ -345,16 +349,19 @@ func archiveStageCleanupFixture(t *testing.T) (*config.Config, cleanupSeed, stri
|
|||||||
return cfg, seed, runID
|
return cfg, seed, runID
|
||||||
}
|
}
|
||||||
|
|
||||||
func writeArchiveFixtureRunFiles(t *testing.T, runWorkDir string) {
|
func writeArchiveFixtureRunFiles(t *testing.T, runWorkDir, sessionRoot string) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
mustWriteFile(t, filepath.Join(runWorkDir, "inputs", "session.yml"), "session_id: 2026-05-03\n")
|
mustWriteFile(t, filepath.Join(runWorkDir, "prepare", "inputs", "session.yml"), "session_id: 2026-05-03\n")
|
||||||
mustWriteFile(t, filepath.Join(runWorkDir, "transcripts", "raw", "speaker.json"), "{}\n")
|
mustWriteFile(t, filepath.Join(runWorkDir, "transcribe", "outputs", "transcripts", "raw", "speaker.json"), "{}\n")
|
||||||
mustWriteFile(t, filepath.Join(runWorkDir, "transcripts", "trimmed.json"), "{}\n")
|
mustWriteFile(t, filepath.Join(runWorkDir, "trim", "outputs", "transcripts", "trimmed.json"), "{}\n")
|
||||||
mustWriteFile(t, filepath.Join(runWorkDir, "artifacts", "session_recap.md"), "# recap\n")
|
mustWriteFile(t, filepath.Join(runWorkDir, "analyze", "outputs", "artifacts", "session_recap.md"), "# recap\n")
|
||||||
mustWriteFile(t, filepath.Join(runWorkDir, "reports", "audita.report.json"), "{}\n")
|
mustWriteFile(t, filepath.Join(runWorkDir, "polish", "reports", "audita.report.json"), "{}\n")
|
||||||
mustWriteFile(t, filepath.Join(runWorkDir, "config", "audita.generated.yml"), "key: value\n")
|
mustWriteFile(t, filepath.Join(runWorkDir, "merge", "config", "seriatim.generated.yml"), "key: value\n")
|
||||||
mustWriteFile(t, filepath.Join(runWorkDir, "logs", "audita.stderr.log"), "stderr\n")
|
mustWriteFile(t, filepath.Join(runWorkDir, "logs", "audita.stderr.log"), "stderr\n")
|
||||||
mustWriteFile(t, filepath.Join(runWorkDir, "manifest.json"), "{}\n")
|
mustWriteFile(t, filepath.Join(runWorkDir, "manifest.json"), "{}\n")
|
||||||
|
|
||||||
|
mustWriteFile(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), "{}\n")
|
||||||
|
mustWriteFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n")
|
||||||
}
|
}
|
||||||
|
|
||||||
type failKeyStore struct {
|
type failKeyStore struct {
|
||||||
|
|||||||
@@ -202,7 +202,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err)
|
return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err)
|
||||||
}
|
}
|
||||||
|
|
||||||
outputs := mapResultOutputs(result)
|
outputs := mapResultOutputs(result, runID)
|
||||||
succeededAt := nowUTC()
|
succeededAt := nowUTC()
|
||||||
m.MarkStageSucceeded(s.Name(), succeededAt, outputs)
|
m.MarkStageSucceeded(s.Name(), succeededAt, outputs)
|
||||||
applyStageResultToManifest(m, s.Name(), result)
|
applyStageResultToManifest(m, s.Name(), result)
|
||||||
@@ -383,11 +383,12 @@ func fileExists(path string) (bool, error) {
|
|||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
|
|
||||||
func mapResultOutputs(result *stage.StageResult) []manifest.ArtifactRecord {
|
func mapResultOutputs(result *stage.StageResult, runID string) []manifest.ArtifactRecord {
|
||||||
if result == nil || len(result.Outputs) == 0 {
|
if result == nil || len(result.Outputs) == 0 {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
runID = strings.TrimSpace(runID)
|
||||||
out := make([]manifest.ArtifactRecord, 0, len(result.Outputs))
|
out := make([]manifest.ArtifactRecord, 0, len(result.Outputs))
|
||||||
for _, ref := range result.Outputs {
|
for _, ref := range result.Outputs {
|
||||||
localPath := ref.AbsolutePath
|
localPath := ref.AbsolutePath
|
||||||
@@ -397,6 +398,7 @@ func mapResultOutputs(result *stage.StageResult) []manifest.ArtifactRecord {
|
|||||||
out = append(out, manifest.ArtifactRecord{
|
out = append(out, manifest.ArtifactRecord{
|
||||||
Kind: ref.Kind,
|
Kind: ref.Kind,
|
||||||
LocalPath: localPath,
|
LocalPath: localPath,
|
||||||
|
ProducerRunID: runID,
|
||||||
RemoteKey: ref.RemoteKey,
|
RemoteKey: ref.RemoteKey,
|
||||||
Checksum: ref.Checksum,
|
Checksum: ref.Checksum,
|
||||||
})
|
})
|
||||||
@@ -446,7 +448,7 @@ func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest, runID stri
|
|||||||
changed = true
|
changed = true
|
||||||
}
|
}
|
||||||
if m.LocalWorkDir == "" && campaign != "" && sessionID != "" && m.RunID != "" {
|
if m.LocalWorkDir == "" && campaign != "" && sessionID != "" && m.RunID != "" {
|
||||||
m.LocalWorkDir = artifacts.SessionRunWorkDir(cfg.Pipeline.Workspace.Root, campaign, sessionID, m.RunID)
|
m.LocalWorkDir = artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, campaign, sessionID, m.RunID)
|
||||||
changed = true
|
changed = true
|
||||||
}
|
}
|
||||||
if m.LocalSpoolDir == "" && campaign != "" && sessionID != "" && m.RunID != "" && strings.TrimSpace(cfg.Pipeline.Spool.Root) != "" {
|
if m.LocalSpoolDir == "" && campaign != "" && sessionID != "" && m.RunID != "" && strings.TrimSpace(cfg.Pipeline.Spool.Root) != "" {
|
||||||
|
|||||||
@@ -463,6 +463,9 @@ func TestExecuteStagesRunLocalArtifactsAndCanonicalPromotion(t *testing.T) {
|
|||||||
if strings.Contains(out.LocalPath, string(filepath.Separator)+"runs"+string(filepath.Separator)) {
|
if strings.Contains(out.LocalPath, string(filepath.Separator)+"runs"+string(filepath.Separator)) {
|
||||||
t.Fatalf("session manifest output should be canonical, got run-local path %q", out.LocalPath)
|
t.Fatalf("session manifest output should be canonical, got run-local path %q", out.LocalPath)
|
||||||
}
|
}
|
||||||
|
if out.ProducerRunID != summary.RunID {
|
||||||
|
t.Fatalf("producer_run_id = %q, want %q", out.ProducerRunID, summary.RunID)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath)
|
runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath)
|
||||||
@@ -480,6 +483,47 @@ func TestExecuteStagesRunLocalArtifactsAndCanonicalPromotion(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestExecuteStagesSkippedStagePreservesExistingOutputsProvenance(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
store := &manifest.LocalStore{}
|
||||||
|
manifestPath := manifestPathFor(cfg)
|
||||||
|
|
||||||
|
existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC))
|
||||||
|
existing.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), []manifest.ArtifactRecord{
|
||||||
|
{
|
||||||
|
Kind: "transcript_raw",
|
||||||
|
LocalPath: "transcripts/raw/alice.json",
|
||||||
|
ProducerRunID: "20260501T000000Z-deadbeef",
|
||||||
|
},
|
||||||
|
})
|
||||||
|
if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil {
|
||||||
|
t.Fatalf("MkdirAll() error = %v", err)
|
||||||
|
}
|
||||||
|
if err := store.Save(context.Background(), manifestPath, existing); err != nil {
|
||||||
|
t.Fatalf("Save manifest error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
summary, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[1]}, RunOptions{})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("executeStages() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(summary.Skipped) != 1 || summary.Skipped[0] != "transcribe" {
|
||||||
|
t.Fatalf("summary = %#v, want skipped transcribe", summary)
|
||||||
|
}
|
||||||
|
|
||||||
|
loaded, err := store.Load(context.Background(), manifestPath)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Load manifest error = %v", err)
|
||||||
|
}
|
||||||
|
got := loaded.Stages["transcribe"]
|
||||||
|
if got == nil || len(got.Outputs) != 1 {
|
||||||
|
t.Fatalf("transcribe outputs = %#v, want one preserved output", got)
|
||||||
|
}
|
||||||
|
if got.Outputs[0].ProducerRunID != "20260501T000000Z-deadbeef" {
|
||||||
|
t.Fatalf("producer_run_id = %q, want preserved value", got.Outputs[0].ProducerRunID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) {
|
func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) {
|
||||||
cases := []struct {
|
cases := []struct {
|
||||||
name string
|
name string
|
||||||
|
|||||||
@@ -29,6 +29,8 @@ type InputRecord struct {
|
|||||||
type ArtifactRecord struct {
|
type ArtifactRecord struct {
|
||||||
Kind string `json:"kind"`
|
Kind string `json:"kind"`
|
||||||
LocalPath string `json:"local_path"`
|
LocalPath string `json:"local_path"`
|
||||||
|
// ProducerRunID identifies the run that produced this durable artifact.
|
||||||
|
ProducerRunID string `json:"producer_run_id,omitempty"`
|
||||||
RemoteKey string `json:"remote_key,omitempty"`
|
RemoteKey string `json:"remote_key,omitempty"`
|
||||||
Checksum string `json:"checksum,omitempty"`
|
Checksum string `json:"checksum,omitempty"`
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -256,6 +256,32 @@ func TestAnalyzeOmitsOptionalPreviousRecapWhenUnavailable(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestAnalyzeUsesRunLocalPathsAndPromotesCanonical(t *testing.T) {
|
||||||
|
env, m, fake := setupAnalyzeEnv(t)
|
||||||
|
env.Config.Session.Campaign = "sample-campaign"
|
||||||
|
m.Campaign = "sample-campaign"
|
||||||
|
m.RunID = "20260518T010203Z-abcdef12"
|
||||||
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
|
writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`)
|
||||||
|
|
||||||
|
result, err := (analyzeStage{}).Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(fake.RunRequests) != 1 {
|
||||||
|
t.Fatalf("run requests = %d, want 1", len(fake.RunRequests))
|
||||||
|
}
|
||||||
|
if !strings.Contains(fake.RunRequests[0].OutputPath, filepath.Join("runs", m.RunID, "analyze", "outputs")) {
|
||||||
|
t.Fatalf("run output path = %q, want run-local path", fake.RunRequests[0].OutputPath)
|
||||||
|
}
|
||||||
|
if len(result.Outputs) != 1 {
|
||||||
|
t.Fatalf("outputs len = %d, want 1", len(result.Outputs))
|
||||||
|
}
|
||||||
|
if strings.Contains(result.Outputs[0].AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) {
|
||||||
|
t.Fatalf("promoted output path = %q, want canonical session path", result.Outputs[0].AbsolutePath)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
type orderedScriptoriumRunner struct {
|
type orderedScriptoriumRunner struct {
|
||||||
Calls []string
|
Calls []string
|
||||||
RenderErr error
|
RenderErr error
|
||||||
|
|||||||
@@ -33,15 +33,6 @@ var archivePrerequisiteStages = []string{
|
|||||||
"analyze",
|
"analyze",
|
||||||
}
|
}
|
||||||
|
|
||||||
var archiveRunUploadDirs = []string{
|
|
||||||
"inputs",
|
|
||||||
"transcripts",
|
|
||||||
"artifacts",
|
|
||||||
"reports",
|
|
||||||
"config",
|
|
||||||
"logs",
|
|
||||||
}
|
|
||||||
|
|
||||||
func (archiveStage) Name() string { return "archive" }
|
func (archiveStage) Name() string { return "archive" }
|
||||||
|
|
||||||
func (archiveStage) Declares() IODecl {
|
func (archiveStage) Declares() IODecl {
|
||||||
@@ -87,16 +78,16 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
return nil, fmt.Errorf("archive: remote object store backend is required when archive run upload is enabled")
|
return nil, fmt.Errorf("archive: remote object store backend is required when archive run upload is enabled")
|
||||||
}
|
}
|
||||||
|
|
||||||
workDir, err := archiveWorkDir(env, m)
|
runRoot, err := resolveArchiveRunRoot(env, m)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("archive: resolve local workdir: %w", err)
|
return nil, fmt.Errorf("archive: resolve run root: %w", err)
|
||||||
}
|
}
|
||||||
workDirInfo, err := os.Stat(workDir)
|
runRootInfo, err := os.Stat(runRoot)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("archive: local workdir %q: %w", workDir, err)
|
return nil, fmt.Errorf("archive: run root %q: %w", runRoot, err)
|
||||||
}
|
}
|
||||||
if !workDirInfo.IsDir() {
|
if !runRootInfo.IsDir() {
|
||||||
return nil, fmt.Errorf("archive: local workdir %q is not a directory", workDir)
|
return nil, fmt.Errorf("archive: run root %q is not a directory", runRoot)
|
||||||
}
|
}
|
||||||
|
|
||||||
runPrefix, err := archiveRunPrefix(env, m)
|
runPrefix, err := archiveRunPrefix(env, m)
|
||||||
@@ -116,17 +107,20 @@ func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
return nil, fmt.Errorf("archive: run id is required")
|
return nil, fmt.Errorf("archive: run id is required")
|
||||||
}
|
}
|
||||||
|
|
||||||
manifestSource, err := resolveArchiveManifestSource(env, m, workDir)
|
manifestSource, err := resolveArchiveRunManifestSource(runRoot)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("archive: resolve manifest source: %w", err)
|
return nil, fmt.Errorf("archive: resolve run manifest source: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
runFiles, err := collectArchiveRunFiles(workDir, 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("archive: collect run files: %w", err)
|
||||||
}
|
}
|
||||||
workDirCandidates := archiveSourceWorkDirs(env, m, workDir)
|
sessionRoot, err := resolveArchiveSessionRoot(env, m)
|
||||||
promotions, err := resolveArchivePromotions(workDirCandidates, env.Config.Pipeline.Archive.PromoteArtifacts)
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("archive: resolve session root for promotions: %w", err)
|
||||||
|
}
|
||||||
|
promotions, err := resolveArchivePromotions(sessionRoot, env.Config.Pipeline.Archive.PromoteArtifacts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
|
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
|
||||||
}
|
}
|
||||||
@@ -249,39 +243,67 @@ func validateArchivePrerequisites(m *manifest.Manifest) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func archiveWorkDir(env *Env, m *manifest.Manifest) (string, error) {
|
func resolveArchiveRunRoot(env *Env, m *manifest.Manifest) (string, error) {
|
||||||
workDir := strings.TrimSpace(m.LocalWorkDir)
|
|
||||||
if workDir != "" {
|
|
||||||
cleaned := filepath.Clean(workDir)
|
|
||||||
if info, err := os.Stat(cleaned); err == nil && info.IsDir() {
|
|
||||||
return cleaned, nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
|
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
|
||||||
if sessionID == "" {
|
if sessionID == "" && m != nil {
|
||||||
sessionID = strings.TrimSpace(m.SessionID)
|
sessionID = strings.TrimSpace(m.SessionID)
|
||||||
}
|
}
|
||||||
campaign := strings.TrimSpace(env.Config.Session.Campaign)
|
campaign := strings.TrimSpace(env.Config.Session.Campaign)
|
||||||
if campaign == "" {
|
if campaign == "" && m != nil {
|
||||||
campaign = strings.TrimSpace(m.Campaign)
|
campaign = strings.TrimSpace(m.Campaign)
|
||||||
}
|
}
|
||||||
runID := strings.TrimSpace(m.RunID)
|
runID := ""
|
||||||
|
if m != nil {
|
||||||
|
runID = strings.TrimSpace(m.RunID)
|
||||||
|
}
|
||||||
|
if sessionID == "" || campaign == "" {
|
||||||
|
return "", fmt.Errorf("campaign and session id are required")
|
||||||
|
}
|
||||||
if runID == "" {
|
if runID == "" {
|
||||||
return "", fmt.Errorf("run id is required")
|
return "", fmt.Errorf("run id is required")
|
||||||
}
|
}
|
||||||
if campaign == "" || sessionID == "" {
|
|
||||||
return "", fmt.Errorf("campaign and session id are required")
|
canonical := filepath.Clean(artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID))
|
||||||
|
legacy := filepath.Clean(artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID))
|
||||||
|
|
||||||
|
canonicalExists, err := directoryExists(canonical)
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("check canonical run root %q: %w", canonical, err)
|
||||||
}
|
}
|
||||||
runScoped := artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID)
|
legacyExists, err := directoryExists(legacy)
|
||||||
if info, err := os.Stat(runScoped); err == nil && info.IsDir() {
|
if err != nil {
|
||||||
return runScoped, nil
|
return "", fmt.Errorf("check legacy run root %q: %w", legacy, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
switch {
|
||||||
|
case canonicalExists && legacyExists && canonical != legacy:
|
||||||
|
return "", fmt.Errorf("ambiguous run roots for campaign %q session %q run %q: canonical=%q legacy=%q", campaign, sessionID, runID, canonical, legacy)
|
||||||
|
case canonicalExists:
|
||||||
|
return canonical, nil
|
||||||
|
case legacyExists:
|
||||||
|
return legacy, nil
|
||||||
|
default:
|
||||||
|
return "", fmt.Errorf("run root not found for campaign %q session %q run %q (checked canonical=%q legacy=%q)", campaign, sessionID, runID, canonical, legacy)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func resolveArchiveSessionRoot(env *Env, m *manifest.Manifest) (string, error) {
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
if sessionID == "" {
|
||||||
|
return "", fmt.Errorf("session id is required")
|
||||||
}
|
}
|
||||||
paths, err := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).ResolveSessionPathsFor(campaign, sessionID)
|
paths, err := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).ResolveSessionPathsFor(campaign, sessionID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
return paths.Root, nil
|
return filepath.Clean(paths.Root), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) {
|
func archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) {
|
||||||
@@ -334,9 +356,10 @@ func archiveBucket(env *Env, m *manifest.Manifest) string {
|
|||||||
return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket)
|
return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket)
|
||||||
}
|
}
|
||||||
|
|
||||||
func resolveArchivePromotions(workDirs []string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) {
|
func resolveArchivePromotions(sessionRoot string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) {
|
||||||
if len(workDirs) == 0 {
|
sessionRoot = filepath.Clean(strings.TrimSpace(sessionRoot))
|
||||||
return nil, fmt.Errorf("at least one workdir candidate is required")
|
if sessionRoot == "" {
|
||||||
|
return nil, fmt.Errorf("session root is required")
|
||||||
}
|
}
|
||||||
out := make([]archivePromotion, 0, len(rules))
|
out := make([]archivePromotion, 0, len(rules))
|
||||||
for _, rule := range rules {
|
for _, rule := range rules {
|
||||||
@@ -344,28 +367,16 @@ func resolveArchivePromotions(workDirs []string, rules []config.ArchivePromotion
|
|||||||
to := strings.TrimSpace(rule.To)
|
to := strings.TrimSpace(rule.To)
|
||||||
required := rule.Required == nil || *rule.Required
|
required := rule.Required == nil || *rule.Required
|
||||||
|
|
||||||
var (
|
resolvedPath, err := resolveWorkDirRelativePath(sessionRoot, from)
|
||||||
localPath string
|
|
||||||
exists bool
|
|
||||||
)
|
|
||||||
for _, candidateRoot := range workDirs {
|
|
||||||
resolvedPath, err := resolveWorkDirRelativePath(candidateRoot, from)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("promotion from %q: %w", from, err)
|
return nil, fmt.Errorf("promotion from %q: %w", from, err)
|
||||||
}
|
}
|
||||||
info, err := os.Stat(resolvedPath)
|
info, err := os.Stat(resolvedPath)
|
||||||
if err == nil && !info.IsDir() {
|
exists := err == nil && !info.IsDir()
|
||||||
localPath = resolvedPath
|
|
||||||
exists = true
|
|
||||||
break
|
|
||||||
}
|
|
||||||
if err != nil && !os.IsNotExist(err) {
|
if err != nil && !os.IsNotExist(err) {
|
||||||
return nil, fmt.Errorf("promotion source %q: %w", from, err)
|
return nil, fmt.Errorf("promotion source %q: %w", from, err)
|
||||||
}
|
}
|
||||||
if localPath == "" {
|
localPath := resolvedPath
|
||||||
localPath = resolvedPath
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
out = append(out, archivePromotion{
|
out = append(out, archivePromotion{
|
||||||
From: from,
|
From: from,
|
||||||
@@ -378,35 +389,6 @@ func resolveArchivePromotions(workDirs []string, rules []config.ArchivePromotion
|
|||||||
return out, nil
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func archiveSourceWorkDirs(env *Env, m *manifest.Manifest, runWorkDir string) []string {
|
|
||||||
candidates := make([]string, 0, 2)
|
|
||||||
if strings.TrimSpace(runWorkDir) != "" {
|
|
||||||
candidates = append(candidates, filepath.Clean(runWorkDir))
|
|
||||||
}
|
|
||||||
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
|
|
||||||
if sessionID == "" && m != nil {
|
|
||||||
sessionID = strings.TrimSpace(m.SessionID)
|
|
||||||
}
|
|
||||||
if sessionID != "" {
|
|
||||||
paths, err := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).ResolveSessionPathsFor(strings.TrimSpace(env.Config.Session.Campaign), sessionID)
|
|
||||||
if err == nil {
|
|
||||||
if cleanRoot := filepath.Clean(paths.Root); cleanRoot != "" {
|
|
||||||
candidates = append(candidates, cleanRoot)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
seen := make(map[string]struct{}, len(candidates))
|
|
||||||
out := make([]string, 0, len(candidates))
|
|
||||||
for _, c := range candidates {
|
|
||||||
if _, ok := seen[c]; ok {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
seen[c] = struct{}{}
|
|
||||||
out = append(out, c)
|
|
||||||
}
|
|
||||||
return out
|
|
||||||
}
|
|
||||||
|
|
||||||
func resolveWorkDirRelativePath(workDir, rel string) (string, error) {
|
func resolveWorkDirRelativePath(workDir, rel string) (string, error) {
|
||||||
rel = filepath.Clean(filepath.FromSlash(strings.TrimSpace(rel)))
|
rel = filepath.Clean(filepath.FromSlash(strings.TrimSpace(rel)))
|
||||||
if rel == "." || rel == "" {
|
if rel == "." || rel == "" {
|
||||||
@@ -425,32 +407,30 @@ func resolveWorkDirRelativePath(workDir, rel string) (string, error) {
|
|||||||
return cleanedFull, nil
|
return cleanedFull, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func collectArchiveRunFiles(workDir, manifestPath string) ([]archiveUploadFile, error) {
|
func collectArchiveRunFiles(runRoot, manifestPath string) ([]archiveUploadFile, error) {
|
||||||
files := make([]archiveUploadFile, 0, 64)
|
files := make([]archiveUploadFile, 0, 64)
|
||||||
|
err := filepath.WalkDir(runRoot, func(path string, d fs.DirEntry, walkErr error) error {
|
||||||
for _, dirName := range archiveRunUploadDirs {
|
|
||||||
fullDir := filepath.Join(workDir, dirName)
|
|
||||||
info, err := os.Stat(fullDir)
|
|
||||||
if err != nil {
|
|
||||||
if os.IsNotExist(err) {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
return nil, fmt.Errorf("stat %q: %w", fullDir, err)
|
|
||||||
}
|
|
||||||
if !info.IsDir() {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := filepath.WalkDir(fullDir, func(path string, d fs.DirEntry, walkErr error) error {
|
|
||||||
if walkErr != nil {
|
if walkErr != nil {
|
||||||
return walkErr
|
return walkErr
|
||||||
}
|
}
|
||||||
if d.IsDir() {
|
if d.IsDir() {
|
||||||
|
if path == runRoot {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
rel, err := filepath.Rel(workDir, path)
|
relDir, err := filepath.Rel(runRoot, path)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("relative path from %q to %q: %w", workDir, path, err)
|
return fmt.Errorf("relative dir from %q to %q: %w", runRoot, path, err)
|
||||||
|
}
|
||||||
|
relDir = filepath.ToSlash(relDir)
|
||||||
|
// Preserve existing behavior: audio is not uploaded in archive run record.
|
||||||
|
if relDir == "audio" || strings.HasPrefix(relDir, "audio/") {
|
||||||
|
return filepath.SkipDir
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
rel, err := filepath.Rel(runRoot, path)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("relative path from %q to %q: %w", runRoot, path, err)
|
||||||
}
|
}
|
||||||
rel = filepath.ToSlash(rel)
|
rel = filepath.ToSlash(rel)
|
||||||
files = append(files, archiveUploadFile{
|
files = append(files, archiveUploadFile{
|
||||||
@@ -458,9 +438,9 @@ func collectArchiveRunFiles(workDir, manifestPath string) ([]archiveUploadFile,
|
|||||||
LocalPath: path,
|
LocalPath: path,
|
||||||
})
|
})
|
||||||
return nil
|
return nil
|
||||||
}); err != nil {
|
})
|
||||||
return nil, fmt.Errorf("walk %q: %w", fullDir, err)
|
if err != nil {
|
||||||
}
|
return nil, fmt.Errorf("walk %q: %w", runRoot, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
manifestInfo, err := os.Stat(manifestPath)
|
manifestInfo, err := os.Stat(manifestPath)
|
||||||
@@ -477,6 +457,14 @@ func collectArchiveRunFiles(workDir, manifestPath string) ([]archiveUploadFile,
|
|||||||
RelativePath: "manifest.json",
|
RelativePath: "manifest.json",
|
||||||
LocalPath: manifestPath,
|
LocalPath: manifestPath,
|
||||||
})
|
})
|
||||||
|
seen := map[string]archiveUploadFile{}
|
||||||
|
for _, file := range files {
|
||||||
|
seen[file.RelativePath] = file
|
||||||
|
}
|
||||||
|
files = files[:0]
|
||||||
|
for _, file := range seen {
|
||||||
|
files = append(files, file)
|
||||||
|
}
|
||||||
|
|
||||||
sort.Slice(files, func(i, j int) bool {
|
sort.Slice(files, func(i, j int) bool {
|
||||||
return files[i].RelativePath < files[j].RelativePath
|
return files[i].RelativePath < files[j].RelativePath
|
||||||
@@ -484,42 +472,30 @@ func collectArchiveRunFiles(workDir, manifestPath string) ([]archiveUploadFile,
|
|||||||
return files, nil
|
return files, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func resolveArchiveManifestSource(env *Env, m *manifest.Manifest, workDir string) (string, error) {
|
func resolveArchiveRunManifestSource(runRoot string) (string, error) {
|
||||||
candidates := make([]string, 0, 3)
|
path := filepath.Join(filepath.Clean(runRoot), "manifest.json")
|
||||||
candidates = append(candidates, filepath.Join(workDir, "manifest.json"))
|
info, err := os.Stat(path)
|
||||||
|
|
||||||
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
|
|
||||||
if sessionID == "" && m != nil {
|
|
||||||
sessionID = strings.TrimSpace(m.SessionID)
|
|
||||||
}
|
|
||||||
if sessionID != "" {
|
|
||||||
paths, err := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).ResolveSessionPathsFor(strings.TrimSpace(env.Config.Session.Campaign), sessionID)
|
|
||||||
if err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
candidates = append(candidates, paths.ManifestPath)
|
|
||||||
}
|
|
||||||
|
|
||||||
seen := make(map[string]struct{}, len(candidates))
|
|
||||||
for _, candidate := range candidates {
|
|
||||||
clean := filepath.Clean(candidate)
|
|
||||||
if _, ok := seen[clean]; ok {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
seen[clean] = struct{}{}
|
|
||||||
info, err := os.Stat(clean)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if os.IsNotExist(err) {
|
if os.IsNotExist(err) {
|
||||||
continue
|
return "", fmt.Errorf("manifest.json not found in run root %q", runRoot)
|
||||||
}
|
}
|
||||||
return "", fmt.Errorf("stat %q: %w", clean, err)
|
return "", fmt.Errorf("stat %q: %w", path, err)
|
||||||
}
|
}
|
||||||
if info.IsDir() {
|
if info.IsDir() {
|
||||||
continue
|
return "", fmt.Errorf("manifest path %q is a directory", path)
|
||||||
}
|
}
|
||||||
return clean, nil
|
return path, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func directoryExists(path string) (bool, error) {
|
||||||
|
info, err := os.Stat(path)
|
||||||
|
if err == nil {
|
||||||
|
return info.IsDir(), nil
|
||||||
}
|
}
|
||||||
return "", fmt.Errorf("manifest.json not found in run workdir or session workdir")
|
if os.IsNotExist(err) {
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
return false, err
|
||||||
}
|
}
|
||||||
|
|
||||||
func writeCurrentManifestSnapshot(m *manifest.Manifest, archiveMetadata map[string]any) (string, error) {
|
func writeCurrentManifestSnapshot(m *manifest.Manifest, archiveMetadata map[string]any) (string, error) {
|
||||||
|
|||||||
@@ -73,14 +73,15 @@ func TestArchiveUploadsRunRecordPromotionsAndCurrentPointer(t *testing.T) {
|
|||||||
runPrefix := m.S3RunPrefix
|
runPrefix := m.S3RunPrefix
|
||||||
sessionPrefix := m.S3SessionPrefix
|
sessionPrefix := m.S3SessionPrefix
|
||||||
wantRunUploads := []string{
|
wantRunUploads := []string{
|
||||||
"artifacts/session_recap.md",
|
"analyze/outputs/artifacts/session_recap.md",
|
||||||
"config/audita.generated.yml",
|
"merge/config/seriatim.generated.yml",
|
||||||
"inputs/session.yml",
|
"prepare/inputs/session.yml",
|
||||||
|
"prepare/outputs/audio/speaker.flac",
|
||||||
"logs/audita.stderr.log",
|
"logs/audita.stderr.log",
|
||||||
"manifest.json",
|
"manifest.json",
|
||||||
"reports/audita.report.json",
|
"polish/reports/audita.report.json",
|
||||||
"transcripts/raw/speaker.json",
|
"transcribe/outputs/transcripts/raw/speaker.json",
|
||||||
"transcripts/trimmed.json",
|
"trim/outputs/transcripts/trimmed.json",
|
||||||
}
|
}
|
||||||
for _, rel := range wantRunUploads {
|
for _, rel := range wantRunUploads {
|
||||||
key := runPrefix + rel
|
key := runPrefix + rel
|
||||||
@@ -132,12 +133,11 @@ func TestArchiveUploadsRunRecordPromotionsAndCurrentPointer(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestArchiveUsesCustomPromotionRules(t *testing.T) {
|
func TestArchiveUsesCustomPromotionRules(t *testing.T) {
|
||||||
env, m, workDir := archiveFixture(t)
|
env, m, _ := archiveFixture(t)
|
||||||
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{
|
env.Config.Pipeline.Archive.PromoteArtifacts = []config.ArchivePromotionRule{
|
||||||
{From: "transcripts/trimmed.json", To: "published/trimmed.json", Required: boolPtr(true)},
|
{From: "transcripts/trimmed.json", To: "published/trimmed.json", Required: boolPtr(true)},
|
||||||
{From: "artifacts/session_recap.md", To: "published/recap.md", Required: boolPtr(true)},
|
{From: "artifacts/session_recap.md", To: "published/recap.md", Required: boolPtr(true)},
|
||||||
}
|
}
|
||||||
writeStageTestFile(t, filepath.Join(workDir, "published", "ignored.txt"), "ignore\n")
|
|
||||||
|
|
||||||
_, err := archiveStage{}.Run(context.Background(), env, m)
|
_, err := archiveStage{}.Run(context.Background(), env, m)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -184,14 +184,9 @@ func TestArchiveFailsWhenRequiredPromotionMissing(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestArchivePromotionFailsOnAmbiguousSessionRoots(t *testing.T) {
|
func TestArchivePromotionFailsOnAmbiguousSessionRoots(t *testing.T) {
|
||||||
env, m, runWorkDir := archiveFixture(t)
|
env, m, _ := archiveFixture(t)
|
||||||
sessionWorkDir := artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, m.SessionID)
|
sessionWorkDir := artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, m.SessionID)
|
||||||
sessionTrimmed := filepath.Join(sessionWorkDir, "transcripts", "trimmed.json")
|
writeStageTestFile(t, filepath.Join(sessionWorkDir, "transcripts", "trimmed.json"), "{}\n")
|
||||||
|
|
||||||
if err := os.Remove(filepath.Join(runWorkDir, "transcripts", "trimmed.json")); err != nil {
|
|
||||||
t.Fatalf("remove run scoped trimmed transcript: %v", err)
|
|
||||||
}
|
|
||||||
writeStageTestFile(t, sessionTrimmed, "{}\n")
|
|
||||||
|
|
||||||
_, err := archiveStage{}.Run(context.Background(), env, m)
|
_, err := archiveStage{}.Run(context.Background(), env, m)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
@@ -202,6 +197,50 @@ func TestArchivePromotionFailsOnAmbiguousSessionRoots(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestArchiveFallsBackToLegacyRunRootWhenCanonicalMissing(t *testing.T) {
|
||||||
|
env, m, runRoot := archiveFixture(t)
|
||||||
|
legacyRoot := artifacts.SessionRunWorkDir(
|
||||||
|
env.Config.Pipeline.Workspace.Root,
|
||||||
|
env.Config.Session.Campaign,
|
||||||
|
env.Config.Session.SessionID,
|
||||||
|
m.RunID,
|
||||||
|
)
|
||||||
|
if err := os.MkdirAll(filepath.Dir(legacyRoot), 0o755); err != nil {
|
||||||
|
t.Fatalf("create legacy run-root parent: %v", err)
|
||||||
|
}
|
||||||
|
if err := os.Rename(runRoot, legacyRoot); err != nil {
|
||||||
|
t.Fatalf("move canonical run root to legacy root: %v", err)
|
||||||
|
}
|
||||||
|
m.LocalWorkDir = legacyRoot
|
||||||
|
|
||||||
|
if _, err := (archiveStage{}).Run(context.Background(), env, m); err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArchiveFailsOnAmbiguousCanonicalAndLegacyRunRoots(t *testing.T) {
|
||||||
|
env, m, runRoot := archiveFixture(t)
|
||||||
|
legacyRoot := artifacts.SessionRunWorkDir(
|
||||||
|
env.Config.Pipeline.Workspace.Root,
|
||||||
|
env.Config.Session.Campaign,
|
||||||
|
env.Config.Session.SessionID,
|
||||||
|
m.RunID,
|
||||||
|
)
|
||||||
|
writeStageTestFile(t, filepath.Join(legacyRoot, "manifest.json"), "{}\n")
|
||||||
|
writeStageTestFile(t, filepath.Join(legacyRoot, "prepare", "inputs", "session.yml"), "session_id: 2026-04-19\n")
|
||||||
|
if runRoot == legacyRoot {
|
||||||
|
t.Fatalf("test requires distinct canonical and legacy run roots, got %q", runRoot)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err := archiveStage{}.Run(context.Background(), env, m)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("expected ambiguity error, got nil")
|
||||||
|
}
|
||||||
|
if !strings.Contains(err.Error(), "ambiguous run roots") {
|
||||||
|
t.Fatalf("error = %v, want run-root ambiguity", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestArchiveDoesNotWriteCurrentPointerWhenPromotionUploadFails(t *testing.T) {
|
func TestArchiveDoesNotWriteCurrentPointerWhenPromotionUploadFails(t *testing.T) {
|
||||||
env, m, _ := archiveFixture(t)
|
env, m, _ := archiveFixture(t)
|
||||||
fake := env.ObjectStore.(*storage.FakeBackend)
|
fake := env.ObjectStore.(*storage.FakeBackend)
|
||||||
@@ -260,22 +299,27 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) {
|
|||||||
runID := "20260516T010203Z-1a2b3c4d"
|
runID := "20260516T010203Z-1a2b3c4d"
|
||||||
campaign := "forsaken"
|
campaign := "forsaken"
|
||||||
sessionID := "2026-04-19"
|
sessionID := "2026-04-19"
|
||||||
workDir := filepath.Join(root, "work", campaign, sessionID, runID)
|
sessionRoot := artifacts.SessionWorkDirForCampaign(root, campaign, sessionID)
|
||||||
|
runRoot := artifacts.SessionRunRootForCampaign(root, campaign, sessionID, runID)
|
||||||
|
|
||||||
writeStageTestFile(t, filepath.Join(workDir, "inputs", "session.yml"), "session_id: 2026-04-19\n")
|
writeStageTestFile(t, filepath.Join(sessionRoot, "transcripts", "trimmed.json"), "{}\n")
|
||||||
writeStageTestFile(t, filepath.Join(workDir, "transcripts", "raw", "speaker.json"), "{}\n")
|
writeStageTestFile(t, filepath.Join(sessionRoot, "artifacts", "session_recap.md"), "# recap\n")
|
||||||
writeStageTestFile(t, filepath.Join(workDir, "transcripts", "trimmed.json"), "{}\n")
|
|
||||||
writeStageTestFile(t, filepath.Join(workDir, "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(workDir, "reports", "audita.report.json"), "{}\n")
|
writeStageTestFile(t, filepath.Join(runRoot, "prepare", "outputs", "audio", "speaker.flac"), "flac\n")
|
||||||
writeStageTestFile(t, filepath.Join(workDir, "config", "audita.generated.yml"), "key: value\n")
|
writeStageTestFile(t, filepath.Join(runRoot, "transcribe", "outputs", "transcripts", "raw", "speaker.json"), "{}\n")
|
||||||
writeStageTestFile(t, filepath.Join(workDir, "logs", "audita.stderr.log"), "stderr\n")
|
writeStageTestFile(t, filepath.Join(runRoot, "trim", "outputs", "transcripts", "trimmed.json"), "{}\n")
|
||||||
writeStageTestFile(t, filepath.Join(workDir, "audio", "speaker.flac"), "flac")
|
writeStageTestFile(t, filepath.Join(runRoot, "analyze", "outputs", "artifacts", "session_recap.md"), "# recap\n")
|
||||||
writeStageTestFile(t, filepath.Join(workDir, "manifest.json"), "{}\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")
|
||||||
|
writeStageTestFile(t, filepath.Join(runRoot, "logs", "audita.stderr.log"), "stderr\n")
|
||||||
|
writeStageTestFile(t, filepath.Join(runRoot, "audio", "speaker.flac"), "flac")
|
||||||
|
writeStageTestFile(t, filepath.Join(runRoot, "manifest.json"), "{}\n")
|
||||||
|
|
||||||
m := manifest.New(sessionID, time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC))
|
m := manifest.New(sessionID, time.Date(2026, 5, 16, 1, 2, 3, 0, time.UTC))
|
||||||
m.Campaign = campaign
|
m.Campaign = campaign
|
||||||
m.RunID = runID
|
m.RunID = runID
|
||||||
m.LocalWorkDir = workDir
|
m.LocalWorkDir = runRoot
|
||||||
m.S3Bucket = "my-dnd-archive"
|
m.S3Bucket = "my-dnd-archive"
|
||||||
m.S3SessionPrefix = artifacts.S3SessionPrefix("dnd", campaign, sessionID)
|
m.S3SessionPrefix = artifacts.S3SessionPrefix("dnd", campaign, sessionID)
|
||||||
m.S3RunPrefix = artifacts.S3RunPrefix(m.S3SessionPrefix, runID)
|
m.S3RunPrefix = artifacts.S3RunPrefix(m.S3SessionPrefix, runID)
|
||||||
@@ -309,7 +353,7 @@ func archiveFixture(t *testing.T) (*Env, *manifest.Manifest, string) {
|
|||||||
},
|
},
|
||||||
ObjectStore: &storage.FakeBackend{},
|
ObjectStore: &storage.FakeBackend{},
|
||||||
}
|
}
|
||||||
return env, m, workDir
|
return env, m, runRoot
|
||||||
}
|
}
|
||||||
|
|
||||||
type promotionFailingStore struct {
|
type promotionFailingStore struct {
|
||||||
|
|||||||
@@ -289,6 +289,39 @@ func TestMergeStageFailsWhenNormalizedOutputInvalid(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestMergeStageUsesRunLocalPathsAndPromotesCanonical(t *testing.T) {
|
||||||
|
env, m := setupMergeEnv(t)
|
||||||
|
env.Config.Session.Campaign = "sample-campaign"
|
||||||
|
m.Campaign = "sample-campaign"
|
||||||
|
m.RunID = "20260518T010203Z-abcdef12"
|
||||||
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
|
|
||||||
|
in := filepath.Join(paths.TranscriptsRawDir, "alice.json")
|
||||||
|
writeFile(t, in, `{"segments":[]}`)
|
||||||
|
writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n")
|
||||||
|
writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n")
|
||||||
|
|
||||||
|
fake := &seriatim.FakeRunner{}
|
||||||
|
env.Seriatim = fake
|
||||||
|
result, err := (mergeStage{}).Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("merge.Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(fake.Requests) != 1 {
|
||||||
|
t.Fatalf("requests = %d, want 1", len(fake.Requests))
|
||||||
|
}
|
||||||
|
req := fake.Requests[0]
|
||||||
|
if !strings.Contains(req.OutputMergedTranscriptPath, filepath.Join("runs", m.RunID, "merge", "outputs")) {
|
||||||
|
t.Fatalf("run output path = %q, want run-local path", req.OutputMergedTranscriptPath)
|
||||||
|
}
|
||||||
|
if len(result.Outputs) == 0 {
|
||||||
|
t.Fatalf("outputs = %#v, want promoted outputs", result.Outputs)
|
||||||
|
}
|
||||||
|
if strings.Contains(result.Outputs[0].AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) {
|
||||||
|
t.Fatalf("promoted output path = %q, want canonical session path", result.Outputs[0].AbsolutePath)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestMergeStageResolvesSessionRelativeManifestOutputs(t *testing.T) {
|
func TestMergeStageResolvesSessionRelativeManifestOutputs(t *testing.T) {
|
||||||
env, m := setupMergeEnv(t)
|
env, m := setupMergeEnv(t)
|
||||||
paths := sessionPathsForEnv(env, m.SessionID)
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
|
|||||||
@@ -211,6 +211,33 @@ func TestNormalizeStageReportEnabledFailsWhenReportMissing(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestNormalizeStageUsesRunLocalPathsAndPromotesCanonical(t *testing.T) {
|
||||||
|
env, m, ser := setupNormalizeEnv(t)
|
||||||
|
env.Config.Session.Campaign = "sample-campaign"
|
||||||
|
m.Campaign = "sample-campaign"
|
||||||
|
m.RunID = "20260518T010203Z-abcdef12"
|
||||||
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
|
writeFile(t, filepath.Join(paths.TranscriptsDir, "processed.json"), `{"segments":[{"id":1}]}`)
|
||||||
|
|
||||||
|
result, err := (normalizeStage{}).Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("normalize.Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(ser.NormalizeRequests) != 1 {
|
||||||
|
t.Fatalf("normalize requests = %d, want 1", len(ser.NormalizeRequests))
|
||||||
|
}
|
||||||
|
req := ser.NormalizeRequests[0]
|
||||||
|
if !strings.Contains(req.OutputNormalizedPath, filepath.Join("runs", m.RunID, "normalize", "outputs")) {
|
||||||
|
t.Fatalf("run output path = %q, want run-local path", req.OutputNormalizedPath)
|
||||||
|
}
|
||||||
|
if len(result.Outputs) == 0 {
|
||||||
|
t.Fatalf("outputs = %#v, want promoted outputs", result.Outputs)
|
||||||
|
}
|
||||||
|
if strings.Contains(result.Outputs[0].AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) {
|
||||||
|
t.Fatalf("promoted output path = %q, want canonical session path", result.Outputs[0].AbsolutePath)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func setupNormalizeEnv(t *testing.T) (*Env, *manifest.Manifest, *seriatim.FakeRunner) {
|
func setupNormalizeEnv(t *testing.T) (*Env, *manifest.Manifest, *seriatim.FakeRunner) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
workspace := t.TempDir()
|
workspace := t.TempDir()
|
||||||
|
|||||||
@@ -64,6 +64,7 @@ func TestStagesReturnExpectedMetadata(t *testing.T) {
|
|||||||
},
|
},
|
||||||
Session: &config.SessionConfig{
|
Session: &config.SessionConfig{
|
||||||
SessionID: "2026-05-03",
|
SessionID: "2026-05-03",
|
||||||
|
Campaign: "sample-campaign",
|
||||||
Inputs: config.SessionInputsConfig{
|
Inputs: config.SessionInputsConfig{
|
||||||
AudioDir: "./audio",
|
AudioDir: "./audio",
|
||||||
SpeakersFile: "./speakers.yml",
|
SpeakersFile: "./speakers.yml",
|
||||||
@@ -88,7 +89,7 @@ func TestStagesReturnExpectedMetadata(t *testing.T) {
|
|||||||
}
|
}
|
||||||
m.RunID = "20260516T000000Z-abcdef12"
|
m.RunID = "20260516T000000Z-abcdef12"
|
||||||
m.Campaign = "sample-campaign"
|
m.Campaign = "sample-campaign"
|
||||||
m.LocalWorkDir = filepath.Join(root, "work", "sample-campaign", "2026-05-03", m.RunID)
|
m.LocalWorkDir = artifacts.SessionRunRootForCampaign(root, "sample-campaign", "2026-05-03", m.RunID)
|
||||||
m.S3RunPrefix = "dnd/campaigns/sample-campaign/sessions/2026-05-03/runs/" + m.RunID + "/"
|
m.S3RunPrefix = "dnd/campaigns/sample-campaign/sessions/2026-05-03/runs/" + m.RunID + "/"
|
||||||
m.S3Bucket = "my-dnd-archive"
|
m.S3Bucket = "my-dnd-archive"
|
||||||
if err := os.MkdirAll(filepath.Join(m.LocalWorkDir, "inputs"), 0o755); err != nil {
|
if err := os.MkdirAll(filepath.Join(m.LocalWorkDir, "inputs"), 0o755); err != nil {
|
||||||
|
|||||||
@@ -238,6 +238,36 @@ func TestPolishStageFailsWhenReportInvalid(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestPolishStageUsesRunLocalPathsAndPromotesCanonical(t *testing.T) {
|
||||||
|
env, m := setupPolishEnv(t)
|
||||||
|
env.Config.Session.Campaign = "sample-campaign"
|
||||||
|
m.Campaign = "sample-campaign"
|
||||||
|
m.RunID = "20260518T010203Z-abcdef12"
|
||||||
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
|
writeFile(t, filepath.Join(paths.TranscriptsDir, "merged.json"), `{"segments":[]}`)
|
||||||
|
writeFile(t, filepath.Join(paths.InputsDir, "glossary.yml"), "terms: []\n")
|
||||||
|
|
||||||
|
fake := &audita.FakeRunner{}
|
||||||
|
env.Audita = fake
|
||||||
|
result, err := (polishStage{}).Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("polish.Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(fake.Requests) != 1 {
|
||||||
|
t.Fatalf("requests = %d, want 1", len(fake.Requests))
|
||||||
|
}
|
||||||
|
req := fake.Requests[0]
|
||||||
|
if !strings.Contains(req.OutputProcessedPath, filepath.Join("runs", m.RunID, "polish", "outputs")) {
|
||||||
|
t.Fatalf("run output path = %q, want run-local path", req.OutputProcessedPath)
|
||||||
|
}
|
||||||
|
if len(result.Outputs) == 0 {
|
||||||
|
t.Fatalf("outputs = %#v, want promoted outputs", result.Outputs)
|
||||||
|
}
|
||||||
|
if strings.Contains(result.Outputs[0].AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) {
|
||||||
|
t.Fatalf("promoted output path = %q, want canonical session path", result.Outputs[0].AbsolutePath)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func setupPolishEnv(t *testing.T) (*Env, *manifest.Manifest) {
|
func setupPolishEnv(t *testing.T) (*Env, *manifest.Manifest) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
workspace := t.TempDir()
|
workspace := t.TempDir()
|
||||||
|
|||||||
@@ -366,7 +366,11 @@ func pathsWorkDirForManifest(env *Env, m *manifest.Manifest, sessionID string) s
|
|||||||
runID = strings.TrimSpace(m.RunID)
|
runID = strings.TrimSpace(m.RunID)
|
||||||
}
|
}
|
||||||
if runID != "" {
|
if runID != "" {
|
||||||
return artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, env.Config.Session.Campaign, sessionID, runID)
|
campaign := strings.TrimSpace(env.Config.Session.Campaign)
|
||||||
|
if campaign != "" {
|
||||||
|
return artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID)
|
||||||
|
}
|
||||||
|
return artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID)
|
||||||
}
|
}
|
||||||
campaign := strings.TrimSpace(env.Config.Session.Campaign)
|
campaign := strings.TrimSpace(env.Config.Session.Campaign)
|
||||||
if campaign == "" {
|
if campaign == "" {
|
||||||
|
|||||||
@@ -191,6 +191,36 @@ func TestTranscribeStageInvalidJSONFails(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestTranscribeStageUsesRunLocalOutputAndPromotesCanonical(t *testing.T) {
|
||||||
|
env, m := setupTranscribeEnv(t, []string{"alice.flac"})
|
||||||
|
env.Config.Session.Campaign = "sample-campaign"
|
||||||
|
m.Campaign = "sample-campaign"
|
||||||
|
m.RunID = "20260518T010203Z-abcdef12"
|
||||||
|
fake := &whisperx.FakeClient{}
|
||||||
|
env.WhisperX = fake
|
||||||
|
|
||||||
|
if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil {
|
||||||
|
t.Fatalf("prepare.Run() error = %v", err)
|
||||||
|
}
|
||||||
|
result, err := (transcribeStage{}).Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("transcribe.Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(fake.Requests) != 1 {
|
||||||
|
t.Fatalf("requests = %d, want 1", len(fake.Requests))
|
||||||
|
}
|
||||||
|
runOut := fake.Requests[0].OutputRawTranscriptPath
|
||||||
|
if !strings.Contains(runOut, filepath.Join("runs", m.RunID, "transcribe", "outputs")) {
|
||||||
|
t.Fatalf("run-local output path = %q, want runs/{run_id}/transcribe/outputs path", runOut)
|
||||||
|
}
|
||||||
|
if len(result.Outputs) != 1 {
|
||||||
|
t.Fatalf("outputs = %#v, want one output", result.Outputs)
|
||||||
|
}
|
||||||
|
if strings.Contains(result.Outputs[0].AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) {
|
||||||
|
t.Fatalf("promoted output path = %q, want canonical session path", result.Outputs[0].AbsolutePath)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func setupTranscribeEnv(t *testing.T, audioFiles []string) (*Env, *manifest.Manifest) {
|
func setupTranscribeEnv(t *testing.T, audioFiles []string) (*Env, *manifest.Manifest) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
|
|||||||
@@ -298,6 +298,39 @@ func TestTrimStageDisabledCopiesNormalizedTranscript(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestTrimStageUsesRunLocalPathsAndPromotesCanonical(t *testing.T) {
|
||||||
|
env, m, scr, ser := setupTrimEnv(t)
|
||||||
|
env.Config.Session.Campaign = "sample-campaign"
|
||||||
|
m.Campaign = "sample-campaign"
|
||||||
|
m.RunID = "20260518T010203Z-abcdef12"
|
||||||
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
|
normalized := filepath.Join(paths.TranscriptsDir, "normalized.json")
|
||||||
|
writeFile(t, normalized, `{"segments":[{"id":10},{"id":11}]}`)
|
||||||
|
scr.BoundsBody = `{"trim_action":"trim","start_segment_id":10,"end_segment_id":11,"warnings":[]}`
|
||||||
|
|
||||||
|
result, err := (trimStage{}).Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("trim.Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(scr.RunRequests) != 1 || len(ser.TrimRequests) != 1 {
|
||||||
|
t.Fatalf("scriptorium run=%d seriatim trim=%d, want 1/1", len(scr.RunRequests), len(ser.TrimRequests))
|
||||||
|
}
|
||||||
|
if !strings.Contains(scr.RunRequests[0].OutputPath, filepath.Join("runs", m.RunID, "trim", "outputs")) {
|
||||||
|
t.Fatalf("bounds run output path = %q, want run-local path", scr.RunRequests[0].OutputPath)
|
||||||
|
}
|
||||||
|
if !strings.Contains(ser.TrimRequests[0].OutputTrimmedPath, filepath.Join("runs", m.RunID, "trim", "outputs")) {
|
||||||
|
t.Fatalf("trim output path = %q, want run-local path", ser.TrimRequests[0].OutputTrimmedPath)
|
||||||
|
}
|
||||||
|
if len(result.Outputs) < 2 {
|
||||||
|
t.Fatalf("outputs = %#v, want promoted trimmed+bounds outputs", result.Outputs)
|
||||||
|
}
|
||||||
|
for _, out := range result.Outputs {
|
||||||
|
if strings.Contains(out.AbsolutePath, string(filepath.Separator)+"runs"+string(filepath.Separator)) {
|
||||||
|
t.Fatalf("promoted output path = %q, want canonical session path", out.AbsolutePath)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
type boundsScriptoriumRunner struct {
|
type boundsScriptoriumRunner struct {
|
||||||
RunRequests []scriptorium.RunArtifactRequest
|
RunRequests []scriptorium.RunArtifactRequest
|
||||||
RenderRequests []scriptorium.RenderArtifactRequest
|
RenderRequests []scriptorium.RenderArtifactRequest
|
||||||
|
|||||||
Reference in New Issue
Block a user