diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index a6a103f..aa06cb9 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; Stage 1 complete. +Status: Implementation in progress; Stages 1-2 complete. ## Scope and Instructions @@ -148,7 +148,15 @@ aggregate status, without activating new stage behavior yet. ## Stage 2: Implement Read-Only Speaker Reconciliation -Status: Pending. +Status: Complete. + +Completion marker (2026-09-21): Extracted shared deterministic transcribe job +construction and added read-only, cancellation-aware per-speaker input and +output reconciliation. Covered unchanged reuse, semantic and operational +settings, changed/added/removed inputs, legacy and forced work, missing/corrupt +outputs, unsafe evidence, and missing audio without connecting runtime skip or +checkpoint behavior. Verified with +`go test ./internal/stage ./internal/manifest` and `git diff --check`. Preparation: `internal/stage/transcribe.go`, `semantic_contracts_initial.go`, `analyze_reconciliation.go`, `extract_resume.go`, their tests, and the artifact diff --git a/internal/stage/transcribe.go b/internal/stage/transcribe.go index 0415556..0e83361 100644 --- a/internal/stage/transcribe.go +++ b/internal/stage/transcribe.go @@ -51,48 +51,26 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) if err != nil { return nil, fmt.Errorf("transcribe: resolve run-stage layout: %w", err) } - audioFiles, err := discoverPreparedAudio(m, paths.AudioDir) + jobs, err := buildTranscribeJobs(m, paths) if err != nil { return nil, fmt.Errorf("transcribe: resolve audio inputs: %w", err) } - if len(audioFiles) == 0 { + if len(jobs) == 0 { return nil, fmt.Errorf("transcribe: no prepared audio files found") } + 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) + } + } concurrency := 1 if env.Config.Pipeline.WhisperX.Concurrency != nil && *env.Config.Pipeline.WhisperX.Concurrency > 0 { concurrency = *env.Config.Pipeline.WhisperX.Concurrency } - if concurrency > len(audioFiles) { - concurrency = len(audioFiles) - } - - type job struct { - speakerID string - audioPath string - outPath string - } - jobs := make([]job, 0, len(audioFiles)) - seenSpeaker := map[string]string{} - for _, audioPath := range audioFiles { - base := strings.TrimSuffix(filepath.Base(audioPath), filepath.Ext(audioPath)) - if base == "" { - return nil, fmt.Errorf("transcribe: could not derive speaker id from %q", audioPath) - } - if prev, ok := seenSpeaker[base]; ok { - return nil, fmt.Errorf("transcribe: duplicate speaker/audio basename %q from %q and %q", base, prev, audioPath) - } - seenSpeaker[base] = audioPath - canonicalOut := filepath.Join(paths.TranscriptsRawDir, base+".json") - runOut, err := runLocalPathForCanonical(runLayout, paths, canonicalOut) - if err != nil { - return nil, fmt.Errorf("transcribe: resolve run-local output path for %q: %w", base, err) - } - jobs = append(jobs, job{ - speakerID: base, - audioPath: audioPath, - outPath: runOut, - }) + if concurrency > len(jobs) { + concurrency = len(jobs) } var ( @@ -111,24 +89,24 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) res, runErr := env.WhisperX.Transcribe(groupCtx, whisperx.TranscribeRequest{ SpeakerID: job.speakerID, AudioPath: job.audioPath, - OutputRawTranscriptPath: job.outPath, + OutputRawTranscriptPath: job.runOutputPath, }) if runErr != nil { return fmt.Errorf("speaker %q (%s): %w", job.speakerID, filepath.Base(job.audioPath), runErr) } - if _, err := authoritativeOutputPath(job.outPath, res.OutputRawTranscriptPath); err != nil { + if _, err := authoritativeOutputPath(job.runOutputPath, res.OutputRawTranscriptPath); err != nil { return fmt.Errorf("speaker %q: %w", job.speakerID, err) } - if err := validateTranscriptJSONFile(job.outPath); err != nil { - return fmt.Errorf("speaker %q output %q invalid: %w", job.speakerID, job.outPath, 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, "duration_ms": res.Duration.Milliseconds(), - "output_path": job.outPath, + "output_path": job.runOutputPath, } mu.Lock() @@ -163,22 +141,21 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) outputPaths := make([]string, 0, len(speakers)) orderedPerFile := make(map[string]any, len(speakers)) for _, speaker := range speakers { - var runOut string + var selected transcribeJob for _, job := range jobs { if job.speakerID == speaker { - runOut = job.outPath + selected = job break } } - runOutputPaths = append(runOutputPaths, runOut) - canonicalOut := filepath.Join(paths.TranscriptsRawDir, speaker+".json") - ref := artifacts.Ref{Kind: "transcript_raw", Category: "transcripts", SessionID: sessionID, AbsolutePath: runOut} - materialized, err := materializeRunLocalOutput(env.ArtifactStore, runOut, canonicalOut, ref) + 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, canonicalOut) + outputPaths = append(outputPaths, selected.canonicalOutputPath) orderedPerFile[speaker] = perFile[speaker] } @@ -191,7 +168,7 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) Outputs: outputs, Metadata: map[string]any{ "stage": "transcribe", - "audio_files_count": len(audioFiles), + "audio_files_count": len(jobs), "language": env.Config.Pipeline.WhisperX.Language, "concurrency": concurrency, "retries": retries, @@ -204,7 +181,39 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) }, nil } -func discoverPreparedAudio(m *manifest.Manifest, audioDir string) ([]string, error) { +type transcribeJob struct { + speakerID string + audioPath string + canonicalOutputPath string + runOutputPath string +} + +func buildTranscribeJobs(m *manifest.Manifest, paths artifacts.SessionPaths) ([]transcribeJob, error) { + audioFiles, err := discoverPreparedAudio(m, paths) + if err != nil { + return nil, err + } + jobs := make([]transcribeJob, 0, len(audioFiles)) + seenSpeaker := map[string]string{} + for _, audioPath := range audioFiles { + base := strings.TrimSuffix(filepath.Base(audioPath), filepath.Ext(audioPath)) + if base == "" { + return nil, fmt.Errorf("could not derive speaker id from %q", audioPath) + } + if previous, exists := seenSpeaker[base]; exists { + return nil, fmt.Errorf("duplicate speaker/audio basename %q from %q and %q", base, previous, audioPath) + } + seenSpeaker[base] = audioPath + jobs = append(jobs, transcribeJob{ + speakerID: base, + audioPath: audioPath, + canonicalOutputPath: filepath.Join(paths.TranscriptsRawDir, base+".json"), + }) + } + return jobs, nil +} + +func discoverPreparedAudio(m *manifest.Manifest, paths artifacts.SessionPaths) ([]string, error) { fromManifest := make([]string, 0) if m != nil { for _, in := range m.Inputs { @@ -215,7 +224,7 @@ func discoverPreparedAudio(m *manifest.Manifest, audioDir string) ([]string, err if p == "" { continue } - fromManifest = append(fromManifest, filepath.Clean(p)) + fromManifest = append(fromManifest, artifacts.ResolveSessionLocalPathForRead(paths, p)) } } if len(fromManifest) > 0 { @@ -226,9 +235,9 @@ func discoverPreparedAudio(m *manifest.Manifest, audioDir string) ([]string, err return fromManifest, nil } - entries, err := os.ReadDir(audioDir) + entries, err := os.ReadDir(paths.AudioDir) if err != nil { - return nil, fmt.Errorf("read audio directory %q: %w", audioDir, err) + return nil, fmt.Errorf("read audio directory %q: %w", paths.AudioDir, err) } out := make([]string, 0) @@ -236,7 +245,7 @@ func discoverPreparedAudio(m *manifest.Manifest, audioDir string) ([]string, err if entry.IsDir() { continue } - p := filepath.Join(audioDir, entry.Name()) + p := filepath.Join(paths.AudioDir, entry.Name()) if !isFlac(p) { continue } diff --git a/internal/stage/transcribe_reconciliation.go b/internal/stage/transcribe_reconciliation.go new file mode 100644 index 0000000..cc72f21 --- /dev/null +++ b/internal/stage/transcribe_reconciliation.go @@ -0,0 +1,339 @@ +package stage + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "os" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" +) + +type transcribeReconciliationJob struct { + transcribeJob + audioRelativePath string + audioSHA256 string + audioSize int64 + outputRelativePath string + fingerprint string +} + +type transcribeReconciliation struct { + jobs []transcribeReconciliationJob + current map[string]manifest.TranscribeSpeakerRecord + work []transcribeReconciliationJob +} + +type transcribeSpeakerFingerprint 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"` +} + +// reconcileTranscribeSpeakers compares the current prepared audio set with +// manifest-owned per-speaker evidence without mutating either. +func reconcileTranscribeSpeakers(ctx context.Context, env *Env, m *manifest.Manifest) (transcribeReconciliation, error) { + if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil { + return transcribeReconciliation{}, fmt.Errorf("transcribe reconciliation requires resolved pipeline and session config") + } + if env.ArtifactStore == nil { + return transcribeReconciliation{}, fmt.Errorf("transcribe reconciliation requires artifact store") + } + if err := contextError(ctx); err != nil { + return transcribeReconciliation{}, err + } + + sessionID := strings.TrimSpace(env.Config.Session.SessionID) + if m != nil && strings.TrimSpace(m.SessionID) != "" { + sessionID = strings.TrimSpace(m.SessionID) + } + if sessionID == "" { + return transcribeReconciliation{}, fmt.Errorf("transcribe reconciliation requires session ID") + } + paths := sessionPathsForEnv(env, sessionID) + baseJobs, err := buildTranscribeJobs(m, paths) + if err != nil { + return transcribeReconciliation{}, fmt.Errorf("resolve audio inputs: %w", err) + } + if len(baseJobs) == 0 { + return transcribeReconciliation{}, fmt.Errorf("no prepared audio files found") + } + semantic, err := (transcribeStage{}).SemanticConfigFingerprint(env) + if err != nil { + return transcribeReconciliation{}, err + } + + result := transcribeReconciliation{ + jobs: make([]transcribeReconciliationJob, 0, len(baseJobs)), + current: map[string]manifest.TranscribeSpeakerRecord{}, + work: make([]transcribeReconciliationJob, 0, len(baseJobs)), + } + for _, base := range baseJobs { + candidate, err := inspectTranscribeJob(ctx, paths, base, semantic) + if err != nil { + return transcribeReconciliation{}, fmt.Errorf("speaker %q: %w", base.speakerID, err) + } + result.jobs = append(result.jobs, candidate) + } + + stageRecord, versioned, err := transcribeManifestState(m) + if err != nil { + return transcribeReconciliation{}, err + } + if env.Force || !versioned { + result.work = append(result.work, result.jobs...) + return result, nil + } + + for _, candidate := range result.jobs { + record, exists := stageRecord.TranscribeSpeakers[candidate.speakerID] + if !exists { + result.work = append(result.work, candidate) + continue + } + if err := manifest.ValidateTranscribeSpeakerCollection( + stageRecord.TranscribeStateVersion, + map[string]manifest.TranscribeSpeakerRecord{candidate.speakerID: record}, + ); err != nil { + return transcribeReconciliation{}, fmt.Errorf("speaker %q manifest evidence: %w", candidate.speakerID, err) + } + if !sameTranscribeInputEvidence(candidate, record) { + result.work = append(result.work, candidate) + continue + } + current, err := inspectTranscribeOutput(ctx, paths, candidate, record) + if err != nil { + return transcribeReconciliation{}, fmt.Errorf("speaker %q output evidence: %w", candidate.speakerID, err) + } + if !current { + result.work = append(result.work, candidate) + continue + } + result.current[candidate.speakerID] = manifest.CloneTranscribeSpeakerCollection( + map[string]manifest.TranscribeSpeakerRecord{candidate.speakerID: record}, + )[candidate.speakerID] + } + return result, nil +} + +func inspectTranscribeJob( + ctx context.Context, + paths artifacts.SessionPaths, + job transcribeJob, + semantic manifest.SemanticConfigFingerprint, +) (transcribeReconciliationJob, error) { + audioRelative, err := pathsafe.SlashRelativeFromRoot(paths.Root, job.audioPath) + if err != nil { + return transcribeReconciliationJob{}, fmt.Errorf("derive session-relative audio path: %w", err) + } + outputRelative, err := pathsafe.SlashRelativeFromRoot(paths.Root, job.canonicalOutputPath) + if err != nil { + return transcribeReconciliationJob{}, fmt.Errorf("derive session-relative output path: %w", err) + } + audioSHA256, audioSize, err := hashRegularFileWithContext(ctx, job.audioPath) + if err != nil { + return transcribeReconciliationJob{}, fmt.Errorf("inspect audio %q: %w", job.audioPath, err) + } + fingerprint, err := fingerprintTranscribeSpeaker(transcribeSpeakerFingerprint{ + SpeakerID: job.speakerID, + AudioPath: audioRelative, + AudioSHA256: audioSHA256, + AudioSize: audioSize, + SemanticConfigVersion: semantic.Version, + SemanticConfigDigest: semantic.Digest, + }) + if err != nil { + return transcribeReconciliationJob{}, err + } + return transcribeReconciliationJob{ + transcribeJob: job, + audioRelativePath: audioRelative, + audioSHA256: audioSHA256, + audioSize: audioSize, + outputRelativePath: outputRelative, + fingerprint: fingerprint, + }, nil +} + +func fingerprintTranscribeSpeaker(payload transcribeSpeakerFingerprint) (string, error) { + fingerprint, err := FingerprintSemanticConfig(manifest.TranscribeFingerprintContractVersion, payload) + if err != nil { + return "", fmt.Errorf("build per-speaker fingerprint: %w", err) + } + return fingerprint.Digest, nil +} + +func transcribeManifestState(m *manifest.Manifest) (*manifest.StageRecord, bool, error) { + if m == nil || m.Stages == nil || m.Stages["transcribe"] == nil { + return nil, false, nil + } + record := m.Stages["transcribe"] + if record.Name != "transcribe" { + return nil, false, fmt.Errorf("transcribe manifest record has stage name %q", record.Name) + } + if record.TranscribeStateVersion == 0 && record.TranscribeSpeakers == nil { + return record, false, nil + } + if record.TranscribeStateVersion != manifest.TranscribeStateContractVersion { + return nil, false, fmt.Errorf("unsupported transcribe state version %d", record.TranscribeStateVersion) + } + return record, true, nil +} + +func sameTranscribeInputEvidence(candidate transcribeReconciliationJob, record manifest.TranscribeSpeakerRecord) bool { + return record.FingerprintVersion == manifest.TranscribeFingerprintContractVersion && + record.Fingerprint == candidate.fingerprint && + record.AudioPath == candidate.audioRelativePath && + record.AudioSHA256 == candidate.audioSHA256 && + record.AudioSize == candidate.audioSize && + record.Output.LocalPath == candidate.outputRelativePath +} + +func inspectTranscribeOutput( + ctx context.Context, + paths artifacts.SessionPaths, + candidate transcribeReconciliationJob, + record manifest.TranscribeSpeakerRecord, +) (bool, error) { + if record.Output.LocalPath != candidate.outputRelativePath { + return false, nil + } + absolute, err := pathsafe.JoinSlashRelativeUnderRoot(paths.Root, record.Output.LocalPath) + if err != nil { + return false, fmt.Errorf("resolve recorded output path: %w", err) + } + exists, err := requireSafeTranscribeOutput(paths.Root, record.Output.LocalPath) + if err != nil { + return false, err + } + if !exists { + return false, nil + } + if err := validateTranscriptJSONFile(absolute); err != nil { + return false, nil + } + checksum, size, err := hashRegularFileWithContext(ctx, absolute) + if err != nil { + return false, fmt.Errorf("verify output: %w", err) + } + return checksum == record.Output.Checksum && size == record.OutputSize, nil +} + +func requireSafeTranscribeOutput(rootPath, relativePath string) (bool, error) { + normalized, err := pathsafe.NormalizeRelativeDestination(relativePath) + if err != nil { + return false, fmt.Errorf("recorded output path is unsafe: %w", err) + } + root, err := os.OpenRoot(rootPath) + if err != nil { + return false, fmt.Errorf("open session root: %w", err) + } + defer func() { _ = root.Close() }() + + parts := strings.Split(normalized, "/") + current := root + opened := make([]*os.Root, 0, len(parts)-1) + defer func() { + for index := len(opened) - 1; index >= 0; index-- { + _ = opened[index].Close() + } + }() + for _, part := range parts[:len(parts)-1] { + info, err := current.Lstat(part) + if errors.Is(err, os.ErrNotExist) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("inspect output ancestor %q: %w", part, err) + } + if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { + return false, fmt.Errorf("output ancestor %q is not a directory without symlinks", part) + } + next, err := current.OpenRoot(part) + if err != nil { + return false, fmt.Errorf("open output ancestor %q: %w", part, err) + } + opened = append(opened, next) + current = next + } + leaf := parts[len(parts)-1] + info, err := current.Lstat(leaf) + if errors.Is(err, os.ErrNotExist) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("inspect output %q: %w", normalized, err) + } + if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() { + return false, fmt.Errorf("output %q is not a regular file without symlinks", normalized) + } + return true, nil +} + +func hashRegularFileWithContext(ctx context.Context, path string) (string, int64, error) { + if err := contextError(ctx); err != nil { + return "", 0, err + } + file, err := os.Open(path) + if err != nil { + return "", 0, err + } + info, err := file.Stat() + if err != nil { + _ = file.Close() + return "", 0, err + } + if !info.Mode().IsRegular() { + _ = file.Close() + return "", 0, fmt.Errorf("path is not a regular file") + } + hash := sha256.New() + size, copyErr := io.Copy(hash, &contextReadChecker{ctx: ctx, reader: file}) + closeErr := file.Close() + if copyErr != nil { + return "", 0, copyErr + } + if closeErr != nil { + return "", 0, closeErr + } + if err := contextError(ctx); err != nil { + return "", 0, err + } + if size != info.Size() { + return "", 0, fmt.Errorf("file size changed while reading") + } + return hex.EncodeToString(hash.Sum(nil)), size, nil +} + +type contextReadChecker struct { + ctx context.Context + reader io.Reader +} + +func (r *contextReadChecker) Read(buffer []byte) (int, error) { + if err := contextError(r.ctx); err != nil { + return 0, err + } + return r.reader.Read(buffer) +} + +func contextError(ctx context.Context) error { + if ctx == nil { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + default: + return nil + } +} diff --git a/internal/stage/transcribe_reconciliation_test.go b/internal/stage/transcribe_reconciliation_test.go new file mode 100644 index 0000000..707ef87 --- /dev/null +++ b/internal/stage/transcribe_reconciliation_test.go @@ -0,0 +1,292 @@ +package stage + +import ( + "context" + "errors" + "os" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func TestReconcileTranscribeSpeakersReusesUnchangedCompletedEvidence(t *testing.T) { + env, m := setupCurrentTranscribeReconciliation(t, []string{"bob.flac", "alice.flac"}) + reconciled, err := reconcileTranscribeSpeakers(context.Background(), env, m) + if err != nil { + t.Fatalf("reconcileTranscribeSpeakers() error = %v", err) + } + if got := reconciliationSpeakerIDs(reconciled.jobs); !slices.Equal(got, []string{"alice", "bob"}) { + t.Fatalf("jobs = %#v, want sorted speakers", got) + } + if len(reconciled.current) != 2 || len(reconciled.work) != 0 { + t.Fatalf("reconciliation = current %d work %d, want 2/0", len(reconciled.current), len(reconciled.work)) + } + + // Manifest paths may be session-relative; reconciliation must not resolve + // them against the process working directory. + for index := range m.Inputs { + if m.Inputs[index].Kind == "audio" { + relative, relErr := filepath.Rel(sessionPathsForEnv(env, m.SessionID).Root, m.Inputs[index].Path) + if relErr != nil { + t.Fatal(relErr) + } + m.Inputs[index].Path = filepath.ToSlash(relative) + } + } + reconciled, err = reconcileTranscribeSpeakers(context.Background(), env, m) + if err != nil || len(reconciled.current) != 2 || len(reconciled.work) != 0 { + t.Fatalf("session-relative reconciliation = %#v, error %v", reconciled, err) + } +} + +func TestReconcileTranscribeSpeakersReplacesOnlyChangedAudio(t *testing.T) { + env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac", "bob.flac"}) + paths := sessionPathsForEnv(env, m.SessionID) + writeFile(t, filepath.Join(paths.AudioDir, "alice.flac"), "replacement audio") + + reconciled, err := reconcileTranscribeSpeakers(context.Background(), env, m) + if err != nil { + t.Fatalf("reconcileTranscribeSpeakers() error = %v", err) + } + if got := reconciliationSpeakerIDs(reconciled.work); !slices.Equal(got, []string{"alice"}) { + t.Fatalf("work = %#v, want alice", got) + } + if _, current := reconciled.current["bob"]; !current || len(reconciled.current) != 1 { + t.Fatalf("current = %#v, want only bob", reconciled.current) + } +} + +func TestReconcileTranscribeSpeakersTracksSemanticAndExcludesOperationalSettings(t *testing.T) { + for _, test := range []struct { + name string + mutate func(*Env) + work bool + }{ + {name: "service URL", mutate: func(env *Env) { env.Config.Pipeline.WhisperX.TranscribeURL = "https://other.example/transcribe" }, work: true}, + {name: "language", mutate: func(env *Env) { env.Config.Pipeline.WhisperX.Language = "fr" }, work: true}, + {name: "timeout", mutate: func(env *Env) { env.Config.Pipeline.WhisperX.Timeout = "45m" }}, + {name: "retry policy", mutate: func(env *Env) { + retries := 9 + env.Config.Pipeline.WhisperX.Retries = &retries + env.Config.Pipeline.WhisperX.RetryDelay = "15s" + }}, + {name: "concurrency", mutate: func(env *Env) { + concurrency := 1 + env.Config.Pipeline.WhisperX.Concurrency = &concurrency + }}, + } { + t.Run(test.name, func(t *testing.T) { + env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac"}) + test.mutate(env) + reconciled, err := reconcileTranscribeSpeakers(context.Background(), env, m) + if err != nil { + t.Fatalf("reconcileTranscribeSpeakers() error = %v", err) + } + if test.work { + if len(reconciled.work) != 1 || len(reconciled.current) != 0 { + t.Fatalf("semantic change reconciliation = current %d work %d", len(reconciled.current), len(reconciled.work)) + } + } else if len(reconciled.work) != 0 || len(reconciled.current) != 1 { + t.Fatalf("operational change reconciliation = current %d work %d", len(reconciled.current), len(reconciled.work)) + } + }) + } +} + +func TestReconcileTranscribeSpeakersHandlesAddedAndRemovedInputs(t *testing.T) { + env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac", "bob.flac"}) + paths := sessionPathsForEnv(env, m.SessionID) + caraPath := filepath.Join(paths.AudioDir, "cara.flac") + writeFile(t, caraPath, "cara audio") + m.Inputs = append(m.Inputs, manifest.InputRecord{Kind: "audio", Path: caraPath}) + + reconciled, err := reconcileTranscribeSpeakers(context.Background(), env, m) + if err != nil { + t.Fatalf("reconcile added input error = %v", err) + } + if got := reconciliationSpeakerIDs(reconciled.work); !slices.Equal(got, []string{"cara"}) || len(reconciled.current) != 2 { + t.Fatalf("added input reconciliation = current %#v work %#v", reconciled.current, got) + } + + inputs := m.Inputs[:0] + for _, input := range m.Inputs { + if input.Kind != "audio" || filepath.Base(input.Path) != "bob.flac" { + inputs = append(inputs, input) + } + } + m.Inputs = inputs + reconciled, err = reconcileTranscribeSpeakers(context.Background(), env, m) + if err != nil { + t.Fatalf("reconcile removed input error = %v", err) + } + if got := reconciliationSpeakerIDs(reconciled.jobs); !slices.Equal(got, []string{"alice", "cara"}) { + t.Fatalf("jobs after removal = %#v", got) + } + if _, retained := reconciled.current["bob"]; retained { + t.Fatalf("removed bob record retained in current collection: %#v", reconciled.current) + } +} + +func TestReconcileTranscribeSpeakersTreatsMissingAndCorruptRegularOutputsAsWork(t *testing.T) { + for _, test := range []struct { + name string + mutate func(*testing.T, string) + }{ + {name: "missing", mutate: func(t *testing.T, path string) { + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + }}, + {name: "invalid JSON", mutate: func(t *testing.T, path string) { writeFile(t, path, "not-json") }}, + {name: "checksum mismatch", mutate: func(t *testing.T, path string) { writeFile(t, path, `{"speaker":"changed"}`) }}, + } { + t.Run(test.name, func(t *testing.T) { + env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac"}) + path := filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "alice.json") + test.mutate(t, path) + reconciled, err := reconcileTranscribeSpeakers(context.Background(), env, m) + if err != nil { + t.Fatalf("reconcileTranscribeSpeakers() error = %v", err) + } + if len(reconciled.current) != 0 || len(reconciled.work) != 1 { + t.Fatalf("reconciliation = current %d work %d, want 0/1", len(reconciled.current), len(reconciled.work)) + } + }) + } +} + +func TestReconcileTranscribeSpeakersRejectsUnsafeEvidenceAndMissingAudio(t *testing.T) { + t.Run("unsafe record path", func(t *testing.T) { + env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac"}) + record := m.Stages["transcribe"].TranscribeSpeakers["alice"] + record.Output.LocalPath = "../alice.json" + m.Stages["transcribe"].TranscribeSpeakers["alice"] = record + _, err := reconcileTranscribeSpeakers(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "unsafe") { + t.Fatalf("error = %v, want unsafe record path", err) + } + }) + + t.Run("non-regular output", func(t *testing.T) { + env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac"}) + path := filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "alice.json") + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(path, 0o755); err != nil { + t.Fatal(err) + } + _, err := reconcileTranscribeSpeakers(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "not a regular file") { + t.Fatalf("error = %v, want non-regular safety error", err) + } + }) + + t.Run("missing audio", func(t *testing.T) { + env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac"}) + if err := os.Remove(filepath.Join(sessionPathsForEnv(env, m.SessionID).AudioDir, "alice.flac")); err != nil { + t.Fatal(err) + } + _, err := reconcileTranscribeSpeakers(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "not found") { + t.Fatalf("error = %v, want missing audio input error", err) + } + }) +} + +func TestReconcileTranscribeSpeakersForceAndLegacyStateSelectAllWork(t *testing.T) { + for _, test := range []struct { + name string + mutate func(*Env, *manifest.Manifest) + }{ + {name: "force", mutate: func(env *Env, _ *manifest.Manifest) { env.Force = true }}, + {name: "legacy aggregate only", mutate: func(_ *Env, m *manifest.Manifest) { + stage := m.Stages["transcribe"] + stage.TranscribeStateVersion = 0 + stage.TranscribeSpeakers = nil + }}, + {name: "versioned empty", mutate: func(_ *Env, m *manifest.Manifest) { + m.Stages["transcribe"].TranscribeSpeakers = map[string]manifest.TranscribeSpeakerRecord{} + }}, + } { + t.Run(test.name, func(t *testing.T) { + env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac", "bob.flac"}) + test.mutate(env, m) + reconciled, err := reconcileTranscribeSpeakers(context.Background(), env, m) + if err != nil { + t.Fatalf("reconcileTranscribeSpeakers() error = %v", err) + } + if len(reconciled.current) != 0 || len(reconciled.work) != 2 { + t.Fatalf("reconciliation = current %d work %d, want 0/2", len(reconciled.current), len(reconciled.work)) + } + }) + } +} + +func TestReconcileTranscribeSpeakersHonorsCancellationWhileInspectingInputs(t *testing.T) { + env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac"}) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err := reconcileTranscribeSpeakers(ctx, env, m) + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context cancellation", err) + } +} + +func setupCurrentTranscribeReconciliation(t *testing.T, audioFiles []string) (*Env, *manifest.Manifest) { + t.Helper() + env, m := setupTranscribeEnv(t, audioFiles) + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + initial, err := reconcileTranscribeSpeakers(context.Background(), env, m) + if err != nil { + t.Fatalf("initial reconciliation error = %v", err) + } + records := make(map[string]manifest.TranscribeSpeakerRecord, len(initial.work)) + for _, candidate := range initial.work { + if err := writeJSONFile(candidate.canonicalOutputPath, map[string]any{"speaker": candidate.speakerID}); err != nil { + t.Fatal(err) + } + checksum, err := artifacts.SHA256File(candidate.canonicalOutputPath) + if err != nil { + t.Fatal(err) + } + info, err := os.Stat(candidate.canonicalOutputPath) + if err != nil { + t.Fatal(err) + } + records[candidate.speakerID] = manifest.TranscribeSpeakerRecord{ + FingerprintVersion: manifest.TranscribeFingerprintContractVersion, + Fingerprint: candidate.fingerprint, + AudioPath: candidate.audioRelativePath, + AudioSHA256: candidate.audioSHA256, + AudioSize: candidate.audioSize, + Output: manifest.ArtifactRecord{ + Kind: "transcript_raw", LocalPath: candidate.outputRelativePath, + Checksum: checksum, ProducerRunID: "run-current", + }, + OutputSize: info.Size(), ProducerRunID: "run-current", + } + } + now := time.Date(2026, 9, 21, 12, 0, 0, 0, time.UTC) + m.Stages["transcribe"] = &manifest.StageRecord{ + Name: "transcribe", Status: manifest.StatusFailed, CreatedAt: now, UpdatedAt: now, + TranscribeStateVersion: manifest.TranscribeStateContractVersion, + TranscribeSpeakers: records, + } + return env, m +} + +func reconciliationSpeakerIDs(jobs []transcribeReconciliationJob) []string { + ids := make([]string, 0, len(jobs)) + for _, job := range jobs { + ids = append(ids, job.speakerID) + } + return ids +}