diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index ddf4a5f..0dc47a0 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -1,6 +1,6 @@ # Implementation Plan: Speaker Checkpoints and Replayable Uploads -Status: Implementation in progress; Stages 1-3 complete. +Status: Implementation in progress; Stages 1-4 complete. ## Scope and Instructions @@ -241,7 +241,18 @@ ordinary, concurrent, canceled, and forced execution. ## Stage 4: Verify Runner, Merge, and Durable Lifecycle Integration -Status: Pending. +Status: Complete. + +Completion marker (2026-09-22): Added real-persistence application coverage for +cross-invocation retry, interrupted running state, failed aggregate saves, +force/reset persistence, downstream invalidation, and one-time legacy upgrade. +Versioned merge input discovery now requires aggregate success and an exact +match between current speaker evidence and declared aggregate outputs, while +legacy/manual directory fallback remains available. Restore coverage preserves +session-relative speaker paths, leaves unpublished raw transcripts absent, and +canonical transcripts remain reusable after run-directory cleanup. Verified +with `go test ./internal/app ./internal/stage ./internal/manifest`, +`go test -race ./internal/app`, and `git diff --check`. Preparation: `internal/app/runner.go`, `runner_runtime.go`, `analyze_checkpoint_recovery_test.go`, `runner_terminal_failure_test.go`, diff --git a/internal/app/restore_execution_test.go b/internal/app/restore_execution_test.go index 453c2c2..cd381de 100644 --- a/internal/app/restore_execution_test.go +++ b/internal/app/restore_execution_test.go @@ -82,6 +82,28 @@ func TestExecuteRestoreRoundTripsPublishedExtractionAndManifestMetadata(t *testi remoteManifest.RunID = "20260519T010203Z-a1b2c3d4" producerRoot := "/prior/workspace/work/sample-campaign/2026-05-03" remoteManifest.LocalWorkDir = filepath.Join(producerRoot, "runs", remoteManifest.RunID) + rawRelative := "transcripts/raw/alice.json" + rawOutput := manifest.ArtifactRecord{ + Kind: "transcript_raw", LocalPath: filepath.Join(producerRoot, filepath.FromSlash(rawRelative)), + ProducerRunID: remoteManifest.RunID, Checksum: strings.Repeat("b", 64), + } + remoteManifest.MarkStageSucceeded("transcribe", time.Now().UTC(), []manifest.ArtifactRecord{rawOutput}) + remoteManifest.Stages["transcribe"].TranscribeStateVersion = manifest.TranscribeStateContractVersion + remoteManifest.Stages["transcribe"].TranscribeSpeakers = map[string]manifest.TranscribeSpeakerRecord{ + "alice": { + FingerprintVersion: manifest.TranscribeFingerprintContractVersion, + Fingerprint: strings.Repeat("c", 64), + AudioPath: "audio/alice.flac", + AudioSHA256: strings.Repeat("a", 64), + AudioSize: 12, + Output: manifest.ArtifactRecord{ + Kind: "transcript_raw", LocalPath: rawRelative, + ProducerRunID: remoteManifest.RunID, Checksum: strings.Repeat("b", 64), + }, + OutputSize: 15, + ProducerRunID: remoteManifest.RunID, + }, + } remoteManifest.Stages["extract"] = &manifest.StageRecord{ Name: "extract", Status: manifest.StatusSucceeded, Outputs: []manifest.ArtifactRecord{{ @@ -121,6 +143,13 @@ func TestExecuteRestoreRoundTripsPublishedExtractionAndManifestMetadata(t *testi if want := filepath.Join(sessionRoot, "artifacts", "encounters.json"); lane.LocalPath != want { t.Fatalf("rebased extraction path = %q, want %q", lane.LocalPath, want) } + speaker := restored.Stages["transcribe"].TranscribeSpeakers["alice"] + if speaker.AudioPath != "audio/alice.flac" || speaker.Output.LocalPath != rawRelative { + t.Fatalf("restored speaker paths = %#v, want session-relative paths", speaker) + } + if _, err := os.Stat(filepath.Join(sessionRoot, filepath.FromSlash(rawRelative))); !os.IsNotExist(err) { + t.Fatalf("unpublished raw transcript should not be restored; stat err=%v", err) + } } func TestExecuteRestoreIncludeAudioRestoresAudio(t *testing.T) { diff --git a/internal/app/runner_test_helpers_test.go b/internal/app/runner_test_helpers_test.go index 09deb46..375bd3b 100644 --- a/internal/app/runner_test_helpers_test.go +++ b/internal/app/runner_test_helpers_test.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "os" "path/filepath" + "slices" "strings" "testing" @@ -126,6 +127,14 @@ func seedCurrentTranscribeEvidence( record := m.Stages["transcribe"] record.TranscribeStateVersion = manifest.TranscribeStateContractVersion record.TranscribeSpeakers = records + if len(record.Outputs) == 0 { + for _, speakerRecord := range records { + record.Outputs = append(record.Outputs, speakerRecord.Output) + } + slices.SortFunc(record.Outputs, func(a, b manifest.ArtifactRecord) int { + return strings.Compare(a.LocalPath, b.LocalPath) + }) + } } func loadConfigForSemanticEvidence(t *testing.T, pipelinePath, campaignPath, sessionPath string) *config.Config { diff --git a/internal/app/transcribe_checkpoint_recovery_test.go b/internal/app/transcribe_checkpoint_recovery_test.go new file mode 100644 index 0000000..8a9cf9b --- /dev/null +++ b/internal/app/transcribe_checkpoint_recovery_test.go @@ -0,0 +1,338 @@ +package app + +import ( + "context" + "errors" + "os" + "path/filepath" + "slices" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/stage" +) + +func TestExecuteStagesRetriesOnlyUncommittedTranscribeSpeakers(t *testing.T) { + cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"a", "b", "c", "d"}) + concurrency := 1 + cfg.Pipeline.WhisperX.Concurrency = &concurrency + cause := errors.New("speaker c failed") + failing := &whisperx.FakeClient{TranscribeFn: func( + _ context.Context, + req whisperx.TranscribeRequest, + ) (whisperx.TranscribeResult, error) { + if req.SpeakerID == "c" { + return whisperx.TranscribeResult{}, cause + } + if req.SpeakerID == "d" { + t.Fatalf("speaker d was dispatched after c failed") + } + return writeTranscribeRecoveryResult(req) + }} + transcribe := mustSelectStage(t, "transcribe") + + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: baseStore, WhisperX: failing}, + }); !errors.Is(err, cause) { + t.Fatalf("failed executeStages() error = %v, want %v", err, cause) + } + failed := loadTranscribeRecoveryManifest(t, baseStore, cfg) + if failed.Stages["transcribe"].Status != manifest.StatusFailed { + t.Fatalf("transcribe status = %q, want failed", failed.Stages["transcribe"].Status) + } + if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"a", "b"}) { + t.Fatalf("checkpointed speakers = %#v, want a and b", got) + } + if failed.Stages["prepare"].Status != manifest.StatusSucceeded { + t.Fatalf("prepare status = %q, want unchanged success", failed.Stages["prepare"].Status) + } + if failed.Stages["merge"].Status != manifest.StatusStale { + t.Fatalf("merge status = %q, want stale after transcribe failure", failed.Stages["merge"].Status) + } + + retry := &whisperx.FakeClient{} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: baseStore, WhisperX: retry}, + }); err != nil { + t.Fatalf("retry executeStages() error = %v", err) + } + if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"c", "d"}) { + t.Fatalf("retry speakers = %#v, want c and d", got) + } + completed := loadTranscribeRecoveryManifest(t, baseStore, cfg) + if completed.Stages["transcribe"].Status != manifest.StatusSucceeded { + t.Fatalf("transcribe status = %q, want succeeded", completed.Stages["transcribe"].Status) + } + if got := transcribeSpeakerKeys(completed); !slices.Equal(got, []string{"a", "b", "c", "d"}) { + t.Fatalf("completed speakers = %#v", got) + } +} + +func TestExecuteStagesRebuildsTranscribeAggregateAfterFinalSaveFailure(t *testing.T) { + cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"}) + store := &transcribeAggregateSaveFailingStore{ + Store: baseStore, + Err: errors.New("aggregate save failed"), + } + fake := &whisperx.FakeClient{} + transcribe := mustSelectStage(t, "transcribe") + + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: store, WhisperX: fake}, + }); !errors.Is(err, store.Err) { + t.Fatalf("executeStages() error = %v, want aggregate save failure", err) + } + failed := loadTranscribeRecoveryManifest(t, baseStore, cfg) + if failed.Stages["transcribe"].Status != manifest.StatusFailed { + t.Fatalf("transcribe status = %q, want failed", failed.Stages["transcribe"].Status) + } + if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"alice", "bob"}) { + t.Fatalf("checkpointed speakers = %#v, want both speakers", got) + } + + retry := &whisperx.FakeClient{} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: baseStore, WhisperX: retry}, + }); err != nil { + t.Fatalf("retry executeStages() error = %v", err) + } + if got := len(retry.RequestsSnapshot()); got != 0 { + t.Fatalf("retry adapter requests = %d, want aggregate-only rebuild", got) + } + completed := loadTranscribeRecoveryManifest(t, baseStore, cfg) + if completed.Stages["transcribe"].Status != manifest.StatusSucceeded || len(completed.Stages["transcribe"].Outputs) != 2 { + t.Fatalf("rebuilt aggregate = %#v", completed.Stages["transcribe"]) + } +} + +func TestExecuteStagesResumesInterruptedRunningTranscribeState(t *testing.T) { + cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"}) + concurrency := 1 + cfg.Pipeline.WhisperX.Concurrency = &concurrency + stageErr := errors.New("speaker b failed") + terminalErr := errors.New("terminal save interrupted") + store := &interruptAfterTranscribeCheckpointStore{Store: baseStore, Err: terminalErr} + fake := &whisperx.FakeClient{TranscribeFn: func( + _ context.Context, + req whisperx.TranscribeRequest, + ) (whisperx.TranscribeResult, error) { + if req.SpeakerID == "bob" { + return whisperx.TranscribeResult{}, stageErr + } + return writeTranscribeRecoveryResult(req) + }} + transcribe := mustSelectStage(t, "transcribe") + + _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: store, WhisperX: fake}, + }) + if !errors.Is(err, stageErr) || !errors.Is(err, terminalErr) { + t.Fatalf("executeStages() error = %v, want stage and terminal failures", err) + } + interrupted := loadTranscribeRecoveryManifest(t, baseStore, cfg) + if interrupted.Stages["transcribe"].Status != manifest.StatusRunning { + t.Fatalf("transcribe status = %q, want durable running state", interrupted.Stages["transcribe"].Status) + } + if got := transcribeSpeakerKeys(interrupted); !slices.Equal(got, []string{"alice"}) { + t.Fatalf("checkpointed speakers = %#v, want alice", got) + } + + retry := &whisperx.FakeClient{} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: baseStore, WhisperX: retry}, + }); err != nil { + t.Fatalf("retry executeStages() error = %v", err) + } + if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"bob"}) { + t.Fatalf("retry speakers = %#v, want bob", got) + } +} + +func TestExecuteStagesPersistsForcedTranscribeResetAcrossRetry(t *testing.T) { + cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"}) + concurrency := 1 + cfg.Pipeline.WhisperX.Concurrency = &concurrency + transcribe := mustSelectStage(t, "transcribe") + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: store, WhisperX: &whisperx.FakeClient{}}, + }); err != nil { + t.Fatalf("baseline executeStages() error = %v", err) + } + + cause := errors.New("forced bob failed") + forced := &whisperx.FakeClient{TranscribeFn: func( + _ context.Context, + req whisperx.TranscribeRequest, + ) (whisperx.TranscribeResult, error) { + if req.SpeakerID == "bob" { + return whisperx.TranscribeResult{}, cause + } + return writeTranscribeRecoveryResult(req) + }} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Force: true, + Env: &Env{ManifestStore: store, WhisperX: forced}, + }); !errors.Is(err, cause) { + t.Fatalf("forced executeStages() error = %v, want %v", err, cause) + } + failed := loadTranscribeRecoveryManifest(t, store, cfg) + if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"alice"}) { + t.Fatalf("speakers after forced failure = %#v, want only replacement alice", got) + } + + retry := &whisperx.FakeClient{} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: store, WhisperX: retry}, + }); err != nil { + t.Fatalf("retry executeStages() error = %v", err) + } + if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"bob"}) { + t.Fatalf("retry speakers = %#v, want bob", got) + } +} + +func TestExecuteStagesUpgradesLegacyTranscribeSuccessOnce(t *testing.T) { + cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice"}) + m := loadTranscribeRecoveryManifest(t, store, cfg) + m.MarkStageSucceeded("transcribe", time.Now().UTC(), []manifest.ArtifactRecord{{ + Kind: "transcript_raw", LocalPath: "transcripts/raw/alice.json", + }}) + if err := store.Save(context.Background(), manifestPathFor(cfg), m); err != nil { + t.Fatalf("save legacy transcribe success: %v", err) + } + transcribe := mustSelectStage(t, "transcribe") + upgrade := &whisperx.FakeClient{} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: store, WhisperX: upgrade}, + }); err != nil { + t.Fatalf("upgrade executeStages() error = %v", err) + } + if got := requestedTranscribeSpeakers(upgrade); !slices.Equal(got, []string{"alice"}) { + t.Fatalf("upgrade speakers = %#v, want alice", got) + } + upgraded := loadTranscribeRecoveryManifest(t, store, cfg) + if upgraded.Stages["transcribe"].TranscribeStateVersion != manifest.TranscribeStateContractVersion { + t.Fatalf("transcribe state version = %d, want current", upgraded.Stages["transcribe"].TranscribeStateVersion) + } + + reused := &whisperx.FakeClient{} + summary, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: store, WhisperX: reused}, + }) + if err != nil { + t.Fatalf("reuse executeStages() error = %v", err) + } + if len(summary.Skipped) != 1 || len(reused.RequestsSnapshot()) != 0 { + t.Fatalf("reuse summary = %#v, requests = %#v, want skipped without HTTP", summary, reused.RequestsSnapshot()) + } +} + +type transcribeAggregateSaveFailingStore struct { + manifest.Store + Err error + Failed bool +} + +func (store *transcribeAggregateSaveFailingStore) Save( + ctx context.Context, + path string, + m *manifest.Manifest, +) error { + record := m.Stages["transcribe"] + if !store.Failed && record != nil && record.Status == manifest.StatusSucceeded && + record.TranscribeStateVersion == manifest.TranscribeStateContractVersion { + store.Failed = true + return store.Err + } + return store.Store.Save(ctx, path, m) +} + +type interruptAfterTranscribeCheckpointStore struct { + manifest.Store + Err error + Interrupted bool +} + +func (store *interruptAfterTranscribeCheckpointStore) Save( + ctx context.Context, + path string, + m *manifest.Manifest, +) error { + if store.Interrupted { + return store.Err + } + record := m.Stages["transcribe"] + err := store.Store.Save(ctx, path, m) + if err == nil && record != nil && record.Status == manifest.StatusRunning && len(record.TranscribeSpeakers) == 1 { + store.Interrupted = true + } + return err +} + +func setupTranscribeRecoveryFixture(t *testing.T, speakers []string) (*config.Config, *manifest.LocalStore) { + t.Helper() + cfg := testConfig(t) + store := &manifest.LocalStore{} + artifactStore := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) + paths, err := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) + if err != nil { + t.Fatalf("prepare session layout: %v", err) + } + m := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + m.Campaign = cfg.Session.Campaign + m.Inputs = make([]manifest.InputRecord, 0, len(speakers)) + for _, speaker := range speakers { + relative := filepath.ToSlash(filepath.Join("audio", speaker+".flac")) + if err := os.WriteFile(filepath.Join(paths.AudioDir, speaker+".flac"), []byte("audio-"+speaker), 0o644); err != nil { + t.Fatalf("write audio fixture: %v", err) + } + m.Inputs = append(m.Inputs, manifest.InputRecord{Kind: "audio", Path: relative}) + } + m.MarkStageSucceeded("prepare", time.Now().UTC(), nil) + m.MarkStageSucceeded("merge", time.Now().UTC(), nil) + if err := store.Save(context.Background(), paths.ManifestPath, m); err != nil { + t.Fatalf("save recovery fixture: %v", err) + } + return cfg, store +} + +func writeTranscribeRecoveryResult(req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) { + if err := os.MkdirAll(filepath.Dir(req.OutputRawTranscriptPath), 0o755); err != nil { + return whisperx.TranscribeResult{}, err + } + if err := os.WriteFile(req.OutputRawTranscriptPath, []byte(`{"segments":[]}`), 0o644); err != nil { + return whisperx.TranscribeResult{}, err + } + return whisperx.TranscribeResult{OutputRawTranscriptPath: req.OutputRawTranscriptPath}, nil +} + +func loadTranscribeRecoveryManifest(t *testing.T, store manifest.Store, cfg *config.Config) *manifest.Manifest { + t.Helper() + m, err := store.Load(context.Background(), manifestPathFor(cfg)) + if err != nil { + t.Fatalf("load transcribe recovery manifest: %v", err) + } + return m +} + +func transcribeSpeakerKeys(m *manifest.Manifest) []string { + keys := make([]string, 0, len(m.Stages["transcribe"].TranscribeSpeakers)) + for key := range m.Stages["transcribe"].TranscribeSpeakers { + keys = append(keys, key) + } + slices.Sort(keys) + return keys +} + +func requestedTranscribeSpeakers(fake *whisperx.FakeClient) []string { + requests := fake.RequestsSnapshot() + speakers := make([]string, 0, len(requests)) + for _, request := range requests { + speakers = append(speakers, request.SpeakerID) + } + return speakers +} diff --git a/internal/stage/merge.go b/internal/stage/merge.go index e1512b6..31f03bc 100644 --- a/internal/stage/merge.go +++ b/internal/stage/merge.go @@ -306,6 +306,12 @@ func normalizeMergeInputs( } func discoverRawTranscripts(m *manifest.Manifest, paths artifacts.SessionPaths) ([]string, error) { + if record, versioned, err := transcribeManifestState(m); err != nil { + return nil, err + } else if versioned { + return discoverVersionedRawTranscripts(record, paths) + } + fromManifest := make([]string, 0) if m != nil && m.Stages != nil { if tr := m.Stages["transcribe"]; tr != nil { @@ -350,6 +356,64 @@ func discoverRawTranscripts(m *manifest.Manifest, paths artifacts.SessionPaths) return dedupeAndSortPaths(out), nil } +func discoverVersionedRawTranscripts( + record *manifest.StageRecord, + paths artifacts.SessionPaths, +) ([]string, error) { + if record.Status != manifest.StatusSucceeded { + return nil, fmt.Errorf("versioned transcribe state is not aggregate-successful (status %q)", record.Status) + } + if err := manifest.ValidateTranscribeSpeakerCollection( + record.TranscribeStateVersion, + record.TranscribeSpeakers, + ); err != nil { + return nil, fmt.Errorf("validate versioned transcribe state: %w", err) + } + if len(record.TranscribeSpeakers) == 0 { + return nil, fmt.Errorf("versioned transcribe state has no completed speakers") + } + + expected := make(map[string]string, len(record.TranscribeSpeakers)) + for speaker, speakerRecord := range record.TranscribeSpeakers { + path := filepath.Clean(artifacts.ResolveSessionLocalPathForRead(paths, speakerRecord.Output.LocalPath)) + if other, exists := expected[path]; exists { + return nil, fmt.Errorf("versioned transcribe speakers %q and %q declare the same output", other, speaker) + } + expected[path] = speaker + } + inputs := make([]string, 0, len(expected)) + for _, output := range record.Outputs { + if output.Kind != "transcript_raw" { + continue + } + path := strings.TrimSpace(output.LocalPath) + if path == "" { + return nil, fmt.Errorf("versioned transcribe aggregate output has an empty local path") + } + path = filepath.Clean(artifacts.ResolveSessionLocalPathForRead(paths, path)) + speaker, declared := expected[path] + if !declared { + return nil, fmt.Errorf("versioned transcribe aggregate output %q is not declared by current speaker state", output.LocalPath) + } + if speaker == "" { + return nil, fmt.Errorf("versioned transcribe aggregate output %q is duplicated", output.LocalPath) + } + expected[path] = "" + inputs = append(inputs, path) + } + missing := make([]string, 0) + for _, speaker := range expected { + if speaker != "" { + missing = append(missing, speaker) + } + } + if len(missing) > 0 { + sort.Strings(missing) + return nil, fmt.Errorf("versioned transcribe aggregate outputs are incomplete for speakers: %s", strings.Join(missing, ", ")) + } + return dedupeAndSortPaths(inputs), nil +} + func dedupeAndSortPaths(paths []string) []string { if len(paths) == 0 { return nil diff --git a/internal/stage/merge_test.go b/internal/stage/merge_test.go index b011f81..0a5ce9a 100644 --- a/internal/stage/merge_test.go +++ b/internal/stage/merge_test.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "path/filepath" + "slices" "strings" "testing" "time" @@ -191,6 +192,70 @@ func TestMergeStageFallsBackToRawDirectoryWhenTranscribeOutputsMissing(t *testin } } +func TestMergeStageRejectsPartialVersionedTranscribeState(t *testing.T) { + env, m := setupMergeEnv(t) + paths := sessionPathsForEnv(env, m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsRawDir, "alice.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.TranscriptsRawDir, "uncommitted.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") + writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") + setVersionedTranscribeMergeState(t, m, manifest.StatusFailed, map[string]string{ + "alice": "transcripts/raw/alice.json", + }) + fake := &seriatim.FakeRunner{} + env.Seriatim = fake + + _, err := (mergeStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "not aggregate-successful") { + t.Fatalf("merge.Run() error = %v, want incomplete aggregate rejection", err) + } + if len(fake.Requests) != 0 || len(fake.NormalizeRequests) != 0 { + t.Fatalf("adapter requests = %#v / %#v, want none", fake.Requests, fake.NormalizeRequests) + } +} + +func TestMergeStageUsesOnlyCurrentVersionedAggregateOutputs(t *testing.T) { + env, m := setupMergeEnv(t) + paths := sessionPathsForEnv(env, m.SessionID) + alice := filepath.Join(paths.TranscriptsRawDir, "alice.json") + writeFile(t, alice, `{"segments":[]}`) + writeFile(t, filepath.Join(paths.TranscriptsRawDir, "removed.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") + writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") + setVersionedTranscribeMergeState(t, m, manifest.StatusSucceeded, map[string]string{ + "alice": "transcripts/raw/alice.json", + }) + fake := &seriatim.FakeRunner{} + env.Seriatim = fake + + if _, err := (mergeStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("merge.Run() error = %v", err) + } + if len(fake.NormalizeRequests) != 1 || fake.NormalizeRequests[0].InputTranscriptPath != alice { + t.Fatalf("normalize requests = %#v, want only current alice", fake.NormalizeRequests) + } +} + +func TestMergeStageRejectsIncompleteVersionedAggregateOutputs(t *testing.T) { + env, m := setupMergeEnv(t) + paths := sessionPathsForEnv(env, m.SessionID) + writeFile(t, filepath.Join(paths.TranscriptsRawDir, "alice.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.TranscriptsRawDir, "bob.json"), `{"segments":[]}`) + writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n") + writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n") + setVersionedTranscribeMergeState(t, m, manifest.StatusSucceeded, map[string]string{ + "alice": "transcripts/raw/alice.json", + "bob": "transcripts/raw/bob.json", + }) + m.Stages["transcribe"].Outputs = m.Stages["transcribe"].Outputs[:1] + env.Seriatim = &seriatim.FakeRunner{} + + _, err := (mergeStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "incomplete for speakers: bob") { + t.Fatalf("merge.Run() error = %v, want incomplete aggregate output rejection", err) + } +} + func TestMergeStageResolvesWorkspaceQualifiedManifestOutputsWithoutDuplication(t *testing.T) { env, m := setupMergeEnvWithRelativeWorkspaceRoot(t) paths := sessionPathsForEnv(env, m.SessionID) @@ -414,3 +479,44 @@ func setupMergeEnvWithWorkspace(t *testing.T, workspace string) (*Env, *manifest ArtifactStore: store, }, manifest.New("2026-05-03", time.Now().UTC()) } + +func setVersionedTranscribeMergeState( + t *testing.T, + m *manifest.Manifest, + status manifest.StageStatus, + outputs map[string]string, +) { + t.Helper() + if m.RunID == "" { + m.RunID = "20260503T010203Z-a1b2c3d4" + } + aggregate := make([]manifest.ArtifactRecord, 0, len(outputs)) + records := make(map[string]manifest.TranscribeSpeakerRecord, len(outputs)) + for speaker, outputPath := range outputs { + output := manifest.ArtifactRecord{ + Kind: "transcript_raw", LocalPath: outputPath, + ProducerRunID: m.RunID, Checksum: strings.Repeat("b", 64), + } + aggregate = append(aggregate, output) + records[speaker] = manifest.TranscribeSpeakerRecord{ + FingerprintVersion: manifest.TranscribeFingerprintContractVersion, + Fingerprint: strings.Repeat("c", 64), + AudioPath: "audio/" + speaker + ".flac", + AudioSHA256: strings.Repeat("a", 64), + AudioSize: 1, + Output: output, + OutputSize: 1, + ProducerRunID: m.RunID, + } + } + slices.SortFunc(aggregate, func(a, b manifest.ArtifactRecord) int { + return strings.Compare(a.LocalPath, b.LocalPath) + }) + m.MarkStageSucceeded("transcribe", time.Now().UTC(), aggregate) + record := m.Stages["transcribe"] + record.TranscribeStateVersion = manifest.TranscribeStateContractVersion + record.TranscribeSpeakers = records + if status != manifest.StatusSucceeded { + m.MarkStageFailed("transcribe", time.Now().UTC(), "incomplete") + } +} diff --git a/internal/stage/transcribe_checkpoint_test.go b/internal/stage/transcribe_checkpoint_test.go index 7a9791a..7e3df62 100644 --- a/internal/stage/transcribe_checkpoint_test.go +++ b/internal/stage/transcribe_checkpoint_test.go @@ -271,6 +271,28 @@ func TestTranscribeRebuildsAggregateFromCurrentSpeakersWithoutRequests(t *testin } } +func TestTranscribeReuseDoesNotDependOnRunDirectory(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac"}) + env.WhisperX = &whisperx.FakeClient{} + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + if _, err := (transcribeStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("Run() error = %v", err) + } + paths := sessionPathsForEnv(env, m.SessionID) + if err := os.RemoveAll(paths.RunsDir); err != nil { + t.Fatalf("remove run directory: %v", err) + } + if _, err := os.Stat(paths.TranscriptsRawDir + "/alice.json"); err != nil { + t.Fatalf("canonical transcript missing after run cleanup: %v", err) + } + validation, err := (transcribeStage{}).ValidateResume(context.Background(), env, m) + if err != nil || !validation.Resumable { + t.Fatalf("ValidateResume() = %#v, %v, want canonical reuse", validation, err) + } +} + type failingTranscribeManifestStore struct { manifest.Store Saves atomic.Int64