diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index aa06cb9..ddf4a5f 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-2 complete. +Status: Implementation in progress; Stages 1-3 complete. ## Scope and Instructions @@ -186,7 +186,17 @@ work, without altering existing runtime behavior prematurely. ## Stage 3: Execute and Checkpoint Speakers Incrementally -Status: Pending. +Status: Complete. + +Completion marker (2026-09-21): Connected reconciliation to bounded execution, +durably checkpointed the starting and per-speaker collections through detached +manifest candidates, preserved committed work across failures and cancellation, +rebuilt deterministic aggregate outputs from reusable evidence, and added +speaker-aware resume validation. Updated production-style fixtures and covered +serialized concurrent saves, force/retry replacement, input rechecks, save +failures, partial recovery, and aggregate reconstruction. Verified with +`go test ./internal/stage ./internal/app`, `go test -race ./internal/stage`, and +`git diff --check`. Preparation: `internal/stage/transcribe.go`, `analyze_execution_state.go`, `run_local.go`, diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 74a3e52..981699e 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -1082,7 +1082,7 @@ func TestExecuteStagesReportsSkippedStage(t *testing.T) { t.Fatalf("Save manifest error = %v", err) } - summary, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[1]}, RunOptions{}) + summary, err := executeStages(context.Background(), cfg, []stage.Stage{resultStage{name: "transcribe"}}, RunOptions{}) if err != nil { t.Fatalf("executeStages() error = %v", err) } @@ -1302,7 +1302,7 @@ func TestExecuteStagesSkippedStagePreservesExistingOutputsProvenance(t *testing. t.Fatalf("Save manifest error = %v", err) } - summary, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[1]}, RunOptions{}) + summary, err := executeStages(context.Background(), cfg, []stage.Stage{resultStage{name: "transcribe"}}, RunOptions{}) if err != nil { t.Fatalf("executeStages() error = %v", err) } diff --git a/internal/app/runner_test_helpers_test.go b/internal/app/runner_test_helpers_test.go index 3dde60e..09deb46 100644 --- a/internal/app/runner_test_helpers_test.go +++ b/internal/app/runner_test_helpers_test.go @@ -2,10 +2,17 @@ package app import ( "context" + "crypto/sha256" + "encoding/hex" + "os" + "path/filepath" + "strings" "testing" + "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/pathsafe" "gitea.maximumdirect.net/eric/narratio/internal/stage" ) @@ -38,9 +45,89 @@ func seedCurrentSemanticEvidence(t *testing.T, cfg *config.Config, m *manifest.M t.Fatalf("fingerprint stage %q: %v", name, err) } record.SemanticConfig = &fingerprint + if name == "transcribe" { + seedCurrentTranscribeEvidence(t, cfg, m, fingerprint) + } } } +func seedCurrentTranscribeEvidence( + t *testing.T, + cfg *config.Config, + m *manifest.Manifest, + semantic manifest.SemanticConfigFingerprint, +) { + t.Helper() + store := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) + paths := store.SessionPathsFor(cfg.Session.Campaign, m.SessionID) + if strings.TrimSpace(m.RunID) == "" { + m.RunID = "20260503T010203Z-a1b2c3d4" + } + records := make(map[string]manifest.TranscribeSpeakerRecord) + for _, input := range m.Inputs { + if input.Kind != "audio" { + continue + } + audioPath := artifacts.ResolveSessionLocalPathForRead(paths, input.Path) + audioBytes, err := os.ReadFile(audioPath) + if err != nil { + t.Fatalf("read transcribe audio fixture %q: %v", audioPath, err) + } + audioRelative, err := pathsafe.SlashRelativeFromRoot(paths.Root, audioPath) + if err != nil { + t.Fatalf("relativize transcribe audio fixture %q: %v", audioPath, err) + } + speaker := strings.TrimSuffix(filepath.Base(audioPath), filepath.Ext(audioPath)) + outputRelative := filepath.ToSlash(filepath.Join("transcripts", "raw", speaker+".json")) + outputPath, err := pathsafe.JoinSlashRelativeUnderRoot(paths.Root, outputRelative) + if err != nil { + t.Fatalf("resolve transcribe output fixture: %v", err) + } + if _, err := os.Stat(outputPath); os.IsNotExist(err) { + mustWriteTestFile(t, outputPath, `{"segments":[]}`) + } else if err != nil { + t.Fatalf("inspect transcribe output fixture %q: %v", outputPath, err) + } + outputBytes, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read transcribe output fixture %q: %v", outputPath, err) + } + audioDigest := sha256.Sum256(audioBytes) + outputDigest := sha256.Sum256(outputBytes) + fingerprint, err := stage.FingerprintSemanticConfig( + manifest.TranscribeFingerprintContractVersion, + struct { + SpeakerID string `json:"speaker_id"` + AudioPath string `json:"audio_path"` + AudioSHA256 string `json:"audio_sha256"` + AudioSize int64 `json:"audio_size"` + SemanticConfigVersion int `json:"semantic_config_version"` + SemanticConfigDigest string `json:"semantic_config_digest"` + }{ + SpeakerID: speaker, AudioPath: audioRelative, + AudioSHA256: hex.EncodeToString(audioDigest[:]), AudioSize: int64(len(audioBytes)), + SemanticConfigVersion: semantic.Version, SemanticConfigDigest: semantic.Digest, + }, + ) + if err != nil { + t.Fatalf("fingerprint transcribe speaker fixture %q: %v", speaker, err) + } + output := manifest.ArtifactRecord{ + Kind: "transcript_raw", LocalPath: outputRelative, + ProducerRunID: m.RunID, Checksum: hex.EncodeToString(outputDigest[:]), + } + records[speaker] = manifest.TranscribeSpeakerRecord{ + FingerprintVersion: manifest.TranscribeFingerprintContractVersion, + Fingerprint: fingerprint.Digest, AudioPath: audioRelative, + AudioSHA256: hex.EncodeToString(audioDigest[:]), AudioSize: int64(len(audioBytes)), + Output: output, OutputSize: int64(len(outputBytes)), ProducerRunID: m.RunID, + } + } + record := m.Stages["transcribe"] + record.TranscribeStateVersion = manifest.TranscribeStateContractVersion + record.TranscribeSpeakers = records +} + func loadConfigForSemanticEvidence(t *testing.T, pipelinePath, campaignPath, sessionPath string) *config.Config { t.Helper() cfg, err := config.Load(pipelinePath, campaignPath, sessionPath) diff --git a/internal/stage/transcribe.go b/internal/stage/transcribe.go index 0e83361..a6bc549 100644 --- a/internal/stage/transcribe.go +++ b/internal/stage/transcribe.go @@ -8,7 +8,6 @@ import ( "path/filepath" "sort" "strings" - "sync" "golang.org/x/sync/errgroup" @@ -28,11 +27,14 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) if env.ArtifactStore == nil { return nil, fmt.Errorf("transcribe: artifact store is required") } + if env.ManifestStore == nil { + return nil, fmt.Errorf("transcribe: manifest store is required") + } if env.Config.Pipeline == nil || env.Config.Session == nil { return nil, fmt.Errorf("transcribe: resolved config must include pipeline and session") } - if env.WhisperX == nil { - return nil, fmt.Errorf("transcribe: whisperx adapter is required") + if m == nil { + return nil, fmt.Errorf("transcribe: session manifest is required") } var sessionID string @@ -47,21 +49,29 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) } paths := sessionPathsForEnv(env, sessionID) - runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "transcribe") + reconciliation, err := reconcileTranscribeSpeakers(ctx, env, m) if err != nil { - return nil, fmt.Errorf("transcribe: resolve run-stage layout: %w", err) + return nil, fmt.Errorf("transcribe: reconcile speakers: %w", err) } - jobs, err := buildTranscribeJobs(m, paths) - if err != nil { - return nil, fmt.Errorf("transcribe: resolve audio inputs: %w", err) + state := newTranscribeExecutionState(env, m, paths, reconciliation.current) + if err := state.checkpointStartingCollection(ctx); err != nil { + return nil, fmt.Errorf("transcribe: checkpoint reconciled speaker collection before execution: %w", err) } - if len(jobs) == 0 { - return nil, fmt.Errorf("transcribe: no prepared audio files found") + + jobs := append([]transcribeReconciliationJob(nil), reconciliation.work...) + if len(jobs) > 0 && env.WhisperX == nil { + return nil, fmt.Errorf("transcribe: whisperx adapter is required") } - for index := range jobs { - jobs[index].runOutputPath, err = runLocalPathForCanonical(runLayout, paths, jobs[index].canonicalOutputPath) + if len(jobs) > 0 { + runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "transcribe") if err != nil { - return nil, fmt.Errorf("transcribe: resolve run-local output path for %q: %w", jobs[index].speakerID, err) + return nil, fmt.Errorf("transcribe: resolve run-stage layout: %w", err) + } + for index := range jobs { + jobs[index].runOutputPath, err = runLocalPathForCanonical(runLayout, paths, jobs[index].canonicalOutputPath) + if err != nil { + return nil, fmt.Errorf("transcribe: resolve run-local output path for %q: %w", jobs[index].speakerID, err) + } } } @@ -69,15 +79,10 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) if env.Config.Pipeline.WhisperX.Concurrency != nil && *env.Config.Pipeline.WhisperX.Concurrency > 0 { concurrency = *env.Config.Pipeline.WhisperX.Concurrency } - if concurrency > len(jobs) { + if len(jobs) > 0 && concurrency > len(jobs) { concurrency = len(jobs) } - var ( - mu sync.Mutex - completed int - perFile = map[string]map[string]any{} - ) group, groupCtx := errgroup.WithContext(ctx) group.SetLimit(concurrency) for _, job := range jobs { @@ -86,6 +91,9 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) if err := groupCtx.Err(); err != nil { return err } + if err := state.checkpointFailure(); err != nil { + return err + } res, runErr := env.WhisperX.Transcribe(groupCtx, whisperx.TranscribeRequest{ SpeakerID: job.speakerID, AudioPath: job.audioPath, @@ -95,13 +103,6 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) return fmt.Errorf("speaker %q (%s): %w", job.speakerID, filepath.Base(job.audioPath), runErr) } - if _, err := authoritativeOutputPath(job.runOutputPath, res.OutputRawTranscriptPath); err != nil { - return fmt.Errorf("speaker %q: %w", job.speakerID, err) - } - if err := validateTranscriptJSONFile(job.runOutputPath); err != nil { - return fmt.Errorf("speaker %q output %q invalid: %w", job.speakerID, job.runOutputPath, err) - } - meta := map[string]any{ "attempts": res.Attempts, "http_status": res.HTTPStatus, @@ -109,17 +110,17 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) "output_path": job.runOutputPath, } - mu.Lock() - perFile[job.speakerID] = meta - completed++ - mu.Unlock() + if err := state.acceptSpeaker(ctx, job, res.OutputRawTranscriptPath, meta); err != nil { + return fmt.Errorf("speaker %q: %w", job.speakerID, err) + } return nil }) } groupErr := group.Wait() - mu.Lock() - completedCount := completed - mu.Unlock() + if err := state.checkpointFailure(); err != nil { + return nil, err + } + completedCount := state.completedCount() if err := ctx.Err(); err != nil { return nil, fmt.Errorf("transcribe: canceled after completing %d of %d audio jobs: %w", completedCount, len(jobs), err) } @@ -129,34 +130,9 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) if completedCount != len(jobs) { return nil, fmt.Errorf("transcribe: completed %d of %d planned audio jobs", completedCount, len(jobs)) } - - speakers := make([]string, 0, len(perFile)) - for speaker := range perFile { - speakers = append(speakers, speaker) - } - sort.Strings(speakers) - - outputs := make([]artifacts.Ref, 0, len(speakers)) - runOutputPaths := make([]string, 0, len(speakers)) - outputPaths := make([]string, 0, len(speakers)) - orderedPerFile := make(map[string]any, len(speakers)) - for _, speaker := range speakers { - var selected transcribeJob - for _, job := range jobs { - if job.speakerID == speaker { - selected = job - break - } - } - runOutputPaths = append(runOutputPaths, selected.runOutputPath) - ref := artifacts.Ref{Kind: "transcript_raw", Category: "transcripts", SessionID: sessionID, AbsolutePath: selected.runOutputPath} - materialized, err := materializeRunLocalOutput(env.ArtifactStore, selected.runOutputPath, selected.canonicalOutputPath, ref) - if err != nil { - return nil, fmt.Errorf("transcribe: materialize %q output: %w", speaker, err) - } - outputs = append(outputs, materialized) - outputPaths = append(outputPaths, selected.canonicalOutputPath) - orderedPerFile[speaker] = perFile[speaker] + outputs, outputPaths, runOutputPaths, orderedPerFile, err := state.successDetails() + if err != nil { + return nil, fmt.Errorf("transcribe: assemble completed outputs: %w", err) } retries := 0 @@ -168,7 +144,7 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) Outputs: outputs, Metadata: map[string]any{ "stage": "transcribe", - "audio_files_count": len(jobs), + "audio_files_count": len(reconciliation.jobs), "language": env.Config.Pipeline.WhisperX.Language, "concurrency": concurrency, "retries": retries, diff --git a/internal/stage/transcribe_checkpoint_test.go b/internal/stage/transcribe_checkpoint_test.go new file mode 100644 index 0000000..7a9791a --- /dev/null +++ b/internal/stage/transcribe_checkpoint_test.go @@ -0,0 +1,299 @@ +package stage + +import ( + "context" + "errors" + "os" + "slices" + "sync/atomic" + "testing" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func TestTranscribeFailureCheckpointsEarlierSpeakersAndRetryRunsOnlyRemaining(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"a.flac", "b.flac", "c.flac", "d.flac"}) + concurrency := 1 + env.Config.Pipeline.WhisperX.Concurrency = &concurrency + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + cause := errors.New("speaker c failed") + env.WhisperX = &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 failure") + } + if err := writeJSONFile(req.OutputRawTranscriptPath, map[string]any{"speaker": req.SpeakerID}); err != nil { + return whisperx.TranscribeResult{}, err + } + return whisperx.TranscribeResult{OutputRawTranscriptPath: req.OutputRawTranscriptPath}, nil + }} + + if _, err := (transcribeStage{}).Run(context.Background(), env, m); !errors.Is(err, cause) { + t.Fatalf("Run() error = %v, want %v", err, cause) + } + durable, err := env.ManifestStore.Load(context.Background(), sessionPathsForEnv(env, m.SessionID).ManifestPath) + if err != nil { + t.Fatalf("load durable manifest: %v", err) + } + if got := sortedTranscribeSpeakerKeys(durable.Stages["transcribe"].TranscribeSpeakers); !slices.Equal(got, []string{"a", "b"}) { + t.Fatalf("durable speakers = %#v, want a and b", got) + } + + retry := &whisperx.FakeClient{} + env.WhisperX = retry + result, err := (transcribeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("retry Run() error = %v", err) + } + requests := retry.RequestsSnapshot() + gotRequests := make([]string, 0, len(requests)) + for _, request := range requests { + gotRequests = append(gotRequests, request.SpeakerID) + } + if !slices.Equal(gotRequests, []string{"c", "d"}) { + t.Fatalf("retry speakers = %#v, want c and d", gotRequests) + } + if len(result.Outputs) != 4 { + t.Fatalf("retry outputs = %d, want all four speakers", len(result.Outputs)) + } +} + +func TestTranscribeConcurrentCheckpointsDoNotLoseCompletions(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"a.flac", "b.flac", "c.flac", "d.flac"}) + concurrency := 4 + env.Config.Pipeline.WhisperX.Concurrency = &concurrency + 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) + } + durable, err := env.ManifestStore.Load(context.Background(), sessionPathsForEnv(env, m.SessionID).ManifestPath) + if err != nil { + t.Fatalf("load durable manifest: %v", err) + } + if got := sortedTranscribeSpeakerKeys(durable.Stages["transcribe"].TranscribeSpeakers); !slices.Equal(got, []string{"a", "b", "c", "d"}) { + t.Fatalf("durable speakers = %#v, want all completions", got) + } +} + +func TestTranscribeStartingCheckpointFailurePreventsRequestsAndCanonicalWrites(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac"}) + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + cause := errors.New("starting checkpoint failed") + base := env.ManifestStore + env.ManifestStore = &failingTranscribeManifestStore{Store: base, FailAt: 1, Err: cause} + fake := &whisperx.FakeClient{} + env.WhisperX = fake + + if _, err := (transcribeStage{}).Run(context.Background(), env, m); !errors.Is(err, cause) { + t.Fatalf("Run() error = %v, want %v", err, cause) + } + if got := len(fake.RequestsSnapshot()); got != 0 { + t.Fatalf("adapter requests = %d, want none", got) + } + canonical := sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir + "/alice.json" + if _, err := os.Stat(canonical); !os.IsNotExist(err) { + t.Fatalf("canonical output exists after starting checkpoint failure: %v", err) + } +} + +func TestTranscribeCheckpointFailureRetainsEarlierAuthority(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"a.flac", "b.flac", "c.flac"}) + concurrency := 1 + env.Config.Pipeline.WhisperX.Concurrency = &concurrency + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + cause := errors.New("speaker checkpoint failed") + base := env.ManifestStore + env.ManifestStore = &failingTranscribeManifestStore{Store: base, FailAt: 3, Err: cause} + fake := &whisperx.FakeClient{} + env.WhisperX = fake + + if _, err := (transcribeStage{}).Run(context.Background(), env, m); !errors.Is(err, cause) { + t.Fatalf("Run() error = %v, want %v", err, cause) + } + requests := fake.RequestsSnapshot() + if len(requests) != 2 { + t.Fatalf("adapter requests = %d, want a and b only", len(requests)) + } + durable, err := base.Load(context.Background(), sessionPathsForEnv(env, m.SessionID).ManifestPath) + if err != nil { + t.Fatalf("load durable manifest: %v", err) + } + if got := sortedTranscribeSpeakerKeys(durable.Stages["transcribe"].TranscribeSpeakers); !slices.Equal(got, []string{"a"}) { + t.Fatalf("durable speakers = %#v, want only a", got) + } + if got := sortedTranscribeSpeakerKeys(m.Stages["transcribe"].TranscribeSpeakers); !slices.Equal(got, []string{"a"}) { + t.Fatalf("live speakers = %#v, want last committed collection", got) + } +} + +func TestTranscribeFailedForceRetryDoesNotResurrectOldSpeakers(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"a.flac", "b.flac"}) + concurrency := 1 + env.Config.Pipeline.WhisperX.Concurrency = &concurrency + 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("seed Run() error = %v", err) + } + + m.RunID = "20260503T020304Z-b2c3d4e5" + env.Force = true + cause := errors.New("forced b failed") + env.WhisperX = &whisperx.FakeClient{TranscribeFn: func( + _ context.Context, + req whisperx.TranscribeRequest, + ) (whisperx.TranscribeResult, error) { + if req.SpeakerID == "b" { + return whisperx.TranscribeResult{}, cause + } + if err := writeJSONFile(req.OutputRawTranscriptPath, map[string]any{"replacement": req.SpeakerID}); err != nil { + return whisperx.TranscribeResult{}, err + } + return whisperx.TranscribeResult{OutputRawTranscriptPath: req.OutputRawTranscriptPath}, nil + }} + if _, err := (transcribeStage{}).Run(context.Background(), env, m); !errors.Is(err, cause) { + t.Fatalf("forced Run() error = %v, want %v", err, cause) + } + if got := sortedTranscribeSpeakerKeys(m.Stages["transcribe"].TranscribeSpeakers); !slices.Equal(got, []string{"a"}) { + t.Fatalf("speakers after failed force = %#v, want only replacement a", got) + } + + env.Force = false + retry := &whisperx.FakeClient{} + env.WhisperX = retry + if _, err := (transcribeStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("retry Run() error = %v", err) + } + requests := retry.RequestsSnapshot() + if len(requests) != 1 || requests[0].SpeakerID != "b" { + t.Fatalf("retry requests = %#v, want only b", requests) + } +} + +func TestTranscribeRejectsAudioChangedDuringRequest(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac"}) + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + env.WhisperX = &whisperx.FakeClient{TranscribeFn: func( + _ context.Context, + req whisperx.TranscribeRequest, + ) (whisperx.TranscribeResult, error) { + if err := writeJSONFile(req.OutputRawTranscriptPath, map[string]any{"speaker": req.SpeakerID}); err != nil { + return whisperx.TranscribeResult{}, err + } + if err := os.WriteFile(req.AudioPath, []byte("changed"), 0o644); err != nil { + return whisperx.TranscribeResult{}, err + } + return whisperx.TranscribeResult{OutputRawTranscriptPath: req.OutputRawTranscriptPath}, nil + }} + if _, err := (transcribeStage{}).Run(context.Background(), env, m); err == nil { + t.Fatal("Run() error = nil, want changed-audio failure") + } + if got := len(m.Stages["transcribe"].TranscribeSpeakers); got != 0 { + t.Fatalf("checkpointed speakers = %d, want none", got) + } +} + +func TestTranscribeResumeValidationUsesSpeakerEvidence(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) + } + validation, err := (transcribeStage{}).ValidateResume(context.Background(), env, m) + if err != nil || !validation.Resumable { + t.Fatalf("ValidateResume() = %#v, %v, want resumable", validation, err) + } + if err := os.WriteFile(m.Inputs[0].Path, []byte("changed"), 0o644); err != nil { + t.Fatalf("change audio: %v", err) + } + validation, err = (transcribeStage{}).ValidateResume(context.Background(), env, m) + if err != nil || validation.Resumable { + t.Fatalf("ValidateResume() = %#v, %v, want execution required", validation, err) + } +} + +func TestTranscribeRebuildsAggregateFromCurrentSpeakersWithoutRequests(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac"}) + fake := &whisperx.FakeClient{} + env.WhisperX = fake + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + seed, err := (transcribeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("seed Run() error = %v", err) + } + originalRunID := m.RunID + m.MarkStageFailed("transcribe", m.UpdatedAt.Add(1), "aggregate save failed") + m.RunID = "20260503T030405Z-c3d4e5f6" + fake = &whisperx.FakeClient{} + env.WhisperX = fake + + result, err := (transcribeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("rebuild Run() error = %v", err) + } + if got := len(fake.RequestsSnapshot()); got != 0 { + t.Fatalf("adapter requests = %d, want none", got) + } + if len(result.Outputs) != len(seed.Outputs) { + t.Fatalf("rebuilt outputs = %d, want %d", len(result.Outputs), len(seed.Outputs)) + } + for index, output := range result.Outputs { + if output.ProducerRunID != originalRunID { + t.Fatalf("output %d producer = %q, want %q", index, output.ProducerRunID, originalRunID) + } + if output.Checksum != seed.Outputs[index].Checksum { + t.Fatalf("output %d checksum = %q, want %q", index, output.Checksum, seed.Outputs[index].Checksum) + } + } +} + +type failingTranscribeManifestStore struct { + manifest.Store + Saves atomic.Int64 + FailAt int64 + Err error +} + +func (store *failingTranscribeManifestStore) Save( + ctx context.Context, + path string, + m *manifest.Manifest, +) error { + if store.Saves.Add(1) == store.FailAt { + return store.Err + } + return store.Store.Save(ctx, path, m) +} + +func sortedTranscribeSpeakerKeys(records map[string]manifest.TranscribeSpeakerRecord) []string { + keys := make([]string, 0, len(records)) + for key := range records { + keys = append(keys, key) + } + slices.Sort(keys) + return keys +} diff --git a/internal/stage/transcribe_execution_state.go b/internal/stage/transcribe_execution_state.go new file mode 100644 index 0000000..640035c --- /dev/null +++ b/internal/stage/transcribe_execution_state.go @@ -0,0 +1,292 @@ +package stage + +import ( + "context" + "fmt" + "reflect" + "sort" + "strings" + "sync" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" +) + +type transcribeExecutionState struct { + env *Env + manifest *manifest.Manifest + paths artifacts.SessionPaths + + mu sync.Mutex + records map[string]manifest.TranscribeSpeakerRecord + checkpointErr error + completed int + perFile map[string]map[string]any + runOutputs map[string]string +} + +func newTranscribeExecutionState( + env *Env, + m *manifest.Manifest, + paths artifacts.SessionPaths, + current map[string]manifest.TranscribeSpeakerRecord, +) *transcribeExecutionState { + return &transcribeExecutionState{ + env: env, + manifest: m, + paths: paths, + records: manifest.CloneTranscribeSpeakerCollection(current), + perFile: make(map[string]map[string]any), + runOutputs: make(map[string]string), + } +} + +func (state *transcribeExecutionState) checkpointStartingCollection(ctx context.Context) error { + state.mu.Lock() + defer state.mu.Unlock() + if record := transcribeStageRecord(state.manifest); record != nil && + record.TranscribeStateVersion == manifest.TranscribeStateContractVersion && + reflect.DeepEqual(record.TranscribeSpeakers, state.records) { + return nil + } + if err := state.checkpointCollectionLocked(ctx, state.records); err != nil { + state.checkpointErr = err + return err + } + return nil +} + +func (state *transcribeExecutionState) acceptSpeaker( + ctx context.Context, + job transcribeReconciliationJob, + adapterOutputPath string, + metadata map[string]any, +) error { + if _, err := authoritativeOutputPath(job.runOutputPath, adapterOutputPath); err != nil { + return err + } + if err := validateTranscriptJSONFile(job.runOutputPath); err != nil { + return fmt.Errorf("output %q invalid: %w", job.runOutputPath, err) + } + audioSHA256, audioSize, err := hashRegularFileWithContext(ctx, job.audioPath) + if err != nil { + return fmt.Errorf("recheck audio %q: %w", job.audioPath, err) + } + if audioSHA256 != job.audioSHA256 || audioSize != job.audioSize { + return fmt.Errorf("audio %q changed during transcription", job.audioPath) + } + if err := state.checkpointFailure(); err != nil { + return err + } + + producerRunID := strings.TrimSpace(state.manifest.RunID) + ref := artifacts.Ref{ + Kind: "transcript_raw", + Category: "transcripts", + SessionID: state.manifest.SessionID, + AbsolutePath: job.runOutputPath, + ProducerRunID: producerRunID, + } + materialized, err := materializeRunLocalOutput( + state.env.ArtifactStore, + job.runOutputPath, + job.canonicalOutputPath, + ref, + ) + if err != nil { + return fmt.Errorf("materialize output: %w", err) + } + outputSHA256, outputSize, err := hashRegularFileWithContext(ctx, job.canonicalOutputPath) + if err != nil { + return fmt.Errorf("inspect materialized output %q: %w", job.canonicalOutputPath, err) + } + if materialized.Checksum != outputSHA256 { + return fmt.Errorf("materialized output checksum changed while inspecting") + } + record := manifest.TranscribeSpeakerRecord{ + FingerprintVersion: manifest.TranscribeFingerprintContractVersion, + Fingerprint: job.fingerprint, + AudioPath: job.audioRelativePath, + AudioSHA256: job.audioSHA256, + AudioSize: job.audioSize, + Output: manifest.ArtifactRecord{ + Kind: "transcript_raw", + LocalPath: job.outputRelativePath, + ArchivePath: materialized.ArchivePath, + ProducerRunID: producerRunID, + Checksum: outputSHA256, + }, + OutputSize: outputSize, + ProducerRunID: producerRunID, + } + + state.mu.Lock() + defer state.mu.Unlock() + if state.checkpointErr != nil { + return state.checkpointErr + } + candidate := manifest.CloneTranscribeSpeakerCollection(state.records) + candidate[job.speakerID] = record + if err := state.checkpointCollectionLocked(ctx, candidate); err != nil { + state.checkpointErr = fmt.Errorf("transcribe: checkpoint speaker %q: %w", job.speakerID, err) + return state.checkpointErr + } + state.completed++ + state.perFile[job.speakerID] = cloneStringAnyMap(metadata) + state.runOutputs[job.speakerID] = job.runOutputPath + return nil +} + +func (state *transcribeExecutionState) checkpointCollectionLocked( + ctx context.Context, + records map[string]manifest.TranscribeSpeakerRecord, +) error { + if state.manifest == nil { + return fmt.Errorf("session manifest is required") + } + if state.env == nil || state.env.ManifestStore == nil { + return fmt.Errorf("manifest store is required") + } + if err := manifest.ValidateTranscribeSpeakerCollection( + manifest.TranscribeStateContractVersion, + records, + ); err != nil { + return fmt.Errorf("validate transcribe speaker collection: %w", err) + } + + candidate := cloneManifestForTranscribeCheckpoint(state.manifest) + if candidate.Stages == nil { + candidate.Stages = make(map[string]*manifest.StageRecord) + } + record := candidate.Stages["transcribe"] + if record == nil { + now := time.Now().UTC() + record = &manifest.StageRecord{ + Name: "transcribe", Status: manifest.StatusRunning, CreatedAt: now, UpdatedAt: now, + } + candidate.Stages["transcribe"] = record + } + record.TranscribeStateVersion = manifest.TranscribeStateContractVersion + record.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(records) + record.UpdatedAt = time.Now().UTC() + if err := state.env.ManifestStore.Save(ctx, state.paths.ManifestPath, candidate); err != nil { + return fmt.Errorf("save session manifest %q: %w", state.paths.ManifestPath, err) + } + + live := transcribeStageRecord(state.manifest) + if live == nil { + live = &manifest.StageRecord{} + if state.manifest.Stages == nil { + state.manifest.Stages = make(map[string]*manifest.StageRecord) + } + state.manifest.Stages["transcribe"] = live + } + live.TranscribeStateVersion = candidate.Stages["transcribe"].TranscribeStateVersion + live.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection( + candidate.Stages["transcribe"].TranscribeSpeakers, + ) + live.UpdatedAt = candidate.Stages["transcribe"].UpdatedAt + state.manifest.UpdatedAt = candidate.UpdatedAt + state.records = manifest.CloneTranscribeSpeakerCollection(live.TranscribeSpeakers) + return nil +} + +func (state *transcribeExecutionState) checkpointFailure() error { + state.mu.Lock() + defer state.mu.Unlock() + return state.checkpointErr +} + +func (state *transcribeExecutionState) completedCount() int { + state.mu.Lock() + defer state.mu.Unlock() + return state.completed +} + +func (state *transcribeExecutionState) successDetails() ( + []artifacts.Ref, + []string, + []string, + map[string]any, + error, +) { + state.mu.Lock() + defer state.mu.Unlock() + + speakers := make([]string, 0, len(state.records)) + for speaker := range state.records { + speakers = append(speakers, speaker) + } + sort.Strings(speakers) + outputs := make([]artifacts.Ref, 0, len(speakers)) + outputPaths := make([]string, 0, len(speakers)) + runOutputPaths := make([]string, 0, len(state.runOutputs)) + orderedPerFile := make(map[string]any, len(state.perFile)) + for _, speaker := range speakers { + record := state.records[speaker] + absolute, err := pathsafe.JoinSlashRelativeUnderRoot(state.paths.Root, record.Output.LocalPath) + if err != nil { + return nil, nil, nil, nil, fmt.Errorf("speaker %q output path: %w", speaker, err) + } + outputs = append(outputs, artifacts.Ref{ + Kind: record.Output.Kind, + Category: "transcripts", + SessionID: state.manifest.SessionID, + RelativePath: record.Output.LocalPath, + AbsolutePath: absolute, + ArchivePath: record.Output.ArchivePath, + RemoteKey: record.Output.RemoteKey, + Checksum: record.Output.Checksum, + ProducerRunID: record.ProducerRunID, + }) + outputPaths = append(outputPaths, absolute) + if runPath, ok := state.runOutputs[speaker]; ok { + runOutputPaths = append(runOutputPaths, runPath) + } + if metadata, ok := state.perFile[speaker]; ok { + orderedPerFile[speaker] = cloneStringAnyMap(metadata) + } + } + return outputs, outputPaths, runOutputPaths, orderedPerFile, nil +} + +func transcribeStageRecord(m *manifest.Manifest) *manifest.StageRecord { + if m == nil || m.Stages == nil { + return nil + } + return m.Stages["transcribe"] +} + +func cloneManifestForTranscribeCheckpoint(source *manifest.Manifest) *manifest.Manifest { + clone := *source + clone.Inputs = append([]manifest.InputRecord(nil), source.Inputs...) + clone.Artifacts = append([]manifest.ArtifactRecord(nil), source.Artifacts...) + clone.Stages = make(map[string]*manifest.StageRecord, len(source.Stages)) + for name, sourceRecord := range source.Stages { + if sourceRecord == nil { + clone.Stages[name] = nil + continue + } + record := *sourceRecord + record.Outputs = append([]manifest.ArtifactRecord(nil), sourceRecord.Outputs...) + record.Logs = append([]string(nil), sourceRecord.Logs...) + record.GeneratedConfigs = append([]string(nil), sourceRecord.GeneratedConfigs...) + record.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(sourceRecord.TranscribeSpeakers) + record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(sourceRecord.AnalyzeArtifacts) + clone.Stages[name] = &record + } + return &clone +} + +func cloneStringAnyMap(source map[string]any) map[string]any { + if source == nil { + return nil + } + clone := make(map[string]any, len(source)) + for key, value := range source { + clone[key] = value + } + return clone +} diff --git a/internal/stage/transcribe_resume.go b/internal/stage/transcribe_resume.go new file mode 100644 index 0000000..f573b0d --- /dev/null +++ b/internal/stage/transcribe_resume.go @@ -0,0 +1,32 @@ +package stage + +import ( + "context" + "fmt" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func (transcribeStage) ValidateResume( + ctx context.Context, + env *Env, + m *manifest.Manifest, +) (ResumeValidation, error) { + reconciliation, err := reconcileTranscribeSpeakers(ctx, env, m) + if err != nil { + return ResumeValidation{}, fmt.Errorf("transcribe resume: reconcile speakers: %w", err) + } + if len(reconciliation.work) == 0 { + return Resumable(), nil + } + speakers := make([]string, 0, len(reconciliation.work)) + for _, job := range reconciliation.work { + speakers = append(speakers, job.speakerID) + } + sort.Strings(speakers) + return NonResumable("speaker transcripts require execution: " + strings.Join(speakers, ", ")), nil +} + +var _ ResumeValidator = transcribeStage{} diff --git a/internal/stage/transcribe_test.go b/internal/stage/transcribe_test.go index 19bb740..a3fac41 100644 --- a/internal/stage/transcribe_test.go +++ b/internal/stage/transcribe_test.go @@ -322,7 +322,7 @@ func TestTranscribeStagePreservesDistinctSameBasenameAudio(t *testing.T) { } } -func TestTranscribeStageCancellationDoesNotMaterializePartialOutputs(t *testing.T) { +func TestTranscribeStageCancellationPreservesCheckpointedOutputs(t *testing.T) { env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac", "cara.flac"}) m.RunID = "20260810T214607Z-a1b2c3d4" fastComplete := make(chan struct{}) @@ -351,6 +351,21 @@ func TestTranscribeStageCancellationDoesNotMaterializePartialOutputs(t *testing. }() select { case <-fastComplete: + manifestPath := sessionPathsForEnv(env, m.SessionID).ManifestPath + deadline := time.Now().Add(time.Second) + for { + durable, err := env.ManifestStore.Load(context.Background(), manifestPath) + if err == nil { + if _, ok := durable.Stages["transcribe"].TranscribeSpeakers["alice"]; ok { + break + } + } + if time.Now().After(deadline) { + cancel() + t.Fatal("alice completion was not checkpointed before cancellation") + } + time.Sleep(5 * time.Millisecond) + } cancel() case <-time.After(time.Second): cancel() @@ -365,8 +380,15 @@ func TestTranscribeStageCancellationDoesNotMaterializePartialOutputs(t *testing. } canonical := filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "alice.json") - if _, err := os.Stat(canonical); !os.IsNotExist(err) { - t.Fatalf("partial canonical output should not exist, stat err = %v", err) + if _, err := os.Stat(canonical); err != nil { + t.Fatalf("checkpointed canonical output missing: %v", err) + } + durable, err := env.ManifestStore.Load(context.Background(), sessionPathsForEnv(env, m.SessionID).ManifestPath) + if err != nil { + t.Fatalf("load checkpointed manifest: %v", err) + } + if _, ok := durable.Stages["transcribe"].TranscribeSpeakers["alice"]; !ok { + t.Fatalf("checkpointed speakers = %#v, want alice", durable.Stages["transcribe"].TranscribeSpeakers) } } @@ -454,8 +476,13 @@ func setupTranscribeEnv(t *testing.T, audioFiles []string) (*Env, *manifest.Mani env := &Env{ Config: cfg, ArtifactStore: artifacts.NewLocalStore(workspace), + ManifestStore: &manifest.LocalStore{}, } - return env, manifest.New("2026-05-03", time.Now().UTC()) + m := manifest.New("2026-05-03", time.Now().UTC()) + m.Campaign = "sample-campaign" + m.RunID = "20260503T010203Z-a1b2c3d4" + m.MarkStageRunning("transcribe", time.Now().UTC()) + return env, m } func writeJSONFile(path string, payload any) error {