Reconcile reusable speaker transcriptions

2026-09-21 16:09:45 +00:00
parent 7b0bf99ed1
commit 26aa0cce29
4 changed files with 700 additions and 52 deletions
--- 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
 		}
--- a/internal/stage/transcribe_reconciliation.go
+++ 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
+	}
+}
--- a/internal/stage/transcribe_reconciliation_test.go
+++ 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
+}